X-Git-Url: https://git.immae.eu/?a=blobdiff_plain;f=server%2Fcontrollers%2Factivitypub%2Finbox.ts;h=c5edf86b757b2c7e573ef65532e76b114865c756;hb=bdd428a6d9138d751f8cde82867022a93f1a76cc;hp=73fbd94aa3fd3a27a822914bd2ea366783f0f113;hpb=ee28cdf1c488ac6ad6f3d06804b9c5e63eea1965;p=github%2FChocobozzz%2FPeerTube.git diff --git a/server/controllers/activitypub/inbox.ts b/server/controllers/activitypub/inbox.ts index 73fbd94aa..c5edf86b7 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,7 +14,7 @@ inboxRouter.post('/inbox', signatureValidator, asyncMiddleware(checkSignature), asyncMiddleware(activityPubValidator), - asyncMiddleware(inboxController) + inboxController ) inboxRouter.post('/accounts/:name/inbox', @@ -21,7 +22,14 @@ inboxRouter.post('/accounts/:name/inbox', asyncMiddleware(checkSignature), asyncMiddleware(localAccountValidator), asyncMiddleware(activityPubValidator), - asyncMiddleware(inboxController) + inboxController +) +inboxRouter.post('/video-channels/:name/inbox', + signatureValidator, + asyncMiddleware(checkSignature), + asyncMiddleware(localVideoChannelValidator), + asyncMiddleware(activityPubValidator), + inboxController ) // --------------------------------------------------------------------------- @@ -32,15 +40,25 @@ export { // --------------------------------------------------------------------------- -async function inboxController (req: express.Request, res: express.Response, next: express.NextFunction) { - const rootActivity: RootActivity = req.body - let activities: Activity[] = [] +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() + }) +}) - console.log(rootActivity) +function inboxController (req: express.Request, res: express.Response) { + const rootActivity: RootActivity = req.body + let activities: Activity[] - if ([ 'Collection', 'CollectionPage' ].indexOf(rootActivity.type) !== -1) { + if ([ 'Collection', 'CollectionPage' ].includes(rootActivity.type)) { activities = (rootActivity as ActivityPubCollection).items - } else if ([ 'OrderedCollection', 'OrderedCollectionPage' ].indexOf(rootActivity.type) !== -1) { + } else if ([ 'OrderedCollection', 'OrderedCollectionPage' ].includes(rootActivity.type)) { activities = (rootActivity as ActivityPubOrderedCollection).orderedItems } else { activities = [ rootActivity as Activity ] @@ -51,16 +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() }