From 8278ef522103a9b52f3ad9791a1866c62469280d Mon Sep 17 00:00:00 2001 From: hided62 Date: Sun, 16 Aug 2026 18:09:04 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20read=20model=20outbox=20=EC=9E=AC?= =?UTF-8?q?=EC=8B=9C=EB=8F=84=20dispatcher=EB=A5=BC=20=EC=B6=94=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- packages/common/src/index.ts | 1 + .../common/src/realtime/readModelOutbox.ts | 135 ++++++++++++++ packages/common/src/realtime/types.ts | 6 +- packages/common/test/readModelOutbox.test.ts | 86 +++++++++ packages/infra/src/index.ts | 1 + .../infra/src/readModelOutboxDispatcher.ts | 175 ++++++++++++++++++ ...dModelOutboxDispatcher.integration.test.ts | 107 +++++++++++ .../test/readModelOutboxDispatcher.test.ts | 97 ++++++++++ packages/infra/vitest.config.ts | 4 + 9 files changed, 611 insertions(+), 1 deletion(-) create mode 100644 packages/common/src/realtime/readModelOutbox.ts create mode 100644 packages/common/test/readModelOutbox.test.ts create mode 100644 packages/infra/src/readModelOutboxDispatcher.ts create mode 100644 packages/infra/test/readModelOutboxDispatcher.integration.test.ts create mode 100644 packages/infra/test/readModelOutboxDispatcher.test.ts diff --git a/packages/common/src/index.ts b/packages/common/src/index.ts index 7f1dcc42..f30d5907 100644 --- a/packages/common/src/index.ts +++ b/packages/common/src/index.ts @@ -19,6 +19,7 @@ export * from './realtime/keys.js'; export * from './realtime/types.js'; export * from './realtime/delta.js'; export * from './realtime/changeJournal.js'; +export * from './realtime/readModelOutbox.js'; export * from './ranking/types.js'; export * from './ranking/legacyColor.js'; export * from './auth/accountIconProjection.js'; diff --git a/packages/common/src/realtime/readModelOutbox.ts b/packages/common/src/realtime/readModelOutbox.ts new file mode 100644 index 00000000..d8dd1184 --- /dev/null +++ b/packages/common/src/realtime/readModelOutbox.ts @@ -0,0 +1,135 @@ +import { + isReadModelDomain, + READ_MODEL_OUTBOX_PAYLOAD_VERSION, + type ReadModelOutboxPayloadV1, +} from './changeJournal.js'; +import { createEmptyRealtimeReadModelChanges, type RealtimeReadModelChanges } from './types.js'; + +const uniqueSortedIds = (values: Iterable): number[] => + [...new Set(values)].sort((left, right) => left - right); + +export const parseReadModelOutboxPayload = (value: unknown): ReadModelOutboxPayloadV1 | null => { + if (!value || typeof value !== 'object' || Array.isArray(value)) { + return null; + } + const record = value as Record; + if (record.version !== READ_MODEL_OUTBOX_PAYLOAD_VERSION || !Array.isArray(record.changes)) { + return null; + } + + const changes: Array = []; + for (const item of record.changes) { + if (!Array.isArray(item) || item.length !== 3) { + return null; + } + const [domain, entityId, revision] = item; + if ( + typeof domain !== 'string' || + !isReadModelDomain(domain) || + !Number.isSafeInteger(entityId) || + (entityId as number) < 0 || + typeof revision !== 'string' || + !/^(?:0|[1-9][0-9]*)$/u.test(revision) + ) { + return null; + } + try { + BigInt(revision); + } catch { + return null; + } + changes.push([domain, entityId as number, revision]); + } + return { version: READ_MODEL_OUTBOX_PAYLOAD_VERSION, changes } as ReadModelOutboxPayloadV1; +}; + +/** + * Converts the durable internal domain envelope back into the legacy internal + * invalidation shape. The result still contains entity IDs and must be + * viewer-filtered before it crosses the public SSE boundary. + */ +export const readModelOutboxPayloadToChanges = ( + payload: ReadModelOutboxPayloadV1 +): RealtimeReadModelChanges => { + const changes = createEmptyRealtimeReadModelChanges(); + const generalIds: number[] = []; + const cityIds: number[] = []; + const nationIds: number[] = []; + const mapGeneralIds: number[] = []; + const frontStatusNationIds: number[] = []; + const frontStatusActorIds: number[] = []; + const lobbyGeneralIds: number[] = []; + const reservedGeneralIds: number[] = []; + const recordGeneralIds: number[] = []; + + for (const [domain, entityId] of payload.changes) { + switch (domain) { + case 'general.content': + generalIds.push(entityId); + break; + case 'city.content': + cityIds.push(entityId); + break; + case 'nation.content': + nationIds.push(entityId); + break; + case 'world.content': + changes.worldChanged = true; + break; + case 'map.world': + changes.mapChanged = true; + break; + case 'map.general': + mapGeneralIds.push(entityId); + break; + case 'records.general': + recordGeneralIds.push(entityId); + break; + case 'records.global': + changes.globalRecordsChanged = true; + break; + case 'records.history': + changes.worldHistoryChanged = true; + break; + case 'front.general': + frontStatusActorIds.push(entityId); + break; + case 'front.nation': + frontStatusNationIds.push(entityId); + break; + case 'front.global': + changes.frontStatusChanged = true; + break; + case 'lobby.world': + changes.lobbyChanged = true; + break; + case 'lobby.general': + lobbyGeneralIds.push(entityId); + break; + case 'contacts.world': + changes.contactsChanged = true; + break; + case 'reserved.general': + reservedGeneralIds.push(entityId); + break; + case 'access.general': + case 'tournament': + case 'betting': + // These domains have no browser-wide dashboard invalidation. + break; + } + } + + return { + ...changes, + generalIds: uniqueSortedIds(generalIds), + cityIds: uniqueSortedIds(cityIds), + nationIds: uniqueSortedIds(nationIds), + mapGeneralIds: uniqueSortedIds(mapGeneralIds), + frontStatusNationIds: uniqueSortedIds(frontStatusNationIds), + frontStatusActorIds: uniqueSortedIds(frontStatusActorIds), + lobbyGeneralIds: uniqueSortedIds(lobbyGeneralIds), + reservedGeneralIds: uniqueSortedIds(reservedGeneralIds), + recordGeneralIds: uniqueSortedIds(recordGeneralIds), + }; +}; diff --git a/packages/common/src/realtime/types.ts b/packages/common/src/realtime/types.ts index c925fdae..9ac59176 100644 --- a/packages/common/src/realtime/types.ts +++ b/packages/common/src/realtime/types.ts @@ -30,6 +30,8 @@ export interface RealtimeReadModelChanges { contactsChanged: boolean; /** A global front-status source such as the active survey changed. */ frontStatusChanged?: boolean; + /** Shared base-map projection changed without implying other world slices. */ + mapChanged?: boolean; lobbyChanged?: boolean; } @@ -125,7 +127,7 @@ export const resolveRealtimeReadModelInvalidation = ( return { context: entityContextChanged, lobby: changes.worldChanged || lobbyChanged || ownLobbyGeneralChanged, - map: changes.worldChanged || mapEntitiesChanged || ownGeneralMapChanged, + map: changes.worldChanged || Boolean(changes.mapChanged) || mapEntitiesChanged || ownGeneralMapChanged, commands: changes.worldChanged || commandEntitiesChanged || ownGeneralChanged, contacts: changes.contactsChanged, boardAccess: ownGeneralChanged || ownNationChanged, @@ -160,6 +162,7 @@ export const createEmptyRealtimeReadModelChanges = (): RealtimeReadModelChanges worldHistoryChanged: false, contactsChanged: false, frontStatusChanged: false, + mapChanged: false, lobbyChanged: false, }); @@ -175,6 +178,7 @@ export const hasRealtimeReadModelChanges = (changes: RealtimeReadModelChanges): changes.worldHistoryChanged || changes.contactsChanged || Boolean(changes.frontStatusChanged) || + Boolean(changes.mapChanged) || Boolean(changes.lobbyChanged); export interface TurnCompletedEvent { diff --git a/packages/common/test/readModelOutbox.test.ts b/packages/common/test/readModelOutbox.test.ts new file mode 100644 index 00000000..2859687d --- /dev/null +++ b/packages/common/test/readModelOutbox.test.ts @@ -0,0 +1,86 @@ +import { describe, expect, it } from 'vitest'; + +import { + hasRealtimeReadModelChanges, + parseReadModelOutboxPayload, + readModelOutboxPayloadToChanges, + resolveRealtimeReadModelInvalidation, +} from '../src/index.js'; + +describe('read-model outbox payload', () => { + it('validates the version, domain, entity ID, and bigint revision', () => { + expect( + parseReadModelOutboxPayload({ + version: 1, + changes: [ + ['general.content', 7, '12'], + ['map.world', 0, '3'], + ], + }) + ).toEqual({ + version: 1, + changes: [ + ['general.content', 7, '12'], + ['map.world', 0, '3'], + ], + }); + expect(parseReadModelOutboxPayload({ version: 2, changes: [] })).toBeNull(); + expect(parseReadModelOutboxPayload({ version: 1, changes: [['unknown', 0, '1']] })).toBeNull(); + expect(parseReadModelOutboxPayload({ version: 1, changes: [['map.world', -1, '1']] })).toBeNull(); + expect(parseReadModelOutboxPayload({ version: 1, changes: [['map.world', 0, '-1']] })).toBeNull(); + }); + + it('reconstructs an idempotent internal invalidation without broadening map-only changes', () => { + const payload = parseReadModelOutboxPayload({ + version: 1, + changes: [ + ['general.content', 7, '2'], + ['map.general', 7, '2'], + ['map.world', 0, '9'], + ['front.general', 7, '3'], + ['front.nation', 4, '5'], + ['records.general', 7, '8'], + ['reserved.general', 7, '4'], + ['lobby.general', 7, '2'], + ['contacts.world', 0, '6'], + ], + }); + if (!payload) throw new Error('valid payload rejected'); + + const changes = readModelOutboxPayloadToChanges(payload); + expect(changes).toMatchObject({ + generalIds: [7], + mapGeneralIds: [7], + mapChanged: true, + frontStatusActorIds: [7], + frontStatusNationIds: [4], + recordGeneralIds: [7], + reservedGeneralIds: [7], + lobbyGeneralIds: [7], + contactsChanged: true, + worldChanged: false, + }); + expect(hasRealtimeReadModelChanges(changes)).toBe(true); + expect(resolveRealtimeReadModelInvalidation(changes, { generalId: 99, cityId: 2, nationId: 9 })).toMatchObject( + { + map: true, + context: false, + commands: false, + lobby: false, + } + ); + }); + + it('keeps access-only, tournament, and betting domains off the dashboard channel', () => { + const payload = parseReadModelOutboxPayload({ + version: 1, + changes: [ + ['access.general', 7, '1'], + ['tournament', 0, '1'], + ['betting', 0, '1'], + ], + }); + if (!payload) throw new Error('valid payload rejected'); + expect(hasRealtimeReadModelChanges(readModelOutboxPayloadToChanges(payload))).toBe(false); + }); +}); diff --git a/packages/infra/src/index.ts b/packages/infra/src/index.ts index 4fc613c3..7286b6ca 100644 --- a/packages/infra/src/index.ts +++ b/packages/infra/src/index.ts @@ -7,3 +7,4 @@ export * from './db.js'; export * from './redis.js'; export * from './turnEngineDb.js'; export * from './readModelChangeJournal.js'; +export * from './readModelOutboxDispatcher.js'; diff --git a/packages/infra/src/readModelOutboxDispatcher.ts b/packages/infra/src/readModelOutboxDispatcher.ts new file mode 100644 index 00000000..26c2e7b6 --- /dev/null +++ b/packages/infra/src/readModelOutboxDispatcher.ts @@ -0,0 +1,175 @@ +import { parseReadModelOutboxPayload, type ReadModelOutboxPayloadV1 } from '@sammo-ts/common'; + +import { GamePrisma, type GamePrismaClient } from './gamePrisma.js'; + +export interface ClaimedReadModelOutbox { + id: bigint; + payload: unknown; + attempts: number; +} + +type ClaimedRow = { + id: bigint; + payload: unknown; + attempts: number; +}; + +export interface ReadModelOutboxDispatchOptions { + owner: string; + limit?: number; + leaseMs?: number; + retryBaseMs?: number; + retryMaxMs?: number; + now?: () => Date; +} + +export interface ReadModelOutboxDispatchResult { + claimed: number; + delivered: number; + failed: number; +} + +const normalizeLimit = (value: number | undefined): number => + Math.min(500, Math.max(1, Math.floor(value ?? 50))); + +const normalizeDuration = (value: number | undefined, fallback: number): number => + Math.max(1, Math.floor(value ?? fallback)); + +const formatDispatchError = (error: unknown): string => { + const text = error instanceof Error ? `${error.name}: ${error.message}` : String(error); + return text.replaceAll(/\s+/gu, ' ').slice(0, 1_000); +}; + +const retryDelayMs = (attempts: number, baseMs: number, maxMs: number): number => { + const exponent = Math.min(20, Math.max(0, attempts - 1)); + return Math.min(maxMs, baseMs * 2 ** exponent); +}; + +export const claimReadModelOutboxBatch = async ( + db: GamePrismaClient, + options: Pick & { now?: Date } +): Promise => { + if (!options.owner.trim()) { + throw new Error('Read-model outbox owner must not be empty.'); + } + const limit = normalizeLimit(options.limit); + const leaseMs = normalizeDuration(options.leaseMs, 30_000); + const now = options.now ?? new Date(); + const leaseExpiredBefore = new Date(now.getTime() - leaseMs); + const rows = await db.$queryRaw(GamePrisma.sql` + WITH candidates AS ( + SELECT "id" + FROM "read_model_outbox" + WHERE "delivered_at" IS NULL + AND "available_at" <= ${now} + AND ("locked_at" IS NULL OR "locked_at" < ${leaseExpiredBefore}) + ORDER BY "id" + LIMIT ${limit} + FOR UPDATE SKIP LOCKED + ) + UPDATE "read_model_outbox" AS outbox + SET + "attempts" = outbox."attempts" + 1, + "locked_at" = ${now}, + "lock_owner" = ${options.owner}, + "last_error" = NULL + FROM candidates + WHERE outbox."id" = candidates."id" + RETURNING outbox."id", outbox."payload", outbox."attempts" + `); + + return rows.map((row) => ({ id: BigInt(row.id), payload: row.payload, attempts: row.attempts })); +}; + +export const markReadModelOutboxDelivered = async ( + db: GamePrismaClient, + input: { id: bigint; owner: string; deliveredAt?: Date } +): Promise => { + const result = await db.readModelOutbox.updateMany({ + where: { id: input.id, lockOwner: input.owner, deliveredAt: null }, + data: { + deliveredAt: input.deliveredAt ?? new Date(), + lockedAt: null, + lockOwner: null, + lastError: null, + }, + }); + return result.count === 1; +}; + +export const releaseReadModelOutbox = async ( + db: GamePrismaClient, + input: { id: bigint; owner: string; error: unknown; availableAt: Date } +): Promise => { + const result = await db.readModelOutbox.updateMany({ + where: { id: input.id, lockOwner: input.owner, deliveredAt: null }, + data: { + availableAt: input.availableAt, + lockedAt: null, + lockOwner: null, + lastError: formatDispatchError(input.error), + }, + }); + return result.count === 1; +}; + +export const dispatchReadModelOutboxBatch = async ( + db: GamePrismaClient, + publish: (payload: ReadModelOutboxPayloadV1, outboxId: bigint) => Promise, + options: ReadModelOutboxDispatchOptions +): Promise => { + const now = options.now ?? (() => new Date()); + const retryBaseMs = normalizeDuration(options.retryBaseMs, 1_000); + const retryMaxMs = Math.max(retryBaseMs, normalizeDuration(options.retryMaxMs, 60_000)); + const claimed = await claimReadModelOutboxBatch(db, { + owner: options.owner, + limit: options.limit, + leaseMs: options.leaseMs, + now: now(), + }); + let delivered = 0; + let failed = 0; + + for (const item of claimed) { + try { + const payload = parseReadModelOutboxPayload(item.payload); + if (!payload) { + throw new Error(`Read-model outbox ${item.id.toString()} has an invalid payload.`); + } + await publish(payload, item.id); + if (!(await markReadModelOutboxDelivered(db, { id: item.id, owner: options.owner, deliveredAt: now() }))) { + throw new Error(`Read-model outbox ${item.id.toString()} lost its delivery lease.`); + } + delivered += 1; + } catch (error) { + failed += 1; + await releaseReadModelOutbox(db, { + id: item.id, + owner: options.owner, + error, + availableAt: new Date(now().getTime() + retryDelayMs(item.attempts, retryBaseMs, retryMaxMs)), + }); + } + } + + return { claimed: claimed.length, delivered, failed }; +}; + +export const pruneDeliveredReadModelOutbox = async ( + db: GamePrismaClient, + input: { deliveredBefore: Date; limit?: number } +): Promise => { + const limit = normalizeLimit(input.limit); + const rows = await db.$queryRaw>(GamePrisma.sql` + DELETE FROM "read_model_outbox" + WHERE "id" IN ( + SELECT "id" + FROM "read_model_outbox" + WHERE "delivered_at" < ${input.deliveredBefore} + ORDER BY "id" + LIMIT ${limit} + ) + RETURNING "id" + `); + return rows.length; +}; diff --git a/packages/infra/test/readModelOutboxDispatcher.integration.test.ts b/packages/infra/test/readModelOutboxDispatcher.integration.test.ts new file mode 100644 index 00000000..754acb57 --- /dev/null +++ b/packages/infra/test/readModelOutboxDispatcher.integration.test.ts @@ -0,0 +1,107 @@ +import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from 'vitest'; + +import { createGamePostgresConnector, type GamePrismaClient } from '../src/gamePrisma.js'; +import { writeReadModelChangeJournal } from '../src/readModelChangeJournal.js'; +import { + claimReadModelOutboxBatch, + dispatchReadModelOutboxBatch, + pruneDeliveredReadModelOutbox, +} from '../src/readModelOutboxDispatcher.js'; + +const databaseUrl = process.env.READ_MODEL_JOURNAL_DATABASE_URL; +const integration = describe.skipIf(!databaseUrl); + +integration('read-model outbox PostgreSQL delivery boundary', () => { + let disconnect: (() => Promise) | undefined; + let prisma: GamePrismaClient; + + beforeAll(async () => { + if (!databaseUrl) throw new Error('READ_MODEL_JOURNAL_DATABASE_URL is required.'); + const connector = createGamePostgresConnector({ url: databaseUrl }); + prisma = connector.prisma; + disconnect = connector.disconnect; + await connector.connect(); + }); + + afterAll(async () => disconnect?.()); + + beforeEach(async () => { + await prisma.$executeRaw`TRUNCATE TABLE "read_model_outbox", "read_model_revision" RESTART IDENTITY`; + }); + + const enqueue = async (entityId: number): Promise => { + await prisma.$transaction((transaction) => + writeReadModelChangeJournal(transaction, [{ domain: 'general.content', entityId }]) + ); + }; + + it('claims concurrent worker batches without overlap', async () => { + await Promise.all(Array.from({ length: 20 }, (_, index) => enqueue(index + 1))); + const now = new Date('2099-08-16T00:00:00.000Z'); + const [left, right] = await Promise.all([ + claimReadModelOutboxBatch(prisma, { owner: 'left', limit: 10, now }), + claimReadModelOutboxBatch(prisma, { owner: 'right', limit: 10, now }), + ]); + const ids = [...left, ...right].map(({ id }) => id); + expect(ids).toHaveLength(20); + expect(new Set(ids).size).toBe(20); + }); + + it('releases a publish failure and later delivers the same row', async () => { + await enqueue(7); + const failedAt = new Date('2099-08-16T00:00:00.000Z'); + await expect( + dispatchReadModelOutboxBatch(prisma, vi.fn().mockRejectedValue(new Error('redis unavailable')), { + owner: 'worker-a', + now: () => failedAt, + retryBaseMs: 1_000, + }) + ).resolves.toEqual({ claimed: 1, delivered: 0, failed: 1 }); + const released = await prisma.readModelOutbox.findUniqueOrThrow({ where: { id: 1n } }); + expect(released).toMatchObject({ attempts: 1, lockedAt: null, lockOwner: null, deliveredAt: null }); + expect(released.lastError).toContain('redis unavailable'); + + const retryAt = new Date('2099-08-16T00:00:01.000Z'); + const publish = vi.fn().mockResolvedValue(undefined); + await expect( + dispatchReadModelOutboxBatch(prisma, publish, { owner: 'worker-b', now: () => retryAt }) + ).resolves.toEqual({ claimed: 1, delivered: 1, failed: 0 }); + expect(publish).toHaveBeenCalledTimes(1); + await expect(prisma.readModelOutbox.findUniqueOrThrow({ where: { id: 1n } })).resolves.toMatchObject({ + attempts: 2, + lockOwner: null, + deliveredAt: retryAt, + }); + }); + + it('allows a lease-expired row to be republished after publish-before-ack crash', async () => { + await enqueue(7); + const first = await claimReadModelOutboxBatch(prisma, { + owner: 'crashed-worker', + leaseMs: 30_000, + now: new Date('2099-08-16T00:00:00.000Z'), + }); + expect(first).toHaveLength(1); + + const second = await claimReadModelOutboxBatch(prisma, { + owner: 'recovery-worker', + leaseMs: 30_000, + now: new Date('2099-08-16T00:00:31.000Z'), + }); + expect(second.map(({ id }) => id)).toEqual(first.map(({ id }) => id)); + expect(second[0]?.attempts).toBe(2); + }); + + it('prunes delivered rows in bounded batches', async () => { + await Promise.all([enqueue(1), enqueue(2), enqueue(3)]); + const deliveredAt = new Date('2099-08-15T00:00:00.000Z'); + await prisma.readModelOutbox.updateMany({ data: { deliveredAt } }); + await expect( + pruneDeliveredReadModelOutbox(prisma, { + deliveredBefore: new Date('2099-08-16T00:00:00.000Z'), + limit: 2, + }) + ).resolves.toBe(2); + await expect(prisma.readModelOutbox.count()).resolves.toBe(1); + }); +}); diff --git a/packages/infra/test/readModelOutboxDispatcher.test.ts b/packages/infra/test/readModelOutboxDispatcher.test.ts new file mode 100644 index 00000000..9e93f05b --- /dev/null +++ b/packages/infra/test/readModelOutboxDispatcher.test.ts @@ -0,0 +1,97 @@ +import { describe, expect, it, vi } from 'vitest'; + +import type { GamePrismaClient } from '../src/gamePrisma.js'; +import { + claimReadModelOutboxBatch, + dispatchReadModelOutboxBatch, + pruneDeliveredReadModelOutbox, +} from '../src/readModelOutboxDispatcher.js'; + +const validPayload = { + version: 1, + changes: [['general.content', 7, '3']], +}; + +const createDb = (rows: readonly object[]) => { + const queryRaw = vi.fn().mockResolvedValue(rows); + const updateMany = vi.fn().mockResolvedValue({ count: 1 }); + return { + db: { $queryRaw: queryRaw, readModelOutbox: { updateMany } } as unknown as GamePrismaClient, + queryRaw, + updateMany, + }; +}; + +describe('read-model outbox dispatcher', () => { + it('claims a bounded lease with one statement and preserves delivery identity', async () => { + const fixture = createDb([{ id: 41n, payload: validPayload, attempts: 2 }]); + await expect( + claimReadModelOutboxBatch(fixture.db, { + owner: 'worker-a', + limit: 25, + leaseMs: 15_000, + now: new Date('2026-08-16T00:00:00.000Z'), + }) + ).resolves.toEqual([{ id: 41n, payload: validPayload, attempts: 2 }]); + expect(fixture.queryRaw).toHaveBeenCalledTimes(1); + }); + + it('publishes and acknowledges a valid payload', async () => { + const fixture = createDb([{ id: 1n, payload: validPayload, attempts: 1 }]); + const publish = vi.fn().mockResolvedValue(undefined); + const result = await dispatchReadModelOutboxBatch(fixture.db, publish, { + owner: 'worker-a', + now: () => new Date('2026-08-16T00:00:00.000Z'), + }); + + expect(result).toEqual({ claimed: 1, delivered: 1, failed: 0 }); + expect(publish).toHaveBeenCalledWith(validPayload, 1n); + expect(fixture.updateMany).toHaveBeenCalledWith( + expect.objectContaining({ + where: { id: 1n, lockOwner: 'worker-a', deliveredAt: null }, + data: expect.objectContaining({ deliveredAt: new Date('2026-08-16T00:00:00.000Z') }), + }) + ); + }); + + it('releases failed and malformed rows with bounded retry state', async () => { + const fixture = createDb([ + { id: 1n, payload: validPayload, attempts: 3 }, + { id: 2n, payload: { version: 99 }, attempts: 1 }, + ]); + const publish = vi.fn().mockRejectedValue(new Error('redis unavailable')); + const result = await dispatchReadModelOutboxBatch(fixture.db, publish, { + owner: 'worker-a', + retryBaseMs: 1_000, + retryMaxMs: 10_000, + now: () => new Date('2026-08-16T00:00:00.000Z'), + }); + + expect(result).toEqual({ claimed: 2, delivered: 0, failed: 2 }); + expect(publish).toHaveBeenCalledTimes(1); + expect(fixture.updateMany).toHaveBeenNthCalledWith( + 1, + expect.objectContaining({ + where: { id: 1n, lockOwner: 'worker-a', deliveredAt: null }, + data: expect.objectContaining({ availableAt: new Date('2026-08-16T00:00:04.000Z') }), + }) + ); + expect(fixture.updateMany).toHaveBeenNthCalledWith( + 2, + expect.objectContaining({ + where: { id: 2n, lockOwner: 'worker-a', deliveredAt: null }, + data: expect.objectContaining({ availableAt: new Date('2026-08-16T00:00:01.000Z') }), + }) + ); + }); + + it('prunes only a bounded delivered batch', async () => { + const fixture = createDb([{ id: 1n }, { id: 2n }]); + await expect( + pruneDeliveredReadModelOutbox(fixture.db, { + deliveredBefore: new Date('2026-08-15T00:00:00.000Z'), + limit: 100, + }) + ).resolves.toBe(2); + }); +}); diff --git a/packages/infra/vitest.config.ts b/packages/infra/vitest.config.ts index 9e8e7606..bb6b2926 100644 --- a/packages/infra/vitest.config.ts +++ b/packages/infra/vitest.config.ts @@ -7,6 +7,10 @@ export default defineConfig({ test: { environment: 'node', globals: true, + // Integration files share the explicitly supplied disposable schema. + // Keep file-level TRUNCATE/setup boundaries from racing each other; + // individual tests still create concurrent writers deliberately. + fileParallelism: false, include: ['test/**/*.test.ts'], }, });