X-Git-Url: https://git.immae.eu/?a=blobdiff_plain;f=server%2Fcontrollers%2Factivitypub%2Finbox.ts;h=3b8fb34a844aaf1ae443d71924e7fe8499c3c488;hb=2ad9dcda240ee843c5e4a5b98cc94f7b2aab2c89;hp=bd0d7a9c8bb6ac4b0f27114664eaeba227afabc6;hpb=285fe7c93072b2a8e6a9af6b7e8ffcdefcffbddf;p=github%2FChocobozzz%2FPeerTube.git diff --git a/server/controllers/activitypub/inbox.ts b/server/controllers/activitypub/inbox.ts index bd0d7a9c8..3b8fb34a8 100644 --- a/server/controllers/activitypub/inbox.ts +++ b/server/controllers/activitypub/inbox.ts @@ -3,9 +3,10 @@ import { Activity, ActivityPubCollection, ActivityPubOrderedCollection, RootActi import { isActivityValid } from '../../helpers/custom-validators/activitypub/activity' import { logger } from '../../helpers/logger' import { processActivities } from '../../lib/activitypub/process/process' -import { asyncMiddleware, checkSignature, localAccountValidator, signatureValidator } from '../../middlewares' +import { asyncMiddleware, checkSignature, localAccountValidator, localVideoChannelValidator, signatureValidator } from '../../middlewares' import { activityPubValidator } from '../../middlewares/validators/activitypub/activity' -import { ActorModel } from '../../models/activitypub/actor' +import { queue } from 'async' +import { MActorDefault, MActorSignature } from '../../typings/models' const inboxRouter = express.Router() @@ -13,15 +14,22 @@ inboxRouter.post('/inbox', signatureValidator, asyncMiddleware(checkSignature), asyncMiddleware(activityPubValidator), - asyncMiddleware(inboxController) + inboxController ) inboxRouter.post('/accounts/:name/inbox', signatureValidator, asyncMiddleware(checkSignature), - localAccountValidator, + asyncMiddleware(localAccountValidator), asyncMiddleware(activityPubValidator), - asyncMiddleware(inboxController) + inboxController +) +inboxRouter.post('/video-channels/:name/inbox', + signatureValidator, + asyncMiddleware(checkSignature), + asyncMiddleware(localVideoChannelValidator), + asyncMiddleware(activityPubValidator), + inboxController ) // --------------------------------------------------------------------------- @@ -32,9 +40,21 @@ export { // --------------------------------------------------------------------------- -async function inboxController (req: express.Request, res: express.Response, next: express.NextFunction) { +type QueueParam = { activities: Activity[], signatureActor?: MActorSignature, inboxActor?: MActorDefault } +const inboxQueue = queue((task, cb) => { + const options = { signatureActor: task.signatureActor, inboxActor: task.inboxActor } + + processActivities(task.activities, options) + .then(() => cb()) + .catch(err => { + logger.error('Error in process activities.', { err }) + cb() + }) +}) + +function inboxController (req: express.Request, res: express.Response) { const rootActivity: RootActivity = req.body - let activities: Activity[] = [] + let activities: Activity[] if ([ 'Collection', 'CollectionPage' ].indexOf(rootActivity.type) !== -1) { activities = (rootActivity as ActivityPubCollection).items @@ -49,14 +69,15 @@ async function inboxController (req: express.Request, res: express.Response, nex activities = activities.filter(a => isActivityValid(a)) logger.debug('We keep %d activities.', activities.length, { activities }) - let specificActor: ActorModel = undefined - if (res.locals.account) { - specificActor = res.locals.account - } else if (res.locals.videoChannel) { - specificActor = res.locals.videoChannel - } + const accountOrChannel = res.locals.account || res.locals.videoChannel + + logger.info('Receiving inbox requests for %d activities by %s.', activities.length, res.locals.signature.actor.url) - await processActivities(activities, res.locals.signature.actor, specificActor) + inboxQueue.push({ + activities, + signatureActor: res.locals.signature.actor, + inboxActor: accountOrChannel ? accountOrChannel.Actor : undefined + }) - res.status(204).end() + return res.status(204).end() }