1/**
2 * Node.js socket layers for Effect Cluster runners.
3 *
4 * The main `layer` builds a sharding layer for socket transport, choosing
5 * serialization, runner health checks, runner storage, message storage, and
6 * optional client-only mode from the supplied options. This module also
7 * re-exports the shared socket client and server protocol layers, and provides
8 * Kubernetes-aware Undici dispatcher and HTTP client layers for runner health
9 * checks.
10 *
11 * @since 4.0.0
12 */
13import { layerClientProtocol, layerSocketServer } from "@effect/platform-node-shared/NodeClusterSocket"
14import * as K8sHttpClient from "effect/cluster/K8sHttpClient"
15import * as MessageStorage from "effect/cluster/MessageStorage"
16import * as RunnerHealth from "effect/cluster/RunnerHealth"
17import * as Runners from "effect/cluster/Runners"
18import * as RunnerStorage from "effect/cluster/RunnerStorage"
19import type { Sharding } from "effect/cluster/Sharding"
20import * as ShardingConfig from "effect/cluster/ShardingConfig"
21import * as SocketRunner from "effect/cluster/SocketRunner"
22import * as SqlMessageStorage from "effect/cluster/SqlMessageStorage"
23import * as SqlRunnerStorage from "effect/cluster/SqlRunnerStorage"
24import type { ConfigError } from "effect/Config"
25import * as Effect from "effect/Effect"
26import * as FileSystem from "effect/FileSystem"
27import * as Layer from "effect/Layer"
28import * as RpcSerialization from "effect/rpc/RpcSerialization"
29import type * as SocketServer from "effect/socket/SocketServer"
30import type { SqlClient } from "effect/sql/SqlClient"
31import * as NodeCrypto from "./NodeCrypto.ts"
32import * as NodeFileSystem from "./NodeFileSystem.ts"
33import * as NodeHttpClient from "./NodeHttpClient.ts"
34
35export {
36 /**
37 * Provides the cluster `RpcClientProtocol` using the shared socket client
38 * implementation.
39 *
40 * @category re-exports
41 * @since 4.0.0
42 */
43 layerClientProtocol,
44 /**
45 * Provides the socket server used by Node cluster runners through the shared
46 * socket server implementation.
47 *
48 * @category re-exports
49 * @since 4.0.0
50 */
51 layerSocketServer
52}
53
54/**
55 * Builds the Node cluster socket sharding layer, configuring RPC
56 * serialization, message storage, runner health checks, and optional
57 * client-only mode.
58 *
59 * @category layers
60 * @since 4.0.0
61 */
62export const layer = <
63 const ClientOnly extends boolean = false,
64 const Storage extends "local" | "sql" | "byo" = never
65>(
66 options?: {
67 readonly serialization?: "binary" | "ndjson" | undefined
68 readonly serializationMaxBufferSize?: number | "unbounded" | undefined
69 readonly clientOnly?: ClientOnly | undefined
70 readonly storage?: Storage | undefined
71 readonly runnerHealth?: "ping" | "k8s" | undefined
72 readonly runnerHealthK8s?: {
73 readonly namespace?: string | undefined
74 readonly labelSelector?: string | undefined
75 } | undefined
76 readonly shardingConfig?: Partial<ShardingConfig.ShardingConfig["Service"]> | undefined
77 }
78): ClientOnly extends true ? Layer.Layer<
79 Sharding | Runners.Runners | ("byo" extends Storage ? never : MessageStorage.MessageStorage),
80 ConfigError,
81 "local" extends Storage ? never
82 : "byo" extends Storage ? (MessageStorage.MessageStorage | RunnerStorage.RunnerStorage)
83 : SqlClient
84 > :
85 Layer.Layer<
86 Sharding | Runners.Runners | ("byo" extends Storage ? never : MessageStorage.MessageStorage),
87 SocketServer.SocketServerError | ConfigError,
88 "local" extends Storage ? never
89 : "byo" extends Storage ? (MessageStorage.MessageStorage | RunnerStorage.RunnerStorage)
90 : SqlClient
91 > =>
92{
93 const layer: Layer.Layer<any, any, any> = options?.clientOnly
94 // client only
95 ? Layer.provide(SocketRunner.layerClientOnly, layerClientProtocol)
96 // with server
97 : Layer.provide(SocketRunner.layer, [layerSocketServer, layerClientProtocol])
98
99 const runnerHealth: Layer.Layer<any, any, any> = options?.clientOnly
100 ? Layer.empty as any
101 : options?.runnerHealth === "k8s"
102 ? RunnerHealth.layerK8s(options.runnerHealthK8s).pipe(
103 Layer.provide(layerK8sHttpClient)
104 )
105 : RunnerHealth.layerPing.pipe(
106 Layer.provide(Runners.layerRpc),
107 Layer.provide(layerClientProtocol)
108 )
109
110 return layer.pipe(
111 Layer.provide(runnerHealth),
112 Layer.provideMerge(
113 options?.storage === "local"
114 ? MessageStorage.layerNoop
115 : options?.storage === "byo"
116 ? Layer.empty
117 : Layer.orDie(SqlMessageStorage.layer).pipe(Layer.provide(NodeCrypto.layer))
118 ),
119 Layer.provide(
120 options?.storage === "local"
121 ? RunnerStorage.layerMemory
122 : options?.storage === "byo"
123 ? Layer.empty
124 : Layer.orDie(SqlRunnerStorage.layer)
125 ),
126 Layer.provide(ShardingConfig.layerFromEnv(options?.shardingConfig)),
127 Layer.provide(
128 options?.serialization === "ndjson"
129 ? RpcSerialization.layerNdjsonWith({ maxBufferSize: options?.serializationMaxBufferSize })
130 : RpcSerialization.layerSchemaBinary({
131 maxFrameSize: options?.serializationMaxBufferSize
132 })
133 )
134 ) as any
135}
136
137/**
138 * Provides an Undici dispatcher for Kubernetes API calls, using the service
139 * account CA certificate when it is available and falling back to the default
140 * dispatcher otherwise.
141 *
142 * @stability unstable
143 * @category layers
144 * @since 4.0.0
145 */
146export const layerDispatcherK8s: Layer.Layer<NodeHttpClient.Dispatcher> = Layer.effect(NodeHttpClient.Dispatcher)(
147 Effect.gen(function*() {
148 const fs = yield* FileSystem.FileSystem
149 const caCertOption = yield* fs.readFileString("/var/run/secrets/kubernetes.io/serviceaccount/ca.crt").pipe(
150 Effect.option
151 )
152 if (caCertOption._tag === "Some") {
153 const Undici = yield* Effect.promise(() => import("./Undici.ts"))
154 return yield* Effect.acquireRelease(
155 Effect.sync(() =>
156 new Undici.Agent({
157 connect: {
158 ca: caCertOption.value
159 }
160 })
161 ),
162 (agent) => Effect.promise(() => agent.destroy())
163 )
164 }
165
166 return yield* NodeHttpClient.makeDispatcher
167 })
168).pipe(
169 Layer.provide(NodeFileSystem.layer)
170)
171
172/**
173 * Provides a `K8sHttpClient` backed by the Undici HTTP client and the
174 * Kubernetes-aware dispatcher.
175 *
176 * @category layers
177 * @since 4.0.0
178 */
179export const layerK8sHttpClient: Layer.Layer<K8sHttpClient.K8sHttpClient> = K8sHttpClient.layer.pipe(
180 Layer.provide(Layer.fresh(NodeHttpClient.layerUndiciNoDispatcher)),
181 Layer.provide(layerDispatcherK8s),
182 Layer.provide(NodeFileSystem.layer)
183)