Mizah/mizzkey
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge branch 'develop' into ed25519

Browse source
This commit is contained in:
tamaina 2024-06-11 14:32:54 +09:00
commit bdaef5f8e1
316 changed files with 12361 additions and 1693 deletions
Download patch file Download diff file Expand all files Collapse all files

6
packages/backend/src/queue/QueueProcessorModule.ts
View file

@ -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';
@ -71,7 +72,8 @@ import { RelationshipProcessorService } from './processors/RelationshipProcessor
DeleteFileProcessorService,
CleanRemoteFilesProcessorService,
RelationshipProcessorService,
WebhookDeliverProcessorService,
UserWebhookDeliverProcessorService,
SystemWebhookDeliverProcessorService,
EndedPollNotificationProcessorService,
DeliverProcessorService,
InboxProcessorService,

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

@ -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';
@ -75,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;
@ -85,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,
@ -135,199 +139,367 @@ 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`);
}
};
const systemLogger = this.logger.createSubLogger('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,
});
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}`));
const logger = this.logger.createSubLogger('system');
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 '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 '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 '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 'deleteAccount': return this.deleteAccountProcessorService.process(job);
default: throw new Error(`unrecognized job type ${job.name} for db`);
}
};
const dbLogger = this.logger.createSubLogger('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,
});
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}`));
const logger = this.logger.createSubLogger('db');
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`);
}
};
const relationshipLogger = this.logger.createSubLogger('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,
},
});
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}`));
const logger = this.logger.createSubLogger('relationship');
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`);
}
};
const objectStorageLogger = this.logger.createSubLogger('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,
});
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}`));
const logger = this.logger.createSubLogger('objectStorage');
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
}
@ -338,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(),
@ -352,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(),

3
packages/backend/src/queue/const.ts
View file

@ -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 {

14
packages/backend/src/queue/processors/DeliverProcessorService.ts
View file

@ -5,6 +5,7 @@
import { Inject, Injectable } from '@nestjs/common';
import * as Bull from 'bullmq';
import { Not } from 'typeorm';
import { DI } from '@/di-symbols.js';
import type { InstancesRepository } from '@/models/_.js';
import type Logger from '@/logger.js';
@ -62,7 +63,7 @@ export class DeliverProcessorService {
if (suspendedHosts == null) {
suspendedHosts = await this.instancesRepository.find({
where: {
isSuspended: true,
suspensionState: Not('none'),
},
});
this.suspendedHostsCache.set(suspendedHosts);
@ -89,6 +90,7 @@ export class DeliverProcessorService {
if (server.isNotResponding) {
this.federatedInstanceService.update(server.id, {
isNotResponding: false,
notRespondingSince: null,
});
}
@ -106,7 +108,15 @@ export class DeliverProcessorService {
if (!i.isNotResponding) {
this.federatedInstanceService.update(i.id, {
isNotResponding: true,
notRespondingSince: new Date(),
});
} else if (i.notRespondingSince) {
// 1週間以上不通ならサスペンド
if (i.suspensionState === 'none' && i.notRespondingSince.getTime() <= Date.now() - 1000 * 60 * 60 * 24 * 7) {
this.federatedInstanceService.update(i.id, {
suspensionState: 'autoSuspendedForNotResponding',
});
}
}
this.apRequestChart.deliverFail();
@ -124,7 +134,7 @@ export class DeliverProcessorService {
if (job.data.isSharedInbox && res.statusCode === 410) {
this.federatedInstanceService.fetch(host).then(i => {
this.federatedInstanceService.update(i.id, {
isSuspended: true,
suspensionState: 'goneSuspended',
});
});
throw new Bull.UnrecoverableError(`${host} is gone`);

1
packages/backend/src/queue/processors/ExportAntennasProcessorService.ts
View file

@ -84,7 +84,6 @@ export class ExportAntennasProcessorService {
excludeBots: antenna.excludeBots,
withReplies: antenna.withReplies,
withFile: antenna.withFile,
notify: antenna.notify,
}));
if (antennas.length - 1 !== index) {
write(', ');

8
packages/backend/src/queue/processors/ImportAntennasProcessorService.ts
View file

@ -47,9 +47,8 @@ const validate = new Ajv().compile({
excludeBots: { type: 'boolean' },
withReplies: { type: 'boolean' },
withFile: { type: 'boolean' },
notify: { type: 'boolean' },
},
required: ['name', 'src', 'keywords', 'excludeKeywords', 'users', 'caseSensitive', 'withReplies', 'withFile', 'notify'],
required: ['name', 'src', 'keywords', 'excludeKeywords', 'users', 'caseSensitive', 'withReplies', 'withFile'],
});
@Injectable()
@ -77,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,
@ -92,8 +91,7 @@ export class ImportAntennasProcessorService {
excludeBots: antenna.excludeBots,
withReplies: antenna.withReplies,
withFile: antenna.withFile,
notify: antenna.notify,
}).then(x => this.antennasRepository.findOneByOrFail(x.identifiers[0]));
});
this.logger.succ('Antenna created: ' + result.id);
this.globalEventService.publishInternalEvent('antennaCreated', result);
}

4
packages/backend/src/queue/processors/ImportUserListsProcessorService.ts
View file

@ -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({

15
packages/backend/src/queue/processors/InboxProcessorService.ts
View file

@ -206,6 +206,8 @@ export class InboxProcessorService {
this.federatedInstanceService.update(i.id, {
latestRequestReceivedAt: new Date(),
isNotResponding: false,
// もしサーバーが死んでるために配信が止まっていた場合には自動的に復活させてあげる
suspensionState: i.suspensionState === 'autoSuspendedForNotResponding' ? 'none' : undefined,
});
this.fetchInstanceMetadataService.fetchInstanceMetadata(i);
@ -220,13 +222,22 @@ export class InboxProcessorService {
// アクティビティを処理
try {
await this.apInboxService.performActivity(authUser.user, activity);
const result = await this.apInboxService.performActivity(authUser.user, activity);
if (result && !result.startsWith('ok')) {
this.logger.warn(`inbox activity ignored (maybe): id=${activity.id} reason=${result}`);
return result;
}
} catch (e) {
if (e instanceof IdentifiableError) {
if (e.id === '689ee33f-f97c-479a-ac49-1b9f8140af99') {
return 'blocked notes with prohibited words';
}
if (e.id === '85ab9bd7-3a41-4530-959d-f07073900109') return 'actor has been suspended';
if (e.id === '85ab9bd7-3a41-4530-959d-f07073900109') {
return 'actor has been suspended';
}
if (e.id === 'd450b8a9-48e4-4dab-ae36-f4db763fda7c') { // invalid Note
return e.message;
}
}
throw e;
}

87
packages/backend/src/queue/processors/SystemWebhookDeliverProcessorService.ts Normal file
View file

@ -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;
}
}
}
}

6
packages/backend/src/queue/processors/WebhookDeliverProcessorService.ts → packages/backend/src/queue/processors/UserWebhookDeliverProcessorService.ts
View file

@ -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}`);

12
packages/backend/src/queue/types.ts
View file

@ -124,7 +124,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'];