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