import {  ORCHESTRATION_WS_METHODS,  type EnvironmentId,  type OrchestrationShellSnapshot,  type OrchestrationShellStreamItem,  type ServerConfig,} from "@t3tools/contracts";import * as Cause from "effect/Cause";import * as Effect from "effect/Effect";import * as Option from "effect/Option";import * as Queue from "effect/Queue";import * as Ref from "effect/Ref";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 { connectionProjectionPhase } from "../connection/model.ts";import { EnvironmentSupervisor } from "../connection/supervisor.ts";import * as ConnectionWakeups from "../connection/wakeups.ts";import { safeErrorLogAttributes } from "../errors/safeLog.ts";import { EnvironmentCacheStore } from "../platform/persistence.ts";import { subscribeDynamic } from "../rpc/client.ts";import type { RpcSession } from "../rpc/session.ts";import { ShellSnapshotLoader } from "./shellSnapshotHttp.ts";import { applyShellStreamEvent } from "./shellReducer.ts";import type { EnvironmentCatalogState } from "./connections.ts";import { followStreamInEnvironment } from "./runtime.ts"; export type EnvironmentShellStatus = "empty" | "cached" | "synchronizing" | "live"; export interface EnvironmentShellState {  readonly snapshot: Option.Option<OrchestrationShellSnapshot>;  readonly status: EnvironmentShellStatus;  readonly error: Option.Option<string>;} const EMPTY_SHELL_STATE: EnvironmentShellState = {  snapshot: Option.none(),  status: "empty",  error: Option.none(),}; function shellStatusForSnapshot(  snapshot: Option.Option<OrchestrationShellSnapshot>,): EnvironmentShellStatus {  return Option.isSome(snapshot) ? "cached" : "empty";} const SHELL_SYNCHRONIZATION_ERROR_MESSAGE = "Could not synchronize environment data."; export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")(function* () {  const supervisor = yield* EnvironmentSupervisor;  const cache = yield* EnvironmentCacheStore;  const snapshotLoader = yield* ShellSnapshotLoader;  const wakeups = yield* Effect.serviceOption(ConnectionWakeups.ConnectionWakeups);  const environmentId = supervisor.target.environmentId;  const cachedSnapshot = yield* cache.loadShell(environmentId).pipe(    Effect.catch((error) =>      Effect.logWarning("Could not load cached environment shell.").pipe(        Effect.annotateLogs({          environmentId,          ...safeErrorLogAttributes(error),        }),        Effect.as(Option.none<OrchestrationShellSnapshot>()),      ),    ),  );  const state = yield* SubscriptionRef.make<EnvironmentShellState>({    snapshot: cachedSnapshot,    status: shellStatusForSnapshot(cachedSnapshot),    error: Option.none(),  });  const awaitingCompletion = yield* Ref.make(false);  const lastAuthoritativeSession = yield* Ref.make<RpcSession | null>(null);  const activeSubscriptionSession = yield* Ref.make<RpcSession | null>(null);  const persistence = yield* Queue.sliding<OrchestrationShellSnapshot>(1);   const persist = Effect.fn("EnvironmentShellState.persist")(function* (    snapshot: OrchestrationShellSnapshot,  ) {    yield* cache.saveShell(environmentId, snapshot).pipe(      Effect.catch((error) =>        Effect.logWarning("Could not persist environment shell cache.").pipe(          Effect.annotateLogs({            environmentId,            ...safeErrorLogAttributes(error),          }),        ),      ),    );  });   yield* Stream.fromQueue(persistence).pipe(    Stream.debounce("500 millis"),    Stream.runForEach(persist),    Effect.forkScoped,  );   const setDisconnected = Ref.set(awaitingCompletion, false).pipe(    Effect.andThen(      SubscriptionRef.update(state, (current) => ({        ...current,        status: shellStatusForSnapshot(current.snapshot),      })),    ),  );  const setSynchronizing = SubscriptionRef.update(state, (current) => ({    ...current,    status: "synchronizing" as const,    error: Option.none(),  }));  const setReady = SubscriptionRef.update(state, (current) =>    current.status === "live"      ? current      : {          ...current,          status: "synchronizing" as const,          error: Option.none(),        },  );  const setStreamError = (error: unknown) =>    Ref.set(awaitingCompletion, false).pipe(      Effect.andThen(Effect.logWarning("Could not synchronize the environment shell.")),      Effect.annotateLogs({        environmentId,        ...safeErrorLogAttributes(error),      }),      Effect.andThen(        SubscriptionRef.update(state, (current) => ({          ...current,          status: shellStatusForSnapshot(current.snapshot),          error: Option.some(SHELL_SYNCHRONIZATION_ERROR_MESSAGE),        })),      ),    );   // Apply each received batch with one state write. The RPC client's bounded  // buffer can split a server chunk, so a bulk action can still need several  // writes, but each write includes every event in that batch.  const applyItems = Effect.fn("EnvironmentShellState.applyItems")(function* (    items: ReadonlyArray<OrchestrationShellStreamItem>,  ) {    const initial = yield* SubscriptionRef.get(state);    let waiting = yield* Ref.get(awaitingCompletion);    let next = initial;    let receivedSnapshot = false;    for (const item of items) {      if (item.kind === "synchronized") {        waiting = false;        if (Option.isSome(next.snapshot)) {          next = { ...next, status: "live", error: Option.none() };        }        continue;      }      const nextSnapshot =        item.kind === "snapshot"          ? item.snapshot          : Option.match(next.snapshot, {              onNone: () => null,              onSome: (snapshot) =>                item.sequence > snapshot.snapshotSequence                  ? applyShellStreamEvent(snapshot, item)                  : snapshot,            });      if (nextSnapshot === null) continue;      receivedSnapshot ||= item.kind === "snapshot";      next = {        snapshot: Option.some(nextSnapshot),        status: waiting ? "synchronizing" : "live",        error: Option.none(),      };    }    yield* Ref.set(awaitingCompletion, waiting);    if (next === initial) return;    yield* SubscriptionRef.set(state, next);    if (receivedSnapshot) {      const session = yield* Ref.get(activeSubscriptionSession);      if (session !== null) {        yield* Ref.set(lastAuthoritativeSession, session);      }    }    if (next.snapshot !== initial.snapshot && Option.isSome(next.snapshot)) {      yield* Queue.offer(persistence, next.snapshot.value);    }  });   const foregroundResubscriptions = Option.match(wakeups, {    onNone: () => Stream.never,    onSome: (service) =>      service.changes.pipe(Stream.filter(ConnectionWakeups.shouldResubscribeAfterWakeup)),  });   yield* setSynchronizing;  yield* Effect.forkScoped(    subscribeDynamic(      ORCHESTRATION_WS_METHODS.subscribeShell,      Effect.fn("EnvironmentShellState.makeSubscribeInput")(function* (session) {        yield* Ref.set(activeSubscriptionSession, session);        const supportsCompletionMarker = yield* session.initialConfig.pipe(          Effect.map((config) => config.shellResumeCompletionMarker === true),          Effect.orElseSucceed(() => false),        );        yield* Ref.set(awaitingCompletion, supportsCompletionMarker);        yield* setSynchronizing;         // Foreground resubscriptions on the same live session can resume from        // the in-memory cursor. A new session reloads the authoritative HTTP        // snapshot so a valid cursor cannot preserve incomplete cached data.        const hasAuthoritativeSnapshot = (yield* Ref.get(lastAuthoritativeSession)) === session;        let canResume = hasAuthoritativeSnapshot;        let current = yield* SubscriptionRef.get(state);        if (!hasAuthoritativeSnapshot || Option.isNone(current.snapshot)) {          const prepared = yield* SubscriptionRef.get(supervisor.prepared).pipe(            Effect.flatMap(              Option.match({                onSome: Effect.succeed,                onNone: () =>                  SubscriptionRef.changes(supervisor.prepared).pipe(                    Stream.filter(Option.isSome),                    Stream.map((value) => value.value),                    Stream.runHead,                    Effect.map(Option.getOrThrow),                  ),              }),            ),          );          const httpSnapshot = yield* snapshotLoader.load(prepared);          if (Option.isSome(httpSnapshot)) {            yield* applyItems([{ kind: "snapshot", snapshot: httpSnapshot.value }]);            canResume = true;            current = yield* SubscriptionRef.get(state);          }        }         // If the authoritative refresh failed, omit the cached cursor so the        // socket fallback sends a complete snapshot for this new session.        if (!canResume || Option.isNone(current.snapshot)) {          return supportsCompletionMarker ? { requestCompletionMarker: true as const } : {};        }        if (!supportsCompletionMarker) {          // Without a completion marker there is no synchronized signal for a          // resumed subscription, so report live immediately, like threads.          yield* SubscriptionRef.update(state, (value) => ({            ...value,            status: "live" as const,            error: Option.none(),          }));        }        return {          afterSequence: current.snapshot.value.snapshotSequence,          ...(supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}),        };      }),      {        onExpectedFailure: (cause) => setStreamError(Cause.squash(cause)),        retryExpectedFailureAfter: "250 millis",        resubscribe: foregroundResubscriptions,      },    ).pipe(Stream.runForEachArray(applyItems)),  );  yield* SubscriptionRef.changes(supervisor.state).pipe(    Stream.runForEach((connectionState) => {      switch (connectionProjectionPhase(connectionState)) {        case "synchronizing":          return setSynchronizing;        case "disconnected":          return setDisconnected;        case "ready":          return setReady;      }    }),    Effect.forkScoped,  );   return state;}); function shellStateChanges(environmentId: EnvironmentId) {  return followStreamInEnvironment(    environmentId,    Stream.unwrap(makeEnvironmentShellState().pipe(Effect.map(SubscriptionRef.changes))),  );} export interface EnvironmentShellSummary {  readonly hasSnapshot: boolean;  readonly hasSynchronizingShell: boolean;  readonly hasCachedShell: boolean;  readonly hasLiveShell: boolean;  readonly firstError: string | null;  readonly latestSnapshotUpdatedAt: string | null;} const EMPTY_ENVIRONMENT_SHELL_SUMMARY: EnvironmentShellSummary = Object.freeze({  hasSnapshot: false,  hasSynchronizingShell: false,  hasCachedShell: false,  hasLiveShell: false,  firstError: null,  latestSnapshotUpdatedAt: null,}); const EMPTY_SERVER_CONFIGS: ReadonlyMap<EnvironmentId, ServerConfig> = new Map(); function shellSummariesEqual(  left: EnvironmentShellSummary,  right: EnvironmentShellSummary,): boolean {  return (    left.hasSnapshot === right.hasSnapshot &&    left.hasSynchronizingShell === right.hasSynchronizingShell &&    left.hasCachedShell === right.hasCachedShell &&    left.hasLiveShell === right.hasLiveShell &&    left.firstError === right.firstError &&    left.latestSnapshotUpdatedAt === right.latestSnapshotUpdatedAt  );} function mapsEqual<K, V>(left: ReadonlyMap<K, V>, right: ReadonlyMap<K, V>): boolean {  if (left.size !== right.size) {    return false;  }  for (const [key, value] of left) {    if (right.get(key) !== value) {      return false;    }  }  return true;} export function createEnvironmentShellSummaryAtom(input: {  readonly catalogValueAtom: Atom.Atom<EnvironmentCatalogState>;  readonly shellStateValueAtom: (environmentId: EnvironmentId) => Atom.Atom<EnvironmentShellState>;}) {  let previousSummary = EMPTY_ENVIRONMENT_SHELL_SUMMARY;  return Atom.make((get) => {    let hasSnapshot = false;    let hasSynchronizingShell = false;    let hasCachedShell = false;    let hasLiveShell = false;    let firstError: string | null = null;    let latestSnapshotUpdatedAt: string | null = null;     for (const environmentId of get(input.catalogValueAtom).entries.keys()) {      const state = get(input.shellStateValueAtom(environmentId));      hasSynchronizingShell ||= state.status === "synchronizing";      hasCachedShell ||= state.status === "cached";      hasLiveShell ||= state.status === "live";      if (firstError === null) {        firstError = Option.getOrNull(state.error);      }      if (Option.isNone(state.snapshot)) {        continue;      }      hasSnapshot = true;      const updatedAt = state.snapshot.value.updatedAt;      if (latestSnapshotUpdatedAt === null || updatedAt > latestSnapshotUpdatedAt) {        latestSnapshotUpdatedAt = updatedAt;      }    }     const next: EnvironmentShellSummary = {      hasSnapshot,      hasSynchronizingShell,      hasCachedShell,      hasLiveShell,      firstError,      latestSnapshotUpdatedAt,    };    if (shellSummariesEqual(previousSummary, next)) {      return previousSummary;    }    previousSummary = next;    return previousSummary;  }).pipe(Atom.withLabel("environment-shell-summary"));} export function createEnvironmentServerConfigsAtom(input: {  readonly catalogValueAtom: Atom.Atom<EnvironmentCatalogState>;  readonly serverConfigValueAtom: (environmentId: EnvironmentId) => Atom.Atom<ServerConfig | null>;}) {  let previousServerConfigs = EMPTY_SERVER_CONFIGS;  return Atom.make((get) => {    const next = new Map<EnvironmentId, ServerConfig>();    for (const environmentId of get(input.catalogValueAtom).entries.keys()) {      const config = get(input.serverConfigValueAtom(environmentId));      if (config !== null) {        next.set(environmentId, config);      }    }    if (mapsEqual(previousServerConfigs, next)) {      return previousServerConfigs;    }    previousServerConfigs = next;    return previousServerConfigs;  }).pipe(Atom.withLabel("environment-server-configs"));} export function createEnvironmentShellAtoms<R, E>(  runtime: Atom.AtomRuntime<    EnvironmentRegistry | EnvironmentCacheStore | ShellSnapshotLoader | R,    E  >,) {  const stateAtom = Atom.family((environmentId: EnvironmentId) =>    runtime.atom(shellStateChanges(environmentId), {      initialValue: EMPTY_SHELL_STATE,    }),  );   const stateValueAtom = Atom.family((environmentId: EnvironmentId) =>    Atom.make((get) =>      Option.getOrElse(AsyncResult.value(get(stateAtom(environmentId))), () => EMPTY_SHELL_STATE),    ).pipe(Atom.withLabel(`environment-shell-state-value:${environmentId}`)),  );   return {    stateAtom,    stateValueAtom,  };} export * from "./models.ts";export * from "./shellCommands.ts";export * from "./shellReducer.ts";export * from "./shellSnapshotHttp.ts";export * from "./snapshots.ts";