summaryrefslogtreecommitdiff
path: root/packages/backend/src/queue
diff options
context:
space:
mode:
authordakkar <dakkar@thenautilus.net>2024-08-02 12:25:58 +0100
committerdakkar <dakkar@thenautilus.net>2024-08-02 12:25:58 +0100
commitcfa9b852df9e0293865b3acbd67d59265962e552 (patch)
treea408ad670956a45c4e162e4ecc97a3624e2b0f20 /packages/backend/src/queue
parentmerge: rate limit all password checks - fixes #540 (!568) (diff)
parentMerge pull request #14233 from misskey-dev/develop (diff)
downloadsharkey-cfa9b852df9e0293865b3acbd67d59265962e552.tar.gz
sharkey-cfa9b852df9e0293865b3acbd67d59265962e552.tar.bz2
sharkey-cfa9b852df9e0293865b3acbd67d59265962e552.zip
Merge remote-tracking branch 'misskey/master' into feature/misskey-2024.07
Diffstat (limited to 'packages/backend/src/queue')
-rw-r--r--packages/backend/src/queue/QueueProcessorModule.ts6
-rw-r--r--packages/backend/src/queue/QueueProcessorService.ts518
-rw-r--r--packages/backend/src/queue/const.ts3
-rw-r--r--packages/backend/src/queue/processors/EndedPollNotificationProcessorService.ts11
-rw-r--r--packages/backend/src/queue/processors/ImportAntennasProcessorService.ts4
-rw-r--r--packages/backend/src/queue/processors/ImportUserListsProcessorService.ts4
-rw-r--r--packages/backend/src/queue/processors/SystemWebhookDeliverProcessorService.ts87
-rw-r--r--packages/backend/src/queue/processors/UserWebhookDeliverProcessorService.ts (renamed from packages/backend/src/queue/processors/WebhookDeliverProcessorService.ts)6
-rw-r--r--packages/backend/src/queue/types.ts12
9 files changed, 465 insertions, 186 deletions
diff --git a/packages/backend/src/queue/QueueProcessorModule.ts b/packages/backend/src/queue/QueueProcessorModule.ts
index d7316e19e3..7daca687a1 100644
--- a/packages/backend/src/queue/QueueProcessorModule.ts
+++ b/packages/backend/src/queue/QueueProcessorModule.ts
@@ -11,7 +11,8 @@ import { QueueProcessorService } from './QueueProcessorService.js';
import { DeliverProcessorService } from './processors/DeliverProcessorService.js';
import { EndedPollNotificationProcessorService } from './processors/EndedPollNotificationProcessorService.js';
import { InboxProcessorService } from './processors/InboxProcessorService.js';
-import { WebhookDeliverProcessorService } from './processors/WebhookDeliverProcessorService.js';
+import { UserWebhookDeliverProcessorService } from './processors/UserWebhookDeliverProcessorService.js';
+import { SystemWebhookDeliverProcessorService } from './processors/SystemWebhookDeliverProcessorService.js';
import { CheckExpiredMutingsProcessorService } from './processors/CheckExpiredMutingsProcessorService.js';
import { CleanChartsProcessorService } from './processors/CleanChartsProcessorService.js';
import { CleanProcessorService } from './processors/CleanProcessorService.js';
@@ -75,7 +76,8 @@ import { RelationshipProcessorService } from './processors/RelationshipProcessor
DeleteFileProcessorService,
CleanRemoteFilesProcessorService,
RelationshipProcessorService,
- WebhookDeliverProcessorService,
+ UserWebhookDeliverProcessorService,
+ SystemWebhookDeliverProcessorService,
EndedPollNotificationProcessorService,
DeliverProcessorService,
InboxProcessorService,
diff --git a/packages/backend/src/queue/QueueProcessorService.ts b/packages/backend/src/queue/QueueProcessorService.ts
index 76b6d7fb05..7a6169bf9c 100644
--- a/packages/backend/src/queue/QueueProcessorService.ts
+++ b/packages/backend/src/queue/QueueProcessorService.ts
@@ -5,11 +5,13 @@
import { Inject, Injectable, OnApplicationShutdown } from '@nestjs/common';
import * as Bull from 'bullmq';
+import * as Sentry from '@sentry/node';
import type { Config } from '@/config.js';
import { DI } from '@/di-symbols.js';
import type Logger from '@/logger.js';
import { bindThis } from '@/decorators.js';
-import { WebhookDeliverProcessorService } from './processors/WebhookDeliverProcessorService.js';
+import { UserWebhookDeliverProcessorService } from './processors/UserWebhookDeliverProcessorService.js';
+import { SystemWebhookDeliverProcessorService } from './processors/SystemWebhookDeliverProcessorService.js';
import { EndedPollNotificationProcessorService } from './processors/EndedPollNotificationProcessorService.js';
import { DeliverProcessorService } from './processors/DeliverProcessorService.js';
import { InboxProcessorService } from './processors/InboxProcessorService.js';
@@ -77,7 +79,8 @@ export class QueueProcessorService implements OnApplicationShutdown {
private dbQueueWorker: Bull.Worker;
private deliverQueueWorker: Bull.Worker;
private inboxQueueWorker: Bull.Worker;
- private webhookDeliverQueueWorker: Bull.Worker;
+ private userWebhookDeliverQueueWorker: Bull.Worker;
+ private systemWebhookDeliverQueueWorker: Bull.Worker;
private relationshipQueueWorker: Bull.Worker;
private objectStorageQueueWorker: Bull.Worker;
private endedPollNotificationQueueWorker: Bull.Worker;
@@ -87,7 +90,8 @@ export class QueueProcessorService implements OnApplicationShutdown {
private config: Config,
private queueLoggerService: QueueLoggerService,
- private webhookDeliverProcessorService: WebhookDeliverProcessorService,
+ private userWebhookDeliverProcessorService: UserWebhookDeliverProcessorService,
+ private systemWebhookDeliverProcessorService: SystemWebhookDeliverProcessorService,
private endedPollNotificationProcessorService: EndedPollNotificationProcessorService,
private deliverProcessorService: DeliverProcessorService,
private inboxProcessorService: InboxProcessorService,
@@ -139,207 +143,375 @@ export class QueueProcessorService implements OnApplicationShutdown {
}
//#region system
- this.systemQueueWorker = new Bull.Worker(QUEUE.SYSTEM, (job) => {
- switch (job.name) {
- case 'tickCharts': return this.tickChartsProcessorService.process();
- case 'resyncCharts': return this.resyncChartsProcessorService.process();
- case 'cleanCharts': return this.cleanChartsProcessorService.process();
- case 'aggregateRetention': return this.aggregateRetentionProcessorService.process();
- case 'checkExpiredMutings': return this.checkExpiredMutingsProcessorService.process();
- case 'clean': return this.cleanProcessorService.process();
- default: throw new Error(`unrecognized job type ${job.name} for system`);
- }
- }, {
- ...baseQueueOptions(this.config, QUEUE.SYSTEM),
- autorun: false,
- });
+ {
+ const processer = (job: Bull.Job) => {
+ switch (job.name) {
+ case 'tickCharts': return this.tickChartsProcessorService.process();
+ case 'resyncCharts': return this.resyncChartsProcessorService.process();
+ case 'cleanCharts': return this.cleanChartsProcessorService.process();
+ case 'aggregateRetention': return this.aggregateRetentionProcessorService.process();
+ case 'checkExpiredMutings': return this.checkExpiredMutingsProcessorService.process();
+ case 'clean': return this.cleanProcessorService.process();
+ default: throw new Error(`unrecognized job type ${job.name} for system`);
+ }
+ };
+
+ this.systemQueueWorker = new Bull.Worker(QUEUE.SYSTEM, (job) => {
+ if (this.config.sentryForBackend) {
+ return Sentry.startSpan({ name: 'Queue: System: ' + job.name }, () => processer(job));
+ } else {
+ return processer(job);
+ }
+ }, {
+ ...baseQueueOptions(this.config, QUEUE.SYSTEM),
+ autorun: false,
+ });
- const systemLogger = this.logger.createSubLogger('system');
+ const logger = this.logger.createSubLogger('system');
- this.systemQueueWorker
- .on('active', (job) => systemLogger.debug(`active id=${job.id}`))
- .on('completed', (job, result) => systemLogger.debug(`completed(${result}) id=${job.id}`))
- .on('failed', (job, err) => systemLogger.warn(`failed(${err.stack}) id=${job ? job.id : '-'}`, { job, e: renderError(err) }))
- .on('error', (err: Error) => systemLogger.error(`error ${err.stack}`, { e: renderError(err) }))
- .on('stalled', (jobId) => systemLogger.warn(`stalled id=${jobId}`));
+ this.systemQueueWorker
+ .on('active', (job) => logger.debug(`active id=${job.id}`))
+ .on('completed', (job, result) => logger.debug(`completed(${result}) id=${job.id}`))
+ .on('failed', (job, err: Error) => {
+ logger.error(`failed(${err.stack}) id=${job ? job.id : '-'}`, { job, e: renderError(err) });
+ if (config.sentryForBackend) {
+ Sentry.captureMessage(`Queue: System: ${job?.name ?? '?'}: ${err.message}`, {
+ level: 'error',
+ extra: { job, err },
+ });
+ }
+ })
+ .on('error', (err: Error) => logger.error(`error ${err.stack}`, { e: renderError(err) }))
+ .on('stalled', (jobId) => logger.warn(`stalled id=${jobId}`));
+ }
//#endregion
//#region db
- this.dbQueueWorker = new Bull.Worker(QUEUE.DB, (job) => {
- switch (job.name) {
- case 'deleteDriveFiles': return this.deleteDriveFilesProcessorService.process(job);
- case 'exportAccountData': return this.exportAccountDataProcessorService.process(job);
- case 'exportCustomEmojis': return this.exportCustomEmojisProcessorService.process(job);
- case 'exportNotes': return this.exportNotesProcessorService.process(job);
- case 'exportClips': return this.exportClipsProcessorService.process(job);
- case 'exportFavorites': return this.exportFavoritesProcessorService.process(job);
- case 'exportFollowing': return this.exportFollowingProcessorService.process(job);
- case 'exportMuting': return this.exportMutingProcessorService.process(job);
- case 'exportBlocking': return this.exportBlockingProcessorService.process(job);
- case 'exportUserLists': return this.exportUserListsProcessorService.process(job);
- case 'exportAntennas': return this.exportAntennasProcessorService.process(job);
- case 'importFollowing': return this.importFollowingProcessorService.process(job);
- case 'importNotes': return this.importNotesProcessorService.process(job);
- case 'importTweetsToDb': return this.importNotesProcessorService.processTwitterDb(job);
- case 'importIGToDb': return this.importNotesProcessorService.processIGDb(job);
- case 'importFBToDb': return this.importNotesProcessorService.processFBDb(job);
- case 'importMastoToDb': return this.importNotesProcessorService.processMastoToDb(job);
- case 'importPleroToDb': return this.importNotesProcessorService.processPleroToDb(job);
- case 'importKeyNotesToDb': return this.importNotesProcessorService.processKeyNotesToDb(job);
- case 'importFollowingToDb': return this.importFollowingProcessorService.processDb(job);
- case 'importMuting': return this.importMutingProcessorService.process(job);
- case 'importBlocking': return this.importBlockingProcessorService.process(job);
- case 'importBlockingToDb': return this.importBlockingProcessorService.processDb(job);
- case 'importUserLists': return this.importUserListsProcessorService.process(job);
- case 'importCustomEmojis': return this.importCustomEmojisProcessorService.process(job);
- case 'importAntennas': return this.importAntennasProcessorService.process(job);
- case 'deleteAccount': return this.deleteAccountProcessorService.process(job);
- default: throw new Error(`unrecognized job type ${job.name} for db`);
- }
- }, {
- ...baseQueueOptions(this.config, QUEUE.DB),
- autorun: false,
- });
+ {
+ const processer = (job: Bull.Job) => {
+ switch (job.name) {
+ case 'deleteDriveFiles': return this.deleteDriveFilesProcessorService.process(job);
+ case 'exportCustomEmojis': return this.exportCustomEmojisProcessorService.process(job);
+ case 'exportNotes': return this.exportNotesProcessorService.process(job);
+ case 'exportClips': return this.exportClipsProcessorService.process(job);
+ case 'exportFavorites': return this.exportFavoritesProcessorService.process(job);
+ case 'exportFollowing': return this.exportFollowingProcessorService.process(job);
+ case 'exportMuting': return this.exportMutingProcessorService.process(job);
+ case 'exportBlocking': return this.exportBlockingProcessorService.process(job);
+ case 'exportUserLists': return this.exportUserListsProcessorService.process(job);
+ case 'exportAntennas': return this.exportAntennasProcessorService.process(job);
+ case 'exportAccountData': return this.exportAccountDataProcessorService.process(job);
+ case 'importFollowing': return this.importFollowingProcessorService.process(job);
+ case 'importFollowingToDb': return this.importFollowingProcessorService.processDb(job);
+ case 'importMuting': return this.importMutingProcessorService.process(job);
+ case 'importBlocking': return this.importBlockingProcessorService.process(job);
+ case 'importBlockingToDb': return this.importBlockingProcessorService.processDb(job);
+ case 'importUserLists': return this.importUserListsProcessorService.process(job);
+ case 'importCustomEmojis': return this.importCustomEmojisProcessorService.process(job);
+ case 'importAntennas': return this.importAntennasProcessorService.process(job);
+ case 'importNotes': return this.importNotesProcessorService.process(job);
+ case 'importTweetsToDb': return this.importNotesProcessorService.processTwitterDb(job);
+ case 'importIGToDb': return this.importNotesProcessorService.processIGDb(job);
+ case 'importFBToDb': return this.importNotesProcessorService.processFBDb(job);
+ case 'importMastoToDb': return this.importNotesProcessorService.processMastoToDb(job);
+ case 'importPleroToDb': return this.importNotesProcessorService.processPleroToDb(job);
+ case 'importKeyNotesToDb': return this.importNotesProcessorService.processKeyNotesToDb(job);
+ case 'deleteAccount': return this.deleteAccountProcessorService.process(job);
+ default: throw new Error(`unrecognized job type ${job.name} for db`);
+ }
+ };
+
+ this.dbQueueWorker = new Bull.Worker(QUEUE.DB, (job) => {
+ if (this.config.sentryForBackend) {
+ return Sentry.startSpan({ name: 'Queue: DB: ' + job.name }, () => processer(job));
+ } else {
+ return processer(job);
+ }
+ }, {
+ ...baseQueueOptions(this.config, QUEUE.DB),
+ autorun: false,
+ });
- const dbLogger = this.logger.createSubLogger('db');
+ const logger = this.logger.createSubLogger('db');
- this.dbQueueWorker
- .on('active', (job) => dbLogger.debug(`active id=${job.id}`))
- .on('completed', (job, result) => dbLogger.debug(`completed(${result}) id=${job.id}`))
- .on('failed', (job, err) => dbLogger.warn(`failed(${err.stack}) id=${job ? job.id : '-'}`, { job, e: renderError(err) }))
- .on('error', (err: Error) => dbLogger.error(`error ${err.stack}`, { e: renderError(err) }))
- .on('stalled', (jobId) => dbLogger.warn(`stalled id=${jobId}`));
+ this.dbQueueWorker
+ .on('active', (job) => logger.debug(`active id=${job.id}`))
+ .on('completed', (job, result) => logger.debug(`completed(${result}) id=${job.id}`))
+ .on('failed', (job, err) => {
+ logger.error(`failed(${err.stack}) id=${job ? job.id : '-'}`, { job, e: renderError(err) });
+ if (config.sentryForBackend) {
+ Sentry.captureMessage(`Queue: DB: ${job?.name ?? '?'}: ${err.message}`, {
+ level: 'error',
+ extra: { job, err },
+ });
+ }
+ })
+ .on('error', (err: Error) => logger.error(`error ${err.stack}`, { e: renderError(err) }))
+ .on('stalled', (jobId) => logger.warn(`stalled id=${jobId}`));
+ }
//#endregion
//#region deliver
- this.deliverQueueWorker = new Bull.Worker(QUEUE.DELIVER, (job) => this.deliverProcessorService.process(job), {
- ...baseQueueOptions(this.config, QUEUE.DELIVER),
- autorun: false,
- concurrency: this.config.deliverJobConcurrency ?? 128,
- limiter: {
- max: this.config.deliverJobPerSec ?? 128,
- duration: 1000,
- },
- settings: {
- backoffStrategy: httpRelatedBackoff,
- },
- });
+ {
+ this.deliverQueueWorker = new Bull.Worker(QUEUE.DELIVER, (job) => {
+ if (this.config.sentryForBackend) {
+ return Sentry.startSpan({ name: 'Queue: Deliver' }, () => this.deliverProcessorService.process(job));
+ } else {
+ return this.deliverProcessorService.process(job);
+ }
+ }, {
+ ...baseQueueOptions(this.config, QUEUE.DELIVER),
+ autorun: false,
+ concurrency: this.config.deliverJobConcurrency ?? 128,
+ limiter: {
+ max: this.config.deliverJobPerSec ?? 128,
+ duration: 1000,
+ },
+ settings: {
+ backoffStrategy: httpRelatedBackoff,
+ },
+ });
- const deliverLogger = this.logger.createSubLogger('deliver');
+ const logger = this.logger.createSubLogger('deliver');
- this.deliverQueueWorker
- .on('active', (job) => deliverLogger.debug(`active ${getJobInfo(job, true)} to=${job.data.to}`))
- .on('completed', (job, result) => deliverLogger.debug(`completed(${result}) ${getJobInfo(job, true)} to=${job.data.to}`))
- .on('failed', (job, err) => deliverLogger.warn(`failed(${err.stack}) ${getJobInfo(job)} to=${job ? job.data.to : '-'}`))
- .on('error', (err: Error) => deliverLogger.error(`error ${err.stack}`, { e: renderError(err) }))
- .on('stalled', (jobId) => deliverLogger.warn(`stalled id=${jobId}`));
+ this.deliverQueueWorker
+ .on('active', (job) => logger.debug(`active ${getJobInfo(job, true)} to=${job.data.to}`))
+ .on('completed', (job, result) => logger.debug(`completed(${result}) ${getJobInfo(job, true)} to=${job.data.to}`))
+ .on('failed', (job, err) => {
+ logger.error(`failed(${err.stack}) ${getJobInfo(job)} to=${job ? job.data.to : '-'}`);
+ if (config.sentryForBackend) {
+ Sentry.captureMessage(`Queue: Deliver: ${err.message}`, {
+ level: 'error',
+ extra: { job, err },
+ });
+ }
+ })
+ .on('error', (err: Error) => logger.error(`error ${err.stack}`, { e: renderError(err) }))
+ .on('stalled', (jobId) => logger.warn(`stalled id=${jobId}`));
+ }
//#endregion
//#region inbox
- this.inboxQueueWorker = new Bull.Worker(QUEUE.INBOX, (job) => this.inboxProcessorService.process(job), {
- ...baseQueueOptions(this.config, QUEUE.INBOX),
- autorun: false,
- concurrency: this.config.inboxJobConcurrency ?? 16,
- limiter: {
- max: this.config.inboxJobPerSec ?? 32,
- duration: 1000,
- },
- settings: {
- backoffStrategy: httpRelatedBackoff,
- },
- });
+ {
+ this.inboxQueueWorker = new Bull.Worker(QUEUE.INBOX, (job) => {
+ if (this.config.sentryForBackend) {
+ return Sentry.startSpan({ name: 'Queue: Inbox' }, () => this.inboxProcessorService.process(job));
+ } else {
+ return this.inboxProcessorService.process(job);
+ }
+ }, {
+ ...baseQueueOptions(this.config, QUEUE.INBOX),
+ autorun: false,
+ concurrency: this.config.inboxJobConcurrency ?? 16,
+ limiter: {
+ max: this.config.inboxJobPerSec ?? 32,
+ duration: 1000,
+ },
+ settings: {
+ backoffStrategy: httpRelatedBackoff,
+ },
+ });
- const inboxLogger = this.logger.createSubLogger('inbox');
+ const logger = this.logger.createSubLogger('inbox');
- this.inboxQueueWorker
- .on('active', (job) => inboxLogger.debug(`active ${getJobInfo(job, true)}`))
- .on('completed', (job, result) => inboxLogger.debug(`completed(${result}) ${getJobInfo(job, true)}`))
- .on('failed', (job, err) => inboxLogger.warn(`failed(${err.stack}) ${getJobInfo(job)} activity=${job ? (job.data.activity ? job.data.activity.id : 'none') : '-'}`, { job, e: renderError(err) }))
- .on('error', (err: Error) => inboxLogger.error(`error ${err.stack}`, { e: renderError(err) }))
- .on('stalled', (jobId) => inboxLogger.warn(`stalled id=${jobId}`));
+ this.inboxQueueWorker
+ .on('active', (job) => logger.debug(`active ${getJobInfo(job, true)}`))
+ .on('completed', (job, result) => logger.debug(`completed(${result}) ${getJobInfo(job, true)}`))
+ .on('failed', (job, err) => {
+ logger.error(`failed(${err.stack}) ${getJobInfo(job)} activity=${job ? (job.data.activity ? job.data.activity.id : 'none') : '-'}`, { job, e: renderError(err) });
+ if (config.sentryForBackend) {
+ Sentry.captureMessage(`Queue: Inbox: ${err.message}`, {
+ level: 'error',
+ extra: { job, err },
+ });
+ }
+ })
+ .on('error', (err: Error) => logger.error(`error ${err.stack}`, { e: renderError(err) }))
+ .on('stalled', (jobId) => logger.warn(`stalled id=${jobId}`));
+ }
//#endregion
- //#region webhook deliver
- this.webhookDeliverQueueWorker = new Bull.Worker(QUEUE.WEBHOOK_DELIVER, (job) => this.webhookDeliverProcessorService.process(job), {
- ...baseQueueOptions(this.config, QUEUE.WEBHOOK_DELIVER),
- autorun: false,
- concurrency: 64,
- limiter: {
- max: 64,
- duration: 1000,
- },
- settings: {
- backoffStrategy: httpRelatedBackoff,
- },
- });
+ //#region user-webhook deliver
+ {
+ this.userWebhookDeliverQueueWorker = new Bull.Worker(QUEUE.USER_WEBHOOK_DELIVER, (job) => {
+ if (this.config.sentryForBackend) {
+ return Sentry.startSpan({ name: 'Queue: UserWebhookDeliver' }, () => this.userWebhookDeliverProcessorService.process(job));
+ } else {
+ return this.userWebhookDeliverProcessorService.process(job);
+ }
+ }, {
+ ...baseQueueOptions(this.config, QUEUE.USER_WEBHOOK_DELIVER),
+ autorun: false,
+ concurrency: 64,
+ limiter: {
+ max: 64,
+ duration: 1000,
+ },
+ settings: {
+ backoffStrategy: httpRelatedBackoff,
+ },
+ });
- const webhookLogger = this.logger.createSubLogger('webhook');
+ const logger = this.logger.createSubLogger('user-webhook');
- this.webhookDeliverQueueWorker
- .on('active', (job) => webhookLogger.debug(`active ${getJobInfo(job, true)} to=${job.data.to}`))
- .on('completed', (job, result) => webhookLogger.debug(`completed(${result}) ${getJobInfo(job, true)} to=${job.data.to}`))
- .on('failed', (job, err) => webhookLogger.warn(`failed(${err.stack}) ${getJobInfo(job)} to=${job ? job.data.to : '-'}`))
- .on('error', (err: Error) => webhookLogger.error(`error ${err.stack}`, { e: renderError(err) }))
- .on('stalled', (jobId) => webhookLogger.warn(`stalled id=${jobId}`));
+ this.userWebhookDeliverQueueWorker
+ .on('active', (job) => logger.debug(`active ${getJobInfo(job, true)} to=${job.data.to}`))
+ .on('completed', (job, result) => logger.debug(`completed(${result}) ${getJobInfo(job, true)} to=${job.data.to}`))
+ .on('failed', (job, err) => {
+ logger.error(`failed(${err.stack}) ${getJobInfo(job)} to=${job ? job.data.to : '-'}`);
+ if (config.sentryForBackend) {
+ Sentry.captureMessage(`Queue: UserWebhookDeliver: ${err.message}`, {
+ level: 'error',
+ extra: { job, err },
+ });
+ }
+ })
+ .on('error', (err: Error) => logger.error(`error ${err.stack}`, { e: renderError(err) }))
+ .on('stalled', (jobId) => logger.warn(`stalled id=${jobId}`));
+ }
+ //#endregion
+
+ //#region system-webhook deliver
+ {
+ this.systemWebhookDeliverQueueWorker = new Bull.Worker(QUEUE.SYSTEM_WEBHOOK_DELIVER, (job) => {
+ if (this.config.sentryForBackend) {
+ return Sentry.startSpan({ name: 'Queue: SystemWebhookDeliver' }, () => this.systemWebhookDeliverProcessorService.process(job));
+ } else {
+ return this.systemWebhookDeliverProcessorService.process(job);
+ }
+ }, {
+ ...baseQueueOptions(this.config, QUEUE.SYSTEM_WEBHOOK_DELIVER),
+ autorun: false,
+ concurrency: 16,
+ limiter: {
+ max: 16,
+ duration: 1000,
+ },
+ settings: {
+ backoffStrategy: httpRelatedBackoff,
+ },
+ });
+
+ const logger = this.logger.createSubLogger('system-webhook');
+
+ this.systemWebhookDeliverQueueWorker
+ .on('active', (job) => logger.debug(`active ${getJobInfo(job, true)} to=${job.data.to}`))
+ .on('completed', (job, result) => logger.debug(`completed(${result}) ${getJobInfo(job, true)} to=${job.data.to}`))
+ .on('failed', (job, err) => {
+ logger.error(`failed(${err.stack}) ${getJobInfo(job)} to=${job ? job.data.to : '-'}`);
+ if (config.sentryForBackend) {
+ Sentry.captureMessage(`Queue: SystemWebhookDeliver: ${err.message}`, {
+ level: 'error',
+ extra: { job, err },
+ });
+ }
+ })
+ .on('error', (err: Error) => logger.error(`error ${err.stack}`, { e: renderError(err) }))
+ .on('stalled', (jobId) => logger.warn(`stalled id=${jobId}`));
+ }
//#endregion
//#region relationship
- this.relationshipQueueWorker = new Bull.Worker(QUEUE.RELATIONSHIP, (job) => {
- switch (job.name) {
- case 'follow': return this.relationshipProcessorService.processFollow(job);
- case 'unfollow': return this.relationshipProcessorService.processUnfollow(job);
- case 'block': return this.relationshipProcessorService.processBlock(job);
- case 'unblock': return this.relationshipProcessorService.processUnblock(job);
- default: throw new Error(`unrecognized job type ${job.name} for relationship`);
- }
- }, {
- ...baseQueueOptions(this.config, QUEUE.RELATIONSHIP),
- autorun: false,
- concurrency: this.config.relationshipJobConcurrency ?? 16,
- limiter: {
- max: this.config.relationshipJobPerSec ?? 64,
- duration: 1000,
- },
- });
+ {
+ const processer = (job: Bull.Job) => {
+ switch (job.name) {
+ case 'follow': return this.relationshipProcessorService.processFollow(job);
+ case 'unfollow': return this.relationshipProcessorService.processUnfollow(job);
+ case 'block': return this.relationshipProcessorService.processBlock(job);
+ case 'unblock': return this.relationshipProcessorService.processUnblock(job);
+ default: throw new Error(`unrecognized job type ${job.name} for relationship`);
+ }
+ };
+
+ this.relationshipQueueWorker = new Bull.Worker(QUEUE.RELATIONSHIP, (job) => {
+ if (this.config.sentryForBackend) {
+ return Sentry.startSpan({ name: 'Queue: Relationship: ' + job.name }, () => processer(job));
+ } else {
+ return processer(job);
+ }
+ }, {
+ ...baseQueueOptions(this.config, QUEUE.RELATIONSHIP),
+ autorun: false,
+ concurrency: this.config.relationshipJobConcurrency ?? 16,
+ limiter: {
+ max: this.config.relationshipJobPerSec ?? 64,
+ duration: 1000,
+ },
+ });
- const relationshipLogger = this.logger.createSubLogger('relationship');
+ const logger = this.logger.createSubLogger('relationship');
- this.relationshipQueueWorker
- .on('active', (job) => relationshipLogger.debug(`active id=${job.id}`))
- .on('completed', (job, result) => relationshipLogger.debug(`completed(${result}) id=${job.id}`))
- .on('failed', (job, err) => relationshipLogger.warn(`failed(${err.stack}) id=${job ? job.id : '-'}`, { job, e: renderError(err) }))
- .on('error', (err: Error) => relationshipLogger.error(`error ${err.stack}`, { e: renderError(err) }))
- .on('stalled', (jobId) => relationshipLogger.warn(`stalled id=${jobId}`));
+ this.relationshipQueueWorker
+ .on('active', (job) => logger.debug(`active id=${job.id}`))
+ .on('completed', (job, result) => logger.debug(`completed(${result}) id=${job.id}`))
+ .on('failed', (job, err) => {
+ logger.error(`failed(${err.stack}) id=${job ? job.id : '-'}`, { job, e: renderError(err) });
+ if (config.sentryForBackend) {
+ Sentry.captureMessage(`Queue: Relationship: ${job?.name ?? '?'}: ${err.message}`, {
+ level: 'error',
+ extra: { job, err },
+ });
+ }
+ })
+ .on('error', (err: Error) => logger.error(`error ${err.stack}`, { e: renderError(err) }))
+ .on('stalled', (jobId) => logger.warn(`stalled id=${jobId}`));
+ }
//#endregion
//#region object storage
- this.objectStorageQueueWorker = new Bull.Worker(QUEUE.OBJECT_STORAGE, (job) => {
- switch (job.name) {
- case 'deleteFile': return this.deleteFileProcessorService.process(job);
- case 'cleanRemoteFiles': return this.cleanRemoteFilesProcessorService.process(job);
- default: throw new Error(`unrecognized job type ${job.name} for objectStorage`);
- }
- }, {
- ...baseQueueOptions(this.config, QUEUE.OBJECT_STORAGE),
- autorun: false,
- concurrency: 16,
- });
+ {
+ const processer = (job: Bull.Job) => {
+ switch (job.name) {
+ case 'deleteFile': return this.deleteFileProcessorService.process(job);
+ case 'cleanRemoteFiles': return this.cleanRemoteFilesProcessorService.process(job);
+ default: throw new Error(`unrecognized job type ${job.name} for objectStorage`);
+ }
+ };
+
+ this.objectStorageQueueWorker = new Bull.Worker(QUEUE.OBJECT_STORAGE, (job) => {
+ if (this.config.sentryForBackend) {
+ return Sentry.startSpan({ name: 'Queue: ObjectStorage: ' + job.name }, () => processer(job));
+ } else {
+ return processer(job);
+ }
+ }, {
+ ...baseQueueOptions(this.config, QUEUE.OBJECT_STORAGE),
+ autorun: false,
+ concurrency: 16,
+ });
- const objectStorageLogger = this.logger.createSubLogger('objectStorage');
+ const logger = this.logger.createSubLogger('objectStorage');
- this.objectStorageQueueWorker
- .on('active', (job) => objectStorageLogger.debug(`active id=${job.id}`))
- .on('completed', (job, result) => objectStorageLogger.debug(`completed(${result}) id=${job.id}`))
- .on('failed', (job, err) => objectStorageLogger.warn(`failed(${err.stack}) id=${job ? job.id : '-'}`, { job, e: renderError(err) }))
- .on('error', (err: Error) => objectStorageLogger.error(`error ${err.stack}`, { e: renderError(err) }))
- .on('stalled', (jobId) => objectStorageLogger.warn(`stalled id=${jobId}`));
+ this.objectStorageQueueWorker
+ .on('active', (job) => logger.debug(`active id=${job.id}`))
+ .on('completed', (job, result) => logger.debug(`completed(${result}) id=${job.id}`))
+ .on('failed', (job, err) => {
+ logger.error(`failed(${err.stack}) id=${job ? job.id : '-'}`, { job, e: renderError(err) });
+ if (config.sentryForBackend) {
+ Sentry.captureMessage(`Queue: ObjectStorage: ${job?.name ?? '?'}: ${err.message}`, {
+ level: 'error',
+ extra: { job, err },
+ });
+ }
+ })
+ .on('error', (err: Error) => logger.error(`error ${err.stack}`, { e: renderError(err) }))
+ .on('stalled', (jobId) => logger.warn(`stalled id=${jobId}`));
+ }
//#endregion
//#region ended poll notification
- this.endedPollNotificationQueueWorker = new Bull.Worker(QUEUE.ENDED_POLL_NOTIFICATION, (job) => this.endedPollNotificationProcessorService.process(job), {
- ...baseQueueOptions(this.config, QUEUE.ENDED_POLL_NOTIFICATION),
- autorun: false,
- });
+ {
+ this.endedPollNotificationQueueWorker = new Bull.Worker(QUEUE.ENDED_POLL_NOTIFICATION, (job) => {
+ if (this.config.sentryForBackend) {
+ return Sentry.startSpan({ name: 'Queue: EndedPollNotification' }, () => this.endedPollNotificationProcessorService.process(job));
+ } else {
+ return this.endedPollNotificationProcessorService.process(job);
+ }
+ }, {
+ ...baseQueueOptions(this.config, QUEUE.ENDED_POLL_NOTIFICATION),
+ autorun: false,
+ });
+ }
//#endregion
}
@@ -350,7 +522,8 @@ export class QueueProcessorService implements OnApplicationShutdown {
this.dbQueueWorker.run(),
this.deliverQueueWorker.run(),
this.inboxQueueWorker.run(),
- this.webhookDeliverQueueWorker.run(),
+ this.userWebhookDeliverQueueWorker.run(),
+ this.systemWebhookDeliverQueueWorker.run(),
this.relationshipQueueWorker.run(),
this.objectStorageQueueWorker.run(),
this.endedPollNotificationQueueWorker.run(),
@@ -364,7 +537,8 @@ export class QueueProcessorService implements OnApplicationShutdown {
this.dbQueueWorker.close(),
this.deliverQueueWorker.close(),
this.inboxQueueWorker.close(),
- this.webhookDeliverQueueWorker.close(),
+ this.userWebhookDeliverQueueWorker.close(),
+ this.systemWebhookDeliverQueueWorker.close(),
this.relationshipQueueWorker.close(),
this.objectStorageQueueWorker.close(),
this.endedPollNotificationQueueWorker.close(),
diff --git a/packages/backend/src/queue/const.ts b/packages/backend/src/queue/const.ts
index 132e916612..67f689b618 100644
--- a/packages/backend/src/queue/const.ts
+++ b/packages/backend/src/queue/const.ts
@@ -14,7 +14,8 @@ export const QUEUE = {
DB: 'db',
RELATIONSHIP: 'relationship',
OBJECT_STORAGE: 'objectStorage',
- WEBHOOK_DELIVER: 'webhookDeliver',
+ USER_WEBHOOK_DELIVER: 'userWebhookDeliver',
+ SYSTEM_WEBHOOK_DELIVER: 'systemWebhookDeliver',
};
export function baseQueueOptions(config: Config, queueName: typeof QUEUE[keyof typeof QUEUE]): Bull.QueueOptions {
diff --git a/packages/backend/src/queue/processors/EndedPollNotificationProcessorService.ts b/packages/backend/src/queue/processors/EndedPollNotificationProcessorService.ts
index 29c1f27bb1..34180e5f2b 100644
--- a/packages/backend/src/queue/processors/EndedPollNotificationProcessorService.ts
+++ b/packages/backend/src/queue/processors/EndedPollNotificationProcessorService.ts
@@ -7,6 +7,7 @@ import { Inject, Injectable } from '@nestjs/common';
import { DI } from '@/di-symbols.js';
import type { PollVotesRepository, NotesRepository } from '@/models/_.js';
import type Logger from '@/logger.js';
+import { CacheService } from '@/core/CacheService.js';
import { NotificationService } from '@/core/NotificationService.js';
import { bindThis } from '@/decorators.js';
import { QueueLoggerService } from '../QueueLoggerService.js';
@@ -24,6 +25,7 @@ export class EndedPollNotificationProcessorService {
@Inject(DI.pollVotesRepository)
private pollVotesRepository: PollVotesRepository,
+ private cacheService: CacheService,
private notificationService: NotificationService,
private queueLoggerService: QueueLoggerService,
) {
@@ -47,9 +49,12 @@ export class EndedPollNotificationProcessorService {
const userIds = [...new Set([note.userId, ...votes.map(v => v.userId)])];
for (const userId of userIds) {
- this.notificationService.createNotification(userId, 'pollEnded', {
- noteId: note.id,
- });
+ const profile = await this.cacheService.userProfileCache.fetch(userId);
+ if (profile.userHost === null) {
+ this.notificationService.createNotification(userId, 'pollEnded', {
+ noteId: note.id,
+ });
+ }
}
}
}
diff --git a/packages/backend/src/queue/processors/ImportAntennasProcessorService.ts b/packages/backend/src/queue/processors/ImportAntennasProcessorService.ts
index e5b7c5ac52..9c033b73e2 100644
--- a/packages/backend/src/queue/processors/ImportAntennasProcessorService.ts
+++ b/packages/backend/src/queue/processors/ImportAntennasProcessorService.ts
@@ -76,7 +76,7 @@ export class ImportAntennasProcessorService {
this.logger.warn('Validation Failed');
continue;
}
- const result = await this.antennasRepository.insert({
+ const result = await this.antennasRepository.insertOne({
id: this.idService.gen(now.getTime()),
lastUsedAt: now,
userId: job.data.user.id,
@@ -91,7 +91,7 @@ export class ImportAntennasProcessorService {
excludeBots: antenna.excludeBots,
withReplies: antenna.withReplies,
withFile: antenna.withFile,
- }).then(x => this.antennasRepository.findOneByOrFail(x.identifiers[0]));
+ });
this.logger.succ('Antenna created: ' + result.id);
this.globalEventService.publishInternalEvent('antennaCreated', result);
}
diff --git a/packages/backend/src/queue/processors/ImportUserListsProcessorService.ts b/packages/backend/src/queue/processors/ImportUserListsProcessorService.ts
index a5992c28c8..db9255b35d 100644
--- a/packages/backend/src/queue/processors/ImportUserListsProcessorService.ts
+++ b/packages/backend/src/queue/processors/ImportUserListsProcessorService.ts
@@ -79,11 +79,11 @@ export class ImportUserListsProcessorService {
});
if (list == null) {
- list = await this.userListsRepository.insert({
+ list = await this.userListsRepository.insertOne({
id: this.idService.gen(),
userId: user.id,
name: listName,
- }).then(x => this.userListsRepository.findOneByOrFail(x.identifiers[0]));
+ });
}
let target = this.utilityService.isSelfHost(host!) ? await this.usersRepository.findOneBy({
diff --git a/packages/backend/src/queue/processors/SystemWebhookDeliverProcessorService.ts b/packages/backend/src/queue/processors/SystemWebhookDeliverProcessorService.ts
new file mode 100644
index 0000000000..f6bef52684
--- /dev/null
+++ b/packages/backend/src/queue/processors/SystemWebhookDeliverProcessorService.ts
@@ -0,0 +1,87 @@
+/*
+ * SPDX-FileCopyrightText: syuilo and misskey-project
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+import { Inject, Injectable } from '@nestjs/common';
+import * as Bull from 'bullmq';
+import { DI } from '@/di-symbols.js';
+import type { SystemWebhooksRepository } from '@/models/_.js';
+import type { Config } from '@/config.js';
+import type Logger from '@/logger.js';
+import { HttpRequestService } from '@/core/HttpRequestService.js';
+import { StatusError } from '@/misc/status-error.js';
+import { bindThis } from '@/decorators.js';
+import { QueueLoggerService } from '../QueueLoggerService.js';
+import { SystemWebhookDeliverJobData } from '../types.js';
+
+@Injectable()
+export class SystemWebhookDeliverProcessorService {
+ private logger: Logger;
+
+ constructor(
+ @Inject(DI.config)
+ private config: Config,
+
+ @Inject(DI.systemWebhooksRepository)
+ private systemWebhooksRepository: SystemWebhooksRepository,
+
+ private httpRequestService: HttpRequestService,
+ private queueLoggerService: QueueLoggerService,
+ ) {
+ this.logger = this.queueLoggerService.logger.createSubLogger('webhook');
+ }
+
+ @bindThis
+ public async process(job: Bull.Job<SystemWebhookDeliverJobData>): Promise<string> {
+ try {
+ this.logger.debug(`delivering ${job.data.webhookId}`);
+
+ const res = await this.httpRequestService.send(job.data.to, {
+ method: 'POST',
+ headers: {
+ 'User-Agent': 'Misskey-Hooks',
+ 'X-Misskey-Host': this.config.host,
+ 'X-Misskey-Hook-Id': job.data.webhookId,
+ 'X-Misskey-Hook-Secret': job.data.secret,
+ 'Content-Type': 'application/json',
+ },
+ body: JSON.stringify({
+ server: this.config.url,
+ hookId: job.data.webhookId,
+ eventId: job.data.eventId,
+ createdAt: job.data.createdAt,
+ type: job.data.type,
+ body: job.data.content,
+ }),
+ });
+
+ this.systemWebhooksRepository.update({ id: job.data.webhookId }, {
+ latestSentAt: new Date(),
+ latestStatus: res.status,
+ });
+
+ return 'Success';
+ } catch (res) {
+ this.logger.error(res as Error);
+
+ this.systemWebhooksRepository.update({ id: job.data.webhookId }, {
+ latestSentAt: new Date(),
+ latestStatus: res instanceof StatusError ? res.statusCode : 1,
+ });
+
+ if (res instanceof StatusError) {
+ // 4xx
+ if (!res.isRetryable) {
+ throw new Bull.UnrecoverableError(`${res.statusCode} ${res.statusMessage}`);
+ }
+
+ // 5xx etc.
+ throw new Error(`${res.statusCode} ${res.statusMessage}`);
+ } else {
+ // DNS error, socket error, timeout ...
+ throw res;
+ }
+ }
+ }
+}
diff --git a/packages/backend/src/queue/processors/WebhookDeliverProcessorService.ts b/packages/backend/src/queue/processors/UserWebhookDeliverProcessorService.ts
index 8c260c0137..9ec630ef70 100644
--- a/packages/backend/src/queue/processors/WebhookDeliverProcessorService.ts
+++ b/packages/backend/src/queue/processors/UserWebhookDeliverProcessorService.ts
@@ -13,10 +13,10 @@ import { HttpRequestService } from '@/core/HttpRequestService.js';
import { StatusError } from '@/misc/status-error.js';
import { bindThis } from '@/decorators.js';
import { QueueLoggerService } from '../QueueLoggerService.js';
-import type { WebhookDeliverJobData } from '../types.js';
+import { UserWebhookDeliverJobData } from '../types.js';
@Injectable()
-export class WebhookDeliverProcessorService {
+export class UserWebhookDeliverProcessorService {
private logger: Logger;
constructor(
@@ -33,7 +33,7 @@ export class WebhookDeliverProcessorService {
}
@bindThis
- public async process(job: Bull.Job<WebhookDeliverJobData>): Promise<string> {
+ public async process(job: Bull.Job<UserWebhookDeliverJobData>): Promise<string> {
try {
this.logger.debug(`delivering ${job.data.webhookId}`);
diff --git a/packages/backend/src/queue/types.ts b/packages/backend/src/queue/types.ts
index 91718898b2..c0d246ebbc 100644
--- a/packages/backend/src/queue/types.ts
+++ b/packages/backend/src/queue/types.ts
@@ -131,7 +131,17 @@ export type EndedPollNotificationJobData = {
noteId: MiNote['id'];
};
-export type WebhookDeliverJobData = {
+export type SystemWebhookDeliverJobData = {
+ type: string;
+ content: unknown;
+ webhookId: MiWebhook['id'];
+ to: string;
+ secret: string;
+ createdAt: number;
+ eventId: string;
+};
+
+export type UserWebhookDeliverJobData = {
type: string;
content: unknown;
webhookId: MiWebhook['id'];