server/lib/job-queue/handlers/activitypub-http-fetcher.ts

   1 import * as kue from 'kue'
   2 import { logger } from '../../../helpers/logger'
   3 import { doRequest } from '../../../helpers/requests'
   4 import { ACTIVITY_PUB, JOB_REQUEST_TIMEOUT } from '../../../initializers'
   5 import { processActivities } from '../../activitypub/process'
   6 import { ActivitypubHttpBroadcastPayload } from './activitypub-http-broadcast'
   7
   8 export type ActivitypubHttpFetcherPayload = {
   9   uris: string[]
  10 }
  11
  12 async function processActivityPubHttpFetcher (job: kue.Job) {
  13   logger.info('Processing ActivityPub fetcher in job %d.', job.id)
  14
  15   const payload = job.data as ActivitypubHttpBroadcastPayload
  16
  17   const options = {
  18     method: 'GET',
  19     uri: '',
  20     json: true,
  21     activityPub: true,
  22     timeout: JOB_REQUEST_TIMEOUT
  23   }
  24
  25   for (const uri of payload.uris) {
  26     options.uri = uri
  27     logger.info('Fetching ActivityPub data on %s.', uri)
  28
  29     const response = await doRequest(options)
  30     const firstBody = response.body
  31
  32     if (firstBody.first && Array.isArray(firstBody.first.orderedItems)) {
  33       const activities = firstBody.first.orderedItems
  34
  35       logger.info('Processing %i items ActivityPub fetcher for %s.', activities.length, options.uri)
  36
  37       await processActivities(activities)
  38     }
  39
  40     let limit = ACTIVITY_PUB.FETCH_PAGE_LIMIT
  41     let i = 0
  42     let nextLink = firstBody.first.next
  43     while (nextLink && i < limit) {
  44       options.uri = nextLink
  45
  46       const { body } = await doRequest(options)
  47       nextLink = body.next
  48       i++
  49
  50       if (Array.isArray(body.orderedItems)) {
  51         const activities = body.orderedItems
  52         logger.info('Processing %i items ActivityPub fetcher for %s.', activities.length, options.uri)
  53
  54         await processActivities(activities)
  55       }
  56     }
  57   }
  58 }
  59
  60 // ---------------------------------------------------------------------------
  61
  62 export {
  63   processActivityPubHttpFetcher
  64 }