feat(catalog): persist idempotent bulk operations and retryable deliveries

integration/database.test.ts

@@ -4,6 +4,7 @@ import { copyFile, mkdir, mkdtemp, rm } from "node:fs/promises";
 import { join, resolve } from "node:path";
 import { pathToFileURL } from "node:url";
 import { promisify } from "node:util";
+import { sql } from "drizzle-orm";
 import type { RowDataPacket } from "mysql2/promise";
 import mysql from "mysql2/promise";
 import {
@@ -18,6 +19,7 @@ const migrations = [
 	"0027_catalog_packages.sql",
 	"0028_history_snapshots.sql",
 	"0029_admin_table_views.sql",
+	"0031_operations_outbox.sql",
 ];
 let maria: StartedTestContainer | undefined;
 let redisContainer: StartedTestContainer | undefined;
@@ -25,6 +27,7 @@ let connection: mysql.Connection | undefined;
 let appDb: typeof import("@/lib/db").db | undefined;
 let appRedis: typeof import("@/lib/redis").redis;
 let commands: typeof import("@/features/catalog/server/bulk-offers");
+let operations: typeof import("@/features/operations/server");
 let cache: typeof import("@/lib/cache");
 let migrationRoot: string | undefined;
 let databaseUrl: string;
@@ -131,6 +134,7 @@ beforeAll(async () => {
 	await appRedis.ping();
 	commands = await import("@/features/catalog/server/bulk-offers");
 	cache = await import("@/lib/cache");
+	operations = await import("@/features/operations/server");
 });
 
 afterAll(async () => {
@@ -159,6 +163,8 @@ afterAll(async () => {
 
 beforeEach(async () => {
 	if (!connection) throw Error("Integration database is not connected");
+	await connection.query("DELETE FROM cms_outbox");
+	await connection.query("DELETE FROM cms_operations");
 	await connection.query("DROP TRIGGER IF EXISTS reject_second_history");
 	await connection.query("DELETE FROM admin_audit_log");
 	await connection.query("DELETE FROM catalog_items");
@@ -195,7 +201,7 @@ describe("MariaDB migrations and catalog transactions", () => {
 			{ name: "Saved" },
 		]);
 		expect((await migrate("--status")).stdout).toContain(
-			"3/3 applied, 0 pending",
+			"4/4 applied, 0 pending",
 		);
 		const columns = await rows(
 			"SELECT COLUMN_NAME, DATA_TYPE FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='admin_audit_log' AND COLUMN_NAME IN ('before','after') ORDER BY COLUMN_NAME",
@@ -306,3 +312,110 @@ describe("Redis application cache", () => {
 		expect(await appRedis?.get(second)).toBe('"second"');
 	});
 });
+
+describe("operation idempotency and transactional outbox", () => {
+	it("replays concurrent identical requests after applying their mutation and effect only once", async () => {
+		const request = {
+			actorId: 7,
+			kind: "integration.test",
+			key: randomUUID(),
+			input: { increment: 3 },
+		};
+		let executions = 0;
+		const work = async (
+			tx: import("@/features/operations/server").OperationTransaction,
+			id: string,
+		) => {
+			executions++;
+			await tx.execute(
+				sql`UPDATE catalog_items SET cost_credits=cost_credits+3 WHERE id=1`,
+			);
+			await operations.enqueueEffect(tx, id, "catalog.refresh");
+			return { operationId: id, changed: 1 };
+		};
+		const result = await Promise.all([
+			operations.runOperation(request, work),
+			operations.runOperation(request, work),
+		]);
+		expect(result[0]).toEqual(result[1]);
+		expect(executions).toBe(1);
+		expect(
+			await rows("SELECT cost_credits FROM catalog_items WHERE id=1"),
+		).toEqual([{ cost_credits: 13 }]);
+		expect(await rows("SELECT id FROM cms_operations")).toHaveLength(1);
+		expect(await rows("SELECT topic,status FROM cms_outbox")).toEqual([
+			{ topic: "catalog.refresh", status: "pending" },
+		]);
+		await expect(
+			operations.runOperation({ ...request, input: { increment: 9 } }, work),
+		).rejects.toThrow();
+		expect(executions).toBe(1);
+		expect(
+			await rows("SELECT cost_credits FROM catalog_items WHERE id=1"),
+		).toEqual([{ cost_credits: 13 }]);
+	});
+
+	it("rolls back the request, mutation and queued effect together, allowing the same request to retry", async () => {
+		const request = {
+			actorId: 7,
+			kind: "integration.test",
+			key: randomUUID(),
+			input: { increment: 3 },
+		};
+		await expect(
+			operations.runOperation(request, async (tx, id) => {
+				await tx.execute(
+					sql`UPDATE catalog_items SET cost_credits=99 WHERE id=1`,
+				);
+				await operations.enqueueEffect(tx, id, "catalog.refresh");
+				throw Error("failure after effect enqueue");
+			}),
+		).rejects.toThrow("failure after effect enqueue");
+		expect(
+			await rows("SELECT cost_credits FROM catalog_items WHERE id=1"),
+		).toEqual([{ cost_credits: 10 }]);
+		expect(await rows("SELECT id FROM cms_operations")).toHaveLength(0);
+		expect(await rows("SELECT id FROM cms_outbox")).toHaveLength(0);
+		await expect(
+			operations.runOperation(request, async (tx, id) => {
+				await tx.execute(
+					sql`UPDATE catalog_items SET cost_credits=13 WHERE id=1`,
+				);
+				await operations.enqueueEffect(tx, id, "catalog.refresh");
+				return { changed: 1 };
+			}),
+		).resolves.toEqual({ changed: 1 });
+		expect(
+			await rows("SELECT cost_credits FROM catalog_items WHERE id=1"),
+		).toEqual([{ cost_credits: 13 }]);
+		expect(await rows("SELECT id FROM cms_outbox")).toHaveLength(1);
+	});
+
+	it("leases a queued effect to only one consumer and ignores a stale completion token", async () => {
+		await operations.runOperation(
+			{ actorId: 7, kind: "integration.test", key: randomUUID(), input: {} },
+			async (tx, id) => {
+				await operations.enqueueEffect(tx, id, "catalog.refresh");
+				return { queued: true };
+			},
+		);
+		const claims = await Promise.all([
+			operations.effectRepository.claim(),
+			operations.effectRepository.claim(),
+		]);
+		expect(claims.filter(Boolean)).toHaveLength(1);
+		const claim = claims.find((value) => value !== null);
+		if (!claim) throw Error("Expected one effect lease");
+		await operations.effectRepository.complete({
+			...claim,
+			token: randomUUID(),
+		});
+		expect(await rows("SELECT status FROM cms_outbox")).toEqual([
+			{ status: "running" },
+		]);
+		await operations.effectRepository.complete(claim);
+		expect(await rows("SELECT status FROM cms_outbox")).toEqual([
+			{ status: "done" },
+		]);
+	});
+});