feat: 내구성 있는 read model 변화 저널 기반을 추가
This commit is contained in:
@@ -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 | 국가/토너먼트 베팅 목록·상태 |
|
||||
|
||||
|
||||
@@ -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';
|
||||
|
||||
@@ -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<string> = 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<ReadModelRevisionKey>
|
||||
): readonly ReadModelRevisionKey[] => {
|
||||
const byDomain = new Map<ReadModelDomain, Set<number>>();
|
||||
for (const key of keys) {
|
||||
assertRevisionKey(key);
|
||||
const entityIds = byDomain.get(key.domain) ?? new Set<number>();
|
||||
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<ReadModelDomain, Set<number>>();
|
||||
|
||||
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<number>();
|
||||
entityIds.add(key.entityId);
|
||||
this.#keys.set(key.domain, entityIds);
|
||||
return this;
|
||||
}
|
||||
|
||||
markAll(keys: Iterable<ReadModelRevisionKey>): 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();
|
||||
}
|
||||
}
|
||||
@@ -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();
|
||||
});
|
||||
});
|
||||
@@ -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"
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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")
|
||||
|
||||
+39
@@ -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);
|
||||
@@ -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`
|
||||
|
||||
@@ -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';
|
||||
|
||||
@@ -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<ReadModelRevisionKey>
|
||||
): Promise<ReadModelJournalWriteResult | null> => {
|
||||
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<ReadModelJournalWriteRow[]>(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);
|
||||
};
|
||||
@@ -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<void>) | 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<Array<{ indexname: string }>>`
|
||||
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();
|
||||
});
|
||||
});
|
||||
@@ -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');
|
||||
});
|
||||
});
|
||||
@@ -5,5 +5,5 @@
|
||||
"composite": true
|
||||
},
|
||||
"include": ["src", "test", "*.ts"],
|
||||
"references": [{ "path": "../logic" }]
|
||||
"references": [{ "path": "../common" }, { "path": "../logic" }]
|
||||
}
|
||||
|
||||
@@ -0,0 +1,12 @@
|
||||
import { defineConfig } from 'vitest/config';
|
||||
|
||||
export default defineConfig({
|
||||
resolve: {
|
||||
tsconfigPaths: true,
|
||||
},
|
||||
test: {
|
||||
environment: 'node',
|
||||
globals: true,
|
||||
include: ['test/**/*.test.ts'],
|
||||
},
|
||||
});
|
||||
Generated
+8
@@ -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':
|
||||
|
||||
@@ -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',
|
||||
|
||||
Reference in New Issue
Block a user