misty.cafe/mistykey
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

merge upstream again

Browse source
This commit is contained in:
Hazelnoot 2025-04-24 14:23:45 -04:00
commit a4dd19fdd4
167 changed files with 6779 additions and 3952 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -47,7 +47,7 @@ import { CleanProcessorService } from './processors/CleanProcessorService.js';
import { AggregateRetentionProcessorService } from './processors/AggregateRetentionProcessorService.js';
import { ScheduleNotePostProcessorService } from './processors/ScheduleNotePostProcessorService.js';
import { QueueLoggerService } from './QueueLoggerService.js';
import { QUEUE, baseQueueOptions } from './const.js';
import { QUEUE, baseWorkerOptions } from './const.js';
import { ImportNotesProcessorService } from './processors/ImportNotesProcessorService.js';
// ref. https://github.com/misskey-dev/misskey/pull/7635#issue-971097019
@ -186,7 +186,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
return processer(job);
}
}, {
...baseQueueOptions(this.config, QUEUE.SYSTEM),
...baseWorkerOptions(this.config, QUEUE.SYSTEM),
autorun: false,
});
@ -251,7 +251,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
return processer(job);
}
}, {
...baseQueueOptions(this.config, QUEUE.DB),
...baseWorkerOptions(this.config, QUEUE.DB),
autorun: false,
});
@ -283,7 +283,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
return this.deliverProcessorService.process(job);
}
}, {
...baseQueueOptions(this.config, QUEUE.DELIVER),
...baseWorkerOptions(this.config, QUEUE.DELIVER),
autorun: false,
concurrency: this.config.deliverJobConcurrency ?? 128,
limiter: {
@ -323,7 +323,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
return this.inboxProcessorService.process(job);
}
}, {
...baseQueueOptions(this.config, QUEUE.INBOX),
...baseWorkerOptions(this.config, QUEUE.INBOX),
autorun: false,
concurrency: this.config.inboxJobConcurrency ?? 16,
limiter: {
@ -363,7 +363,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
return this.userWebhookDeliverProcessorService.process(job);
}
}, {
...baseQueueOptions(this.config, QUEUE.USER_WEBHOOK_DELIVER),
...baseWorkerOptions(this.config, QUEUE.USER_WEBHOOK_DELIVER),
autorun: false,
concurrency: 64,
limiter: {
@ -403,7 +403,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
return this.systemWebhookDeliverProcessorService.process(job);
}
}, {
...baseQueueOptions(this.config, QUEUE.SYSTEM_WEBHOOK_DELIVER),
...baseWorkerOptions(this.config, QUEUE.SYSTEM_WEBHOOK_DELIVER),
autorun: false,
concurrency: 16,
limiter: {
@ -453,7 +453,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
return processer(job);
}
}, {
...baseQueueOptions(this.config, QUEUE.RELATIONSHIP),
...baseWorkerOptions(this.config, QUEUE.RELATIONSHIP),
autorun: false,
concurrency: this.config.relationshipJobConcurrency ?? 16,
limiter: {
@ -498,7 +498,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
return processer(job);
}
}, {
...baseQueueOptions(this.config, QUEUE.OBJECT_STORAGE),
...baseWorkerOptions(this.config, QUEUE.OBJECT_STORAGE),
autorun: false,
concurrency: 16,
});
@ -531,7 +531,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
return this.endedPollNotificationProcessorService.process(job);
}
}, {
...baseQueueOptions(this.config, QUEUE.ENDED_POLL_NOTIFICATION),
...baseWorkerOptions(this.config, QUEUE.ENDED_POLL_NOTIFICATION),
autorun: false,
});
}

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

@ -3,6 +3,7 @@
* SPDX-License-Identifier: AGPL-3.0-only
*/
import { MetricsTime } from 'bullmq';
import { Config } from '@/config.js';
import type * as Bull from 'bullmq';
@ -28,3 +29,12 @@ export function baseQueueOptions(config: Config, queueName: typeof QUEUE[keyof t
prefix: config.redisForJobQueue.prefix ? `${config.redisForJobQueue.prefix}:queue:${queueName}` : `queue:${queueName}`,
};
}
export function baseWorkerOptions(config: Config, queueName: typeof QUEUE[keyof typeof QUEUE]): Bull.WorkerOptions {
return {
...baseQueueOptions(config, queueName),
metrics: {
maxDataPoints: MetricsTime.ONE_WEEK,
},
};
}