diff --git a/apps/server/src/orchestration/engine.ts b/apps/server/src/orchestration/engine.ts index 28ef5e976e45741f193047d5eaaf2bbdebebf52b..9021aa45030dad21a7d2bb9046e586eb18949dfe 100644 --- a/apps/server/src/orchestration/engine.ts +++ b/apps/server/src/orchestration/engine.ts @@ -9,6 +9,7 @@ import { OrchestrationCommandSchema } from "@t3tools/contracts"; import { PubSub, Queue, Schema, Stream, Effect, Fiber, Runtime, Either } from "effect"; import { createLogger } from "../logger"; +import type { OrchestrationEventStore } from "./eventStore"; import { SqliteEventStore } from "./eventStore"; import { createEmptyReadModel, reduceEvent } from "./reducer"; import { UI_ENTITY_CONTRACTS } from "./uiContractInventory"; @@ -190,7 +191,7 @@ function mapCommandToEvent(command: OrchestrationCommand): Omit; @@ -200,9 +201,15 @@ export class OrchestrationEngine { private readonly readModelListeners = new Set<(snapshot: OrchestrationReadModel) => void>(); private readonly domainEventListeners = new Set<(event: OrchestrationEvent) => void>(); - constructor(stateDir: string) { - const dbPath = path.join(stateDir, "orchestration.sqlite"); - this.eventStore = new SqliteEventStore(dbPath); + constructor(stateDir: string); + constructor(eventStore: OrchestrationEventStore); + constructor(stateDirOrEventStore: string | OrchestrationEventStore) { + if (typeof stateDirOrEventStore === "string") { + const dbPath = path.join(stateDirOrEventStore, "orchestration.sqlite"); + this.eventStore = new SqliteEventStore(dbPath); + } else { + this.eventStore = stateDirOrEventStore; + } this.readModel = createEmptyReadModel(new Date().toISOString()); this.commandQueue = Runtime.runSync(this.runtime)(Queue.unbounded()); this.readModelPubSub = Runtime.runSync(this.runtime)( diff --git a/apps/server/src/orchestration/eventStore.ts b/apps/server/src/orchestration/eventStore.ts index 636879785c1f258d7b0dbf04829b80376f090620..ee0c21da6fc9339eafcc80f39ca60151e6da76b8 100644 --- a/apps/server/src/orchestration/eventStore.ts +++ b/apps/server/src/orchestration/eventStore.ts @@ -6,6 +6,13 @@ import type { SQLOutputValue } from "node:sqlite"; import type { OrchestrationEvent } from "@t3tools/contracts"; import { Effect } from "effect"; +export interface OrchestrationEventStore { + append(event: Omit): Effect.Effect; + readFromSequence(sequenceExclusive: number, limit?: number): Effect.Effect; + readAll(): Effect.Effect; + close(): void; +} + interface EventRow { sequence: number; event_id: string; @@ -50,7 +57,7 @@ function readRequiredNumber( throw new Error(`Invalid SQLite value for ${String(column)}: expected number`); } -export class SqliteEventStore { +export class SqliteEventStore implements OrchestrationEventStore { private readonly db: DatabaseSync; private readonly insertStatement: StatementSync; private closed = false; diff --git a/apps/server/src/orchestration/layers.test.ts b/apps/server/src/orchestration/layers.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..da24a9b7c79c7b7bd6278cbbbe3a9b15015eb203 --- /dev/null +++ b/apps/server/src/orchestration/layers.test.ts @@ -0,0 +1,47 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +import { afterEach, describe, expect, it } from "vitest"; + +import { createOrchestrationEngine } from "./runtime"; + +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("orchestration layers", () => { + it("creates an engine through Context.Tag wiring", async () => { + const stateDir = makeTempDir("t3code-orchestration-layer-"); + const createdAt = new Date().toISOString(); + const engine = createOrchestrationEngine(stateDir); + + await engine.start(); + await engine.dispatch({ + type: "project.create", + commandId: "cmd-layer-1", + projectId: "project-layer-1", + name: "layer", + cwd: "/tmp/layer", + model: "gpt-5-codex", + createdAt, + }); + const firstSnapshot = engine.getSnapshot(); + await engine.stop(); + + const restarted = createOrchestrationEngine(stateDir); + await restarted.start(); + expect(restarted.getSnapshot()).toEqual(firstSnapshot); + await restarted.stop(); + }); +}); diff --git a/apps/server/src/orchestration/layers.ts b/apps/server/src/orchestration/layers.ts new file mode 100644 index 0000000000000000000000000000000000000000..7c0220fd8a825fa952a147f29ae99a3e8b7e0772 --- /dev/null +++ b/apps/server/src/orchestration/layers.ts @@ -0,0 +1,28 @@ +import path from "node:path"; + +import { Effect, Layer } from "effect"; + +import { OrchestrationEngine } from "./engine"; +import { SqliteEventStore } from "./eventStore"; +import { + OrchestrationConfig, + OrchestrationEngineService, + OrchestrationEventStoreService, +} from "./services"; + +export const OrchestrationEventStoreLive = Layer.effect( + OrchestrationEventStoreService, + Effect.map(OrchestrationConfig, ({ stateDir }) => { + const dbPath = path.join(stateDir, "orchestration.sqlite"); + return new SqliteEventStore(dbPath); + }), +); + +export const OrchestrationEngineLive = Layer.effect( + OrchestrationEngineService, + Effect.map(OrchestrationEventStoreService, (eventStore) => new OrchestrationEngine(eventStore)), +); + +export const OrchestrationLive = OrchestrationEngineLive.pipe( + Layer.provide(OrchestrationEventStoreLive), +); diff --git a/apps/server/src/orchestration/runtime.ts b/apps/server/src/orchestration/runtime.ts new file mode 100644 index 0000000000000000000000000000000000000000..cbb875b5f907d9a29a1f51c1500718561ad28f0d --- /dev/null +++ b/apps/server/src/orchestration/runtime.ts @@ -0,0 +1,14 @@ +import { Effect, Layer, Runtime } from "effect"; + +import type { OrchestrationEngine } from "./engine"; +import { OrchestrationLive } from "./layers"; +import { OrchestrationConfig, OrchestrationEngineService } from "./services"; + +export function createOrchestrationEngine(stateDir: string): OrchestrationEngine { + const orchestrationLayer = OrchestrationLive.pipe( + Layer.provide(Layer.succeed(OrchestrationConfig, { stateDir })), + ); + return Runtime.runSync(Runtime.defaultRuntime)( + Effect.provide(OrchestrationEngineService, orchestrationLayer), + ); +} diff --git a/apps/server/src/orchestration/services.ts b/apps/server/src/orchestration/services.ts new file mode 100644 index 0000000000000000000000000000000000000000..5218c7a5884ab83e085556cc8466d2b24a5b4f01 --- /dev/null +++ b/apps/server/src/orchestration/services.ts @@ -0,0 +1,22 @@ +import { Context } from "effect"; + +import { OrchestrationEngine } from "./engine"; +import type { OrchestrationEventStore } from "./eventStore"; + +export interface OrchestrationConfigShape { + readonly stateDir: string; +} + +export class OrchestrationConfig extends Context.Tag("orchestration/Config")< + OrchestrationConfig, + OrchestrationConfigShape +>() {} + +export class OrchestrationEventStoreService extends Context.Tag( + "orchestration/EventStore", +)() {} + +export class OrchestrationEngineService extends Context.Tag("orchestration/Engine")< + OrchestrationEngineService, + OrchestrationEngine +>() {} diff --git a/apps/server/src/wsServer.ts b/apps/server/src/wsServer.ts index e0e811c7b698016ad6de63b9902d5aea995dd018..cdee56f031e6e9624dfbeae62573461f3aca8911 100644 --- a/apps/server/src/wsServer.ts +++ b/apps/server/src/wsServer.ts @@ -36,7 +36,7 @@ import { import { TerminalManager } from "./terminalManager"; import { loadResolvedKeybindingsConfig, upsertKeybindingRule } from "./keybindings"; import { searchWorkspaceEntries } from "./workspaceEntries"; -import { OrchestrationEngine } from "./orchestration/engine"; +import { createOrchestrationEngine } from "./orchestration/runtime"; const MIME_TYPES: Record = { ".html": "text/html; charset=utf-8", @@ -107,7 +107,7 @@ export function createServer(options: ServerOptions) { const projectRegistry = providedRegistry ?? new ProjectRegistry(path.join(os.homedir(), ".t3", "userdata")); const gitManager = providedGitManager ?? new GitManager(); - const orchestrationEngine = new OrchestrationEngine( + const orchestrationEngine = createOrchestrationEngine( stateDir ?? path.join(os.homedir(), ".t3", "userdata"), ); const clients = new Set();