server/lib/jobs/activitypub-http-job-scheduler/activitypub-http-fetcher-handler.ts

   1 import { logger } from '../../../helpers/logger'
   2 import { doRequest } from '../../../helpers/requests'
   3 import { ACTIVITY_PUB } from '../../../initializers'
   4 import { processActivities } from '../../activitypub/process'
   5 import { ActivityPubHttpPayload } from './activitypub-http-job-scheduler'
   6
   7 async function process (payload: ActivityPubHttpPayload, jobId: number) {
   8   logger.info('Processing ActivityPub fetcher in job %d.', jobId)
   9
  10   const options = {
  11     method: 'GET',
  12     uri: '',
  13     json: true,
  14     activityPub: true
  15   }
  16
  17   for (const uri of payload.uris) {
  18     options.uri = uri
  19     logger.info('Fetching ActivityPub data on %s.', uri)
  20
  21     const response = await doRequest(options)
  22     const firstBody = response.body
  23
  24     if (firstBody.first && Array.isArray(firstBody.first.orderedItems)) {
  25       const activities = firstBody.first.orderedItems
  26
  27       logger.info('Processing %i items ActivityPub fetcher for %s.', activities.length, options.uri)
  28
  29       await processActivities(activities)
  30     }
  31
  32     let limit = ACTIVITY_PUB.FETCH_PAGE_LIMIT
  33     let i = 0
  34     let nextLink = firstBody.first.next
  35     while (nextLink && i < limit) {
  36       options.uri = nextLink
  37
  38       const { body } = await doRequest(options)
  39       nextLink = body.next
  40       i++
  41
  42       if (Array.isArray(body.orderedItems)) {
  43         const activities = body.orderedItems
  44         logger.info('Processing %i items ActivityPub fetcher for %s.', activities.length, options.uri)
  45
  46         await processActivities(activities)
  47       }
  48     }
  49   }
  50 }
  51
  52 function onError (err: Error, jobId: number) {
  53   logger.error('Error when fetcher ActivityPub request in job %d.', jobId, err)
  54   return Promise.resolve()
  55 }
  56
  57 function onSuccess (jobId: number) {
  58   logger.info('Job %d is a success.', jobId)
  59   return Promise.resolve()
  60 }
  61
  62 // ---------------------------------------------------------------------------
  63
  64 export {
  65   process,
  66   onError,
  67   onSuccess
  68 }