Refactor HttpApi workspace routing and proxy boundaries (#25006)

2026-04-29 16:50:54 -04:00
parent c9db964e9b
commit 09c65338df
27 changed files with 980 additions and 345 deletions
@@ -0,0 +1,71 @@
+import { Effect, Encoding, Layer, Redacted, Schema } from "effect"
+import { HttpApiMiddleware, HttpApiSecurity } from "effect/unstable/httpapi"
+import { Flag } from "@opencode-ai/core/flag/flag"
+
+class Unauthorized extends Schema.TaggedErrorClass<Unauthorized>()(
+  "Unauthorized",
+  { message: Schema.String },
+  { httpApiStatus: 401 },
+) {}
+
+export class Authorization extends HttpApiMiddleware.Service<Authorization>()(
+  "@opencode/ExperimentalHttpApiAuthorization",
+  {
+    error: Unauthorized,
+    security: {
+      basic: HttpApiSecurity.basic,
+      authToken: HttpApiSecurity.apiKey({ in: "query", key: "auth_token" }),
+    },
+  },
+) {}
+
+const emptyCredential = {
+  username: "",
+  password: Redacted.make(""),
+}
+
+function validateCredential<A, E, R>(
+  effect: Effect.Effect<A, E, R>,
+  credential: { readonly username: string; readonly password: typeof emptyCredential.password },
+) {
+  return Effect.gen(function* () {
+    if (!Flag.OPENCODE_SERVER_PASSWORD) return yield* effect
+
+    if (credential.username !== (Flag.OPENCODE_SERVER_USERNAME ?? "opencode")) {
+      return yield* new Unauthorized({ message: "Unauthorized" })
+    }
+    if (Redacted.value(credential.password) !== Flag.OPENCODE_SERVER_PASSWORD) {
+      return yield* new Unauthorized({ message: "Unauthorized" })
+    }
+    return yield* effect
+  })
+}
+
+function decodeCredential(input: string) {
+  return Encoding.decodeBase64String(input)
+    .asEffect()
+    .pipe(
+      Effect.match({
+        onFailure: () => emptyCredential,
+        onSuccess: (header) => {
+          const parts = header.split(":")
+          if (parts.length !== 2) return emptyCredential
+          return {
+            username: parts[0],
+            password: Redacted.make(parts[1]),
+          }
+        },
+      }),
+    )
+}
+
+export const authorizationLayer = Layer.succeed(
+  Authorization,
+  Authorization.of({
+    basic: (effect, { credential }) => validateCredential(effect, credential),
+    authToken: (effect, { credential }) =>
+      Effect.gen(function* () {
+        return yield* validateCredential(effect, yield* decodeCredential(Redacted.value(credential)))
+      }),
+  }),
+)
@@ -0,0 +1,55 @@
+import { InstanceRef, WorkspaceRef } from "@/effect/instance-ref"
+import { AppRuntime } from "@/effect/app-runtime"
+import { InstanceBootstrap } from "@/project/bootstrap"
+import { Instance } from "@/project/instance"
+import type { InstanceContext } from "@/project/instance"
+import { Filesystem } from "@/util/filesystem"
+import { Effect, Layer } from "effect"
+import { HttpRouter, HttpServerResponse } from "effect/unstable/http"
+import { HttpApiMiddleware } from "effect/unstable/httpapi"
+import { WorkspaceRouteContext } from "./workspace-routing"
+
+export class InstanceContextMiddleware extends HttpApiMiddleware.Service<
+  InstanceContextMiddleware,
+  {
+    requires: WorkspaceRouteContext
+  }
+>()("@opencode/ExperimentalHttpApiInstanceContext") {}
+
+function decode(input: string): string {
+  try {
+    return decodeURIComponent(input)
+  } catch {
+    return input
+  }
+}
+
+function makeInstanceContext(directory: string): Effect.Effect<InstanceContext> {
+  return Effect.promise(() =>
+    Instance.provide({
+      directory: Filesystem.resolve(decode(directory)),
+      init: () => AppRuntime.runPromise(InstanceBootstrap),
+      fn: () => Instance.current,
+    }),
+  )
+}
+
+function provideInstanceContext<E>(
+  effect: Effect.Effect<HttpServerResponse.HttpServerResponse, E>,
+): Effect.Effect<HttpServerResponse.HttpServerResponse, E, WorkspaceRouteContext> {
+  return Effect.gen(function* () {
+    const route = yield* WorkspaceRouteContext
+    const ctx = yield* makeInstanceContext(route.directory)
+    return yield* effect.pipe(
+      Effect.provideService(InstanceRef, ctx),
+      Effect.provideService(WorkspaceRef, route.workspaceID),
+    )
+  })
+}
+
+export const instanceContextLayer = Layer.succeed(
+  InstanceContextMiddleware,
+  InstanceContextMiddleware.of((effect) => provideInstanceContext(effect)),
+)
+
+export const instanceRouterMiddleware = HttpRouter.middleware()((effect) => provideInstanceContext(effect))
@@ -0,0 +1,86 @@
+import { ProxyUtil } from "@/server/proxy-util"
+import { Effect, Stream } from "effect"
+import { FetchHttpClient, HttpBody, HttpClient, HttpClientRequest, HttpServerRequest, HttpServerResponse } from "effect/unstable/http"
+import * as Socket from "effect/unstable/socket/Socket"
+
+function webSource(request: HttpServerRequest.HttpServerRequest): Request | undefined {
+  return request.source instanceof Request ? request.source : undefined
+}
+
+function requestBody(request: HttpServerRequest.HttpServerRequest) {
+  if (request.method === "GET" || request.method === "HEAD") return HttpBody.empty
+  const len = request.headers["content-length"]
+  return HttpBody.raw(webSource(request)?.body ?? null, {
+    contentType: request.headers["content-type"],
+    contentLength: len ? Number(len) : undefined,
+  })
+}
+
+export function websocket(
+  request: HttpServerRequest.HttpServerRequest,
+  target: string | URL,
+): Effect.Effect<HttpServerResponse.HttpServerResponse, never, Socket.WebSocketConstructor> {
+  return Effect.scoped(
+    Effect.gen(function* () {
+      const inbound = yield* Effect.orDie(request.upgrade)
+      const outbound = yield* Socket.makeWebSocket(ProxyUtil.websocketTargetURL(target), {
+        protocols: ProxyUtil.websocketProtocols(request.headers),
+      })
+      const writeInbound = yield* inbound.writer
+      const writeOutbound = yield* outbound.writer
+
+      yield* outbound
+        .runRaw((message) => writeInbound(message))
+        .pipe(
+          Effect.catchReason("SocketError", "SocketCloseError", (reason) =>
+            writeInbound(new Socket.CloseEvent(reason.code, reason.closeReason)).pipe(Effect.catch(() => Effect.void)),
+          ),
+          Effect.catch(() => writeInbound(new Socket.CloseEvent(1011, "proxy error")).pipe(Effect.catch(() => Effect.void))),
+          Effect.forkScoped,
+        )
+
+      yield* inbound
+        .runRaw((message) => {
+          return writeOutbound(typeof message === "string" ? message : message.slice())
+        })
+        .pipe(
+          Effect.catch(() => Effect.void),
+          Effect.ensuring(writeOutbound(new Socket.CloseEvent()).pipe(Effect.catch(() => Effect.void))),
+        )
+      return HttpServerResponse.empty()
+    }).pipe(Effect.orDie),
+  )
+}
+
+function statusText(response: unknown) {
+  return (response as { source?: Response }).source?.statusText
+}
+
+export function http(
+  url: string | URL,
+  extra: HeadersInit | undefined,
+  request: HttpServerRequest.HttpServerRequest,
+): Effect.Effect<HttpServerResponse.HttpServerResponse> {
+  return Effect.gen(function* () {
+    const response = yield* HttpClient.execute(
+      HttpClientRequest.make(request.method as never)(url, {
+        headers: ProxyUtil.headers(request.headers as HeadersInit, extra),
+        body: requestBody(request),
+      }),
+    )
+    const headers = new Headers(response.headers as HeadersInit)
+    headers.delete("content-encoding")
+    headers.delete("content-length")
+
+    return HttpServerResponse.stream(response.stream.pipe(Stream.catchCause(() => Stream.empty)), {
+      status: response.status,
+      statusText: statusText(response),
+      headers,
+    })
+  }).pipe(
+    Effect.provide(FetchHttpClient.layer),
+    Effect.catch(() => Effect.succeed(HttpServerResponse.empty({ status: 500 }))),
+  )
+}
+
+export * as HttpApiProxy from "./proxy"
@@ -0,0 +1,212 @@
+import { getAdaptor } from "@/control-plane/adaptors"
+import { WorkspaceID } from "@/control-plane/schema"
+import type { Target } from "@/control-plane/types"
+import { Workspace } from "@/control-plane/workspace"
+import { Instance } from "@/project/instance"
+import { Session } from "@/session/session"
+import { HttpApiProxy } from "./proxy"
+import * as Fence from "@/server/fence"
+import { getWorkspaceRouteSessionID, isLocalWorkspaceRoute, workspaceProxyURL } from "@/server/workspace"
+import { Flag } from "@opencode-ai/core/flag/flag"
+import { Context, Data, Effect, Layer } from "effect"
+import { HttpRouter, HttpServerRequest, HttpServerResponse } from "effect/unstable/http"
+import { HttpApiMiddleware } from "effect/unstable/httpapi"
+import * as Socket from "effect/unstable/socket/Socket"
+
+type RemoteTarget = Extract<Target, { type: "remote" }>
+
+type RequestPlan = Data.TaggedEnum<{
+  MissingWorkspace: { readonly workspaceID: WorkspaceID }
+  Local: { readonly directory: string; readonly workspaceID?: WorkspaceID }
+  Remote: {
+    readonly request: HttpServerRequest.HttpServerRequest
+    readonly workspace: Workspace.Info
+    readonly target: RemoteTarget
+    readonly url: URL
+  }
+}>
+const RequestPlan = Data.taggedEnum<RequestPlan>()
+
+export class WorkspaceRouteContext extends Context.Service<WorkspaceRouteContext, {
+  readonly directory: string
+  readonly workspaceID?: WorkspaceID
+}>()("@opencode/ExperimentalHttpApiWorkspaceRouteContext") {}
+
+export class WorkspaceRoutingMiddleware extends HttpApiMiddleware.Service<
+  WorkspaceRoutingMiddleware,
+  {
+    provides: WorkspaceRouteContext
+    requires: Session.Service
+  }
+>()("@opencode/ExperimentalHttpApiWorkspaceRouting") {}
+
+function currentDirectory(): string {
+  try {
+    return Instance.directory
+  } catch {
+    return process.cwd()
+  }
+}
+
+function requestURL(request: HttpServerRequest.HttpServerRequest): URL {
+  return new URL(request.url, "http://localhost")
+}
+
+function configuredWorkspaceID(): WorkspaceID | undefined {
+  return Flag.OPENCODE_WORKSPACE_ID ? WorkspaceID.make(Flag.OPENCODE_WORKSPACE_ID) : undefined
+}
+
+function selectedWorkspaceID(url: URL, sessionWorkspaceID?: WorkspaceID): WorkspaceID | undefined {
+  const workspaceParam = url.searchParams.get("workspace")
+  return sessionWorkspaceID ?? (workspaceParam ? WorkspaceID.make(workspaceParam) : undefined)
+}
+
+function defaultDirectory(request: HttpServerRequest.HttpServerRequest, url: URL): string {
+  return url.searchParams.get("directory") || request.headers["x-opencode-directory"] || currentDirectory()
+}
+
+function shouldStayOnControlPlane(request: HttpServerRequest.HttpServerRequest, url: URL): boolean {
+  return isLocalWorkspaceRoute(request.method, url.pathname) || url.pathname.startsWith("/console")
+}
+
+function resolveWorkspace(
+  id: WorkspaceID | undefined,
+  envWorkspaceID: WorkspaceID | undefined,
+): Effect.Effect<Workspace.Info | void> {
+  if (!id || envWorkspaceID) return Effect.void
+  return Effect.promise(() => Workspace.get(id))
+}
+
+function missingWorkspaceResponse(id: WorkspaceID): HttpServerResponse.HttpServerResponse {
+  return HttpServerResponse.text(`Workspace not found: ${id}`, {
+    status: 500,
+    contentType: "text/plain; charset=utf-8",
+  })
+}
+
+function resolveTarget(workspace: Workspace.Info): Effect.Effect<Target> {
+  return Effect.gen(function* () {
+    const adaptor = yield* Effect.promise(() => getAdaptor(workspace.projectID, workspace.type))
+    return yield* Effect.promise(() => Promise.resolve(adaptor.target(workspace)))
+  })
+}
+
+function proxyRemote(
+  request: HttpServerRequest.HttpServerRequest,
+  workspace: Workspace.Info,
+  target: RemoteTarget,
+  url: URL,
+): Effect.Effect<HttpServerResponse.HttpServerResponse, never, Socket.WebSocketConstructor> {
+  return Effect.gen(function* () {
+    const syncing = yield* Effect.promise(() => Workspace.isSyncing(workspace.id))
+    if (!syncing) {
+      return HttpServerResponse.text(`broken sync connection for workspace: ${workspace.id}`, {
+        status: 503,
+        contentType: "text/plain; charset=utf-8",
+      })
+    }
+    const proxyURL = workspaceProxyURL(target.url, url)
+    const headers = request.headers as Record<string, string>
+    if (headers["upgrade"]?.toLowerCase() === "websocket") return yield* HttpApiProxy.websocket(request, proxyURL)
+    const response = yield* HttpApiProxy.http(proxyURL, target.headers, request)
+    const sync = Fence.parse(new Headers(response.headers))
+    if (sync) yield* Effect.promise(() => Fence.wait(workspace.id, sync, request.source instanceof Request ? request.source.signal : undefined))
+    return response
+  })
+}
+
+function planWorkspaceRequest(
+  request: HttpServerRequest.HttpServerRequest,
+  url: URL,
+  workspace: Workspace.Info,
+): Effect.Effect<RequestPlan> {
+  return Effect.gen(function* () {
+    const target = yield* resolveTarget(workspace)
+    if (target.type === "remote") return RequestPlan.Remote({ request, workspace, target, url })
+    return RequestPlan.Local({ directory: target.directory, workspaceID: workspace.id })
+  })
+}
+
+function planRequest(
+  request: HttpServerRequest.HttpServerRequest,
+  sessionWorkspaceID?: WorkspaceID,
+): Effect.Effect<RequestPlan> {
+  return Effect.gen(function* () {
+    const url = requestURL(request)
+    const envWorkspaceID = configuredWorkspaceID()
+    const workspaceID = selectedWorkspaceID(url, sessionWorkspaceID)
+    const workspace = yield* resolveWorkspace(workspaceID, envWorkspaceID)
+
+    if (workspaceID && workspace === undefined && !envWorkspaceID) {
+      return RequestPlan.MissingWorkspace({ workspaceID })
+    }
+
+    if (workspace !== undefined && !envWorkspaceID && !shouldStayOnControlPlane(request, url)) {
+      return yield* planWorkspaceRequest(request, url, workspace)
+    }
+
+    return RequestPlan.Local({ directory: defaultDirectory(request, url), workspaceID: envWorkspaceID ?? workspaceID })
+  })
+}
+
+function routeWorkspace<E>(
+  effect: Effect.Effect<HttpServerResponse.HttpServerResponse, E, WorkspaceRouteContext>,
+  plan: RequestPlan,
+): Effect.Effect<HttpServerResponse.HttpServerResponse, E, Socket.WebSocketConstructor> {
+  return RequestPlan.$match(plan, {
+    MissingWorkspace: ({ workspaceID }) => Effect.succeed(missingWorkspaceResponse(workspaceID)),
+    Remote: ({ request, workspace, target, url }) => proxyRemote(request, workspace, target, url),
+    Local: ({ directory, workspaceID }) =>
+      effect.pipe(
+        Effect.provideService(WorkspaceRouteContext, WorkspaceRouteContext.of({ directory, workspaceID })),
+      ),
+  })
+}
+
+function routeWorkspaceRequest<E>(
+  effect: Effect.Effect<HttpServerResponse.HttpServerResponse, E, WorkspaceRouteContext>,
+  request: HttpServerRequest.HttpServerRequest,
+  sessionWorkspaceID?: WorkspaceID,
+): Effect.Effect<HttpServerResponse.HttpServerResponse, E, Socket.WebSocketConstructor> {
+  return Effect.flatMap(planRequest(request, sessionWorkspaceID), (plan) => routeWorkspace(effect, plan))
+}
+
+function routeHttpApiWorkspace<E>(
+  effect: Effect.Effect<HttpServerResponse.HttpServerResponse, E, WorkspaceRouteContext>,
+): Effect.Effect<
+  HttpServerResponse.HttpServerResponse,
+  E,
+  Session.Service | HttpServerRequest.HttpServerRequest | Socket.WebSocketConstructor
+> {
+  return Effect.gen(function* () {
+    const request = yield* HttpServerRequest.HttpServerRequest
+    const sessionID = getWorkspaceRouteSessionID(requestURL(request))
+    const session = sessionID
+      ? yield* Session.Service.use((svc) => svc.get(sessionID)).pipe(Effect.catchDefect(() => Effect.void))
+      : undefined
+    return yield* routeWorkspaceRequest(effect, request, session?.workspaceID)
+  })
+}
+
+export const workspaceRoutingLayer = Layer.effect(
+  WorkspaceRoutingMiddleware,
+  Effect.gen(function* () {
+    const makeWebSocket = yield* Socket.WebSocketConstructor
+    return WorkspaceRoutingMiddleware.of((effect) =>
+      routeHttpApiWorkspace(effect).pipe(Effect.provideService(Socket.WebSocketConstructor, makeWebSocket)),
+    )
+  }),
+)
+
+export const workspaceRouterMiddleware = HttpRouter.middleware<{ provides: WorkspaceRouteContext }>()(
+  Effect.gen(function* () {
+    const makeWebSocket = yield* Socket.WebSocketConstructor
+    return (effect) =>
+      Effect.gen(function* () {
+        const request = yield* HttpServerRequest.HttpServerRequest
+        return yield* routeWorkspaceRequest(effect, request).pipe(
+          Effect.provideService(Socket.WebSocketConstructor, makeWebSocket),
+        )
+      })
+  }),
+)