feat: API 변화 저널과 outbox worker를 연결

입력 이벤트의 업무 변경, 성공 원장, revision 및 outbox를 한 transaction에서 커밋한다. 설문의 commit 전 Redis 발행을 제거하고 접속 점수용 private revision과 재시도·retention을 갖춘 dispatcher lifecycle을 추가한다.
2026-08-16 18:22:57 +00:00
parent b41b161e9c
commit f858ca43f4
15 changed files with 606 additions and 52 deletions
@@ -1,4 +1,5 @@
 import { z } from 'zod';
+import type { ChangeJournal } from '@sammo-ts/common';
 import type { GameSessionTokenPayload } from '@sammo-ts/common/auth/gameToken';
 import type { DatabaseClient as InfraDatabaseClient, RedisConnector, GamePrisma } from '@sammo-ts/infra';
 import { normalizeScenarioEffect, SCENARIO_EFFECT_KEYS } from '@sammo-ts/logic';
@@ -10,6 +11,7 @@ import type { RedisAccessTokenStore } from './auth/accessTokenStore.js';
 import type { AccountIconSource } from './auth/accountIconSource.js';
 import type { ProfileStatusSource } from './auth/profileStatusSource.js';
 import type { ContentImageUploadStore } from './services/remoteContentImageStore.js';
