authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-21 19:25:31-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-26 13:20:36-08:00
log2bece01a3be88f5bde92e7ae5633b2d3181955a1
tree1dc0804fe0c86f9899b026500807fe2b98624c02
parent4f4691676acad990c6a51221e2522d4ffb131415

more effect: stage 1

7 files changed, 132 insertions(+), 7 deletions(-)

apps/server/src/orchestration/engine.ts+11-4
...@@ -9,6 +9,7 @@ import { OrchestrationCommandSchema } from "@t3tools/contracts";...@@ -9,6 +9,7 @@ import { OrchestrationCommandSchema } from "@t3tools/contracts";
9import { PubSub, Queue, Schema, Stream, Effect, Fiber, Runtime, Either } from "effect";9import { PubSub, Queue, Schema, Stream, Effect, Fiber, Runtime, Either } from "effect";
1010
11import { createLogger } from "../logger";11import { createLogger } from "../logger";
12import type { OrchestrationEventStore } from "./eventStore";
12import { SqliteEventStore } from "./eventStore";13import { SqliteEventStore } from "./eventStore";
13import { createEmptyReadModel, reduceEvent } from "./reducer";14import { createEmptyReadModel, reduceEvent } from "./reducer";
14import { UI_ENTITY_CONTRACTS } from "./uiContractInventory";15import { UI_ENTITY_CONTRACTS } from "./uiContractInventory";
...@@ -190,7 +191,7 @@ function mapCommandToEvent(command: OrchestrationCommand): Omit<OrchestrationEve...@@ -190,7 +191,7 @@ function mapCommandToEvent(command: OrchestrationCommand): Omit<OrchestrationEve
190export class OrchestrationEngine {191export class OrchestrationEngine {
191 private readonly logger = createLogger("orchestration");192 private readonly logger = createLogger("orchestration");
192 private readonly runtime = Runtime.defaultRuntime;193 private readonly runtime = Runtime.defaultRuntime;
193 private readonly eventStore: SqliteEventStore;194 private readonly eventStore: OrchestrationEventStore;
194195
195 private readModel: OrchestrationReadModel;196 private readModel: OrchestrationReadModel;
196 private commandQueue: Queue.Queue<CommandEnvelope>;197 private commandQueue: Queue.Queue<CommandEnvelope>;
...@@ -200,9 +201,15 @@ export class OrchestrationEngine {...@@ -200,9 +201,15 @@ export class OrchestrationEngine {
200 private readonly readModelListeners = new Set<(snapshot: OrchestrationReadModel) => void>();201 private readonly readModelListeners = new Set<(snapshot: OrchestrationReadModel) => void>();
201 private readonly domainEventListeners = new Set<(event: OrchestrationEvent) => void>();202 private readonly domainEventListeners = new Set<(event: OrchestrationEvent) => void>();
202203
203 constructor(stateDir: string) {204 constructor(stateDir: string);
204 const dbPath = path.join(stateDir, "orchestration.sqlite");205 constructor(eventStore: OrchestrationEventStore);
205 this.eventStore = new SqliteEventStore(dbPath);206 constructor(stateDirOrEventStore: string | OrchestrationEventStore) {
207 if (typeof stateDirOrEventStore === "string") {
208 const dbPath = path.join(stateDirOrEventStore, "orchestration.sqlite");
209 this.eventStore = new SqliteEventStore(dbPath);
210 } else {
211 this.eventStore = stateDirOrEventStore;
212 }
206 this.readModel = createEmptyReadModel(new Date().toISOString());213 this.readModel = createEmptyReadModel(new Date().toISOString());
207 this.commandQueue = Runtime.runSync(this.runtime)(Queue.unbounded<CommandEnvelope>());214 this.commandQueue = Runtime.runSync(this.runtime)(Queue.unbounded<CommandEnvelope>());
208 this.readModelPubSub = Runtime.runSync(this.runtime)(215 this.readModelPubSub = Runtime.runSync(this.runtime)(
apps/server/src/orchestration/eventStore.ts+8-1
...@@ -6,6 +6,13 @@ import type { SQLOutputValue } from "node:sqlite";...@@ -6,6 +6,13 @@ import type { SQLOutputValue } from "node:sqlite";
6import type { OrchestrationEvent } from "@t3tools/contracts";6import type { OrchestrationEvent } from "@t3tools/contracts";
7import { Effect } from "effect";7import { Effect } from "effect";
88
9export interface OrchestrationEventStore {
10 append(event: Omit<OrchestrationEvent, "sequence">): Effect.Effect<OrchestrationEvent>;
11 readFromSequence(sequenceExclusive: number, limit?: number): Effect.Effect<OrchestrationEvent[]>;
12 readAll(): Effect.Effect<OrchestrationEvent[]>;
13 close(): void;
14}
15
9interface EventRow {16interface EventRow {
10 sequence: number;17 sequence: number;
11 event_id: string;18 event_id: string;
...@@ -50,7 +57,7 @@ function readRequiredNumber(...@@ -50,7 +57,7 @@ function readRequiredNumber(
50 throw new Error(`Invalid SQLite value for ${String(column)}: expected number`);57 throw new Error(`Invalid SQLite value for ${String(column)}: expected number`);
51}58}
5259
53export class SqliteEventStore {60export class SqliteEventStore implements OrchestrationEventStore {
54 private readonly db: DatabaseSync;61 private readonly db: DatabaseSync;
55 private readonly insertStatement: StatementSync;62 private readonly insertStatement: StatementSync;
56 private closed = false;63 private closed = false;
apps/server/src/orchestration/layers.test.ts created+47
...@@ -0,0 +1,47 @@
1import fs from "node:fs";
2import os from "node:os";
3import path from "node:path";
4
5import { afterEach, describe, expect, it } from "vitest";
6
7import { createOrchestrationEngine } from "./runtime";
8
9const tempDirs: string[] = [];
10
11function makeTempDir(prefix: string): string {
12 const dir = fs.mkdtempSync(path.join(os.tmpdir(), prefix));
13 tempDirs.push(dir);
14 return dir;
15}
16
17afterEach(() => {
18 for (const dir of tempDirs.splice(0, tempDirs.length)) {
19 fs.rmSync(dir, { recursive: true, force: true });
20 }
21});
22
23describe("orchestration layers", () => {
24 it("creates an engine through Context.Tag wiring", async () => {
25 const stateDir = makeTempDir("t3code-orchestration-layer-");
26 const createdAt = new Date().toISOString();
27 const engine = createOrchestrationEngine(stateDir);
28
29 await engine.start();
30 await engine.dispatch({
31 type: "project.create",
32 commandId: "cmd-layer-1",
33 projectId: "project-layer-1",
34 name: "layer",
35 cwd: "/tmp/layer",
36 model: "gpt-5-codex",
37 createdAt,
38 });
39 const firstSnapshot = engine.getSnapshot();
40 await engine.stop();
41
42 const restarted = createOrchestrationEngine(stateDir);
43 await restarted.start();
44 expect(restarted.getSnapshot()).toEqual(firstSnapshot);
45 await restarted.stop();
46 });
47});
apps/server/src/orchestration/layers.ts created+28
...@@ -0,0 +1,28 @@
1import path from "node:path";
2
3import { Effect, Layer } from "effect";
4
5import { OrchestrationEngine } from "./engine";
6import { SqliteEventStore } from "./eventStore";
7import {
8 OrchestrationConfig,
9 OrchestrationEngineService,
10 OrchestrationEventStoreService,
11} from "./services";
12
13export const OrchestrationEventStoreLive = Layer.effect(
14 OrchestrationEventStoreService,
15 Effect.map(OrchestrationConfig, ({ stateDir }) => {
16 const dbPath = path.join(stateDir, "orchestration.sqlite");
17 return new SqliteEventStore(dbPath);
18 }),
19);
20
21export const OrchestrationEngineLive = Layer.effect(
22 OrchestrationEngineService,
23 Effect.map(OrchestrationEventStoreService, (eventStore) => new OrchestrationEngine(eventStore)),
24);
25
26export const OrchestrationLive = OrchestrationEngineLive.pipe(
27 Layer.provide(OrchestrationEventStoreLive),
28);
apps/server/src/orchestration/runtime.ts created+14
...@@ -0,0 +1,14 @@
1import { Effect, Layer, Runtime } from "effect";
2
3import type { OrchestrationEngine } from "./engine";
4import { OrchestrationLive } from "./layers";
5import { OrchestrationConfig, OrchestrationEngineService } from "./services";
6
7export function createOrchestrationEngine(stateDir: string): OrchestrationEngine {
8 const orchestrationLayer = OrchestrationLive.pipe(
9 Layer.provide(Layer.succeed(OrchestrationConfig, { stateDir })),
10 );
11 return Runtime.runSync(Runtime.defaultRuntime)(
12 Effect.provide(OrchestrationEngineService, orchestrationLayer),
13 );
14}
apps/server/src/orchestration/services.ts created+22
...@@ -0,0 +1,22 @@
1import { Context } from "effect";
2
3import { OrchestrationEngine } from "./engine";
4import type { OrchestrationEventStore } from "./eventStore";
5
6export interface OrchestrationConfigShape {
7 readonly stateDir: string;
8}
9
10export class OrchestrationConfig extends Context.Tag("orchestration/Config")<
11 OrchestrationConfig,
12 OrchestrationConfigShape
13>() {}
14
15export class OrchestrationEventStoreService extends Context.Tag(
16 "orchestration/EventStore",
17)<OrchestrationEventStoreService, OrchestrationEventStore>() {}
18
19export class OrchestrationEngineService extends Context.Tag("orchestration/Engine")<
20 OrchestrationEngineService,
21 OrchestrationEngine
22>() {}
apps/server/src/wsServer.ts+2-2
...@@ -36,7 +36,7 @@ import {...@@ -36,7 +36,7 @@ import {
36import { TerminalManager } from "./terminalManager";36import { TerminalManager } from "./terminalManager";
37import { loadResolvedKeybindingsConfig, upsertKeybindingRule } from "./keybindings";37import { loadResolvedKeybindingsConfig, upsertKeybindingRule } from "./keybindings";
38import { searchWorkspaceEntries } from "./workspaceEntries";38import { searchWorkspaceEntries } from "./workspaceEntries";
39import { OrchestrationEngine } from "./orchestration/engine";39import { createOrchestrationEngine } from "./orchestration/runtime";
4040
41const MIME_TYPES: Record<string, string> = {41const MIME_TYPES: Record<string, string> = {
42 ".html": "text/html; charset=utf-8",42 ".html": "text/html; charset=utf-8",
...@@ -107,7 +107,7 @@ export function createServer(options: ServerOptions) {...@@ -107,7 +107,7 @@ export function createServer(options: ServerOptions) {
107 const projectRegistry =107 const projectRegistry =
108 providedRegistry ?? new ProjectRegistry(path.join(os.homedir(), ".t3", "userdata"));108 providedRegistry ?? new ProjectRegistry(path.join(os.homedir(), ".t3", "userdata"));
109 const gitManager = providedGitManager ?? new GitManager();109 const gitManager = providedGitManager ?? new GitManager();
110 const orchestrationEngine = new OrchestrationEngine(110 const orchestrationEngine = createOrchestrationEngine(
111 stateDir ?? path.join(os.homedir(), ".t3", "userdata"),111 stateDir ?? path.join(os.homedir(), ".t3", "userdata"),
112 );112 );
113 const clients = new Set<WebSocket>();113 const clients = new Set<WebSocket>();