fix(api): proxy mission-control routes to orchestrator

apps/api/src/mission-control-proxy/mission-control-proxy.controller.ts

@@ -0,0 +1,286 @@
+import {
+  Body,
+  Controller,
+  Get,
+  Logger,
+  Param,
+  Post,
+  Req,
+  Res,
+  ServiceUnavailableException,
+  UseGuards,
+} from "@nestjs/common";
+import type { Request, Response } from "express";
+import { AuthGuard } from "../auth/guards/auth.guard";
+
+const ORCHESTRATOR_URL_KEY = "ORCHESTRATOR_URL";
+const ORCHESTRATOR_API_KEY = "ORCHESTRATOR_API_KEY";
+
+@Controller("mission-control")
+@UseGuards(AuthGuard)
+export class MissionControlProxyController {
+  private readonly logger = new Logger(MissionControlProxyController.name);
+  private readonly orchestratorUrl: string;
+  private readonly orchestratorApiKey: string;
+
+  constructor() {
+    this.orchestratorUrl = this.requireEnv(ORCHESTRATOR_URL_KEY);
+    this.orchestratorApiKey = this.requireEnv(ORCHESTRATOR_API_KEY);
+  }
+
+  @Get("sessions")
+  proxySessions(@Req() req: Request, @Res() res: Response): Promise<void> {
+    return this.proxyRequest("GET", "sessions", req, res);
+  }
+
+  @Get("sessions/:sessionId")
+  proxySession(
+    @Param("sessionId") sessionId: string,
+    @Req() req: Request,
+    @Res() res: Response
+  ): Promise<void> {
+    return this.proxyRequest("GET", `sessions/${sessionId}`, req, res);
+  }
+
+  @Get("sessions/:sessionId/messages")
+  proxyMessages(
+    @Param("sessionId") sessionId: string,
+    @Req() req: Request,
+    @Res() res: Response
+  ): Promise<void> {
+    return this.proxyRequest("GET", `sessions/${sessionId}/messages`, req, res);
+  }
+
+  @Get("audit-log")
+  proxyAuditLog(@Req() req: Request, @Res() res: Response): Promise<void> {
+    return this.proxyRequest("GET", "audit-log", req, res);
+  }
+
+  @Post("sessions/:sessionId/inject")
+  proxyInject(
+    @Param("sessionId") sessionId: string,
+    @Body() body: unknown,
+    @Req() req: Request,
+    @Res() res: Response
+  ): Promise<void> {
+    return this.proxyRequest("POST", `sessions/${sessionId}/inject`, req, res, body);
+  }
+
+  @Post("sessions/:sessionId/pause")
+  proxyPause(
+    @Param("sessionId") sessionId: string,
+    @Req() req: Request,
+    @Res() res: Response
+  ): Promise<void> {
+    return this.proxyRequest("POST", `sessions/${sessionId}/pause`, req, res);
+  }
+
+  @Post("sessions/:sessionId/resume")
+  proxyResume(
+    @Param("sessionId") sessionId: string,
+    @Req() req: Request,
+    @Res() res: Response
+  ): Promise<void> {
+    return this.proxyRequest("POST", `sessions/${sessionId}/resume`, req, res);
+  }
+
+  @Post("sessions/:sessionId/kill")
+  proxyKill(
+    @Param("sessionId") sessionId: string,
+    @Body() body: unknown,
+    @Req() req: Request,
+    @Res() res: Response
+  ): Promise<void> {
+    return this.proxyRequest("POST", `sessions/${sessionId}/kill`, req, res, body);
+  }
+
+  @Get("sessions/:sessionId/stream")
+  async proxySessionStream(
+    @Param("sessionId") sessionId: string,
+    @Req() req: Request,
+    @Res() res: Response
+  ): Promise<void> {
+    const abortController = new AbortController();
+    req.once("close", () => {
+      abortController.abort();
+    });
+
+    try {
+      const upstream = await this.fetchUpstream(
+        "GET",
+        `sessions/${sessionId}/stream`,
+        req.query,
+        undefined,
+        abortController.signal
+      );
+
+      if (!upstream.ok || !upstream.body) {
+        await this.sendStandardResponse(upstream, res);
+        return;
+      }
+
+      res.status(upstream.status);
+      this.copyHeaderIfPresent(upstream, res, "content-type");
+      this.copyHeaderIfPresent(upstream, res, "cache-control");
+      this.copyHeaderIfPresent(upstream, res, "connection");
+      res.setHeader("X-Accel-Buffering", "no");
+
+      if (typeof res.flushHeaders === "function") {
+        res.flushHeaders();
+      }
+
+      for await (const chunk of upstream.body as unknown as AsyncIterable<Uint8Array>) {
+        if (res.writableEnded || res.destroyed) {
+          break;
+        }
+
+        res.write(Buffer.from(chunk));
+      }
+
+      if (!res.writableEnded && !res.destroyed) {
+        res.end();
+      }
+    } catch (error: unknown) {
+      if (this.isAbortError(error)) {
+        return;
+      }
+
+      const message = error instanceof Error ? error.message : String(error);
+      this.logger.warn(`Mission Control stream proxy request failed: ${message}`);
+
+      if (!res.headersSent) {
+        res.status(503).json({ message: "Failed to proxy Mission Control request" });
+      } else if (!res.writableEnded && !res.destroyed) {
+        res.end();
+      }
+    }
+  }
+
+  private async proxyRequest(
+    method: "GET" | "POST",
+    path: string,
+    req: Request,
+    res: Response,
+    body?: unknown
+  ): Promise<void> {
+    const abortController = new AbortController();
+    req.once("close", () => {
+      abortController.abort();
+    });
+
+    try {
+      const upstream = await this.fetchUpstream(
+        method,
+        path,
+        req.query,
+        body,
+        abortController.signal
+      );
+      await this.sendStandardResponse(upstream, res);
+    } catch (error: unknown) {
+      if (this.isAbortError(error)) {
+        return;
+      }
+
+      this.handleProxyError(error);
+    }
+  }
+
+  private async fetchUpstream(
+    method: "GET" | "POST",
+    path: string,
+    query: Request["query"],
+    body: unknown,
+    signal: AbortSignal
+  ): Promise<globalThis.Response> {
+    const url = new URL(`/api/mission-control/${path}`, this.orchestratorUrl);
+    this.appendQueryParams(url.searchParams, query);
+
+    const headers: Record<string, string> = {
+      "X-API-Key": this.orchestratorApiKey,
+    };
+
+    const requestInit: RequestInit = {
+      method,
+      headers,
+      signal,
+    };
+
+    if (method === "POST" && body !== undefined) {
+      headers["Content-Type"] = "application/json";
+      requestInit.body = JSON.stringify(body);
+    }
+
+    return fetch(url.toString(), requestInit);
+  }
+
+  private appendQueryParams(searchParams: URLSearchParams, query: Request["query"]): void {
+    for (const [key, value] of Object.entries(query)) {
+      this.appendQueryValue(searchParams, key, value);
+    }
+  }
+
+  private appendQueryValue(searchParams: URLSearchParams, key: string, value: unknown): void {
+    if (value === undefined || value === null) {
+      return;
+    }
+
+    if (Array.isArray(value)) {
+      for (const item of value) {
+        this.appendQueryValue(searchParams, key, item);
+      }
+      return;
+    }
+
+    if (typeof value === "string" || typeof value === "number" || typeof value === "boolean") {
+      searchParams.append(key, String(value));
+    }
+  }
+
+  private async sendStandardResponse(upstream: globalThis.Response, res: Response): Promise<void> {
+    res.status(upstream.status);
+    this.copyHeaderIfPresent(upstream, res, "content-type");
+    this.copyHeaderIfPresent(upstream, res, "cache-control");
+    this.copyHeaderIfPresent(upstream, res, "location");
+
+    const responseText = await upstream.text();
+
+    if (responseText.length === 0) {
+      res.end();
+      return;
+    }
+
+    res.send(responseText);
+  }
+
+  private copyHeaderIfPresent(
+    upstream: globalThis.Response,
+    res: Response,
+    headerName: string
+  ): void {
+    const value = upstream.headers.get(headerName);
+    if (value) {
+      res.setHeader(headerName, value);
+    }
+  }
+
+  private handleProxyError(error: unknown): never {
+    const message = error instanceof Error ? error.message : String(error);
+    this.logger.warn(`Mission Control proxy request failed: ${message}`);
+    throw new ServiceUnavailableException("Failed to proxy Mission Control request");
+  }
+
+  private isAbortError(error: unknown): boolean {
+    return error instanceof Error && error.name === "AbortError";
+  }
+
+  private requireEnv(key: string): string {
+    const value = process.env[key];
+
+    if (typeof value !== "string" || value.trim().length === 0) {
+      throw new Error(`@mosaic/api: ${key} is required. Set it in your config or via ${key}.`);
+    }
+
+    return value;
+  }
+}

apps/api/src/mission-control-proxy/mission-control-proxy.module.ts

@@ -0,0 +1,9 @@
+import { Module } from "@nestjs/common";
+import { AuthModule } from "../auth/auth.module";
+import { MissionControlProxyController } from "./mission-control-proxy.controller";
+
+@Module({
+  imports: [AuthModule],
+  controllers: [MissionControlProxyController],
+})
+export class MissionControlProxyModule {}