Move persistence to server-side with Effect and SQLite

apps/server/package.json

@@ -17,6 +17,9 @@
     "test": "vitest run"
   },
   "dependencies": {
+    "@t3tools/core": "workspace:*",
+    "@t3tools/infra-sqlite": "workspace:*",
+    "effect": "^3.18.4",
     "node-pty": "^1.1.0",
     "open": "^10.1.0",
     "ws": "^8.18.0"

apps/server/src/coreRuntime.ts

@@ -0,0 +1,108 @@
+import path from "node:path";
+import crypto from "node:crypto";
+import type { ProviderEvent, ProviderSession, TerminalEvent } from "@t3tools/contracts";
+import { EffectFanout, OrchestrationEngine, QueueProjector, type AppViewState } from "@t3tools/core";
+import { createSqliteStores } from "@t3tools/infra-sqlite";
+
+import type { ProviderManager } from "./providerManager";
+import type { TerminalManager } from "./terminalManager";
+
+function nowIso(): string {
+  return new Date().toISOString();
+}
+
+export class CoreRuntime {
+  private readonly fanout = new EffectFanout();
+  private readonly stores;
+  private readonly projector: QueueProjector;
+  private readonly engine: OrchestrationEngine;
+
+  constructor(dbDir: string) {
+    const dbPath = path.join(dbDir, "event-store.sqlite");
+    this.stores = createSqliteStores(dbPath);
+    this.projector = new QueueProjector(this.stores.projectionStore, this.fanout);
+    this.engine = new OrchestrationEngine(this.stores.eventStore, this.stores.projectionStore, this.projector);
+  }
+
+  async start(cwd: string, projectName: string): Promise<void> {
+    await this.engine.start();
+    await this.engine.execute({
+      id: crypto.randomUUID(),
+      type: "app.bootstrap",
+      issuedAt: nowIso(),
+      payload: { cwd, projectName },
+    });
+  }
+
+  async stop(): Promise<void> {
+    await this.engine.stop();
+    this.stores.db.close();
+  }
+
+  async state(): Promise<AppViewState> {
+    return this.engine.currentState();
+  }
+
+  async dispatch(command: Parameters<OrchestrationEngine["execute"]>[0]): Promise<AppViewState> {
+    return this.engine.execute(command);
+  }
+
+  subscribe() {
+    return this.fanout.subscribe();
+  }
+
+  bindProviderEvents(providerManager: ProviderManager): void {
+    providerManager.on("event", (event: ProviderEvent) => {
+      void this.ingestProviderEvent(event);
+    });
+  }
+
+  bindTerminalEvents(terminalManager: TerminalManager): void {
+    terminalManager.on("event", (event: TerminalEvent) => {
+      if (event.type !== "activity" && event.type !== "error" && event.type !== "exited") return;
+      void this.dispatch({
+        id: crypto.randomUUID(),
+        type: "thread.setTerminalActivity",
+        issuedAt: event.createdAt,
+        payload: {
+          threadId: event.threadId,
+          terminalId: event.terminalId,
+          running: event.type === "activity" ? event.hasRunningSubprocess : false,
+        },
+      });
+    });
+  }
+
+  async bindProviderSession(threadId: string, session: ProviderSession): Promise<void> {
+    await this.dispatch({
+      id: crypto.randomUUID(),
+      type: "thread.updateProviderSession",
+      issuedAt: nowIso(),
+      payload: { threadId, session },
+    });
+  }
+
+  async clearProviderSession(threadId: string): Promise<void> {
+    await this.dispatch({
+      id: crypto.randomUUID(),
+      type: "thread.updateProviderSession",
+      issuedAt: nowIso(),
+      payload: { threadId, session: null },
+    });
+  }
+
+  async ingestProviderEvent(event: ProviderEvent): Promise<void> {
+    const state = await this.state();
+    const target = state.threads.find((thread) => thread.session?.sessionId === event.sessionId);
+    if (!target) return;
+    await this.dispatch({
+      id: crypto.randomUUID(),
+      type: "thread.recordProviderEvent",
+      issuedAt: event.createdAt,
+      payload: {
+        threadId: target.id,
+        event,
+      },
+    });
+  }
+}

apps/server/src/index.ts

@@ -121,6 +121,7 @@ async function main() {
     staticDir,
     devUrl,
     projectRegistry,
+    stateDir,
     authToken,
   });
   await server.start();