1/**
2 * Bun 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 * `layerK8sHttpClient` for Kubernetes runner health checks.
9 *
10 * @since 4.0.0
11 */
12import { layerClientProtocol, layerSocketServer } from "@effect/platform-node-shared/NodeClusterSocket"
13import * as K8sHttpClient from "effect/cluster/K8sHttpClient"
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 SocketRunner from "effect/cluster/SocketRunner"
21import * as SqlMessageStorage from "effect/cluster/SqlMessageStorage"
22import * as SqlRunnerStorage from "effect/cluster/SqlRunnerStorage"
23import type * as Config from "effect/Config"
24import * as Effect from "effect/Effect"
25import * as FileSystem from "effect/FileSystem"
26import * as FetchHttpClient from "effect/http/FetchHttpClient"
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 BunCrypto from "./BunCrypto.ts"
32import * as BunFileSystem from "./BunFileSystem.ts"
33
34export {
35 /**
36 * Provides the cluster `RpcClientProtocol` using the shared socket client
37 * implementation.
38 *
39 * @category re-exports
40 * @since 4.0.0
41 */
42 layerClientProtocol,
43 /**
44 * Provides the socket server used by Bun cluster runners through the shared
45 * socket server implementation.
46 *
47 * @category re-exports
48 * @since 4.0.0
49 */
50 layerSocketServer
51}
52
53/**
54 * Creates Bun socket cluster layers, configuring serialization, storage, runner health, and optional client-only mode.
55 *
56 * @category layers
57 * @since 4.0.0
58 */
59export const layer = <
60 const ClientOnly extends boolean = false,
61 const Storage extends "local" | "sql" | "byo" = never
62>(
63 options?: {
64 readonly serialization?: "binary" | "ndjson" | undefined
65 readonly serializationMaxBufferSize?: number | "unbounded" | undefined
66 readonly clientOnly?: ClientOnly | undefined
67 readonly storage?: Storage | undefined
68 readonly runnerHealth?: "ping" | "k8s" | undefined
69 readonly runnerHealthK8s?: {
70 readonly namespace?: string | undefined
71 readonly labelSelector?: string | undefined
72 } | undefined
73 readonly shardingConfig?: Partial<ShardingConfig.ShardingConfig["Service"]> | undefined
74 }
75): ClientOnly extends true ? Layer.Layer<
76 Sharding | Runners.Runners | ("byo" extends Storage ? never : MessageStorage.MessageStorage),
77 Config.ConfigError,
78 "local" extends Storage ? never
79 : "byo" extends Storage ? (MessageStorage.MessageStorage | RunnerStorage.RunnerStorage)
80 : SqlClient
81 > :
82 Layer.Layer<
83 Sharding | Runners.Runners | ("byo" extends Storage ? never : MessageStorage.MessageStorage),
84 SocketServer.SocketServerError | Config.ConfigError,
85 "local" extends Storage ? never
86 : "byo" extends Storage ? (MessageStorage.MessageStorage | RunnerStorage.RunnerStorage)
87 : SqlClient
88 > =>
89{
90 const layer: Layer.Layer<any, any, any> = options?.clientOnly
91 // client only
92 ? Layer.provide(SocketRunner.layerClientOnly, layerClientProtocol)
93 // with server
94 : Layer.provide(SocketRunner.layer, [layerSocketServer, layerClientProtocol])
95
96 const runnerHealth: Layer.Layer<any, any, any> = options?.clientOnly
97 ? Layer.empty as any
98 : options?.runnerHealth === "k8s"
99 ? RunnerHealth.layerK8s(options.runnerHealthK8s).pipe(
100 Layer.provide([BunFileSystem.layer, layerK8sHttpClient])
101 )
102 : RunnerHealth.layerPing.pipe(
103 Layer.provide(Runners.layerRpc),
104 Layer.provide(layerClientProtocol)
105 )
106
107 return layer.pipe(
108 Layer.provide(runnerHealth),
109 Layer.provideMerge(
110 options?.storage === "local"
111 ? MessageStorage.layerNoop
112 : options?.storage === "byo"
113 ? Layer.empty
114 : Layer.orDie(SqlMessageStorage.layer).pipe(Layer.provide(BunCrypto.layer))
115 ),
116 Layer.provide(
117 options?.storage === "local"
118 ? RunnerStorage.layerMemory
119 : options?.storage === "byo"
120 ? Layer.empty
121 : Layer.orDie(SqlRunnerStorage.layer)
122 ),
123 Layer.provide(ShardingConfig.layerFromEnv(options?.shardingConfig)),
124 Layer.provide(
125 options?.serialization === "ndjson"
126 ? RpcSerialization.layerNdjsonWith({ maxBufferSize: options?.serializationMaxBufferSize })
127 : RpcSerialization.layerSchemaBinary({
128 maxFrameSize: options?.serializationMaxBufferSize
129 })
130 )
131 ) as any
132}
133
134/**
135 * Layer that provides `K8sHttpClient`, using the Kubernetes service-account CA certificate when it is available.
136 *
137 * @category layers
138 * @since 4.0.0
139 */
140export const layerK8sHttpClient: Layer.Layer<K8sHttpClient.K8sHttpClient> = K8sHttpClient.layer.pipe(
141 Layer.provide(Layer.unwrap(Effect.gen(function*() {
142 const fs = yield* FileSystem.FileSystem
143 const caCertOption = yield* fs.readFile("/var/run/secrets/kubernetes.io/serviceaccount/ca.crt").pipe(
144 Effect.option
145 )
146 if (caCertOption._tag === "None") {
147 return FetchHttpClient.layer
148 }
149
150 return Layer.fresh(FetchHttpClient.layer).pipe(
151 Layer.provide(Layer.succeed(FetchHttpClient.RequestInit, {
152 tls: {
153 ca: caCertOption.value
154 }
155 } as any))
156 )
157 }))),
158 Layer.provide(BunFileSystem.layer)
159)