perf(worker): stop ledgering the two per-message tasks (#2136)

fluxer_api/src/api/channel/services/message/MessageMentionService.ts

@@ -499,7 +499,7 @@ export class MessageMentionService {
 			mentionEveryone ||
 			(message.reference && message.type === MessageTypes.REPLY);
 		if (hasMentions) {
-			await this.workerService.addJob('handleMentions', taskData);
+			await this.workerService.addJob('handleMentions', taskData, {skipLedger: true});
 		}
 	}
 }

fluxer_api/src/api/infrastructure/EmbedService.ts

@@ -507,7 +507,10 @@ export class EmbedService {
 				nsfwMode,
 				...(expectedContentHash ? {expectedContentHash} : {}),
 			},
-			{jobKey: expectedContentHash ? `${messageId.toString()}:${expectedContentHash}` : messageId.toString()},
+			{
+				jobKey: expectedContentHash ? `${messageId.toString()}:${expectedContentHash}` : messageId.toString(),
+				skipLedger: true,
+			},
 		);
 	}
 

fluxer_api/src/api/jobs/JobLedgerRepository.ts

@@ -71,7 +71,7 @@ export class JobLedgerRepository extends IJobLedgerRepository {
 		batch.addPrepared(JobsById.insert(idRow));
 		batch.addPrepared(JobsByDayBucket.insert(bucketRow));
 		batch.addPrepared(JobsActive.insert(activeRow));
-		await batch.executeChunked(10, true);
+		await batch.executeChunked(10, false);
 	}
 
 	async getJob(jobId: bigint): Promise<JobByIdRow | null> {

fluxer_api/src/api/worker/tests/PerMessageJobLedger.test.ts

@@ -0,0 +1,74 @@
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+import {MessageTypes} from '@fluxer/constants/src/ChannelConstants';
+import type {IWorkerService} from '@pkgs/worker/src/contracts/IWorkerService';
+import type {WorkerJobOptions, WorkerJobPayload} from '@pkgs/worker/src/contracts/WorkerTypes';
+import {describe, expect, it} from 'vitest';
+import {createChannelID, createMessageID, createRoleID, createUserID} from '../../BrandedTypes';
+import {MessageMentionService} from '../../channel/services/message/MessageMentionService';
+import {EmbedService} from '../../infrastructure/EmbedService';
+import type {Message} from '../../models/Message';
+import type {WorkerTaskName} from '../WorkerLaneConfig';
+
+class RecordingWorkerService implements IWorkerService<WorkerTaskName> {
+	readonly jobs: Array<{taskType: WorkerTaskName; options: WorkerJobOptions | undefined}> = [];
+
+	async addJob<TPayload extends WorkerJobPayload = WorkerJobPayload>(
+		taskType: WorkerTaskName,
+		_payload: TPayload,
+		options?: WorkerJobOptions,
+	): Promise<bigint> {
+		this.jobs.push({taskType, options});
+		return 1n;
+	}
+
+	async cancelJob(): Promise<boolean> {
+		return false;
+	}
+
+	async retryDeadLetterJob(): Promise<boolean> {
+		return false;
+	}
+}
+
+function makeMentionMessage(): Message {
+	return {
+		id: createMessageID(2n),
+		channelId: createChannelID(3n),
+		type: MessageTypes.DEFAULT,
+		reference: null,
+		mentionEveryone: false,
+		mentionedUserIds: new Set([createUserID(4n)]),
+		mentionedRoleIds: new Set([createRoleID(5n)]),
+	} as never;
+}
+
+describe('per-message worker jobs', () => {
+	it('enqueues handleMentions without a ledger row', async () => {
+		const workerService = new RecordingWorkerService();
+		const mentionService = new MessageMentionService(
+			null as never,
+			null as never,
+			null as never,
+			workerService,
+			null as never,
+		);
+		await mentionService.handleMentionTasks({
+			guildId: null,
+			message: makeMentionMessage(),
+			authorId: createUserID(1n),
+		});
+		expect(workerService.jobs).toHaveLength(1);
+		expect(workerService.jobs[0]!.taskType).toBe('handleMentions');
+		expect(workerService.jobs[0]!.options?.skipLedger).toBe(true);
+	});
+
+	it('enqueues extractEmbeds without a ledger row', async () => {
+		const workerService = new RecordingWorkerService();
+		const embedService = new EmbedService(null as never, null as never, null as never, workerService);
+		await embedService.enqueueUrlEmbedExtraction(createChannelID(3n), createMessageID(2n), null, 'block');
+		expect(workerService.jobs).toHaveLength(1);
+		expect(workerService.jobs[0]!.taskType).toBe('extractEmbeds');
+		expect(workerService.jobs[0]!.options?.skipLedger).toBe(true);
+	});
+});

tools/dev/cassandra_target_schema.json

@@ -6284,7 +6284,7 @@
 				}
 			],
 			"primary_key": "(bucket_day, created_at, job_id)",
-			"options": "CLUSTERING ORDER BY (created_at DESC, job_id DESC)"
+			"options": "CLUSTERING ORDER BY (created_at DESC, job_id DESC) AND default_time_to_live = 7776000"
 		},
 		{
 			"name": "jobs_by_id",
@@ -6375,7 +6375,7 @@
 				}
 			],
 			"primary_key": "(job_id)",
-			"options": ""
+			"options": "default_time_to_live = 7776000"
 		},
 		{
 			"name": "latest_risk_context_by_user",