perf(desktop): talk to the main process without the Effect runtime in the renderer (#50112)

packages/desktop/src/renderer/ipc-client.ts

@@ -1,10 +1,14 @@
-import { Context, Effect, Layer, ManagedRuntime, Queue, Schema, Stream } from "effect"
-import { RpcClient, RpcMessage } from "effect/unstable/rpc"
-import { DesktopRpcs, type DesktopRpcClient } from "../shared/ipc-rpc"
+import type { Effect } from "effect"
+import type { RpcMessage } from "effect/unstable/rpc"
+import type { DesktopRpcClient } from "../shared/ipc-rpc"
 import type { DesktopEvent } from "../shared/ipc-rpc/events"
 import { IpcTransportPort } from "../shared/ipc-transport"
 
-class DesktopClient extends Context.Service<DesktopClient, DesktopRpcClient>()("opencode/desktop/DesktopClient") {}
+// The main process serves Effect's RpcServer over a MessagePort; this side speaks its wire format
+// directly. Messages cross by structured clone (no serialization layer, binary stays binary), every
+// payload in the contract is JSON-native or a Uint8Array, and the main process is trusted, so the
+// renderer needs neither the Effect runtime nor the contract's schemas to talk to it. Keeping them
+// out of the renderer's initial module graph is worth about a third of its startup script.
 
 type EventTag = DesktopEvent["_tag"]
 type InvokeTag = Exclude<keyof DesktopRpcClient, "DesktopEvents">
@@ -13,55 +17,48 @@ type InvokeResult<Tag extends InvokeTag> =
   ReturnType<DesktopRpcClient[Tag]> extends Effect.Effect<infer Value, unknown> ? Value : never
 type EventValue<Tag extends EventTag> = Extract<DesktopEvent, { readonly _tag: Tag }>
 
+type Pending = {
+  readonly resolve: (value: unknown) => void
+  readonly reject: (error: unknown) => void
+  readonly chunk?: (values: ReadonlyArray<unknown>) => void
+}
+
+const pending = new Map<number, Pending>()
+const listeners = new Map<EventTag, Set<(value: unknown) => void>>()
+const beforeDispose = new Set<() => Promise<unknown> | void>()
+let nextId = 0
+
 const port = new Promise<MessagePort>((resolve) => {
   const onMessage = (event: MessageEvent) => {
     if (event.source !== window || event.data !== IpcTransportPort) return
     const value = event.ports[0]
     if (!value) return
     window.removeEventListener("message", onMessage)
+    value.addEventListener("message", (message) => receive(value, message.data as RpcMessage.FromServerEncoded))
+    value.start()
     resolve(value)
   }
   window.addEventListener("message", onMessage)
 })
 
-const ClientProtocolLive = Layer.unwrap(Effect.promise(() => port).pipe(Effect.map((value) => clientProtocol(value))))
-const ClientLive = Layer.effect(DesktopClient, RpcClient.make(DesktopRpcs)).pipe(Layer.provide(ClientProtocolLive))
-const runtime = ManagedRuntime.make(ClientLive)
-const listeners = new Map<EventTag, Set<(value: unknown) => void>>()
-const beforeDispose = new Set<() => Promise<unknown> | void>()
-// Let queued work (storage flushes) hand its messages to the port before the runtime goes away.
+// Let queued work (storage flushes) hand its messages to the port before it closes.
 window.addEventListener(
   "pagehide",
-  () => void Promise.allSettled([...beforeDispose].map((callback) => callback())).then(() => runtime.dispose()),
+  () => void Promise.allSettled([...beforeDispose].map((callback) => callback())).then(() => port.then((p) => p.close())),
   { once: true },
 )
 
+void request("DesktopEvents", null, (values) => {
+  for (const value of values as ReadonlyArray<DesktopEvent>) listeners.get(value._tag)?.forEach((fn) => fn(value))
+})
+
 export function onBeforeDispose(callback: () => Promise<unknown> | void) {
   beforeDispose.add(callback)
   return () => beforeDispose.delete(callback)
 }
 
-runtime.runFork(
-  Effect.gen(function* () {
-    const client = yield* DesktopClient
-    yield* client
-      .DesktopEvents()
-      .pipe(
-        Stream.runForEach((event) =>
-          Effect.sync(() => listeners.get(event._tag)?.forEach((listener) => listener(event))),
-        ),
-      )
-  }),
-)
-
 export function invoke<Tag extends InvokeTag>(tag: Tag, ...payload: InvokeArgs<Tag>): Promise<InvokeResult<Tag>> {
-  return runtime.runPromise(
-    Effect.gen(function* () {
-      const client = yield* DesktopClient
-      const method = client[tag] as unknown as (...args: ReadonlyArray<unknown>) => Effect.Effect<unknown, unknown>
-      return yield* method(...payload)
-    }),
-  ) as Promise<InvokeResult<Tag>>
+  return request(tag, payload[0] ?? null) as Promise<InvokeResult<Tag>>
 }
 
 export function send<Tag extends InvokeTag>(tag: Tag, ...payload: InvokeArgs<Tag>) {
@@ -79,42 +76,49 @@ export function listen<Tag extends EventTag>(tag: Tag, listener: (value: EventVa
   }
 }
 
-// Structured clone over the port, like Effect's worker protocol: no serialization layer, so binary
-// payloads stay binary. Buffers are cloned rather than transferred: Electron's MessagePortMain
-// drops transferred ArrayBuffers, so a request carrying one would never arrive.
-function clientProtocol(value: MessagePort) {
-  return Layer.effect(
-    RpcClient.Protocol,
-    RpcClient.Protocol.make(
-      Effect.fnUntraced(function* (writeResponse, clientIds) {
-        const inbound = yield* Queue.unbounded<RpcMessage.FromServerEncoded>()
-        const onMessage = (event: MessageEvent) => {
-          Queue.offerUnsafe(inbound, event.data as RpcMessage.FromServerEncoded)
-        }
-        value.addEventListener("message", onMessage)
-        value.start()
-        yield* Effect.addFinalizer(() =>
-          Effect.sync(() => {
-            value.removeEventListener("message", onMessage)
-            value.close()
-          }),
-        )
-        yield* Stream.fromQueue(inbound).pipe(
-          Stream.runForEach((message) =>
-            Effect.forEach(clientIds, (clientId) => writeResponse(clientId, message), { discard: true }),
-          ),
-          Effect.forkScoped,
-        )
-        return {
-          codecFor: Schema.toCodecJson,
-          send: (_clientId, request) =>
-            Effect.sync(() => {
-              value.postMessage(request)
-            }),
-          supportsAck: true,
-          supportsTransferables: false,
-        }
-      }),
-    ),
-  )
+function request(tag: string, payload: unknown, chunk?: Pending["chunk"]) {
+  const id = nextId++
+  return new Promise<unknown>((resolve, reject) => {
+    pending.set(id, { resolve, reject, chunk })
+    const message: RpcMessage.RequestEncoded = { _tag: "Request", id, tag, payload, headers: [] }
+    void port.then((p) => p.postMessage(message))
+  })
+}
+
+function receive(p: MessagePort, message: RpcMessage.FromServerEncoded) {
+  switch (message._tag) {
+    case "Chunk": {
+      pending.get(Number(message.requestId))?.chunk?.(message.values)
+      p.postMessage({ _tag: "Ack", requestId: message.requestId } satisfies RpcMessage.AckEncoded)
+      return
+    }
+    case "Exit": {
+      const id = Number(message.requestId)
+      const entry = pending.get(id)
+      pending.delete(id)
+      if (!entry) return
+      if (message.exit._tag === "Success") return entry.resolve(message.exit.value)
+      return entry.reject(failure(message.exit.cause))
+    }
+    case "Defect": {
+      const error = new Error("Desktop IPC defect", { cause: message.defect })
+      pending.forEach((entry) => entry.reject(error))
+      pending.clear()
+      return
+    }
+    case "ClientProtocolError": {
+      console.error("[desktop-ipc] protocol error", message.error)
+      return
+    }
+  }
+}
+
+// The RPC failure a caller sees is the encoded error the handler failed with, as before; defects
+// and interrupts surface as errors.
+function failure(cause: ReadonlyArray<{ readonly _tag: string; readonly error?: unknown; readonly defect?: unknown }>) {
+  const failed = cause.find((item) => item._tag === "Fail")
+  if (failed) return failed.error
+  const died = cause.find((item) => item._tag === "Die")
+  if (died) return new Error("Desktop IPC handler failed", { cause: died.defect })
+  return new Error("Desktop IPC request interrupted")
 }