1import * as Cause from "effect/Cause";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Option from "effect/Option";
5import * as Result from "effect/Result";
6import * as Sink from "effect/Sink";
7import * as Stream from "effect/Stream";
8import * as HttpClientRequest from "effect/http/HttpClientRequest";
9import type * as HttpClientResponse from "effect/http/HttpClientResponse";
10import { HttpServerRequest } from "effect/http/HttpServerRequest";
11import * as HttpServerResponse from "effect/http/HttpServerResponse";
12import * as Socket from "effect/socket/Socket";
13import type { HttpEffect } from "./Http.ts";
14
15export type Rpc<Shape> = {
16 "~alchemy/rpc": Shape;
17};
18
19/**
20 * Recover the user's RPC `Shape` from any of the forms a caller might pass
21 * to {@link toRpcAsync}:
22 *
23 * - the Worker class value's type, e.g. `typeof Backend`, which extends
24 * `Effect.Effect<Worker & Rpc<Shape>, …>`
25 * - the unwrapped `Worker & Rpc<Shape>` type
26 * - a bare `Shape` (when the caller types it explicitly)
27 */
28export declare namespace Rpc {
29 export type Shape<W> =
30 W extends Effect.Effect<infer R, any, any>
31 ? R extends Rpc<infer Shape>
32 ? Shape
33 : R
34 : W extends Rpc<infer Shape>
35 ? Shape
36 : W;
37}
38
39// ---------------------------------------------------------------------------
40// Wire protocol shared by every transport (native Cloudflare JSRPC and the
41// plain-`fetch` transport below). The marker tags travel as structured-clone
42// objects over JSRPC, or as NDJSON / response headers over `fetch`.
43// ---------------------------------------------------------------------------
44
45export const StreamTag = "~alchemy/rpc/stream";
46export const ErrorTag = "~alchemy/rpc/error";
47export const StreamErrorTag = "~alchemy/rpc/stream-error";
48
49export type StreamEncoding = "bytes" | "jsonl";
50
51export type RpcStreamEnvelope = {
52 _tag: typeof StreamTag;
53 encoding: StreamEncoding;
54 body: ReadableStream<Uint8Array>;
55};
56
57export type RpcErrorEnvelope = {
58 _tag: typeof ErrorTag;
59 error: unknown;
60};
61
62export type RpcStreamErrorMarker = {
63 _tag: typeof StreamErrorTag;
64 error: unknown;
65};
66
67export class RpcDecodeError extends Data.TaggedError("RpcDecodeError")<{
68 readonly cause: unknown;
69}> {
70 override get message() {
71 return this.cause instanceof Error
72 ? this.cause.message
73 : String(this.cause);
74 }
75}
76
77export class RpcCallError extends Data.TaggedError("RpcCallError")<{
78 readonly method: string;
79 readonly cause: unknown;
80}> {
81 override get message() {
82 return `RPC call to "${this.method}" failed: ${
83 this.cause instanceof Error ? this.cause.message : String(this.cause)
84 }`;
85 }
86}
87
88export class RpcRemoteStreamError extends Data.TaggedError(
89 "RpcRemoteStreamError",
90)<{
91 readonly error: unknown;
92}> {}
93
94export const isRpcStreamErrorMarker = (
95 value: unknown,
96): value is RpcStreamErrorMarker =>
97 typeof value === "object" &&
98 value !== null &&
99 "_tag" in value &&
100 value._tag === StreamErrorTag &&
101 "error" in value;
102
103export const isRpcErrorEnvelope = (value: unknown): value is RpcErrorEnvelope =>
104 typeof value === "object" &&
105 value !== null &&
106 "_tag" in value &&
107 value._tag === ErrorTag &&
108 "error" in value;
109
110export const isRpcStreamEnvelope = (
111 value: unknown,
112): value is RpcStreamEnvelope =>
113 typeof value === "object" &&
114 value !== null &&
115 "_tag" in value &&
116 value._tag === StreamTag &&
117 "encoding" in value &&
118 (value.encoding === "bytes" || value.encoding === "jsonl") &&
119 "body" in value &&
120 value.body instanceof ReadableStream;
121
122/**
123 * Normalize an error value into a plain, structured-clone-safe object.
124 * Tagged errors keep `_tag` and all own enumerable fields.
125 * Plain `Error` instances keep `name`, `message`, and `stack`.
126 */
127export const encodeRpcError = (error: unknown): unknown => {
128 if (error === null || error === undefined) return error;
129 if (typeof error !== "object") return error;
130
131 const obj = error as Record<string, unknown>;
132 if ("_tag" in obj && typeof obj._tag === "string") {
133 const out: Record<string, unknown> = {};
134 for (const key of Object.keys(obj)) {
135 out[key] = obj[key];
136 }
137 if (error instanceof Error && !("message" in out)) {
138 out.message = (error as Error).message;
139 }
140 return out;
141 }
142
143 if (error instanceof Error) {
144 return { name: error.name, message: error.message, stack: error.stack };
145 }
146
147 return error;
148};
149
150// ---------------------------------------------------------------------------
151// Stream decoding (client side): turn a wire byte-stream into a typed Effect
152// `Stream`, lifting embedded error markers into the error channel.
153// ---------------------------------------------------------------------------
154
155/**
156 * Decode a wire byte stream into the original values. `bytes` streams pass
157 * through untouched; `jsonl` streams are split per-line, JSON-parsed, and any
158 * embedded {@link RpcStreamErrorMarker} is lifted into the error channel.
159 */
160export const decodeRpcByteStream = <E>(
161 bytes: Stream.Stream<Uint8Array, E>,
162 encoding: StreamEncoding,
163): Stream.Stream<any, E | RpcDecodeError | RpcRemoteStreamError> => {
164 if (encoding === "bytes") {
165 return bytes;
166 }
167 return bytes.pipe(
168 Stream.decodeText,
169 Stream.splitLines,
170 Stream.filter((line) => line.length > 0),
171 Stream.mapEffect((line) =>
172 Effect.try({
173 try: () => JSON.parse(line),
174 catch: (cause) => new RpcDecodeError({ cause }),
175 }),
176 ),
177 Stream.flatMap((value) =>
178 isRpcStreamErrorMarker(value)
179 ? Stream.fail(new RpcRemoteStreamError({ error: value.error }))
180 : Stream.succeed(value),
181 ),
182 );
183};
184
185export const fromRpcReadableStream = (
186 body: ReadableStream<Uint8Array>,
187 encoding: StreamEncoding,
188): Stream.Stream<
189 any,
190 Socket.SocketError | RpcDecodeError | RpcRemoteStreamError
191> =>
192 decodeRpcByteStream(
193 Stream.fromReadableStream({
194 evaluate: () => body,
195 onError: (cause) =>
196 Socket.isSocketError(cause)
197 ? cause
198 : new Socket.SocketError({
199 reason: new Socket.SocketReadError({ cause }),
200 }),
201 }),
202 encoding,
203 );
204
205export const fromRpcStreamEnvelope = (
206 envelope: RpcStreamEnvelope,
207): Stream.Stream<
208 any,
209 Socket.SocketError | RpcDecodeError | RpcRemoteStreamError
210> => fromRpcReadableStream(envelope.body, envelope.encoding);
211
212export const decodeRpcValue = (value: unknown) => {
213 if (isRpcStreamEnvelope(value)) {
214 return fromRpcReadableStream(value.body, value.encoding);
215 }
216
217 if (value instanceof ReadableStream) {
218 return fromRpcReadableStream(value, "bytes");
219 }
220
221 return value;
222};
223
224/**
225 * A tagged-error class an RPC surface can declare (e.g. a Durable Object's
226 * `errors` prop) so failures crossing the RPC boundary are reconstructed as
227 * real class instances instead of the plain `{ _tag, ...fields }` objects
228 * {@link encodeRpcError} produces. `identifier` is the static tag that
229 * `Schema.TaggedError` classes carry; other classes match by class `name`.
230 */
231export interface RpcErrorClass {
232 new (props: never): unknown;
233 readonly identifier?: string;
234}
235
236/**
237 * Build a reviver from declared error classes: a failure shaped
238 * `{ _tag, ...fields }` whose tag matches a declared class is
239 * reconstructed via that class's constructor (both sides of the RPC import
240 * the same class declaration, so the schema is shared by construction).
241 * Unmatched or already-revived errors pass through untouched.
242 */
243export const makeRpcErrorReviver = (
244 errors: ReadonlyArray<RpcErrorClass> | undefined,
245): ((error: unknown) => unknown) => {
246 if (errors === undefined || errors.length === 0) {
247 return (error) => error;
248 }
249 const byTag = new Map<string, RpcErrorClass>();
250 for (const cls of errors) {
251 byTag.set(cls.identifier ?? cls.name, cls);
252 }
253 return (error) => {
254 if (typeof error !== "object" || error === null || !("_tag" in error)) {
255 return error;
256 }
257 const cls = byTag.get((error as { _tag: string })._tag);
258 if (cls === undefined || error instanceof cls) {
259 return error;
260 }
261 try {
262 return new (cls as new (props: unknown) => unknown)(error);
263 } catch {
264 // A malformed wire object must not turn into a defect here — the
265 // caller still sees the raw envelope error.
266 return error;
267 }
268 };
269};
270
271/**
272 * Decode an RPC return value, lifting error envelopes into the Effect
273 * error channel so that remote `Effect.fail(...)` values are recoverable.
274 * When `revive` is given (see {@link makeRpcErrorReviver}), envelope errors
275 * are reconstructed as their declared error classes.
276 */
277export const decodeRpcResult = (
278 value: unknown,
279 revive?: (error: unknown) => unknown,
280): Effect.Effect<unknown, unknown> => {
281 if (isRpcErrorEnvelope(value)) {
282 return Effect.fail(
283 revive === undefined ? value.error : revive(value.error),
284 );
285 }
286 return Effect.succeed(decodeRpcValue(value));
287};
288
289// ---------------------------------------------------------------------------
290// Stream encoding (server side): turn a typed Effect `Stream` into a wire
291// byte stream, embedding any failure as a trailing error marker.
292// ---------------------------------------------------------------------------
293
294const encodeStreamErrorMarker = (cause: Cause.Cause<unknown>): string => {
295 const failReason = cause.reasons.find(Cause.isFailReason);
296 const error = failReason ? encodeRpcError(failReason.error) : undefined;
297 return (
298 JSON.stringify({
299 _tag: StreamErrorTag,
300 error,
301 } satisfies RpcStreamErrorMarker) + "\n"
302 );
303};
304
305const appendStreamErrors = <R>(s: Stream.Stream<string, unknown, R>) =>
306 s.pipe(
307 Stream.catchCause((cause) =>
308 Stream.succeed(encodeStreamErrorMarker(cause)),
309 ),
310 );
311
312export const toRpcStream = (stream: Stream.Stream<any, any, any>) =>
313 Effect.scoped(
314 Effect.gen(function* () {
315 const [head, rest] = yield* Stream.peel(stream, Sink.head());
316
317 if (Option.isSome(head) && head.value instanceof Uint8Array) {
318 return {
319 _tag: StreamTag,
320 encoding: "bytes",
321 body: Stream.toReadableStream(
322 rest.pipe(Stream.prepend([head.value])),
323 ),
324 } satisfies RpcStreamEnvelope;
325 }
326
327 const body = Option.isSome(head)
328 ? rest.pipe(Stream.prepend([head.value]))
329 : rest;
330
331 return {
332 _tag: StreamTag,
333 encoding: "jsonl",
334 body: Stream.toReadableStream(
335 appendStreamErrors(
336 body.pipe(Stream.map((value) => JSON.stringify(value) + "\n")),
337 ).pipe(Stream.encodeText),
338 ),
339 } satisfies RpcStreamEnvelope;
340 }),
341 ).pipe(
342 Effect.catchCause((cause) => {
343 const failReason = cause.reasons.find(Cause.isFailReason);
344 if (failReason) {
345 return Effect.succeed({
346 _tag: StreamTag,
347 encoding: "jsonl",
348 body: Stream.toReadableStream(
349 Stream.succeed(encodeStreamErrorMarker(cause)).pipe(
350 Stream.encodeText,
351 ),
352 ),
353 } satisfies RpcStreamEnvelope);
354 }
355 return Effect.die(Cause.squash(cause));
356 }),
357 );
358
359// ---------------------------------------------------------------------------
360// `fetch`-transport stream codec. Unlike the JSRPC path above (which can peel
361// to pick a raw-`bytes` vs `jsonl` encoding because its `ReadableStream` body
362// is materialized eagerly), an HTTP response body is consumed lazily AFTER the
363// handler returns — so peeling here would tie the source to the handler scope
364// and truncate the body to its first element once that scope closes. Instead
365// this codec is uniformly NDJSON and FULLY LAZY (no peel, no scope): each line
366// is a JSON value, `Uint8Array` chunks are tagged + base64-encoded so binary
367// streams round-trip, and a failure is appended as a trailing error marker.
368// ---------------------------------------------------------------------------
369
370const BytesTag = "~alchemy/rpc/bytes";
371
372type RpcBytesChunk = { _tag: typeof BytesTag; b64: string };
373
374const isRpcBytesChunk = (value: unknown): value is RpcBytesChunk =>
375 typeof value === "object" &&
376 value !== null &&
377 "_tag" in value &&
378 value._tag === BytesTag &&
379 "b64" in value;
380
381const toBase64 = (bytes: Uint8Array): string => {
382 let binary = "";
383 for (const byte of bytes) binary += String.fromCharCode(byte);
384 return btoa(binary);
385};
386
387const fromBase64 = (b64: string): Uint8Array => {
388 const binary = atob(b64);
389 const out = new Uint8Array(binary.length);
390 for (let i = 0; i < binary.length; i++) out[i] = binary.charCodeAt(i);
391 return out;
392};
393
394/**
395 * Encode a `Stream` as a lazy NDJSON byte stream for an HTTP response body.
396 * No peeking, so nothing is held open across the handler→body-streaming
397 * boundary. `Uint8Array` elements are tagged + base64-encoded; a source
398 * failure is appended as a trailing {@link RpcStreamErrorMarker}.
399 */
400export const encodeRpcResponseStream = (
401 stream: Stream.Stream<any, any, any>,
402): Stream.Stream<Uint8Array, never, any> =>
403 appendStreamErrors(
404 stream.pipe(
405 Stream.map((value) =>
406 value instanceof Uint8Array
407 ? JSON.stringify({ _tag: BytesTag, b64: toBase64(value) }) + "\n"
408 : JSON.stringify(value) + "\n",
409 ),
410 ),
411 ).pipe(Stream.encodeText);
412
413/**
414 * Decode an NDJSON byte stream produced by {@link encodeRpcResponseStream}:
415 * tagged byte chunks become `Uint8Array`, a {@link RpcStreamErrorMarker} is
416 * lifted into the error channel, everything else is the decoded JSON value.
417 */
418export const decodeRpcResponseStream = <E>(
419 bytes: Stream.Stream<Uint8Array, E>,
420): Stream.Stream<any, E | RpcDecodeError | RpcRemoteStreamError> =>
421 bytes.pipe(
422 Stream.decodeText,
423 Stream.splitLines,
424 Stream.filter((line) => line.length > 0),
425 Stream.mapEffect((line) =>
426 Effect.try({
427 try: () => JSON.parse(line),
428 catch: (cause) => new RpcDecodeError({ cause }),
429 }),
430 ),
431 Stream.flatMap((value) =>
432 isRpcStreamErrorMarker(value)
433 ? Stream.fail(new RpcRemoteStreamError({ error: value.error }))
434 : Stream.succeed(
435 isRpcBytesChunk(value) ? fromBase64(value.b64) : value,
436 ),
437 ),
438 );
439
440// ---------------------------------------------------------------------------
441// `asEffectOrStream`: a single value usable as BOTH an `Effect` and a `Stream`.
442// ---------------------------------------------------------------------------
443
444// Effect's internal Stream brand. `Stream.isStream` recognises a value by
445// this property and reads its `channel`. An RPC stub method can't know
446// synchronously whether the remote method returns a value or a `Stream`
447// (the call is async), yet its declared type mirrors the remote `Shape`
448// verbatim — value methods are typed `Effect<A>`, streaming methods `Stream<A>`.
449// We satisfy BOTH by handing back the call `Effect` augmented with the Stream
450// brand + channel, so the single return value can be `yield*`-ed / `.pipe`d as
451// an Effect (value methods, e.g. `stub.put(k, v).pipe(Effect.orDie)`) AND piped
452// through `Stream.*` combinators (streaming methods, e.g.
453// `stub.tick(n).pipe(Stream.map(...))`).
454const StreamTypeId = "~effect/Stream";
455
456export const asEffectOrStream = (
457 call: Effect.Effect<unknown, unknown>,
458): Effect.Effect<unknown, unknown> => {
459 const streamForm = Stream.unwrap(
460 Effect.map(call, (value) =>
461 Stream.isStream(value) ? value : Stream.succeed(value),
462 ),
463 );
464 return Object.assign(call, {
465 [StreamTypeId]: (streamForm as any)[StreamTypeId],
466 channel: (streamForm as any).channel,
467 });
468};
469
470// ---------------------------------------------------------------------------
471// `fetch` transport: a transport-agnostic RPC over plain HTTP. Each method is
472// a `POST {RPC_PATH_PREFIX}{name}` whose request body is the JSON-encoded
473// argument array; the response is either a JSON value, a JSON error envelope,
474// or a streamed body (NDJSON / raw bytes) flagged by response headers.
475// ---------------------------------------------------------------------------
476
477/** Path prefix under which RPC methods are dispatched. */
478export const RPC_PATH_PREFIX = "/__rpc__/";
479/** Response header flag marking an NDJSON streamed body (vs a JSON value). */
480export const RPC_STREAM_HEADER = "x-alchemy-rpc-stream";
481
482/**
483 * Build a typed RPC stub over a plain `fetch` transport. Any property that
484 * isn't an own property of `base` is treated as a remote method: calling it
485 * `POST`s `{baseUrl}{RPC_PATH_PREFIX}{name}` with the JSON-encoded arguments
486 * and decodes the response into an {@link asEffectOrStream} value (so value
487 * methods `yield*` as `Effect`s and streaming methods pipe as `Stream`s).
488 */
489export const makeFetchRpcStub = <Shape>(options: {
490 readonly fetch: (
491 request: HttpClientRequest.HttpClientRequest,
492 ) => Effect.Effect<HttpClientResponse.HttpClientResponse, unknown>;
493 readonly baseUrl?: string;
494 /** Own properties that take precedence over remote-method dispatch. */
495 readonly base?: Record<string, unknown>;
496}): Shape => {
497 const baseUrl = options.baseUrl ?? "http://alchemy-rpc";
498 const target: Record<string, unknown> = options.base ?? {};
499
500 return new Proxy(target, {
501 get: (obj, prop) => {
502 if (prop in obj) return obj[prop as keyof typeof obj];
503 if (typeof prop !== "string") return undefined;
504 return (...args: unknown[]) =>
505 asEffectOrStream(
506 Effect.gen(function* () {
507 const request = HttpClientRequest.post(
508 `${baseUrl}${RPC_PATH_PREFIX}${encodeURIComponent(prop)}`,
509 ).pipe(
510 HttpClientRequest.bodyText(
511 JSON.stringify(args),
512 "application/json",
513 ),
514 );
515 const response = yield* options
516 .fetch(request)
517 .pipe(
518 Effect.mapError(
519 (cause) => new RpcCallError({ method: prop, cause }),
520 ),
521 );
522
523 const headers = response.headers as Record<
524 string,
525 string | undefined
526 >;
527 if (headers[RPC_STREAM_HEADER] !== undefined) {
528 return decodeRpcResponseStream(response.stream);
529 }
530
531 const value = yield* response.json.pipe(
532 Effect.mapError(
533 (cause) => new RpcCallError({ method: prop, cause }),
534 ),
535 );
536 return yield* decodeRpcResult(value);
537 }),
538 );
539 },
540 }) as Shape;
541};
542
543/**
544 * Serve the RPC methods on `shape` over the {@link RPC_PATH_PREFIX} route,
545 * delegating every other request to `fallback`. The mirror of {@link
546 * makeFetchRpcStub}: method arguments are read from the JSON request body, the
547 * method is invoked, and the result is encoded as a JSON value, a JSON error
548 * envelope, or a streamed body (flagged via {@link RPC_STREAM_HEADER}).
549 */
550export const serveRpc = <Req = never>(
551 shape: Record<string, unknown>,
552 fallback: HttpEffect<Req>,
553): HttpEffect<Req> =>
554 Effect.gen(function* () {
555 const request = yield* HttpServerRequest;
556 const prefixAt = request.url.indexOf(RPC_PATH_PREFIX);
557 if (prefixAt === -1) {
558 return yield* fallback;
559 }
560
561 let name = request.url.slice(prefixAt + RPC_PATH_PREFIX.length);
562 const queryAt = name.indexOf("?");
563 if (queryAt !== -1) name = name.slice(0, queryAt);
564 name = decodeURIComponent(name);
565
566 const method = shape[name];
567 if (typeof method !== "function") {
568 return HttpServerResponse.text(`Unknown RPC method "${name}"`, {
569 status: 404,
570 });
571 }
572
573 const text = yield* request.text;
574 // Keep arg-decode failures out of the handler's error channel — a
575 // malformed argument payload is a client error, so answer 400.
576 const argsResult = yield* Effect.result(
577 Effect.try({
578 try: () => (text.length > 0 ? (JSON.parse(text) as unknown[]) : []),
579 catch: (cause) => new RpcDecodeError({ cause }),
580 }),
581 );
582 if (Result.isFailure(argsResult)) {
583 return HttpServerResponse.text(
584 `Invalid RPC arguments for "${name}": ${argsResult.failure.message}`,
585 { status: 400 },
586 );
587 }
588 const args = argsResult.success;
589
590 const invoked = (method as (...a: unknown[]) => unknown)(...args);
591
592 // Fully-lazy NDJSON body: nothing is held open across the
593 // handler→body-streaming boundary, so the stream isn't truncated. The
594 // body's requirements are the shape method's phantom `R` (e.g.
595 // `RuntimeContext`), satisfied by the surrounding runtime; erase it for
596 // the response constructor.
597 const streamResponse = (stream: Stream.Stream<unknown, unknown, unknown>) =>
598 HttpServerResponse.stream(
599 encodeRpcResponseStream(stream) as Stream.Stream<Uint8Array, never>,
600 { headers: { [RPC_STREAM_HEADER]: "ndjson" } },
601 );
602
603 // A *genuine* `Stream` (not an Effect) is encoded directly. A nested-RPC
604 // value built by {@link asEffectOrStream} is BOTH an Effect and a Stream —
605 // it MUST be run as an effect (below), otherwise a forwarded value method
606 // (e.g. a DO method returning `container.readObject(key)`) would be
607 // mis-served as a stream envelope and decoded by the caller as a `Stream`
608 // instead of the value.
609 if (Stream.isStream(invoked) && !Effect.isEffect(invoked)) {
610 return streamResponse(invoked);
611 }
612
613 const result = yield* Effect.result(
614 invoked as Effect.Effect<unknown, unknown>,
615 );
616 if (Result.isSuccess(result)) {
617 // The resolved value may itself be a `Stream` (e.g. a forwarded nested
618 // *streaming* RPC, where the inner call resolves to a `Stream`) — encode
619 // that as a stream body too.
620 return Stream.isStream(result.success)
621 ? streamResponse(result.success)
622 : yield* HttpServerResponse.json(result.success ?? null);
623 }
624 return yield* HttpServerResponse.json({
625 _tag: ErrorTag,
626 error: encodeRpcError(result.failure),
627 } satisfies RpcErrorEnvelope);
628 });