1import * as Cause from "effect/Cause";
2import * as Context from "effect/Context";
3import * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as Exit from "effect/Exit";
6import { flow } from "effect/Function";
7import * as Redacted from "effect/Redacted";
8import * as Schema from "effect/Schema";
9import * as Stream from "effect/Stream";
10import * as NodeUtil from "node:util";
11import * as Output from "../Output.ts";
12import { isRedactedMarker, type RedactedMarker } from "../RuntimeContext.ts";
13import {
14 decodeDuration,
15 DURATION_MARKER,
16 encodeState,
17} from "../State/StateEncoding.ts";
18
19type RpcEffectHandler<Args extends Array<any>, Success, Error> = (
20 ...args: Args
21) => Effect.Effect<Success, Error>;
22
23type RpcWrappedEffectHandler<Args extends Array<any>, Success, Error> = (
24 args: Args,
25) => Promise<RpcSerializedExit<Success, Error>>;
26
27type RpcStreamHandler<Args extends Array<any>, Success, Error> = (
28 ...args: Args
29) => Stream.Stream<Success, Error>;
30
31type RpcWrappedStreamHandler<Args extends Array<any>, Success, Error> = (
32 args: Args,
33) => RpcSerializedStream<Success, Error>;
34
35type RpcSerializedStream<Success, _Error> = ReadableStream<Success>;
36
37type RpcSerializedExit<Success, Error> =
38 | { _tag: "Success"; value: Success }
39 | { _tag: "Failure"; cause: Array<RpcSerializedCause<Error>> };
40
41type RpcSerializedCause<Error> =
42 | { _tag: "Fail"; error: Error }
43 | { _tag: "Die"; defect: unknown }
44 | { _tag: "Interrupt"; fiberId: number | undefined };
45
46export type RpcWrapped<T> =
47 T extends RpcEffectHandler<infer Args, infer Success, infer Error>
48 ? RpcWrappedEffectHandler<Args, Success, Error>
49 : T extends RpcStreamHandler<infer Args, infer Success, infer Error>
50 ? RpcWrappedStreamHandler<Args, Success, Error>
51 : T extends Record<string, any>
52 ? { [K in keyof T]: RpcWrapped<T[K]> }
53 : T;
54
55export type RpcUnwrapped<T> =
56 T extends RpcWrappedEffectHandler<infer Args, infer Success, infer Error>
57 ? RpcEffectHandler<Args, Success, Error>
58 : T extends RpcWrappedStreamHandler<infer Args, infer Success, infer Error>
59 ? RpcStreamHandler<Args, Success, Error>
60 : T extends Record<string, any>
61 ? { [K in keyof T]: RpcUnwrapped<T[K]> }
62 : T;
63
64export const wrapRpcHandlers = <T extends Record<string, any>>(
65 handlers: T,
66 streamKeys?: Array<keyof T>,
67): RpcWrapped<T> => {
68 return Object.fromEntries(
69 Object.entries(handlers).map(([key, value]) => [
70 key,
71 typeof value === "function"
72 ? streamKeys?.includes(key)
73 ? wrapRpcStreamHandler(value)
74 : wrapRpcEffectHandler(value)
75 : typeof value === "object" && value !== null && !Array.isArray(value)
76 ? wrapRpcHandlers(value)
77 : value,
78 ]),
79 ) as RpcWrapped<T>;
80};
81
82export const unwrapRpcHandlers = <T extends Record<string, any>>(
83 handlers: RpcWrapped<T>,
84 streamKeys?: Array<keyof T>,
85): RpcUnwrapped<T> => {
86 return Object.fromEntries(
87 Object.entries(handlers).map(([key, value]) => [
88 key,
89 typeof value === "function"
90 ? streamKeys?.includes(key)
91 ? unwrapRpcStreamHandler(value)
92 : unwrapRpcEffectHandler(value)
93 : typeof value === "object" && value !== null && !Array.isArray(value)
94 ? unwrapRpcHandlers(value)
95 : value,
96 ]),
97 ) as RpcUnwrapped<T>;
98};
99
100const serializeError = Schema.encodeSync(Schema.Defect());
101
102const wrapRpcEffectHandler = <Args extends Array<any>, Success, Error>(
103 handler: RpcEffectHandler<Args, Success, Error>,
104): RpcWrappedEffectHandler<Args, Success, Error> =>
105 flow(
106 (args) => deserializeRpcArgs(args) as Args,
107 (args) => handler(...args),
108 Effect.exit,
109 Effect.map((exit): RpcSerializedExit<Success, Error> => {
110 if (exit._tag === "Success") {
111 // Success values need the same marker treatment as args: provider
112 // attributes can legitimately carry `Redacted` secrets (e.g. a local
113 // container's bound env), and a raw Redacted reaching capnweb dies
114 // with `TypeError: Cannot serialize value: <redacted>`.
115 return {
116 _tag: "Success",
117 value: serializeRpcArgs(exit.value) as Success,
118 };
119 }
120 return {
121 _tag: "Failure",
122 cause: exit.cause.reasons.map((reason): RpcSerializedCause<Error> => {
123 switch (reason._tag) {
124 case "Fail":
125 return {
126 _tag: "Fail",
127 error: serializeError(reason.error) as Error,
128 };
129 case "Die":
130 return { _tag: "Die", defect: serializeError(reason.defect) };
131 case "Interrupt":
132 return { _tag: "Interrupt", fiberId: reason.fiberId };
133 }
134 }),
135 };
136 }),
137 Effect.runPromise,
138 );
139
140const wrapRpcStreamHandler = <Args extends Array<any>, Success, Error>(
141 handler: RpcStreamHandler<Args, Success, Error>,
142): RpcWrappedStreamHandler<Args, Success, Error> =>
143 flow(
144 (args) => deserializeRpcArgs(args) as Args,
145 (args) => handler(...args),
146 (stream) => Stream.toReadableStream(stream),
147 );
148
149const unwrapRpcEffectHandler = <Args extends Array<any>, Success, Error>(
150 handler: RpcWrappedEffectHandler<Args, Success, Error>,
151): RpcEffectHandler<Args, Success, Error> =>
152 flow(
153 (...args) => serializeRpcArgs(args) as Args,
154 (args) => Effect.promise(() => handler(args)),
155 Effect.flatMap((exit): Exit.Exit<Success, Error> => {
156 if (exit._tag === "Success") {
157 // Mirror of the wrap side: rebuild `Redacted` wrappers from their
158 // wire markers so callers get the same shape an in-process provider
159 // returns.
160 return Exit.succeed(deserializeRpcArgs(exit.value) as Success);
161 }
162 return Exit.failCause(
163 Cause.fromReasons(
164 exit.cause.map((reason): Cause.Reason<Error> => {
165 switch (reason._tag) {
166 case "Fail":
167 return Cause.makeFailReason(reason.error);
168 case "Die":
169 return Cause.makeDieReason(reason.defect);
170 case "Interrupt":
171 return Cause.makeInterruptReason(reason.fiberId);
172 }
173 }),
174 ),
175 );
176 }),
177 );
178
179const unwrapRpcStreamHandler = <Args extends Array<any>, Success, Error>(
180 handler: RpcWrappedStreamHandler<Args, Success, Error>,
181): RpcStreamHandler<Args, Success, Error> =>
182 flow(
183 (...args) => serializeRpcArgs(args) as Args,
184 (args) => handler(args),
185 (stream) =>
186 Stream.fromReadableStream({
187 evaluate: () => stream,
188 onError: (error) => error as Error,
189 }),
190 );
191
192const serializeRpcArgs = (value: unknown): unknown => {
193 if (Redacted.isRedacted(value)) {
194 return {
195 _tag: "Redacted",
196 value: Redacted.value(value),
197 } satisfies RedactedMarker;
198 }
199 // Output must be tested BEFORE Effect.isEffect — Output exprs are yieldable
200 // and would otherwise be misclassified as plain Effects.
201 if (Output.isOutput(value)) {
202 return {
203 _tag: "Output",
204 description: NodeUtil.inspect(value),
205 };
206 }
207 // Runtime-only effect values riding in props (e.g. a Worker's `exports`
208 // carries each DO's `constructor` Effect and captured `services` Context —
209 // passed through prop resolution by identity since #1094). They cannot
210 // cross the wire (function-valued internals kill capnweb) and the sidecar
211 // never runs them (`stripEffects` drops them from `news` provider-side), so
212 // ship a marker that deserializes back into an equivalent leaf: an Effect
213 // that dies loudly if it IS ever run, keeping `isResolved`/`stripEffects`
214 // semantics identical on both sides of the boundary.
215 if (Effect.isEffect(value)) {
216 return {
217 _tag: "~alchemy/Rpc/Effect",
218 description: NodeUtil.inspect(value),
219 } satisfies EffectMarker;
220 }
221 if (Context.isContext(value)) {
222 return {
223 _tag: "~alchemy/Rpc/Context",
224 description: NodeUtil.inspect(value),
225 } satisfies ContextMarker;
226 }
227 // Duration has `toJSON` (so the structural walk below skips it) but
228 // capnweb cannot serialize the live Effect Duration object — it dies
229 // with `TypeError: Cannot serialize value: 15000 millis`. Use the same
230 // `{ __duration__: toJSON() }` envelope as persisted state.
231 if (Duration.isDuration(value)) {
232 return encodeState(value);
233 }
234 if (typeof value === "function") {
235 return null;
236 }
237 if (Array.isArray(value)) {
238 return value.map(serializeRpcArgs);
239 }
240 if (value && typeof value === "object" && !("toJSON" in value)) {
241 return Object.fromEntries(
242 Object.entries(value).map(([key, child]) => [
243 key,
244 serializeRpcArgs(child),
245 ]),
246 );
247 }
248 return value;
249};
250
251interface EffectMarker {
252 readonly _tag: "~alchemy/Rpc/Effect";
253 readonly description: string;
254}
255
256interface ContextMarker {
257 readonly _tag: "~alchemy/Rpc/Context";
258 readonly description: string;
259}
260
261const isEffectMarker = (value: object): value is EffectMarker =>
262 "_tag" in value &&
263 value._tag === "~alchemy/Rpc/Effect" &&
264 "description" in value &&
265 typeof value.description === "string";
266
267const isContextMarker = (value: object): value is ContextMarker =>
268 "_tag" in value &&
269 value._tag === "~alchemy/Rpc/Context" &&
270 "description" in value &&
271 typeof value.description === "string";
272
273const isDurationEnvelope = (
274 value: object,
275): value is { [DURATION_MARKER]: unknown } =>
276 DURATION_MARKER in value && Object.keys(value).length === 1;
277
278const deserializeRpcArgs = (value: unknown): unknown => {
279 if (Array.isArray(value)) {
280 return value.map(deserializeRpcArgs);
281 } else if (typeof value === "object" && value !== null) {
282 // These values are serialized as `{_tag: "Redacted", value: ...}` and `{_tag: "Output", description: ...}`,
283 // so we need to detect them manually - Redacted.isRedacted and Output.isOutput do not work.
284 if (isRedactedMarker(value)) {
285 return Redacted.make(value.value);
286 } else if (
287 "_tag" in value &&
288 value._tag === "Output" &&
289 "description" in value &&
290 typeof value.description === "string"
291 ) {
292 return new Output.NamedExpr(
293 new Output.EffectExpr(Output.VoidExpr, () => Effect.never),
294 value.description,
295 );
296 } else if (isEffectMarker(value)) {
297 // Rebuild an Effect-typed leaf so `isResolved`/`stripEffects` classify
298 // it exactly like the original; dies loudly if anything ever runs it.
299 const description = value.description;
300 return Effect.suspend(() =>
301 Effect.die(
302 new Error(
303 `An Effect from the resource's props cannot cross the RPC provider boundary and was replaced by a placeholder: ${description}`,
304 ),
305 ),
306 );
307 } else if (isContextMarker(value)) {
308 return Context.empty();
309 } else if (isDurationEnvelope(value)) {
310 return decodeDuration(value[DURATION_MARKER]) ?? value;
311 }
312 return Object.fromEntries(
313 Object.entries(value).map(([key, child]) => [
314 key,
315 deserializeRpcArgs(child),
316 ]),
317 );
318 }
319 return value;
320};