1import {
2 ORCHESTRATION_V2_WS_METHODS,
3 type EnvironmentId,
4 type OrchestrationV2ShellSnapshot,
5 type OrchestrationV2ShellStreamItem,
6 type OrchestrationV2ThreadShell,
7 type ServerConfig,
8 type ThreadId,
9 type ThreadPullRequestLink,
10} from "@t3tools/contracts";
11import * as Cause from "effect/Cause";
12import * as Effect from "effect/Effect";
13import * as Fiber from "effect/Fiber";
14import * as Option from "effect/Option";
15import * as Queue from "effect/Queue";
16import * as Ref from "effect/Ref";
17import * as Semaphore from "effect/Semaphore";
18import * as Stream from "effect/Stream";
19import * as SubscriptionRef from "effect/SubscriptionRef";
20import { AsyncResult, Atom } from "effect/reactivity";
21
22import * as EnvironmentRegistry from "../connection/registry.ts";
23import { connectionProjectionPhase } from "../connection/model.ts";
24import * as EnvironmentSupervisor from "../connection/supervisor.ts";
25import * as ConnectionWakeups from "../connection/wakeups.ts";
26import * as Hub from "../connection/hub.ts";
27import { applyHubInboxEnvironment } from "./hub.ts";
28import { safeErrorLogAttributes } from "../errors/safeLog.ts";
29import * as Persistence from "../platform/persistence.ts";
30import { runCachePersistence } from "./cachePersistence.ts";
31import { subscribeDynamic } from "../rpc/client.ts";
32import type { RpcSession } from "../rpc/session.ts";
33import * as ShellSnapshotLoader from "./shellSnapshotHttp.ts";
34import type { DeferredShellSnapshot } from "./shellPullRequests.ts";
35import {
36 applyShellStreamEvent,
37 mergeShellSnapshotProjects,
38 reuseUnchangedThreadShells,
39 sameThreadPullRequests,
40} from "./shellReducer.ts";
41import { type EnvironmentCatalogState, enabledEnvironmentIds } from "./connections.ts";
42import { followStreamInEnvironment } from "./runtime.ts";
43
44export type EnvironmentShellStatus = "empty" | "cached" | "synchronizing" | "live";
45
46export interface EnvironmentShellState {
47 readonly snapshot: Option.Option<OrchestrationV2ShellSnapshot>;
48 readonly status: EnvironmentShellStatus;
49 readonly error: Option.Option<string>;
50}
51
52const EMPTY_SHELL_STATE: EnvironmentShellState = {
53 snapshot: Option.none(),
54 status: "empty",
55 error: Option.none(),
56};
57
58function shellStatusForSnapshot(
59 snapshot: Option.Option<OrchestrationV2ShellSnapshot>,
60): EnvironmentShellStatus {
61 return Option.isSome(snapshot) ? "cached" : "empty";
62}
63
64function withoutDeferredPullRequests(value: DeferredShellSnapshot): OrchestrationV2ShellSnapshot {
65 if (value.loadPullRequests === undefined) return value;
66 const { loadPullRequests: _deferred, ...snapshot } = value;
67 return snapshot;
68}
69
70const SHELL_SYNCHRONIZATION_ERROR_MESSAGE = "Could not synchronize environment data.";
71
72export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")(function* () {
73 const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor;
74 const cache = yield* Persistence.EnvironmentCacheStore;
75 const snapshotLoader = yield* ShellSnapshotLoader.ShellSnapshotLoader;
76 const wakeups = yield* Effect.serviceOption(ConnectionWakeups.ConnectionWakeups);
77 const environmentId = supervisor.target.environmentId;
78 const cached = yield* cache.loadShell(environmentId).pipe(
79 Effect.catch((error) =>
80 Effect.logWarning("Could not load cached environment shell.").pipe(
81 Effect.annotateLogs({
82 environmentId,
83 ...safeErrorLogAttributes(error),
84 }),
85 Effect.as(Option.none<DeferredShellSnapshot>()),
86 ),
87 ),
88 );
89 const cachedSnapshot = Option.map(cached, withoutDeferredPullRequests);
90 const state = yield* SubscriptionRef.make<EnvironmentShellState>({
91 snapshot: cachedSnapshot,
92 status: shellStatusForSnapshot(cachedSnapshot),
93 error: Option.none(),
94 });
95 const awaitingCompletion = yield* Ref.make(false);
96 const lastAuthoritativeSession = yield* Ref.make<RpcSession | null>(null);
97 const activeSubscriptionSession = yield* Ref.make<RpcSession | null>(null);
98 const latestLiveSnapshot = yield* Ref.make<Option.Option<OrchestrationV2ShellSnapshot>>(
99 Option.none(),
100 );
101 const persistence = yield* Queue.sliding<OrchestrationV2ShellSnapshot>(1);
102
103 const persistenceLock = yield* Semaphore.make(1);
104 let lastPersisted: OrchestrationV2ShellSnapshot | undefined;
105 // Rows still waiting for their deferred pull request links, which must not be saved:
106 // the cache would keep them without links until the next server refresh.
107 let pendingLinkRows: ReadonlySet<OrchestrationV2ThreadShell> = new Set();
108 const persistLatest = Effect.fn("EnvironmentShellState.persistLatest")(function* (
109 flush: boolean,
110 ) {
111 while (true) {
112 const latest = yield* Ref.get(latestLiveSnapshot);
113 if (Option.isNone(latest) || latest.value === lastPersisted) return;
114 if (latest.value.threads.some((thread) => pendingLinkRows.has(thread))) return;
115 const snapshot = latest.value;
116 const saved = yield* cache.saveShell(environmentId, snapshot).pipe(
117 Effect.as(true),
118 Effect.catch((error) =>
119 Effect.logWarning("Could not persist environment shell cache.").pipe(
120 Effect.annotateLogs({ environmentId, ...safeErrorLogAttributes(error) }),
121 Effect.as(false),
122 ),
123 ),
124 );
125 if (!saved) return;
126 lastPersisted = snapshot;
127 // Only lifecycle flushes chase updates that arrived during an in-flight save.
128 // The regular worker leaves those updates for the next write window.
129 if (!flush) return;
130 }
131 }, persistenceLock.withPermit);
132 const persist = () => persistLatest(false);
133 const flushLiveShellSnapshot = persistLatest(true);
134
135 // Register before scoped worker fibers so reverse finalizer order interrupts
136 // those fibers first and this flush sees a stable latestLiveSnapshot.
137 yield* Effect.addFinalizer(() => flushLiveShellSnapshot);
138
139 yield* runCachePersistence(persistence, persist).pipe(Effect.forkScoped);
140
141 const setDisconnected = Ref.set(awaitingCompletion, false).pipe(
142 Effect.andThen(
143 SubscriptionRef.update(state, (current) => ({
144 ...current,
145 status: shellStatusForSnapshot(current.snapshot),
146 })),
147 ),
148 Effect.andThen(flushLiveShellSnapshot.pipe(Effect.forkScoped)),
149 Effect.asVoid,
150 );
151 const setSynchronizing = SubscriptionRef.update(state, (current) => ({
152 ...current,
153 status: "synchronizing" as const,
154 error: Option.none(),
155 }));
156 const setReady = SubscriptionRef.update(state, (current) =>
157 current.status === "live"
158 ? current
159 : {
160 ...current,
161 status: "synchronizing" as const,
162 error: Option.none(),
163 },
164 );
165 const setStreamError = (error: unknown) =>
166 Ref.set(awaitingCompletion, false).pipe(
167 Effect.andThen(Effect.logWarning("Could not synchronize the environment shell.")),
168 Effect.annotateLogs({
169 environmentId,
170 ...safeErrorLogAttributes(error),
171 }),
172 Effect.andThen(
173 SubscriptionRef.update(state, (current) => ({
174 ...current,
175 status: shellStatusForSnapshot(current.snapshot),
176 error: Option.some(SHELL_SYNCHRONIZATION_ERROR_MESSAGE),
177 })),
178 ),
179 );
180
181 // Server items and deferred link fills both read the snapshot and write it back; one at a
182 // time, so neither can overwrite what the other wrote in between.
183 const snapshotWriteLock = yield* Semaphore.make(1);
184
185 // Apply each received batch with one state write. The RPC client's bounded
186 // buffer can split a server chunk, so a bulk action can still need several
187 // writes, but each write includes every event in that batch.
188 const applyItems = Effect.fn("EnvironmentShellState.applyItems")(function* (
189 items: ReadonlyArray<OrchestrationV2ShellStreamItem>,
190 // Ids of rows whose links will fill later; they are held out of saves before one queues.
191 pendingLinkIds?: ReadonlySet<ThreadId>,
192 ) {
193 const initial = yield* SubscriptionRef.get(state);
194 let waiting = yield* Ref.get(awaitingCompletion);
195 let next = initial;
196 let receivedSnapshot = false;
197 for (const item of items) {
198 if (item.kind === "synchronized") {
199 waiting = false;
200 if (Option.isSome(next.snapshot)) {
201 next = { ...next, status: "live", error: Option.none() };
202 }
203 continue;
204 }
205 const nextSnapshot =
206 item.kind === "snapshot"
207 ? mergeShellSnapshotProjects(
208 Option.getOrNull(next.snapshot),
209 item.snapshot,
210 item.resolvedRepositoryIdentityRoots === undefined
211 ? undefined
212 : {
213 resolvedRepositoryIdentityRoots: item.resolvedRepositoryIdentityRoots,
214 },
215 )
216 : Option.match(next.snapshot, {
217 onNone: () => null,
218 onSome: (snapshot) =>
219 item.sequence > snapshot.snapshotSequence
220 ? applyShellStreamEvent(snapshot, item)
221 : snapshot,
222 });
223 if (nextSnapshot === null) continue;
224 receivedSnapshot ||= item.kind === "snapshot";
225 next = {
226 snapshot: Option.some(nextSnapshot),
227 status: waiting ? "synchronizing" : "live",
228 error: Option.none(),
229 };
230 }
231 yield* Ref.set(awaitingCompletion, waiting);
232 if (next === initial) return;
233 if (pendingLinkIds !== undefined && Option.isSome(next.snapshot)) {
234 pendingLinkRows = new Set(
235 next.snapshot.value.threads.filter((thread) => pendingLinkIds.has(thread.id)),
236 );
237 }
238 if (Option.isSome(next.snapshot)) {
239 yield* Ref.set(latestLiveSnapshot, next.snapshot);
240 }
241 yield* SubscriptionRef.set(state, next);
242 if (receivedSnapshot) {
243 const session = yield* Ref.get(activeSubscriptionSession);
244 if (session !== null) {
245 yield* Ref.set(lastAuthoritativeSession, session);
246 }
247 }
248 if (next.snapshot !== initial.snapshot && Option.isSome(next.snapshot)) {
249 yield* Queue.offer(persistence, next.snapshot.value);
250 }
251 }, snapshotWriteLock.withPermit);
252
253 // Rows can arrive before their pull request links (see DeferredShellSnapshot), so the list
254 // paints first. The links then fill into the rows that still hold the object they arrived
255 // with; a row the server has replaced since carries its own links.
256 // Captured so fills started from the subscription callback still end with this state.
257 const stateScope = yield* Effect.scope;
258 // Each full snapshot supersedes the previous one's fill; links from an older snapshot must
259 // not land on rows a newer snapshot reused. A cancelled fill's rows stay pending until the
260 // next snapshot replaces the pending set.
261 let activeFill: Fiber.Fiber<void> | undefined;
262 const cancelActiveFill = Effect.suspend(() => {
263 const fill = activeFill;
264 activeFill = undefined;
265 return fill === undefined ? Effect.void : Fiber.interrupt(fill);
266 });
267 const fillDeferredPullRequests = Effect.fnUntraced(function* (
268 source: DeferredShellSnapshot,
269 applied: OrchestrationV2ShellSnapshot,
270 ) {
271 yield* cancelActiveFill;
272 if (source.loadPullRequests === undefined) return;
273 const arrivedRows = new Set(applied.threads);
274 const arrivedIds = new Set(source.threads.map((thread) => thread.id));
275 pendingLinkRows = new Set(applied.threads.filter((thread) => arrivedIds.has(thread.id)));
276 activeFill = yield* source.loadPullRequests.pipe(
277 Effect.flatMap((linksByThreadId) =>
278 writeFilledLinks(linksByThreadId, arrivedRows, arrivedIds),
279 ),
280 Effect.forkIn(stateScope),
281 );
282 });
283 const writeFilledLinks = Effect.fnUntraced(
284 function* (
285 linksByThreadId: ReadonlyMap<ThreadId, ReadonlyArray<ThreadPullRequestLink>>,
286 arrivedRows: ReadonlySet<OrchestrationV2ThreadShell>,
287 arrivedIds: ReadonlySet<ThreadId>,
288 ) {
289 const filled = yield* SubscriptionRef.modify(
290 state,
291 (current): [Option.Option<OrchestrationV2ShellSnapshot>, EnvironmentShellState] => {
292 if (Option.isNone(current.snapshot)) return [Option.none(), current];
293 const snapshot = current.snapshot.value;
294 let changed = false;
295 const threads = snapshot.threads.map((thread) => {
296 if (!arrivedRows.has(thread) || !arrivedIds.has(thread.id)) return thread;
297 const links = linksByThreadId.get(thread.id);
298 // A reused row may already hold these links from an earlier fill.
299 if (sameThreadPullRequests(thread.pullRequests, links)) return thread;
300 changed = true;
301 if (links === undefined) {
302 const { pullRequests: _previous, ...row } = thread;
303 return row;
304 }
305 return { ...thread, pullRequests: links };
306 });
307 if (!changed) return [Option.none(), current];
308 const next = { ...snapshot, threads };
309 return [Option.some(next), { ...current, snapshot: Option.some(next) }];
310 },
311 );
312 // Saves skipped while links were pending are queued again. Cached rows are never live:
313 // until a server snapshot arrives there is nothing new to save.
314 pendingLinkRows = new Set();
315 const live = yield* Ref.get(latestLiveSnapshot);
316 if (Option.isNone(live)) return;
317 const latest = Option.isSome(filled) ? filled : live;
318 yield* Ref.set(latestLiveSnapshot, latest);
319 yield* Queue.offer(persistence, latest.value);
320 },
321 (effect) => snapshotWriteLock.withPermit(effect),
322 );
323 if (Option.isSome(cached) && Option.isSome(cachedSnapshot)) {
324 yield* fillDeferredPullRequests(cached.value, cachedSnapshot.value);
325 }
326
327 const foregroundResubscriptions = Option.match(wakeups, {
328 onNone: () => Stream.never,
329 onSome: (service) =>
330 service.changes.pipe(Stream.filter(ConnectionWakeups.shouldResubscribeAfterWakeup)),
331 });
332
333 yield* setSynchronizing;
334 yield* Effect.forkScoped(
335 subscribeDynamic(
336 ORCHESTRATION_V2_WS_METHODS.subscribeShell,
337 Effect.fn("EnvironmentShellState.makeSubscribeInput")(function* (session) {
338 yield* Ref.set(activeSubscriptionSession, session);
339 const supportsCompletionMarker = yield* session.initialConfig.pipe(
340 Effect.map((config) => config.shellResumeCompletionMarker === true),
341 Effect.orElseSucceed(() => false),
342 );
343 yield* Ref.set(awaitingCompletion, supportsCompletionMarker);
344 yield* setSynchronizing;
345
346 // Foreground resubscriptions on the same live session can resume from
347 // the in-memory cursor. A new session reloads the authoritative HTTP
348 // snapshot so a valid cursor cannot preserve incomplete cached data.
349 const hasAuthoritativeSnapshot = (yield* Ref.get(lastAuthoritativeSession)) === session;
350 let canResume = hasAuthoritativeSnapshot;
351 let current = yield* SubscriptionRef.get(state);
352 if (!hasAuthoritativeSnapshot || Option.isNone(current.snapshot)) {
353 const prepared = yield* SubscriptionRef.get(supervisor.prepared).pipe(
354 Effect.flatMap(
355 Option.match({
356 onSome: Effect.succeed,
357 onNone: () =>
358 SubscriptionRef.changes(supervisor.prepared).pipe(
359 Stream.filter(Option.isSome),
360 Stream.map((value) => value.value),
361 Stream.runHead,
362 Effect.map(Option.getOrThrow),
363 ),
364 }),
365 ),
366 );
367 const httpSnapshot = yield* snapshotLoader.load(prepared);
368 if (Option.isSome(httpSnapshot)) {
369 yield* cancelActiveFill;
370 const previous = yield* SubscriptionRef.get(state);
371 const snapshot = reuseUnchangedThreadShells(
372 Option.getOrNull(previous.snapshot),
373 withoutDeferredPullRequests(httpSnapshot.value),
374 );
375 const pendingLinkIds = new Set(
376 httpSnapshot.value.loadPullRequests === undefined
377 ? []
378 : httpSnapshot.value.threads.map((thread) => thread.id),
379 );
380 yield* applyItems([{ kind: "snapshot", snapshot }], pendingLinkIds);
381 canResume = true;
382 current = yield* SubscriptionRef.get(state);
383 if (Option.isSome(current.snapshot)) {
384 yield* fillDeferredPullRequests(httpSnapshot.value, current.snapshot.value);
385 }
386 }
387 }
388
389 // If the authoritative refresh failed, omit the cached cursor so the
390 // socket fallback sends a complete snapshot for this new session.
391 if (!canResume || Option.isNone(current.snapshot)) {
392 return supportsCompletionMarker ? { requestCompletionMarker: true as const } : {};
393 }
394 if (!supportsCompletionMarker) {
395 // Without a completion marker there is no synchronized signal for a
396 // resumed subscription, so report live immediately, like threads.
397 yield* SubscriptionRef.update(state, (value) => ({
398 ...value,
399 status: "live" as const,
400 error: Option.none(),
401 }));
402 }
403 return {
404 afterSequence: current.snapshot.value.snapshotSequence,
405 ...(supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}),
406 };
407 }),
408 {
409 onExpectedFailure: (cause) => setStreamError(Cause.squash(cause)),
410 retryExpectedFailureAfter: "250 millis",
411 resubscribe: foregroundResubscriptions,
412 },
413 ).pipe(Stream.runForEachArray((items) => applyItems(items))),
414 );
415 yield* SubscriptionRef.changes(supervisor.state).pipe(
416 Stream.runForEach((connectionState) => {
417 switch (connectionProjectionPhase(connectionState)) {
418 case "synchronizing":
419 return setSynchronizing;
420 case "disconnected":
421 return setDisconnected;
422 case "ready":
423 return setReady;
424 }
425 }),
426 Effect.forkScoped,
427 );
428
429 return state;
430});
431
432function shellStateChanges(environmentId: EnvironmentId) {
433 return Stream.unwrap(
434 Effect.gen(function* () {
435 const source = followStreamInEnvironment(
436 environmentId,
437 Stream.unwrap(makeEnvironmentShellState().pipe(Effect.map(SubscriptionRef.changes))),
438 );
439 const hub = yield* Effect.serviceOption(Hub.HubConnection);
440 if (Option.isNone(hub)) return source;
441 return Stream.zipLatest(source, SubscriptionRef.changes(hub.value.state)).pipe(
442 Stream.map(([state, central]) =>
443 applyHubInboxEnvironment(
444 state,
445 central.inbox.environments.find((entry) => entry.environmentId === environmentId),
446 ),
447 ),
448 );
449 }),
450 );
451}
452
453export interface EnvironmentShellSummary {
454 readonly hasSnapshot: boolean;
455 readonly hasSynchronizingShell: boolean;
456 readonly hasCachedShell: boolean;
457 readonly hasLiveShell: boolean;
458 readonly firstError: string | null;
459}
460
461const EMPTY_ENVIRONMENT_SHELL_SUMMARY: EnvironmentShellSummary = Object.freeze({
462 hasSnapshot: false,
463 hasSynchronizingShell: false,
464 hasCachedShell: false,
465 hasLiveShell: false,
466 firstError: null,
467});
468
469const EMPTY_SERVER_CONFIGS: ReadonlyMap<EnvironmentId, ServerConfig> = new Map();
470
471function shellSummariesEqual(
472 left: EnvironmentShellSummary,
473 right: EnvironmentShellSummary,
474): boolean {
475 return (
476 left.hasSnapshot === right.hasSnapshot &&
477 left.hasSynchronizingShell === right.hasSynchronizingShell &&
478 left.hasCachedShell === right.hasCachedShell &&
479 left.hasLiveShell === right.hasLiveShell &&
480 left.firstError === right.firstError
481 );
482}
483
484function mapsEqual<K, V>(left: ReadonlyMap<K, V>, right: ReadonlyMap<K, V>): boolean {
485 if (left.size !== right.size) {
486 return false;
487 }
488 for (const [key, value] of left) {
489 if (right.get(key) !== value) {
490 return false;
491 }
492 }
493 return true;
494}
495
496export function createEnvironmentShellSummaryAtom(input: {
497 readonly catalogValueAtom: Atom.Atom<EnvironmentCatalogState>;
498 readonly shellStateValueAtom: (environmentId: EnvironmentId) => Atom.Atom<EnvironmentShellState>;
499}) {
500 let previousSummary = EMPTY_ENVIRONMENT_SHELL_SUMMARY;
501 return Atom.make((get) => {
502 let hasSnapshot = false;
503 let hasSynchronizingShell = false;
504 let hasCachedShell = false;
505 let hasLiveShell = false;
506 let firstError: string | null = null;
507
508 for (const environmentId of enabledEnvironmentIds(get(input.catalogValueAtom))) {
509 const state = get(input.shellStateValueAtom(environmentId));
510 hasSynchronizingShell ||= state.status === "synchronizing";
511 hasCachedShell ||= state.status === "cached";
512 hasLiveShell ||= state.status === "live";
513 if (firstError === null) {
514 firstError = Option.getOrNull(state.error);
515 }
516 if (Option.isNone(state.snapshot)) {
517 continue;
518 }
519 hasSnapshot = true;
520 }
521
522 const next: EnvironmentShellSummary = {
523 hasSnapshot,
524 hasSynchronizingShell,
525 hasCachedShell,
526 hasLiveShell,
527 firstError,
528 };
529 if (shellSummariesEqual(previousSummary, next)) {
530 return previousSummary;
531 }
532 previousSummary = next;
533 return previousSummary;
534 }).pipe(Atom.withLabel("environment-shell-summary"));
535}
536
537export function createEnvironmentServerConfigsAtom(input: {
538 readonly catalogValueAtom: Atom.Atom<EnvironmentCatalogState>;
539 readonly serverConfigValueAtom: (environmentId: EnvironmentId) => Atom.Atom<ServerConfig | null>;
540}) {
541 let previousServerConfigs = EMPTY_SERVER_CONFIGS;
542 return Atom.make((get) => {
543 const next = new Map<EnvironmentId, ServerConfig>();
544 for (const environmentId of enabledEnvironmentIds(get(input.catalogValueAtom))) {
545 const config = get(input.serverConfigValueAtom(environmentId));
546 if (config !== null) {
547 next.set(environmentId, config);
548 }
549 }
550 if (mapsEqual(previousServerConfigs, next)) {
551 return previousServerConfigs;
552 }
553 previousServerConfigs = next;
554 return previousServerConfigs;
555 }).pipe(Atom.withLabel("environment-server-configs"));
556}
557
558export function createEnvironmentShellAtoms<R, E>(
559 runtime: Atom.AtomRuntime<
560 | EnvironmentRegistry.EnvironmentRegistry
561 | Persistence.EnvironmentCacheStore
562 | ShellSnapshotLoader.ShellSnapshotLoader
563 | R,
564 E
565 >,
566) {
567 const stateAtom = Atom.family((environmentId: EnvironmentId) =>
568 runtime.atom(shellStateChanges(environmentId), {
569 initialValue: EMPTY_SHELL_STATE,
570 }),
571 );
572
573 const stateValueAtom = Atom.family((environmentId: EnvironmentId) =>
574 Atom.make((get) =>
575 Option.getOrElse(AsyncResult.value(get(stateAtom(environmentId))), () => EMPTY_SHELL_STATE),
576 ).pipe(Atom.withLabel(`environment-shell-state-value:${environmentId}`)),
577 );
578
579 return {
580 stateAtom,
581 stateValueAtom,
582 };
583}
584
585export * from "./models.ts";
586export * from "./shellCommands.ts";
587export * from "./shellReducer.ts";
588export * from "./shellPullRequests.ts";
589export * as ShellSnapshotLoader from "./shellSnapshotHttp.ts";
590export * from "./snapshots.ts";