fix: tweak retention rate aggregation

packages/backend/migration/1678869617549-retention-date-key.js

@@ -0,0 +1,14 @@
+export class retentionDateKey1678869617549 {
+    name = 'retentionDateKey1678869617549'
+
+    async up(queryRunner) {
+			await queryRunner.query(`TRUNCATE TABLE "retention_aggregation"`, undefined);
+        await queryRunner.query(`ALTER TABLE "retention_aggregation" ADD "dateKey" character varying(512) NOT NULL`);
+        await queryRunner.query(`CREATE UNIQUE INDEX "IDX_f7c3576b37bd2eec966ae24477" ON "retention_aggregation" ("dateKey") `);
+    }
+
+    async down(queryRunner) {
+        await queryRunner.query(`DROP INDEX "public"."IDX_f7c3576b37bd2eec966ae24477"`);
+        await queryRunner.query(`ALTER TABLE "retention_aggregation" DROP COLUMN "dateKey"`);
+    }
+}

packages/backend/src/models/entities/RetentionAggregation.ts

@@ -18,6 +18,12 @@ export class RetentionAggregation {
 	})
 	public updatedAt: Date;
 
+	@Index({ unique: true })
+	@Column('varchar', {
+		length: 512, nullable: false,
+	})
+	public dateKey: string;
+
 	@Column({
 		...id(),
 		array: true,

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

@@ -7,6 +7,7 @@ import { bindThis } from '@/decorators.js';
 import type { RetentionAggregationsRepository, UsersRepository } from '@/models/index.js';
 import { deepClone } from '@/misc/clone.js';
 import { IdService } from '@/core/IdService.js';
+import { isDuplicateKeyValueError } from '@/misc/is-duplicate-key-value-error.js';
 import { QueueLoggerService } from '../QueueLoggerService.js';
 import type Bull from 'bull';
 
@@ -49,13 +50,21 @@ export class AggregateRetentionProcessorService {
 		});
 		const targetUserIds = targetUsers.map(u => u.id);
 
-		await this.retentionAggregationsRepository.insert({
-			id: this.idService.genId(),
-			createdAt: now,
-			updatedAt: now,
-			userIds: targetUserIds,
-			usersCount: targetUserIds.length,
-		});
+		try {
+			await this.retentionAggregationsRepository.insert({
+				id: this.idService.genId(),
+				createdAt: now,
+				updatedAt: now,
+				dateKey,
+				userIds: targetUserIds,
+				usersCount: targetUserIds.length,
+			});
+		} catch (err) {
+			if (isDuplicateKeyValueError(err)) {
+				this.logger.succ('Skip because it has already been processed by another worker.');
+				done();
+			}
+		}
 
 		// 今日活動したユーザーを全て取得
 		const activeUsers = await this.usersRepository.findBy({