1import * as Cloudflare from "@/Cloudflare";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as Stream from "effect/Stream";
5import * as RpcSerialization from "effect/rpc/RpcSerialization";
6import * as RpcServer from "effect/rpc/RpcServer";
7import { DoRpcs } from "./group.ts";
8
9/**
10 * Durable Object backing the `*DO` RPCs on the Worker. It exposes the
11 * `InnerRpcs` group via `RpcServer.toHttpEffect`. The Worker constructs
12 * an `RpcClient` whose transport calls into this DO's `fetch`, so the
13 * Worker's `*DO` handlers transparently delegate here.
14 */
15export default class RpcHttpTestObject extends Cloudflare.DurableObject<RpcHttpTestObject>()(
16 "RpcHttpTestObject",
17 Effect.succeed(
18 Effect.gen(function* () {
19 let counter = 0;
20
21 const handlersLayer = DoRpcs.toLayer({
22 PingDO: ({ message }) => {
23 console.log("Ping");
24 return Effect.sync(() => {
25 console.log("Ping inside");
26 return {
27 echo: message,
28 n: ++counter,
29 };
30 });
31 },
32 CountDO: ({ upto }) =>
33 Stream.fromReadableStream<number, never>({
34 evaluate: () => {
35 let next = 1;
36 return new ReadableStream<number>({
37 pull(controller) {
38 if (next > upto) {
39 controller.close();
40 return;
41 }
42 controller.enqueue(next++);
43 },
44 });
45 },
46 onError: (cause) => cause as never,
47 }),
48 EchoDO: ({ messages }) =>
49 Stream.fromIterable(
50 messages.map((message, index) => ({ index, message })),
51 ),
52 });
53
54 return {
55 fetch: RpcServer.toHttpEffect(DoRpcs).pipe(
56 Effect.provide(
57 Layer.mergeAll(handlersLayer, RpcSerialization.layerNdjson),
58 ),
59 ),
60 };
61 }),
62 ),
63) {}