de-duplicate import jobs

packages/backend/src/core/QueueService.ts

@@ -17,6 +17,7 @@ import { bindThis } from '@/decorators.js';
 import type { Antenna } from '@/server/api/endpoints/i/import-antennas.js';
 import { ApRequestCreator } from '@/core/activitypub/ApRequestService.js';
 import { type SystemWebhookPayload } from '@/core/SystemWebhookService.js';
+import { MiNote } from '@/models/Note.js';
 import { type UserWebhookPayload } from './UserWebhookService.js';
 import type {
 	DbJobData,
@@ -40,7 +41,6 @@ import type {
 } from './QueueModule.js';
 import type httpSignature from '@peertube/http-signature';
 import type * as Bull from 'bullmq';
-import { MiNote } from '@/models/Note.js';
 
 export const QUEUE_TYPES = [
 	'system',
@@ -424,6 +424,9 @@ export class QueueService {
 				age: 3600 * 24 * 7, // keep up to 7 days
 				count: 100,
 			},
+			deduplication: {
+				id: `${user.id}_${fileId}_${withReplies ?? false}`,
+			},
 		});
 	}
 
@@ -436,6 +439,9 @@ export class QueueService {
 		}, {
 			removeOnComplete: true,
 			removeOnFail: true,
+			deduplication: {
+				id: `${user.id}_${fileId}_${type ?? null}`,
+			},
 		});
 	}
 
@@ -495,6 +501,9 @@ export class QueueService {
 				age: 3600 * 24 * 7, // keep up to 7 days
 				count: 100,
 			},
+			deduplication: {
+				id: `${user.id}_${fileId}`,
+			},
 		});
 	}
 
@@ -512,6 +521,9 @@ export class QueueService {
 				age: 3600 * 24 * 7, // keep up to 7 days
 				count: 100,
 			},
+			deduplication: {
+				id: `${user.id}_${fileId}`,
+			},
 		});
 	}
 
@@ -557,6 +569,9 @@ export class QueueService {
 				age: 3600 * 24 * 7, // keep up to 7 days
 				count: 100,
 			},
+			deduplication: {
+				id: `${user.id}_${fileId}`,
+			},
 		});
 	}
 
@@ -574,14 +589,18 @@ export class QueueService {
 				age: 3600 * 24 * 7, // keep up to 7 days
 				count: 100,
 			},
+			deduplication: {
+				id: `${user.id}_${fileId}`,
+			},
 		});
 	}
 
 	@bindThis
-	public createImportAntennasJob(user: ThinUser, antenna: Antenna) {
+	public createImportAntennasJob(user: ThinUser, antenna: Antenna, fileId: MiDriveFile['id']) {
 		return this.dbQueue.add('importAntennas', {
 			user: { id: user.id },
 			antenna,
+			fileId,
 		}, {
 			removeOnComplete: {
 				age: 3600 * 24 * 7, // keep up to 7 days
@@ -591,6 +610,9 @@ export class QueueService {
 				age: 3600 * 24 * 7, // keep up to 7 days
 				count: 100,
 			},
+			deduplication: {
+				id: `${user.id}_${fileId}`,
+			},
 		});
 	}
 

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

@@ -11,8 +11,8 @@ import Logger from '@/logger.js';
 import type { AntennasRepository, UsersRepository } from '@/models/_.js';
 import { DI } from '@/di-symbols.js';
 import { bindThis } from '@/decorators.js';
-import { QueueLoggerService } from '../QueueLoggerService.js';
 import { NotificationService } from '@/core/NotificationService.js';
+import { QueueLoggerService } from '../QueueLoggerService.js';
 import { DBAntennaImportJobData } from '../types.js';
 import type * as Bull from 'bullmq';
 
@@ -79,7 +79,7 @@ export class ImportAntennasProcessorService {
 			return;
 		}
 
-		this.logger.debug(`Importing blocking of ${job.data.user.id} ...`);
+		this.logger.debug(`Importing antennas of ${job.data.user.id} ...`);
 
 		const now = new Date();
 		try {

packages/backend/src/queue/types.ts

@@ -110,6 +110,7 @@ export type DbNoteImportJobData = {
 export type DBAntennaImportJobData = {
 	user: ThinUser,
 	antenna: Antenna
+	fileId: MiDriveFile['id'];
 };
 
 export type DbUserImportToDbJobData = {

packages/backend/src/server/api/endpoints/i/import-antennas.ts

@@ -82,7 +82,7 @@ export default class extends Endpoint<typeof meta, typeof paramDef> {
 			if (currentAntennasCount + antennas.length >= (await this.roleService.getUserPolicies(me.id)).antennaLimit) {
 				throw new ApiError(meta.errors.tooManyAntennas);
 			}
-			this.queueService.createImportAntennasJob(me, antennas);
+			await this.queueService.createImportAntennasJob(me, antennas, file.id);
 		});
 	}
 }