import type { EnvironmentId, ServerConfig } from "@t3tools/contracts";import * as Effect from "effect/Effect";import * as Option from "effect/Option";import * as Stream from "effect/Stream";import * as SubscriptionRef from "effect/SubscriptionRef";import { AsyncResult, Atom } from "effect/unstable/reactivity"; import { EnvironmentRegistry } from "../connection/registry.ts";import type { PreparedConnection } from "../connection/model.ts";import { EnvironmentSupervisor } from "../connection/supervisor.ts";import { followStreamInEnvironment } from "./runtime.ts"; export function initialConfigOption<E>(  initialConfig: Effect.Effect<ServerConfig, E>,): Effect.Effect<Option.Option<ServerConfig>> {  return initialConfig.pipe(    Effect.map(Option.some),    Effect.catch((error) =>      Effect.logWarning("Could not load the initial environment configuration.", {        error,      }).pipe(Effect.as(Option.none<ServerConfig>())),    ),  );} export function createEnvironmentSessionAtoms<R, E>(  runtime: Atom.AtomRuntime<EnvironmentRegistry | R, E>,) {  const configAtom = Atom.family((environmentId: EnvironmentId) =>    runtime.atom(      followStreamInEnvironment(        environmentId,        Stream.unwrap(          EnvironmentSupervisor.pipe(            Effect.map((supervisor) =>              SubscriptionRef.changes(supervisor.session).pipe(                Stream.mapEffect(                  Option.match({                    onNone: () => Effect.succeed(Option.none<ServerConfig>()),                    onSome: (session) => initialConfigOption(session.initialConfig),                  }),                ),              ),            ),          ),        ),      ),      { initialValue: Option.none() },    ),  );   const configValueAtom = Atom.family((environmentId: EnvironmentId) =>    Atom.make((get): ServerConfig | null =>      Option.getOrNull(        Option.getOrElse(AsyncResult.value(get(configAtom(environmentId))), () => Option.none()),      ),    ).pipe(Atom.withLabel(`environment-config-value:${environmentId}`)),  );   const preparedConnectionAtom = Atom.family((environmentId: EnvironmentId) =>    runtime.atom(      followStreamInEnvironment(        environmentId,        Stream.unwrap(          EnvironmentSupervisor.pipe(            Effect.map((supervisor) => SubscriptionRef.changes(supervisor.prepared)),          ),        ),      ),      { initialValue: Option.none<PreparedConnection>() },    ),  );   const preparedConnectionValueAtom = Atom.family((environmentId: EnvironmentId) =>    Atom.make((get) =>      Option.getOrElse(AsyncResult.value(get(preparedConnectionAtom(environmentId))), () =>        Option.none<PreparedConnection>(),      ),    ).pipe(Atom.withLabel(`environment-prepared-connection:${environmentId}`)),  );   return {    configAtom,    configValueAtom,    preparedConnectionAtom,    preparedConnectionValueAtom,  };}