1import type * as cf from "@cloudflare/workers-types";
2
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as HttpClient from "effect/http/HttpClient";
6import {
7 RpcClient,
8 RpcSerialization,
9 type Rpc,
10 type RpcGroup,
11} from "effect/rpc";
12import type * as RpcClientError from "effect/rpc/RpcClientError";
13import {
14 asEffectOrStream,
15 decodeRpcResult,
16 makeRpcErrorReviver,
17 RpcCallError,
18 type RpcErrorClass,
19} from "../../Rpc.ts";
20import { isYieldableEffect } from "../../Util/effect.ts";
21import { fromCloudflareFetcher } from "../Fetcher.ts";
22
23// The transport-agnostic RPC wire protocol (envelopes, error types, stream
24// encode/decode, `asEffectOrStream`, and the plain-`fetch` client/server) now
25// lives in `src/Rpc.ts` so non-Cloudflare runtimes (e.g. Containers) can reuse
26// it. Re-export it here so existing `Cloudflare.*` consumers keep working.
27export * from "../../Rpc.ts";
28
29/**
30 * Wrap a Cloudflare service-binding stub (or an `Effect` that resolves
31 * to one — useful when the stub depends on a service like
32 * `WorkerEnvironment` that's only available at *exec* phase) into an
33 * Effect-typed RPC client.
34 *
35 * `Service.fetch`/`Service.connect` are passed through eagerly when the
36 * stub is already resolved; everything else is treated as an RPC method
37 * whose dispatch is deferred until call time, so the user effect runs in
38 * the right runtime layer (which is what `bindWorker` actually wants —
39 * its methods are called at exec, even though it's *defined* at init).
40 */
41export const makeRpcStub = <Shape>(
42 stubSource: unknown | Effect.Effect<unknown, never, never>,
43 options?: {
44 /**
45 * Declared error classes (see {@link RpcErrorClass}): failed method
46 * results whose `_tag` matches one are reconstructed as real class
47 * instances instead of the plain objects RPC serialization produces.
48 */
49 readonly errors?: ReadonlyArray<RpcErrorClass> | undefined;
50 },
51): Shape => {
52 const isLazy = isYieldableEffect(stubSource);
53 const eagerFetcher = isLazy
54 ? undefined
55 : fromCloudflareFetcher(stubSource as cf.Fetcher);
56 const proxyTarget: object = eagerFetcher ?? {};
57 const revive = makeRpcErrorReviver(options?.errors);
58
59 return new Proxy(proxyTarget, {
60 get: (target: any, prop) => {
61 if (!isLazy && prop in target) return target[prop];
62 if (typeof prop !== "string" && typeof prop !== "symbol") {
63 return target[prop];
64 }
65 return (...args: any[]) =>
66 asEffectOrStream(
67 Effect.gen(function* () {
68 const stub = isLazy
69 ? yield* stubSource as Effect.Effect<any>
70 : stubSource;
71 return yield* Effect.tryPromise({
72 try: () => (stub as any)[prop](...args),
73 catch: (cause) =>
74 new RpcCallError({ method: String(prop), cause }),
75 }).pipe(Effect.flatMap((value) => decodeRpcResult(value, revive)));
76 }),
77 );
78 },
79 }) as Shape;
80};
81
82export const bindEffectRpc = <Rpcs extends Rpc.Any>(
83 namespace: {
84 readonly getByName: (
85 id: string,
86 options?: cf.DurableObjectNamespaceGetDurableObjectOptions,
87 ) => { readonly fetch: any };
88 },
89 group: RpcGroup.RpcGroup<Rpcs>,
90 options?: {
91 /**
92 * Override the rpc serialization layer. Defaults to NDJSON, which
93 * is required when any rpc in the group is a streaming rpc.
94 */
95 readonly serialization?: Layer.Layer<RpcSerialization.RpcSerialization>;
96 },
97): {
98 readonly getByName: (
99 id: string,
100 options?: cf.DurableObjectNamespaceGetDurableObjectOptions,
101 ) => Effect.Effect<
102 RpcClient.RpcClient<Rpcs, RpcClientError.RpcClientError>,
103 never,
104 Rpc.MiddlewareClient<Rpcs>
105 >;
106} => {
107 const serialization = options?.serialization ?? RpcSerialization.layerNdjson;
108
109 return {
110 // Wrap the cached `RpcClient` Effect in a chainable Proxy so callers
111 // can `yield* counter.getByName(id).method(args)` directly. The proxy
112 // records the `.method(args)` ops and replays them against the
113 // resolved client when the chain is yielded.
114 getByName: Effect.fn(function* (
115 id: string,
116 options?: cf.DurableObjectNamespaceGetDurableObjectOptions,
117 ) {
118 const httpClient = HttpClient.layerMergedContext(
119 Effect.sync(() => {
120 const stub = namespace.getByName(id, options);
121 return HttpClient.make((request) => stub.fetch(request));
122 }),
123 );
124 const protocol = RpcClient.layerProtocolHttp({
125 url: "http://alchemy-rpc/",
126 }).pipe(Layer.provide(serialization), Layer.provide(httpClient));
127 return yield* RpcClient.make(group).pipe(Effect.provide(protocol));
128 }) as any,
129 };
130};