server/lib/activitypub/send/shared/send-utils.ts

   1 import { Transaction } from 'sequelize'
   2 import { ActorFollowHealthCache } from '@server/lib/actor-follow-health-cache'
   3 import { getServerActor } from '@server/models/application/application'
   4 import { Activity, ActivityAudience } from '@shared/models'
   5 import { ContextType } from '@shared/models/activitypub/context'
   6 import { afterCommitIfTransaction } from '../../../../helpers/database-utils'
   7 import { logger } from '../../../../helpers/logger'
   8 import { ActorModel } from '../../../../models/actor/actor'
   9 import { ActorFollowModel } from '../../../../models/actor/actor-follow'
  10 import { MActor, MActorId, MActorLight, MActorWithInboxes, MVideoAccountLight, MVideoId, MVideoImmutable } from '../../../../types/models'
  11 import { JobQueue } from '../../../job-queue'
  12 import { getActorsInvolvedInVideo, getAudienceFromFollowersOf, getOriginVideoAudience } from './audience-utils'
  13
  14 async function sendVideoRelatedActivity (activityBuilder: (audience: ActivityAudience) => Activity, options: {
  15   byActor: MActorLight
  16   video: MVideoImmutable | MVideoAccountLight
  17   transaction?: Transaction
  18   contextType?: ContextType
  19 }) {
  20   const { byActor, video, transaction, contextType } = options
  21
  22   const actorsInvolvedInVideo = await getActorsInvolvedInVideo(video, transaction)
  23
  24   // Send to origin
  25   if (video.isOwned() === false) {
  26     return sendVideoActivityToOrigin(activityBuilder, options)
  27   }
  28
  29   // Send to followers
  30   const audience = getAudienceFromFollowersOf(actorsInvolvedInVideo)
  31   const activity = activityBuilder(audience)
  32
  33   const actorsException = [ byActor ]
  34
  35   return broadcastToFollowers(activity, byActor, actorsInvolvedInVideo, transaction, actorsException, contextType)
  36 }
  37
  38 async function sendVideoActivityToOrigin (activityBuilder: (audience: ActivityAudience) => Activity, options: {
  39   byActor: MActorLight
  40   video: MVideoImmutable | MVideoAccountLight
  41   actorsInvolvedInVideo?: MActorLight[]
  42   transaction?: Transaction
  43   contextType?: ContextType
  44 }) {
  45   const { byActor, video, actorsInvolvedInVideo, transaction, contextType } = options
  46
  47   if (video.isOwned()) throw new Error('Cannot send activity to owned video origin ' + video.url)
  48
  49   let accountActor: MActorLight = (video as MVideoAccountLight).VideoChannel?.Account?.Actor
  50   if (!accountActor) accountActor = await ActorModel.loadAccountActorByVideoId(video.id, transaction)
  51
  52   const audience = getOriginVideoAudience(accountActor, actorsInvolvedInVideo)
  53   const activity = activityBuilder(audience)
  54
  55   return afterCommitIfTransaction(transaction, () => {
  56     return unicastTo(activity, byActor, accountActor.getSharedInbox(), contextType)
  57   })
  58 }
  59
  60 // ---------------------------------------------------------------------------
  61
  62 async function forwardVideoRelatedActivity (
  63   activity: Activity,
  64   t: Transaction,
  65   followersException: MActorWithInboxes[],
  66   video: MVideoId
  67 ) {
  68   // Mastodon does not add our announces in audience, so we forward to them manually
  69   const additionalActors = await getActorsInvolvedInVideo(video, t)
  70   const additionalFollowerUrls = additionalActors.map(a => a.followersUrl)
  71
  72   return forwardActivity(activity, t, followersException, additionalFollowerUrls)
  73 }
  74
  75 async function forwardActivity (
  76   activity: Activity,
  77   t: Transaction,
  78   followersException: MActorWithInboxes[] = [],
  79   additionalFollowerUrls: string[] = []
  80 ) {
  81   logger.info('Forwarding activity %s.', activity.id)
  82
  83   const to = activity.to || []
  84   const cc = activity.cc || []
  85
  86   const followersUrls = additionalFollowerUrls
  87   for (const dest of to.concat(cc)) {
  88     if (dest.endsWith('/followers')) {
  89       followersUrls.push(dest)
  90     }
  91   }
  92
  93   const toActorFollowers = await ActorModel.listByFollowersUrls(followersUrls, t)
  94   const uris = await computeFollowerUris(toActorFollowers, followersException, t)
  95
  96   if (uris.length === 0) {
  97     logger.info('0 followers for %s, no forwarding.', toActorFollowers.map(a => a.id).join(', '))
  98     return undefined
  99   }
 100
 101   logger.debug('Creating forwarding job.', { uris })
 102
 103   const payload = {
 104     uris,
 105     body: activity
 106   }
 107   return afterCommitIfTransaction(t, () => JobQueue.Instance.createJob({ type: 'activitypub-http-broadcast', payload }))
 108 }
 109
 110 // ---------------------------------------------------------------------------
 111
 112 async function broadcastToFollowers (
 113   data: any,
 114   byActor: MActorId,
 115   toFollowersOf: MActorId[],
 116   t: Transaction,
 117   actorsException: MActorWithInboxes[] = [],
 118   contextType?: ContextType
 119 ) {
 120   const uris = await computeFollowerUris(toFollowersOf, actorsException, t)
 121
 122   return afterCommitIfTransaction(t, () => broadcastTo(uris, data, byActor, contextType))
 123 }
 124
 125 async function broadcastToActors (
 126   data: any,
 127   byActor: MActorId,
 128   toActors: MActor[],
 129   t?: Transaction,
 130   actorsException: MActorWithInboxes[] = [],
 131   contextType?: ContextType
 132 ) {
 133   const uris = await computeUris(toActors, actorsException)
 134   return afterCommitIfTransaction(t, () => broadcastTo(uris, data, byActor, contextType))
 135 }
 136
 137 function broadcastTo (uris: string[], data: any, byActor: MActorId, contextType?: ContextType) {
 138   if (uris.length === 0) return undefined
 139
 140   const broadcastUris: string[] = []
 141   const unicastUris: string[] = []
 142
 143   // Bad URIs could be slow to respond, prefer to process them in a dedicated queue
 144   for (const uri of uris) {
 145     if (ActorFollowHealthCache.Instance.isBadInbox(uri)) {
 146       unicastUris.push(uri)
 147     } else {
 148       broadcastUris.push(uri)
 149     }
 150   }
 151
 152   logger.debug('Creating broadcast job.', { broadcastUris, unicastUris })
 153
 154   if (broadcastUris.length !== 0) {
 155     const payload = {
 156       uris: broadcastUris,
 157       signatureActorId: byActor.id,
 158       body: data,
 159       contextType
 160     }
 161
 162     JobQueue.Instance.createJob({ type: 'activitypub-http-broadcast', payload })
 163   }
 164
 165   for (const unicastUri of unicastUris) {
 166     const payload = {
 167       uri: unicastUri,
 168       signatureActorId: byActor.id,
 169       body: data,
 170       contextType
 171     }
 172
 173     JobQueue.Instance.createJob({ type: 'activitypub-http-unicast', payload })
 174   }
 175 }
 176
 177 function unicastTo (data: any, byActor: MActorId, toActorUrl: string, contextType?: ContextType) {
 178   logger.debug('Creating unicast job.', { uri: toActorUrl })
 179
 180   const payload = {
 181     uri: toActorUrl,
 182     signatureActorId: byActor.id,
 183     body: data,
 184     contextType
 185   }
 186
 187   JobQueue.Instance.createJob({ type: 'activitypub-http-unicast', payload })
 188 }
 189
 190 // ---------------------------------------------------------------------------
 191
 192 export {
 193   broadcastToFollowers,
 194   unicastTo,
 195   forwardActivity,
 196   broadcastToActors,
 197   sendVideoActivityToOrigin,
 198   forwardVideoRelatedActivity,
 199   sendVideoRelatedActivity
 200 }
 201
 202 // ---------------------------------------------------------------------------
 203
 204 async function computeFollowerUris (toFollowersOf: MActorId[], actorsException: MActorWithInboxes[], t: Transaction) {
 205   const toActorFollowerIds = toFollowersOf.map(a => a.id)
 206
 207   const result = await ActorFollowModel.listAcceptedFollowerSharedInboxUrls(toActorFollowerIds, t)
 208   const sharedInboxesException = await buildSharedInboxesException(actorsException)
 209
 210   return result.data.filter(sharedInbox => sharedInboxesException.includes(sharedInbox) === false)
 211 }
 212
 213 async function computeUris (toActors: MActor[], actorsException: MActorWithInboxes[] = []) {
 214   const serverActor = await getServerActor()
 215   const targetUrls = toActors
 216     .filter(a => a.id !== serverActor.id) // Don't send to ourselves
 217     .map(a => a.getSharedInbox())
 218
 219   const toActorSharedInboxesSet = new Set(targetUrls)
 220
 221   const sharedInboxesException = await buildSharedInboxesException(actorsException)
 222   return Array.from(toActorSharedInboxesSet)
 223               .filter(sharedInbox => sharedInboxesException.includes(sharedInbox) === false)
 224 }
 225
 226 async function buildSharedInboxesException (actorsException: MActorWithInboxes[]) {
 227   const serverActor = await getServerActor()
 228
 229   return actorsException
 230     .map(f => f.getSharedInbox())
 231     .concat([ serverActor.sharedInboxUrl ])
 232 }