1import type * as Effect from "effect/Effect";
2import type * as Stream from "effect/Stream";
3import type { Rpc } from "../../Rpc.ts";
4import { isRpcErrorEnvelope, isRpcStreamEnvelope } from "./Rpc.ts";
5
6export type RpcAsync<Shape> = {
7 [K in keyof Shape as K extends "fetch" ? never : K]: Shape[K] extends (
8 ...args: infer A
9 ) => Effect.Effect<infer T, any, any>
10 ? (...args: A) => Promise<T>
11 : Shape[K] extends (...args: infer A) => Stream.Stream<any, any, any>
12 ? (...args: A) => Promise<ReadableStream<Uint8Array>>
13 : Shape[K] extends Effect.Effect<infer T, any, any>
14 ? Promise<T>
15 : Shape[K] extends Stream.Stream<any, any, any>
16 ? Promise<ReadableStream<Uint8Array>>
17 : Shape[K] extends (...args: infer A) => infer R
18 ? (...args: A) => Promise<Awaited<R>>
19 : Promise<Shape[K]>;
20};
21
22/**
23 * Wrap a Cloudflare Worker RPC stub in a `Promise<T>`-shaped proxy so callers
24 * outside the Effect runtime (TanStack Start route handlers, plain Workers,
25 * test code) can `await` RPC methods directly.
26 *
27 * The proxy:
28 * - leaves `fetch` / `connect` / `Symbol.dispose` and other non-string keys
29 * passing through to the underlying binding unchanged,
30 * - turns each `Effect<T>` / `Stream<T>` method into a `Promise<T>` /
31 * `Promise<ReadableStream<Uint8Array>>`,
32 * - unwraps the wire envelopes produced by the Effect-side RPC bridge:
33 * `RpcErrorEnvelope` is rethrown via {@link decodeRpcThrowable} so
34 * `try/catch` sees an `Error` (or a tagged error object), and
35 * `RpcStreamEnvelope` is flattened to its `ReadableStream` body.
36 *
37 * @example Three ways to reach a value from a plain Worker — direct binding, fetch, RPC
38 * ```ts
39 * import { toRpcAsync } from "alchemy/Cloudflare/Bridge";
40 * import type Backend from "./backend.ts";
41 *
42 * interface Env {
43 * BUCKET: Bucket;
44 * BACKEND: Service<Backend>;
45 * }
46 *
47 * export default {
48 * async fetch(request: Request, env: Env): Promise<Response> {
49 * const url = new URL(request.url);
50 * const key = url.searchParams.get("key");
51 * const via = url.searchParams.get("via") ?? "binding";
52 * if (!key) return new Response("Missing 'key'", { status: 400 });
53 *
54 * // option 1 — use the async binding directly
55 * if (via === "binding") {
56 * const object = await env.BUCKET.get(key);
57 * if (!object) return new Response("Not found", { status: 404 });
58 * return new Response(object.body);
59 * }
60 *
61 * // option 2 — call the Effect worker's fetch handler
62 * if (via === "fetch") {
63 * return env.BACKEND.fetch(
64 * `https://backend/?key=${encodeURIComponent(key)}`,
65 * );
66 * }
67 *
68 * // option 3 — call an Effect worker RPC method as a Promise
69 * const backend = toRpcAsync<Backend>(env.BACKEND);
70 * const value = await backend.hello(key);
71 * if (value === null) return new Response("Not found", { status: 404 });
72 * return new Response(value);
73 * },
74 * };
75 * ```
76 */
77export const toRpcAsync = <W>(stub: any): RpcAsync<Rpc.Shape<W>> & Service =>
78 new Proxy(stub, {
79 get: (target, prop) => {
80 // `Service` methods (fetch/connect) and any non-string keys (Symbol.dispose, etc.)
81 // pass through to the underlying Cloudflare binding unchanged.
82 if (typeof prop !== "string" || prop === "fetch" || prop === "connect") {
83 const value = (target as any)[prop];
84 return typeof value === "function" ? value.bind(target) : value;
85 }
86
87 return async (...args: unknown[]) => {
88 const result = await (target as any)[prop](...args);
89 if (isRpcErrorEnvelope(result)) {
90 throw decodeRpcThrowable(result.error);
91 }
92 if (isRpcStreamEnvelope(result)) {
93 return result.body;
94 }
95 return result;
96 };
97 },
98 }) as any;
99
100/**
101 * Reconstruct a throwable from {@link encodeRpcError}'s wire form. Plain
102 * `Error` payloads are rebuilt as `Error` instances so `.message`/`.name`/
103 * `.stack` survive. Tagged errors (anything with a `_tag`) and primitives
104 * are thrown as-is so `try { ... } catch (e) { if (e._tag === ...) }` keeps
105 * working.
106 */
107const decodeRpcThrowable = (error: unknown): unknown => {
108 if (error === null || typeof error !== "object") return error;
109 const obj = error as Record<string, unknown>;
110 if ("_tag" in obj) return obj;
111 if (typeof obj.message === "string" || typeof obj.name === "string") {
112 const e = new Error(typeof obj.message === "string" ? obj.message : "");
113 if (typeof obj.name === "string") e.name = obj.name;
114 if (typeof obj.stack === "string") e.stack = obj.stack;
115 return e;
116 }
117 return error;
118};