/** * @since 1.0.0 */ import * as HttpRunner from "@effect/cluster/HttpRunner" import * as MessageStorage from "@effect/cluster/MessageStorage" import * as RunnerHealth from "@effect/cluster/RunnerHealth" import * as Runners from "@effect/cluster/Runners" import * as RunnerStorage from "@effect/cluster/RunnerStorage" import type { Sharding } from "@effect/cluster/Sharding" import * as ShardingConfig from "@effect/cluster/ShardingConfig" import * as SqlMessageStorage from "@effect/cluster/SqlMessageStorage" import * as SqlRunnerStorage from "@effect/cluster/SqlRunnerStorage" import type * as Etag from "@effect/platform/Etag" import * as FetchHttpClient from "@effect/platform/FetchHttpClient" import type { HttpPlatform } from "@effect/platform/HttpPlatform" import type { HttpServer } from "@effect/platform/HttpServer" import type { ServeError } from "@effect/platform/HttpServerError" import * as RpcSerialization from "@effect/rpc/RpcSerialization" import type { SqlClient } from "@effect/sql/SqlClient" import type { ConfigError } from "effect/ConfigError" import * as Effect from "effect/Effect" import * as Layer from "effect/Layer" import * as Option from "effect/Option" import type { BunContext } from "./BunContext.js" import * as BunHttpServer from "./BunHttpServer.js" import * as BunSocket from "./BunSocket.js" /** * @since 1.0.0 * @category Layers */ export const layerHttpServer: Layer.Layer< | HttpPlatform | Etag.Generator | BunContext | HttpServer, ServeError, ShardingConfig.ShardingConfig > = Effect.gen(function*() { const config = yield* ShardingConfig.ShardingConfig const listenAddress = Option.orElse(config.runnerListenAddress, () => config.runnerAddress) if (listenAddress._tag === "None") { return yield* Effect.die("BunClusterHttp.layerHttpServer: ShardingConfig.runnerAddress is None") } return BunHttpServer.layer(listenAddress.value) }).pipe(Layer.unwrapEffect) /** * @since 1.0.0 * @category Layers */ export const layer = < const ClientOnly extends boolean = false, const Storage extends "local" | "sql" | "byo" = never >(options: { readonly transport: "http" | "websocket" readonly serialization?: "msgpack" | "ndjson" | undefined readonly clientOnly?: ClientOnly | undefined readonly storage?: Storage | undefined readonly shardingConfig?: Partial | undefined }): ClientOnly extends true ? Layer.Layer< Sharding | Runners.Runners | ("byo" extends Storage ? never : MessageStorage.MessageStorage), ConfigError, "local" extends Storage ? never : "byo" extends Storage ? (MessageStorage.MessageStorage | RunnerStorage.RunnerStorage) : SqlClient > : Layer.Layer< Sharding | Runners.Runners | MessageStorage.MessageStorage, ServeError | ConfigError, "local" extends Storage ? never : "byo" extends Storage ? (MessageStorage.MessageStorage | RunnerStorage.RunnerStorage) : SqlClient > => { const layer: Layer.Layer = options.clientOnly // client only ? options.transport === "http" ? Layer.provide(HttpRunner.layerHttpClientOnly, FetchHttpClient.layer) : Layer.provide(HttpRunner.layerWebsocketClientOnly, BunSocket.layerWebSocketConstructor) // with server : options.transport === "http" ? Layer.provide(HttpRunner.layerHttp, [layerHttpServer, FetchHttpClient.layer]) : Layer.provide(HttpRunner.layerWebsocket, [layerHttpServer, BunSocket.layerWebSocketConstructor]) const runnerHealth: Layer.Layer = options?.clientOnly ? Layer.empty as any // TODO: when bun supports adding custom CA certificates // : options?.runnerHealth === "k8s" // ? RunnerHealth.layerK8s().pipe( // Layer.provide([NodeFileSystem.layer, layerHttpClientK8s]) // ) : RunnerHealth.layerPing.pipe( Layer.provide(Runners.layerRpc), Layer.provide( options.transport === "http" ? HttpRunner.layerClientProtocolHttpDefault.pipe(Layer.provide(FetchHttpClient.layer)) : HttpRunner.layerClientProtocolWebsocketDefault.pipe(Layer.provide(BunSocket.layerWebSocketConstructor)) ) ) return layer.pipe( Layer.provide(runnerHealth), Layer.provideMerge( options?.storage === "local" ? MessageStorage.layerNoop : options?.storage === "byo" ? Layer.empty : Layer.orDie(SqlMessageStorage.layer) ), Layer.provide( options?.storage === "local" ? RunnerStorage.layerMemory : options?.storage === "byo" ? Layer.empty : Layer.orDie(SqlRunnerStorage.layer) ), Layer.provide(ShardingConfig.layerFromEnv(options?.shardingConfig)), Layer.provide( options?.serialization === "ndjson" ? RpcSerialization.layerNdjson : RpcSerialization.layerMsgPack ) ) as any }