From b41b161e9ca13581e6fbb23e54f817055b9dda42 Mon Sep 17 00:00:00 2001 From: hided62 Date: Sun, 16 Aug 2026 18:18:08 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20=EC=97=94=EC=A7=84=20=EB=B3=80=ED=99=94?= =?UTF-8?q?=20=EC=A0=80=EB=84=90=EC=9D=84=20=EC=83=81=ED=83=9C=20=EC=BB=A4?= =?UTF-8?q?=EB=B0=8B=EA=B3=BC=20=EC=9B=90=EC=9E=90=ED=99=94?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 최종 read model projection과 저장된 로그를 소유 transaction 안에서 판정하고 revision 및 outbox와 함께 커밋한다. clock 전용 변화와 rollback은 revision을 남기지 않으며 커밋 receipt를 유실 없이 큐잉한다. --- app/game-engine/src/turn/databaseHooks.ts | 198 +++++++++++-- app/game-engine/src/turn/turnDaemon.ts | 21 +- ...angeJournalPersistence.integration.test.ts | 267 ++++++++++++++++++ .../test/realtimeReadModelChanges.test.ts | 94 ++++++ 4 files changed, 546 insertions(+), 34 deletions(-) create mode 100644 app/game-engine/test/readModelChangeJournalPersistence.integration.test.ts diff --git a/app/game-engine/src/turn/databaseHooks.ts b/app/game-engine/src/turn/databaseHooks.ts index 64948f22..bbc99cc6 100644 --- a/app/game-engine/src/turn/databaseHooks.ts +++ b/app/game-engine/src/turn/databaseHooks.ts @@ -1,7 +1,9 @@ import { createGamePostgresConnector, GamePrisma, + writeReadModelChangeJournal, type InputJsonValue, + type ReadModelJournalWriteResult, type TurnEngineCityUpdateInput, type TurnEngineDiplomacyCreateManyInput, type TurnEngineDiplomacyUpdateInput, @@ -24,7 +26,13 @@ import { type MessageRecordDraft, type Nation, } from '@sammo-ts/logic'; -import { asRecord, type RealtimeReadModelChanges } from '@sammo-ts/common'; +import { + asRecord, + ChangeJournal, + type CommittedReadModelInvalidation, + type ReadModelDomain, + type RealtimeReadModelChanges, +} from '@sammo-ts/common'; import type { TurnDaemonCommandResult, TurnDaemonHooks } from '../lifecycle/types.js'; import type { InMemoryTurnWorld, TurnWorldChanges } from './inMemoryWorld.js'; @@ -44,9 +52,17 @@ import { persistYearbookSnapshot } from './yearbookPersistence.js'; export interface DatabaseTurnHooks { hooks: TurnDaemonHooks; takeCommittedReadModelChanges(): RealtimeReadModelChanges | null; + takeCommittedReadModelChangeReceipt(): CommittedReadModelChangeReceipt | null; close(): Promise; } +export interface CommittedReadModelChangeReceipt { + /** Delivery identity only; this is not a projection revision. */ + outboxId: bigint; + invalidation: CommittedReadModelInvalidation; + changes: RealtimeReadModelChanges; +} + const uniqueSortedIds = (values: Iterable): number[] => [...new Set(values)] .filter((value) => Number.isSafeInteger(value) && value > 0) @@ -94,6 +110,50 @@ const canonicalizeReadModelValue = (value: unknown): unknown => { const signature = (value: unknown): string => JSON.stringify(canonicalizeReadModelValue(value)) ?? 'undefined'; +const CLOCK_ONLY_WORLD_META_KEYS = new Set([ + 'clockBaseTime', + 'clock_base_time', + 'clockMode', + 'clock_mode', + 'clockTick', + 'clock_tick', + 'clockWallAnchor', + 'clock_wall_anchor', + 'heartbeat', + 'heartbeatAt', + 'heartbeat_at', + 'lastExecuted', + 'last_executed', + 'lastTurnTick', + 'last_turn_tick', + 'lastTurnTime', + 'last_turn_time', + 'lease', + 'leaseOwner', + 'lease_owner', + 'leaseUntil', + 'lease_until', +]); + +const projectWorldMeta = (meta: Record): Record => + Object.fromEntries(Object.entries(meta).filter(([key]) => !CLOCK_ONLY_WORLD_META_KEYS.has(key))); + +/** + * Canonical fields consumed by world-scoped screens. Moving clock cursors and + * lease/heartbeat bookkeeping are deliberately absent, while turn term, + * calendar, scenario config, and gameplay meta remain visible. + */ +export const createWorldReadModelSignature = (world: InMemoryTurnWorld): string => { + const state = world.getState(); + return signature({ + currentYear: state.currentYear, + currentMonth: state.currentMonth, + tickSeconds: state.tickSeconds, + config: world.getScenarioConfig(), + meta: projectWorldMeta(state.meta), + }); +}; + const generalSignatures = (general: TurnGeneral): ReadModelSignatures => ({ content: signature(general), map: signature({ cityId: general.cityId, nationId: general.nationId }), @@ -351,6 +411,58 @@ export const mergePersistedVisibleLogChanges = ( rows.some((entry) => entry.scope === LogScope.SYSTEM && entry.category === LogCategory.HISTORY), }); +const markIds = (journal: ChangeJournal, domain: ReadModelDomain, ids: readonly number[]): void => { + for (const id of ids) { + journal.mark(domain, id); + } +}; + +/** + * Converts the final legacy internal invalidation into durable domain keys. + * City/nation map changes are shared-map changes; general movement remains + * actor-targeted. General name/nation changes affect the global online list, + * while frontStatusActorIds is the private actor projection. + */ +export const createReadModelChangeJournal = (changes: RealtimeReadModelChanges): ChangeJournal => { + const journal = new ChangeJournal(); + markIds(journal, 'general.content', changes.generalIds); + markIds(journal, 'city.content', changes.cityIds); + markIds(journal, 'nation.content', changes.nationIds); + markIds(journal, 'map.general', changes.mapGeneralIds ?? []); + markIds(journal, 'front.nation', changes.frontStatusNationIds ?? []); + markIds(journal, 'front.general', changes.frontStatusActorIds ?? []); + markIds(journal, 'lobby.general', changes.lobbyGeneralIds ?? []); + markIds(journal, 'reserved.general', changes.reservedGeneralIds); + markIds(journal, 'records.general', changes.recordGeneralIds); + + if (changes.worldChanged) { + journal.mark('world.content').mark('map.world'); + } + if ( + changes.mapChanged || + (changes.mapCityIds ?? []).length > 0 || + (changes.mapNationIds ?? []).length > 0 + ) { + journal.mark('map.world'); + } + if ((changes.frontStatusGeneralIds ?? []).length > 0 || changes.frontStatusChanged) { + journal.mark('front.global'); + } + if (changes.globalRecordsChanged) { + journal.mark('records.global'); + } + if (changes.worldHistoryChanged) { + journal.mark('records.history'); + } + if (changes.contactsChanged) { + journal.mark('contacts.world'); + } + if (changes.lobbyChanged) { + journal.mark('lobby.world'); + } + return journal; +}; + export const excludeDeletedReservedTurnQueues = ( changes: ReservedTurnChanges, deletedGeneralIds: readonly number[], @@ -901,14 +1013,42 @@ export const createDatabaseTurnHooks = async ( // closely. A populated season can finish the turn but expire while flushing // it, which rolls the transaction back and marks the profile PAUSED. const transactionOptions = { timeout: options?.transactionTimeoutMs ?? 30_000 }; - let committedReadModelChanges: RealtimeReadModelChanges | null = null; const readModelBaseline = createRealtimeReadModelBaseline(world); + let worldReadModelBaseline = createWorldReadModelSignature(world); + const committedReceipts = new Map(); + + const enqueueCommittedReceipt = ( + changes: RealtimeReadModelChanges, + journalWrite: ReadModelJournalWriteResult | null + ): void => { + if (!journalWrite) { + return; + } + committedReceipts.set(journalWrite.outboxId, { + outboxId: journalWrite.outboxId, + invalidation: journalWrite.invalidation, + changes, + }); + }; + const takeCommittedReceipt = (): CommittedReadModelChangeReceipt | null => { + const next = committedReceipts.entries().next(); + if (next.done) { + return null; + } + const [outboxId, receipt] = next.value; + committedReceipts.delete(outboxId); + return receipt; + }; const persistChanges = async ( transaction?: GamePrisma.TransactionClient, commandCompletion?: { requestId: string; result: TurnDaemonCommandResult }, directLogFloor?: number - ): Promise<{ acknowledge: () => void; readModelChanges: RealtimeReadModelChanges }> => { + ): Promise<{ + acknowledge: () => void; + readModelChanges: RealtimeReadModelChanges; + journalWrite: ReadModelJournalWriteResult | null; + }> => { const state = world.getState(); const changes = world.peekDirtyState(); let persistedVisibleLogs: PersistedVisibleLogRow[] = []; @@ -956,7 +1096,13 @@ export const createDatabaseTurnHooks = async ( lastTurnTick: BigInt(state.lastTurnTick ?? world.dateToGameTick(state.lastTurnTime)), meta: asJson(state.meta), }; - const persist = async (prisma: GamePrisma.TransactionClient): Promise => { + const persist = async ( + prisma: GamePrisma.TransactionClient + ): Promise<{ + readModelChanges: RealtimeReadModelChanges; + journalWrite: ReadModelJournalWriteResult | null; + worldReadModelSignature: string; + }> => { visibleLogFloor ??= ( await prisma.logEntry.findFirst({ @@ -1399,17 +1545,26 @@ export const createDatabaseTurnHooks = async ( orderBy: { id: 'asc' }, select: { id: true, scope: true, category: true, generalId: true }, }); - }; - if (transaction) { - await persist(transaction); - } else { - await prisma.$transaction(persist, transactionOptions); - } - const readModelChanges = mergePersistedVisibleLogChanges( - summarizeRealtimeReadModelChanges(changes, persistedReservedTurnChanges, readModelBaseline), - persistedVisibleLogs - ); + const worldReadModelSignature = createWorldReadModelSignature(world); + const readModelChanges = mergePersistedVisibleLogChanges( + summarizeRealtimeReadModelChanges(changes, persistedReservedTurnChanges, readModelBaseline), + persistedVisibleLogs + ); + if (worldReadModelSignature !== worldReadModelBaseline) { + readModelChanges.worldChanged = true; + } + const journal = createReadModelChangeJournal(readModelChanges); + markIds(journal, 'access.general', accessScoreResetGeneralIds); + if (pendingNationBettingOpens.length > 0 || pendingNationBettingFinishes.length > 0) { + journal.mark('betting'); + } + const journalWrite = await writeReadModelChangeJournal(prisma, journal.snapshot()); + return { readModelChanges, journalWrite, worldReadModelSignature }; + }; + const persisted = transaction + ? await persist(transaction) + : await prisma.$transaction(persist, transactionOptions); return { acknowledge: () => { world.acknowledgeDirtyState(changes); @@ -1417,8 +1572,10 @@ export const createDatabaseTurnHooks = async ( options.reservedTurns.acknowledgeDirtyState(reservedTurnChanges); } applyRealtimeReadModelBaseline(readModelBaseline, changes); + worldReadModelBaseline = persisted.worldReadModelSignature; }, - readModelChanges, + readModelChanges: persisted.readModelChanges, + journalWrite: persisted.journalWrite, }; }; @@ -1426,12 +1583,12 @@ export const createDatabaseTurnHooks = async ( flushChanges: async () => { const committed = await persistChanges(); committed.acknowledge(); - committedReadModelChanges = committed.readModelChanges; + enqueueCommittedReceipt(committed.readModelChanges, committed.journalWrite); }, commitCommand: async (requestId, result) => { const committed = await persistChanges(undefined, { requestId, result }); committed.acknowledge(); - committedReadModelChanges = committed.readModelChanges; + enqueueCommittedReceipt(committed.readModelChanges, committed.journalWrite); }, executeCommand: async (requestId, execute) => { const committed = await prisma.$transaction(async (transaction) => { @@ -1447,7 +1604,7 @@ export const createDatabaseTurnHooks = async ( return { result, persisted }; }, transactionOptions); committed.persisted.acknowledge(); - committedReadModelChanges = committed.persisted.readModelChanges; + enqueueCommittedReceipt(committed.persisted.readModelChanges, committed.persisted.journalWrite); return committed.result; }, }; @@ -1455,10 +1612,9 @@ export const createDatabaseTurnHooks = async ( return { hooks, takeCommittedReadModelChanges: () => { - const changes = committedReadModelChanges; - committedReadModelChanges = null; - return changes; + return takeCommittedReceipt()?.changes ?? null; }, + takeCommittedReadModelChangeReceipt: takeCommittedReceipt, close: () => connector.disconnect(), }; }; diff --git a/app/game-engine/src/turn/turnDaemon.ts b/app/game-engine/src/turn/turnDaemon.ts index ac1bd723..ae30b141 100644 --- a/app/game-engine/src/turn/turnDaemon.ts +++ b/app/game-engine/src/turn/turnDaemon.ts @@ -20,7 +20,7 @@ import type { Clock, TurnDaemonControlQueue, TurnDaemonHooks, TurnRunBudget } fr import { TurnDaemonLifecycle } from '../lifecycle/turnDaemonLifecycle.js'; import { DatabaseTurnDaemonCommandQueue } from '../lifecycle/databaseCommandQueue.js'; import type { MapLoaderOptions } from '../scenario/mapLoader.js'; -import { createDatabaseTurnHooks } from './databaseHooks.js'; +import { createDatabaseTurnHooks, type CommittedReadModelChangeReceipt } from './databaseHooks.js'; import type { GeneralTurnHandler, InMemoryTurnWorldOptions, TurnCalendarHandler } from './inMemoryWorld.js'; import { InMemoryTurnWorld } from './inMemoryWorld.js'; import { InMemoryTurnProcessor } from './inMemoryTurnProcessor.js'; @@ -488,7 +488,7 @@ const createRealtimeRuntime = async (options: { redisUrl?: string; profileName: string; hooks?: TurnDaemonHooks; - takeCommittedReadModelChanges: (() => RealtimeReadModelChanges | null) | null; + takeCommittedReadModelChangeReceipt: (() => CommittedReadModelChangeReceipt | null) | null; }): Promise<{ redisConnector: RedisConnector | null; hooks?: TurnDaemonHooks }> => { const redisConfig = resolveRedisConfig(options.redisUrl); if (!redisConfig) { @@ -526,10 +526,8 @@ const createRealtimeRuntime = async (options: { ...options.hooks, publishEvents: async (result) => { try { - const changes = options.takeCommittedReadModelChanges?.() ?? createEmptyRealtimeReadModelChanges(); - if (result.processedTurns > 0) { - changes.worldChanged = true; - } + const changes = + options.takeCommittedReadModelChangeReceipt?.()?.changes ?? createEmptyRealtimeReadModelChanges(); const revision = await publishCommittedChanges(changes); await publishRealtimeEvent({ type: 'turnCompleted', @@ -545,10 +543,7 @@ const createRealtimeRuntime = async (options: { }, publishCommandEvents: async (result) => { try { - const changes = options.takeCommittedReadModelChanges?.(); - if (changes && result.type === 'shiftSchedule' && result.ok) { - changes.lobbyChanged = true; - } + const changes = options.takeCommittedReadModelChangeReceipt?.()?.changes; if (changes && hasRealtimeReadModelChanges(changes)) { const revision = await publishCommittedChanges(changes); if (revision !== undefined) { @@ -814,7 +809,7 @@ const createTurnDaemonRuntimeWithLease = async ( const controlQueue = options.controlQueue ?? new InMemoryControlQueue(); let hooks: TurnDaemonHooks | undefined; - let takeCommittedReadModelChanges: (() => RealtimeReadModelChanges | null) | null = null; + let takeCommittedReadModelChangeReceipt: (() => CommittedReadModelChangeReceipt | null) | null = null; let close = async () => {}; let auctionFinalizer: Awaited> | null = null; let auctionBidder: Awaited> | null = null; @@ -858,7 +853,7 @@ const createTurnDaemonRuntimeWithLease = async ( await gatewayGate?.markPaused(error); }, }; - takeCommittedReadModelChanges = dbHooks.takeCommittedReadModelChanges; + takeCommittedReadModelChangeReceipt = dbHooks.takeCommittedReadModelChangeReceipt; close = async () => { if (auctionBidder) { await auctionBidder.close(); @@ -908,7 +903,7 @@ const createTurnDaemonRuntimeWithLease = async ( redisUrl: options.redisUrl, profileName: options.profileName ?? options.profile, hooks, - takeCommittedReadModelChanges, + takeCommittedReadModelChangeReceipt, }); redisConnector = realtimeRuntime.redisConnector; hooks = realtimeRuntime.hooks; diff --git a/app/game-engine/test/readModelChangeJournalPersistence.integration.test.ts b/app/game-engine/test/readModelChangeJournalPersistence.integration.test.ts new file mode 100644 index 00000000..eeb686aa --- /dev/null +++ b/app/game-engine/test/readModelChangeJournalPersistence.integration.test.ts @@ -0,0 +1,267 @@ +import { afterAll, beforeAll, describe, expect, it } from 'vitest'; + +import type { TurnDaemonCommandResult, TurnRunResult } from '@sammo-ts/common'; +import { createGamePostgresConnector, type GamePrisma, type GamePrismaClient } from '@sammo-ts/infra'; +import { LogCategory, LogFormat, LogScope, type MapDefinition, type ScenarioConfig } from '@sammo-ts/logic'; + +import { createDatabaseTurnHooks, type DatabaseTurnHooks } from '../src/turn/databaseHooks.js'; +import { InMemoryTurnWorld } from '../src/turn/inMemoryWorld.js'; +import type { TurnWorldSnapshot, TurnWorldState } from '../src/turn/types.js'; + +const databaseUrl = process.env.READ_MODEL_JOURNAL_DATABASE_URL; +const integration = describe.skipIf(!databaseUrl); +const worldId = 991_816; +const directLogGeneralId = 991_817; +const requestId = 'integration:engine:read-model-journal:direct-logs'; +const rollbackConstraint = 'read_model_outbox_engine_rollback_test'; + +const assertDedicatedDatabase = (rawUrl: string): void => { + const schema = new URL(rawUrl).searchParams.get('schema'); + if (!schema?.endsWith('read_model_journal_integration')) { + throw new Error(`Refusing to mutate non-dedicated schema: ${schema ?? '(missing)'}`); + } +}; + +const scenarioConfig: ScenarioConfig = { + stat: { total: 300, min: 10, max: 100, npcTotal: 150, npcMax: 50, npcMin: 10, chiefMin: 65 }, + iconPath: '', + map: {}, + const: { feature: 'stable' }, + environment: { mapName: 'che', unitSet: 'che' }, +}; + +const map: MapDefinition = { + id: 'read-model-journal-integration', + name: 'read-model journal integration', + cities: [], +}; + +const initialState: TurnWorldState = { + id: worldId, + currentYear: 190, + currentMonth: 1, + tickSeconds: 600, + lastTurnTime: new Date('2026-08-16T00:00:00.000Z'), + meta: { + lastTurnTime: '2026-08-16T00:00:00.000Z', + scenarioMeta: { + title: 'read-model journal integration', + startYear: 190, + life: null, + fiction: null, + history: [], + ignoreDefaultEvents: false, + }, + }, +}; + +const turnRunResult = (world: InMemoryTurnWorld): TurnRunResult => ({ + lastTurnTime: world.getState().lastTurnTime.toISOString(), + processedGenerals: 0, + processedTurns: 0, + durationMs: 0, + partial: false, +}); + +integration('game-engine read-model journal PostgreSQL transaction', () => { + let db: GamePrismaClient; + let disconnect: (() => Promise) | undefined; + let hooks: DatabaseTurnHooks | undefined; + + beforeAll(async () => { + if (!databaseUrl) { + throw new Error('READ_MODEL_JOURNAL_DATABASE_URL is required.'); + } + assertDedicatedDatabase(databaseUrl); + const connector = createGamePostgresConnector({ url: databaseUrl }); + db = connector.prisma; + disconnect = connector.disconnect; + await connector.connect(); + + await db.$executeRawUnsafe(`ALTER TABLE read_model_outbox DROP CONSTRAINT IF EXISTS ${rollbackConstraint}`); + await db.inputEvent.deleteMany({ where: { requestId } }); + await db.logEntry.deleteMany({ where: { text: { startsWith: '[read-model-journal]' } } }); + await db.worldState.deleteMany({ where: { id: worldId } }); + await db.$executeRaw`TRUNCATE TABLE "read_model_outbox", "read_model_revision" RESTART IDENTITY`; + await db.worldState.create({ + data: { + id: worldId, + scenarioCode: 'read-model-journal-integration', + currentYear: initialState.currentYear, + currentMonth: initialState.currentMonth, + tickSeconds: initialState.tickSeconds, + config: JSON.parse(JSON.stringify(scenarioConfig)) as GamePrisma.InputJsonValue, + meta: initialState.meta as GamePrisma.InputJsonValue, + }, + }); + }); + + afterAll(async () => { + await hooks?.close(); + if (db) { + await db.$executeRawUnsafe( + `ALTER TABLE read_model_outbox DROP CONSTRAINT IF EXISTS ${rollbackConstraint}` + ); + await db.inputEvent.deleteMany({ where: { requestId } }); + await db.logEntry.deleteMany({ where: { text: { startsWith: '[read-model-journal]' } } }); + await db.worldState.deleteMany({ where: { id: worldId } }); + await db.$executeRaw`TRUNCATE TABLE "read_model_outbox", "read_model_revision" RESTART IDENTITY`; + } + await disconnect?.(); + }); + + it('commits visible state/log domains atomically and excludes clock-only or rolled-back work', async () => { + const snapshot: TurnWorldSnapshot = { + generals: [], + cities: [], + nations: [], + troops: [], + diplomacy: [], + events: [], + initialEvents: [], + scenarioConfig, + scenarioMeta: initialState.meta.scenarioMeta as TurnWorldSnapshot['scenarioMeta'], + map, + }; + const world = new InMemoryTurnWorld(initialState, snapshot, { + schedule: { entries: [{ startMinute: 0, tickMinutes: 10 }] }, + }); + hooks = await createDatabaseTurnHooks(databaseUrl!, world); + + world.setLastTurnTime(new Date('2026-08-16T00:10:00.000Z')); + await hooks.hooks.flushChanges?.(turnRunResult(world)); + expect(hooks.takeCommittedReadModelChangeReceipt()).toBeNull(); + await hooks.hooks.flushChanges?.(turnRunResult(world)); + expect(hooks.takeCommittedReadModelChangeReceipt()).toBeNull(); + await expect(db.readModelRevision.count()).resolves.toBe(0); + await expect(db.readModelOutbox.count()).resolves.toBe(0); + + await db.$executeRawUnsafe(` + ALTER TABLE read_model_outbox + ADD CONSTRAINT ${rollbackConstraint} + CHECK ((payload->>'version')::integer <> 1) + `); + world.updateWorldMeta({ durableFixture: 'must-rollback' }); + world.pushLog({ + scope: LogScope.SYSTEM, + category: LogCategory.SUMMARY, + format: LogFormat.RAWTEXT, + text: '[read-model-journal] rollback', + }); + await expect(hooks.hooks.flushChanges?.(turnRunResult(world))).rejects.toThrow(rollbackConstraint); + expect(hooks.takeCommittedReadModelChangeReceipt()).toBeNull(); + await expect(db.logEntry.count({ where: { text: '[read-model-journal] rollback' } })).resolves.toBe(0); + await expect(db.readModelRevision.count()).resolves.toBe(0); + await expect(db.readModelOutbox.count()).resolves.toBe(0); + expect((await db.worldState.findUniqueOrThrow({ where: { id: worldId } })).meta).not.toMatchObject({ + durableFixture: 'must-rollback', + }); + + await db.$executeRawUnsafe(`ALTER TABLE read_model_outbox DROP CONSTRAINT ${rollbackConstraint}`); + await hooks.hooks.flushChanges?.(turnRunResult(world)); + const stateAndLogReceipt = hooks.takeCommittedReadModelChangeReceipt(); + expect(stateAndLogReceipt?.changes).toMatchObject({ + worldChanged: true, + globalRecordsChanged: true, + }); + expect(stateAndLogReceipt?.invalidation.revisions).toEqual([ + { domain: 'map.world', entityId: 0, revision: 1n }, + { domain: 'records.global', entityId: 0, revision: 1n }, + { domain: 'world.content', entityId: 0, revision: 1n }, + ]); + await expect(db.logEntry.count({ where: { text: '[read-model-journal] rollback' } })).resolves.toBe(1); + await expect(db.readModelOutbox.count()).resolves.toBe(1); + + await world.advanceMonth(new Date('2026-08-16T00:20:00.000Z')); + await hooks.hooks.flushChanges?.(turnRunResult(world)); + const monthReceipt = hooks.takeCommittedReadModelChangeReceipt(); + expect(monthReceipt?.changes.worldChanged).toBe(true); + expect(monthReceipt?.invalidation.revisions).toEqual([ + { domain: 'map.world', entityId: 0, revision: 2n }, + { domain: 'world.content', entityId: 0, revision: 2n }, + ]); + await expect(db.worldState.findUniqueOrThrow({ where: { id: worldId } })).resolves.toMatchObject({ + currentYear: 190, + currentMonth: 2, + }); + + await db.inputEvent.create({ + data: { + requestId, + target: 'ENGINE', + eventType: 'shiftSchedule', + status: 'PROCESSING', + payload: {}, + }, + }); + const directResult: TurnDaemonCommandResult = { + type: 'shiftSchedule', + ok: true, + actionId: 'read-model-journal-direct-log', + deltaMinutes: 0, + lastTurnTime: world.getState().lastTurnTime.toISOString(), + shiftedGenerals: 0, + shiftedAuctions: 0, + }; + await hooks.hooks.executeCommand?.(requestId, async ({ db: transaction }) => { + if (!transaction) { + throw new Error('Expected the direct-log command transaction.'); + } + await transaction.logEntry.createMany({ + data: [ + { + scope: LogScope.GENERAL, + category: LogCategory.ACTION, + year: 190, + month: 2, + text: '[read-model-journal] direct general', + generalId: directLogGeneralId, + }, + { + scope: LogScope.SYSTEM, + category: LogCategory.SUMMARY, + year: 190, + month: 2, + text: '[read-model-journal] direct global', + }, + { + scope: LogScope.SYSTEM, + category: LogCategory.HISTORY, + year: 190, + month: 2, + text: '[read-model-journal] direct history', + }, + ], + }); + return directResult; + }); + const directLogReceipt = hooks.takeCommittedReadModelChangeReceipt(); + expect(directLogReceipt?.changes).toMatchObject({ + recordGeneralIds: [directLogGeneralId], + globalRecordsChanged: true, + worldHistoryChanged: true, + worldChanged: false, + }); + expect(directLogReceipt?.invalidation.revisions).toEqual([ + { domain: 'records.general', entityId: directLogGeneralId, revision: 1n }, + { domain: 'records.global', entityId: 0, revision: 2n }, + { domain: 'records.history', entityId: 0, revision: 1n }, + ]); + await expect(db.readModelOutbox.count()).resolves.toBe(3); + await expect(db.logEntry.count({ where: { text: { startsWith: '[read-model-journal] direct' } } })).resolves.toBe( + 3 + ); + + world.updateWorldMeta({ queueProbe: 1 }); + await hooks.hooks.flushChanges?.(turnRunResult(world)); + world.updateWorldMeta({ queueProbe: 2 }); + await hooks.hooks.flushChanges?.(turnRunResult(world)); + const firstQueuedReceipt = hooks.takeCommittedReadModelChangeReceipt(); + const secondQueuedReceipt = hooks.takeCommittedReadModelChangeReceipt(); + expect(firstQueuedReceipt?.outboxId).toBeLessThan(secondQueuedReceipt?.outboxId ?? 0n); + expect(firstQueuedReceipt?.changes.worldChanged).toBe(true); + expect(secondQueuedReceipt?.changes.worldChanged).toBe(true); + expect(hooks.takeCommittedReadModelChangeReceipt()).toBeNull(); + await expect(db.readModelOutbox.count()).resolves.toBe(5); + }); +}); diff --git a/app/game-engine/test/realtimeReadModelChanges.test.ts b/app/game-engine/test/realtimeReadModelChanges.test.ts index 3aa4b27e..83f07bbc 100644 --- a/app/game-engine/test/realtimeReadModelChanges.test.ts +++ b/app/game-engine/test/realtimeReadModelChanges.test.ts @@ -4,7 +4,9 @@ import { LogCategory, LogFormat, LogScope } from '@sammo-ts/logic'; import { applyRealtimeReadModelBaseline, + createReadModelChangeJournal, createRealtimeReadModelBaseline, + createWorldReadModelSignature, mergePersistedVisibleLogChanges, summarizeRealtimeReadModelChanges, } from '../src/turn/databaseHooks.js'; @@ -12,6 +14,98 @@ import type { InMemoryTurnWorld } from '../src/turn/inMemoryWorld.js'; import type { TurnWorldChanges } from '../src/turn/inMemoryWorld.js'; import type { ReservedTurnChanges } from '../src/turn/reservedTurnStore.js'; +describe('durable read-model change journal mapping', () => { + it('maps every final engine invalidation to its precise durable domain', () => { + const changes = { + ...createEmptyRealtimeReadModelChanges(), + generalIds: [7], + cityIds: [3], + nationIds: [2], + mapGeneralIds: [8], + mapCityIds: [4], + mapNationIds: [5], + frontStatusGeneralIds: [9], + frontStatusNationIds: [2], + frontStatusActorIds: [7], + lobbyGeneralIds: [7], + reservedGeneralIds: [7], + recordGeneralIds: [7], + worldChanged: true, + globalRecordsChanged: true, + worldHistoryChanged: true, + contactsChanged: true, + frontStatusChanged: true, + mapChanged: true, + lobbyChanged: true, + }; + + expect(createReadModelChangeJournal(changes).snapshot()).toEqual([ + { domain: 'city.content', entityId: 3 }, + { domain: 'contacts.world', entityId: 0 }, + { domain: 'front.general', entityId: 7 }, + { domain: 'front.global', entityId: 0 }, + { domain: 'front.nation', entityId: 2 }, + { domain: 'general.content', entityId: 7 }, + { domain: 'lobby.general', entityId: 7 }, + { domain: 'lobby.world', entityId: 0 }, + { domain: 'map.general', entityId: 8 }, + { domain: 'map.world', entityId: 0 }, + { domain: 'nation.content', entityId: 2 }, + { domain: 'records.general', entityId: 7 }, + { domain: 'records.global', entityId: 0 }, + { domain: 'records.history', entityId: 0 }, + { domain: 'reserved.general', entityId: 7 }, + { domain: 'world.content', entityId: 0 }, + ]); + }); + + it('ignores moving clock and lease metadata but detects month, turn-term, config, and gameplay meta', () => { + const state = { + currentYear: 190, + currentMonth: 1, + tickSeconds: 600, + lastTurnTime: new Date('2026-08-16T00:00:00.000Z'), + clockBaseTime: new Date('2026-08-16T00:00:00.000Z'), + clockTick: 1, + clockMode: 'realtime' as const, + clockWallAnchor: new Date('2026-08-16T00:00:00.000Z'), + lastTurnTick: 1, + meta: { + lastTurnTime: '2026-08-16T00:00:00.000Z', + heartbeatAt: '2026-08-16T00:00:00.000Z', + leaseUntil: '2026-08-16T00:01:00.000Z', + killturn: 24, + }, + }; + const config = { const: { feature: 'old' } }; + const world = { + getState: () => ({ ...state, meta: { ...state.meta } }), + getScenarioConfig: () => config, + } as unknown as InMemoryTurnWorld; + const baseline = createWorldReadModelSignature(world); + + state.lastTurnTime = new Date('2026-08-16T00:05:00.000Z'); + state.clockTick = 2; + state.lastTurnTick = 2; + state.meta.lastTurnTime = '2026-08-16T00:05:00.000Z'; + state.meta.heartbeatAt = '2026-08-16T00:00:10.000Z'; + state.meta.leaseUntil = '2026-08-16T00:01:10.000Z'; + expect(createWorldReadModelSignature(world)).toBe(baseline); + + state.currentMonth = 2; + expect(createWorldReadModelSignature(world)).not.toBe(baseline); + state.currentMonth = 1; + state.tickSeconds = 300; + expect(createWorldReadModelSignature(world)).not.toBe(baseline); + state.tickSeconds = 600; + config.const.feature = 'new'; + expect(createWorldReadModelSignature(world)).not.toBe(baseline); + config.const.feature = 'old'; + state.meta.killturn = 25; + expect(createWorldReadModelSignature(world)).not.toBe(baseline); + }); +}); + describe('summarizeRealtimeReadModelChanges', () => { it('classifies the committed log rows even when they bypass in-memory log drafts', () => { expect(