Prepare Effect HttpApi backend parity (#24853)

2026-04-29 09:34:50 -04:00
parent 65ba1f6c13
commit 6015084fa2
98 changed files with 4290 additions and 2766 deletions
@@ -0,0 +1,156 @@
+import { Config } from "@/config/config"
+import { GlobalBus, type GlobalEvent as GlobalBusEvent } from "@/bus/global"
+import { Installation } from "@/installation"
+import { Instance } from "@/project/instance"
+import { InstallationVersion } from "@opencode-ai/core/installation/version"
+import * as Log from "@opencode-ai/core/util/log"
+import { Effect, Queue, Schema } from "effect"
+import * as Stream from "effect/Stream"
+import { HttpServerRequest, HttpServerResponse } from "effect/unstable/http"
+import { HttpApiBuilder } from "effect/unstable/httpapi"
+import * as Sse from "effect/unstable/encoding/Sse"
+import { RootHttpApi } from "../api"
+import { GlobalUpgradeInput } from "../groups/global"
+
+const log = Log.create({ service: "server" })
+
+function eventData(data: unknown): Sse.Event {
+  return {
+    _tag: "Event",
+    event: "message",
+    id: undefined,
+    data: JSON.stringify(data),
+  }
+}
+
+function parseBody(body: string) {
+  try {
+    return JSON.parse(body || "{}") as unknown
+  } catch {
+    return undefined
+  }
+}
+
+function eventResponse() {
+  log.info("global event connected")
+  const events = Stream.callback<GlobalBusEvent>((queue) => {
+    const handler = (event: GlobalBusEvent) => Queue.offerUnsafe(queue, event)
+    return Effect.acquireRelease(
+      Effect.sync(() => GlobalBus.on("event", handler)),
+      () => Effect.sync(() => GlobalBus.off("event", handler)),
+    )
+  })
+  const heartbeat = Stream.tick("10 seconds").pipe(
+    Stream.drop(1),
+    Stream.map(() => ({ payload: { type: "server.heartbeat", properties: {} } })),
+  )
+
+  return HttpServerResponse.stream(
+    Stream.make({ payload: { type: "server.connected", properties: {} } }).pipe(
+      Stream.concat(events.pipe(Stream.merge(heartbeat, { haltStrategy: "left" }))),
+      Stream.map(eventData),
+      Stream.pipeThroughChannel(Sse.encode()),
+      Stream.encodeText,
+      Stream.ensuring(Effect.sync(() => log.info("global event disconnected"))),
+    ),
+    {
+      contentType: "text/event-stream",
+      headers: {
+        "Cache-Control": "no-cache, no-transform",
+        "X-Accel-Buffering": "no",
+        "X-Content-Type-Options": "nosniff",
+      },
+    },
+  )
+}
+
+export const globalHandlers = HttpApiBuilder.group(RootHttpApi, "global", (handlers) =>
+    Effect.gen(function* () {
+      const config = yield* Config.Service
+      const installation = yield* Installation.Service
+
+      const health = Effect.fn("GlobalHttpApi.health")(function* () {
+        return { healthy: true as const, version: InstallationVersion }
+      })
+
+      const event = Effect.fn("GlobalHttpApi.event")(function* () {
+        return eventResponse()
+      })
+
+      const configGet = Effect.fn("GlobalHttpApi.configGet")(function* () {
+        return yield* config.getGlobal()
+      })
+
+      const configUpdate = Effect.fn("GlobalHttpApi.configUpdate")(function* (ctx) {
+        return yield* config.updateGlobal(ctx.payload)
+      })
+
+      const dispose = Effect.fn("GlobalHttpApi.dispose")(function* () {
+        yield* Effect.promise(() => Instance.disposeAll())
+        GlobalBus.emit("event", {
+          directory: "global",
+          payload: { type: "global.disposed", properties: {} },
+        })
+        return true
+      })
+
+      const upgrade = Effect.fn("GlobalHttpApi.upgrade")(function* (ctx: { payload: typeof GlobalUpgradeInput.Type }) {
+        const method = yield* installation.method()
+        if (method === "unknown") {
+          return {
+            status: 400,
+            body: { success: false as const, error: "Unknown installation method" },
+          }
+        }
+        const target = ctx.payload.target || (yield* installation.latest(method))
+        const result = yield* installation.upgrade(method, target).pipe(
+          Effect.as({ status: 200, body: { success: true as const, version: target } }),
+          Effect.catch((err) =>
+            Effect.succeed({
+              status: 500,
+              body: {
+                success: false as const,
+                error: err instanceof Error ? err.message : String(err),
+              },
+            }),
+          ),
+        )
+        if (!result.body.success) return result
+        GlobalBus.emit("event", {
+          directory: "global",
+          payload: {
+            type: Installation.Event.Updated.type,
+            properties: { version: target },
+          },
+        })
+        return result
+      })
+
+      const upgradeRaw = Effect.fn("GlobalHttpApi.upgradeRaw")(function* (ctx: {
+        request: HttpServerRequest.HttpServerRequest
+      }) {
+        const body = yield* Effect.orDie(ctx.request.text)
+        const json = parseBody(body)
+        if (json === undefined) {
+          return HttpServerResponse.jsonUnsafe({ success: false, error: "Invalid request body" }, { status: 400 })
+        }
+        const payload = yield* Schema.decodeUnknownEffect(GlobalUpgradeInput)(json).pipe(
+          Effect.map((payload) => ({ valid: true as const, payload })),
+          Effect.catch(() => Effect.succeed({ valid: false as const })),
+        )
+        if (!payload.valid) {
+          return HttpServerResponse.jsonUnsafe({ success: false, error: "Invalid request body" }, { status: 400 })
+        }
+        const result = yield* upgrade({ payload: payload.payload })
+        return HttpServerResponse.jsonUnsafe(result.body, { status: result.status })
+      })
+
+      return handlers
+        .handle("health", health)
+        .handleRaw("event", event)
+        .handle("configGet", configGet)
+        .handle("configUpdate", configUpdate)
+        .handle("dispose", dispose)
+        .handleRaw("upgrade", upgradeRaw)
+    }),
+)