1import { EnvironmentId, type OrchestrationV2ShellSnapshot } from "@t3tools/contracts";
2import * as Option from "effect/Option";
3import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity";
4import { afterEach, beforeEach, describe, expect, it, vi } from "vite-plus/test";
5
6import {
7 AVAILABLE_CONNECTION_STATE,
8 PrimaryConnectionTarget,
9 type SupervisorConnectionState,
10} from "../connection/model.ts";
11import type { EnvironmentCatalogState } from "./connections.ts";
12import { v2ShellSnapshot } from "./orchestrationV2TestFixtures.ts";
13import { EMPTY_SHELL_STATE, type EnvironmentShellState } from "./shell.ts";
14import {
15 createSteadyShellAtoms,
16 EMPTY_STEADY_SHELLS_STATE,
17 stepSteadyShells,
18 type SteadyShellInput,
19 type SteadyShellsState,
20} from "./steadyShells.ts";
21
22const A = EnvironmentId.make("a");
23const B = EnvironmentId.make("b");
24const WINDOW = 2_000;
25
26const snapshot = (snapshotSequence: number): OrchestrationV2ShellSnapshot => ({
27 ...v2ShellSnapshot,
28 snapshotSequence,
29});
30const shell = (
31 environmentId: EnvironmentId,
32 value: OrchestrationV2ShellSnapshot | null,
33 phase: "connecting" | "disconnected" | "live",
34 generation = 0,
35 rowActions = 0,
36): SteadyShellInput => ({
37 environmentId,
38 snapshot: value,
39 serverSequence: value?.snapshotSequence ?? null,
40 status: phase === "live" ? "live" : phase === "connecting" ? "synchronizing" : "cached",
41 generation,
42 connecting: phase !== "disconnected",
43 rowActions,
44});
45const shown = (state: SteadyShellsState, environmentId: EnvironmentId) =>
46 state.shells.get(environmentId)?.shown;
47
48describe("stepSteadyShells", () => {
49 it("shows cached rows at once and holds server rows until the environment is synchronized", () => {
50 const cached = snapshot(7);
51 const withLinks = { ...cached };
52 const fromServer = snapshot(9);
53 let state = stepSteadyShells(EMPTY_STEADY_SHELLS_STATE, [shell(A, cached, "connecting")], 0);
54 expect(shown(state, A)).toBe(cached);
55
56 // The cached links fill in without new server data.
57 state = stepSteadyShells(state, [shell(A, withLinks, "connecting")], 10);
58 expect(shown(state, A)).toBe(withLinks);
59
60 state = stepSteadyShells(state, [shell(A, fromServer, "connecting")], 20);
61 expect(shown(state, A)).toBe(withLinks);
62
63 state = stepSteadyShells(state, [shell(A, fromServer, "live")], 30);
64 expect(shown(state, A)).toBe(fromServer);
65 expect(state.releaseAt).toBeNull();
66 });
67
68 it("shows a connection's server snapshot when it arrives, and the events after it when live", () => {
69 const [cached, fromServer, withEvents] = [snapshot(7), snapshot(9), snapshot(10)];
70 let state = stepSteadyShells(EMPTY_STEADY_SHELLS_STATE, [shell(A, cached, "connecting")], 0);
71 state = stepSteadyShells(state, [shell(A, fromServer, "connecting", 1)], 10);
72 expect(shown(state, A)).toBe(fromServer);
73 state = stepSteadyShells(state, [shell(A, withEvents, "connecting", 1)], 20);
74 expect(shown(state, A)).toBe(fromServer);
75 state = stepSteadyShells(state, [shell(A, withEvents, "live", 1)], 30);
76 expect(shown(state, A)).toBe(withEvents);
77 });
78
79 it("does not wait for an environment that already shows its server snapshot", () => {
80 const [cachedA, cachedB, serverA, serverB] = [
81 snapshot(1),
82 snapshot(2),
83 snapshot(3),
84 snapshot(4),
85 ];
86 let state = stepSteadyShells(
87 EMPTY_STEADY_SHELLS_STATE,
88 [shell(A, cachedA, "connecting"), shell(B, cachedB, "connecting")],
89 0,
90 );
91 state = stepSteadyShells(
92 state,
93 [shell(A, cachedA, "connecting"), shell(B, serverB, "connecting", 1)],
94 10,
95 );
96 expect(shown(state, B)).toBe(cachedB);
97 state = stepSteadyShells(
98 state,
99 [shell(A, serverA, "connecting", 2), shell(B, serverB, "connecting", 1)],
100 20,
101 );
102 expect(shown(state, A)).toBe(serverA);
103 expect(shown(state, B)).toBe(serverB);
104 // B is not live yet, and A's next rows do not wait for it.
105 const liveA = snapshot(5);
106 state = stepSteadyShells(
107 state,
108 [shell(A, liveA, "live", 2), shell(B, serverB, "connecting", 1)],
109 30,
110 );
111 expect(shown(state, A)).toBe(liveA);
112 });
113
114 it("shows rows that wait when their environment disconnects", () => {
115 const [cachedA, cachedB, serverA] = [snapshot(7), snapshot(2), snapshot(9)];
116 let state = stepSteadyShells(
117 EMPTY_STEADY_SHELLS_STATE,
118 [shell(A, cachedA, "connecting"), shell(B, cachedB, "connecting")],
119 0,
120 );
121 state = stepSteadyShells(
122 state,
123 [shell(A, serverA, "connecting", 1), shell(B, cachedB, "connecting")],
124 10,
125 );
126 expect(shown(state, A)).toBe(cachedA);
127 const disconnected = [shell(A, serverA, "disconnected", 1), shell(B, cachedB, "connecting")];
128 state = stepSteadyShells(state, disconnected, 20);
129 expect(shown(state, A)).toBe(cachedA);
130 expect(state.releaseAt).toBe(10 + WINDOW);
131 state = stepSteadyShells(state, disconnected, 10 + WINDOW);
132 expect(shown(state, A)).toBe(serverA);
133 });
134
135 it("holds replayed events after a reconnect until the environment is live again", () => {
136 const [first, replayed] = [snapshot(1), snapshot(4)];
137 let state = stepSteadyShells(EMPTY_STEADY_SHELLS_STATE, [shell(A, first, "live", 1)], 0);
138 state = stepSteadyShells(state, [shell(A, replayed, "connecting", 1)], 10);
139 expect(shown(state, A)).toBe(first);
140 state = stepSteadyShells(state, [shell(A, replayed, "live", 1)], 20);
141 expect(shown(state, A)).toBe(replayed);
142 });
143
144 it("shows the cache of an environment that was first seen before its cache was read", () => {
145 const cached = snapshot(7);
146 let state = stepSteadyShells(EMPTY_STEADY_SHELLS_STATE, [shell(A, null, "connecting")], 0);
147 state = stepSteadyShells(state, [shell(A, cached, "connecting")], 10);
148 expect(shown(state, A)).toBe(cached);
149 state = stepSteadyShells(state, [shell(A, snapshot(9), "connecting")], 20);
150 expect(shown(state, A)).toBe(cached);
151 });
152
153 it("shows the rows of environments that synchronize within the window together", () => {
154 const [cachedA, cachedB, liveA, liveB] = [snapshot(1), snapshot(2), snapshot(3), snapshot(4)];
155 let state = stepSteadyShells(
156 EMPTY_STEADY_SHELLS_STATE,
157 [shell(A, cachedA, "connecting"), shell(B, cachedB, "connecting")],
158 0,
159 );
160 state = stepSteadyShells(
161 state,
162 [shell(A, liveA, "live"), shell(B, cachedB, "connecting")],
163 500,
164 );
165 expect(shown(state, A)).toBe(cachedA);
166 expect(state.releaseAt).toBe(500 + WINDOW);
167
168 state = stepSteadyShells(state, [shell(A, liveA, "live"), shell(B, liveB, "live")], 900);
169 expect(shown(state, A)).toBe(liveA);
170 expect(shown(state, B)).toBe(liveB);
171 expect(state.releaseAt).toBeNull();
172 });
173
174 it("stops waiting for a slow environment when the window ends, then shows it on its own", () => {
175 const [cachedA, cachedB, liveA, liveB] = [snapshot(1), snapshot(2), snapshot(3), snapshot(4)];
176 let state = stepSteadyShells(
177 EMPTY_STEADY_SHELLS_STATE,
178 [shell(A, cachedA, "connecting"), shell(B, cachedB, "connecting")],
179 0,
180 );
181 const waiting = [shell(A, liveA, "live"), shell(B, cachedB, "connecting")];
182 state = stepSteadyShells(state, waiting, 500);
183 expect(stepSteadyShells(state, waiting, 500 + WINDOW - 1)).toBe(state);
184
185 state = stepSteadyShells(state, waiting, 500 + WINDOW);
186 expect(shown(state, A)).toBe(liveA);
187 expect(shown(state, B)).toBe(cachedB);
188
189 state = stepSteadyShells(state, [shell(A, liveA, "live"), shell(B, liveB, "live")], 9_000);
190 expect(shown(state, B)).toBe(liveB);
191 });
192
193 it("does not wait for an environment that already failed", () => {
194 const [cachedA, cachedB, liveA] = [snapshot(1), snapshot(2), snapshot(3)];
195 let state = stepSteadyShells(
196 EMPTY_STEADY_SHELLS_STATE,
197 [shell(A, cachedA, "connecting"), shell(B, cachedB, "disconnected")],
198 0,
199 );
200 state = stepSteadyShells(
201 state,
202 [shell(A, liveA, "live"), shell(B, cachedB, "disconnected")],
203 500,
204 );
205 expect(shown(state, A)).toBe(liveA);
206 });
207
208 it("follows a synchronized environment, and holds its rows again while it reconnects", () => {
209 const [first, second, missed, caughtUp] = [snapshot(1), snapshot(2), snapshot(5), snapshot(6)];
210 let state = stepSteadyShells(EMPTY_STEADY_SHELLS_STATE, [shell(A, first, "live")], 0);
211 state = stepSteadyShells(state, [shell(A, second, "live")], 10);
212 expect(shown(state, A)).toBe(second);
213
214 state = stepSteadyShells(state, [shell(A, second, "connecting")], 20);
215 state = stepSteadyShells(state, [shell(A, missed, "connecting")], 30);
216 expect(shown(state, A)).toBe(second);
217
218 state = stepSteadyShells(state, [shell(A, caughtUp, "live")], 40);
219 expect(shown(state, A)).toBe(caughtUp);
220 });
221
222 it("follows an environment's live rows once the user moves or removes a row there", () => {
223 const [cachedA, cachedB, liveA, confirmedA] = [
224 snapshot(1),
225 snapshot(2),
226 snapshot(3),
227 snapshot(4),
228 ];
229 const [replayedB, archivedB, laterB] = [snapshot(5), snapshot(6), snapshot(7)];
230 const settledA = { ...liveA };
231 let state = stepSteadyShells(
232 EMPTY_STEADY_SHELLS_STATE,
233 [shell(A, cachedA, "connecting"), shell(B, cachedB, "connecting")],
234 0,
235 );
236 state = stepSteadyShells(state, [shell(A, liveA, "live"), shell(B, cachedB, "connecting")], 10);
237 expect(shown(state, A)).toBe(cachedA);
238
239 // The user settles a thread of A: its rows show at once, and B's rows still wait.
240 const heldB = shell(B, replayedB, "connecting");
241 state = stepSteadyShells(state, [shell(A, settledA, "live", 0, 1), heldB], 20);
242 expect(shown(state, A)).toBe(settledA);
243 expect(shown(state, B)).toBe(cachedB);
244 expect(state.releaseAt).toBeNull();
245 state = stepSteadyShells(state, [shell(A, confirmedA, "live", 0, 1), heldB], 30);
246 expect(shown(state, A)).toBe(confirmedA);
247
248 // B still synchronizes when the user archives a thread there. Its rows follow the
249 // replayed events until it is live, then wait again on the next reconnect.
250 const a = shell(A, confirmedA, "live", 0, 1);
251 state = stepSteadyShells(state, [a, shell(B, replayedB, "connecting", 0, 1)], 40);
252 expect(shown(state, B)).toBe(replayedB);
253 state = stepSteadyShells(state, [a, shell(B, archivedB, "connecting", 0, 1)], 50);
254 expect(shown(state, B)).toBe(archivedB);
255 state = stepSteadyShells(state, [a, shell(B, archivedB, "live", 0, 1)], 60);
256 state = stepSteadyShells(state, [a, shell(B, laterB, "connecting", 0, 1)], 70);
257 expect(shown(state, B)).toBe(archivedB);
258 });
259
260 it("drops an environment that left the list of enabled environments", () => {
261 const state = stepSteadyShells(
262 stepSteadyShells(EMPTY_STEADY_SHELLS_STATE, [shell(A, snapshot(1), "live")], 0),
263 [],
264 10,
265 );
266 expect(state.shells.size).toBe(0);
267 });
268});
269
270describe("createSteadyShellAtoms", () => {
271 beforeEach(() => {
272 vi.useFakeTimers();
273 vi.setSystemTime(10_000);
274 });
275 afterEach(() => {
276 vi.clearAllTimers();
277 vi.useRealTimers();
278 });
279
280 it("shows held rows when the window ends without another change", () => {
281 const entry = (environmentId: EnvironmentId) => ({
282 target: new PrimaryConnectionTarget({
283 environmentId,
284 label: environmentId,
285 httpBaseUrl: "https://example.test",
286 wsBaseUrl: "wss://example.test",
287 }),
288 enabled: true,
289 profile: Option.none(),
290 });
291 const catalog = Atom.make<EnvironmentCatalogState>({
292 isReady: true,
293 entries: new Map([
294 [A, entry(A)],
295 [B, entry(B)],
296 ]),
297 });
298 const [cachedA, liveA] = [snapshot(1), snapshot(3)];
299 const snapshots = Atom.family((environmentId: EnvironmentId) =>
300 Atom.make<OrchestrationV2ShellSnapshot | null>(environmentId === A ? cachedA : snapshot(2)),
301 );
302 const rowActions = Atom.family((_: EnvironmentId) => Atom.make(0));
303 const shellStates = Atom.family((_: EnvironmentId) =>
304 Atom.make<EnvironmentShellState>({ ...EMPTY_SHELL_STATE, status: "synchronizing" }),
305 );
306 const connections = Atom.family((_: EnvironmentId) =>
307 Atom.make(
308 AsyncResult.success<SupervisorConnectionState>({
309 ...AVAILABLE_CONNECTION_STATE,
310 desired: true,
311 phase: "connecting",
312 attempt: 1,
313 }),
314 ),
315 );
316 const steady = createSteadyShellAtoms({
317 catalogValueAtom: catalog,
318 snapshotAtom: snapshots,
319 shellStateAtom: shellStates,
320 rowActionsAtom: rowActions,
321 connectionStateAtom: connections,
322 });
323 const registry = AtomRegistry.make();
324 const release = registry.mount(steady.snapshotAtom(A));
325
326 registry.set(snapshots(A), liveA);
327 registry.set(shellStates(A), { ...EMPTY_SHELL_STATE, status: "live", generation: 1 });
328 registry.set(
329 connections(A),
330 AsyncResult.success({ ...AVAILABLE_CONNECTION_STATE, desired: true, phase: "connected" }),
331 );
332 expect(registry.get(steady.snapshotAtom(A))).toBe(cachedA);
333
334 vi.advanceTimersByTime(WINDOW);
335 expect(registry.get(steady.snapshotAtom(A))).toBe(liveA);
336 release();
337 });
338});