1/**
2 * Node.js HTTP and WebSocket layers for Effect Cluster runners.
3 *
4 * The main `layer` builds a sharding layer for HTTP or WebSocket transport,
5 * choosing serialization, runner health checks, runner storage, message
6 * storage, and optional client-only mode from the supplied options.
7 * `layerHttpServer` provides the Node HTTP server used by cluster runners, and
8 * this module re-exports the Kubernetes HTTP client layer used by runner health
9 * checks.
10 *
11 * @since 4.0.0
12 */
13import * as HttpRunner from "effect/cluster/HttpRunner"
14import * as MessageStorage from "effect/cluster/MessageStorage"
15import * as RunnerHealth from "effect/cluster/RunnerHealth"
16import * as Runners from "effect/cluster/Runners"
17import * as RunnerStorage from "effect/cluster/RunnerStorage"
18import type { Sharding } from "effect/cluster/Sharding"
19import * as ShardingConfig from "effect/cluster/ShardingConfig"
20import * as SqlMessageStorage from "effect/cluster/SqlMessageStorage"
21import * as SqlRunnerStorage from "effect/cluster/SqlRunnerStorage"
22import type * as Config from "effect/Config"
23import * as Effect from "effect/Effect"
24import type * as Etag from "effect/http/Etag"
25import type { HttpPlatform } from "effect/http/HttpPlatform"
26import type { HttpServer } from "effect/http/HttpServer"
27import type { ServeError } from "effect/http/HttpServerError"
28import * as Layer from "effect/Layer"
29import * as Option from "effect/Option"
30import * as RpcSerialization from "effect/rpc/RpcSerialization"
31import type { SqlClient } from "effect/sql/SqlClient"
32import { createServer } from "node:http"
33import { layerK8sHttpClient } from "./NodeClusterSocket.ts"
34import * as NodeCrypto from "./NodeCrypto.ts"
35import * as NodeHttpClient from "./NodeHttpClient.ts"
36import * as NodeHttpServer from "./NodeHttpServer.ts"
37import type { NodeServices } from "./NodeServices.ts"
38import * as NodeSocket from "./NodeSocket.ts"
39
40export {
41 /**
42 * Provides the Kubernetes HTTP client layer used by Kubernetes runner health checks.
43 *
44 * @category re-exports
45 * @since 4.0.0
46 */
47 layerK8sHttpClient
48} from "./NodeClusterSocket.ts"
49
50/**
51 * Builds the Node cluster HTTP/WebSocket sharding layer, configuring runner
52 * transport, RPC serialization, message storage, runner health checks, and
53 * optional client-only mode.
54 *
55 * @category layers
56 * @since 4.0.0
57 */
58export const layer = <
59 const ClientOnly extends boolean = false,
60 const Storage extends "local" | "sql" | "byo" = never
61>(options: {
62 readonly transport: "http" | "websocket"
63 readonly serialization?: "binary" | "ndjson" | undefined
64 readonly serializationMaxBufferSize?: number | "unbounded" | undefined
65 readonly clientOnly?: ClientOnly | undefined
66 readonly storage?: Storage | undefined
67 readonly runnerHealth?: "ping" | "k8s" | undefined
68 readonly runnerHealthK8s?: {
69 readonly namespace?: string | undefined
70 readonly labelSelector?: string | undefined
71 } | undefined
72 readonly shardingConfig?: Partial<ShardingConfig.ShardingConfig["Service"]> | undefined
73}): ClientOnly extends true ? Layer.Layer<
74 Sharding | Runners.Runners | ("byo" extends Storage ? never : MessageStorage.MessageStorage),
75 Config.ConfigError,
76 "local" extends Storage ? never
77 : "byo" extends Storage ? (MessageStorage.MessageStorage | RunnerStorage.RunnerStorage)
78 : SqlClient
79 > :
80 Layer.Layer<
81 Sharding | Runners.Runners | ("byo" extends Storage ? never : MessageStorage.MessageStorage),
82 ServeError | Config.ConfigError,
83 "local" extends Storage ? never
84 : "byo" extends Storage ? (MessageStorage.MessageStorage | RunnerStorage.RunnerStorage)
85 : SqlClient
86 > =>
87{
88 const layer: Layer.Layer<any, any, any> = options.clientOnly
89 // client only
90 ? options.transport === "http"
91 ? Layer.provide(HttpRunner.layerHttpClientOnly, NodeHttpClient.layerUndici)
92 : Layer.provide(HttpRunner.layerWebsocketClientOnly, NodeSocket.layerWebSocketConstructor)
93 // with server
94 : options.transport === "http"
95 ? Layer.provide(HttpRunner.layerHttp, [layerHttpServer, NodeHttpClient.layerUndici])
96 : Layer.provide(HttpRunner.layerWebsocket, [layerHttpServer, NodeSocket.layerWebSocketConstructor])
97
98 const runnerHealth: Layer.Layer<any, any, any> = options?.clientOnly
99 ? Layer.empty as any
100 : options?.runnerHealth === "k8s"
101 ? RunnerHealth.layerK8s(options.runnerHealthK8s).pipe(
102 Layer.provide(layerK8sHttpClient)
103 )
104 : RunnerHealth.layerPing.pipe(
105 Layer.provide(Runners.layerRpc),
106 Layer.provide(
107 options.transport === "http"
108 ? HttpRunner.layerClientProtocolHttpDefault.pipe(Layer.provide(NodeHttpClient.layerUndici))
109 : HttpRunner.layerClientProtocolWebsocketDefault.pipe(Layer.provide(NodeSocket.layerWebSocketConstructor))
110 )
111 )
112
113 return layer.pipe(
114 Layer.provide(runnerHealth),
115 Layer.provideMerge(
116 options?.storage === "local"
117 ? MessageStorage.layerNoop
118 : options?.storage === "byo"
119 ? Layer.empty
120 : Layer.orDie(SqlMessageStorage.layer).pipe(Layer.provide(NodeCrypto.layer))
121 ),
122 Layer.provide(
123 options?.storage === "local"
124 ? RunnerStorage.layerMemory
125 : options?.storage === "byo"
126 ? Layer.empty
127 : Layer.orDie(SqlRunnerStorage.layer)
128 ),
129 Layer.provide(ShardingConfig.layerFromEnv(options?.shardingConfig)),
130 Layer.provide(
131 options?.serialization === "ndjson"
132 ? RpcSerialization.layerNdjsonWith({ maxBufferSize: options.serializationMaxBufferSize })
133 : RpcSerialization.layerSchemaBinary({ maxFrameSize: options.serializationMaxBufferSize })
134 )
135 ) as any
136}
137
138/**
139 * Provides the HTTP server and Node HTTP services used by cluster runners,
140 * listening on `ShardingConfig.runnerListenAddress` or `runnerAddress`.
141 *
142 * @category layers
143 * @since 4.0.0
144 */
145export const layerHttpServer: Layer.Layer<
146 | HttpPlatform
147 | Etag.Generator
148 | NodeServices
149 | HttpServer,
150 ServeError,
151 ShardingConfig.ShardingConfig
152> = Effect.gen(function*() {
153 const config = yield* ShardingConfig.ShardingConfig
154 const listenAddress = Option.orElse(config.runnerListenAddress, () => config.runnerAddress)
155 if (Option.isNone(listenAddress)) {
156 return yield* Effect.die("NodeClusterHttp.layerHttpServer: ShardingConfig.runnerAddress is None")
157 }
158 return NodeHttpServer.layer(createServer, listenAddress.value)
159}).pipe(Layer.unwrap)