diff --git a/apps/server/src/orchestration/engine.test.ts b/apps/server/src/orchestration/engine.test.ts index 5325ae35a86ecbe67033e5bbb17cc8a42b6d9f0c..59789bfa744c9d1071753c543133fc2cc8f32ab8 100644 --- a/apps/server/src/orchestration/engine.test.ts +++ b/apps/server/src/orchestration/engine.test.ts @@ -6,8 +6,8 @@ import type { OrchestrationEvent } from "@t3tools/contracts"; import { Effect } from "effect"; import { afterEach, describe, expect, it } from "vitest"; +import type { OrchestrationEventRepositoryShape } from "../persistence/Services/OrchestrationEvents"; import { OrchestrationEngine } from "./engine"; -import type { OrchestrationEventRepositoryShape } from "./eventRepository"; const tempDirs: string[] = []; diff --git a/apps/server/src/orchestration/engine.ts b/apps/server/src/orchestration/engine.ts index 826b5143f5e6387effa77a4e4b752fe018b52061..56ba03ace3848cf0d2431176ae39e591fad66df9 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 { OrchestrationEventRepositoryShape } from "./eventRepository"; -import { makeSqliteOrchestrationEventRepository } from "./eventStore"; +import type { OrchestrationEventRepositoryShape } from "../persistence/Services/OrchestrationEvents"; +import { makeSqliteOrchestrationEventRepository } from "../persistence/Layers/OrchestrationEvents"; import { createEmptyReadModel, reduceEvent } from "./reducer"; import { UI_ENTITY_CONTRACTS } from "./uiContractInventory"; diff --git a/apps/server/src/orchestration/eventRepository.ts b/apps/server/src/orchestration/eventRepository.ts deleted file mode 100644 index 503df4c18ca4c963f4570c8348cd1b00d3bfffb7..0000000000000000000000000000000000000000 --- a/apps/server/src/orchestration/eventRepository.ts +++ /dev/null @@ -1,20 +0,0 @@ -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 deleted file mode 100644 index e75b94927608275ae62756d32665c8d19791d285..0000000000000000000000000000000000000000 --- a/apps/server/src/orchestration/eventStore.test.ts +++ /dev/null @@ -1,90 +0,0 @@ -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 deleted file mode 100644 index 6ec7572dc7363339ec7f64c65045cbe5e5a9cac9..0000000000000000000000000000000000000000 --- a/apps/server/src/orchestration/eventStore.ts +++ /dev/null @@ -1,190 +0,0 @@ -import fs from "node:fs"; -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 SqlSchema from "@effect/sql/SqlSchema"; -import * as SqliteClient from "@effect/sql-sqlite-node/SqliteClient"; -import { Effect, ManagedRuntime, Schema } from "effect"; - -import type { OrchestrationEventRepositoryShape } from "./eventRepository"; -import { runOrchestrationMigrations } from "./migrations"; - -const decodeEvent = Schema.decodeUnknownSync(OrchestrationEventSchema); - -const EventRowSchema = Schema.Struct({ - sequence: Schema.Number, - eventId: Schema.String, - type: Schema.String, - aggregateType: Schema.String, - aggregateId: Schema.String, - occurredAt: Schema.String, - commandId: Schema.NullOr(Schema.String), - payloadJson: Schema.String, -}); - -type EventRow = Schema.Schema.Type; - -const AppendEventRequestSchema = Schema.Struct({ - eventId: Schema.String, - type: Schema.String, - aggregateType: Schema.String, - aggregateId: Schema.String, - occurredAt: Schema.String, - commandId: Schema.NullOr(Schema.String), - payloadJson: Schema.String, -}); - -const ReadFromSequenceRequestSchema = Schema.Struct({ - sequenceExclusive: Schema.Number, - limit: Schema.Number, -}); - -const appendEventRow = SqlSchema.single({ - Request: AppendEventRequestSchema, - Result: EventRowSchema, - execute: (request) => - Effect.flatMap(SqlClient.SqlClient, (sql) => - sql` - INSERT INTO orchestration_events ( - event_id, - event_type, - aggregate_type, - aggregate_id, - occurred_at, - command_id, - payload_json - ) - VALUES ( - ${request.eventId}, - ${request.type}, - ${request.aggregateType}, - ${request.aggregateId}, - ${request.occurredAt}, - ${request.commandId}, - ${request.payloadJson} - ) - RETURNING - sequence, - event_id AS "eventId", - event_type AS "type", - aggregate_type AS "aggregateType", - aggregate_id AS "aggregateId", - occurred_at AS "occurredAt", - command_id AS "commandId", - payload_json AS "payloadJson" - `, - ), -}); - -const readEventRowsFromSequence = SqlSchema.findAll({ - Request: ReadFromSequenceRequestSchema, - Result: EventRowSchema, - execute: (request) => - Effect.flatMap(SqlClient.SqlClient, (sql) => - sql` - SELECT - sequence, - event_id AS "eventId", - event_type AS "type", - aggregate_type AS "aggregateType", - aggregate_id AS "aggregateId", - occurred_at AS "occurredAt", - command_id AS "commandId", - payload_json AS "payloadJson" - FROM orchestration_events - WHERE sequence > ${request.sequenceExclusive} - ORDER BY sequence ASC - LIMIT ${request.limit} - `, - ), -}); - -function eventRowToOrchestrationEvent(row: EventRow): OrchestrationEvent { - return decodeEvent({ - sequence: row.sequence, - eventId: row.eventId, - type: row.type, - aggregateType: row.aggregateType, - aggregateId: row.aggregateId, - occurredAt: row.occurredAt, - commandId: row.commandId, - payload: JSON.parse(row.payloadJson) as unknown, - }); -} - -export function makeSqliteOrchestrationEventRepository( - dbPath: string, -): OrchestrationEventRepositoryShape { - fs.mkdirSync(path.dirname(dbPath), { recursive: true }); - - const runtime: ManagedRuntime.ManagedRuntime = ManagedRuntime.make( - SqliteClient.layer({ - filename: dbPath, - }), - ); - let closed = false; - - const provideSql = ( - effect: Effect.Effect, - ): Effect.Effect => Effect.provide(effect, runtime); - - const initialize = Effect.gen(function* () { - const sql = yield* SqlClient.SqlClient; - yield* sql`PRAGMA journal_mode = WAL;`; - yield* sql`PRAGMA foreign_keys = ON;`; - yield* runOrchestrationMigrations; - }); - - const initialized = runtime.runPromise(initialize).then(() => undefined); - const ensureInitialized = Effect.promise(() => initialized); - - const append: OrchestrationEventRepositoryShape["append"] = (event) => - Effect.gen(function* () { - yield* ensureInitialized; - const row = yield* provideSql( - appendEventRow({ - eventId: event.eventId, - type: event.type, - aggregateType: event.aggregateType, - aggregateId: event.aggregateId, - occurredAt: event.occurredAt, - commandId: event.commandId, - payloadJson: JSON.stringify(event.payload), - }), - ); - return eventRowToOrchestrationEvent(row); - }).pipe(Effect.orDie); - - const readFromSequence: OrchestrationEventRepositoryShape["readFromSequence"] = ( - sequenceExclusive, - limit = 1_000, - ) => - Effect.gen(function* () { - yield* ensureInitialized; - const rows = yield* provideSql( - readEventRowsFromSequence({ - sequenceExclusive, - limit, - }), - ); - return rows.map(eventRowToOrchestrationEvent); - }).pipe(Effect.orDie); - - const close = (): void => { - if (closed) { - return; - } - 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 d4037a3a0cad4904281ecf51ebed99d86b823051..e2b4d2d1c82dce22d6613b7a56bfcac3d324e6fd 100644 --- a/apps/server/src/orchestration/layers.ts +++ b/apps/server/src/orchestration/layers.ts @@ -2,9 +2,9 @@ import path from "node:path"; import { Effect, Layer } from "effect"; -import { OrchestrationEventRepository } from "./eventRepository"; +import { OrchestrationEventRepository } from "../persistence/Services/OrchestrationEvents"; +import { makeSqliteOrchestrationEventRepository } from "../persistence/Layers/OrchestrationEvents"; import { OrchestrationEngine } from "./engine"; -import { makeSqliteOrchestrationEventRepository } from "./eventStore"; import { OrchestrationConfig, OrchestrationEngineService } from "./services"; export const OrchestrationEventRepositoryLive = Layer.effect( diff --git a/apps/server/src/orchestration/migrations.ts b/apps/server/src/orchestration/migrations.ts deleted file mode 100644 index 7d651dc4eb3bd07cd5f47df53c2034d281f254fa..0000000000000000000000000000000000000000 --- a/apps/server/src/orchestration/migrations.ts +++ /dev/null @@ -1,31 +0,0 @@ -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/persistence/Layers/OrchestrationEvents.test.ts b/apps/server/src/persistence/Layers/OrchestrationEvents.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..60ee3a7a4272b3e9a17ad43298009a7471986070 --- /dev/null +++ b/apps/server/src/persistence/Layers/OrchestrationEvents.test.ts @@ -0,0 +1,94 @@ +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 "./OrchestrationEvents"; + +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 IN ('effect_sql_migrations', 'orchestration_sql_migrations') + ORDER BY name ASC + `, + ) + .get() as { name: string } | undefined; + expect(tableRow?.name).toBeDefined(); + if (!tableRow) { + throw new Error("Expected migrations table to exist"); + } + + const migrationRows = db + .prepare( + ` + SELECT * + FROM ${tableRow.name} + `, + ) + .all() as Array>; + expect(migrationRows).toHaveLength(1); + expect(Object.values(migrationRows[0] ?? {}).length).toBeGreaterThan(0); + db.close(); + }); +}); diff --git a/apps/server/src/persistence/Layers/OrchestrationEvents.ts b/apps/server/src/persistence/Layers/OrchestrationEvents.ts new file mode 100644 index 0000000000000000000000000000000000000000..bafdcd2a434c0978c28b1dd95e8ad28d284a2b44 --- /dev/null +++ b/apps/server/src/persistence/Layers/OrchestrationEvents.ts @@ -0,0 +1,190 @@ +import fs from "node:fs"; +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 SqlSchema from "@effect/sql/SqlSchema"; +import * as SqliteClient from "@effect/sql-sqlite-node/SqliteClient"; +import { Effect, ManagedRuntime, Schema } from "effect"; + +import type { OrchestrationEventRepositoryShape } from "../Services/OrchestrationEvents"; +import { runMigrations } from "../Migrations"; + +const decodeEvent = Schema.decodeUnknownSync(OrchestrationEventSchema); + +const EventRowSchema = Schema.Struct({ + sequence: Schema.Number, + eventId: Schema.String, + type: Schema.String, + aggregateType: Schema.String, + aggregateId: Schema.String, + occurredAt: Schema.String, + commandId: Schema.NullOr(Schema.String), + payloadJson: Schema.String, +}); + +type EventRow = Schema.Schema.Type; + +const AppendEventRequestSchema = Schema.Struct({ + eventId: Schema.String, + type: Schema.String, + aggregateType: Schema.String, + aggregateId: Schema.String, + occurredAt: Schema.String, + commandId: Schema.NullOr(Schema.String), + payloadJson: Schema.String, +}); + +const ReadFromSequenceRequestSchema = Schema.Struct({ + sequenceExclusive: Schema.Number, + limit: Schema.Number, +}); + +const appendEventRow = SqlSchema.single({ + Request: AppendEventRequestSchema, + Result: EventRowSchema, + execute: (request) => + Effect.flatMap(SqlClient.SqlClient, (sql) => + sql` + INSERT INTO orchestration_events ( + event_id, + event_type, + aggregate_type, + aggregate_id, + occurred_at, + command_id, + payload_json + ) + VALUES ( + ${request.eventId}, + ${request.type}, + ${request.aggregateType}, + ${request.aggregateId}, + ${request.occurredAt}, + ${request.commandId}, + ${request.payloadJson} + ) + RETURNING + sequence, + event_id AS "eventId", + event_type AS "type", + aggregate_type AS "aggregateType", + aggregate_id AS "aggregateId", + occurred_at AS "occurredAt", + command_id AS "commandId", + payload_json AS "payloadJson" + `, + ), +}); + +const readEventRowsFromSequence = SqlSchema.findAll({ + Request: ReadFromSequenceRequestSchema, + Result: EventRowSchema, + execute: (request) => + Effect.flatMap(SqlClient.SqlClient, (sql) => + sql` + SELECT + sequence, + event_id AS "eventId", + event_type AS "type", + aggregate_type AS "aggregateType", + aggregate_id AS "aggregateId", + occurred_at AS "occurredAt", + command_id AS "commandId", + payload_json AS "payloadJson" + FROM orchestration_events + WHERE sequence > ${request.sequenceExclusive} + ORDER BY sequence ASC + LIMIT ${request.limit} + `, + ), +}); + +function eventRowToOrchestrationEvent(row: EventRow): OrchestrationEvent { + return decodeEvent({ + sequence: row.sequence, + eventId: row.eventId, + type: row.type, + aggregateType: row.aggregateType, + aggregateId: row.aggregateId, + occurredAt: row.occurredAt, + commandId: row.commandId, + payload: JSON.parse(row.payloadJson) as unknown, + }); +} + +export function makeSqliteOrchestrationEventRepository( + dbPath: string, +): OrchestrationEventRepositoryShape { + fs.mkdirSync(path.dirname(dbPath), { recursive: true }); + + const runtime: ManagedRuntime.ManagedRuntime = ManagedRuntime.make( + SqliteClient.layer({ + filename: dbPath, + }), + ); + let closed = false; + + const provideSql = ( + effect: Effect.Effect, + ): Effect.Effect => Effect.provide(effect, runtime); + + const initialize = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`PRAGMA journal_mode = WAL;`; + yield* sql`PRAGMA foreign_keys = ON;`; + yield* runMigrations; + }); + + const initialized = runtime.runPromise(initialize).then(() => undefined); + const ensureInitialized = Effect.promise(() => initialized); + + const append: OrchestrationEventRepositoryShape["append"] = (event) => + Effect.gen(function* () { + yield* ensureInitialized; + const row = yield* provideSql( + appendEventRow({ + eventId: event.eventId, + type: event.type, + aggregateType: event.aggregateType, + aggregateId: event.aggregateId, + occurredAt: event.occurredAt, + commandId: event.commandId, + payloadJson: JSON.stringify(event.payload), + }), + ); + return eventRowToOrchestrationEvent(row); + }).pipe(Effect.orDie); + + const readFromSequence: OrchestrationEventRepositoryShape["readFromSequence"] = ( + sequenceExclusive, + limit = 1_000, + ) => + Effect.gen(function* () { + yield* ensureInitialized; + const rows = yield* provideSql( + readEventRowsFromSequence({ + sequenceExclusive, + limit, + }), + ); + return rows.map(eventRowToOrchestrationEvent); + }).pipe(Effect.orDie); + + const close = (): void => { + if (closed) { + return; + } + closed = true; + void runtime.dispose(); + }; + + return { + append, + readFromSequence, + readAll: () => readFromSequence(0, Number.MAX_SAFE_INTEGER), + close, + }; +} diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts new file mode 100644 index 0000000000000000000000000000000000000000..129a6f3c6adc9385161ee1852f5702acb386d496 --- /dev/null +++ b/apps/server/src/persistence/Migrations.ts @@ -0,0 +1,67 @@ +/** + * MigrationsLive - Migration runner with inline loader + * + * Uses Migrator.make with fromRecord to define migrations inline. + * All migrations are statically imported - no dynamic file system loading. + * + * Migrations run automatically when the MigrationLayer is provided, + * ensuring the database schema is always up-to-date before the application starts. + */ + +import * as Migrator from "@effect/sql/Migrator"; +import * as Layer from "effect/Layer"; + +// Import all migrations statically +import Migration0001 from "./Migrations/001_OrchestrationEvents.ts"; + + +/** + * Migration loader with all migrations defined inline. + * + * Key format: "{id}_{name}" where: + * - id: numeric migration ID (determines execution order) + * - name: descriptive name for the migration + * + * Uses Migrator.fromRecord which parses the key format and + * returns migrations sorted by ID. + */ +const loader = Migrator.fromRecord({ + "1_OrchestrationEvents": Migration0001, +}); + +/** + * Migrator run function - no schema dumping needed + * Uses the base Migrator.make without platform dependencies + */ +const run = Migrator.make({}); + +/** + * Run all pending migrations. + * + * Creates the migrations tracking table (effect_sql_migrations) if it doesn't exist, + * then runs any migrations with ID greater than the latest recorded migration. + * + * Returns array of [id, name] tuples for migrations that were run. + * + * @returns Effect containing array of executed migrations + */ +export const runMigrations = run({ loader }); + +/** + * Layer that runs migrations when the layer is built. + * + * Use this to ensure migrations run before your application starts. + * Migrations are run automatically - no separate script is needed. + * + * @example + * ```typescript + * import { MigrationsLive } from "@acme/db/Migrations" + * import { PgClient } from "@effect/sql-pg" + * + * // Migrations run automatically when PgClient is provided + * const AppLayer = MigrationsLive.pipe( + * Layer.provideMerge(PgClient.layer({ url: Redacted.make("postgresql://...") })) + * ) + * ``` + */ +export const MigrationsLive = Layer.effectDiscard(runMigrations); \ No newline at end of file diff --git a/apps/server/src/persistence/Migrations/001_OrchestrationEvents.ts b/apps/server/src/persistence/Migrations/001_OrchestrationEvents.ts new file mode 100644 index 0000000000000000000000000000000000000000..f2c3b9f9a2c2c8e4f5f17586419a4575bb4c5a81 --- /dev/null +++ b/apps/server/src/persistence/Migrations/001_OrchestrationEvents.ts @@ -0,0 +1,24 @@ +import { SqlClient } from "@effect/sql"; +import * as Effect from "effect/Effect"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + 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) + `; +}); \ No newline at end of file diff --git a/apps/server/src/persistence/Services/OrchestrationEvents.ts b/apps/server/src/persistence/Services/OrchestrationEvents.ts new file mode 100644 index 0000000000000000000000000000000000000000..503df4c18ca4c963f4570c8348cd1b00d3bfffb7 --- /dev/null +++ b/apps/server/src/persistence/Services/OrchestrationEvents.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/tsconfig.json b/apps/server/tsconfig.json index bf7111184a650760ebe01ca7e5e6461033212d4f..776a47cce04100d562403014f72b25d17ca41c46 100644 --- a/apps/server/tsconfig.json +++ b/apps/server/tsconfig.json @@ -3,7 +3,9 @@ "compilerOptions": { "composite": true, "types": ["node"], - "lib": ["ES2023", "esnext.disposable"] + "lib": ["ES2023", "esnext.disposable"], + "noEmit": true, + "allowImportingTsExtensions": true }, "include": ["src", "tsdown.config.ts"] }