import type { EnvironmentId as EnvironmentIdType } 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 * as EnvironmentRegistry from "../connection/registry.ts";import type { ConnectionCatalogEntry } from "../connection/catalog.ts";import { AVAILABLE_CONNECTION_STATE } from "../connection/model.ts";import * as EnvironmentSupervisor from "../connection/supervisor.ts";import { createAtomCommandScheduler, createRuntimeCommand, followStreamInEnvironment,} from "./runtime.ts"; export interface EnvironmentCatalogState { readonly isReady: boolean; readonly entries: ReadonlyMap<EnvironmentIdType, ConnectionCatalogEntry>;} const EMPTY_ENVIRONMENT_CATALOG_STATE: EnvironmentCatalogState = Object.freeze({ isReady: false, entries: new Map(),}); export function createEnvironmentCatalogAtoms<R, E>( runtime: Atom.AtomRuntime<EnvironmentRegistry.EnvironmentRegistry | R, E>,) { const commandScheduler = createAtomCommandScheduler(); const serial = { mode: "serial" as const, key: () => "environment-catalog" }; const catalogAtom = runtime.atom( Stream.unwrap( EnvironmentRegistry.EnvironmentRegistry.pipe( Effect.map((registry) => SubscriptionRef.changes(registry.entries).pipe( Stream.map((entries) => ({ isReady: true, entries, })), ), ), ), ), { initialValue: EMPTY_ENVIRONMENT_CATALOG_STATE }, ); const catalogValueAtom = Atom.make((get) => Option.getOrElse(AsyncResult.value(get(catalogAtom)), () => EMPTY_ENVIRONMENT_CATALOG_STATE), ).pipe(Atom.withLabel("environment-catalog-value")); const networkStatusAtom = runtime.atom( Stream.unwrap( EnvironmentRegistry.EnvironmentRegistry.pipe( Effect.map((registry) => SubscriptionRef.changes(registry.networkStatus)), ), ), { initialValue: "unknown" as const }, ); const networkStatusValueAtom = Atom.make((get) => Option.getOrElse(AsyncResult.value(get(networkStatusAtom)), () => "unknown" as const), ).pipe(Atom.withLabel("environment-network-status-value")); const stateAtom = Atom.family((environmentId: EnvironmentIdType) => runtime.atom( followStreamInEnvironment( environmentId, Stream.unwrap( EnvironmentSupervisor.EnvironmentSupervisor.pipe( Effect.map((supervisor) => SubscriptionRef.changes(supervisor.state)), ), ), ), { initialValue: AVAILABLE_CONNECTION_STATE }, ), ); const register = createRuntimeCommand(runtime, { label: "environment-catalog:register", scheduler: commandScheduler, concurrency: serial, execute: ( target: Parameters<EnvironmentRegistry.EnvironmentRegistry["Service"]["register"]>[0], ) => EnvironmentRegistry.EnvironmentRegistry.pipe( Effect.flatMap((registry) => registry.register(target)), ), }); const remove = createRuntimeCommand(runtime, { label: "environment-catalog:remove", scheduler: commandScheduler, concurrency: serial, execute: (environmentId: EnvironmentIdType) => EnvironmentRegistry.EnvironmentRegistry.pipe( Effect.flatMap((registry) => registry.remove(environmentId)), ), }); const removeRelayEnvironments = createRuntimeCommand(runtime, { label: "environment-catalog:remove-relay-environments", scheduler: commandScheduler, concurrency: serial, execute: (_input: void) => EnvironmentRegistry.EnvironmentRegistry.pipe( Effect.flatMap((registry) => registry.removeRelayEnvironments()), ), }); const retryNow = createRuntimeCommand(runtime, { label: "environment-catalog:retry-now", scheduler: commandScheduler, concurrency: serial, execute: (environmentId: EnvironmentIdType) => EnvironmentRegistry.EnvironmentRegistry.pipe( Effect.flatMap((registry) => registry.retryNow(environmentId)), ), }); return { catalogAtom, catalogValueAtom, networkStatusAtom, networkStatusValueAtom, stateAtom, register, remove, removeRelayEnvironments, retryNow, };}
import type { EnvironmentId as EnvironmentIdType } 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 * as EnvironmentRegistry from "../connection/registry.ts";
import type { ConnectionCatalogEntry } from "../connection/catalog.ts";
import { AVAILABLE_CONNECTION_STATE } from "../connection/model.ts";
import * as EnvironmentSupervisor from "../connection/supervisor.ts";
import {
createAtomCommandScheduler,
createRuntimeCommand,
followStreamInEnvironment,
} from "./runtime.ts";
export interface EnvironmentCatalogState {
readonly isReady: boolean;
readonly entries: ReadonlyMap<EnvironmentIdType, ConnectionCatalogEntry>;
}
const EMPTY_ENVIRONMENT_CATALOG_STATE: EnvironmentCatalogState = Object.freeze({
isReady: false,
entries: new Map(),
});
export function createEnvironmentCatalogAtoms<R, E>(
runtime: Atom.AtomRuntime<EnvironmentRegistry.EnvironmentRegistry | R, E>,
) {
const commandScheduler = createAtomCommandScheduler();
const serial = { mode: "serial" as const, key: () => "environment-catalog" };
const catalogAtom = runtime.atom(
Stream.unwrap(
EnvironmentRegistry.EnvironmentRegistry.pipe(
Effect.map((registry) =>
SubscriptionRef.changes(registry.entries).pipe(
Stream.map((entries) => ({
isReady: true,
entries,
})),
),
),
),
),
{ initialValue: EMPTY_ENVIRONMENT_CATALOG_STATE },
);
const catalogValueAtom = Atom.make((get) =>
Option.getOrElse(AsyncResult.value(get(catalogAtom)), () => EMPTY_ENVIRONMENT_CATALOG_STATE),
).pipe(Atom.withLabel("environment-catalog-value"));
const networkStatusAtom = runtime.atom(
Stream.unwrap(
EnvironmentRegistry.EnvironmentRegistry.pipe(
Effect.map((registry) => SubscriptionRef.changes(registry.networkStatus)),
),
),
{ initialValue: "unknown" as const },
);
const networkStatusValueAtom = Atom.make((get) =>
Option.getOrElse(AsyncResult.value(get(networkStatusAtom)), () => "unknown" as const),
).pipe(Atom.withLabel("environment-network-status-value"));
const stateAtom = Atom.family((environmentId: EnvironmentIdType) =>
runtime.atom(
followStreamInEnvironment(
environmentId,
Stream.unwrap(
EnvironmentSupervisor.EnvironmentSupervisor.pipe(
Effect.map((supervisor) => SubscriptionRef.changes(supervisor.state)),
),
),
),
{ initialValue: AVAILABLE_CONNECTION_STATE },
),
);
const register = createRuntimeCommand(runtime, {
label: "environment-catalog:register",
scheduler: commandScheduler,
concurrency: serial,
execute: (
target: Parameters<EnvironmentRegistry.EnvironmentRegistry["Service"]["register"]>[0],
) =>
EnvironmentRegistry.EnvironmentRegistry.pipe(
Effect.flatMap((registry) => registry.register(target)),
),
});
const remove = createRuntimeCommand(runtime, {
label: "environment-catalog:remove",
scheduler: commandScheduler,
concurrency: serial,
execute: (environmentId: EnvironmentIdType) =>
EnvironmentRegistry.EnvironmentRegistry.pipe(
Effect.flatMap((registry) => registry.remove(environmentId)),
),
});
const removeRelayEnvironments = createRuntimeCommand(runtime, {
label: "environment-catalog:remove-relay-environments",
scheduler: commandScheduler,
concurrency: serial,
execute: (_input: void) =>
EnvironmentRegistry.EnvironmentRegistry.pipe(
Effect.flatMap((registry) => registry.removeRelayEnvironments()),
),
});
const retryNow = createRuntimeCommand(runtime, {
label: "environment-catalog:retry-now",
scheduler: commandScheduler,
concurrency: serial,
execute: (environmentId: EnvironmentIdType) =>
EnvironmentRegistry.EnvironmentRegistry.pipe(
Effect.flatMap((registry) => registry.retryNow(environmentId)),
),
});
return {
catalogAtom,
catalogValueAtom,
networkStatusAtom,
networkStatusValueAtom,
stateAtom,
register,
remove,
removeRelayEnvironments,
retryNow,
};
}