1/**
2 * Connects cluster runner RPCs to HTTP and WebSocket transports.
3 *
4 * Runner nodes communicate through the `Runners.Rpcs` protocol. This module
5 * provides client protocol layers for dialing runner addresses over HTTP or
6 * WebSocket, HTTP effects that serve runner RPC handlers, route layers for
7 * installing runner endpoints into an `HttpRouter`, and ready-made layers for
8 * HTTP or WebSocket runner communication.
9 *
10 * @stability unstable
11 * @since 4.0.0
12 */
13import * as Effect from "../Effect.ts"
14import * as HttpClient from "../http/HttpClient.ts"
15import * as HttpClientRequest from "../http/HttpClientRequest.ts"
16import * as HttpRouter from "../http/HttpRouter.ts"
17import type * as HttpServer from "../http/HttpServer.ts"
18import type { HttpServerRequest } from "../http/HttpServerRequest.ts"
19import type { HttpServerResponse } from "../http/HttpServerResponse.ts"
20import * as Layer from "../Layer.ts"
21import * as NetAddress from "../net/NetAddress.ts"
22import * as RpcClient from "../rpc/RpcClient.ts"
23import * as RpcSerialization from "../rpc/RpcSerialization.ts"
24import * as RpcServer from "../rpc/RpcServer.ts"
25import type { Scope } from "../Scope.ts"
26import * as Socket from "../socket/Socket.ts"
27import type { MessageStorage } from "./MessageStorage.ts"
28import type { RunnerHealth } from "./RunnerHealth.ts"
29import * as Runners from "./Runners.ts"
30import { RpcClientProtocol } from "./Runners.ts"
31import * as RunnerServer from "./RunnerServer.ts"
32import type { RunnerStorage } from "./RunnerStorage.ts"
33import * as Sharding from "./Sharding.ts"
34import type * as ShardingConfig from "./ShardingConfig.ts"
35
36/**
37 * Provides a runner RPC client protocol that connects to runner addresses over
38 * HTTP.
39 *
40 * **Details**
41 *
42 * The configured path is appended to each runner address, and `https` switches
43 * the generated URL from `http` to `https`.
44 *
45 * @stability unstable
46 * @category layers
47 * @since 4.0.0
48 */
49export const layerClientProtocolHttp = (options: {
50 readonly path: string
51 readonly https?: boolean | undefined
52}): Layer.Layer<
53 RpcClientProtocol,
54 never,
55 RpcSerialization.RpcSerialization | HttpClient.HttpClient
56> =>
57 Layer.effect(RpcClientProtocol)(
58 Effect.gen(function*() {
59 const serialization = yield* RpcSerialization.RpcSerialization
60 const client = yield* HttpClient.HttpClient
61 const https = options.https ?? false
62 const path = options.path.startsWith("/") ? options.path : `/${options.path}`
63 return {
64 codecFor: serialization.codecFor,
65 make: (address) => {
66 const clientWithUrl = HttpClient.mapRequest(
67 client,
68 HttpClientRequest.prependUrl(
69 `http${https ? "s" : ""}://${NetAddress.formatUrlHostString(address.host)}:${address.port}${path}`
70 )
71 )
72 return RpcClient.makeProtocolHttp(clientWithUrl).pipe(
73 Effect.provideService(RpcSerialization.RpcSerialization, serialization)
74 )
75 }
76 }
77 })
78 )
79
80/**
81 * Default HTTP runner client protocol layer using path `/`.
82 *
83 * @stability unstable
84 * @category layers
85 * @since 4.0.0
86 */
87export const layerClientProtocolHttpDefault: Layer.Layer<
88 Runners.RpcClientProtocol,
89 never,
90 RpcSerialization.RpcSerialization | HttpClient.HttpClient
91> = layerClientProtocolHttp({ path: "/" })
92
93/**
94 * Provides a runner RPC client protocol that connects to runner addresses over
95 * WebSocket.
96 *
97 * **Details**
98 *
99 * The configured path is appended to each runner address, and `https` switches
100 * the generated URL from `ws` to `wss`.
101 *
102 * @stability unstable
103 * @category layers
104 * @since 4.0.0
105 */
106export const layerClientProtocolWebsocket = (options: {
107 readonly path: string
108 readonly https?: boolean | undefined
109}): Layer.Layer<
110 RpcClientProtocol,
111 never,
112 RpcSerialization.RpcSerialization | Socket.WebSocketConstructor
113> =>
114 Layer.effect(RpcClientProtocol)(
115 Effect.gen(function*() {
116 const serialization = yield* RpcSerialization.RpcSerialization
117 const https = options.https ?? false
118 const path = options.path.startsWith("/") ? options.path : `/${options.path}`
119 const constructor = yield* Socket.WebSocketConstructor
120 return {
121 codecFor: serialization.codecFor,
122 make: Effect.fnUntraced(function*(address) {
123 const socket = yield* Socket.makeWebSocket(
124 `ws${https ? "s" : ""}://${NetAddress.formatUrlHostString(address.host)}:${address.port}${path}`
125 ).pipe(
126 Effect.provideService(Socket.WebSocketConstructor, constructor)
127 )
128 return yield* RpcClient.makeProtocolSocket().pipe(
129 Effect.provideService(Socket.Socket, socket),
130 Effect.provideService(RpcSerialization.RpcSerialization, serialization)
131 )
132 })
133 }
134 })
135 )
136
137/**
138 * Default WebSocket runner client protocol layer using path `/`.
139 *
140 * @stability unstable
141 * @category layers
142 * @since 4.0.0
143 */
144export const layerClientProtocolWebsocketDefault: Layer.Layer<
145 Runners.RpcClientProtocol,
146 never,
147 RpcSerialization.RpcSerialization | Socket.WebSocketConstructor
148> = layerClientProtocolWebsocket({ path: "/" })
149
150/**
151 * Builds an HTTP effect that serves runner RPCs over the HTTP protocol.
152 *
153 * **Details**
154 *
155 * The returned effect is produced from `RunnerServer.layerHandlers` and the
156 * cluster runner RPC group.
157 *
158 * @stability unstable
159 * @category running
160 * @since 4.0.0
161 */
162export const toHttpEffect: Effect.Effect<
163 Effect.Effect<HttpServerResponse, never, Scope | HttpServerRequest>,
164 never,
165 Scope | RpcSerialization.RpcSerialization | Sharding.Sharding | MessageStorage
166> = Effect.gen(function*() {
167 const { httpEffect, protocol } = yield* RpcServer.makeProtocolWithHttpEffect()
168 const handlers = yield* Layer.build(RunnerServer.layerHandlers).pipe(
169 Effect.provideService(RpcServer.Protocol, protocol)
170 )
171 yield* RpcServer.make(Runners.Rpcs, {
172 spanPrefix: "RunnerServer",
173 disableTracing: true,
174 disableFatalDefects: true
175 }).pipe(
176 Effect.provideContext(handlers),
177 Effect.provideService(RpcServer.Protocol, protocol),
178 Effect.forkScoped
179 )
180 return httpEffect
181})
182
183/**
184 * Builds an HTTP effect that serves runner RPCs over WebSocket.
185 *
186 * **Details**
187 *
188 * The returned effect is produced from `RunnerServer.layerHandlers` and the
189 * cluster runner RPC group.
190 *
191 * @stability unstable
192 * @category running
193 * @since 4.0.0
194 */
195export const toHttpEffectWebsocket: Effect.Effect<
196 Effect.Effect<HttpServerResponse, never, Scope | HttpServerRequest>,
197 never,
198 Scope | RpcSerialization.RpcSerialization | Sharding.Sharding | MessageStorage
199> = Effect.gen(function*() {
200 const { httpEffect, protocol } = yield* RpcServer.makeProtocolWithHttpEffectWebsocket
201 const handlers = yield* Layer.build(RunnerServer.layerHandlers).pipe(
202 Effect.provideService(RpcServer.Protocol, protocol)
203 )
204 yield* RpcServer.make(Runners.Rpcs, {
205 spanPrefix: "RunnerServer",
206 disableTracing: true,
207 disableFatalDefects: true
208 }).pipe(
209 Effect.provideContext(handlers),
210 Effect.provideService(RpcServer.Protocol, protocol),
211 Effect.forkScoped
212 )
213 return httpEffect
214})
215
216/**
217 * Layer that provides `Sharding` and `Runners` using the configured runner RPC
218 * client protocol and storage services.
219 *
220 * @stability unstable
221 * @category layers
222 * @since 4.0.0
223 */
224export const layerClient: Layer.Layer<
225 Sharding.Sharding | Runners.Runners,
226 never,
227 ShardingConfig.ShardingConfig | Runners.RpcClientProtocol | MessageStorage | RunnerStorage | RunnerHealth
228> = Sharding.layer.pipe(
229 Layer.provideMerge(Runners.layerRpc)
230)
231
232/**
233 * Layer that adds HTTP runner routes to the provided `HttpRouter`.
234 *
235 * @stability unstable
236 * @category layers
237 * @since 4.0.0
238 */
239export const layerHttpOptions = (options: {
240 readonly path: HttpRouter.PathInput
241}): Layer.Layer<
242 Sharding.Sharding | Runners.Runners,
243 never,
244 | RunnerStorage
245 | RunnerHealth
246 | RpcSerialization.RpcSerialization
247 | MessageStorage
248 | ShardingConfig.ShardingConfig
249 | Runners.RpcClientProtocol
250 | HttpRouter.HttpRouter
251> =>
252 RunnerServer.layerWithClients.pipe(
253 Layer.provide(RpcServer.layerProtocolHttp(options))
254 )
255
256/**
257 * Layer that adds WebSocket runner routes to the provided `HttpRouter`.
258 *
259 * @stability unstable
260 * @category layers
261 * @since 4.0.0
262 */
263export const layerWebsocketOptions = (options: {
264 readonly path: HttpRouter.PathInput
265}): Layer.Layer<
266 Sharding.Sharding | Runners.Runners,
267 never,
268 | ShardingConfig.ShardingConfig
269 | Runners.RpcClientProtocol
270 | MessageStorage
271 | RunnerStorage
272 | RunnerHealth
273 | RpcSerialization.RpcSerialization
274 | HttpRouter.HttpRouter
275> =>
276 RunnerServer.layerWithClients.pipe(
277 Layer.provide(RpcServer.layerProtocolWebsocket(options))
278 )
279
280/**
281 * Layer that serves runner routes at `/` and configures HTTP runner clients.
282 *
283 * **Details**
284 *
285 * It serves runner routes at `/` and configures runner clients to communicate
286 * over HTTP.
287 *
288 * @stability unstable
289 * @category layers
290 * @since 4.0.0
291 */
292export const layerHttp: Layer.Layer<
293 Sharding.Sharding | Runners.Runners,
294 never,
295 | RpcSerialization.RpcSerialization
296 | ShardingConfig.ShardingConfig
297 | HttpClient.HttpClient
298 | HttpServer.HttpServer
299 | MessageStorage
300 | RunnerStorage
301 | RunnerHealth
302> = HttpRouter.serve(layerHttpOptions({ path: "/" })).pipe(
303 Layer.provide(layerClientProtocolHttpDefault)
304)
305
306/**
307 * Provides a client-only HTTP runner layer.
308 *
309 * **When to use**
310 *
311 * Use to provide runner clients over HTTP from a process that should not serve
312 * runner routes.
313 *
314 * **Details**
315 *
316 * It configures runner clients to communicate over HTTP without serving runner
317 * HTTP routes.
318 *
319 * @stability unstable
320 * @category layers
321 * @since 4.0.0
322 */
323export const layerHttpClientOnly: Layer.Layer<
324 Sharding.Sharding | Runners.Runners,
325 never,
326 | RpcSerialization.RpcSerialization
327 | ShardingConfig.ShardingConfig
328 | HttpClient.HttpClient
329 | MessageStorage
330 | RunnerStorage
331> = RunnerServer.layerClientOnly.pipe(
332 Layer.provide(layerClientProtocolHttpDefault)
333)
334
335/**
336 * Layer that serves runner routes at `/` and configures WebSocket runner clients.
337 *
338 * **Details**
339 *
340 * It serves runner routes at `/` and configures runner clients to communicate
341 * over WebSocket.
342 *
343 * @stability unstable
344 * @category layers
345 * @since 4.0.0
346 */
347export const layerWebsocket: Layer.Layer<
348 Sharding.Sharding | Runners.Runners,
349 never,
350 | RpcSerialization.RpcSerialization
351 | ShardingConfig.ShardingConfig
352 | Socket.WebSocketConstructor
353 | HttpServer.HttpServer
354 | MessageStorage
355 | RunnerStorage
356 | RunnerHealth
357> = HttpRouter.serve(layerWebsocketOptions({ path: "/" })).pipe(
358 Layer.provide(layerClientProtocolWebsocketDefault)
359)
360
361/**
362 * Provides a client-only WebSocket runner layer.
363 *
364 * **When to use**
365 *
366 * Use to provide runner clients over WebSocket from a process that should not
367 * serve runner routes.
368 *
369 * **Details**
370 *
371 * It configures runner clients to communicate over WebSocket without serving
372 * runner WebSocket routes.
373 *
374 * @stability unstable
375 * @category layers
376 * @since 4.0.0
377 */
378export const layerWebsocketClientOnly: Layer.Layer<
379 Sharding.Sharding | Runners.Runners,
380 never,
381 | ShardingConfig.ShardingConfig
382 | MessageStorage
383 | RunnerStorage
384 | RpcSerialization.RpcSerialization
385 | Socket.WebSocketConstructor
386> = RunnerServer.layerClientOnly.pipe(
387 Layer.provide(layerClientProtocolWebsocketDefault)
388)