1// @effect-diagnostics globalTimers:off - The release timer runs in web and native clients, outside an Effect runtime.
2import type { EnvironmentId, OrchestrationV2ShellSnapshot } from "@t3tools/contracts";
3import { AsyncResult, Atom } from "effect/reactivity";
4
5import { AVAILABLE_CONNECTION_STATE, type SupervisorConnectionState } from "../connection/model.ts";
6import { presentConnectionState } from "../connection/presentation.ts";
7import { type EnvironmentCatalogState, enabledEnvironmentIds } from "./connections.ts";
8import type { EnvironmentShellState, EnvironmentShellStatus } from "./shell.ts";
9
10/**
11 * How long fresh rows wait for environments that are still connecting. Each environment
12 * gets its server data at its own time, and a list that follows each one moves once per
13 * environment; within this time they move together.
14 */
15const STEADY_SHELL_RELEASE_WINDOW_MS = 2_000;
16
17export interface SteadyShellInput {
18 readonly environmentId: EnvironmentId;
19 readonly snapshot: OrchestrationV2ShellSnapshot | null;
20 /**
21 * The snapshot's server sequence. It moves with new server data and with nothing else, and
22 * it comes from the snapshot itself: a second atom could be one update behind the rows.
23 */
24 readonly serverSequence: number | null;
25 readonly status: EnvironmentShellStatus;
26 /** The shell state's generation: it grows with each server snapshot. */
27 readonly generation: number;
28 /** The connection is up or on its first attempt, so fresh rows can come soon. */
29 readonly connecting: boolean;
30 /** How many times the user moved or removed a row of this environment. */
31 readonly rowActions: number;
32}
33
34interface SteadyShell {
35 readonly shown: OrchestrationV2ShellSnapshot | null;
36 readonly shownSequence: number | null;
37 readonly shownGeneration: number;
38 /** The rows were shown while live, so each live change shows at once. */
39 readonly following: boolean;
40 /** The rows hold a server snapshot of the connection that is synchronizing. */
41 readonly caughtUp: boolean;
42 /** The `rowActions` count when the rows last showed while live. */
43 readonly rowActions: number;
44}
45
46export interface SteadyShellsState {
47 readonly shells: ReadonlyMap<EnvironmentId, SteadyShell>;
48 /** When held rows show at the latest; null while none wait. */
49 readonly releaseAt: number | null;
50}
51
52export const EMPTY_STEADY_SHELLS_STATE: SteadyShellsState = { shells: new Map(), releaseAt: null };
53
54/**
55 * Decides which snapshot of each environment a thread list shows. New server rows wait for
56 * a complete state (a server snapshot, live, or disconnected), then show with every other
57 * environment that has one: at once when no other owes its rows, else when the window ends.
58 * Returns `previous` itself when nothing changes.
59 */
60export function stepSteadyShells(
61 previous: SteadyShellsState,
62 inputs: ReadonlyArray<SteadyShellInput>,
63 now: number,
64 windowMs = STEADY_SHELL_RELEASE_WINDOW_MS,
65): SteadyShellsState {
66 const shells = new Map<EnvironmentId, SteadyShell>();
67 const waiting: Array<SteadyShellInput> = [];
68 let othersOweRows = false;
69 const show = (
70 input: SteadyShellInput,
71 caughtUp: boolean,
72 current: SteadyShell | undefined,
73 ): SteadyShell => ({
74 shown: input.snapshot,
75 shownSequence: input.serverSequence,
76 shownGeneration: input.generation,
77 following: input.status === "live",
78 caughtUp,
79 rowActions:
80 input.status === "live" || current === undefined ? input.rowActions : current.rowActions,
81 });
82 const keep = (input: SteadyShellInput, next: SteadyShell, current: SteadyShell | undefined) => {
83 othersOweRows ||= input.status === "synchronizing" && input.connecting && !next.caughtUp;
84 const same =
85 current !== undefined &&
86 current.shown === next.shown &&
87 current.shownSequence === next.shownSequence &&
88 current.shownGeneration === next.shownGeneration &&
89 current.following === next.following &&
90 current.caughtUp === next.caughtUp &&
91 current.rowActions === next.rowActions;
92 shells.set(input.environmentId, same ? current : next);
93 };
94 for (const input of inputs) {
95 const current = previous.shells.get(input.environmentId);
96 const synchronizing = input.status === "synchronizing";
97 if (current === undefined) {
98 keep(input, show(input, false, current), current);
99 continue;
100 }
101 const newSnapshot = input.generation > current.shownGeneration;
102 // Without new server data the change is local (cached links, rows loaded on demand), and
103 // an environment that shows nothing (its cache is not read yet) has nothing to hold back.
104 const local =
105 current.shown === null || (input.serverSequence === current.shownSequence && !newSnapshot);
106 // The hold stops changes the user did not cause. After the user moves or removes a row,
107 // the environment's rows follow the live data until it is live.
108 const userActed = input.rowActions !== current.rowActions;
109 if (local || userActed || (input.status === "live" && current.following)) {
110 const caughtUp = synchronizing && (current.caughtUp || newSnapshot);
111 keep(input, show(input, caughtUp, current), current);
112 } else if (!synchronizing || newSnapshot) {
113 waiting.push(input);
114 shells.set(input.environmentId, current);
115 } else {
116 // Events that a synchronizing environment replays are complete only when it is live.
117 keep(input, { ...current, following: false }, current);
118 }
119 }
120
121 let releaseAt: number | null = null;
122 if (waiting.length > 0) {
123 releaseAt = previous.releaseAt ?? now + windowMs;
124 if (!othersOweRows || now >= releaseAt) {
125 releaseAt = null;
126 for (const input of waiting) {
127 const current = shells.get(input.environmentId);
128 shells.set(input.environmentId, show(input, input.status === "synchronizing", current));
129 }
130 }
131 }
132
133 const unchanged =
134 releaseAt === previous.releaseAt &&
135 shells.size === previous.shells.size &&
136 [...shells].every(([environmentId, shell]) => previous.shells.get(environmentId) === shell);
137 return unchanged ? previous : { shells, releaseAt };
138}
139
140/**
141 * Snapshots for lists that arrange threads across environments; see `stepSteadyShells`.
142 * Point reads (a thread's own shell, route checks) keep the live snapshot, where a row
143 * that the server just sent must not look missing.
144 */
145export function createSteadyShellAtoms<E>(input: {
146 readonly catalogValueAtom: Atom.Atom<EnvironmentCatalogState>;
147 readonly snapshotAtom: (
148 environmentId: EnvironmentId,
149 ) => Atom.Atom<OrchestrationV2ShellSnapshot | null>;
150 readonly shellStateAtom: (environmentId: EnvironmentId) => Atom.Atom<EnvironmentShellState>;
151 readonly rowActionsAtom: (environmentId: EnvironmentId) => Atom.Atom<number>;
152 readonly connectionStateAtom: (
153 environmentId: EnvironmentId,
154 ) => Atom.Atom<AsyncResult.AsyncResult<SupervisorConnectionState, E>>;
155}) {
156 const now = () => {
157 // @effect-diagnostics-next-line globalDate:off - The release window is wall-clock time.
158 return Date.now();
159 };
160 // Outlives the atom's rebuilds, and its unmount while no list is on screen.
161 let state = EMPTY_STEADY_SHELLS_STATE;
162 const stateAtom = Atom.make((get) => {
163 const inputs: Array<SteadyShellInput> = [];
164 for (const environmentId of enabledEnvironmentIds(get(input.catalogValueAtom))) {
165 const snapshot = get(input.snapshotAtom(environmentId));
166 const shell = get(input.shellStateAtom(environmentId));
167 const connection = AsyncResult.getOrElse(
168 get(input.connectionStateAtom(environmentId)),
169 () => AVAILABLE_CONNECTION_STATE,
170 );
171 // An enabled environment starts out connecting, before its supervisor reports.
172 const phase =
173 connection === AVAILABLE_CONNECTION_STATE
174 ? "connecting"
175 : presentConnectionState(connection).phase;
176 inputs.push({
177 environmentId,
178 snapshot,
179 serverSequence: snapshot?.snapshotSequence ?? null,
180 status: shell.status,
181 generation: shell.generation,
182 // A retry after a failure can take minutes, so it is not waited for until it connects.
183 connecting: phase === "connecting" || phase === "connected",
184 rowActions: get(input.rowActionsAtom(environmentId)),
185 });
186 }
187 state = stepSteadyShells(state, inputs, now());
188 if (state.releaseAt !== null) {
189 const timer = setTimeout(() => get.refreshSelf(), Math.max(0, state.releaseAt - now()));
190 get.addFinalizer(() => clearTimeout(timer));
191 }
192 return state;
193 }).pipe(Atom.withLabel("steady-shells"));
194
195 const snapshotAtom = Atom.family((environmentId: EnvironmentId) =>
196 Atom.make(
197 (get): OrchestrationV2ShellSnapshot | null =>
198 get(stateAtom).shells.get(environmentId)?.shown ?? null,
199 ).pipe(Atom.withLabel(`steady-shell-snapshot:${environmentId}`)),
200 );
201
202 // Reads the live snapshots too, so it is current when a handler reads it once.
203 const holdingAtom = Atom.make((get) => {
204 const { shells } = get(stateAtom);
205 for (const environmentId of enabledEnvironmentIds(get(input.catalogValueAtom))) {
206 if (shells.get(environmentId)?.shown !== get(input.snapshotAtom(environmentId))) return true;
207 }
208 return false;
209 }).pipe(Atom.withLabel("steady-shells-holding"));
210
211 return {
212 /** The snapshot a thread list shows for an environment. */
213 snapshotAtom,
214 /** True while a list shows rows older than the server data the client holds. */
215 holdingAtom,
216 };
217}