From 86733b99b7d920cb29829d06e143fce8213c5775 Mon Sep 17 00:00:00 2001 From: hided62 Date: Sun, 16 Aug 2026 18:01:22 +0000 Subject: [PATCH] =?UTF-8?q?feat:=20=EB=82=B4=EA=B5=AC=EC=84=B1=20=EC=9E=88?= =?UTF-8?q?=EB=8A=94=20read=20model=20=EB=B3=80=ED=99=94=20=EC=A0=80?= =?UTF-8?q?=EB=84=90=20=EA=B8=B0=EB=B0=98=EC=9D=84=20=EC=B6=94=EA=B0=80?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/architecture/realtime-change-journal.md | 4 + packages/common/src/index.ts | 1 + packages/common/src/realtime/changeJournal.ts | 155 ++++++++++++++++++ packages/common/test/changeJournal.test.ts | 86 ++++++++++ packages/infra/package.json | 6 +- packages/infra/prisma/game.prisma | 32 ++++ .../migration.sql | 39 +++++ packages/infra/prisma/migrations/README.md | 2 + packages/infra/src/index.ts | 1 + packages/infra/src/readModelChangeJournal.ts | 123 ++++++++++++++ ...readModelChangeJournal.integration.test.ts | 129 +++++++++++++++ .../infra/test/readModelChangeJournal.test.ts | 55 +++++++ packages/infra/tsconfig.json | 2 +- packages/infra/vitest.config.ts | 12 ++ pnpm-lock.yaml | 8 + tools/check-package-boundaries.mjs | 2 +- 16 files changed, 653 insertions(+), 4 deletions(-) create mode 100644 packages/common/src/realtime/changeJournal.ts create mode 100644 packages/common/test/changeJournal.test.ts create mode 100644 packages/infra/prisma/migrations/20260816000000_add_read_model_change_journal/migration.sql create mode 100644 packages/infra/src/readModelChangeJournal.ts create mode 100644 packages/infra/test/readModelChangeJournal.integration.test.ts create mode 100644 packages/infra/test/readModelChangeJournal.test.ts create mode 100644 packages/infra/vitest.config.ts diff --git a/docs/architecture/realtime-change-journal.md b/docs/architecture/realtime-change-journal.md index 5ae7b77d..000b3be8 100644 --- a/docs/architecture/realtime-change-journal.md +++ b/docs/architecture/realtime-change-journal.md @@ -169,6 +169,7 @@ statement 수와 row lock 시간을 제한한다. 없는 key의 revision은 0으 | `nation.content` | nation ID | 현재 국가 context/command/board dependency | | `world.content` | 0 | 연월, scenario/config/catalog 성격의 dependency | | `map.world` | 0 | shared base map projection | +| `map.general` | general ID | 현재 장수 이동처럼 actor별 map wake-up에 필요한 변화 | | `records.general` | general ID | 개인 최근 기록 | | `records.global` | 0 | 장수 동향 | | `records.history` | 0 | 중원 정세 | @@ -177,6 +178,9 @@ statement 수와 row lock 시간을 제한한다. 없는 key의 revision은 0으 | `front.global` | 0 | 설문과 global front status | | `access.general` | general ID | 접속 점수와 제한 상태, public fan-out 없음 | | `lobby.world` | 0 | NPC/국가 수, 공용 lobby projection | +| `lobby.general` | general ID | 본인 lobby에 보이는 이름·아이콘 projection | +| `contacts.world` | 0 | 장수 목록·외교 연락처 공용 projection | +| `reserved.general` | general ID | 장수 예약 명령 projection | | `tournament` | 0 | 토너먼트 stage/state | | `betting` | 0 | 국가/토너먼트 베팅 목록·상태 | diff --git a/packages/common/src/index.ts b/packages/common/src/index.ts index 4bc508a7..7f1dcc42 100644 --- a/packages/common/src/index.ts +++ b/packages/common/src/index.ts @@ -18,6 +18,7 @@ export * from './turnDaemon/types.js'; export * from './realtime/keys.js'; export * from './realtime/types.js'; export * from './realtime/delta.js'; +export * from './realtime/changeJournal.js'; export * from './ranking/types.js'; export * from './ranking/legacyColor.js'; export * from './auth/accountIconProjection.js'; diff --git a/packages/common/src/realtime/changeJournal.ts b/packages/common/src/realtime/changeJournal.ts new file mode 100644 index 00000000..0adee42f --- /dev/null +++ b/packages/common/src/realtime/changeJournal.ts @@ -0,0 +1,155 @@ +export const READ_MODEL_DOMAINS = [ + 'general.content', + 'city.content', + 'nation.content', + 'world.content', + 'map.world', + 'map.general', + 'records.general', + 'records.global', + 'records.history', + 'front.general', + 'front.nation', + 'front.global', + 'access.general', + 'lobby.world', + 'lobby.general', + 'contacts.world', + 'reserved.general', + 'tournament', + 'betting', +] as const; + +export type ReadModelDomain = (typeof READ_MODEL_DOMAINS)[number]; + +export interface ReadModelRevisionKey { + domain: ReadModelDomain; + entityId: number; +} + +export interface CommittedReadModelRevision extends ReadModelRevisionKey { + revision: bigint; +} + +/** + * Internal, post-commit invalidation contract. Entity IDs and durable + * revisions are intentionally retained here for server-side projection and + * viewer filtering; this value must not cross the public SSE boundary. + */ +export interface CommittedReadModelInvalidation { + revisions: readonly CommittedReadModelRevision[]; +} + +export const READ_MODEL_OUTBOX_PAYLOAD_VERSION = 1 as const; + +export type ReadModelOutboxChange = readonly [domain: ReadModelDomain, entityId: number, revision: string]; + +export interface ReadModelOutboxPayloadV1 { + version: typeof READ_MODEL_OUTBOX_PAYLOAD_VERSION; + changes: readonly ReadModelOutboxChange[]; +} + +const READ_MODEL_DOMAIN_SET: ReadonlySet = new Set(READ_MODEL_DOMAINS); + +export const isReadModelDomain = (value: string): value is ReadModelDomain => READ_MODEL_DOMAIN_SET.has(value); + +const assertRevisionKey = (key: ReadModelRevisionKey): void => { + if (!isReadModelDomain(key.domain)) { + throw new TypeError(`Unknown read-model domain: ${String(key.domain)}`); + } + if (!Number.isSafeInteger(key.entityId) || key.entityId < 0) { + throw new RangeError(`Read-model entity ID must be a non-negative safe integer: ${String(key.entityId)}`); + } +}; + +const compareRevisionKeys = (left: ReadModelRevisionKey, right: ReadModelRevisionKey): number => { + const domainOrder = left.domain < right.domain ? -1 : left.domain > right.domain ? 1 : 0; + return domainOrder === 0 ? left.entityId - right.entityId : domainOrder; +}; + +/** + * Dedupe and sort keys before acquiring revision row locks. Stable lock order + * is part of the transaction contract for concurrent writers. + */ +export const normalizeReadModelRevisionKeys = ( + keys: Iterable +): readonly ReadModelRevisionKey[] => { + const byDomain = new Map>(); + for (const key of keys) { + assertRevisionKey(key); + const entityIds = byDomain.get(key.domain) ?? new Set(); + entityIds.add(key.entityId); + byDomain.set(key.domain, entityIds); + } + + const normalized: ReadModelRevisionKey[] = []; + for (const [domain, entityIds] of byDomain) { + for (const entityId of entityIds) { + normalized.push({ domain, entityId }); + } + } + return normalized.sort(compareRevisionKeys); +}; + +export const createReadModelOutboxPayload = ( + invalidation: CommittedReadModelInvalidation +): ReadModelOutboxPayloadV1 => ({ + version: READ_MODEL_OUTBOX_PAYLOAD_VERSION, + changes: [...invalidation.revisions] + .sort(compareRevisionKeys) + .map(({ domain, entityId, revision }) => [domain, entityId, revision.toString()] as const), +}); + +/** Mutable transaction-local collector. It has no I/O and exposes snapshots. */ +export class ChangeJournal { + readonly #keys = new Map>(); + + mark(domain: ReadModelDomain, entityId = 0): this { + return this.markKey({ domain, entityId }); + } + + markKey(key: ReadModelRevisionKey): this { + assertRevisionKey(key); + const entityIds = this.#keys.get(key.domain) ?? new Set(); + entityIds.add(key.entityId); + this.#keys.set(key.domain, entityIds); + return this; + } + + markAll(keys: Iterable): this { + for (const key of keys) { + this.markKey(key); + } + return this; + } + + merge(other: ChangeJournal): this { + return this.markAll(other.snapshot()); + } + + get size(): number { + let size = 0; + for (const entityIds of this.#keys.values()) { + size += entityIds.size; + } + return size; + } + + get isEmpty(): boolean { + return this.size === 0; + } + + snapshot(): readonly ReadModelRevisionKey[] { + const keys: ReadModelRevisionKey[] = []; + for (const [domain, entityIds] of this.#keys) { + for (const entityId of entityIds) { + keys.push({ domain, entityId }); + } + } + return normalizeReadModelRevisionKeys(keys); + } + + clear(): void { + this.#keys.clear(); + } +} diff --git a/packages/common/test/changeJournal.test.ts b/packages/common/test/changeJournal.test.ts new file mode 100644 index 00000000..2629a598 --- /dev/null +++ b/packages/common/test/changeJournal.test.ts @@ -0,0 +1,86 @@ +import { describe, expect, it } from 'vitest'; + +import { + ChangeJournal, + createReadModelOutboxPayload, + normalizeReadModelRevisionKeys, + READ_MODEL_DOMAINS, + type ReadModelDomain, +} from '../src/realtime/changeJournal.js'; + +describe('ChangeJournal', () => { + it('keeps viewer-filtering semantics as distinct durable domains', () => { + expect(READ_MODEL_DOMAINS).toEqual( + expect.arrayContaining(['map.general', 'lobby.general', 'contacts.world', 'reserved.general']) + ); + }); + + it('dedupes keys and returns them in a stable lock order', () => { + const journal = new ChangeJournal() + .mark('nation.content', 4) + .mark('general.content', 9) + .mark('nation.content', 4) + .mark('general.content', 2) + .mark('map.world'); + + expect(journal.size).toBe(4); + expect(journal.snapshot()).toEqual([ + { domain: 'general.content', entityId: 2 }, + { domain: 'general.content', entityId: 9 }, + { domain: 'map.world', entityId: 0 }, + { domain: 'nation.content', entityId: 4 }, + ]); + }); + + it('merges another journal without sharing mutable state', () => { + const source = new ChangeJournal().mark('records.general', 7); + const destination = new ChangeJournal().mark('records.global').merge(source); + + source.clear(); + + expect(source.isEmpty).toBe(true); + expect(destination.snapshot()).toEqual([ + { domain: 'records.general', entityId: 7 }, + { domain: 'records.global', entityId: 0 }, + ]); + }); + + it('rejects unknown domains and unstable entity IDs at runtime', () => { + expect(() => new ChangeJournal().mark('unknown' as ReadModelDomain, 1)).toThrow(TypeError); + expect(() => new ChangeJournal().mark('general.content', -1)).toThrow(RangeError); + expect(() => new ChangeJournal().mark('general.content', Number.MAX_SAFE_INTEGER + 1)).toThrow(RangeError); + }); +}); + +describe('durable read-model invalidation contract', () => { + it('normalizes an arbitrary iterable independently of the collector', () => { + expect( + normalizeReadModelRevisionKeys([ + { domain: 'front.nation', entityId: 3 }, + { domain: 'front.global', entityId: 0 }, + { domain: 'front.nation', entityId: 3 }, + ]) + ).toEqual([ + { domain: 'front.global', entityId: 0 }, + { domain: 'front.nation', entityId: 3 }, + ]); + }); + + it('serializes bigint revisions to a compact JSON-safe payload', () => { + const payload = createReadModelOutboxPayload({ + revisions: [ + { domain: 'world.content', entityId: 0, revision: 12n }, + { domain: 'general.content', entityId: 7, revision: 4n }, + ], + }); + + expect(payload).toEqual({ + version: 1, + changes: [ + ['general.content', 7, '4'], + ['world.content', 0, '12'], + ], + }); + expect(() => JSON.stringify(payload)).not.toThrow(); + }); +}); diff --git a/packages/infra/package.json b/packages/infra/package.json index 91338992..b52a7cd0 100644 --- a/packages/infra/package.json +++ b/packages/infra/package.json @@ -10,7 +10,7 @@ "dev": "tsdown -c ../../tsdown.config.ts -F @sammo-ts/infra --watch", "lint": "eslint .", "lint:fix": "eslint . --fix", - "test": "node -e \"console.log('test not configured')\"", + "test": "vitest run --config vitest.config.ts", "prisma:generate": "pnpm prisma:generate:game && pnpm prisma:generate:gateway", "typecheck": "pnpm -w tsc7 -b packages/infra/tsconfig.json", "prisma:generate:game": "prisma generate --schema prisma/game.prisma", @@ -28,6 +28,7 @@ "@prisma/adapter-pg": "^7.9.1", "@prisma/client": "^7.9.1", "@prisma/client-runtime-utils": "^7.9.1", + "@sammo-ts/common": "workspace:*", "@sammo-ts/logic": "workspace:*", "es-toolkit": "^1.43.0", "pg": "^8.16.3", @@ -37,6 +38,7 @@ "@types/pg": "^8.21.0", "dotenv": "^17.4.2", "prisma": "^7.9.1", - "tsdown": "^0.22.14" + "tsdown": "^0.22.14", + "vitest": "^4.1.10" } } diff --git a/packages/infra/prisma/game.prisma b/packages/infra/prisma/game.prisma index 6cafa1b2..56fb603f 100644 --- a/packages/infra/prisma/game.prisma +++ b/packages/infra/prisma/game.prisma @@ -83,6 +83,38 @@ model InputEvent { @@map("input_event") } +model ReadModelRevision { + domain String + entityId Int @default(0) @map("entity_id") + revision BigInt @default(0) + updatedAt DateTime @updatedAt @map("updated_at") + + @@id([domain, entityId]) + @@map("read_model_revision") +} + +model ReadModelOutbox { + id BigInt @id @default(autoincrement()) + payload Json + attempts Int @default(0) + availableAt DateTime @default(now()) @map("available_at") + lockedAt DateTime? @map("locked_at") + lockOwner String? @map("lock_owner") + deliveredAt DateTime? @map("delivered_at") + lastError String? @map("last_error") + createdAt DateTime @default(now()) @map("created_at") + + @@index([deliveredAt, availableAt, id], map: "read_model_outbox_delivered_at_available_at_id_idx") + @@map("read_model_outbox") +} + +model ReadModelRevisionMeta { + id Int @id + coverageVersion Int @default(0) @map("coverage_version") + + @@map("read_model_revision_meta") +} + model TurnDaemonLease { profile String @id ownerId String @map("owner_id") diff --git a/packages/infra/prisma/migrations/20260816000000_add_read_model_change_journal/migration.sql b/packages/infra/prisma/migrations/20260816000000_add_read_model_change_journal/migration.sql new file mode 100644 index 00000000..4c67aadd --- /dev/null +++ b/packages/infra/prisma/migrations/20260816000000_add_read_model_change_journal/migration.sql @@ -0,0 +1,39 @@ +CREATE TABLE "read_model_revision" ( + "domain" TEXT NOT NULL, + "entity_id" INTEGER NOT NULL DEFAULT 0, + "revision" BIGINT NOT NULL DEFAULT 0, + "updated_at" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "read_model_revision_pkey" PRIMARY KEY ("domain", "entity_id"), + CONSTRAINT "read_model_revision_entity_id_check" CHECK ("entity_id" >= 0), + CONSTRAINT "read_model_revision_revision_check" CHECK ("revision" >= 0) +); + +CREATE TABLE "read_model_outbox" ( + "id" BIGSERIAL NOT NULL, + "payload" JSONB NOT NULL, + "attempts" INTEGER NOT NULL DEFAULT 0, + "available_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "locked_at" TIMESTAMP(3), + "lock_owner" TEXT, + "delivered_at" TIMESTAMP(3), + "last_error" TEXT, + "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "read_model_outbox_pkey" PRIMARY KEY ("id"), + CONSTRAINT "read_model_outbox_attempts_check" CHECK ("attempts" >= 0) +); + +CREATE INDEX "read_model_outbox_delivered_at_available_at_id_idx" + ON "read_model_outbox"("delivered_at", "available_at", "id"); + +CREATE TABLE "read_model_revision_meta" ( + "id" INTEGER NOT NULL, + "coverage_version" INTEGER NOT NULL DEFAULT 0, + + CONSTRAINT "read_model_revision_meta_pkey" PRIMARY KEY ("id"), + CONSTRAINT "read_model_revision_meta_coverage_version_check" CHECK ("coverage_version" >= 0) +); + +INSERT INTO "read_model_revision_meta" ("id", "coverage_version") +VALUES (1, 0); diff --git a/packages/infra/prisma/migrations/README.md b/packages/infra/prisma/migrations/README.md index 98ba9cb3..acc8d17e 100644 --- a/packages/infra/prisma/migrations/README.md +++ b/packages/infra/prisma/migrations/README.md @@ -27,6 +27,8 @@ chain을 적용하고 두 번째 실행은 `No pending migrations to apply`여 - `world_state`, `nation`, `city`, `general`, `message`, `troop` - `general_turn`, `nation_turn`과 revision·lease field - `input_event`, `turn_daemon_lease` +- `read_model_revision`, `read_model_outbox`, `read_model_revision_meta` +- `read_model_revision_meta.id=1`의 `coverage_version=0` - `diplomacy`, `event`, `log_entry`, `error_log` - auction, board, vote, yearbook, archive와 inheritance table - `nation.chief_general_id` diff --git a/packages/infra/src/index.ts b/packages/infra/src/index.ts index 5e24780f..4fc613c3 100644 --- a/packages/infra/src/index.ts +++ b/packages/infra/src/index.ts @@ -6,3 +6,4 @@ export type { GatewayPrismaClient } from './gatewayPrisma.js'; export * from './db.js'; export * from './redis.js'; export * from './turnEngineDb.js'; +export * from './readModelChangeJournal.js'; diff --git a/packages/infra/src/readModelChangeJournal.ts b/packages/infra/src/readModelChangeJournal.ts new file mode 100644 index 00000000..186dc14c --- /dev/null +++ b/packages/infra/src/readModelChangeJournal.ts @@ -0,0 +1,123 @@ +import { + isReadModelDomain, + normalizeReadModelRevisionKeys, + READ_MODEL_OUTBOX_PAYLOAD_VERSION, + type CommittedReadModelInvalidation, + type ReadModelRevisionKey, +} from '@sammo-ts/common'; + +import { GamePrisma } from './gamePrisma.js'; + +interface ReadModelJournalWriteRow { + domain: string; + entityId: number; + revision: bigint; + outboxId: bigint; +} + +export interface ReadModelJournalWriteResult { + invalidation: CommittedReadModelInvalidation; + /** Delivery identity only. It must not be used as a projection revision. */ + outboxId: bigint; +} + +const assertWriteRows = ( + keys: readonly ReadModelRevisionKey[], + rows: readonly ReadModelJournalWriteRow[] +): ReadModelJournalWriteResult => { + if (rows.length !== keys.length) { + throw new Error(`Read-model journal wrote ${rows.length} revisions for ${keys.length} keys.`); + } + + const first = rows[0]; + if (!first) { + throw new Error('Read-model journal write did not return an outbox ID.'); + } + const outboxId = BigInt(first.outboxId); + const revisions = rows.map((row, index) => { + const key = keys[index]; + if (!key || !isReadModelDomain(row.domain) || row.domain !== key.domain || row.entityId !== key.entityId) { + throw new Error('Read-model journal write returned revisions in an unexpected key order.'); + } + if (BigInt(row.outboxId) !== outboxId) { + throw new Error('Read-model journal write returned more than one outbox row.'); + } + return { + domain: row.domain, + entityId: row.entityId, + revision: BigInt(row.revision), + }; + }); + + return { + invalidation: { revisions }, + outboxId, + }; +}; + +/** + * Atomically increments every normalized revision key and stores exactly one + * compact outbox payload. The caller must pass the existing Prisma transaction + * that owns the domain mutation; this function never opens or commits one. + */ +export const writeReadModelChangeJournal = async ( + transaction: GamePrisma.TransactionClient, + candidates: Iterable +): Promise => { + const keys = normalizeReadModelRevisionKeys(candidates); + if (keys.length === 0) { + return null; + } + + const requestedRows = keys.map(({ domain, entityId }) => GamePrisma.sql`(${domain}::text, ${entityId}::integer)`); + const rows = await transaction.$queryRaw(GamePrisma.sql` + WITH requested("domain", "entity_id") AS ( + VALUES ${GamePrisma.join(requestedRows)} + ), + bumped AS ( + INSERT INTO "read_model_revision" ( + "domain", + "entity_id", + "revision", + "updated_at" + ) + SELECT + requested."domain", + requested."entity_id", + 1, + CURRENT_TIMESTAMP + FROM requested + ORDER BY requested."domain", requested."entity_id" + ON CONFLICT ("domain", "entity_id") DO UPDATE + SET + "revision" = "read_model_revision"."revision" + 1, + "updated_at" = CURRENT_TIMESTAMP + RETURNING "domain", "entity_id", "revision" + ), + outbox_payload AS ( + SELECT jsonb_build_object( + 'version', ${READ_MODEL_OUTBOX_PAYLOAD_VERSION}::integer, + 'changes', jsonb_agg( + jsonb_build_array("domain", "entity_id", "revision"::text) + ORDER BY "domain", "entity_id" + ) + ) AS "payload" + FROM bumped + ), + inserted_outbox AS ( + INSERT INTO "read_model_outbox" ("payload") + SELECT "payload" FROM outbox_payload + RETURNING "id" + ) + SELECT + bumped."domain", + bumped."entity_id" AS "entityId", + bumped."revision", + inserted_outbox."id" AS "outboxId" + FROM bumped + CROSS JOIN inserted_outbox + ORDER BY bumped."domain", bumped."entity_id" + `); + + return assertWriteRows(keys, rows); +}; diff --git a/packages/infra/test/readModelChangeJournal.integration.test.ts b/packages/infra/test/readModelChangeJournal.integration.test.ts new file mode 100644 index 00000000..099b7935 --- /dev/null +++ b/packages/infra/test/readModelChangeJournal.integration.test.ts @@ -0,0 +1,129 @@ +import { afterAll, beforeAll, beforeEach, describe, expect, it } from 'vitest'; + +import { ChangeJournal } from '@sammo-ts/common'; + +import { createGamePostgresConnector, type GamePrismaClient } from '../src/gamePrisma.js'; +import { writeReadModelChangeJournal } from '../src/readModelChangeJournal.js'; + +const databaseUrl = process.env.READ_MODEL_JOURNAL_DATABASE_URL; +const integration = describe.skipIf(!databaseUrl); + +integration('read-model change journal PostgreSQL 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 () => { + await disconnect?.(); + }); + + beforeEach(async () => { + await prisma.$executeRaw`TRUNCATE TABLE "read_model_outbox", "read_model_revision" RESTART IDENTITY`; + }); + + it('commits normalized revisions and one compact outbox payload', async () => { + const journal = new ChangeJournal().mark('map.world').mark('general.content', 7).mark('map.world'); + + const receipt = await prisma.$transaction((transaction) => + writeReadModelChangeJournal(transaction, journal.snapshot()) + ); + const outboxes = await prisma.readModelOutbox.findMany(); + + expect(receipt).toEqual({ + invalidation: { + revisions: [ + { domain: 'general.content', entityId: 7, revision: 1n }, + { domain: 'map.world', entityId: 0, revision: 1n }, + ], + }, + outboxId: 1n, + }); + expect(outboxes).toHaveLength(1); + expect(outboxes[0]?.payload).toEqual({ + version: 1, + changes: [ + ['general.content', 7, '1'], + ['map.world', 0, '1'], + ], + }); + }); + + it('rolls revision and outbox changes back with the owner transaction', async () => { + await expect( + prisma.$transaction(async (transaction) => { + await writeReadModelChangeJournal(transaction, [{ domain: 'nation.content', entityId: 2 }]); + throw new Error('rollback fixture'); + }) + ).rejects.toThrow('rollback fixture'); + + await expect(prisma.readModelRevision.count()).resolves.toBe(0); + await expect(prisma.readModelOutbox.count()).resolves.toBe(0); + }); + + it('does not lose increments from concurrent writers of the same key', async () => { + const writerCount = 32; + await Promise.all( + Array.from({ length: writerCount }, () => + prisma.$transaction((transaction) => + writeReadModelChangeJournal(transaction, [{ domain: 'world.content', entityId: 0 }]) + ) + ) + ); + + const revision = await prisma.readModelRevision.findUniqueOrThrow({ + where: { domain_entityId: { domain: 'world.content', entityId: 0 } }, + }); + expect(revision.revision).toBe(BigInt(writerCount)); + await expect(prisma.readModelOutbox.count()).resolves.toBe(writerCount); + }); + + it('keeps revision-first coverage disabled by default', async () => { + const meta = await prisma.readModelRevisionMeta.findUniqueOrThrow({ where: { id: 1 } }); + expect(meta.coverageVersion).toBe(0); + }); + + it('uses the Prisma-declared outbox dispatch index name', async () => { + const indexes = await prisma.$queryRaw>` + SELECT "indexname" + FROM "pg_indexes" + WHERE "schemaname" = current_schema() + AND "tablename" = 'read_model_outbox' + `; + expect(indexes.map(({ indexname }) => indexname)).toContain( + 'read_model_outbox_delivered_at_available_at_id_idx' + ); + }); + + it('rejects negative revision, entity, retry, and coverage state at the database boundary', async () => { + await expect( + prisma.readModelRevision.create({ + data: { domain: 'general.content', entityId: -1 }, + }) + ).rejects.toThrow(); + await expect( + prisma.readModelRevision.create({ + data: { domain: 'general.content', entityId: 1, revision: -1 }, + }) + ).rejects.toThrow(); + await expect( + prisma.readModelOutbox.create({ + data: { payload: {}, attempts: -1 }, + }) + ).rejects.toThrow(); + await expect( + prisma.readModelRevisionMeta.update({ + where: { id: 1 }, + data: { coverageVersion: -1 }, + }) + ).rejects.toThrow(); + }); +}); diff --git a/packages/infra/test/readModelChangeJournal.test.ts b/packages/infra/test/readModelChangeJournal.test.ts new file mode 100644 index 00000000..7d3ecfb2 --- /dev/null +++ b/packages/infra/test/readModelChangeJournal.test.ts @@ -0,0 +1,55 @@ +import { describe, expect, it, vi } from 'vitest'; + +import type { GamePrisma } from '../src/gamePrisma.js'; +import { writeReadModelChangeJournal } from '../src/readModelChangeJournal.js'; + +const createTransaction = (rows: readonly object[]) => { + const queryRaw = vi.fn().mockResolvedValue(rows); + return { + transaction: { $queryRaw: queryRaw } as unknown as GamePrisma.TransactionClient, + queryRaw, + }; +}; + +describe('writeReadModelChangeJournal', () => { + it('dedupes keys, writes one statement, and returns committed revisions', async () => { + const { transaction, queryRaw } = createTransaction([ + { domain: 'general.content', entityId: 3, revision: 8n, outboxId: 41n }, + { domain: 'map.world', entityId: 0, revision: 2n, outboxId: 41n }, + ]); + + const result = await writeReadModelChangeJournal(transaction, [ + { domain: 'map.world', entityId: 0 }, + { domain: 'general.content', entityId: 3 }, + { domain: 'map.world', entityId: 0 }, + ]); + + expect(queryRaw).toHaveBeenCalledTimes(1); + expect(result).toEqual({ + invalidation: { + revisions: [ + { domain: 'general.content', entityId: 3, revision: 8n }, + { domain: 'map.world', entityId: 0, revision: 2n }, + ], + }, + outboxId: 41n, + }); + }); + + it('does not touch the transaction for an empty journal', async () => { + const { transaction, queryRaw } = createTransaction([]); + + await expect(writeReadModelChangeJournal(transaction, [])).resolves.toBeNull(); + expect(queryRaw).not.toHaveBeenCalled(); + }); + + it('rejects a malformed database receipt instead of publishing partial state', async () => { + const { transaction } = createTransaction([ + { domain: 'general.content', entityId: 4, revision: 1n, outboxId: 1n }, + ]); + + await expect( + writeReadModelChangeJournal(transaction, [{ domain: 'general.content', entityId: 3 }]) + ).rejects.toThrow('unexpected key order'); + }); +}); diff --git a/packages/infra/tsconfig.json b/packages/infra/tsconfig.json index 847688b7..a2d789b0 100644 --- a/packages/infra/tsconfig.json +++ b/packages/infra/tsconfig.json @@ -5,5 +5,5 @@ "composite": true }, "include": ["src", "test", "*.ts"], - "references": [{ "path": "../logic" }] + "references": [{ "path": "../common" }, { "path": "../logic" }] } diff --git a/packages/infra/vitest.config.ts b/packages/infra/vitest.config.ts new file mode 100644 index 00000000..9e8e7606 --- /dev/null +++ b/packages/infra/vitest.config.ts @@ -0,0 +1,12 @@ +import { defineConfig } from 'vitest/config'; + +export default defineConfig({ + resolve: { + tsconfigPaths: true, + }, + test: { + environment: 'node', + globals: true, + include: ['test/**/*.test.ts'], + }, +}); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index b3a2e056..9707828c 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -433,6 +433,9 @@ importers: '@prisma/client-runtime-utils': specifier: ^7.9.1 version: 7.9.1 + '@sammo-ts/common': + specifier: workspace:* + version: link:../common '@sammo-ts/logic': specifier: workspace:* version: link:../logic @@ -458,6 +461,9 @@ importers: tsdown: specifier: ^0.22.14 version: 0.22.14(@volar/typescript@2.4.28(typescript@6.0.3))(tsx@4.23.12)(typescript@6.0.3)(unrun@0.2.22(@emnapi/core@1.11.2)(@emnapi/runtime@1.11.2)(synckit@0.11.13))(vue-tsc@3.3.9(typescript@6.0.3)) + vitest: + specifier: ^4.1.10 + version: 4.1.10(@types/node@26.2.0)(vite@8.2.1(@types/node@26.2.0)(esbuild@0.28.2)(jiti@2.7.0)(tsx@4.23.12)) packages/logic: dependencies: @@ -555,6 +561,8 @@ importers: specifier: ^4.1.10 version: 4.1.10(@types/node@26.2.0)(vite@8.2.1(@types/node@26.2.0)(esbuild@0.28.2)(jiti@2.7.0)(tsx@4.23.12)) + tools/load-tests: {} + packages: '@algolia/abtesting@1.22.0': diff --git a/tools/check-package-boundaries.mjs b/tools/check-package-boundaries.mjs index 7e8dd1ad..a835c256 100644 --- a/tools/check-package-boundaries.mjs +++ b/tools/check-package-boundaries.mjs @@ -8,7 +8,7 @@ const workspaceRoot = path.resolve(path.dirname(fileURLToPath(import.meta.url)), const zones = [ { name: 'common', root: 'packages/common/src', allowed: [] }, { name: 'logic', root: 'packages/logic/src', allowed: ['@sammo-ts/common'] }, - { name: 'infra', root: 'packages/infra/src', allowed: ['@sammo-ts/logic'] }, + { name: 'infra', root: 'packages/infra/src', allowed: ['@sammo-ts/common', '@sammo-ts/logic'] }, { name: 'game-engine', root: 'app/game-engine/src',