diff --git a/apps/server/src/orchestration/engine.test.ts b/apps/server/src/orchestration/engine.test.ts index 05cf57c72f2824c33fc0276f2bacd901e74c8886..5325ae35a86ecbe67033e5bbb17cc8a42b6d9f0c 100644 --- a/apps/server/src/orchestration/engine.test.ts +++ b/apps/server/src/orchestration/engine.test.ts @@ -7,7 +7,7 @@ import { Effect } from "effect"; import { afterEach, describe, expect, it } from "vitest"; import { OrchestrationEngine } from "./engine"; -import type { OrchestrationEventStore } from "./eventStore"; +import type { OrchestrationEventRepositoryShape } from "./eventRepository"; const tempDirs: string[] = []; @@ -319,7 +319,7 @@ describe("OrchestrationEngine", () => { let nextSequence = 1; let shouldFailFirstAppend = true; - const flakyStore: OrchestrationEventStore = { + const flakyStore: OrchestrationEventRepositoryShape = { append(event) { if (shouldFailFirstAppend) { shouldFailFirstAppend = false; diff --git a/apps/server/src/orchestration/engine.ts b/apps/server/src/orchestration/engine.ts index 5b8590eea5fc8b4b954756dbcedbb6c319c7e72b..826b5143f5e6387effa77a4e4b752fe018b52061 100644 --- a/apps/server/src/orchestration/engine.ts +++ b/apps/server/src/orchestration/engine.ts @@ -20,8 +20,8 @@ import { } from "effect"; import { createLogger } from "../logger"; -import type { OrchestrationEventStore } from "./eventStore"; -import { SqliteEventStore } from "./eventStore"; +import type { OrchestrationEventRepositoryShape } from "./eventRepository"; +import { makeSqliteOrchestrationEventRepository } from "./eventStore"; import { createEmptyReadModel, reduceEvent } from "./reducer"; import { UI_ENTITY_CONTRACTS } from "./uiContractInventory"; @@ -205,7 +205,7 @@ function mapCommandToEvent(command: OrchestrationCommand): Omit; @@ -216,11 +216,11 @@ export class OrchestrationEngine { private readonly domainEventListeners = new Set<(event: OrchestrationEvent) => void>(); constructor(stateDir: string); - constructor(eventStore: OrchestrationEventStore); - constructor(stateDirOrEventStore: string | OrchestrationEventStore) { + constructor(eventStore: OrchestrationEventRepositoryShape); + constructor(stateDirOrEventStore: string | OrchestrationEventRepositoryShape) { if (typeof stateDirOrEventStore === "string") { const dbPath = path.join(stateDirOrEventStore, "orchestration.sqlite"); - this.eventStore = new SqliteEventStore(dbPath); + this.eventStore = makeSqliteOrchestrationEventRepository(dbPath); } else { this.eventStore = stateDirOrEventStore; } diff --git a/apps/server/src/orchestration/eventRepository.ts b/apps/server/src/orchestration/eventRepository.ts new file mode 100644 index 0000000000000000000000000000000000000000..503df4c18ca4c963f4570c8348cd1b00d3bfffb7 --- /dev/null +++ b/apps/server/src/orchestration/eventRepository.ts @@ -0,0 +1,20 @@ +import type { OrchestrationEvent } from "@t3tools/contracts"; +import { Context } from "effect"; +import type { Effect } from "effect"; + +export interface OrchestrationEventRepositoryShape { + readonly append: ( + event: Omit, + ) => Effect.Effect; + readonly readFromSequence: ( + sequenceExclusive: number, + limit?: number, + ) => Effect.Effect; + readonly readAll: () => Effect.Effect; + readonly close: () => void; +} + +export class OrchestrationEventRepository extends Context.Tag("orchestration/EventRepository")< + OrchestrationEventRepository, + OrchestrationEventRepositoryShape +>() {} diff --git a/apps/server/src/orchestration/eventStore.test.ts b/apps/server/src/orchestration/eventStore.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..e75b94927608275ae62756d32665c8d19791d285 --- /dev/null +++ b/apps/server/src/orchestration/eventStore.test.ts @@ -0,0 +1,90 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { DatabaseSync } from "node:sqlite"; + +import { Runtime } from "effect"; +import { afterEach, describe, expect, it } from "vitest"; + +import { makeSqliteOrchestrationEventRepository } from "./eventStore"; + +const tempDirs: string[] = []; + +function makeTempDir(prefix: string): string { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), prefix)); + tempDirs.push(dir); + return dir; +} + +afterEach(() => { + for (const dir of tempDirs.splice(0, tempDirs.length)) { + fs.rmSync(dir, { recursive: true, force: true }); + } +}); + +describe("SqliteEventStore", () => { + it("persists and replays events across restarts", async () => { + const stateDir = makeTempDir("t3code-event-store-"); + const dbPath = path.join(stateDir, "orchestration.sqlite"); + const runtime = Runtime.defaultRuntime; + const createdAt = new Date().toISOString(); + + const first = makeSqliteOrchestrationEventRepository(dbPath); + const saved = await Runtime.runPromise(runtime)( + first.append({ + eventId: "event-1", + type: "project.created", + aggregateType: "project", + aggregateId: "project-1", + occurredAt: createdAt, + commandId: "cmd-1", + payload: { id: "project-1", name: "demo" }, + }), + ); + expect(saved.sequence).toBe(1); + first.close(); + + const second = makeSqliteOrchestrationEventRepository(dbPath); + const replayed = await Runtime.runPromise(runtime)(second.readFromSequence(0)); + expect(replayed).toEqual([saved]); + second.close(); + }); + + it("creates and reuses the migrator tracking table", async () => { + const stateDir = makeTempDir("t3code-event-store-migrations-"); + const dbPath = path.join(stateDir, "orchestration.sqlite"); + const runtime = Runtime.defaultRuntime; + + const first = makeSqliteOrchestrationEventRepository(dbPath); + await Runtime.runPromise(runtime)(first.readAll()); + first.close(); + + const second = makeSqliteOrchestrationEventRepository(dbPath); + await Runtime.runPromise(runtime)(second.readAll()); + second.close(); + + const db = new DatabaseSync(dbPath, { readOnly: true }); + const tableRow = db + .prepare( + ` + SELECT name + FROM sqlite_master + WHERE type = 'table' AND name = 'orchestration_sql_migrations' + `, + ) + .get() as { name: string } | undefined; + expect(tableRow?.name).toBe("orchestration_sql_migrations"); + + const migrationRows = db + .prepare( + ` + SELECT * + FROM orchestration_sql_migrations + `, + ) + .all() as Array>; + expect(migrationRows).toHaveLength(1); + expect(Object.values(migrationRows[0] ?? {}).length).toBeGreaterThan(0); + db.close(); + }); +}); diff --git a/apps/server/src/orchestration/eventStore.ts b/apps/server/src/orchestration/eventStore.ts index 09086d959e81b2c19c4836815fb90d39b3220cfc..6ec7572dc7363339ec7f64c65045cbe5e5a9cac9 100644 --- a/apps/server/src/orchestration/eventStore.ts +++ b/apps/server/src/orchestration/eventStore.ts @@ -4,17 +4,13 @@ import path from "node:path"; import type { OrchestrationEvent } from "@t3tools/contracts"; import { OrchestrationEventSchema } from "@t3tools/contracts"; import type { SqlClient as SqlClientService } from "@effect/sql/SqlClient"; -import * as SqlClient from "@effect/sql/SqlClient"; +import * as SqlClient from "@effect/sql/SqlClient"; import * as SqlSchema from "@effect/sql/SqlSchema"; import * as SqliteClient from "@effect/sql-sqlite-node/SqliteClient"; import { Effect, ManagedRuntime, Schema } from "effect"; -export interface OrchestrationEventStore { - append(event: Omit): Effect.Effect; - readFromSequence(sequenceExclusive: number, limit?: number): Effect.Effect; - readAll(): Effect.Effect; - close(): void; -} +import type { OrchestrationEventRepositoryShape } from "./eventRepository"; +import { runOrchestrationMigrations } from "./migrations"; const decodeEvent = Schema.decodeUnknownSync(OrchestrationEventSchema); @@ -119,59 +115,36 @@ function eventRowToOrchestrationEvent(row: EventRow): OrchestrationEvent { }); } -export class SqliteEventStore implements OrchestrationEventStore { - private readonly runtime: ManagedRuntime.ManagedRuntime; - private readonly migrated: Promise; - private closed = false; +export function makeSqliteOrchestrationEventRepository( + dbPath: string, +): OrchestrationEventRepositoryShape { + fs.mkdirSync(path.dirname(dbPath), { recursive: true }); - constructor(dbPath: string) { - fs.mkdirSync(path.dirname(dbPath), { recursive: true }); - this.runtime = ManagedRuntime.make( - SqliteClient.layer({ - filename: dbPath, - }), - ); - this.migrated = this.runtime.runPromise(this.migrate()).then(() => undefined); - } + const runtime: ManagedRuntime.ManagedRuntime = ManagedRuntime.make( + SqliteClient.layer({ + filename: dbPath, + }), + ); + let closed = false; - private provideSql( + const provideSql = ( effect: Effect.Effect, - ): Effect.Effect { - return Effect.provide(effect, this.runtime); - } + ): Effect.Effect => Effect.provide(effect, runtime); - private ensureMigrated(): Effect.Effect { - return Effect.promise(() => this.migrated); - } + const initialize = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`PRAGMA journal_mode = WAL;`; + yield* sql`PRAGMA foreign_keys = ON;`; + yield* runOrchestrationMigrations; + }); - private migrate(): Effect.Effect { - return Effect.gen(function* () { - const sql = yield* SqlClient.SqlClient; - yield* sql`PRAGMA journal_mode = WAL;`; - yield* sql`PRAGMA foreign_keys = ON;`; - yield* sql` - CREATE TABLE IF NOT EXISTS orchestration_events ( - sequence INTEGER PRIMARY KEY AUTOINCREMENT, - event_id TEXT NOT NULL UNIQUE, - event_type TEXT NOT NULL, - aggregate_type TEXT NOT NULL, - aggregate_id TEXT NOT NULL, - occurred_at TEXT NOT NULL, - command_id TEXT, - payload_json TEXT NOT NULL - ) - `; - yield* sql` - CREATE INDEX IF NOT EXISTS idx_orch_events_aggregate - ON orchestration_events(aggregate_type, aggregate_id, sequence) - `; - }); - } + const initialized = runtime.runPromise(initialize).then(() => undefined); + const ensureInitialized = Effect.promise(() => initialized); - append(event: Omit): Effect.Effect { - return Effect.gen(this, function* () { - yield* this.ensureMigrated(); - const row = yield* this.provideSql( + const append: OrchestrationEventRepositoryShape["append"] = (event) => + Effect.gen(function* () { + yield* ensureInitialized; + const row = yield* provideSql( appendEventRow({ eventId: event.eventId, type: event.type, @@ -184,12 +157,14 @@ export class SqliteEventStore implements OrchestrationEventStore { ); return eventRowToOrchestrationEvent(row); }).pipe(Effect.orDie); - } - readFromSequence(sequenceExclusive: number, limit = 1_000): Effect.Effect { - return Effect.gen(this, function* () { - yield* this.ensureMigrated(); - const rows = yield* this.provideSql( + const readFromSequence: OrchestrationEventRepositoryShape["readFromSequence"] = ( + sequenceExclusive, + limit = 1_000, + ) => + Effect.gen(function* () { + yield* ensureInitialized; + const rows = yield* provideSql( readEventRowsFromSequence({ sequenceExclusive, limit, @@ -197,17 +172,19 @@ export class SqliteEventStore implements OrchestrationEventStore { ); return rows.map(eventRowToOrchestrationEvent); }).pipe(Effect.orDie); - } - readAll(): Effect.Effect { - return this.readFromSequence(0, Number.MAX_SAFE_INTEGER); - } - - close(): void { - if (this.closed) { + const close = (): void => { + if (closed) { return; } - this.closed = true; - void this.runtime.dispose(); - } + closed = true; + void runtime.dispose(); + }; + + return { + append, + readFromSequence, + readAll: () => readFromSequence(0, Number.MAX_SAFE_INTEGER), + close, + }; } diff --git a/apps/server/src/orchestration/layers.ts b/apps/server/src/orchestration/layers.ts index 7c0220fd8a825fa952a147f29ae99a3e8b7e0772..d4037a3a0cad4904281ecf51ebed99d86b823051 100644 --- a/apps/server/src/orchestration/layers.ts +++ b/apps/server/src/orchestration/layers.ts @@ -2,27 +2,26 @@ import path from "node:path"; import { Effect, Layer } from "effect"; +import { OrchestrationEventRepository } from "./eventRepository"; import { OrchestrationEngine } from "./engine"; -import { SqliteEventStore } from "./eventStore"; -import { - OrchestrationConfig, - OrchestrationEngineService, - OrchestrationEventStoreService, -} from "./services"; +import { makeSqliteOrchestrationEventRepository } from "./eventStore"; +import { OrchestrationConfig, OrchestrationEngineService } from "./services"; -export const OrchestrationEventStoreLive = Layer.effect( - OrchestrationEventStoreService, +export const OrchestrationEventRepositoryLive = Layer.effect( + OrchestrationEventRepository, Effect.map(OrchestrationConfig, ({ stateDir }) => { const dbPath = path.join(stateDir, "orchestration.sqlite"); - return new SqliteEventStore(dbPath); + return makeSqliteOrchestrationEventRepository(dbPath); }), ); export const OrchestrationEngineLive = Layer.effect( OrchestrationEngineService, - Effect.map(OrchestrationEventStoreService, (eventStore) => new OrchestrationEngine(eventStore)), + Effect.map(OrchestrationEventRepository, (eventRepository) => { + return new OrchestrationEngine(eventRepository); + }), ); export const OrchestrationLive = OrchestrationEngineLive.pipe( - Layer.provide(OrchestrationEventStoreLive), + Layer.provide(OrchestrationEventRepositoryLive), ); diff --git a/apps/server/src/orchestration/migrations.ts b/apps/server/src/orchestration/migrations.ts new file mode 100644 index 0000000000000000000000000000000000000000..7d651dc4eb3bd07cd5f47df53c2034d281f254fa --- /dev/null +++ b/apps/server/src/orchestration/migrations.ts @@ -0,0 +1,31 @@ +import * as Migrator from "@effect/sql/Migrator"; +import * as SqlClient from "@effect/sql/SqlClient"; +import { Effect } from "effect"; + +const ORCHESTRATION_MIGRATIONS = Migrator.fromRecord({ + "0001_create_orchestration_events": Effect.flatMap(SqlClient.SqlClient, (sql) => + Effect.gen(function* () { + yield* sql` + CREATE TABLE IF NOT EXISTS orchestration_events ( + sequence INTEGER PRIMARY KEY AUTOINCREMENT, + event_id TEXT NOT NULL UNIQUE, + event_type TEXT NOT NULL, + aggregate_type TEXT NOT NULL, + aggregate_id TEXT NOT NULL, + occurred_at TEXT NOT NULL, + command_id TEXT, + payload_json TEXT NOT NULL + ) + `; + yield* sql` + CREATE INDEX IF NOT EXISTS idx_orch_events_aggregate + ON orchestration_events(aggregate_type, aggregate_id, sequence) + `; + }), + ), +}); + +export const runOrchestrationMigrations = Migrator.make({})({ + loader: ORCHESTRATION_MIGRATIONS, + table: "orchestration_sql_migrations", +}); diff --git a/apps/server/src/orchestration/services.ts b/apps/server/src/orchestration/services.ts index 5218c7a5884ab83e085556cc8466d2b24a5b4f01..148a83ef8bdb000fa7f720c7b1b01b580a8fa573 100644 --- a/apps/server/src/orchestration/services.ts +++ b/apps/server/src/orchestration/services.ts @@ -1,7 +1,6 @@ import { Context } from "effect"; import { OrchestrationEngine } from "./engine"; -import type { OrchestrationEventStore } from "./eventStore"; export interface OrchestrationConfigShape { readonly stateDir: string; @@ -12,10 +11,6 @@ export class OrchestrationConfig extends Context.Tag("orchestration/Config")< OrchestrationConfigShape >() {} -export class OrchestrationEventStoreService extends Context.Tag( - "orchestration/EventStore", -)() {} - export class OrchestrationEngineService extends Context.Tag("orchestration/Engine")< OrchestrationEngineService, OrchestrationEngine