Open History to see the last commit
1import * as NodeUtil from "node:util";
2import * as Cause from "effect/Cause";
3import * as Context from "effect/Context";
4import * as Duration from "effect/Duration";
5import * as Effect from "effect/Effect";
6import * as Exit from "effect/Exit";
7import { flow } from "effect/Function";
8import * as Redacted from "effect/Redacted";
9import * as Schema from "effect/Schema";
10import type * as Scope from "effect/Scope";
11import * as Stream from "effect/Stream";
12import * as Output from "../Output.ts";
13import { isRedactedMarker, type RedactedMarker } from "../RuntimeContext.ts";
14import { decodeDuration, DURATION_MARKER, encodeState } from "../State/StateEncoding.ts";
15import {
16 currentTraceParent,
17 type TraceParent,
18 withStreamTraceParent,
19 withTraceParent,
20} from "../Util/TraceParent.ts";
21
22type RpcEffectHandler<Args extends Array<any>, Success, Error> = (
23 ...args: Args
24) => Effect.Effect<Success, Error>;
25
26// `parent`: the caller's span, so the sidecar's work joins its trace.
27type RpcWrappedEffectHandler<Args extends Array<any>, Success, Error> = (
28 args: Args,
29 parent?: TraceParent,
30) => Promise<RpcSerializedExit<Success, Error>>;
31
32type RpcStreamHandler<Args extends Array<any>, Success, Error> = (
33 ...args: Args
34) => Stream.Stream<Success, Error>;
35
36type RpcWrappedStreamHandler<Args extends Array<any>, Success, Error> = (
37 args: Args,
38 parent?: TraceParent,
39) => RpcSerializedStream<Success, Error>;
40
41type RpcSerializedStream<Success, _Error> = ReadableStream<Success>;
42
43type RpcSerializedExit<Success, Error> =
44 | { _tag: "Success"; value: Success }
45 | { _tag: "Failure"; cause: Array<RpcSerializedCause<Error>> };
46
47type RpcSerializedCause<Error> =
48 | { _tag: "Fail"; error: Error }
49 | { _tag: "Die"; defect: unknown }
50 | { _tag: "Interrupt"; fiberId: number | undefined };
51
52export type RpcWrapped<T> =
53 T extends RpcEffectHandler<infer Args, infer Success, infer Error>
54 ? RpcWrappedEffectHandler<Args, Success, Error>
55 : T extends RpcStreamHandler<infer Args, infer Success, infer Error>
56 ? RpcWrappedStreamHandler<Args, Success, Error>
57 : T extends Record<string, any>
58 ? { [K in keyof T]: RpcWrapped<T[K]> }
59 : T;
60
61export type RpcUnwrapped<T> =
62 T extends RpcWrappedEffectHandler<infer Args, infer Success, infer Error>
63 ? RpcEffectHandler<Args, Success, Error>
64 : T extends RpcWrappedStreamHandler<infer Args, infer Success, infer Error>
65 ? RpcStreamHandler<Args, Success, Error>
66 : T extends Record<string, any>
67 ? { [K in keyof T]: RpcUnwrapped<T[K]> }
68 : T;
69
70export const wrapRpcHandlers = <T extends Record<string, any>>(
71 handlers: T,
72 streamKeys?: Array<keyof T>,
73 registerCall?: Effect.Effect<void, never, Scope.Scope>,
74): RpcWrapped<T> =>
75 Object.fromEntries(
76 Object.entries(handlers).map(([key, value]) => [
77 key,
78 typeof value === "function"
79 ? streamKeys?.includes(key)
80 ? wrapRpcStreamHandler(value, registerCall)
81 : wrapRpcEffectHandler(value, registerCall)
82 : typeof value === "object" && value !== null && !Array.isArray(value)
83 ? wrapRpcHandlers(value, undefined, registerCall)
84 : value,
85 ]),
86 ) as RpcWrapped<T>;
87
88export const unwrapRpcHandlers = <T extends Record<string, any>>(
89 handlers: RpcWrapped<T>,
90 streamKeys?: Array<keyof T>,
91): RpcUnwrapped<T> =>
92 Object.fromEntries(
93 Object.entries(handlers).map(([key, value]) => [
94 key,
95 typeof value === "function"
96 ? streamKeys?.includes(key)
97 ? unwrapRpcStreamHandler(value)
98 : unwrapRpcEffectHandler(value)
99 : typeof value === "object" && value !== null && !Array.isArray(value)
100 ? unwrapRpcHandlers(value)
101 : value,
102 ]),
103 ) as RpcUnwrapped<T>;
104
105const serializeError = (value: unknown) =>
106 Schema.encodeEffect(Schema.Defect())(value).pipe(Effect.orDie);
107
108/** Wire form of a tagged failure: its tag plus every field the error carries. */
109interface TaggedErrorMarker {
110 readonly _tag: "~alchemy/Rpc/TaggedError";
111 readonly tag: string;
112 readonly name: string | undefined;
113 readonly message: string | undefined;
114 readonly stack: string | undefined;
115 readonly fields: Record<string, unknown>;
116}
117
118const taggedErrorClasses = new Map<string, new (...args: any) => object>();
119
120/**
121 * Registers tagged error classes that a provider may fail with across the RPC
122 * sidecar boundary, so the caller receives instances of the real class (for
123 * `instanceof`) rather than plain tagged objects. Registration is by `_tag`;
124 * call it next to the class definitions so both processes register them.
125 */
126export const registerRpcErrorClasses = (
127 ...classes: ReadonlyArray<new (...args: any) => { readonly _tag: string }>
128): void => {
129 for (const cls of classes) {
130 // `Data.TaggedError` sets `_tag` per instance, so read it from a probe.
131 const probe = Reflect.construct(cls, [{}]) as { readonly _tag: string };
132 taggedErrorClasses.set(probe._tag, cls);
133 }
134};
135
136const hasStringTag = (value: unknown): value is { readonly _tag: string } =>
137 typeof value === "object" && value !== null && "_tag" in value && typeof value._tag === "string";
138
139const serializeFailure = (error: unknown): Effect.Effect<unknown> => {
140 // Keep a tagged failure's tag and fields so the caller can still
141 // `catchTag` on it (Schema.Defect alone reduces it to name + message).
142 if (!hasStringTag(error)) return serializeError(error);
143 const fields: Record<string, unknown> = {};
144 for (const [key, value] of Object.entries(error)) {
145 if (key === "_tag") continue;
146 fields[key] = serializeRpcArgs(value);
147 }
148 const asError = error instanceof Error ? error : undefined;
149 return Effect.succeed({
150 _tag: "~alchemy/Rpc/TaggedError",
151 tag: error._tag,
152 name: asError?.name,
153 message: asError?.message,
154 stack: asError?.stack,
155 fields,
156 } satisfies TaggedErrorMarker);
157};
158
159const isTaggedErrorMarker = (value: unknown): value is TaggedErrorMarker =>
160 typeof value === "object" &&
161 value !== null &&
162 "_tag" in value &&
163 value._tag === "~alchemy/Rpc/TaggedError" &&
164 "tag" in value &&
165 typeof value.tag === "string";
166
167const deserializeFailure = (error: unknown): unknown => {
168 if (!isTaggedErrorMarker(error)) return error;
169 const fields = deserializeRpcArgs(error.fields) as Record<string, unknown>;
170 const cls = taggedErrorClasses.get(error.tag);
171 const revived: Record<string, unknown> =
172 cls === undefined
173 ? error.message === undefined
174 ? {}
175 : (new Error(error.message) as unknown as Record<string, unknown>)
176 : (Object.create(cls.prototype) as Record<string, unknown>);
177 Object.assign(revived, fields);
178 revived._tag = error.tag;
179 for (const key of ["name", "message", "stack"] as const) {
180 const value = error[key];
181 if (value !== undefined && !(key in fields)) {
182 Object.defineProperty(revived, key, {
183 value,
184 writable: true,
185 configurable: true,
186 enumerable: false,
187 });
188 }
189 }
190 return revived;
191};
192
193const wrapRpcEffectHandler = <Args extends Array<any>, Success, Error>(
194 handler: RpcEffectHandler<Args, Success, Error>,
195 registerCall?: Effect.Effect<void, never, Scope.Scope>,
196): RpcWrappedEffectHandler<Args, Success, Error> =>
197 flow(
198 (args, parent) => {
199 const call = Effect.suspend(() => handler(...(deserializeRpcArgs(args) as Args)));
200 return (
201 registerCall === undefined ? call : Effect.scoped(registerCall.pipe(Effect.andThen(call)))
202 ).pipe(withTraceParent(parent));
203 },
204 Effect.exit,
205 Effect.flatMap((exit): Effect.Effect<RpcSerializedExit<Success, Error>> => {
206 if (exit._tag === "Success") {
207 // Success values need the same marker treatment as args: provider
208 // attributes can legitimately carry `Redacted` secrets (e.g. a local
209 // container's bound env), and a raw Redacted reaching capnweb dies
210 // with `TypeError: Cannot serialize value: <redacted>`.
211 return Effect.succeed({
212 _tag: "Success",
213 value: serializeRpcArgs(exit.value) as Success,
214 });
215 }
216 return Effect.forEach(
217 exit.cause.reasons,
218 (reason): Effect.Effect<RpcSerializedCause<Error>> => {
219 switch (reason._tag) {
220 case "Fail":
221 return serializeFailure(reason.error).pipe(
222 Effect.map((error) => ({ _tag: "Fail", error: error as Error })),
223 );
224 case "Die":
225 return serializeError(reason.defect).pipe(
226 Effect.map((defect) => ({ _tag: "Die", defect })),
227 );
228 case "Interrupt":
229 return Effect.succeed({ _tag: "Interrupt", fiberId: reason.fiberId });
230 }
231 },
232 ).pipe(Effect.map((cause) => ({ _tag: "Failure", cause })));
233 }),
234 Effect.runPromise,
235 );
236
237const wrapRpcStreamHandler = <Args extends Array<any>, Success, Error>(
238 handler: RpcStreamHandler<Args, Success, Error>,
239 registerCall?: Effect.Effect<void, never, Scope.Scope>,
240): RpcWrappedStreamHandler<Args, Success, Error> =>
241 flow(
242 (args, parent) => {
243 const call = Effect.sync(() => handler(...(deserializeRpcArgs(args) as Args)));
244 return Stream.unwrap(
245 registerCall === undefined ? call : registerCall.pipe(Effect.andThen(call)),
246 ).pipe(withStreamTraceParent(parent));
247 },
248 (stream) => Stream.toReadableStream(stream),
249 );
250
251const unwrapRpcEffectHandler = <Args extends Array<any>, Success, Error>(
252 handler: RpcWrappedEffectHandler<Args, Success, Error>,
253): RpcEffectHandler<Args, Success, Error> =>
254 flow(
255 (...args) => serializeRpcArgs(args) as Args,
256 (args) =>
257 Effect.flatMap(currentTraceParent, (parent) => Effect.promise(() => handler(args, parent))),
258 Effect.flatMap((exit): Exit.Exit<Success, Error> => {
259 if (exit._tag === "Success") {
260 // Mirror of the wrap side: rebuild `Redacted` wrappers from their
261 // wire markers so callers get the same shape an in-process provider
262 // returns.
263 return Exit.succeed(deserializeRpcArgs(exit.value) as Success);
264 }
265 return Exit.failCause(
266 Cause.fromReasons(
267 exit.cause.map((reason): Cause.Reason<Error> => {
268 switch (reason._tag) {
269 case "Fail":
270 return Cause.makeFailReason(deserializeFailure(reason.error) as Error);
271 case "Die":
272 return Cause.makeDieReason(reason.defect);
273 case "Interrupt":
274 return Cause.makeInterruptReason(reason.fiberId);
275 }
276 }),
277 ),
278 );
279 }),
280 );
281
282const unwrapRpcStreamHandler = <Args extends Array<any>, Success, Error>(
283 handler: RpcWrappedStreamHandler<Args, Success, Error>,
284): RpcStreamHandler<Args, Success, Error> =>
285 flow(
286 (...args) => serializeRpcArgs(args) as Args,
287 (args) =>
288 Stream.unwrap(
289 Effect.map(currentTraceParent, (parent) =>
290 Stream.fromReadableStream({
291 evaluate: () => handler(args, parent),
292 onError: (error) => error as Error,
293 }),
294 ),
295 ),
296 );
297
298const serializeRpcArgs = (value: unknown): unknown => {
299 if (Redacted.isRedacted(value)) {
300 return {
301 _tag: "Redacted",
302 value: Redacted.value(value),
303 } satisfies RedactedMarker;
304 }
305 // Output must be tested BEFORE Effect.isEffect — Output exprs are yieldable
306 // and would otherwise be misclassified as plain Effects.
307 if (Output.isOutput(value)) {
308 return {
309 _tag: "Output",
310 description: NodeUtil.inspect(value),
311 };
312 }
313 // Runtime-only effect values riding in props (e.g. a Worker's `exports`
314 // carries each DO's `constructor` Effect and captured `services` Context —
315 // passed through prop resolution by identity since #1094). They cannot
316 // cross the wire (function-valued internals kill capnweb) and the sidecar
317 // never runs them (`stripEffects` drops them from `news` provider-side), so
318 // ship a marker that deserializes back into an equivalent leaf: an Effect
319 // that dies loudly if it IS ever run, keeping `isResolved`/`stripEffects`
320 // semantics identical on both sides of the boundary.
321 if (Effect.isEffect(value)) {
322 return {
323 _tag: "~alchemy/Rpc/Effect",
324 description: NodeUtil.inspect(value),
325 } satisfies EffectMarker;
326 }
327 if (Context.isContext(value)) {
328 return {
329 _tag: "~alchemy/Rpc/Context",
330 description: NodeUtil.inspect(value),
331 } satisfies ContextMarker;
332 }
333 // Duration has `toJSON` (so the structural walk below skips it) but
334 // capnweb cannot serialize the live Effect Duration object — it dies
335 // with `TypeError: Cannot serialize value: 15000 millis`. Use the same
336 // `{ __duration__: toJSON() }` envelope as persisted state.
337 if (Duration.isDuration(value)) {
338 return encodeState(value);
339 }
340 if (typeof value === "function") {
341 return null;
342 }
343 if (Array.isArray(value)) {
344 return value.map(serializeRpcArgs);
345 }
346 if (value && typeof value === "object" && !("toJSON" in value)) {
347 return Object.fromEntries(
348 Object.entries(value).map(([key, child]) => [key, serializeRpcArgs(child)]),
349 );
350 }
351 return value;
352};
353
354interface EffectMarker {
355 readonly _tag: "~alchemy/Rpc/Effect";
356 readonly description: string;
357}
358
359interface ContextMarker {
360 readonly _tag: "~alchemy/Rpc/Context";
361 readonly description: string;
362}
363
364const isEffectMarker = (value: object): value is EffectMarker =>
365 "_tag" in value &&
366 value._tag === "~alchemy/Rpc/Effect" &&
367 "description" in value &&
368 typeof value.description === "string";
369
370const isContextMarker = (value: object): value is ContextMarker =>
371 "_tag" in value &&
372 value._tag === "~alchemy/Rpc/Context" &&
373 "description" in value &&
374 typeof value.description === "string";
375
376const isDurationEnvelope = (value: object): value is { [DURATION_MARKER]: unknown } =>
377 DURATION_MARKER in value && Object.keys(value).length === 1;
378
379const deserializeRpcArgs = (value: unknown): unknown => {
380 if (Array.isArray(value)) {
381 return value.map(deserializeRpcArgs);
382 } else if (typeof value === "object" && value !== null) {
383 // These values are serialized as `{_tag: "Redacted", value: ...}` and `{_tag: "Output", description: ...}`,
384 // so we need to detect them manually - Redacted.isRedacted and Output.isOutput do not work.
385 if (isRedactedMarker(value)) {
386 return Redacted.make(value.value);
387 } else if (
388 "_tag" in value &&
389 value._tag === "Output" &&
390 "description" in value &&
391 typeof value.description === "string"
392 ) {
393 return new Output.NamedExpr(
394 new Output.EffectExpr(Output.VoidExpr, () => Effect.never),
395 value.description,
396 );
397 } else if (isEffectMarker(value)) {
398 // Rebuild an Effect-typed leaf so `isResolved`/`stripEffects` classify
399 // it exactly like the original; dies loudly if anything ever runs it.
400 const description = value.description;
401 return Effect.suspend(() =>
402 Effect.die(
403 new Error(
404 `An Effect from the resource's props cannot cross the RPC provider boundary and was replaced by a placeholder: ${description}`,
405 ),
406 ),
407 );
408 } else if (isContextMarker(value)) {
409 return Context.empty();
410 } else if (isDurationEnvelope(value)) {
411 return decodeDuration(value[DURATION_MARKER]) ?? value;
412 }
413 return Object.fromEntries(
414 Object.entries(value).map(([key, child]) => [key, deserializeRpcArgs(child)]),
415 );
416 }
417 return value;
418};