|
|
|
@@ -10,7 +10,8 @@ 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';
|
|
|
|
@@ -76,7 +77,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;
|
|
|
|
@@ -86,7 +88,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,
|
|
|
|
@@ -160,13 +163,13 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
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('active', (job) => logger.debug(`active id=${job.id}`))
|
|
|
|
|
.on('completed', (job, result) => logger.debug(`completed(${result}) id=${job.id}`))
|
|
|
|
|
.on('failed', (job, err: Error) => {
|
|
|
|
|
systemLogger.error(`failed(${err.stack}) id=${job ? job.id : '-'}`, { job, e: renderError(err) });
|
|
|
|
|
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',
|
|
|
|
@@ -174,8 +177,8 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
})
|
|
|
|
|
.on('error', (err: Error) => systemLogger.error(`error ${err.stack}`, { e: renderError(err) }))
|
|
|
|
|
.on('stalled', (jobId) => systemLogger.warn(`stalled id=${jobId}`));
|
|
|
|
|
.on('error', (err: Error) => logger.error(`error ${err.stack}`, { e: renderError(err) }))
|
|
|
|
|
.on('stalled', (jobId) => logger.warn(`stalled id=${jobId}`));
|
|
|
|
|
}
|
|
|
|
|
//#endregion
|
|
|
|
|
|
|
|
|
@@ -217,13 +220,13 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
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('active', (job) => logger.debug(`active id=${job.id}`))
|
|
|
|
|
.on('completed', (job, result) => logger.debug(`completed(${result}) id=${job.id}`))
|
|
|
|
|
.on('failed', (job, err) => {
|
|
|
|
|
dbLogger.error(`failed(${err.stack}) id=${job ? job.id : '-'}`, { job, e: renderError(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',
|
|
|
|
@@ -231,8 +234,8 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
})
|
|
|
|
|
.on('error', (err: Error) => dbLogger.error(`error ${err.stack}`, { e: renderError(err) }))
|
|
|
|
|
.on('stalled', (jobId) => dbLogger.warn(`stalled id=${jobId}`));
|
|
|
|
|
.on('error', (err: Error) => logger.error(`error ${err.stack}`, { e: renderError(err) }))
|
|
|
|
|
.on('stalled', (jobId) => logger.warn(`stalled id=${jobId}`));
|
|
|
|
|
}
|
|
|
|
|
//#endregion
|
|
|
|
|
|
|
|
|
@@ -257,13 +260,13 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
},
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
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('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) => {
|
|
|
|
|
deliverLogger.error(`failed(${err.stack}) ${getJobInfo(job)} to=${job ? job.data.to : '-'}`);
|
|
|
|
|
logger.error(`failed(${err.stack}) ${getJobInfo(job)} to=${job ? job.data.to : '-'}`);
|
|
|
|
|
if (config.sentryForBackend) {
|
|
|
|
|
Sentry.captureMessage(`Queue: Deliver: ${err.message}`, {
|
|
|
|
|
level: 'error',
|
|
|
|
@@ -271,8 +274,8 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
})
|
|
|
|
|
.on('error', (err: Error) => deliverLogger.error(`error ${err.stack}`, { e: renderError(err) }))
|
|
|
|
|
.on('stalled', (jobId) => deliverLogger.warn(`stalled id=${jobId}`));
|
|
|
|
|
.on('error', (err: Error) => logger.error(`error ${err.stack}`, { e: renderError(err) }))
|
|
|
|
|
.on('stalled', (jobId) => logger.warn(`stalled id=${jobId}`));
|
|
|
|
|
}
|
|
|
|
|
//#endregion
|
|
|
|
|
|
|
|
|
@@ -297,13 +300,13 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
},
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
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('active', (job) => logger.debug(`active ${getJobInfo(job, true)}`))
|
|
|
|
|
.on('completed', (job, result) => logger.debug(`completed(${result}) ${getJobInfo(job, true)}`))
|
|
|
|
|
.on('failed', (job, err) => {
|
|
|
|
|
inboxLogger.error(`failed(${err.stack}) ${getJobInfo(job)} activity=${job ? (job.data.activity ? job.data.activity.id : 'none') : '-'}`, { job, e: renderError(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',
|
|
|
|
@@ -311,21 +314,21 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
})
|
|
|
|
|
.on('error', (err: Error) => inboxLogger.error(`error ${err.stack}`, { e: renderError(err) }))
|
|
|
|
|
.on('stalled', (jobId) => inboxLogger.warn(`stalled id=${jobId}`));
|
|
|
|
|
.on('error', (err: Error) => logger.error(`error ${err.stack}`, { e: renderError(err) }))
|
|
|
|
|
.on('stalled', (jobId) => logger.warn(`stalled id=${jobId}`));
|
|
|
|
|
}
|
|
|
|
|
//#endregion
|
|
|
|
|
|
|
|
|
|
//#region webhook deliver
|
|
|
|
|
//#region user-webhook deliver
|
|
|
|
|
{
|
|
|
|
|
this.webhookDeliverQueueWorker = new Bull.Worker(QUEUE.WEBHOOK_DELIVER, (job) => {
|
|
|
|
|
this.userWebhookDeliverQueueWorker = new Bull.Worker(QUEUE.USER_WEBHOOK_DELIVER, (job) => {
|
|
|
|
|
if (this.config.sentryForBackend) {
|
|
|
|
|
return Sentry.startSpan({ name: 'Queue: WebhookDeliver' }, () => this.webhookDeliverProcessorService.process(job));
|
|
|
|
|
return Sentry.startSpan({ name: 'Queue: UserWebhookDeliver' }, () => this.userWebhookDeliverProcessorService.process(job));
|
|
|
|
|
} else {
|
|
|
|
|
return this.webhookDeliverProcessorService.process(job);
|
|
|
|
|
return this.userWebhookDeliverProcessorService.process(job);
|
|
|
|
|
}
|
|
|
|
|
}, {
|
|
|
|
|
...baseQueueOptions(this.config, QUEUE.WEBHOOK_DELIVER),
|
|
|
|
|
...baseQueueOptions(this.config, QUEUE.USER_WEBHOOK_DELIVER),
|
|
|
|
|
autorun: false,
|
|
|
|
|
concurrency: 64,
|
|
|
|
|
limiter: {
|
|
|
|
@@ -337,22 +340,62 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
},
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
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}`))
|
|
|
|
|
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) => {
|
|
|
|
|
webhookLogger.error(`failed(${err.stack}) ${getJobInfo(job)} to=${job ? job.data.to : '-'}`);
|
|
|
|
|
logger.error(`failed(${err.stack}) ${getJobInfo(job)} to=${job ? job.data.to : '-'}`);
|
|
|
|
|
if (config.sentryForBackend) {
|
|
|
|
|
Sentry.captureMessage(`Queue: WebhookDeliver: ${err.message}`, {
|
|
|
|
|
Sentry.captureMessage(`Queue: UserWebhookDeliver: ${err.message}`, {
|
|
|
|
|
level: 'error',
|
|
|
|
|
extra: { job, err },
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
})
|
|
|
|
|
.on('error', (err: Error) => webhookLogger.error(`error ${err.stack}`, { e: renderError(err) }))
|
|
|
|
|
.on('stalled', (jobId) => webhookLogger.warn(`stalled id=${jobId}`));
|
|
|
|
|
.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
|
|
|
|
|
|
|
|
|
@@ -384,13 +427,13 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
},
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
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('active', (job) => logger.debug(`active id=${job.id}`))
|
|
|
|
|
.on('completed', (job, result) => logger.debug(`completed(${result}) id=${job.id}`))
|
|
|
|
|
.on('failed', (job, err) => {
|
|
|
|
|
relationshipLogger.error(`failed(${err.stack}) id=${job ? job.id : '-'}`, { job, e: renderError(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',
|
|
|
|
@@ -398,8 +441,8 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
})
|
|
|
|
|
.on('error', (err: Error) => relationshipLogger.error(`error ${err.stack}`, { e: renderError(err) }))
|
|
|
|
|
.on('stalled', (jobId) => relationshipLogger.warn(`stalled id=${jobId}`));
|
|
|
|
|
.on('error', (err: Error) => logger.error(`error ${err.stack}`, { e: renderError(err) }))
|
|
|
|
|
.on('stalled', (jobId) => logger.warn(`stalled id=${jobId}`));
|
|
|
|
|
}
|
|
|
|
|
//#endregion
|
|
|
|
|
|
|
|
|
@@ -425,13 +468,13 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
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('active', (job) => logger.debug(`active id=${job.id}`))
|
|
|
|
|
.on('completed', (job, result) => logger.debug(`completed(${result}) id=${job.id}`))
|
|
|
|
|
.on('failed', (job, err) => {
|
|
|
|
|
objectStorageLogger.error(`failed(${err.stack}) id=${job ? job.id : '-'}`, { job, e: renderError(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',
|
|
|
|
@@ -439,8 +482,8 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
})
|
|
|
|
|
.on('error', (err: Error) => objectStorageLogger.error(`error ${err.stack}`, { e: renderError(err) }))
|
|
|
|
|
.on('stalled', (jobId) => objectStorageLogger.warn(`stalled id=${jobId}`));
|
|
|
|
|
.on('error', (err: Error) => logger.error(`error ${err.stack}`, { e: renderError(err) }))
|
|
|
|
|
.on('stalled', (jobId) => logger.warn(`stalled id=${jobId}`));
|
|
|
|
|
}
|
|
|
|
|
//#endregion
|
|
|
|
|
|
|
|
|
@@ -467,7 +510,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(),
|
|
|
|
@@ -481,7 +525,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(),
|
|
|
|
|