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