keith/barkey
1
0
Fork 0
mirror of https://codeberg.org/yeentown/barkey synced 2024-11-22 17:15:13 +00:00
Code Issues Projects Releases Packages Wiki Activity

enhance(backend): sentry integration for job queues

Browse source
This commit is contained in:
syuilo 2024-06-06 09:32:04 +09:00
parent 43cccaaee9
commit d4a8c63264
1 changed files with 231 additions and 158 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

89
packages/backend/src/queue/QueueProcessorService.ts
View file

@ -5,6 +5,7 @@
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';
@ -135,7 +136,8 @@ export class QueueProcessorService implements OnApplicationShutdown {
}
//#region system
this.systemQueueWorker = new Bull.Worker(QUEUE.SYSTEM, (job) => {
{
const processer = (job: Bull.Job) => {
switch (job.name) {
case 'tickCharts': return this.tickChartsProcessorService.process();
case 'resyncCharts': return this.resyncChartsProcessorService.process();
@ -145,6 +147,14 @@ export class QueueProcessorService implements OnApplicationShutdown {
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,
@ -158,10 +168,12 @@ export class QueueProcessorService implements OnApplicationShutdown {
.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}`));
}
//#endregion
//#region db
this.dbQueueWorker = new Bull.Worker(QUEUE.DB, (job) => {
{
const processer = (job: Bull.Job) => {
switch (job.name) {
case 'deleteDriveFiles': return this.deleteDriveFilesProcessorService.process(job);
case 'exportCustomEmojis': return this.exportCustomEmojisProcessorService.process(job);
@ -184,6 +196,14 @@ export class QueueProcessorService implements OnApplicationShutdown {
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,
@ -197,10 +217,18 @@ export class QueueProcessorService implements OnApplicationShutdown {
.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}`));
}
//#endregion
//#region deliver
this.deliverQueueWorker = new Bull.Worker(QUEUE.DELIVER, (job) => this.deliverProcessorService.process(job), {
{
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,
@ -221,10 +249,18 @@ export class QueueProcessorService implements OnApplicationShutdown {
.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}`));
}
//#endregion
//#region inbox
this.inboxQueueWorker = new Bull.Worker(QUEUE.INBOX, (job) => this.inboxProcessorService.process(job), {
{
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,
@ -245,10 +281,18 @@ export class QueueProcessorService implements OnApplicationShutdown {
.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}`));
}
//#endregion
//#region webhook deliver
this.webhookDeliverQueueWorker = new Bull.Worker(QUEUE.WEBHOOK_DELIVER, (job) => this.webhookDeliverProcessorService.process(job), {
{
this.webhookDeliverQueueWorker = new Bull.Worker(QUEUE.WEBHOOK_DELIVER, (job) => {
if (this.config.sentryForBackend) {
return Sentry.startSpan({ name: 'Queue: WebhookDeliver' }, () => this.webhookDeliverProcessorService.process(job));
} else {
return this.webhookDeliverProcessorService.process(job);
}
}, {
...baseQueueOptions(this.config, QUEUE.WEBHOOK_DELIVER),
autorun: false,
concurrency: 64,
@ -269,10 +313,12 @@ export class QueueProcessorService implements OnApplicationShutdown {
.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}`));
}
//#endregion
//#region relationship
this.relationshipQueueWorker = new Bull.Worker(QUEUE.RELATIONSHIP, (job) => {
{
const processer = (job: Bull.Job) => {
switch (job.name) {
case 'follow': return this.relationshipProcessorService.processFollow(job);
case 'unfollow': return this.relationshipProcessorService.processUnfollow(job);
@ -280,6 +326,14 @@ export class QueueProcessorService implements OnApplicationShutdown {
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,
@ -298,15 +352,25 @@ export class QueueProcessorService implements OnApplicationShutdown {
.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}`));
}
//#endregion
//#region object storage
this.objectStorageQueueWorker = new Bull.Worker(QUEUE.OBJECT_STORAGE, (job) => {
{
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,
@ -321,13 +385,22 @@ export class QueueProcessorService implements OnApplicationShutdown {
.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}`));
}
//#endregion
//#region ended poll notification
this.endedPollNotificationQueueWorker = new Bull.Worker(QUEUE.ENDED_POLL_NOTIFICATION, (job) => this.endedPollNotificationProcessorService.process(job), {
{
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
}