1import { newWebSocketRpcSession } from "capnweb";
2import * as Cache from "effect/Cache";
3import * as Config from "effect/Config";
4import * as Context from "effect/Context";
5import * as Effect from "effect/Effect";
6import * as Layer from "effect/Layer";
7import * as HttpBody from "effect/http/HttpBody";
8import * as HttpClient from "effect/http/HttpClient";
9import { AlchemyContext } from "../AlchemyContext.ts";
10import type { ProviderService } from "../Provider.ts";
11import type { ResourceLike } from "../Resource.ts";
12import { Stack } from "../Stack.ts";
13import { unwrapRpcHandlers } from "./RpcSerialization.ts";
14import type { RpcProxyApi } from "./RpcServer.ts";
15import {
16 encodeSessionEnvironment,
17 SESSION_ENV_PARAM,
18} from "./RpcServerEnvironment.ts";
19import type { RpcSpawnPayload } from "./RpcSpawner.ts";
20
21export class RpcProviderProxy extends Context.Service<
22 RpcProviderProxy,
23 {
24 /**
25 * The provider for `providerName`, served by the dev sidecar. `providersUrl`
26 * is the URL of the module whose default export is the provider group's
27 * layer (see `Local/Sidecar.ts`); the sidecar imports it on first use.
28 */
29 readonly get: <R extends ResourceLike>(
30 providersUrl: string,
31 providerName: R["Type"],
32 ) => Effect.Effect<ProviderService<R>, never, AlchemyContext | Stack>;
33 }
34>()("alchemy/Local/RpcProviderProxy") {}
35
36export const SPAWNER_URL_ENV_KEY = "ALCHEMY_RPC_SPAWNER_URL" as const;
37
38/**
39 * The one sidecar entry every RPC-backed provider is served from.
40 * Resolve through package exports. The active export conditions select `src/`
41 * under Bun or the dev loader and `lib/` in a published Node install.
42 */
43export const SIDECAR_ENTRY_URL = import.meta.resolve("alchemy/Local/Sidecar");
44
45const make = Effect.fn(function* (spawnerUrl: string) {
46 const client = yield* HttpClient.HttpClient;
47
48 const getSession = Effect.fn(
49 function* (sessionEnv: string) {
50 const payload: RpcSpawnPayload = { serverEntryUrl: SIDECAR_ENTRY_URL };
51 const response = yield* client.post(spawnerUrl, {
52 body: yield* HttpBody.json(payload),
53 });
54 // The spawner returns the one shared sidecar; the stack-specific
55 // environment rides the session websocket so the child can build (and
56 // memoize) a provider context per stack and provider group.
57 const body = yield* response.text;
58 if (response.status !== 200) {
59 return yield* Effect.fail(
60 new Error(
61 `RPC spawner POST ${spawnerUrl} returned ${response.status}: ${body.slice(0, 300)}`,
62 ),
63 );
64 }
65 let websocketUrl: URL;
66 try {
67 websocketUrl = new URL(body);
68 } catch {
69 return yield* Effect.fail(
70 new Error(
71 `RPC spawner POST ${spawnerUrl} did not return a websocket URL (got ${JSON.stringify(body.slice(0, 200))})`,
72 ),
73 );
74 }
75 if (websocketUrl.protocol !== "ws:" && websocketUrl.protocol !== "wss:") {
76 return yield* Effect.fail(
77 new Error(
78 `RPC spawner POST ${spawnerUrl} returned a non-websocket URL: ${websocketUrl.toString()}`,
79 ),
80 );
81 }
82 websocketUrl.searchParams.set(SESSION_ENV_PARAM, sessionEnv);
83 return newWebSocketRpcSession<RpcProxyApi>(websocketUrl.toString());
84 },
85 (effect) =>
86 Effect.catch(effect, (error) =>
87 Effect.die(
88 new Error(
89 "Failed to create a provider RPC session with the sidecar",
90 {
91 cause: error,
92 },
93 ),
94 ),
95 ),
96 );
97
98 // A websocket that drops (sidecar crash/restart, abnormal 1006 close)
99 // permanently breaks the capnweb session, and a cached broken session would
100 // poison every subsequent call — including test-runner retries.
101 // `onRpcBroken` fires on disconnect and evicts the entry, so the next `get`
102 // re-registers with the spawner (which respawns the sidecar child if it
103 // died). Assigned after the cache exists; the callback only fires on live
104 // sessions, which the cache must already contain.
105 let evictBrokenSession: (key: string) => void = () => {};
106 // One session per stack environment, shared by every provider group.
107 const cache = yield* Cache.make({
108 lookup: (sessionEnv: string) =>
109 getSession(sessionEnv).pipe(
110 Effect.tap((session) =>
111 Effect.sync(() =>
112 session.onRpcBroken(() => evictBrokenSession(sessionEnv)),
113 ),
114 ),
115 ),
116 capacity: Infinity,
117 });
118 evictBrokenSession = (key) => Effect.runFork(Cache.invalidate(cache, key));
119
120 return RpcProviderProxy.of({
121 get: Effect.fn(function* (providersUrl, providerName) {
122 const alchemyContext = yield* AlchemyContext;
123 const stack = yield* Stack;
124 const key = encodeSessionEnvironment({
125 alchemyContext,
126 stack: { name: stack.name, stage: stack.stage },
127 });
128 const fetchProvider = Effect.gen(function* () {
129 const session = yield* Cache.get(cache, key);
130 return yield* Effect.tryPromise(
131 () =>
132 session.getProvider(providerName, providersUrl) as ReturnType<
133 RpcProxyApi["getProvider"]
134 >,
135 );
136 });
137 // One in-place reconnect: if the session broke mid-call (the broken
138 // callback may not have evicted it yet), drop it and re-register once
139 // before giving up.
140 const provider = yield* fetchProvider.pipe(
141 Effect.catch(() =>
142 Cache.invalidate(cache, key).pipe(Effect.andThen(fetchProvider)),
143 ),
144 Effect.orDie,
145 );
146 // The served shape omits the process-local `mode`/`modes` variant
147 // machinery (see RpcProviderService); the unwrapped stub is a plain
148 // (mode-agnostic) ProviderService.
149 return unwrapRpcHandlers(provider, ["tail"]) as ProviderService<any>;
150 }),
151 });
152});
153
154export const layer = (url: string) => Layer.effect(RpcProviderProxy, make(url));
155
156export const fromEnv = () =>
157 Layer.effect(
158 RpcProviderProxy,
159 Config.String(SPAWNER_URL_ENV_KEY).pipe(Effect.flatMap(make)),
160 );