/** * Server - HTTP/WebSocket server service interface. * * Owns startup and shutdown lifecycle of the HTTP server, static asset serving, * and WebSocket request routing. * * @module Server */import http from "node:http";import type { Duplex } from "node:stream"; import Mime from "@effect/platform-node/Mime";import {  CommandId,  ORCHESTRATION_WS_CHANNELS,  ORCHESTRATION_WS_METHODS,  ProjectId,  ThreadId,  TerminalEvent,  WS_CHANNELS,  WS_METHODS,  WebSocketRequest,  WsPush,  WsResponse,} from "@t3tools/contracts";import * as NodeHttpServer from "@effect/platform-node/NodeHttpServer";import {  Cause,  Effect,  Exit,  FileSystem,  Layer,  Path,  Ref,  Schema,  Scope,  ServiceMap,  Stream,  Struct,} from "effect";import { WebSocketServer, type WebSocket } from "ws"; import { createLogger } from "./logger";import { GitManager } from "./git/Services/GitManager.ts";import { TerminalManager } from "./terminal/Services/Manager.ts";import { Keybindings } from "./keybindings";import { searchWorkspaceEntries } from "./workspaceEntries";import { OrchestrationEngineService } from "./orchestration/Services/OrchestrationEngine";import { ProjectionSnapshotQuery } from "./orchestration/Services/ProjectionSnapshotQuery";import { OrchestrationReactor } from "./orchestration/Services/OrchestrationReactor";import { ProviderService } from "./provider/Services/ProviderService";import { CheckpointDiffQuery } from "./checkpointing/Services/CheckpointDiffQuery";import { clamp } from "effect/Number";import { Open } from "./open";import { ServerConfig } from "./config";import { GitCore } from "./git/Services/GitCore.ts";import { ATTACHMENTS_ROUTE_PREFIX, tryHandleProjectFaviconRequest } from "./projectFaviconRoute"; /** * ServerShape - Service API for server lifecycle control. */export interface ServerShape {  /**   * Start HTTP and WebSocket listeners.   */  readonly start: Effect.Effect<    http.Server,    ServerLifecycleError,    Scope.Scope | ServerRuntimeServices | ServerConfig | FileSystem.FileSystem | Path.Path  >;   /**   * Wait for process shutdown signals.   */  readonly stopSignal: Effect.Effect<void, never>;} /** * Server - Service tag for HTTP/WebSocket lifecycle management. */export class Server extends ServiceMap.Service<Server, ServerShape>()("t3/wsServer/Server") {} const isServerNotRunningError = (error: unknown): boolean => {  if (!(error instanceof Error)) return false;  const maybeCode = (error as NodeJS.ErrnoException).code;  return (    maybeCode === "ERR_SERVER_NOT_RUNNING" || error.message.toLowerCase().includes("not running")  );}; function rejectUpgrade(socket: Duplex, statusCode: number, message: string): void {  socket.end(    `HTTP/1.1 ${statusCode} ${statusCode === 401 ? "Unauthorized" : "Bad Request"}\r\n` +      "Connection: close\r\n" +      "Content-Type: text/plain\r\n" +      `Content-Length: ${Buffer.byteLength(message)}\r\n` +      "\r\n" +      message,  );} function websocketRawToString(raw: unknown): string | null {  if (typeof raw === "string") {    return raw;  }  if (raw instanceof Uint8Array) {    return Buffer.from(raw).toString("utf8");  }  if (raw instanceof ArrayBuffer) {    return Buffer.from(new Uint8Array(raw)).toString("utf8");  }  if (Array.isArray(raw)) {    const chunks: string[] = [];    for (const chunk of raw) {      if (typeof chunk === "string") {        chunks.push(chunk);        continue;      }      if (chunk instanceof Uint8Array) {        chunks.push(Buffer.from(chunk).toString("utf8"));        continue;      }      if (chunk instanceof ArrayBuffer) {        chunks.push(Buffer.from(new Uint8Array(chunk)).toString("utf8"));        continue;      }      return null;    }    return chunks.join("");  }  return null;} function stripRequestTag<T extends { _tag: string }>(body: T) {  return Struct.omit(body, ["_tag"]);} export type ServerCoreRuntimeServices =  | OrchestrationEngineService  | ProjectionSnapshotQuery  | CheckpointDiffQuery  | OrchestrationReactor  | ProviderService; export type ServerRuntimeServices =  | ServerCoreRuntimeServices  | GitManager  | GitCore  | TerminalManager  | Keybindings  | Open; export class ServerLifecycleError extends Schema.TaggedErrorClass<ServerLifecycleError>()(  "ServerLifecycleError",  {    operation: Schema.String,    cause: Schema.optional(Schema.Defect),  },) {} class RouteRequestError extends Schema.TaggedErrorClass<RouteRequestError>()("RouteRequestError", {  message: Schema.String,}) {} export const createServer = Effect.fn(function* (): Effect.fn.Return<  http.Server,  ServerLifecycleError,  Scope.Scope | ServerRuntimeServices | ServerConfig | FileSystem.FileSystem | Path.Path> {  const serverConfig = yield* ServerConfig;  const {    port,    cwd,    keybindingsConfigPath,    staticDir,    devUrl,    authToken,    host,    logWebSocketEvents,    autoBootstrapProjectFromCwd,  } = serverConfig;   const gitManager = yield* GitManager;  const terminalManager = yield* TerminalManager;  const keybindingsManager = yield* Keybindings;  const git = yield* GitCore;  const fileSystem = yield* FileSystem.FileSystem;  const path = yield* Path.Path;   yield* keybindingsManager.syncDefaultKeybindingsOnStartup.pipe(    Effect.catch((error) =>      Effect.logWarning("failed to sync keybindings defaults on startup", {        path: error.configPath,        detail: error.detail,        cause: error.cause,      }),    ),  );   const clients = yield* Ref.make(new Set<WebSocket>());  const logger = createLogger("ws");   function logOutgoingPush(push: WsPush, recipients: number) {    if (!logWebSocketEvents) return;    logger.event("outgoing push", {      channel: push.channel,      recipients,      payload: push.data,    });  }   const encodePush = Schema.encodeEffect(Schema.fromJsonString(WsPush));  const broadcastPush = Effect.fnUntraced(function* (push: WsPush) {    const message = yield* encodePush(push);    let recipients = 0;    for (const client of yield* Ref.get(clients)) {      if (client.readyState === client.OPEN) {        client.send(message);        recipients += 1;      }    }    logOutgoingPush(push, recipients);  });   const onTerminalEvent = Effect.fnUntraced(function* (event: TerminalEvent) {    yield* broadcastPush({      type: "push",      channel: WS_CHANNELS.terminalEvent,      data: event,    });  });   // HTTP server — serves static files or redirects to Vite dev server  const httpServer = http.createServer((req, res) => {    const respond = (      statusCode: number,      headers: Record<string, string>,      body?: string | Uint8Array,    ) => {      res.writeHead(statusCode, headers);      res.end(body);    };     void Effect.runPromise(      Effect.gen(function* () {        const url = new URL(req.url ?? "/", `http://localhost:${port}`);        if (tryHandleProjectFaviconRequest(url, res)) {          return;        }         if (url.pathname.startsWith(ATTACHMENTS_ROUTE_PREFIX)) {          const attachmentsRoot = path.resolve(path.join(serverConfig.stateDir, "attachments"));          const rawRelativePath = url.pathname.slice(ATTACHMENTS_ROUTE_PREFIX.length);          const normalizedRelativePath = path.normalize(rawRelativePath).replace(/^[/\\]+/, "");           if (            normalizedRelativePath.length === 0 ||            normalizedRelativePath.startsWith("..") ||            normalizedRelativePath.includes("\0")          ) {            respond(400, { "Content-Type": "text/plain" }, "Invalid attachment path");            return;          }           const filePath = path.resolve(path.join(attachmentsRoot, normalizedRelativePath));          if (!filePath.startsWith(`${attachmentsRoot}${path.sep}`)) {            respond(400, { "Content-Type": "text/plain" }, "Invalid attachment path");            return;          }           const fileInfo = yield* fileSystem            .stat(filePath)            .pipe(Effect.catch(() => Effect.succeed(null)));          if (!fileInfo || fileInfo.type !== "File") {            respond(404, { "Content-Type": "text/plain" }, "Not Found");            return;          }           const contentType = Mime.getType(filePath) ?? "application/octet-stream";          res.writeHead(200, {            "Content-Type": contentType,            "Cache-Control": "public, max-age=31536000, immutable",          });          const streamExit = yield* Stream.runForEach(fileSystem.stream(filePath), (chunk) =>            Effect.sync(() => {              if (!res.destroyed) {                res.write(chunk);              }            }),          ).pipe(Effect.exit);          if (streamExit._tag === "Failure") {            if (!res.destroyed) {              res.destroy();            }            return;          }          if (!res.writableEnded) {            res.end();          }          return;        }         // In dev mode, redirect to Vite dev server        if (devUrl) {          respond(302, { Location: devUrl.href });          return;        }         // Serve static files from the web app build        if (!staticDir) {          respond(            503,            { "Content-Type": "text/plain" },            "No static directory configured and no dev URL set.",          );          return;        }         const staticRoot = path.resolve(staticDir);        const staticRequestPath = url.pathname === "/" ? "/index.html" : url.pathname;        const rawStaticRelativePath = staticRequestPath.replace(/^[/\\]+/, "");        const hasRawLeadingParentSegment = rawStaticRelativePath.startsWith("..");        const staticRelativePath = path.normalize(rawStaticRelativePath).replace(/^[/\\]+/, "");        const hasPathTraversalSegment = staticRelativePath.startsWith("..");        if (          staticRelativePath.length === 0 ||          hasRawLeadingParentSegment ||          hasPathTraversalSegment ||          staticRelativePath.includes("\0")        ) {          respond(400, { "Content-Type": "text/plain" }, "Invalid static file path");          return;        }         const isWithinStaticRoot = (candidate: string) =>          candidate === staticRoot ||          candidate.startsWith(            staticRoot.endsWith(path.sep) ? staticRoot : `${staticRoot}${path.sep}`,          );         let filePath = path.resolve(staticRoot, staticRelativePath);        if (!isWithinStaticRoot(filePath)) {          respond(400, { "Content-Type": "text/plain" }, "Invalid static file path");          return;        }         const ext = path.extname(filePath);        if (!ext) {          filePath = path.resolve(filePath, "index.html");          if (!isWithinStaticRoot(filePath)) {            respond(400, { "Content-Type": "text/plain" }, "Invalid static file path");            return;          }        }         const fileInfo = yield* fileSystem          .stat(filePath)          .pipe(Effect.catch(() => Effect.succeed(null)));        if (!fileInfo || fileInfo.type !== "File") {          const indexPath = path.resolve(staticRoot, "index.html");          const indexData = yield* fileSystem            .readFile(indexPath)            .pipe(Effect.catch(() => Effect.succeed(null)));          if (!indexData) {            respond(404, { "Content-Type": "text/plain" }, "Not Found");            return;          }          respond(200, { "Content-Type": "text/html; charset=utf-8" }, indexData);          return;        }         const contentType = Mime.getType(filePath) ?? "application/octet-stream";        const data = yield* fileSystem          .readFile(filePath)          .pipe(Effect.catch(() => Effect.succeed(null)));        if (!data) {          respond(500, { "Content-Type": "text/plain" }, "Internal Server Error");          return;        }        respond(200, { "Content-Type": contentType }, data);      }),    ).catch(() => {      if (!res.headersSent) {        respond(500, { "Content-Type": "text/plain" }, "Internal Server Error");      }    });  });   // WebSocket server — upgrades from the HTTP server  const wss = new WebSocketServer({ noServer: true });   const closeWebSocketServer = Effect.callback<void, ServerLifecycleError>((resume) => {    wss.close((error) => {      if (error && !isServerNotRunningError(error)) {        resume(          Effect.fail(            new ServerLifecycleError({ operation: "closeWebSocketServer", cause: error }),          ),        );      } else {        resume(Effect.void);      }    });  });   const closeAllClients = Ref.get(clients).pipe(    Effect.flatMap(Effect.forEach((client) => Effect.sync(() => client.close()))),    Effect.flatMap(() => Ref.set(clients, new Set())),  );   const listenOptions = host ? { host, port } : { port };   const orchestrationEngine = yield* OrchestrationEngineService;  const projectionReadModelQuery = yield* ProjectionSnapshotQuery;  const checkpointDiffQuery = yield* CheckpointDiffQuery;  const liveProviderService = yield* ProviderService;  const orchestrationReactor = yield* OrchestrationReactor;  const { openInEditor } = yield* Open;   const subscriptionsScope = yield* Scope.make("sequential");  yield* Effect.addFinalizer(() => Scope.close(subscriptionsScope, Exit.void));   yield* Stream.runForEach(orchestrationEngine.streamDomainEvents, (event) =>    broadcastPush({      type: "push",      channel: ORCHESTRATION_WS_CHANNELS.domainEvent,      data: event,    }),  ).pipe(Effect.forkIn(subscriptionsScope));   yield* Stream.runForEach(keybindingsManager.changes, (event) =>    broadcastPush({      type: "push",      channel: WS_CHANNELS.serverConfigUpdated,      data: event,    }),  ).pipe(Effect.forkIn(subscriptionsScope));   yield* Scope.provide(orchestrationReactor.start, subscriptionsScope);   let welcomeBootstrapProjectId: ProjectId | undefined;  let welcomeBootstrapThreadId: ThreadId | undefined;   if (autoBootstrapProjectFromCwd) {    yield* Effect.gen(function* () {      const snapshot = yield* projectionReadModelQuery.getSnapshot();      const existingProject = snapshot.projects.find(        (project) => project.workspaceRoot === cwd && project.deletedAt === null,      );      let bootstrapProjectId: ProjectId;      let bootstrapProjectDefaultModel: string;       if (!existingProject) {        const createdAt = new Date().toISOString();        bootstrapProjectId = ProjectId.makeUnsafe(crypto.randomUUID());        const bootstrapProjectTitle = path.basename(cwd) || "project";        bootstrapProjectDefaultModel = "gpt-5-codex";        yield* orchestrationEngine.dispatch({          type: "project.create",          commandId: CommandId.makeUnsafe(crypto.randomUUID()),          projectId: bootstrapProjectId,          title: bootstrapProjectTitle,          workspaceRoot: cwd,          defaultModel: bootstrapProjectDefaultModel,          createdAt,        });      } else {        bootstrapProjectId = existingProject.id;        bootstrapProjectDefaultModel = existingProject.defaultModel ?? "gpt-5-codex";      }       const existingThread = snapshot.threads.find(        (thread) => thread.projectId === bootstrapProjectId && thread.deletedAt === null,      );      if (!existingThread) {        const createdAt = new Date().toISOString();        const threadId = ThreadId.makeUnsafe(crypto.randomUUID());        yield* orchestrationEngine.dispatch({          type: "thread.create",          commandId: CommandId.makeUnsafe(crypto.randomUUID()),          threadId,          projectId: bootstrapProjectId,          title: "New thread",          model: bootstrapProjectDefaultModel,          branch: null,          worktreePath: null,          createdAt,        });        welcomeBootstrapProjectId = bootstrapProjectId;        welcomeBootstrapThreadId = threadId;      } else {        welcomeBootstrapProjectId = bootstrapProjectId;        welcomeBootstrapThreadId = existingThread.id;      }    }).pipe(      Effect.mapError(        (cause) => new ServerLifecycleError({ operation: "autoBootstrapProject", cause }),      ),    );  }

Lines 1–500 of 791. Load more