1import * as Console from "effect/Console";
2import * as Context from "effect/Context";
3import * as Deferred from "effect/Deferred";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Logger from "effect/Logger";
7import type * as Scope from "effect/Scope";
8import * as FetchHttpClient from "effect/http/FetchHttpClient";
9import { makePlainConsoleSink } from "../Util/ConsoleSink.ts";
10import type { HttpClient } from "effect/http/HttpClient";
11import { ArtifactStore, createArtifactStore } from "../Artifacts.ts";
12import type { ProviderService } from "../Provider.ts";
13import type { ResourceLike } from "../Resource.ts";
14import {
15 platformLayer,
16 PlatformServices,
17 runMain,
18} from "../Util/PlatformServices.ts";
19import * as RpcSerialization from "./RpcSerialization.ts";
20import * as RpcServerEnvironment from "./RpcServerEnvironment.ts";
21import type { SessionEnvironment } from "./RpcServerEnvironment.ts";
22import {
23 makeServerRpcSession,
24 type ServerRpcSession,
25 type ServerWebSocketLike,
26} from "./RpcServerSession.ts";
27
28/**
29 * A service that exposes one or more resource providers over RPC.
30 * This returns `never` because it is meant to be used with `Layer.launch` (see {@link launch}).
31 */
32export class RpcServer extends Context.Service<RpcServer, never>()(
33 "alchemy/Local/RpcServer",
34) {}
35
36/**
37 * The provider shape served over RPC. The `mode`/`modes` variant machinery
38 * (lazy Layer-built Effects, see `ProviderLayer.dual`) is process-local and
39 * cannot cross the RPC boundary — the sidecar serves the concrete provider
40 * implementation, never the mode-dispatching wrapper.
41 */
42export type RpcProviderService<R extends ResourceLike> = Omit<
43 ProviderService<R>,
44 "mode" | "modes"
45>;
46
47/**
48 * The RPC API that is implemented by the server and consumed by {@link RpcProviderProxy}.
49 */
50export interface RpcProxyApi {
51 /**
52 * Retrieves a provider from the RPC server context.
53 * The consumer must unwrap the provider using {@link RpcSerialization.unwrapRpcHandlers} before using it.
54 *
55 * `group` names the provider group the type belongs to: for a server
56 * launched with a group loader (the dev sidecar, see `Local/Sidecar.ts`)
57 * it is the URL of the module whose default export is that group's
58 * provider layer, imported and built on first use per session. A server
59 * launched with a static layer ignores it.
60 */
61 readonly getProvider: <R extends ResourceLike>(
62 type: R["Type"],
63 group: string,
64 ) => Promise<RpcSerialization.RpcWrapped<RpcProviderService<R>>>;
65}
66
67/** The layer shape a served provider group must have. */
68export type ProviderLayer = Layer.Layer<
69 any,
70 any,
71 | Scope.Scope
72 | RpcServerEnvironment.RpcEnvironmentServices
73 | PlatformServices
74 | HttpClient
75 | ArtifactStore
76>;
77
78/**
79 * Resolves a provider group to its layer. Receives the `group` the client
80 * passed to {@link RpcProxyApi.getProvider}.
81 */
82export type ProviderGroupLoader = (
83 group: string,
84) => Effect.Effect<ProviderLayer, unknown>;
85
86const serverPlatformLayer = platformLayer({
87 bun: async () => {
88 const { RpcServerBun } = await import("./RpcServerBun.ts");
89 return RpcServerBun;
90 },
91 node: async () => {
92 const { RpcServerNode } = await import("./RpcServerNode.ts");
93 return RpcServerNode;
94 },
95});
96
97/**
98 * Per-session provider contexts. One sidecar process serves every stack in
99 * a run (the test harness shares a single child across all test files) and
100 * every provider group, so each group's layer is built lazily per distinct
101 * {@link SessionEnvironment} — each build gets its own MemoMap (a shared
102 * one would dedupe the whole providers layer to the first stack's build)
103 * and lives in the process's root scope. A group whose types a session
104 * never asks for is never loaded.
105 */
106export class SessionProviders extends Context.Service<
107 SessionProviders,
108 {
109 readonly get: (
110 sessionEnv: string | undefined,
111 type: string,
112 group: string,
113 ) => Promise<RpcSerialization.RpcWrapped<RpcProviderService<any>>>;
114 }
115>()("alchemy/Local/SessionProviders") {}
116
117const sessionProviders = (resolve: ProviderGroupLoader) =>
118 Layer.effect(
119 SessionProviders,
120 Effect.gen(function* () {
121 const scope = yield* Effect.scope;
122 // Capture the ambient platform context (PlatformServices, HttpClient,
123 // ArtifactStore — provided by `launch`) so deferred per-session builds
124 // can run inside a capnweb promise callback.
125 const ambient = yield* Effect.context<never>();
126 const base = yield* RpcServerEnvironment.fromProcessEnv.pipe(
127 Effect.orDie,
128 );
129 // Built contexts by session environment, then by provider group.
130 const builds = new Map<
131 string | undefined,
132 Map<string, Promise<Context.Context<any>>>
133 >();
134
135 const contextFor = (
136 sessionEnv: string | undefined,
137 group: string,
138 ): Promise<Context.Context<any>> => {
139 const session = builds.get(sessionEnv) ?? new Map();
140 builds.set(sessionEnv, session);
141 const existing = session.get(group);
142 if (existing !== undefined) {
143 return existing;
144 }
145 const resolved: SessionEnvironment | undefined =
146 sessionEnv !== undefined
147 ? RpcServerEnvironment.decodeSessionEnvironment(sessionEnv)
148 : base.stack !== undefined && base.alchemyContext !== undefined
149 ? { stack: base.stack, alchemyContext: base.alchemyContext }
150 : undefined;
151 if (resolved === undefined) {
152 return Promise.reject(
153 new Error(
154 "RPC session carried no session environment and the server was booted without a default one",
155 ),
156 );
157 }
158 const build = Effect.runPromise(
159 resolve(group).pipe(
160 Effect.flatMap((providers) =>
161 Layer.buildWithScope(
162 providers.pipe(
163 Layer.provide(
164 RpcServerEnvironment.layer({
165 profile: base.profile,
166 envFile: base.envFile,
167 ...resolved,
168 }),
169 ),
170 ),
171 scope,
172 ),
173 ),
174 Effect.provideContext(ambient as Context.Context<any>),
175 ) as Effect.Effect<Context.Context<any>>,
176 );
177 session.set(group, build);
178 // Don't poison the memo with a transient build failure — the next
179 // session for this stack retries.
180 build.catch(() => {
181 if (session.get(group) === build) {
182 session.delete(group);
183 }
184 });
185 return build;
186 };
187
188 return SessionProviders.of({
189 get: async (sessionEnv, type, group) => {
190 const context = await contextFor(sessionEnv, group);
191 const provider = context.mapUnsafe.get(type) as
192 | ProviderService<any>
193 | undefined;
194 if (!provider) {
195 throw new Error(
196 `Provider "${type}" not found in provider group ${group}`,
197 );
198 }
199 // Strip the process-local variant machinery (see
200 // RpcProviderService above) — lazy Effects don't serialize.
201 const { mode: _mode, modes: _modes, ...serializable } = provider;
202 return RpcSerialization.wrapRpcHandlers(
203 serializable as RpcProviderService<any>,
204 ["tail"],
205 );
206 },
207 });
208 }),
209 );
210
211/**
212 * Launches an RPC server that serves providers.
213 * Alchemy globals such as `AlchemyContext`, `Profile`, and `Stack` are inherited from the parent via {@link RpcServerEnvironment.fromEnv} and should not be provided manually.
214 * `PlatformServices` and `HttpClient` are also included.
215 *
216 * Pass a layer to serve a fixed set of providers, or a
217 * {@link ProviderGroupLoader} to resolve the group each client names — the
218 * dev sidecar (`Local/Sidecar.ts`) imports the group module on demand, so
219 * one process serves every provider group without loading the ones a run
220 * never touches.
221 *
222 * @example
223 * ```ts
224 * RpcServer.launch(
225 * Layer.merge(
226 * FunctionProvider,
227 * QueueProvider,
228 * ),
229 * );
230 * ```
231 *
232 * @param providers - A layer containing the providers to serve, or a loader
233 * from group to layer.
234 */
235export const launch = (providers: ProviderLayer | ProviderGroupLoader) =>
236 serverPlatformLayer.pipe(
237 Layer.provide(
238 sessionProviders(
239 Layer.isLayer(providers) ? () => Effect.succeed(providers) : providers,
240 ),
241 ),
242 Layer.provide(
243 Layer.mergeAll(
244 PlatformServices,
245 FetchHttpClient.layer,
246 Layer.sync(ArtifactStore, createArtifactStore),
247 ),
248 ),
249 // Sidecar stdio is piped, so effect's default pretty logger disables
250 // colors (it only checks `isTTY`, never FORCE_COLOR). The spawner sets
251 // FORCE_COLOR exactly when the destination terminal supports color —
252 // honor it here so sidecar log lines match the rest of the dev output.
253 Layer.provide(
254 process.env.FORCE_COLOR
255 ? Logger.layer([makePlainConsoleSink(true)])
256 : Layer.empty,
257 ),
258 Layer.launch,
259 Effect.scoped,
260 runMain,
261 );
262
263/**
264 * Constructs an `RpcServer` layer using the given server implementation.
265 * @param serve - A function that spawns a websocket server and returns its URL.
266 * @returns An `RpcServer` layer.
267 */
268export const layerServer = (
269 serve: (handlers: {
270 /**
271 * Creates a new RPC session over the given websocket. `sessionEnv` is
272 * the raw {@link SessionEnvironment} JSON from the websocket URL's
273 * `SESSION_ENV_PARAM` query parameter, when the client sent one.
274 */
275 createRpcSession: (
276 ws: ServerWebSocketLike,
277 sessionEnv?: string,
278 ) => ServerRpcSession<RpcProxyApi>;
279 /** Called when the parent connection, indicated by the `/parent` path, is established. */
280 parentConnected: () => void;
281 /** Called when the parent disconnects. The server will shut down when this is called. */
282 parentDisconnected: () => void;
283 }) => Effect.Effect<{ readonly url: string }, never, Scope.Scope>,
284) =>
285 Layer.effect(
286 RpcServer,
287 Effect.gen(function* () {
288 const providers = yield* SessionProviders;
289 const connected = yield* Deferred.make<void>();
290 const disconnected = yield* Deferred.make<void>();
291 const { url } = yield* serve({
292 createRpcSession: (ws, sessionEnv) =>
293 makeServerRpcSession<RpcProxyApi>(ws, {
294 getProvider: (<R extends ResourceLike>(
295 type: R["Type"],
296 group: string,
297 ) =>
298 providers.get(
299 sessionEnv,
300 type,
301 group,
302 )) as RpcProxyApi["getProvider"],
303 }),
304 parentConnected: () => Deferred.doneUnsafe(connected, Effect.void),
305 parentDisconnected: () =>
306 Deferred.doneUnsafe(disconnected, Effect.void),
307 });
308 yield* Console.log(`<ALCHEMY_RPC_ADDRESS>${url}</ALCHEMY_RPC_ADDRESS>`);
309 yield* Deferred.await(connected).pipe(Effect.timeout("10 seconds")); // TODO(john): should the timeout be shorter?
310 yield* Deferred.await(disconnected);
311 return yield* Effect.interrupt;
312 }),
313 );