123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280 |
- import { Transaction } from 'sequelize'
- import { ActorFollowHealthCache } from '@server/lib/actor-follow-health-cache'
- import { getServerActor } from '@server/models/application/application'
- import { Activity, ActivityAudience, ActivitypubHttpBroadcastPayload } from '@shared/models'
- import { ContextType } from '@shared/models/activitypub/context'
- 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, getOriginVideoAudience } from './audience-utils'
- async function sendVideoRelatedActivity (activityBuilder: (audience: ActivityAudience) => Activity, options: {
- byActor: MActorLight
- video: MVideoImmutable | MVideoAccountLight
- contextType: ContextType
- transaction?: Transaction
- }) {
- const { byActor, video, transaction, contextType } = options
- // Send to origin
- if (video.isOwned() === false) {
- return sendVideoActivityToOrigin(activityBuilder, options)
- }
- const actorsInvolvedInVideo = await getActorsInvolvedInVideo(video, transaction)
- // Send to followers
- const audience = getAudienceFromFollowersOf(actorsInvolvedInVideo)
- const activity = activityBuilder(audience)
- const actorsException = [ byActor ]
- return broadcastToFollowers({
- data: activity,
- byActor,
- toFollowersOf: actorsInvolvedInVideo,
- transaction,
- actorsException,
- contextType
- })
- }
- async function sendVideoActivityToOrigin (activityBuilder: (audience: ActivityAudience) => Activity, options: {
- byActor: MActorLight
- video: MVideoImmutable | MVideoAccountLight
- contextType: ContextType
- actorsInvolvedInVideo?: MActorLight[]
- transaction?: Transaction
- }) {
- const { byActor, video, actorsInvolvedInVideo, transaction, contextType } = options
- if (video.isOwned()) throw new Error('Cannot send activity to owned video origin ' + video.url)
- let accountActor: MActorLight = (video as MVideoAccountLight).VideoChannel?.Account?.Actor
- if (!accountActor) accountActor = await ActorModel.loadAccountActorByVideoId(video.id, transaction)
- const audience = getOriginVideoAudience(accountActor, actorsInvolvedInVideo)
- const activity = activityBuilder(audience)
- return afterCommitIfTransaction(transaction, () => {
- return unicastTo({
- data: activity,
- byActor,
- toActorUrl: accountActor.getSharedInbox(),
- contextType
- })
- })
- }
- // ---------------------------------------------------------------------------
- async function forwardVideoRelatedActivity (
- activity: Activity,
- t: Transaction,
- followersException: MActorWithInboxes[],
- video: MVideoId
- ) {
- // Mastodon does not add our announces in audience, so we forward to them manually
- const additionalActors = await getActorsInvolvedInVideo(video, t)
- const additionalFollowerUrls = additionalActors.map(a => a.followersUrl)
- return forwardActivity(activity, t, followersException, additionalFollowerUrls)
- }
- async function forwardActivity (
- activity: Activity,
- t: Transaction,
- followersException: MActorWithInboxes[] = [],
- additionalFollowerUrls: string[] = []
- ) {
- 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: ActivitypubHttpBroadcastPayload = {
- uris,
- body: activity,
- contextType: null
- }
- return afterCommitIfTransaction(t, () => JobQueue.Instance.createJob({ type: 'activitypub-http-broadcast', payload }))
- }
- // ---------------------------------------------------------------------------
- async function broadcastToFollowers (options: {
- data: any
- byActor: MActorId
- toFollowersOf: MActorId[]
- transaction: Transaction
- contextType: ContextType
- actorsException?: MActorWithInboxes[]
- }) {
- const { data, byActor, toFollowersOf, transaction, contextType, actorsException = [] } = options
- const uris = await computeFollowerUris(toFollowersOf, actorsException, transaction)
- return afterCommitIfTransaction(transaction, () => {
- return broadcastTo({
- uris,
- data,
- byActor,
- contextType
- })
- })
- }
- async function broadcastToActors (options: {
- data: any
- byActor: MActorId
- toActors: MActor[]
- transaction: Transaction
- contextType: ContextType
- actorsException?: MActorWithInboxes[]
- }) {
- const { data, byActor, toActors, transaction, contextType, actorsException = [] } = options
- const uris = await computeUris(toActors, actorsException)
- return afterCommitIfTransaction(transaction, () => {
- return broadcastTo({
- uris,
- data,
- byActor,
- contextType
- })
- })
- }
- function broadcastTo (options: {
- uris: string[]
- data: any
- byActor: MActorId
- contextType: ContextType
- }) {
- const { uris, data, byActor, contextType } = options
- if (uris.length === 0) return undefined
- const broadcastUris: string[] = []
- const unicastUris: string[] = []
- // Bad URIs could be slow to respond, prefer to process them in a dedicated queue
- 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 (options: {
- data: any
- byActor: MActorId
- toActorUrl: string
- contextType: ContextType
- }) {
- const { data, byActor, toActorUrl, contextType } = options
- 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,
- sendVideoActivityToOrigin,
- forwardVideoRelatedActivity,
- sendVideoRelatedActivity
- }
- // ---------------------------------------------------------------------------
- async function computeFollowerUris (toFollowersOf: MActorId[], actorsException: MActorWithInboxes[], t: Transaction) {
- 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: MActor[], actorsException: MActorWithInboxes[] = []) {
- const serverActor = await getServerActor()
- const targetUrls = toActors
- .filter(a => a.id !== serverActor.id) // Don't send to ourselves
- .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: MActorWithInboxes[]) {
- const serverActor = await getServerActor()
- return actorsException
- .map(f => f.getSharedInbox())
- .concat([ serverActor.sharedInboxUrl ])
- }
|