1import { safeHttpEffect } from "@/Http";
2import * as Rpc from "@/Rpc";
3import { describe, expect, it } from "alchemy-test";
4import * as Data from "effect/Data";
5import * as Effect from "effect/Effect";
6import * as Stream from "effect/Stream";
7import * as FetchHttpClient from "effect/http/FetchHttpClient";
8import * as HttpClient from "effect/http/HttpClient";
9import * as HttpEffect from "effect/http/HttpEffect";
10import * as HttpServerResponse from "effect/http/HttpServerResponse";
11
12// ---------------------------------------------------------------------------
13// In-memory loopback: the server `serveRpc` handler is turned into a Web
14// `Request` handler and injected as the `fetch` implementation behind the
15// real `FetchHttpClient`. This exercises the full wire protocol (headers,
16// JSON bodies, streamed NDJSON / raw bytes) with zero sockets and zero
17// container — just the protocol round-tripping in process.
18// ---------------------------------------------------------------------------
19
20class BoomError extends Data.TaggedError("BoomError")<{
21 readonly code: number;
22}> {}
23
24// The fallback handler stands in for the user's non-RPC routes.
25const fallback = Effect.succeed(
26 HttpServerResponse.text("fallback", { status: 200 }),
27);
28
29const withRpc = <A, E>(
30 shape: Record<string, unknown>,
31 use: (stub: any) => Effect.Effect<A, E, any>,
32): Effect.Effect<A, E> => {
33 const webHandler = HttpEffect.toWebHandler(
34 safeHttpEffect(Rpc.serveRpc(shape, fallback)),
35 );
36 const fetchImpl = ((url: any, init?: any) =>
37 webHandler(new Request(url, init))) as typeof globalThis.fetch;
38
39 return Effect.gen(function* () {
40 const client = yield* HttpClient.HttpClient;
41 const stub = Rpc.makeFetchRpcStub<any>({
42 fetch: (request) => client.execute(request),
43 baseUrl: "http://rpc",
44 });
45 return yield* use(stub);
46 }).pipe(
47 Effect.provide(FetchHttpClient.layer),
48 Effect.provideService(FetchHttpClient.Fetch, fetchImpl),
49 ) as Effect.Effect<A, E>;
50};
51
52const shape = {
53 ping: () => Effect.succeed("pong"),
54 echo: (a: string, b: number) => Effect.succeed({ a, b }),
55 maybe: (key: string) => Effect.succeed(key === "present" ? "value" : null),
56 identity: (value: unknown) => Effect.succeed(value),
57 boomTagged: () => Effect.fail(new BoomError({ code: 42 })),
58 boomPlain: () => Effect.fail(new Error("plain boom")),
59 countJsonl: (n: number) =>
60 Stream.range(1, n).pipe(Stream.map((i) => ({ i }))),
61 emptyStream: () => Stream.empty,
62 // An INFINITE source: only a truly streaming (lazily-pulled) transport can
63 // return a finite prefix of this. A buffer-then-stringify implementation
64 // would try to realize the whole stream and never produce a result.
65 infinite: () => Stream.iterate(0, (n) => n + 1),
66 bytes: () =>
67 Stream.fromIterable([new Uint8Array([1, 2, 3]), new Uint8Array([4, 5])]),
68 failingStream: () =>
69 Stream.range(1, 2).pipe(
70 Stream.map((i) => ({ i })),
71 Stream.concat(Stream.fail(new BoomError({ code: 7 }))),
72 ),
73};
74
75describe("Rpc fetch protocol", { tags: ["unit", "local"] }, () => {
76 describe("value methods (Effect)", () => {
77 it.effect("returns a primitive value", () =>
78 withRpc(shape, (stub) =>
79 Effect.gen(function* () {
80 expect(yield* stub.ping()).toBe("pong");
81 }),
82 ),
83 );
84
85 it.effect("round-trips arguments and an object result", () =>
86 withRpc(shape, (stub) =>
87 Effect.gen(function* () {
88 expect(yield* stub.echo("hi", 7)).toEqual({ a: "hi", b: 7 });
89 }),
90 ),
91 );
92
93 it.effect("supports a null result", () =>
94 withRpc(shape, (stub) =>
95 Effect.gen(function* () {
96 expect(yield* stub.maybe("absent")).toBe(null);
97 expect(yield* stub.maybe("present")).toBe("value");
98 }),
99 ),
100 );
101
102 it.effect("round-trips nested/array/null argument shapes", () =>
103 withRpc(shape, (stub) =>
104 Effect.gen(function* () {
105 const value = { a: [1, 2, { b: "x" }], c: null, d: true };
106 expect(yield* stub.identity(value)).toEqual(value);
107 }),
108 ),
109 );
110
111 it.effect("is usable as an Effect via .pipe (orDie)", () =>
112 withRpc(shape, (stub) =>
113 Effect.gen(function* () {
114 expect(yield* stub.ping().pipe(Effect.orDie)).toBe("pong");
115 }),
116 ),
117 );
118 });
119
120 describe("error channel", () => {
121 it.effect("lifts a remote tagged failure into the error channel", () =>
122 withRpc(shape, (stub) =>
123 Effect.gen(function* () {
124 const error = (yield* stub.boomTagged().pipe(Effect.flip)) as {
125 _tag: string;
126 code: number;
127 };
128 expect(error._tag).toBe("BoomError");
129 expect(error.code).toBe(42);
130 }),
131 ),
132 );
133
134 it.effect("lifts a remote plain Error into the error channel", () =>
135 withRpc(shape, (stub) =>
136 Effect.gen(function* () {
137 const error = (yield* stub.boomPlain().pipe(Effect.flip)) as {
138 name: string;
139 message: string;
140 };
141 expect(error.message).toBe("plain boom");
142 }),
143 ),
144 );
145
146 it.effect("fails an unknown method (404 → RpcCallError)", () =>
147 withRpc(shape, (stub) =>
148 Effect.gen(function* () {
149 const error = (yield* stub.doesNotExist().pipe(Effect.flip)) as {
150 _tag: string;
151 };
152 expect(error._tag).toBe("RpcCallError");
153 }),
154 ),
155 );
156 });
157
158 describe("streaming methods (Stream)", () => {
159 const collect = <T>(streamLike: any) =>
160 Effect.gen(function* () {
161 const out: T[] = [];
162 yield* Stream.runForEach(streamLike, (v: T) =>
163 Effect.sync(() => {
164 out.push(v);
165 }),
166 );
167 return out;
168 });
169
170 it.effect("round-trips an NDJSON object stream", () =>
171 withRpc(shape, (stub) =>
172 Effect.gen(function* () {
173 const items = yield* collect(stub.countJsonl(3));
174 expect(items).toEqual([{ i: 1 }, { i: 2 }, { i: 3 }]);
175 }),
176 ),
177 );
178
179 it.effect("handles an empty stream", () =>
180 withRpc(shape, (stub) =>
181 Effect.gen(function* () {
182 expect(yield* collect(stub.emptyStream())).toEqual([]);
183 }),
184 ),
185 );
186
187 it.effect("round-trips a raw byte stream (content preserved)", () =>
188 withRpc(shape, (stub) =>
189 Effect.gen(function* () {
190 const chunks = yield* collect<Uint8Array>(stub.bytes());
191 const flat = chunks.flatMap((c) => Array.from(c));
192 expect(flat).toEqual([1, 2, 3, 4, 5]);
193 }),
194 ),
195 );
196
197 it.effect("consumes an infinite stream incrementally (not buffered)", () =>
198 withRpc(shape, (stub) =>
199 Effect.gen(function* () {
200 // If the server buffered + stringified the whole stream before
201 // responding, an infinite source would never produce a result and
202 // this would hit the timeout below. Taking a finite prefix only
203 // succeeds because the body is pulled lazily, element by element.
204 const out = yield* collect<number>(
205 stub.infinite().pipe(Stream.take(5)),
206 );
207 expect(out).toEqual([0, 1, 2, 3, 4]);
208 }),
209 ).pipe(Effect.timeout("15 seconds")),
210 );
211
212 it.effect(
213 "server stops producing once the client stops consuming (backpressure)",
214 () =>
215 Effect.gen(function* () {
216 // The shape runs in-process, so this closure-captured array sees
217 // exactly how far the server stream was pulled.
218 const produced: number[] = [];
219 const localShape = {
220 counter: () =>
221 Stream.iterate(0, (n) => n + 1).pipe(
222 Stream.tap((n) => Effect.sync(() => produced.push(n))),
223 ),
224 };
225
226 const out = yield* withRpc(localShape, (stub) =>
227 collect<number>(stub.counter().pipe(Stream.take(3))),
228 ).pipe(Effect.timeout("15 seconds"));
229
230 expect(out).toEqual([0, 1, 2]);
231 // The client only consumed 3 elements. A streaming transport pulls
232 // lazily under backpressure, so the server produces the consumed
233 // prefix plus at most a bounded buffer — never the infinite tail a
234 // buffer-then-stringify implementation would.
235 expect(produced.length).toBeGreaterThanOrEqual(3);
236 expect(produced.length).toBeLessThan(10_000);
237 }),
238 );
239
240 it.effect("lifts a mid-stream failure into RpcRemoteStreamError", () =>
241 withRpc(shape, (stub) =>
242 Effect.gen(function* () {
243 const result = yield* stub
244 .failingStream()
245 .pipe(Stream.runCollect, Effect.flip);
246 expect((result as { _tag: string })._tag).toBe(
247 "RpcRemoteStreamError",
248 );
249 const error = (result as { error: { _tag: string; code: number } })
250 .error;
251 expect(error._tag).toBe("BoomError");
252 expect(error.code).toBe(7);
253 }),
254 ),
255 );
256
257 it.effect("delivers values before a mid-stream failure", () =>
258 withRpc(shape, (stub) =>
259 Effect.gen(function* () {
260 const seen: Array<{ i: number }> = [];
261 yield* stub.failingStream().pipe(
262 Stream.runForEach((v: { i: number }) =>
263 Effect.sync(() => {
264 seen.push(v);
265 }),
266 ),
267 Effect.flip,
268 );
269 expect(seen).toEqual([{ i: 1 }, { i: 2 }]);
270 }),
271 ),
272 );
273 });
274
275 describe("fallback routing", () => {
276 it.effect("non-RPC requests fall through to the fallback handler", () =>
277 Effect.gen(function* () {
278 const webHandler = HttpEffect.toWebHandler(
279 safeHttpEffect(Rpc.serveRpc(shape, fallback)),
280 );
281 const res = yield* Effect.promise(() =>
282 webHandler(new Request("http://rpc/not-rpc")),
283 );
284 expect(res.status).toBe(200);
285 expect(yield* Effect.promise(() => res.text())).toBe("fallback");
286 }),
287 );
288 });
289
290 // -------------------------------------------------------------------------
291 // Nested RPC: a shape method that *forwards* another RPC call returns an
292 // `asEffectOrStream` value — an `Effect` that is ALSO branded as a `Stream`
293 // so a single return is usable either way. This is exactly the shape of a
294 // Worker→DO→Container chain (e.g. `readObjectRpc: (k) => container.readObject(k)`).
295 // The dispatcher MUST run it as an effect rather than mis-detecting the
296 // brand and serving it as a stream envelope (which decodes to `{channel:{}}`
297 // on the caller). See `serveRpc` and the Worker/DO bridges.
298 // -------------------------------------------------------------------------
299 describe("nested RPC (asEffectOrStream forwarding)", () => {
300 const nested = {
301 // value method forwarding a nested call that resolves to a value
302 forwardValue: (k: string) =>
303 Rpc.asEffectOrStream(Effect.succeed(`deep:${k}`)),
304 // value method forwarding a nested call that resolves to a Stream
305 forwardStream: (n: number) =>
306 Rpc.asEffectOrStream(
307 Effect.succeed(Stream.range(1, n).pipe(Stream.map((i) => ({ i })))),
308 ),
309 };
310
311 it.effect("a forwarded value round-trips as a value (not a Stream)", () =>
312 withRpc(nested, (stub) =>
313 Effect.gen(function* () {
314 const value = yield* stub.forwardValue("x");
315 expect(value).toBe("deep:x");
316 }),
317 ),
318 );
319
320 it.effect("a forwarded Stream round-trips as a Stream", () =>
321 withRpc(nested, (stub) =>
322 Effect.gen(function* () {
323 const out: Array<{ i: number }> = [];
324 yield* Stream.runForEach(stub.forwardStream(3), (v: { i: number }) =>
325 Effect.sync(() => {
326 out.push(v);
327 }),
328 );
329 expect(out).toEqual([{ i: 1 }, { i: 2 }, { i: 3 }]);
330 }),
331 ),
332 );
333 });
334
335 // -------------------------------------------------------------------------
336 // Benchmark: the loopback exercises the FULL wire protocol (serialize args →
337 // serveRpc → JSON response → decode) with zero sockets and zero container,
338 // so it isolates the RPC layer's per-call overhead from network latency.
339 // This is the quantitative backstop for "RPC is not the bottleneck": a
340 // catastrophic regression (accidental buffering, O(n)-per-call work, a real
341 // socket sneaking in) blows the per-call ceiling, while normal network jitter
342 // (the 400-700ms seen against a live container's R2 read) lives entirely
343 // outside this measurement.
344 // -------------------------------------------------------------------------
345 describe("performance", () => {
346 const N = 2_000;
347 // In-process protocol overhead is sub-millisecond; a generous ceiling so
348 // the assertion only trips on a real regression, never on a slow CI box.
349 const MAX_MS_PER_CALL = 5;
350
351 const benchmark = (
352 label: string,
353 stub: any,
354 call: (stub: any) => Effect.Effect<unknown, unknown, any>,
355 expected: unknown,
356 ) =>
357 Effect.gen(function* () {
358 // Warm the JIT + connection-less path, and assert correctness once so
359 // the timing loop below isn't measuring a broken (e.g. erroring) path.
360 expect(yield* call(stub)).toEqual(expected);
361 const start = yield* Effect.sync(() => performance.now());
362 for (let i = 0; i < N; i++) {
363 yield* call(stub);
364 }
365 const elapsed = yield* Effect.sync(() => performance.now() - start);
366 const perCall = elapsed / N;
367 yield* Effect.logInfo(
368 `RPC ${label}: ${perCall.toFixed(4)} ms/call ` +
369 `(${N} calls in ${elapsed.toFixed(0)} ms)`,
370 );
371 expect(perCall).toBeLessThan(MAX_MS_PER_CALL);
372 });
373
374 it.effect(
375 "no-arg value-method RPC overhead stays sub-millisecond",
376 () =>
377 withRpc(shape, (stub) =>
378 benchmark("ping", stub, (s) => s.ping(), "pong"),
379 ),
380 { timeout: 60_000 },
381 );
382
383 it.effect(
384 "arg + object-result RPC overhead stays sub-millisecond",
385 () =>
386 withRpc(shape, (stub) =>
387 benchmark("echo", stub, (s) => s.echo("hi", 7), { a: "hi", b: 7 }),
388 ),
389 { timeout: 60_000 },
390 );
391 });
392});