1/**
2 * Bun HTTP and WebSocket layers for Effect Cluster runners.
3 *
4 * `layerHttpServer` provides the Bun HTTP server used by cluster runners. The
5 * main `layer` builds a sharding layer for HTTP or WebSocket transport,
6 * choosing serialization, runner health checks, runner storage, message
7 * storage, and optional client-only mode from the supplied options.
8 *
9 * @since 4.0.0
10 */
11import * as HttpRunner from "effect/cluster/HttpRunner"
12import * as MessageStorage from "effect/cluster/MessageStorage"
13import * as RunnerHealth from "effect/cluster/RunnerHealth"
14import * as Runners from "effect/cluster/Runners"
15import * as RunnerStorage from "effect/cluster/RunnerStorage"
16import type { Sharding } from "effect/cluster/Sharding"
17import * as ShardingConfig from "effect/cluster/ShardingConfig"
18import * as SqlMessageStorage from "effect/cluster/SqlMessageStorage"
19import * as SqlRunnerStorage from "effect/cluster/SqlRunnerStorage"
20import type * as Config from "effect/Config"
21import * as Effect from "effect/Effect"
22import type * as Etag from "effect/http/Etag"
23import * as FetchHttpClient from "effect/http/FetchHttpClient"
24import type { HttpPlatform } from "effect/http/HttpPlatform"
25import type { HttpServer } from "effect/http/HttpServer"
26import type { ServeError } from "effect/http/HttpServerError"
27import * as Layer from "effect/Layer"
28import * as Option from "effect/Option"
29import * as RpcSerialization from "effect/rpc/RpcSerialization"
30import type { SqlClient } from "effect/sql/SqlClient"
31import { layerK8sHttpClient } from "./BunClusterSocket.ts"
32import * as BunCrypto from "./BunCrypto.ts"
33import * as BunFileSystem from "./BunFileSystem.ts"
34import * as BunHttpServer from "./BunHttpServer.ts"
35import type { BunServices } from "./BunServices.ts"
36import * as BunSocket from "./BunSocket.ts"
37
38export {
39 /**
40 * Layer that provides a Kubernetes HTTP client for runner health checks.
41 *
42 * @category re-exports
43 * @since 4.0.0
44 */
45 layerK8sHttpClient
46}
47
48/**
49 * Layer that provides a Bun HTTP server for cluster runners.
50 *
51 * @category layers
52 * @since 4.0.0
53 */
54export const layerHttpServer: Layer.Layer<
55 | HttpPlatform
56 | Etag.Generator
57 | BunServices
58 | HttpServer,
59 ServeError,
60 ShardingConfig.ShardingConfig
61> = Effect.gen(function*() {
62 const config = yield* ShardingConfig.ShardingConfig
63 const listenAddress = Option.orElse(config.runnerListenAddress, () => config.runnerAddress)
64 if (Option.isNone(listenAddress)) {
65 return yield* Effect.die("BunClusterHttp.layerHttpServer: ShardingConfig.runnerAddress is None")
66 }
67 return BunHttpServer.layer({ hostname: listenAddress.value.host, port: listenAddress.value.port })
68}).pipe(Layer.unwrap)
69
70/**
71 * Creates Bun cluster layers for HTTP or WebSocket transport, configuring serialization, storage, runner health, and optional client-only mode.
72 *
73 * **When to use**
74 *
75 * Use to install the complete Bun HTTP or WebSocket cluster layer, including
76 * client-only cluster access when a process should connect without serving
77 * runner RPCs.
78 *
79 * **Details**
80 *
81 * `serialization` defaults to SchemaBinary, `runnerHealth` defaults to ping
82 * checks, SQL-backed storage is used by default, and `shardingConfig` is
83 * overlaid on environment-loaded sharding configuration. `local` storage uses
84 * no-op message storage plus in-memory runner storage, while `byo` leaves both
85 * message and runner storage for the caller to provide.
86 *
87 * **Gotchas**
88 *
89 * `clientOnly` does not start the HTTP server or receive shard assignments.
90 * Non-client-only mode listens with `runnerListenAddress` when present, falling
91 * back to `runnerAddress`. HTTP and WebSocket runner RPCs use the default
92 * `HttpRunner` route.
93 *
94 * @see {@link layerHttpServer} for the server layer used by non-client-only transports
95 * @see {@link layerK8sHttpClient} for Kubernetes runner health support
96 *
97 * @category layers
98 * @since 4.0.0
99 */
100export const layer = <
101 const ClientOnly extends boolean = false,
102 const Storage extends "local" | "sql" | "byo" = never
103>(options: {
104 readonly transport: "http" | "websocket"
105 readonly serialization?: "binary" | "ndjson" | undefined
106 readonly serializationMaxBufferSize?: number | "unbounded" | undefined
107 readonly clientOnly?: ClientOnly | undefined
108 readonly storage?: Storage | undefined
109 readonly runnerHealth?: "ping" | "k8s" | undefined
110 readonly runnerHealthK8s?: {
111 readonly namespace?: string | undefined
112 readonly labelSelector?: string | undefined
113 } | undefined
114 readonly shardingConfig?: Partial<ShardingConfig.ShardingConfig["Service"]> | undefined
115}): ClientOnly extends true ? Layer.Layer<
116 Sharding | Runners.Runners | ("byo" extends Storage ? never : MessageStorage.MessageStorage),
117 Config.ConfigError,
118 "local" extends Storage ? never
119 : "byo" extends Storage ? (MessageStorage.MessageStorage | RunnerStorage.RunnerStorage)
120 : SqlClient
121 > :
122 Layer.Layer<
123 Sharding | Runners.Runners | MessageStorage.MessageStorage,
124 ServeError | Config.ConfigError,
125 "local" extends Storage ? never
126 : "byo" extends Storage ? (MessageStorage.MessageStorage | RunnerStorage.RunnerStorage)
127 : SqlClient
128 > =>
129{
130 const layer: Layer.Layer<any, any, any> = options.clientOnly
131 // client only
132 ? options.transport === "http"
133 ? Layer.provide(HttpRunner.layerHttpClientOnly, FetchHttpClient.layer)
134 : Layer.provide(HttpRunner.layerWebsocketClientOnly, BunSocket.layerWebSocketConstructor)
135 // with server
136 : options.transport === "http"
137 ? Layer.provide(HttpRunner.layerHttp, [layerHttpServer, FetchHttpClient.layer])
138 : Layer.provide(HttpRunner.layerWebsocket, [layerHttpServer, BunSocket.layerWebSocketConstructor])
139
140 const runnerHealth: Layer.Layer<any, any, any> = options?.clientOnly
141 ? Layer.empty as any
142 : options?.runnerHealth === "k8s"
143 ? RunnerHealth.layerK8s(options.runnerHealthK8s).pipe(
144 Layer.provide([BunFileSystem.layer, layerK8sHttpClient])
145 )
146 : RunnerHealth.layerPing.pipe(
147 Layer.provide(Runners.layerRpc),
148 Layer.provide(
149 options.transport === "http"
150 ? HttpRunner.layerClientProtocolHttpDefault.pipe(Layer.provide(FetchHttpClient.layer))
151 : HttpRunner.layerClientProtocolWebsocketDefault.pipe(Layer.provide(BunSocket.layerWebSocketConstructor))
152 )
153 )
154
155 return layer.pipe(
156 Layer.provide(runnerHealth),
157 Layer.provideMerge(
158 options?.storage === "local"
159 ? MessageStorage.layerNoop
160 : options?.storage === "byo"
161 ? Layer.empty
162 : Layer.orDie(SqlMessageStorage.layer).pipe(Layer.provide(BunCrypto.layer))
163 ),
164 Layer.provide(
165 options?.storage === "local"
166 ? RunnerStorage.layerMemory
167 : options?.storage === "byo"
168 ? Layer.empty
169 : Layer.orDie(SqlRunnerStorage.layer)
170 ),
171 Layer.provide(ShardingConfig.layerFromEnv(options?.shardingConfig)),
172 Layer.provide(
173 options?.serialization === "ndjson"
174 ? RpcSerialization.layerNdjsonWith({ maxBufferSize: options.serializationMaxBufferSize })
175 : RpcSerialization.layerSchemaBinary({ maxFrameSize: options.serializationMaxBufferSize })
176 )
177 ) as any
178}