import type { EnvironmentId, OrchestrationShellSnapshot, OrchestrationShellStreamEvent, ServerConfig, ServerLifecycleWelcomePayload, TerminalEvent, } from "@t3tools/contracts"; import type { KnownEnvironment } from "@t3tools/client-runtime"; import type { WsRpcClient } from "~/rpc/wsRpcClient"; export interface EnvironmentConnection { readonly kind: "primary" | "saved"; readonly environmentId: EnvironmentId; readonly knownEnvironment: KnownEnvironment; readonly client: WsRpcClient; readonly ensureBootstrapped: () => Promise; readonly reconnect: () => Promise; readonly dispose: () => Promise; } interface OrchestrationHandlers { readonly applyShellEvent: ( event: OrchestrationShellStreamEvent, environmentId: EnvironmentId, ) => void; readonly syncShellSnapshot: ( snapshot: OrchestrationShellSnapshot, environmentId: EnvironmentId, ) => void; readonly applyTerminalEvent: (event: TerminalEvent, environmentId: EnvironmentId) => void; } interface EnvironmentConnectionInput extends OrchestrationHandlers { readonly kind: "primary" | "saved"; readonly knownEnvironment: KnownEnvironment; readonly client: WsRpcClient; readonly refreshMetadata?: () => Promise; readonly onConfigSnapshot?: (config: ServerConfig) => void; readonly onWelcome?: (payload: ServerLifecycleWelcomePayload) => void; } function createBootstrapGate() { let resolve: (() => void) | null = null; let reject: ((error: unknown) => void) | null = null; let promise = new Promise((nextResolve, nextReject) => { resolve = nextResolve; reject = nextReject; }); return { wait: () => promise, resolve: () => { resolve?.(); resolve = null; reject = null; }, reject: (error: unknown) => { reject?.(error); resolve = null; reject = null; }, reset: () => { promise = new Promise((nextResolve, nextReject) => { resolve = nextResolve; reject = nextReject; }); }, }; } export function createEnvironmentConnection( input: EnvironmentConnectionInput, ): EnvironmentConnection { const environmentId = input.knownEnvironment.environmentId; if (!environmentId) { throw new Error( `Known environment ${input.knownEnvironment.label} is missing its environmentId.`, ); } let disposed = false; const bootstrapGate = createBootstrapGate(); const observeEnvironmentIdentity = (nextEnvironmentId: EnvironmentId, source: string) => { if (environmentId !== nextEnvironmentId) { throw new Error( `Environment connection ${environmentId} changed identity to ${nextEnvironmentId} via ${source}.`, ); } }; const unsubLifecycle = input.client.server.subscribeLifecycle( (event: Parameters[0]>[0]) => { if (event.type !== "welcome") { return; } observeEnvironmentIdentity( event.payload.environment.environmentId, "server lifecycle welcome", ); input.onWelcome?.(event.payload); }, ); const unsubConfig = input.client.server.subscribeConfig( (event: Parameters[0]>[0]) => { if (event.type !== "snapshot") { return; } observeEnvironmentIdentity(event.config.environment.environmentId, "server config snapshot"); input.onConfigSnapshot?.(event.config); }, ); const unsubShell = input.client.orchestration.subscribeShell( (item: Parameters[0]>[0]) => { if (item.kind === "snapshot") { input.syncShellSnapshot(item.snapshot, environmentId); bootstrapGate.resolve(); return; } input.applyShellEvent(item, environmentId); }, { onResubscribe: () => { if (disposed) { return; } bootstrapGate.reset(); }, }, ); const unsubTerminalEvent = input.client.terminal.onEvent( (event: Parameters[0]>[0]) => { input.applyTerminalEvent(event, environmentId); }, ); const cleanup = () => { disposed = true; unsubShell(); unsubTerminalEvent(); unsubLifecycle(); unsubConfig(); }; return { kind: input.kind, environmentId, knownEnvironment: input.knownEnvironment, client: input.client, ensureBootstrapped: () => bootstrapGate.wait(), reconnect: async () => { bootstrapGate.reset(); try { await input.client.reconnect(); await input.refreshMetadata?.(); await bootstrapGate.wait(); } catch (error) { bootstrapGate.reject(error); throw error; } }, dispose: async () => { cleanup(); await input.client.dispose(); }, }; }