import { Transaction } from 'sequelize' ; import { ActorFollowHealthCache } from '@server/lib/actor-follow-health-cache' ; import { getServerActor } from '@server/models/application/application' ; import { ContextType } from '@shared/models/activitypub/context' ; import { Activity , ActivityAudience } from '../../../../shared/models/activitypub' ; import { afterCommitIfTransaction } from '../../../helpers/database-utils' ; import { logger } from '../../../helpers/logger' ; import { ActorModel } from '../../../models/actor/actor' ; import { ActorFollowModel } from '../../../models/actor/actor-follow' ; import { MActor , MActorId , MActorLight , MActorWithInboxes , MVideoAccountLight , MVideoId , MVideoImmutable } from '../../../types/models' ; import { JobQueue } from '../../job-queue' ; import { getActorsInvolvedInVideo , getAudienceFromFollowersOf , getRemoteVideoAudience } from '../audience' ; async function sendVideoRelatedActivity ( activityBuilder , options ) { const { byActor , video , transaction , contextType } = options ; const actorsInvolvedInVideo = await getActorsInvolvedInVideo ( video , transaction ) ; if ( video . isOwned ( ) === false ) { let accountActor = ( video as MVideoAccountLight ) . VideoChannel ?. Account ?. Actor ; if ( ! accountActor ) accountActor = await ActorModel . loadAccountActorByVideoId ( video . id , transaction ) ; const audience = getRemoteVideoAudience ( accountActor , actorsInvolvedInVideo ) ; const activity = activityBuilder ( audience ) ; return afterCommitIfTransaction ( transaction , ( ) => { return unicastTo ( activity , byActor , accountActor . getSharedInbox ( ) , contextType ) ; } ) ; } const audience = getAudienceFromFollowersOf ( actorsInvolvedInVideo ) ; const activity = activityBuilder ( audience ) ; const actorsException = [ byActor ] ; return broadcastToFollowers ( activity , byActor , actorsInvolvedInVideo , transaction , actorsException , contextType ) ; } async function forwardVideoRelatedActivity ( activity , t , followersException , video ) { const additionalActors = await getActorsInvolvedInVideo ( video , t ) ; const additionalFollowerUrls = additionalActors . map ( a => a . followersUrl ) ; return forwardActivity ( activity , t , followersException , additionalFollowerUrls ) ; } async function forwardActivity ( activity , t , followersException = [ ] , additionalFollowerUrls = [ ] ) { logger . info ( 'Forwarding activity %s.' , activity . id ) ; const to = activity . to || [ ] ; const cc = activity . cc || [ ] ; const followersUrls = additionalFollowerUrls ; for ( const dest of to . concat ( cc ) ) { if ( dest . endsWith ( '/followers' ) ) { followersUrls . push ( dest ) ; } } const toActorFollowers = await ActorModel . listByFollowersUrls ( followersUrls , t ) ; const uris = await computeFollowerUris ( toActorFollowers , followersException , t ) ; if ( uris . length === 0 ) { logger . info ( '0 followers for %s, no forwarding.' , toActorFollowers . map ( a => a . id ) . join ( ', ' ) ) ; return undefined ; } logger . debug ( 'Creating forwarding job.' , { uris } ) ; const payload = { uris , body : activity } ; return afterCommitIfTransaction ( t , ( ) => JobQueue . Instance . createJob ( { type : 'activitypub-http-broadcast' , payload } ) ) ; } async function broadcastToFollowers ( data , byActor , toFollowersOf , t , actorsException = [ ] , contextType ? ) { const uris = await computeFollowerUris ( toFollowersOf , actorsException , t ) ; return afterCommitIfTransaction ( t , ( ) => broadcastTo ( uris , data , byActor , contextType ) ) ; } async function broadcastToActors ( data , byActor , toActors , t ? , actorsException = [ ] , contextType ? ) { const uris = await computeUris ( toActors , actorsException ) ; return afterCommitIfTransaction ( t , ( ) => broadcastTo ( uris , data , byActor , contextType ) ) ; } function broadcastTo ( uris , data , byActor , contextType ? ) { if ( uris . length === 0 ) return undefined ; const broadcastUris = [ ] ; const unicastUris = [ ] ; for ( const uri of uris ) { if ( ActorFollowHealthCache . Instance . isBadInbox ( uri ) ) { unicastUris . push ( uri ) ; } else { broadcastUris . push ( uri ) ; } } logger . debug ( 'Creating broadcast job.' , { broadcastUris , unicastUris } ) ; if ( broadcastUris . length !== 0 ) { const payload = { uris : broadcastUris , signatureActorId : byActor . id , body : data , contextType } ; JobQueue . Instance . createJob ( { type : 'activitypub-http-broadcast' , payload } ) ; } for ( const unicastUri of unicastUris ) { const payload = { uri : unicastUri , signatureActorId : byActor . id , body : data , contextType } ; JobQueue . Instance . createJob ( { type : 'activitypub-http-unicast' , payload } ) ; } } function unicastTo ( data , byActor , toActorUrl , contextType ? ) { logger . debug ( 'Creating unicast job.' , { uri : toActorUrl } ) ; const payload = { uri : toActorUrl , signatureActorId : byActor . id , body : data , contextType } ; JobQueue . Instance . createJob ( { type : 'activitypub-http-unicast' , payload } ) ; } export { broadcastToFollowers , unicastTo , forwardActivity , broadcastToActors , forwardVideoRelatedActivity , sendVideoRelatedActivity } ; async function computeFollowerUris ( toFollowersOf , actorsException , t ) { const toActorFollowerIds = toFollowersOf . map ( a => a . id ) ; const result = await ActorFollowModel . listAcceptedFollowerSharedInboxUrls ( toActorFollowerIds , t ) ; const sharedInboxesException = await buildSharedInboxesException ( actorsException ) ; return result . data . filter ( sharedInbox => sharedInboxesException . includes ( sharedInbox ) === false ) ; } async function computeUris ( toActors , actorsException = [ ] ) { const serverActor = await getServerActor ( ) ; const targetUrls = toActors . filter ( a => a . id !== serverActor . id ) . map ( a => a . getSharedInbox ( ) ) ; const toActorSharedInboxesSet = new Set ( targetUrls ) ; const sharedInboxesException = await buildSharedInboxesException ( actorsException ) ; return Array . from ( toActorSharedInboxesSet ) . filter ( sharedInbox => sharedInboxesException . includes ( sharedInbox ) === false ) ; } async function buildSharedInboxesException ( actorsException ) { const serverActor = await getServerActor ( ) ; return actorsException . map ( f => f . getSharedInbox ( ) ) . concat ( [ serverActor . sharedInboxUrl ] ) ; }