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