|
|
|
@@ -5,7 +5,6 @@
|
|
|
|
|
|
|
|
|
|
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';
|
|
|
|
@@ -157,6 +156,13 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
};
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
let Sentry: typeof import('@sentry/node') | undefined;
|
|
|
|
|
if (Sentry != null) {
|
|
|
|
|
import('@sentry/node').then((mod) => {
|
|
|
|
|
Sentry = mod;
|
|
|
|
|
});
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
//#region system
|
|
|
|
|
{
|
|
|
|
|
const processer = (job: Bull.Job) => {
|
|
|
|
@@ -175,7 +181,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
this.systemQueueWorker = new Bull.Worker(QUEUE.SYSTEM, (job) => {
|
|
|
|
|
if (this.config.sentryForBackend) {
|
|
|
|
|
if (Sentry != null) {
|
|
|
|
|
return Sentry.startSpan({ name: 'Queue: System: ' + job.name }, () => processer(job));
|
|
|
|
|
} else {
|
|
|
|
|
return processer(job);
|
|
|
|
@@ -192,7 +198,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
.on('completed', (job, result) => logger.debug(`completed(${result}) id=${job.id}`))
|
|
|
|
|
.on('failed', (job, err: Error) => {
|
|
|
|
|
logger.error(`failed(${err.name}: ${err.message}) id=${job?.id ?? '?'}`, { job: renderJob(job), e: renderError(err) });
|
|
|
|
|
if (config.sentryForBackend) {
|
|
|
|
|
if (Sentry != null) {
|
|
|
|
|
Sentry.captureMessage(`Queue: System: ${job?.name ?? '?'}: ${err.name}: ${err.message}`, {
|
|
|
|
|
level: 'error',
|
|
|
|
|
extra: { job, err },
|
|
|
|
@@ -232,7 +238,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
this.dbQueueWorker = new Bull.Worker(QUEUE.DB, (job) => {
|
|
|
|
|
if (this.config.sentryForBackend) {
|
|
|
|
|
if (Sentry != null) {
|
|
|
|
|
return Sentry.startSpan({ name: 'Queue: DB: ' + job.name }, () => processer(job));
|
|
|
|
|
} else {
|
|
|
|
|
return processer(job);
|
|
|
|
@@ -249,7 +255,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
.on('completed', (job, result) => logger.debug(`completed(${result}) id=${job.id}`))
|
|
|
|
|
.on('failed', (job, err) => {
|
|
|
|
|
logger.error(`failed(${err.name}: ${err.message}) id=${job?.id ?? '?'}`, { job: renderJob(job), e: renderError(err) });
|
|
|
|
|
if (config.sentryForBackend) {
|
|
|
|
|
if (Sentry != null) {
|
|
|
|
|
Sentry.captureMessage(`Queue: DB: ${job?.name ?? '?'}: ${err.name}: ${err.message}`, {
|
|
|
|
|
level: 'error',
|
|
|
|
|
extra: { job, err },
|
|
|
|
@@ -264,7 +270,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
//#region deliver
|
|
|
|
|
{
|
|
|
|
|
this.deliverQueueWorker = new Bull.Worker(QUEUE.DELIVER, (job) => {
|
|
|
|
|
if (this.config.sentryForBackend) {
|
|
|
|
|
if (Sentry != null) {
|
|
|
|
|
return Sentry.startSpan({ name: 'Queue: Deliver' }, () => this.deliverProcessorService.process(job));
|
|
|
|
|
} else {
|
|
|
|
|
return this.deliverProcessorService.process(job);
|
|
|
|
@@ -289,7 +295,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
.on('completed', (job, result) => logger.debug(`completed(${result}) ${getJobInfo(job, true)} to=${job.data.to}`))
|
|
|
|
|
.on('failed', (job, err) => {
|
|
|
|
|
logger.error(`failed(${err.name}: ${err.message}) ${getJobInfo(job)} to=${job ? job.data.to : '-'}`);
|
|
|
|
|
if (config.sentryForBackend) {
|
|
|
|
|
if (Sentry != null) {
|
|
|
|
|
Sentry.captureMessage(`Queue: Deliver: ${err.name}: ${err.message}`, {
|
|
|
|
|
level: 'error',
|
|
|
|
|
extra: { job, err },
|
|
|
|
@@ -304,7 +310,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
//#region inbox
|
|
|
|
|
{
|
|
|
|
|
this.inboxQueueWorker = new Bull.Worker(QUEUE.INBOX, (job) => {
|
|
|
|
|
if (this.config.sentryForBackend) {
|
|
|
|
|
if (Sentry != null) {
|
|
|
|
|
return Sentry.startSpan({ name: 'Queue: Inbox' }, () => this.inboxProcessorService.process(job));
|
|
|
|
|
} else {
|
|
|
|
|
return this.inboxProcessorService.process(job);
|
|
|
|
@@ -329,7 +335,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
.on('completed', (job, result) => logger.debug(`completed(${result}) ${getJobInfo(job, true)}`))
|
|
|
|
|
.on('failed', (job, err) => {
|
|
|
|
|
logger.error(`failed(${err.name}: ${err.message}) ${getJobInfo(job)} activity=${job ? (job.data.activity ? job.data.activity.id : 'none') : '-'}`, { job: renderJob(job), e: renderError(err) });
|
|
|
|
|
if (config.sentryForBackend) {
|
|
|
|
|
if (Sentry != null) {
|
|
|
|
|
Sentry.captureMessage(`Queue: Inbox: ${err.name}: ${err.message}`, {
|
|
|
|
|
level: 'error',
|
|
|
|
|
extra: { job, err },
|
|
|
|
@@ -344,7 +350,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
//#region user-webhook deliver
|
|
|
|
|
{
|
|
|
|
|
this.userWebhookDeliverQueueWorker = new Bull.Worker(QUEUE.USER_WEBHOOK_DELIVER, (job) => {
|
|
|
|
|
if (this.config.sentryForBackend) {
|
|
|
|
|
if (Sentry != null) {
|
|
|
|
|
return Sentry.startSpan({ name: 'Queue: UserWebhookDeliver' }, () => this.userWebhookDeliverProcessorService.process(job));
|
|
|
|
|
} else {
|
|
|
|
|
return this.userWebhookDeliverProcessorService.process(job);
|
|
|
|
@@ -369,7 +375,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
.on('completed', (job, result) => logger.debug(`completed(${result}) ${getJobInfo(job, true)} to=${job.data.to}`))
|
|
|
|
|
.on('failed', (job, err) => {
|
|
|
|
|
logger.error(`failed(${err.name}: ${err.message}) ${getJobInfo(job)} to=${job ? job.data.to : '-'}`);
|
|
|
|
|
if (config.sentryForBackend) {
|
|
|
|
|
if (Sentry != null) {
|
|
|
|
|
Sentry.captureMessage(`Queue: UserWebhookDeliver: ${err.name}: ${err.message}`, {
|
|
|
|
|
level: 'error',
|
|
|
|
|
extra: { job, err },
|
|
|
|
@@ -384,7 +390,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
//#region system-webhook deliver
|
|
|
|
|
{
|
|
|
|
|
this.systemWebhookDeliverQueueWorker = new Bull.Worker(QUEUE.SYSTEM_WEBHOOK_DELIVER, (job) => {
|
|
|
|
|
if (this.config.sentryForBackend) {
|
|
|
|
|
if (Sentry != null) {
|
|
|
|
|
return Sentry.startSpan({ name: 'Queue: SystemWebhookDeliver' }, () => this.systemWebhookDeliverProcessorService.process(job));
|
|
|
|
|
} else {
|
|
|
|
|
return this.systemWebhookDeliverProcessorService.process(job);
|
|
|
|
@@ -409,7 +415,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
.on('completed', (job, result) => logger.debug(`completed(${result}) ${getJobInfo(job, true)} to=${job.data.to}`))
|
|
|
|
|
.on('failed', (job, err) => {
|
|
|
|
|
logger.error(`failed(${err.name}: ${err.message}) ${getJobInfo(job)} to=${job ? job.data.to : '-'}`);
|
|
|
|
|
if (config.sentryForBackend) {
|
|
|
|
|
if (Sentry != null) {
|
|
|
|
|
Sentry.captureMessage(`Queue: SystemWebhookDeliver: ${err.name}: ${err.message}`, {
|
|
|
|
|
level: 'error',
|
|
|
|
|
extra: { job, err },
|
|
|
|
@@ -434,7 +440,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
this.relationshipQueueWorker = new Bull.Worker(QUEUE.RELATIONSHIP, (job) => {
|
|
|
|
|
if (this.config.sentryForBackend) {
|
|
|
|
|
if (Sentry != null) {
|
|
|
|
|
return Sentry.startSpan({ name: 'Queue: Relationship: ' + job.name }, () => processer(job));
|
|
|
|
|
} else {
|
|
|
|
|
return processer(job);
|
|
|
|
@@ -456,7 +462,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
.on('completed', (job, result) => logger.debug(`completed(${result}) id=${job.id}`))
|
|
|
|
|
.on('failed', (job, err) => {
|
|
|
|
|
logger.error(`failed(${err.name}: ${err.message}) id=${job?.id ?? '?'}`, { job: renderJob(job), e: renderError(err) });
|
|
|
|
|
if (config.sentryForBackend) {
|
|
|
|
|
if (Sentry != null) {
|
|
|
|
|
Sentry.captureMessage(`Queue: Relationship: ${job?.name ?? '?'}: ${err.name}: ${err.message}`, {
|
|
|
|
|
level: 'error',
|
|
|
|
|
extra: { job, err },
|
|
|
|
@@ -479,7 +485,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
this.objectStorageQueueWorker = new Bull.Worker(QUEUE.OBJECT_STORAGE, (job) => {
|
|
|
|
|
if (this.config.sentryForBackend) {
|
|
|
|
|
if (Sentry != null) {
|
|
|
|
|
return Sentry.startSpan({ name: 'Queue: ObjectStorage: ' + job.name }, () => processer(job));
|
|
|
|
|
} else {
|
|
|
|
|
return processer(job);
|
|
|
|
@@ -497,7 +503,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
.on('completed', (job, result) => logger.debug(`completed(${result}) id=${job.id}`))
|
|
|
|
|
.on('failed', (job, err) => {
|
|
|
|
|
logger.error(`failed(${err.name}: ${err.message}) id=${job?.id ?? '?'}`, { job: renderJob(job), e: renderError(err) });
|
|
|
|
|
if (config.sentryForBackend) {
|
|
|
|
|
if (Sentry != null) {
|
|
|
|
|
Sentry.captureMessage(`Queue: ObjectStorage: ${job?.name ?? '?'}: ${err.name}: ${err.message}`, {
|
|
|
|
|
level: 'error',
|
|
|
|
|
extra: { job, err },
|
|
|
|
@@ -512,7 +518,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
//#region ended poll notification
|
|
|
|
|
{
|
|
|
|
|
this.endedPollNotificationQueueWorker = new Bull.Worker(QUEUE.ENDED_POLL_NOTIFICATION, (job) => {
|
|
|
|
|
if (this.config.sentryForBackend) {
|
|
|
|
|
if (Sentry != null) {
|
|
|
|
|
return Sentry.startSpan({ name: 'Queue: EndedPollNotification' }, () => this.endedPollNotificationProcessorService.process(job));
|
|
|
|
|
} else {
|
|
|
|
|
return this.endedPollNotificationProcessorService.process(job);
|
|
|
|
@@ -527,7 +533,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
|
|
|
|
|
//#region post scheduled note
|
|
|
|
|
{
|
|
|
|
|
this.postScheduledNoteQueueWorker = new Bull.Worker(QUEUE.POST_SCHEDULED_NOTE, async (job) => {
|
|
|
|
|
if (this.config.sentryForBackend) {
|
|
|
|
|
if (Sentry != null) {
|
|
|
|
|
return Sentry.startSpan({ name: 'Queue: PostScheduledNote' }, () => this.postScheduledNoteProcessorService.process(job));
|
|
|
|
|
} else {
|
|
|
|
|
return this.postScheduledNoteProcessorService.process(job);
|
|
|
|
|