wip

packages/backend/src/core/QueueService.ts

@@ -87,6 +87,12 @@ export class QueueService {
 			repeat: { pattern: '*/5 * * * *' },
 			removeOnComplete: true,
 		});
+
+		this.systemQueue.add('bakeBufferedReactions', {
+		}, {
+			repeat: { pattern: '0 0 * * *' },
+			removeOnComplete: true,
+		});
 	}
 
 	@bindThis

packages/backend/src/core/ReactionsBufferingService.ts

@@ -34,9 +34,9 @@ export class ReactionsBufferingService {
 	}
 
 	@bindThis
-	public async get(noteId: MiNote['id']) {
+	public async get(noteId: MiNote['id']): Promise<Record<string, number>> {
 		const result = await this.redisClient.hgetall(`${REDIS_PREFIX}:${noteId}`);
-		const delta = {};
+		const delta = {} as Record<string, number>;
 		for (const [name, count] of Object.entries(result)) {
 			delta[name] = parseInt(count);
 		}
@@ -44,7 +44,7 @@ export class ReactionsBufferingService {
 	}
 
 	@bindThis
-	public async getMany(noteIds: MiNote['id'][]) {
+	public async getMany(noteIds: MiNote['id'][]): Promise<Map<MiNote['id'], Record<string, number>>> {
 		const deltas = new Map<MiNote['id'], Record<string, number>>();
 
 		const pipeline = this.redisClient.pipeline();
@@ -55,8 +55,8 @@ export class ReactionsBufferingService {
 
 		for (let i = 0; i < noteIds.length; i++) {
 			const noteId = noteIds[i];
-			const result = results![i][1];
-			const delta = {};
+			const result = results![i][1] as Record<string, string>;
+			const delta = {} as Record<string, number>;
 			for (const [name, count] of Object.entries(result)) {
 				delta[name] = parseInt(count);
 			}
@@ -68,7 +68,7 @@ export class ReactionsBufferingService {
 
 	// TODO: scanは重い可能性があるので、別途 bufferedNoteIds を直接Redis上に持っておいてもいいかもしれない
 	@bindThis
-	public async bake() {
+	public async bake(): Promise<void> {
 		const bufferedNoteIds = [];
 		let cursor = '0';
 		do {

packages/backend/src/queue/QueueProcessorModule.ts

@@ -14,6 +14,7 @@ import { InboxProcessorService } from './processors/InboxProcessorService.js';
 import { UserWebhookDeliverProcessorService } from './processors/UserWebhookDeliverProcessorService.js';
 import { SystemWebhookDeliverProcessorService } from './processors/SystemWebhookDeliverProcessorService.js';
 import { CheckExpiredMutingsProcessorService } from './processors/CheckExpiredMutingsProcessorService.js';
+import { BakeBufferedReactionsProcessorService } from './processors/BakeBufferedReactionsProcessorService.js';
 import { CleanChartsProcessorService } from './processors/CleanChartsProcessorService.js';
 import { CleanProcessorService } from './processors/CleanProcessorService.js';
 import { CleanRemoteFilesProcessorService } from './processors/CleanRemoteFilesProcessorService.js';
@@ -51,6 +52,7 @@ import { RelationshipProcessorService } from './processors/RelationshipProcessor
 		ResyncChartsProcessorService,
 		CleanChartsProcessorService,
 		CheckExpiredMutingsProcessorService,
+		BakeBufferedReactionsProcessorService,
 		CleanProcessorService,
 		DeleteDriveFilesProcessorService,
 		ExportCustomEmojisProcessorService,

packages/backend/src/queue/QueueProcessorService.ts

@@ -39,6 +39,7 @@ import { TickChartsProcessorService } from './processors/TickChartsProcessorServ
 import { ResyncChartsProcessorService } from './processors/ResyncChartsProcessorService.js';
 import { CleanChartsProcessorService } from './processors/CleanChartsProcessorService.js';
 import { CheckExpiredMutingsProcessorService } from './processors/CheckExpiredMutingsProcessorService.js';
+import { BakeBufferedReactionsProcessorService } from './processors/BakeBufferedReactionsProcessorService.js';
 import { CleanProcessorService } from './processors/CleanProcessorService.js';
 import { AggregateRetentionProcessorService } from './processors/AggregateRetentionProcessorService.js';
 import { QueueLoggerService } from './QueueLoggerService.js';
@@ -118,6 +119,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
 		private cleanChartsProcessorService: CleanChartsProcessorService,
 		private aggregateRetentionProcessorService: AggregateRetentionProcessorService,
 		private checkExpiredMutingsProcessorService: CheckExpiredMutingsProcessorService,
+		private bakeBufferedReactionsProcessorService: BakeBufferedReactionsProcessorService,
 		private cleanProcessorService: CleanProcessorService,
 	) {
 		this.logger = this.queueLoggerService.logger;
@@ -147,6 +149,7 @@ export class QueueProcessorService implements OnApplicationShutdown {
 					case 'cleanCharts': return this.cleanChartsProcessorService.process();
 					case 'aggregateRetention': return this.aggregateRetentionProcessorService.process();
 					case 'checkExpiredMutings': return this.checkExpiredMutingsProcessorService.process();
+					case 'bakeBufferedReactions': return this.bakeBufferedReactionsProcessorService.process();
 					case 'clean': return this.cleanProcessorService.process();
 					default: throw new Error(`unrecognized job type ${job.name} for system`);
 				}

packages/backend/src/queue/processors/BakeBufferedReactionsProcessorService.ts

@@ -0,0 +1,32 @@
+/*
+ * SPDX-FileCopyrightText: syuilo and misskey-project
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+import { Inject, Injectable } from '@nestjs/common';
+import type Logger from '@/logger.js';
+import { bindThis } from '@/decorators.js';
+import { ReactionsBufferingService } from '@/core/ReactionsBufferingService.js';
+import { QueueLoggerService } from '../QueueLoggerService.js';
+import type * as Bull from 'bullmq';
+
+@Injectable()
+export class BakeBufferedReactionsProcessorService {
+	private logger: Logger;
+
+	constructor(
+		private reactionsBufferingService: ReactionsBufferingService,
+		private queueLoggerService: QueueLoggerService,
+	) {
+		this.logger = this.queueLoggerService.logger.createSubLogger('bake-buffered-reactions');
+	}
+
+	@bindThis
+	public async process(): Promise<void> {
+		this.logger.info('Baking buffered reactions...');
+
+		await this.reactionsBufferingService.bake();
+
+		this.logger.succ('All buffered reactions baked.');
+	}
+}