authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-21 19:55:48-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-26 13:20:36-08:00
logbbf4e6bd3cf4dc1f73f3a0735092cf2f5e88c347
treed75e2e4cf6eb23d803b13165cd73ff19cff0bf5d
parentfa1663c1314760505d59cb043562345816f5eb1c

more ffect: Service/Layer separation

13 files changed, 403 insertions(+), 337 deletions(-)

apps/server/src/orchestration/engine.test.ts+1-1
...@@ -6,8 +6,8 @@ import type { OrchestrationEvent } from "@t3tools/contracts";...@@ -6,8 +6,8 @@ import type { OrchestrationEvent } from "@t3tools/contracts";
6import { Effect } from "effect";6import { Effect } from "effect";
7import { afterEach, describe, expect, it } from "vitest";7import { afterEach, describe, expect, it } from "vitest";
88
9import type { OrchestrationEventRepositoryShape } from "../persistence/Services/OrchestrationEvents";
9import { OrchestrationEngine } from "./engine";10import { OrchestrationEngine } from "./engine";
10import type { OrchestrationEventRepositoryShape } from "./eventRepository";
1111
12const tempDirs: string[] = [];12const tempDirs: string[] = [];
1313
apps/server/src/orchestration/engine.ts+2-2
...@@ -20,8 +20,8 @@ import {...@@ -20,8 +20,8 @@ import {
20} from "effect";20} from "effect";
2121
22import { createLogger } from "../logger";22import { createLogger } from "../logger";
23import type { OrchestrationEventRepositoryShape } from "./eventRepository";23import type { OrchestrationEventRepositoryShape } from "../persistence/Services/OrchestrationEvents";
24import { makeSqliteOrchestrationEventRepository } from "./eventStore";24import { makeSqliteOrchestrationEventRepository } from "../persistence/Layers/OrchestrationEvents";
25import { createEmptyReadModel, reduceEvent } from "./reducer";25import { createEmptyReadModel, reduceEvent } from "./reducer";
26import { UI_ENTITY_CONTRACTS } from "./uiContractInventory";26import { UI_ENTITY_CONTRACTS } from "./uiContractInventory";
2727
apps/server/src/orchestration/eventRepository.ts deleted-20
...@@ -1,20 +0,0 @@
1import type { OrchestrationEvent } from "@t3tools/contracts";
2import { Context } from "effect";
3import type { Effect } from "effect";
4
5export 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
17export class OrchestrationEventRepository extends Context.Tag("orchestration/EventRepository")<
18 OrchestrationEventRepository,
19 OrchestrationEventRepositoryShape
20>() {}
apps/server/src/orchestration/eventStore.test.ts deleted-90
...@@ -1,90 +0,0 @@
1import fs from "node:fs";
2import os from "node:os";
3import path from "node:path";
4import { DatabaseSync } from "node:sqlite";
5
6import { Runtime } from "effect";
7import { afterEach, describe, expect, it } from "vitest";
8
9import { makeSqliteOrchestrationEventRepository } from "./eventStore";
10
11const tempDirs: string[] = [];
12
13function makeTempDir(prefix: string): string {
14 const dir = fs.mkdtempSync(path.join(os.tmpdir(), prefix));
15 tempDirs.push(dir);
16 return dir;
17}
18
19afterEach(() => {
20 for (const dir of tempDirs.splice(0, tempDirs.length)) {
21 fs.rmSync(dir, { recursive: true, force: true });
22 }
23});
24
25describe("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 deleted-190
...@@ -1,190 +0,0 @@
1import fs from "node:fs";
2import path from "node:path";
3
4import type { OrchestrationEvent } from "@t3tools/contracts";
5import { OrchestrationEventSchema } from "@t3tools/contracts";
6import type { SqlClient as SqlClientService } from "@effect/sql/SqlClient";
7import * as SqlClient from "@effect/sql/SqlClient";
8import * as SqlSchema from "@effect/sql/SqlSchema";
9import * as SqliteClient from "@effect/sql-sqlite-node/SqliteClient";
10import { Effect, ManagedRuntime, Schema } from "effect";
11
12import type { OrchestrationEventRepositoryShape } from "./eventRepository";
13import { runOrchestrationMigrations } from "./migrations";
14
15const decodeEvent = Schema.decodeUnknownSync(OrchestrationEventSchema);
16
17const EventRowSchema = Schema.Struct({
18 sequence: Schema.Number,
19 eventId: Schema.String,
20 type: Schema.String,
21 aggregateType: Schema.String,
22 aggregateId: Schema.String,
23 occurredAt: Schema.String,
24 commandId: Schema.NullOr(Schema.String),
25 payloadJson: Schema.String,
26});
27
28type EventRow = Schema.Schema.Type<typeof EventRowSchema>;
29
30const AppendEventRequestSchema = Schema.Struct({
31 eventId: Schema.String,
32 type: Schema.String,
33 aggregateType: Schema.String,
34 aggregateId: Schema.String,
35 occurredAt: Schema.String,
36 commandId: Schema.NullOr(Schema.String),
37 payloadJson: Schema.String,
38});
39
40const ReadFromSequenceRequestSchema = Schema.Struct({
41 sequenceExclusive: Schema.Number,
42 limit: Schema.Number,
43});
44
45const appendEventRow = SqlSchema.single({
46 Request: AppendEventRequestSchema,
47 Result: EventRowSchema,
48 execute: (request) =>
49 Effect.flatMap(SqlClient.SqlClient, (sql) =>
50 sql`
51 INSERT INTO orchestration_events (
52 event_id,
53 event_type,
54 aggregate_type,
55 aggregate_id,
56 occurred_at,
57 command_id,
58 payload_json
59 )
60 VALUES (
61 ${request.eventId},
62 ${request.type},
63 ${request.aggregateType},
64 ${request.aggregateId},
65 ${request.occurredAt},
66 ${request.commandId},
67 ${request.payloadJson}
68 )
69 RETURNING
70 sequence,
71 event_id AS "eventId",
72 event_type AS "type",
73 aggregate_type AS "aggregateType",
74 aggregate_id AS "aggregateId",
75 occurred_at AS "occurredAt",
76 command_id AS "commandId",
77 payload_json AS "payloadJson"
78 `,
79 ),
80});
81
82const readEventRowsFromSequence = SqlSchema.findAll({
83 Request: ReadFromSequenceRequestSchema,
84 Result: EventRowSchema,
85 execute: (request) =>
86 Effect.flatMap(SqlClient.SqlClient, (sql) =>
87 sql`
88 SELECT
89 sequence,
90 event_id AS "eventId",
91 event_type AS "type",
92 aggregate_type AS "aggregateType",
93 aggregate_id AS "aggregateId",
94 occurred_at AS "occurredAt",
95 command_id AS "commandId",
96 payload_json AS "payloadJson"
97 FROM orchestration_events
98 WHERE sequence > ${request.sequenceExclusive}
99 ORDER BY sequence ASC
100 LIMIT ${request.limit}
101 `,
102 ),
103});
104
105function eventRowToOrchestrationEvent(row: EventRow): OrchestrationEvent {
106 return decodeEvent({
107 sequence: row.sequence,
108 eventId: row.eventId,
109 type: row.type,
110 aggregateType: row.aggregateType,
111 aggregateId: row.aggregateId,
112 occurredAt: row.occurredAt,
113 commandId: row.commandId,
114 payload: JSON.parse(row.payloadJson) as unknown,
115 });
116}
117
118export 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>(
131 effect: Effect.Effect<A, E, SqlClientService>,
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(
148 appendEventRow({
149 eventId: event.eventId,
150 type: event.type,
151 aggregateType: event.aggregateType,
152 aggregateId: event.aggregateId,
153 occurredAt: event.occurredAt,
154 commandId: event.commandId,
155 payloadJson: JSON.stringify(event.payload),
156 }),
157 );
158 return eventRowToOrchestrationEvent(row);
159 }).pipe(Effect.orDie);
160
161 const readFromSequence: OrchestrationEventRepositoryShape["readFromSequence"] = (
162 sequenceExclusive,
163 limit = 1_000,
164 ) =>
165 Effect.gen(function* () {
166 yield* ensureInitialized;
167 const rows = yield* provideSql(
168 readEventRowsFromSequence({
169 sequenceExclusive,
170 limit,
171 }),
172 );
173 return rows.map(eventRowToOrchestrationEvent);
174 }).pipe(Effect.orDie);
175
176 const close = (): void => {
177 if (closed) {
178 return;
179 }
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 };
190}
apps/server/src/orchestration/layers.ts+2-2
...@@ -2,9 +2,9 @@ import path from "node:path";...@@ -2,9 +2,9 @@ import path from "node:path";
22
3import { Effect, Layer } from "effect";3import { Effect, Layer } from "effect";
44
5import { OrchestrationEventRepository } from "./eventRepository";5import { OrchestrationEventRepository } from "../persistence/Services/OrchestrationEvents";
6import { makeSqliteOrchestrationEventRepository } from "../persistence/Layers/OrchestrationEvents";
6import { OrchestrationEngine } from "./engine";7import { OrchestrationEngine } from "./engine";
7import { makeSqliteOrchestrationEventRepository } from "./eventStore";
8import { OrchestrationConfig, OrchestrationEngineService } from "./services";8import { OrchestrationConfig, OrchestrationEngineService } from "./services";
99
10export const OrchestrationEventRepositoryLive = Layer.effect(10export const OrchestrationEventRepositoryLive = Layer.effect(
apps/server/src/orchestration/migrations.ts deleted-31
...@@ -1,31 +0,0 @@
1import * as Migrator from "@effect/sql/Migrator";
2import * as SqlClient from "@effect/sql/SqlClient";
3import { Effect } from "effect";
4
5const 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
28export const runOrchestrationMigrations = Migrator.make({})({
29 loader: ORCHESTRATION_MIGRATIONS,
30 table: "orchestration_sql_migrations",
31});
apps/server/src/persistence/Layers/OrchestrationEvents.test.ts created+94
...@@ -0,0 +1,94 @@
1import fs from "node:fs";
2import os from "node:os";
3import path from "node:path";
4import { DatabaseSync } from "node:sqlite";
5
6import { Runtime } from "effect";
7import { afterEach, describe, expect, it } from "vitest";
8
9import { makeSqliteOrchestrationEventRepository } from "./OrchestrationEvents";
10
11const tempDirs: string[] = [];
12
13function makeTempDir(prefix: string): string {
14 const dir = fs.mkdtempSync(path.join(os.tmpdir(), prefix));
15 tempDirs.push(dir);
16 return dir;
17}
18
19afterEach(() => {
20 for (const dir of tempDirs.splice(0, tempDirs.length)) {
21 fs.rmSync(dir, { recursive: true, force: true });
22 }
23});
24
25describe("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 IN ('effect_sql_migrations', 'orchestration_sql_migrations')
73 ORDER BY name ASC
74 `,
75 )
76 .get() as { name: string } | undefined;
77 expect(tableRow?.name).toBeDefined();
78 if (!tableRow) {
79 throw new Error("Expected migrations table to exist");
80 }
81
82 const migrationRows = db
83 .prepare(
84 `
85 SELECT *
86 FROM ${tableRow.name}
87 `,
88 )
89 .all() as Array<Record<string, unknown>>;
90 expect(migrationRows).toHaveLength(1);
91 expect(Object.values(migrationRows[0] ?? {}).length).toBeGreaterThan(0);
92 db.close();
93 });
94});
apps/server/src/persistence/Layers/OrchestrationEvents.ts created+190
...@@ -0,0 +1,190 @@
1import fs from "node:fs";
2import path from "node:path";
3
4import type { OrchestrationEvent } from "@t3tools/contracts";
5import { OrchestrationEventSchema } from "@t3tools/contracts";
6import type { SqlClient as SqlClientService } from "@effect/sql/SqlClient";
7import * as SqlClient from "@effect/sql/SqlClient";
8import * as SqlSchema from "@effect/sql/SqlSchema";
9import * as SqliteClient from "@effect/sql-sqlite-node/SqliteClient";
10import { Effect, ManagedRuntime, Schema } from "effect";
11
12import type { OrchestrationEventRepositoryShape } from "../Services/OrchestrationEvents";
13import { runMigrations } from "../Migrations";
14
15const decodeEvent = Schema.decodeUnknownSync(OrchestrationEventSchema);
16
17const EventRowSchema = Schema.Struct({
18 sequence: Schema.Number,
19 eventId: Schema.String,
20 type: Schema.String,
21 aggregateType: Schema.String,
22 aggregateId: Schema.String,
23 occurredAt: Schema.String,
24 commandId: Schema.NullOr(Schema.String),
25 payloadJson: Schema.String,
26});
27
28type EventRow = Schema.Schema.Type<typeof EventRowSchema>;
29
30const AppendEventRequestSchema = Schema.Struct({
31 eventId: Schema.String,
32 type: Schema.String,
33 aggregateType: Schema.String,
34 aggregateId: Schema.String,
35 occurredAt: Schema.String,
36 commandId: Schema.NullOr(Schema.String),
37 payloadJson: Schema.String,
38});
39
40const ReadFromSequenceRequestSchema = Schema.Struct({
41 sequenceExclusive: Schema.Number,
42 limit: Schema.Number,
43});
44
45const appendEventRow = SqlSchema.single({
46 Request: AppendEventRequestSchema,
47 Result: EventRowSchema,
48 execute: (request) =>
49 Effect.flatMap(SqlClient.SqlClient, (sql) =>
50 sql`
51 INSERT INTO orchestration_events (
52 event_id,
53 event_type,
54 aggregate_type,
55 aggregate_id,
56 occurred_at,
57 command_id,
58 payload_json
59 )
60 VALUES (
61 ${request.eventId},
62 ${request.type},
63 ${request.aggregateType},
64 ${request.aggregateId},
65 ${request.occurredAt},
66 ${request.commandId},
67 ${request.payloadJson}
68 )
69 RETURNING
70 sequence,
71 event_id AS "eventId",
72 event_type AS "type",
73 aggregate_type AS "aggregateType",
74 aggregate_id AS "aggregateId",
75 occurred_at AS "occurredAt",
76 command_id AS "commandId",
77 payload_json AS "payloadJson"
78 `,
79 ),
80});
81
82const readEventRowsFromSequence = SqlSchema.findAll({
83 Request: ReadFromSequenceRequestSchema,
84 Result: EventRowSchema,
85 execute: (request) =>
86 Effect.flatMap(SqlClient.SqlClient, (sql) =>
87 sql`
88 SELECT
89 sequence,
90 event_id AS "eventId",
91 event_type AS "type",
92 aggregate_type AS "aggregateType",
93 aggregate_id AS "aggregateId",
94 occurred_at AS "occurredAt",
95 command_id AS "commandId",
96 payload_json AS "payloadJson"
97 FROM orchestration_events
98 WHERE sequence > ${request.sequenceExclusive}
99 ORDER BY sequence ASC
100 LIMIT ${request.limit}
101 `,
102 ),
103});
104
105function eventRowToOrchestrationEvent(row: EventRow): OrchestrationEvent {
106 return decodeEvent({
107 sequence: row.sequence,
108 eventId: row.eventId,
109 type: row.type,
110 aggregateType: row.aggregateType,
111 aggregateId: row.aggregateId,
112 occurredAt: row.occurredAt,
113 commandId: row.commandId,
114 payload: JSON.parse(row.payloadJson) as unknown,
115 });
116}
117
118export 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>(
131 effect: Effect.Effect<A, E, SqlClientService>,
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* runMigrations;
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(
148 appendEventRow({
149 eventId: event.eventId,
150 type: event.type,
151 aggregateType: event.aggregateType,
152 aggregateId: event.aggregateId,
153 occurredAt: event.occurredAt,
154 commandId: event.commandId,
155 payloadJson: JSON.stringify(event.payload),
156 }),
157 );
158 return eventRowToOrchestrationEvent(row);
159 }).pipe(Effect.orDie);
160
161 const readFromSequence: OrchestrationEventRepositoryShape["readFromSequence"] = (
162 sequenceExclusive,
163 limit = 1_000,
164 ) =>
165 Effect.gen(function* () {
166 yield* ensureInitialized;
167 const rows = yield* provideSql(
168 readEventRowsFromSequence({
169 sequenceExclusive,
170 limit,
171 }),
172 );
173 return rows.map(eventRowToOrchestrationEvent);
174 }).pipe(Effect.orDie);
175
176 const close = (): void => {
177 if (closed) {
178 return;
179 }
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 };
190}
apps/server/src/persistence/Migrations.ts created+67
...@@ -0,0 +1,67 @@
1/**
2 * MigrationsLive - Migration runner with inline loader
3 *
4 * Uses Migrator.make with fromRecord to define migrations inline.
5 * All migrations are statically imported - no dynamic file system loading.
6 *
7 * Migrations run automatically when the MigrationLayer is provided,
8 * ensuring the database schema is always up-to-date before the application starts.
9 */
10
11import * as Migrator from "@effect/sql/Migrator";
12import * as Layer from "effect/Layer";
13
14// Import all migrations statically
15import Migration0001 from "./Migrations/001_OrchestrationEvents.ts";
16
17
18/**
19 * Migration loader with all migrations defined inline.
20 *
21 * Key format: "{id}_{name}" where:
22 * - id: numeric migration ID (determines execution order)
23 * - name: descriptive name for the migration
24 *
25 * Uses Migrator.fromRecord which parses the key format and
26 * returns migrations sorted by ID.
27 */
28const loader = Migrator.fromRecord({
29 "1_OrchestrationEvents": Migration0001,
30});
31
32/**
33 * Migrator run function - no schema dumping needed
34 * Uses the base Migrator.make without platform dependencies
35 */
36const run = Migrator.make({});
37
38/**
39 * Run all pending migrations.
40 *
41 * Creates the migrations tracking table (effect_sql_migrations) if it doesn't exist,
42 * then runs any migrations with ID greater than the latest recorded migration.
43 *
44 * Returns array of [id, name] tuples for migrations that were run.
45 *
46 * @returns Effect containing array of executed migrations
47 */
48export const runMigrations = run({ loader });
49
50/**
51 * Layer that runs migrations when the layer is built.
52 *
53 * Use this to ensure migrations run before your application starts.
54 * Migrations are run automatically - no separate script is needed.
55 *
56 * @example
57 * ```typescript
58 * import { MigrationsLive } from "@acme/db/Migrations"
59 * import { PgClient } from "@effect/sql-pg"
60 *
61 * // Migrations run automatically when PgClient is provided
62 * const AppLayer = MigrationsLive.pipe(
63 * Layer.provideMerge(PgClient.layer({ url: Redacted.make("postgresql://...") }))
64 * )
65 * ```
66 */
67export const MigrationsLive = Layer.effectDiscard(runMigrations);
\ No newline at end of file
apps/server/src/persistence/Migrations/001_OrchestrationEvents.ts created+24
...@@ -0,0 +1,24 @@
1import { SqlClient } from "@effect/sql";
2import * as Effect from "effect/Effect";
3
4export default Effect.gen(function* () {
5 const sql = yield* SqlClient.SqlClient;
6
7 yield* sql`
8 CREATE TABLE IF NOT EXISTS orchestration_events (
9 sequence INTEGER PRIMARY KEY AUTOINCREMENT,
10 event_id TEXT NOT NULL UNIQUE,
11 event_type TEXT NOT NULL,
12 aggregate_type TEXT NOT NULL,
13 aggregate_id TEXT NOT NULL,
14 occurred_at TEXT NOT NULL,
15 command_id TEXT,
16 payload_json TEXT NOT NULL
17 )
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});
\ No newline at end of file
apps/server/src/persistence/Services/OrchestrationEvents.ts created+20
...@@ -0,0 +1,20 @@
1import type { OrchestrationEvent } from "@t3tools/contracts";
2import { Context } from "effect";
3import type { Effect } from "effect";
4
5export 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
17export class OrchestrationEventRepository extends Context.Tag("orchestration/EventRepository")<
18 OrchestrationEventRepository,
19 OrchestrationEventRepositoryShape
20>() {}
apps/server/tsconfig.json+3-1
...@@ -3,7 +3,9 @@...@@ -3,7 +3,9 @@
3 "compilerOptions": {3 "compilerOptions": {
4 "composite": true,4 "composite": true,
5 "types": ["node"],5 "types": ["node"],
6 "lib": ["ES2023", "esnext.disposable"]6 "lib": ["ES2023", "esnext.disposable"],
7 "noEmit": true,
8 "allowImportingTsExtensions": true
7 },9 },
8 "include": ["src", "tsdown.config.ts"]10 "include": ["src", "tsdown.config.ts"]
9}11}