+import type { ReadModelOutboxWakeup } from './realtime/outboxWorker.js';

 export interface GameProfile {
    id: string;
@@ -87,6 +89,10 @@ export interface GameApiContext {
    generalAccessTracking?: boolean;
    /** Request-local identity already resolved by the realtime access gate. */
    realtimeAccessGeneralId?: number;
+    /** Set only while an API input-event transaction owns the mutation. */
+    changeJournal?: ChangeJournal;
+    /** Post-commit scheduling hint for the durable outbox dispatcher. */
+    readModelOutbox?: ReadModelOutboxWakeup;
    db: DatabaseClient;
    redis: RedisConnector['client'];
    turnDaemon: TurnDaemonTransport;
@@ -125,6 +131,7 @@ export const createGameApiContext = (options: {
    gameTokenSecret: string;
    accountIconSource?: AccountIconSource;
    profileStatusSource: ProfileStatusSource;
+    readModelOutbox?: ReadModelOutboxWakeup;
 }): GameApiContext => {
    return {
        requestId: options.requestId,
@@ -145,5 +152,6 @@ export const createGameApiContext = (options: {
        gameTokenSecret: options.gameTokenSecret,
        ...(options.accountIconSource ? { accountIconSource: options.accountIconSource } : {}),
        profileStatusSource: options.profileStatusSource,
+        ...(options.readModelOutbox ? { readModelOutbox: options.readModelOutbox } : {}),
    };
 };
@@ -29,6 +29,7 @@ export * from './tournament/keys.js';
 export * from './tournament/store.js';
 export * from './tournament/types.js';
 export * from './tournament/worker.js';
+export * from './realtime/outboxWorker.js';

 // Types for TRPC consumer
 export type { MessageView } from './messages/store.js';
@@ -0,0 +1,164 @@
+import { randomUUID } from 'node:crypto';
+import {
+    readModelOutboxPayloadToChanges,
+    type ReadModelDomain,
+} from '@sammo-ts/common';
+import {
+    dispatchReadModelOutboxBatch,
+    pruneDeliveredReadModelOutbox,
+    type ReadModelOutboxDatabase,
+    type ReadModelOutboxDispatchResult,
+    type RedisConnector,
+} from '@sammo-ts/infra';
+
+import { publishRealtimeReadModelChanges } from './publisher.js';
+
+// access.general is an authoritative DB-only source revision. Tournament and
+// betting still have separate Redis-owned source revisions. None of the three
+// should wake the legacy dashboard channel solely because its outbox row ran.
+const NON_DASHBOARD_DOMAINS: ReadonlySet<ReadModelDomain> = new Set([
+    'access.general',
+    'tournament',
+    'betting',
+]);
+
+export interface ReadModelOutboxWakeup {
+    wake(): void;
+}
+
+export interface ReadModelOutboxWorkerOptions {
+    intervalMs?: number;
+    batchSize?: number;
+    leaseMs?: number;
+    retentionMs?: number;
+    pruneIntervalMs?: number;
+    pruneLimit?: number;
+    owner?: string;
+    now?: () => Date;
+    onError?: (error: unknown) => void;
+}
+
+const normalizePositiveInteger = (value: number | undefined, fallback: number): number =>
+    Math.max(1, Math.floor(value ?? fallback));
+
+/**
+ * Polls the durable API/engine outbox without overlapping batches. `wake()` is
+ * only a scheduling hint; delivery always reclaims committed PostgreSQL rows.
+ */
+export class ReadModelOutboxWorker implements ReadModelOutboxWakeup {
+    private timer: NodeJS.Timeout | null = null;
+    private inFlight: Promise<void> | null = null;
+    private rerunRequested = false;
+    private running = false;
+    private readonly intervalMs: number;
+    private readonly batchSize: number;
+    private readonly leaseMs: number;
+    private readonly retentionMs: number;
+    private readonly pruneIntervalMs: number;
+    private readonly pruneLimit: number;
+    private readonly owner: string;
+    private readonly now: () => Date;
+    private readonly onError: (error: unknown) => void;
+    private nextPruneAt: number;
+
+    constructor(
+        private readonly db: ReadModelOutboxDatabase,
+        private readonly redis: RedisConnector['client'],
+        private readonly profileName: string,
+        options: ReadModelOutboxWorkerOptions = {}
+    ) {
+        this.intervalMs = normalizePositiveInteger(options.intervalMs, 1_000);
+        this.batchSize = normalizePositiveInteger(options.batchSize, 50);
+        this.leaseMs = normalizePositiveInteger(options.leaseMs, 30_000);
+        this.retentionMs = normalizePositiveInteger(options.retentionMs, 24 * 60 * 60 * 1_000);
+        this.pruneIntervalMs = normalizePositiveInteger(options.pruneIntervalMs, 60_000);
+        this.pruneLimit = normalizePositiveInteger(options.pruneLimit, 100);
+        this.owner = options.owner ?? `game-api:${profileName}:${process.pid}:${randomUUID()}`;
+        this.now = options.now ?? (() => new Date());
+        this.onError = options.onError ?? (() => undefined);
+        this.nextPruneAt = this.now().getTime() + this.pruneIntervalMs;
+    }
+
+    private async dispatchOnce(): Promise<ReadModelOutboxDispatchResult> {
+        const result = await dispatchReadModelOutboxBatch(
+            this.db,
+            async (payload) => {
+                if (payload.changes.every(([domain]) => NON_DASHBOARD_DOMAINS.has(domain))) {
+                    return;
+                }
+                const changes = readModelOutboxPayloadToChanges(payload);
+                await publishRealtimeReadModelChanges(this.redis, this.profileName, changes);
+            },
+            {
+                owner: this.owner,
+                limit: this.batchSize,
+                leaseMs: this.leaseMs,
+            }
+        );
+        if (result.failed > 0) {
+            this.reportError(new Error(`${result.failed} read-model outbox delivery attempt(s) failed.`));
+        }
+
+        const now = this.now();
+        if (now.getTime() >= this.nextPruneAt) {
+            this.nextPruneAt = now.getTime() + this.pruneIntervalMs;
+            await pruneDeliveredReadModelOutbox(this.db, {
+                deliveredBefore: new Date(now.getTime() - this.retentionMs),
+                limit: this.pruneLimit,
+            });
+        }
+        return result;
+    }
+
+    private reportError(error: unknown): void {
+        try {
+            this.onError(error);
+        } catch {
+            // Observability callbacks must not stop durable retry polling.
+        }
+    }
+
+    private runScheduledBatch(): void {
+        if (!this.running || this.inFlight) {
+            return;
+        }
+        this.rerunRequested = false;
+        this.inFlight = this.dispatchOnce()
+            .then(() => undefined)
+            .catch((error: unknown) => this.reportError(error))
+            .finally(() => {
+                this.inFlight = null;
+                if (this.running && this.rerunRequested) {
+                    this.runScheduledBatch();
+                }
+            });
+    }
+
+    start(): void {
+        if (this.running) {
+            return;
+        }
+        this.running = true;
+        this.timer = setInterval(() => this.wake(), this.intervalMs);
+        this.timer.unref?.();
+        this.wake();
+    }
+
+    wake(): void {
+        if (!this.running) {
+            return;
+        }
+        this.rerunRequested = true;
+        this.runScheduledBatch();
+    }
+
+    async stop(): Promise<void> {
+        this.running = false;
+        this.rerunRequested = false;
+        if (this.timer) {
+            clearInterval(this.timer);
+            this.timer = null;
+        }
+        await this.inFlight;
+    }
+}
@@ -1,7 +1,7 @@
 import { TRPCError } from '@trpc/server';
 import { z } from 'zod';

-import { asRecord, createEmptyRealtimeReadModelChanges, LiteHashDRBG, RandUtil } from '@sammo-ts/common';
+import { asRecord, LiteHashDRBG, RandUtil } from '@sammo-ts/common';
 import { GamePrisma } from '@sammo-ts/infra';
 import {
    ITEM_KEYS,
@@ -17,24 +17,8 @@ import {
 } from '@sammo-ts/logic';

 import { authedProcedure, router } from '../../trpc.js';
-import type { GameApiContext } from '../../context.js';
 import { getMyGeneral } from '../shared/general.js';
 import { loadCurrentGameTime, type CurrentGameTime } from '../../services/gameClock.js';
-import { publishRealtimeReadModelChanges } from '../../realtime/publisher.js';
-
-const publishFrontStatusChange = async (
-    ctx: GameApiContext,
-    options: { generalId?: number; global?: boolean }
-): Promise<void> => {
-    const changes = createEmptyRealtimeReadModelChanges();
-    if (options.generalId) changes.frontStatusActorIds = [options.generalId];
-    if (options.global) changes.frontStatusChanged = true;
-    try {
-        await publishRealtimeReadModelChanges(ctx.redis, ctx.profile.name, changes);
-    } catch {
-        // 설문 DB mutation은 이미 commit되었으므로 실시간 알림 실패로 되돌리지 않는다.
-    }
-};

 const hasAdminRole = (roles: string[], profileName: string): boolean => {
    if (roles.includes('superuser') || roles.includes('admin') || roles.includes('admin.superuser')) {
@@ -523,7 +507,7 @@ export const voteRouter = router({
                throw new TRPCError({ code: 'BAD_REQUEST', message: rewardResult.reason });
            }

-            await publishFrontStatusChange(ctx, { generalId: general.id });
+            ctx.changeJournal?.mark('front.general', general.id);
            return { ok: true, wonLottery: rewardResult.awardedUnique };
        }),
    addComment: authedProcedure
@@ -634,7 +618,7 @@ export const voteRouter = router({
                )
            `);

-            await publishFrontStatusChange(ctx, { global: true });
+            ctx.changeJournal?.mark('front.global');
            return { ok: true };
        }),
    updatePoll: adminProcedure
@@ -732,7 +716,7 @@ export const voteRouter = router({
            `);

            if (input.title !== undefined || endAt !== undefined) {
-                await publishFrontStatusChange(ctx, { global: true });
+                ctx.changeJournal?.mark('front.global');
            }
            return { ok: true };
        }),
@@ -748,7 +732,7 @@ export const voteRouter = router({
            if (!rows[0]?.id) {
                throw new TRPCError({ code: 'NOT_FOUND', message: '설문조사가 없습니다.' });
            }
-            await publishFrontStatusChange(ctx, { global: true });
+            ctx.changeJournal?.mark('front.global');
            return { ok: true };
        }),
    getAdminStatus: adminProcedure.query(async () => ({ ok: true })),
@@ -30,6 +30,7 @@ import { createAdminProfileIconResetFlushHandler } from './services/accountIconS
 import { AccountIconResetReconciler } from './services/accountIconResetReconciler.js';
 import { createBestEffortResourceCloser } from './services/bestEffortResourceCloser.js';
 import { RemoteContentImageStore } from './services/remoteContentImageStore.js';
+import { ReadModelOutboxWorker } from './realtime/outboxWorker.js';

 const extractBearerToken = (value: string | string[] | undefined): string | null => {
    if (!value) {
@@ -140,9 +141,16 @@ export const createGameApiServer = async () => {
        throw error;
    }
    const realtimeHub = new RedisRealtimeEventHub(realtimeSubscriberClient, buildGameEventChannel(config.profileName));
+    const readModelOutboxWorker = new ReadModelOutboxWorker(postgres.prisma, redis.client, config.profileName, {
+        onError: (error) => app.log.error({ err: error }, 'read-model outbox dispatch failed'),
+    });
    let flushSubscriberStarted = false;
    let realtimeHubStarted = false;
    const closeResources = createBestEffortResourceCloser([
+        {
+            name: 'read-model-outbox-worker',
+            run: () => readModelOutboxWorker.stop(),
+        },
        {
            name: 'account-icon-reset-reconciler',
            run: () => accountIconResetReconciler.stop(),
@@ -220,6 +228,7 @@ export const createGameApiServer = async () => {
                    gameTokenSecret: config.gameTokenSecret,
                    accountIconSource,
                    profileStatusSource,
+                    readModelOutbox: readModelOutboxWorker,
                });
            },
        },
@@ -334,6 +343,7 @@ export const createGameApiServer = async () => {
        realtimeHubStarted = true;
        await flushSubscriber.start();
        flushSubscriberStarted = true;
+        readModelOutboxWorker.start();
        accountIconResetReconciler.start();
    } catch (error) {
        await closeResources();
@@ -1,5 +1,5 @@
 import { asRecord, resolveAccessLimitLevel, resolveAccessRefreshLimit, type AccessLimitLevel } from '@sammo-ts/common';
-import { GamePrisma } from '@sammo-ts/infra';
+import { GamePrisma, writeReadModelChangeJournal } from '@sammo-ts/infra';

 import type { GameApiContext } from '../context.js';

@@ -371,17 +371,21 @@ export const upsertGeneralAccess = async (
                        general_access_log.refresh_score_total + EXCLUDED.refresh_score_total
            `
        );
+
+        await writeReadModelChangeJournal(transaction, [
+            { domain: 'access.general', entityId: input.generalId },
+        ]);
    });
 };

 export const recordGeneralAccess = async (
-    ctx: Pick<GameApiContext, 'auth' | 'db' | 'profile' | 'profileStatusSource'>,
+    ctx: Pick<GameApiContext, 'auth' | 'db' | 'profile' | 'profileStatusSource' | 'readModelOutbox'>,
    page: AccessPage,
    now = new Date()
 ): Promise<boolean> => recordGeneralAccessWeight(ctx, accessPageWeights[page], now);

 export const recordGeneralAccessWeight = async (
-    ctx: Pick<GameApiContext, 'auth' | 'db' | 'profile' | 'profileStatusSource'>,
+    ctx: Pick<GameApiContext, 'auth' | 'db' | 'profile' | 'profileStatusSource' | 'readModelOutbox'>,
    weight: number,
    now = new Date()
 ): Promise<boolean> => {
@@ -450,5 +454,6 @@ export const recordGeneralAccessWeight = async (
        periodStartedAt,
        scoreStartedAt,
    });
+    ctx.readModelOutbox?.wake();
    return true;
 };
@@ -1,6 +1,8 @@
 import { randomUUID } from 'node:crypto';
 import { initTRPC, TRPCError } from '@trpc/server';
+import { ChangeJournal } from '@sammo-ts/common';
 import { isGameAccessBlocked } from '@sammo-ts/common/auth/sanctions';
+import { writeReadModelChangeJournal } from '@sammo-ts/infra';

 import type { GameApiContext } from './context.js';
 import { IdempotentTurnDaemonTransport } from './daemon/idempotentTransport.js';
@@ -45,8 +47,10 @@ const inputEventMiddleware = t.middleware(async ({ ctx, type, path, next }) => {
    }

    const requestId = `${ctx.requestId ?? randomUUID()}:${path}`;
+    const changeJournal = new ChangeJournal();
+    let journalPersisted = false;
    try {
-        return await executeInputEvent({
+        const result = await executeInputEvent({
            db: ctx.db,
            requestId,
            eventType: path,
@@ -56,15 +60,21 @@ const inputEventMiddleware = t.middleware(async ({ ctx, type, path, next }) => {
                    ctx: {
                        ...ctx,
                        db: transaction,
+                        changeJournal,
                        turnDaemon: new IdempotentTurnDaemonTransport(ctx.turnDaemon, requestId),
                    },
                });
                if (!result.ok) {
                    throw result.error;
                }
+                journalPersisted = Boolean(await writeReadModelChangeJournal(transaction, changeJournal.snapshot()));
                return result;
            },
        });
+        if (journalPersisted) {
+            ctx.readModelOutbox?.wake();
+        }
+        return result;
    } catch (error) {
        if (error instanceof DuplicateInputEventError) {
            throw new TRPCError({
@@ -47,7 +47,13 @@ const buildDb = (
    access: { lastRefresh: Date | null; refreshScore: number } | null = null
 ) => {
    const executeRaw = vi.fn(async (_query: unknown) => 1);
-    const queryRaw = vi.fn(async (_query: unknown) => [{ id: 41 }]);
+    const queryRaw = vi.fn(async (query: unknown) => {
+        const sql = (query as { sql?: string }).sql ?? '';
+        if (sql.includes('read_model_revision')) {
+            return [{ domain: 'access.general', entityId: 7, revision: 1n, outboxId: 1n }];
+        }
+        return [{ id: 41 }];
+    });
    const transaction = vi.fn(
        async (
            callback: (client: { $executeRaw: typeof executeRaw; $queryRaw: typeof queryRaw }) => Promise<unknown>
@@ -157,7 +163,7 @@ describe('general access tracking', () => {
            select: { id: true, userId: true, turnTime: true },
        });
        expect(transaction).toHaveBeenCalledTimes(1);
-        expect(queryRaw).toHaveBeenCalledTimes(1);
+        expect(queryRaw).toHaveBeenCalledTimes(2);
        expect(executeRaw).toHaveBeenCalledTimes(2);

        const periodStatement = queryRaw.mock.calls[0]![0] as { sql: string; values: unknown[] };
@@ -185,6 +191,11 @@ describe('general access tracking', () => {
        expect(accessStatement.values).toContain(2);
        expect(accessStatement.values).toContain(now);
        expect(accessStatement.values).toContainEqual(new Date('2026-07-26T03:00:00.000Z'));
+
+        const journalStatement = queryRaw.mock.calls[1]![0] as { sql: string; values: unknown[] };
+        expect(journalStatement.sql).toContain('INSERT INTO "read_model_outbox"');
+        expect(journalStatement.values).toContain('access.general');
+        expect(journalStatement.values).toContain(7);
    });

    it('accepts legacy weight zero to refresh timestamps without incrementing counters', async () => {
@@ -242,7 +253,12 @@ describe('general access tracking', () => {
        const events: string[] = [];
        let transactionCount = 0;
        const transactionClient = {
-            $queryRaw: vi.fn(async () => [{ id: 41 }]),
+            $queryRaw: vi.fn(async (query: unknown) => {
+                const sql = (query as { sql?: string }).sql ?? '';
+                return sql.includes('read_model_revision')
+                    ? [{ domain: 'access.general', entityId: 7, revision: 1n, outboxId: 1n }]
+                    : [{ id: 41 }];
+            }),
            $executeRaw: vi.fn(async () => 1),
            inputEvent: {
                update: vi.fn(async () => ({})),
@@ -1,7 +1,10 @@
-import { afterAll, beforeAll, describe, expect, it } from 'vitest';
+import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest';
+import { z } from 'zod';
 import { createGamePostgresConnector, type GamePrismaClient } from '@sammo-ts/infra';

+import type { GameApiContext } from '../src/context.js';
 import { DuplicateInputEventError, executeInputEvent } from '../src/inputEventBoundary.js';
+import { procedure, router } from '../src/trpc.js';
 import {
    ConflictingTurnDaemonCommandError,
    DatabaseTurnDaemonTransport,
@@ -10,10 +13,36 @@ import {

 const databaseUrl = process.env.INPUT_EVENT_DATABASE_URL;
 const integration = describe.skipIf(!databaseUrl);
+const journalGeneralIds = [9_980_081, 9_980_082] as const;
+
+const journalBoundaryRouter = router({
+    mutate: procedure
+        .input(z.object({ generalId: z.number().int(), fail: z.boolean().optional().default(false) }))
+        .mutation(({ ctx, input }) => {
+            ctx.changeJournal?.mark('front.general', input.generalId);
+            if (input.fail) {
+                throw new Error('injected journal rollback');
+            }
+            return { ok: true };
+        }),
+});
+
+const payloadHasGeneral = (payload: unknown, generalId: number): boolean => {
+    if (!payload || typeof payload !== 'object' || !('changes' in payload)) return false;
+    const changes = (payload as { changes?: unknown }).changes;
+    return (
+        Array.isArray(changes) &&
+        changes.some(
+            (change) =>
+                Array.isArray(change) && change[0] === 'front.general' && change[1] === generalId
+        )
+    );
+};

 integration('API input event boundary', () => {
    let close: (() => Promise<void>) | undefined;
    let db: GamePrismaClient;
+    const createdOutboxIds: bigint[] = [];

    beforeAll(async () => {
        const connector = createGamePostgresConnector({ url: databaseUrl! });
@@ -23,12 +52,21 @@ integration('API input event boundary', () => {
        await db.inputEvent.deleteMany({
            where: { requestId: { startsWith: 'integration:api:' } },
        });
+        await db.readModelRevision.deleteMany({
+            where: { domain: 'front.general', entityId: { in: [...journalGeneralIds] } },
+        });
    });

    afterAll(async () => {
        await db.inputEvent.deleteMany({
            where: { requestId: { startsWith: 'integration:api:' } },
        });
+        if (createdOutboxIds.length > 0) {
+            await db.readModelOutbox.deleteMany({ where: { id: { in: createdOutboxIds } } });
+        }
+        await db.readModelRevision.deleteMany({
+            where: { domain: 'front.general', entityId: { in: [...journalGeneralIds] } },
+        });
        await close?.();
    });

@@ -64,6 +102,80 @@ integration('API input event boundary', () => {
        expect(marker.status).toBe('PENDING');
    });

+    it('persists an API journal with SUCCEEDED and only wakes delivery after commit', async () => {
+        const requestId = 'integration:api:journal-success';
+        const redisPublish = vi.fn();
+        let wakeSnapshot: Promise<unknown> | undefined;
+        const context = {
+            db,
+            requestId,
+            redis: { publish: redisPublish },
+            readModelOutbox: {
+                wake: () => {
+                    wakeSnapshot = Promise.all([
+                        db.inputEvent.findUniqueOrThrow({ where: { requestId: `${requestId}:mutate` } }),
+                        db.readModelRevision.findUniqueOrThrow({
+                            where: {
+                                domain_entityId: {
+                                    domain: 'front.general',
+                                    entityId: journalGeneralIds[0],
+                                },
+                            },
+                        }),
+                    ]);
+                },
+            },
+        } as unknown as GameApiContext;
+
+        await expect(
+            journalBoundaryRouter.createCaller(context).mutate({ generalId: journalGeneralIds[0] })
+        ).resolves.toEqual({ ok: true });
+
+        expect(wakeSnapshot).toBeDefined();
+        const [event, revision] = (await wakeSnapshot) as [
+            { status: string },
+            { revision: bigint },
+        ];
+        expect(event.status).toBe('SUCCEEDED');
+        expect(revision.revision).toBe(1n);
+        expect(redisPublish).not.toHaveBeenCalled();
+
+        const outboxes = await db.readModelOutbox.findMany({ select: { id: true, payload: true } });
+        const outbox = outboxes.find(({ payload }) => payloadHasGeneral(payload, journalGeneralIds[0]));
+        expect(outbox).toBeDefined();
+        if (outbox) createdOutboxIds.push(outbox.id);
+    });
+
+    it('rolls back an API journal and never schedules delivery when the handler fails', async () => {
+        const requestId = 'integration:api:journal-rollback';
+        const wake = vi.fn();
+        const context = {
+            db,
+            requestId,
+            readModelOutbox: { wake },
+        } as unknown as GameApiContext;
+
+        await expect(
+            journalBoundaryRouter
+                .createCaller(context)
+                .mutate({ generalId: journalGeneralIds[1], fail: true })
+        ).rejects.toThrow('injected journal rollback');
+
+        await expect(
+            db.readModelRevision.findUnique({
+                where: {
+                    domain_entityId: {
+                        domain: 'front.general',
+                        entityId: journalGeneralIds[1],
+                    },
+                },
+            })
+        ).resolves.toBeNull();
+        const outboxes = await db.readModelOutbox.findMany({ select: { payload: true } });
+        expect(outboxes.some(({ payload }) => payloadHasGeneral(payload, journalGeneralIds[1]))).toBe(false);
+        expect(wake).not.toHaveBeenCalled();
+    });
+
    it('rolls back business writes, records failure, and permits one explicit retry', async () => {
        const requestId = 'integration:api:retry';
        const markerId = 'integration:api:retry:marker';
@@ -0,0 +1,100 @@
+import { describe, expect, it, vi } from 'vitest';
+import { z } from 'zod';
+
+import type { GameApiContext } from '../src/context.js';
+import { procedure, router } from '../src/trpc.js';
+
+const testRouter = router({
+    mutate: procedure
+        .input(z.object({ fail: z.boolean().optional().default(false) }))
+        .mutation(({ ctx, input }) => {
+            (ctx as GameApiContext & { testOrder: string[] }).testOrder.push('handler');
+            ctx.changeJournal?.mark('front.general', 7);
+            if (input.fail) throw new Error('injected rollback');
+            return { ok: true };
+        }),
+});
+
+const createContext = () => {
+    const order: string[] = [];
+    const queryRaw = vi.fn(async () => {
+        order.push('journal');
+        return [{ domain: 'front.general', entityId: 7, revision: 1n, outboxId: 11n }];
+    });
+    const transaction = {
+        $queryRaw: queryRaw,
+        inputEvent: {
+            update: vi.fn(async () => {
+                order.push('succeeded');
+                return {};
+            }),
+        },
+    };
+    const db = {
+        inputEvent: {
+            create: vi.fn(async () => {
+                order.push('accepted');
+                return {};
+            }),
+            updateMany: vi.fn(async () => ({ count: 0 })),
+            update: vi.fn(async () => {
+                order.push('failed');
+                return {};
+            }),
+        },
+        $transaction: vi.fn(async (callback: (db: typeof transaction) => Promise<unknown>) => {
+            order.push('transaction-begin');
+            try {
+                const result = await callback(transaction);
+                order.push('commit');
+                return result;
+            } catch (error) {
+                order.push('rollback');
+                throw error;
+            }
+        }),
+    };
+    const redisPublish = vi.fn();
+    const wake = vi.fn(() => order.push('wake'));
+    const context = {
+        requestId: 'journal-unit',
+        db,
+        redis: { publish: redisPublish },
+        readModelOutbox: { wake },
+        testOrder: order,
+    } as unknown as GameApiContext & { testOrder: string[] };
+    return { context, order, queryRaw, redisPublish, wake };
+};
+
+describe('API input-event change journal boundary', () => {
+    it('writes the journal with SUCCEEDED, commits, and only then schedules delivery', async () => {
+        const fixture = createContext();
+
+        await expect(testRouter.createCaller(fixture.context).mutate({})).resolves.toEqual({ ok: true });
+
+        expect(fixture.order).toEqual([
+            'accepted',
+            'transaction-begin',
+            'handler',
+            'journal',
+            'succeeded',
+            'commit',
+            'wake',
+        ]);
+        expect(fixture.redisPublish).not.toHaveBeenCalled();
+        expect(fixture.wake).toHaveBeenCalledTimes(1);
+    });
+
+    it('rolls back a handler mark without writing or scheduling an outbox row', async () => {
+        const fixture = createContext();
+
+        await expect(testRouter.createCaller(fixture.context).mutate({ fail: true })).rejects.toThrow(
+            'injected rollback'
+        );
+
+        expect(fixture.order).toEqual(['accepted', 'transaction-begin', 'handler', 'rollback', 'failed']);
+        expect(fixture.queryRaw).not.toHaveBeenCalled();
+        expect(fixture.redisPublish).not.toHaveBeenCalled();
+        expect(fixture.wake).not.toHaveBeenCalled();
+    });
+});
@@ -0,0 +1,143 @@
+import { describe, expect, it, vi } from 'vitest';
+
+import type { RedisConnector } from '@sammo-ts/infra';
+import type { ReadModelOutboxDatabase } from '@sammo-ts/infra';
+
+import { ReadModelOutboxWorker } from '../src/realtime/outboxWorker.js';
+
+const payload = (domain: 'front.general' | 'access.general' | 'tournament' | 'betting') => ({
+    version: 1,
+    changes: [[domain, domain === 'front.general' || domain === 'access.general' ? 7 : 0, '1']],
+});
+
+const createFixture = (rows: readonly object[]) => {
+    const queryRaw = vi.fn().mockResolvedValueOnce(rows).mockResolvedValue([]);
+    const updateMany = vi.fn().mockResolvedValue({ count: 1 });
+    const incr = vi.fn().mockResolvedValue(41);
+    const publish = vi.fn().mockResolvedValue(1);
+    const db = {
+        $queryRaw: queryRaw,
+        readModelOutbox: { updateMany },
+    } as unknown as ReadModelOutboxDatabase;
+    const redis = { incr, publish } as unknown as RedisConnector['client'];
+    return { db, redis, queryRaw, updateMany, incr, publish };
+};
+
+describe('ReadModelOutboxWorker', () => {
+    it('publishes a legacy internal readModelChanged event and acknowledges the durable row', async () => {
+        const fixture = createFixture([{ id: 11n, payload: payload('front.general'), attempts: 1 }]);
+        const worker = new ReadModelOutboxWorker(fixture.db, fixture.redis, 'che:default', {
+            owner: 'worker-test',
+            intervalMs: 60_000,
+        });
+
+        worker.start();
+        await vi.waitFor(() => expect(fixture.updateMany).toHaveBeenCalledTimes(1));
+        await worker.stop();
+
+        expect(fixture.incr).toHaveBeenCalledWith('sammo:che:default:read-model:revision');
+        const event = JSON.parse(String(fixture.publish.mock.calls[0]?.[1]));
+        expect(event).toMatchObject({
+            type: 'readModelChanged',
+            revision: 41,
+            changes: { frontStatusActorIds: [7] },
+        });
+        expect(fixture.updateMany).toHaveBeenCalledWith(
+            expect.objectContaining({ where: { id: 11n, lockOwner: 'worker-test', deliveredAt: null } })
+        );
+    });
+
+    it.each(['access.general', 'tournament', 'betting'] as const)(
+        'marks a %s-only envelope delivered without dashboard Redis publish',
+        async (domain) => {
+            const fixture = createFixture([{ id: 12n, payload: payload(domain), attempts: 1 }]);
+            const worker = new ReadModelOutboxWorker(fixture.db, fixture.redis, 'che:default', {
+                owner: 'worker-test',
+                intervalMs: 60_000,
+            });
+
+            worker.start();
+            await vi.waitFor(() => expect(fixture.updateMany).toHaveBeenCalledTimes(1));
+            await worker.stop();
+
+            expect(fixture.incr).not.toHaveBeenCalled();
+            expect(fixture.publish).not.toHaveBeenCalled();
+        }
+    );
+
+    it('coalesces repeated wakeups into one trailing batch and waits for it on shutdown', async () => {
+        let releaseFirst: (() => void) | undefined;
+        const first = new Promise<readonly object[]>((resolve) => {
+            releaseFirst = () => resolve([]);
+        });
+        const fixture = createFixture([]);
+        fixture.queryRaw.mockReset();
+        fixture.queryRaw.mockReturnValueOnce(first).mockResolvedValue([]);
+        const worker = new ReadModelOutboxWorker(fixture.db, fixture.redis, 'che:default', {
+            owner: 'worker-test',
+            intervalMs: 60_000,
+        });
+
+        worker.start();
+        worker.wake();
+        worker.wake();
+        expect(fixture.queryRaw).toHaveBeenCalledTimes(1);
+
+        releaseFirst?.();
+        await vi.waitFor(() => expect(fixture.queryRaw).toHaveBeenCalledTimes(2));
+        await worker.stop();
+        expect(fixture.queryRaw).toHaveBeenCalledTimes(2);
+    });
+
+    it('reports item failures while leaving the row released for dispatcher retry', async () => {
+        const fixture = createFixture([{ id: 13n, payload: payload('front.general'), attempts: 1 }]);
+        fixture.publish.mockRejectedValueOnce(new Error('redis unavailable'));
+        const onError = vi.fn();
+        const worker = new ReadModelOutboxWorker(fixture.db, fixture.redis, 'che:default', {
+            owner: 'worker-test',
+            intervalMs: 60_000,
+            onError,
+        });
+
+        worker.start();
+        await vi.waitFor(() => expect(fixture.updateMany).toHaveBeenCalledTimes(1));
+        await worker.stop();
+
+        expect(onError).toHaveBeenCalledWith(
+            expect.objectContaining({ message: '1 read-model outbox delivery attempt(s) failed.' })
+        );
+        expect(fixture.updateMany).toHaveBeenCalledWith(
+            expect.objectContaining({
+                where: { id: 13n, lockOwner: 'worker-test', deliveredAt: null },
+                data: expect.objectContaining({
+                    lockedAt: null,
+                    lockOwner: null,
+                    lastError: expect.stringContaining('redis unavailable'),
+                }),
+            })
+        );
+    });
+
+    it('prunes only a bounded retention batch on the lower-frequency cadence', async () => {
+        let now = new Date('2026-08-16T00:00:00.000Z');
+        const fixture = createFixture([]);
+        const worker = new ReadModelOutboxWorker(fixture.db, fixture.redis, 'che:default', {
+            owner: 'worker-test',
+            intervalMs: 60_000,
+            retentionMs: 24 * 60 * 60 * 1_000,
+            pruneIntervalMs: 60_000,
+            pruneLimit: 100,
+            now: () => now,
+        });
+        now = new Date('2026-08-16T00:01:00.000Z');
+
+        worker.start();
+        await vi.waitFor(() => expect(fixture.queryRaw).toHaveBeenCalledTimes(2));
+        await worker.stop();
+
+        const pruneQuery = fixture.queryRaw.mock.calls[1]?.[0] as { sql: string; values: unknown[] };
+        expect(pruneQuery.sql).toContain('DELETE FROM "read_model_outbox"');
+        expect(pruneQuery.values).toContainEqual(new Date('2026-08-15T00:01:00.000Z'));
+        expect(pruneQuery.values).toContain(100);
+    });
+});
@@ -1,5 +1,6 @@
 import { describe, expect, it, vi } from 'vitest';

+import { ChangeJournal } from '@sammo-ts/common';
 import type { GameSessionTokenPayload } from '@sammo-ts/common/auth/gameToken';
 import type { GamePrisma, RedisConnector } from '@sammo-ts/infra';

@@ -106,6 +107,7 @@ const buildContext = (options: {
    }));
    const redisIncr = vi.fn(async (_key: string) => 41);
    const redisPublish = vi.fn(async (_channel: string, _message: string) => 1);
+    const changeJournal = new ChangeJournal();
    const queryRaw = vi.fn(async (query: GamePrisma.Sql) => {
        const text = sqlText(query);
        if (text.includes('FROM vote_poll') && text.includes('LIMIT 1')) {
@@ -193,8 +195,9 @@ const buildContext = (options: {
        accessTokenStore,
        flushStore: new InMemoryFlushStore(),
        gameTokenSecret: 'test-secret',
+        changeJournal,
    };
-    return { context, requestCommand, queryRaw, db, redisIncr, redisPublish };
+    return { context, requestCommand, queryRaw, db, redisIncr, redisPublish, changeJournal };
 };

 describe('vote router actor and permission boundaries', () => {
@@ -221,16 +224,9 @@ describe('vote router actor and permission boundaries', () => {
                goldReward: 90,
            })
        );
-        expect(fixture.redisIncr).toHaveBeenCalledWith('sammo:che:default:read-model:revision');
-        const published = JSON.parse(String(fixture.redisPublish.mock.calls[0]?.[1]));
-        expect(published).toMatchObject({
-            type: 'readModelChanged',
-            revision: 41,
-            changes: {
-                frontStatusActorIds: [7],
-                frontStatusChanged: false,
-            },
-        });
+        expect(fixture.changeJournal.snapshot()).toEqual([{ domain: 'front.general', entityId: 7 }]);
+        expect(fixture.redisIncr).not.toHaveBeenCalled();
+        expect(fixture.redisPublish).not.toHaveBeenCalled();
    });

    it('publishes a global front-status projection after creating a survey', async () => {
@@ -244,11 +240,9 @@ describe('vote router actor and permission boundaries', () => {
            })
        ).resolves.toEqual({ ok: true });

-        const published = JSON.parse(String(fixture.redisPublish.mock.calls[0]?.[1]));
-        expect(published).toMatchObject({
-            type: 'readModelChanged',
-            changes: { frontStatusChanged: true },
-        });
+        expect(fixture.changeJournal.snapshot()).toEqual([{ domain: 'front.global', entityId: 0 }]);
+        expect(fixture.redisIncr).not.toHaveBeenCalled();
+        expect(fixture.redisPublish).not.toHaveBeenCalled();
    });

    it('uses the current world develcost for the legacy five-times survey reward', async () => {
@@ -44,4 +44,5 @@ export interface DatabaseClient {
    vote: GamePrisma.VoteDelegate;
    inputEvent: GamePrisma.InputEventDelegate;
    turnDaemonLease: GamePrisma.TurnDaemonLeaseDelegate;
+    readModelOutbox: GamePrisma.ReadModelOutboxDelegate;
 }
@@ -6,7 +6,9 @@ import {
    type ReadModelRevisionKey,
 } from '@sammo-ts/common';

-import { GamePrisma } from './gamePrisma.js';
+import { GamePrisma, type GamePrismaClient } from './gamePrisma.js';
+
+export type ReadModelJournalDatabase = Pick<GamePrismaClient, '$queryRaw'>;

 interface ReadModelJournalWriteRow {
    domain: string;
@@ -61,7 +63,7 @@ const assertWriteRows = (
 * that owns the domain mutation; this function never opens or commits one.
 */
 export const writeReadModelChangeJournal = async (
-    transaction: GamePrisma.TransactionClient,
+    transaction: ReadModelJournalDatabase,
    candidates: Iterable<ReadModelRevisionKey>
 ): Promise<ReadModelJournalWriteResult | null> => {
    const keys = normalizeReadModelRevisionKeys(candidates);
@@ -2,6 +2,10 @@ import { parseReadModelOutboxPayload, type ReadModelOutboxPayloadV1 } from '@sam

 import { GamePrisma, type GamePrismaClient } from './gamePrisma.js';

+export interface ReadModelOutboxDatabase extends Pick<GamePrismaClient, '$queryRaw'> {
+    readModelOutbox: GamePrisma.ReadModelOutboxDelegate;
+}
+
 export interface ClaimedReadModelOutbox {
    id: bigint;
    payload: unknown;
@@ -46,7 +50,7 @@ const retryDelayMs = (attempts: number, baseMs: number, maxMs: number): number =
 };

 export const claimReadModelOutboxBatch = async (
-    db: GamePrismaClient,
+    db: ReadModelOutboxDatabase,
    options: Pick<ReadModelOutboxDispatchOptions, 'owner' | 'limit' | 'leaseMs'> & { now?: Date }
 ): Promise<ClaimedReadModelOutbox[]> => {
    if (!options.owner.trim()) {
@@ -82,7 +86,7 @@ export const claimReadModelOutboxBatch = async (
 };

 export const markReadModelOutboxDelivered = async (
-    db: GamePrismaClient,
+    db: ReadModelOutboxDatabase,
    input: { id: bigint; owner: string; deliveredAt?: Date }
 ): Promise<boolean> => {
    const result = await db.readModelOutbox.updateMany({
@@ -98,7 +102,7 @@ export const markReadModelOutboxDelivered = async (
 };

 export const releaseReadModelOutbox = async (
-    db: GamePrismaClient,
+    db: ReadModelOutboxDatabase,
    input: { id: bigint; owner: string; error: unknown; availableAt: Date }
 ): Promise<boolean> => {
    const result = await db.readModelOutbox.updateMany({
@@ -114,7 +118,7 @@ export const releaseReadModelOutbox = async (
 };

 export const dispatchReadModelOutboxBatch = async (
-    db: GamePrismaClient,
+    db: ReadModelOutboxDatabase,
    publish: (payload: ReadModelOutboxPayloadV1, outboxId: bigint) => Promise<void>,
    options: ReadModelOutboxDispatchOptions
 ): Promise<ReadModelOutboxDispatchResult> => {
@@ -156,7 +160,7 @@ export const dispatchReadModelOutboxBatch = async (
 };

 export const pruneDeliveredReadModelOutbox = async (
-    db: GamePrismaClient,
+    db: ReadModelOutboxDatabase,
    input: { deliveredBefore: Date; limit?: number }
 ): Promise<number> => {
    const limit = normalizeLimit(input.limit);