1/**
2 * Native Deno 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. Unlike Node sockets,
7 * Deno connections have no native idle-timeout option, so peer connections use
8 * only the one-second open timeout.
9 *
10 * @since 4.0.0
11 */
12import * as K8sHttpClient from "effect/cluster/K8sHttpClient"
13import * as MessageStorage from "effect/cluster/MessageStorage"
14import * as RunnerHealth from "effect/cluster/RunnerHealth"
15import * as Runners from "effect/cluster/Runners"
16import * as RunnerStorage from "effect/cluster/RunnerStorage"
17import type { Sharding } from "effect/cluster/Sharding"
18import * as ShardingConfig from "effect/cluster/ShardingConfig"
19import * as SocketRunner from "effect/cluster/SocketRunner"
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 * as FileSystem from "effect/FileSystem"
25import * as Layer from "effect/Layer"
26import * as Option from "effect/Option"
27import * as RpcClient from "effect/rpc/RpcClient"
28import * as RpcSerialization from "effect/rpc/RpcSerialization"
29import { Socket } from "effect/socket/Socket"
30import type * as SocketServer from "effect/socket/SocketServer"
31import type { SqlClient } from "effect/sql/SqlClient"
32import * as DenoCrypto from "./DenoCrypto.ts"
33import * as DenoFileSystem from "./DenoFileSystem.ts"
34import * as DenoHttpClient from "./DenoHttpClient.ts"
35import * as DenoSocket from "./DenoSocket.ts"
36import * as DenoSocketServer from "./DenoSocketServer.ts"
37
38/**
39 * Provides the cluster `RpcClientProtocol` using native Deno TCP sockets.
40 *
41 * @category layers
42 * @since 4.0.0
43 */
44export const layerClientProtocol: Layer.Layer<
45 Runners.RpcClientProtocol,
46 never,
47 RpcSerialization.RpcSerialization
48> = Layer.effect(Runners.RpcClientProtocol)(
49 Effect.gen(function*() {
50 const serialization = yield* RpcSerialization.RpcSerialization
51 return {
52 codecFor: serialization.codecFor,
53 make: Effect.fnUntraced(function*(address) {
54 const socket = yield* DenoSocket.makeTcp({
55 openTimeout: 1000,
56 hostname: address.host,
57 port: address.port
58 })
59 return yield* RpcClient.makeProtocolSocket().pipe(
60 Effect.provideService(Socket, socket),
61 Effect.provideService(RpcSerialization.RpcSerialization, serialization)
62 )
63 }, Effect.orDie)
64 }
65 })
66)
67
68/**
69 * Provides the socket server used by cluster runners, listening on
70 * `ShardingConfig.runnerListenAddress` or `runnerAddress`.
71 *
72 * @category layers
73 * @since 4.0.0
74 */
75export const layerSocketServer: Layer.Layer<
76 SocketServer.SocketServer,
77 SocketServer.SocketServerError,
78 ShardingConfig.ShardingConfig
79> = Effect.gen(function*() {
80 const config = yield* ShardingConfig.ShardingConfig
81 const listenAddress = Option.orElse(config.runnerListenAddress, () => config.runnerAddress)
82 if (Option.isNone(listenAddress)) {
83 return yield* Effect.die("layerSocketServer: ShardingConfig.runnerListenAddress is None")
84 }
85 return DenoSocketServer.layer({
86 host: listenAddress.value.host,
87 port: listenAddress.value.port
88 })
89}).pipe(Layer.unwrap)
90
91/**
92 * Creates Deno socket cluster layers, configuring serialization, storage,
93 * runner health, and optional client-only mode.
94 *
95 * @category layers
96 * @since 4.0.0
97 */
98export const layer = <
99 const ClientOnly extends boolean = false,
100 const Storage extends "local" | "sql" | "byo" = never
101>(
102 options?: {
103 readonly serialization?: "binary" | "ndjson" | undefined
104 readonly serializationMaxBufferSize?: number | "unbounded" | undefined
105 readonly clientOnly?: ClientOnly | undefined
106 readonly storage?: Storage | undefined
107 readonly runnerHealth?: "ping" | "k8s" | undefined
108 readonly runnerHealthK8s?: {
109 readonly namespace?: string | undefined
110 readonly labelSelector?: string | undefined
111 } | undefined
112 readonly shardingConfig?: Partial<ShardingConfig.ShardingConfig["Service"]> | undefined
113 }
114): ClientOnly extends true ? Layer.Layer<
115 Sharding | Runners.Runners | ("byo" extends Storage ? never : MessageStorage.MessageStorage),
116 Config.ConfigError,
117 "local" extends Storage ? never
118 : "byo" extends Storage ? (MessageStorage.MessageStorage | RunnerStorage.RunnerStorage)
119 : SqlClient
120 > :
121 Layer.Layer<
122 Sharding | Runners.Runners | ("byo" extends Storage ? never : MessageStorage.MessageStorage),
123 SocketServer.SocketServerError | Config.ConfigError,
124 "local" extends Storage ? never
125 : "byo" extends Storage ? (MessageStorage.MessageStorage | RunnerStorage.RunnerStorage)
126 : SqlClient
127 > =>
128{
129 const layer: Layer.Layer<any, any, any> = options?.clientOnly
130 ? Layer.provide(SocketRunner.layerClientOnly, layerClientProtocol)
131 : Layer.provide(SocketRunner.layer, [layerSocketServer, layerClientProtocol])
132
133 const runnerHealth: Layer.Layer<any, any, any> = options?.clientOnly
134 ? Layer.empty as any
135 : options?.runnerHealth === "k8s"
136 ? RunnerHealth.layerK8s(options.runnerHealthK8s).pipe(
137 Layer.provide(layerK8sHttpClient)
138 )
139 : RunnerHealth.layerPing.pipe(
140 Layer.provide(Runners.layerRpc),
141 Layer.provide(layerClientProtocol)
142 )
143
144 return layer.pipe(
145 Layer.provide(runnerHealth),
146 Layer.provideMerge(
147 options?.storage === "local"
148 ? MessageStorage.layerNoop
149 : options?.storage === "byo"
150 ? Layer.empty
151 : Layer.orDie(SqlMessageStorage.layer).pipe(Layer.provide(DenoCrypto.layer))
152 ),
153 Layer.provide(
154 options?.storage === "local"
155 ? RunnerStorage.layerMemory
156 : options?.storage === "byo"
157 ? Layer.empty
158 : Layer.orDie(SqlRunnerStorage.layer)
159 ),
160 Layer.provide(ShardingConfig.layerFromEnv(options?.shardingConfig)),
161 Layer.provide(
162 options?.serialization === "ndjson"
163 ? RpcSerialization.layerNdjsonWith({ maxBufferSize: options?.serializationMaxBufferSize })
164 : RpcSerialization.layerSchemaBinary({
165 maxFrameSize: options?.serializationMaxBufferSize
166 })
167 )
168 ) as any
169}
170
171/**
172 * Layer that provides `K8sHttpClient`, using a scoped native Deno HTTP client
173 * with the Kubernetes service-account CA certificate when it is available.
174 *
175 * @category layers
176 * @since 4.0.0
177 */
178export const layerK8sHttpClient: Layer.Layer<K8sHttpClient.K8sHttpClient> = K8sHttpClient.layer.pipe(
179 Layer.provide(
180 Layer.fresh(DenoHttpClient.layer).pipe(
181 Layer.provide(Layer.effect(
182 DenoHttpClient.Fetch,
183 Effect.gen(function*() {
184 const fs = yield* FileSystem.FileSystem
185 const caCertOption = yield* fs.readFileString("/var/run/secrets/kubernetes.io/serviceaccount/ca.crt").pipe(
186 Effect.option
187 )
188 if (Option.isNone(caCertOption)) {
189 return globalThis.fetch
190 }
191
192 const client = yield* Effect.acquireRelease(
193 Effect.sync(() => Deno.createHttpClient({ caCerts: [caCertOption.value] })),
194 (client) => Effect.sync(() => client.close())
195 )
196 return ((input, init) => globalThis.fetch(input, { ...init, client } as any)) as typeof globalThis.fetch
197 })
198 ))
199 )
200 ),
201 Layer.provide(DenoFileSystem.layer)
202)