History

migrations and repository

Julius Marminge committed 8 months ago

Commit fa1663c1 parent 9b046fe8 files · +210 −98Browse files

Unified · Split

8 files
  1. apps/server/src/orchestration/engine.test.ts +2 −2
  2. apps/server/src/orchestration/engine.ts +6 −6
  3. apps/server/src/orchestration/eventRepository.ts +20 −0
  4. apps/server/src/orchestration/eventStore.test.ts +90 −0
  5. apps/server/src/orchestration/eventStore.ts +51 −74
  6. apps/server/src/orchestration/layers.ts +10 −11
  7. apps/server/src/orchestration/migrations.ts +31 −0
  8. apps/server/src/orchestration/services.ts +0 −5
apps/server/src/orchestration/engine.test.ts+2−2
@@ -7,7 +7,7 @@ import { Effect } from "effect"; · 6 unmodified lines
77 import { afterEach, describe, expect, it } from "vitest";
88
99 import { OrchestrationEngine } from "./engine";
10−import type { OrchestrationEventStore } from "./eventStore";
10+import type { OrchestrationEventRepositoryShape } from "./eventRepository";
1111
1212 const tempDirs: string[] = [];
1313
@@ -319,7 +319,7 @@ describe("OrchestrationEngine", () => { · 305 unmodified lines
319319 let nextSequence = 1;
320320 let shouldFailFirstAppend = true;
321321
322− const flakyStore: OrchestrationEventStore = {
322+ const flakyStore: OrchestrationEventRepositoryShape = {
323323 append(event) {
324324 if (shouldFailFirstAppend) {
325325 shouldFailFirstAppend = false;
apps/server/src/orchestration/engine.ts+6−6
@@ -20,8 +20,8 @@ import { · 19 unmodified lines
2020 } from "effect";
2121
2222 import { createLogger } from "../logger";
23−import type { OrchestrationEventStore } from "./eventStore";
24−import { SqliteEventStore } from "./eventStore";
23+import type { OrchestrationEventRepositoryShape } from "./eventRepository";
24+import { makeSqliteOrchestrationEventRepository } from "./eventStore";
2525 import { createEmptyReadModel, reduceEvent } from "./reducer";
2626 import { UI_ENTITY_CONTRACTS } from "./uiContractInventory";
2727
@@ -205,7 +205,7 @@ function mapCommandToEvent(command: OrchestrationCommand): Omit<OrchestrationEve · 177 unmodified lines
205205 export class OrchestrationEngine {
206206 private readonly logger = createLogger("orchestration");
207207 private readonly runtime = Runtime.defaultRuntime;
208− private readonly eventStore: OrchestrationEventStore;
208+ private readonly eventStore: OrchestrationEventRepositoryShape;
209209
210210 private readModel: OrchestrationReadModel;
211211 private commandQueue: Queue.Queue<CommandEnvelope>;
@@ -216,11 +216,11 @@ export class OrchestrationEngine { · 4 unmodified lines
216216 private readonly domainEventListeners = new Set<(event: OrchestrationEvent) => void>();
217217
218218 constructor(stateDir: string);
219− constructor(eventStore: OrchestrationEventStore);
220− constructor(stateDirOrEventStore: string | OrchestrationEventStore) {
219+ constructor(eventStore: OrchestrationEventRepositoryShape);
220+ constructor(stateDirOrEventStore: string | OrchestrationEventRepositoryShape) {
221221 if (typeof stateDirOrEventStore === "string") {
222222 const dbPath = path.join(stateDirOrEventStore, "orchestration.sqlite");
223− this.eventStore = new SqliteEventStore(dbPath);
223+ this.eventStore = makeSqliteOrchestrationEventRepository(dbPath);
224224 } else {
225225 this.eventStore = stateDirOrEventStore;
226226 }
apps/server/src/orchestration/eventRepository.ts+20−0
@@ -0,0 +1,20 @@
1+import type { OrchestrationEvent } from "@t3tools/contracts";
2+import { Context } from "effect";
3+import type { Effect } from "effect";
4+
5+export interface OrchestrationEventRepositoryShape {
6+ readonly append: (
7+ event: Omit<OrchestrationEvent, "sequence">,
8+ ) => Effect.Effect<OrchestrationEvent>;
9+ readonly readFromSequence: (
10+ sequenceExclusive: number,
11+ limit?: number,
12+ ) => Effect.Effect<OrchestrationEvent[]>;
13+ readonly readAll: () => Effect.Effect<OrchestrationEvent[]>;
14+ readonly close: () => void;
15+}
16+
17+export class OrchestrationEventRepository extends Context.Tag("orchestration/EventRepository")<
18+ OrchestrationEventRepository,
19+ OrchestrationEventRepositoryShape
20+>() {}
apps/server/src/orchestration/eventStore.test.ts+90−0
@@ -0,0 +1,90 @@
1+import fs from "node:fs";
2+import os from "node:os";
3+import path from "node:path";
4+import { DatabaseSync } from "node:sqlite";
5+
6+import { Runtime } from "effect";
7+import { afterEach, describe, expect, it } from "vitest";
8+
9+import { makeSqliteOrchestrationEventRepository } from "./eventStore";
10+
11+const tempDirs: string[] = [];
12+
13+function makeTempDir(prefix: string): string {
14+ const dir = fs.mkdtempSync(path.join(os.tmpdir(), prefix));
15+ tempDirs.push(dir);
16+ return dir;
17+}
18+
19+afterEach(() => {
20+ for (const dir of tempDirs.splice(0, tempDirs.length)) {
21+ fs.rmSync(dir, { recursive: true, force: true });
22+ }
23+});
24+
25+describe("SqliteEventStore", () => {
26+ it("persists and replays events across restarts", async () => {
27+ const stateDir = makeTempDir("t3code-event-store-");
28+ const dbPath = path.join(stateDir, "orchestration.sqlite");
29+ const runtime = Runtime.defaultRuntime;
30+ const createdAt = new Date().toISOString();
31+
32+ const first = makeSqliteOrchestrationEventRepository(dbPath);
33+ const saved = await Runtime.runPromise(runtime)(
34+ first.append({
35+ eventId: "event-1",
36+ type: "project.created",
37+ aggregateType: "project",
38+ aggregateId: "project-1",
39+ occurredAt: createdAt,
40+ commandId: "cmd-1",
41+ payload: { id: "project-1", name: "demo" },
42+ }),
43+ );
44+ expect(saved.sequence).toBe(1);
45+ first.close();
46+
47+ const second = makeSqliteOrchestrationEventRepository(dbPath);
48+ const replayed = await Runtime.runPromise(runtime)(second.readFromSequence(0));
49+ expect(replayed).toEqual([saved]);
50+ second.close();
51+ });
52+
53+ it("creates and reuses the migrator tracking table", async () => {
54+ const stateDir = makeTempDir("t3code-event-store-migrations-");
55+ const dbPath = path.join(stateDir, "orchestration.sqlite");
56+ const runtime = Runtime.defaultRuntime;
57+
58+ const first = makeSqliteOrchestrationEventRepository(dbPath);
59+ await Runtime.runPromise(runtime)(first.readAll());
60+ first.close();
61+
62+ const second = makeSqliteOrchestrationEventRepository(dbPath);
63+ await Runtime.runPromise(runtime)(second.readAll());
64+ second.close();
65+
66+ const db = new DatabaseSync(dbPath, { readOnly: true });
67+ const tableRow = db
68+ .prepare(
69+ `
70+ SELECT name
71+ FROM sqlite_master
72+ WHERE type = 'table' AND name = 'orchestration_sql_migrations'
73+ `,
74+ )
75+ .get() as { name: string } | undefined;
76+ expect(tableRow?.name).toBe("orchestration_sql_migrations");
77+
78+ const migrationRows = db
79+ .prepare(
80+ `
81+ SELECT *
82+ FROM orchestration_sql_migrations
83+ `,
84+ )
85+ .all() as Array<Record<string, unknown>>;
86+ expect(migrationRows).toHaveLength(1);
87+ expect(Object.values(migrationRows[0] ?? {}).length).toBeGreaterThan(0);
88+ db.close();
89+ });
90+});
apps/server/src/orchestration/eventStore.ts+51−74
@@ -4,17 +4,13 @@ import path from "node:path"; · 3 unmodified lines
44 import type { OrchestrationEvent } from "@t3tools/contracts";
55 import { OrchestrationEventSchema } from "@t3tools/contracts";
66 import type { SqlClient as SqlClientService } from "@effect/sql/SqlClient";
7−import * as SqlClient from "@effect/sql/SqlClient";
7+import * as SqlClient from "@effect/sql/SqlClient";
88 import * as SqlSchema from "@effect/sql/SqlSchema";
99 import * as SqliteClient from "@effect/sql-sqlite-node/SqliteClient";
1010 import { Effect, ManagedRuntime, Schema } from "effect";
1111
12−export interface OrchestrationEventStore {
13− append(event: Omit<OrchestrationEvent, "sequence">): Effect.Effect<OrchestrationEvent>;
14− readFromSequence(sequenceExclusive: number, limit?: number): Effect.Effect<OrchestrationEvent[]>;
15− readAll(): Effect.Effect<OrchestrationEvent[]>;
16− close(): void;
17−}
12+import type { OrchestrationEventRepositoryShape } from "./eventRepository";
13+import { runOrchestrationMigrations } from "./migrations";
1814
1915 const decodeEvent = Schema.decodeUnknownSync(OrchestrationEventSchema);
2016
@@ -119,59 +115,36 @@ function eventRowToOrchestrationEvent(row: EventRow): OrchestrationEvent { · 98 unmodified lines
119115 });
120116 }
121117
122−export class SqliteEventStore implements OrchestrationEventStore {
123− private readonly runtime: ManagedRuntime.ManagedRuntime<SqlClientService, unknown>;
124− private readonly migrated: Promise<void>;
125− private closed = false;
126−
127− constructor(dbPath: string) {
128− fs.mkdirSync(path.dirname(dbPath), { recursive: true });
129− this.runtime = ManagedRuntime.make(
130− SqliteClient.layer({
131− filename: dbPath,
132− }),
133− );
134− this.migrated = this.runtime.runPromise(this.migrate()).then(() => undefined);
135− }
136−
137− private provideSql<A, E>(
118+export function makeSqliteOrchestrationEventRepository(
119+ dbPath: string,
120+): OrchestrationEventRepositoryShape {
121+ fs.mkdirSync(path.dirname(dbPath), { recursive: true });
122+
123+ const runtime: ManagedRuntime.ManagedRuntime<SqlClientService, unknown> = ManagedRuntime.make(
124+ SqliteClient.layer({
125+ filename: dbPath,
126+ }),
127+ );
128+ let closed = false;
129+
130+ const provideSql = <A, E>(
138131 effect: Effect.Effect<A, E, SqlClientService>,
139− ): Effect.Effect<A, E | unknown> {
140− return Effect.provide(effect, this.runtime);
141− }
142−
143− private ensureMigrated(): Effect.Effect<void> {
144− return Effect.promise(() => this.migrated);
145− }
146−
147− private migrate(): Effect.Effect<void, unknown, SqlClientService> {
148− return Effect.gen(function* () {
149− const sql = yield* SqlClient.SqlClient;
150− yield* sql`PRAGMA journal_mode = WAL;`;
151− yield* sql`PRAGMA foreign_keys = ON;`;
152− yield* sql`
153− CREATE TABLE IF NOT EXISTS orchestration_events (
154− sequence INTEGER PRIMARY KEY AUTOINCREMENT,
155− event_id TEXT NOT NULL UNIQUE,
156− event_type TEXT NOT NULL,
157− aggregate_type TEXT NOT NULL,
158− aggregate_id TEXT NOT NULL,
159− occurred_at TEXT NOT NULL,
160− command_id TEXT,
161− payload_json TEXT NOT NULL
162− )
163− `;
164− yield* sql`
165− CREATE INDEX IF NOT EXISTS idx_orch_events_aggregate
166− ON orchestration_events(aggregate_type, aggregate_id, sequence)
167− `;
168− });
169− }
170−
171− append(event: Omit<OrchestrationEvent, "sequence">): Effect.Effect<OrchestrationEvent> {
172− return Effect.gen(this, function* () {
173− yield* this.ensureMigrated();
174− const row = yield* this.provideSql(
132+ ): Effect.Effect<A, E | unknown> => Effect.provide(effect, runtime);
133+
134+ const initialize = Effect.gen(function* () {
135+ const sql = yield* SqlClient.SqlClient;
136+ yield* sql`PRAGMA journal_mode = WAL;`;
137+ yield* sql`PRAGMA foreign_keys = ON;`;
138+ yield* runOrchestrationMigrations;
139+ });
140+
141+ const initialized = runtime.runPromise(initialize).then(() => undefined);
142+ const ensureInitialized = Effect.promise(() => initialized);
143+
144+ const append: OrchestrationEventRepositoryShape["append"] = (event) =>
145+ Effect.gen(function* () {
146+ yield* ensureInitialized;
147+ const row = yield* provideSql(
175148 appendEventRow({
176149 eventId: event.eventId,
177150 type: event.type,
@@ -184,12 +157,14 @@ export class SqliteEventStore implements OrchestrationEventStore { · 6 unmodified lines
184157 );
185158 return eventRowToOrchestrationEvent(row);
186159 }).pipe(Effect.orDie);
187− }
188160
189− readFromSequence(sequenceExclusive: number, limit = 1_000): Effect.Effect<OrchestrationEvent[]> {
190− return Effect.gen(this, function* () {
191− yield* this.ensureMigrated();
192− const rows = yield* this.provideSql(
161+ const readFromSequence: OrchestrationEventRepositoryShape["readFromSequence"] = (
162+ sequenceExclusive,
163+ limit = 1_000,
164+ ) =>
165+ Effect.gen(function* () {
166+ yield* ensureInitialized;
167+ const rows = yield* provideSql(
193168 readEventRowsFromSequence({
194169 sequenceExclusive,
195170 limit,
@@ -197,17 +172,19 @@ export class SqliteEventStore implements OrchestrationEventStore { · 1 unmodified line
197172 );
198173 return rows.map(eventRowToOrchestrationEvent);
199174 }).pipe(Effect.orDie);
200− }
201−
202− readAll(): Effect.Effect<OrchestrationEvent[]> {
203− return this.readFromSequence(0, Number.MAX_SAFE_INTEGER);
204− }
205175
206− close(): void {
207− if (this.closed) {
176+ const close = (): void => {
177+ if (closed) {
208178 return;
209179 }
210− this.closed = true;
211− void this.runtime.dispose();
212− }
180+ closed = true;
181+ void runtime.dispose();
182+ };
183+
184+ return {
185+ append,
186+ readFromSequence,
187+ readAll: () => readFromSequence(0, Number.MAX_SAFE_INTEGER),
188+ close,
189+ };
213190 }
apps/server/src/orchestration/layers.ts+10−11
@@ -2,27 +2,26 @@ import path from "node:path"; · 1 unmodified line
22
33 import { Effect, Layer } from "effect";
44
5+import { OrchestrationEventRepository } from "./eventRepository";
56 import { OrchestrationEngine } from "./engine";
6−import { SqliteEventStore } from "./eventStore";
7−import {
8− OrchestrationConfig,
9− OrchestrationEngineService,
10− OrchestrationEventStoreService,
11−} from "./services";
7+import { makeSqliteOrchestrationEventRepository } from "./eventStore";
8+import { OrchestrationConfig, OrchestrationEngineService } from "./services";
129
13−export const OrchestrationEventStoreLive = Layer.effect(
14− OrchestrationEventStoreService,
10+export const OrchestrationEventRepositoryLive = Layer.effect(
11+ OrchestrationEventRepository,
1512 Effect.map(OrchestrationConfig, ({ stateDir }) => {
1613 const dbPath = path.join(stateDir, "orchestration.sqlite");
17− return new SqliteEventStore(dbPath);
14+ return makeSqliteOrchestrationEventRepository(dbPath);
1815 }),
1916 );
2017
2118 export const OrchestrationEngineLive = Layer.effect(
2219 OrchestrationEngineService,
23− Effect.map(OrchestrationEventStoreService, (eventStore) => new OrchestrationEngine(eventStore)),
20+ Effect.map(OrchestrationEventRepository, (eventRepository) => {
21+ return new OrchestrationEngine(eventRepository);
22+ }),
2423 );
2524
2625 export const OrchestrationLive = OrchestrationEngineLive.pipe(
27− Layer.provide(OrchestrationEventStoreLive),
26+ Layer.provide(OrchestrationEventRepositoryLive),
2827 );
apps/server/src/orchestration/migrations.ts+31−0
@@ -0,0 +1,31 @@
1+import * as Migrator from "@effect/sql/Migrator";
2+import * as SqlClient from "@effect/sql/SqlClient";
3+import { Effect } from "effect";
4+
5+const ORCHESTRATION_MIGRATIONS = Migrator.fromRecord({
6+ "0001_create_orchestration_events": Effect.flatMap(SqlClient.SqlClient, (sql) =>
7+ Effect.gen(function* () {
8+ yield* sql`
9+ CREATE TABLE IF NOT EXISTS orchestration_events (
10+ sequence INTEGER PRIMARY KEY AUTOINCREMENT,
11+ event_id TEXT NOT NULL UNIQUE,
12+ event_type TEXT NOT NULL,
13+ aggregate_type TEXT NOT NULL,
14+ aggregate_id TEXT NOT NULL,
15+ occurred_at TEXT NOT NULL,
16+ command_id TEXT,
17+ payload_json TEXT NOT NULL
18+ )
19+ `;
20+ yield* sql`
21+ CREATE INDEX IF NOT EXISTS idx_orch_events_aggregate
22+ ON orchestration_events(aggregate_type, aggregate_id, sequence)
23+ `;
24+ }),
25+ ),
26+});
27+
28+export const runOrchestrationMigrations = Migrator.make({})({
29+ loader: ORCHESTRATION_MIGRATIONS,
30+ table: "orchestration_sql_migrations",
31+});
apps/server/src/orchestration/services.ts+0−5
@@ -1,7 +1,6 @@
11 import { Context } from "effect";
22
33 import { OrchestrationEngine } from "./engine";
4−import type { OrchestrationEventStore } from "./eventStore";
54
65 export interface OrchestrationConfigShape {
76 readonly stateDir: string;
@@ -12,10 +11,6 @@ export class OrchestrationConfig extends Context.Tag("orchestration/Config")< · 4 unmodified lines
1211 OrchestrationConfigShape
1312 >() {}
1413
15−export class OrchestrationEventStoreService extends Context.Tag(
16− "orchestration/EventStore",
17−)<OrchestrationEventStoreService, OrchestrationEventStore>() {}
18−
1914 export class OrchestrationEngineService extends Context.Tag("orchestration/Engine")<
2015 OrchestrationEngineService,
2116 OrchestrationEngine