1import {
2 isListedThreadLineage,
3 isOlderSettledThreadShell,
4 ORCHESTRATION_V2_WS_METHODS,
5 type EnvironmentId,
6 type OrchestrationV2OlderSettledCursor,
7 type OrchestrationV2OlderSettledThreadsPage,
8 type OrchestrationV2SearchThreadsInput,
9 type OrchestrationV2ThreadFamily,
10 type OrchestrationV2ThreadShell,
11 type ProjectId,
12 type ThreadId,
13} from "@t3tools/contracts";
14import * as DateTime from "effect/DateTime";
15import * as Effect from "effect/Effect";
16import * as Option from "effect/Option";
17import { AsyncResult, Atom, type AtomRegistry } from "effect/reactivity";
18
19import type * as EnvironmentRegistry from "../connection/registry.ts";
20import { request } from "../rpc/client.ts";
21import { createEnvironmentQueryAtomFamily, selectEnvironmentQueryData } from "./runtime.ts";
22import type { EnvironmentShellState } from "./shell.ts";
23import { threadFamilyRead } from "./threadFamilies.ts";
24
25// Rows per page of older settled threads: one "Show more" step of the Settled shelf.
26const OLDER_SETTLED_PAGE_SIZE = 25;
27// A burst of changes (a PR sync touching many old threads) costs one count read.
28const COUNT_REREAD_DELAY = "500 millis";
29
30/** Where an environment's shell cut settled threads, and how many are older per project. */
31export interface OlderSettled {
32 readonly before: string;
33 readonly beforeMs: number;
34 readonly counts: ReadonlyMap<ProjectId, number>;
35}
36
37/** The scope's projects in one environment, or null for all of them. */
38export interface OlderSettledScope {
39 readonly environmentId: EnvironmentId;
40 readonly projectIds: ReadonlyArray<ProjectId> | null;
41}
42
43/**
44 * The pages read below an environment's cut. Everything at or above `frontier` is local: pages
45 * hold the rows that were there when read, and shell deltas keep that range current. A null
46 * frontier means no more can load: every older thread is loaded, or this session has no snapshot.
47 */
48export interface OlderSettledPages {
49 readonly reads: ReadonlyArray<OrchestrationV2ThreadFamily>;
50 readonly frontier: OrchestrationV2OlderSettledCursor | null;
51 readonly status: "ready" | "pending" | "failed";
52}
53
54const NO_PAGES: OlderSettledPages = { reads: [], frontier: null, status: "ready" };
55
56type OlderSettledRow = Pick<
57 OrchestrationV2ThreadShell,
58 "id" | "projectId" | "lineage" | "settledOverride" | "settledAt"
59>;
60
61export interface OlderSettledShelfEnvironment {
62 readonly older: OlderSettled | null;
63 readonly pages: OlderSettledPages;
64}
65
66/**
67 * Scopes in one order, so a shelf keyed on them keeps its pages when the caller's project order
68 * changes. Loading a page can reorder the sidebar's projects, which would otherwise drop the page.
69 */
70export function sortOlderSettledScopes(
71 scopes: ReadonlyArray<OlderSettledScope>,
72): ReadonlyArray<OlderSettledScope> {
73 return scopes
74 .map((scope) => ({
75 environmentId: scope.environmentId,
76 projectIds: scope.projectIds === null ? null : [...scope.projectIds].sort(),
77 }))
78 .sort((left, right) => left.environmentId.localeCompare(right.environmentId));
79}
80
81/**
82 * Changes whenever the older count can: any removal (by largest sequence, so a repeat counts), an
83 * older row that arrives or changes, or an old row arriving outside the set (unsettled). Threads
84 * created after the cut never enter it, as threads settle after creation. Archived-thread events
85 * arrive as removals too, each costing one count read.
86 */
87export function olderSettledCountKey(
88 threads: ReadonlyArray<OrchestrationV2ThreadShell>,
89 beforeMs: number,
90 removedAt: ReadonlyMap<ThreadId, number>,
91): string {
92 let lastRemoval = 0;
93 for (const sequence of removedAt.values()) lastRemoval = Math.max(lastRemoval, sequence);
94 const rows: string[] = [];
95 for (const thread of threads) {
96 if (!isListedThreadLineage(thread.lineage)) continue;
97 const older = isOlderSettledThreadShell(thread, beforeMs);
98 if (older || DateTime.toEpochMillis(thread.createdAt) < beforeMs) {
99 rows.push(`${thread.id}:${older ? 1 : 0}`);
100 }
101 }
102 return `${lastRemoval}|${rows.sort().join(",")}`;
103}
104
105/** Whether a row sits at or above a page cursor in the server's order (settle time desc, id asc). */
106function atOrAboveCursor(
107 thread: Pick<OrchestrationV2ThreadShell, "id" | "settledAt">,
108 cursor: OrchestrationV2OlderSettledCursor,
109): boolean {
110 if (thread.settledAt === null) return true;
111 const settledAt = DateTime.formatIso(thread.settledAt);
112 return !(
113 settledAt < cursor.settledAt ||
114 (cursor.threadId !== undefined && settledAt === cursor.settledAt && thread.id > cursor.threadId)
115 );
116}
117
118/**
119 * Where a shelf over these environments stops listing: the highest frontier of any environment
120 * that can still load older threads (its cut until its first page). Rows below it could sort under
121 * rows not loaded yet, so the merged list is a correct prefix even while a page loads. A failed
122 * environment holds no one back; its Retry rejoins it.
123 */
124export function settledShelfFrontier(
125 environments: ReadonlyArray<OlderSettledShelfEnvironment>,
126): OrchestrationV2OlderSettledCursor | null {
127 let highest: OrchestrationV2OlderSettledCursor | null = null;
128 for (const { older, pages } of environments) {
129 if (older === null || pages.frontier === null || pages.status === "failed") continue;
130 const frontier = pages.frontier;
131 // On a settle-time tie, a cursor with an id stops above one without, and a smaller id higher.
132 if (
133 highest === null ||
134 frontier.settledAt > highest.settledAt ||
135 (frontier.settledAt === highest.settledAt &&
136 frontier.threadId !== undefined &&
137 (highest.threadId === undefined || frontier.threadId < highest.threadId))
138 ) {
139 highest = frontier;
140 }
141 }
142 return highest;
143}
144
145/**
146 * Whether the Settled shelf may list a row: at or above the shelf's frontier. So a row that turns
147 * older mid-session (a backdated auto-settle, a snooze that ends, an unarchive) waits for the
148 * range that holds it.
149 */
150export function isListedSettledThread(
151 thread: Pick<OrchestrationV2ThreadShell, "id" | "settledAt">,
152 frontier: OrchestrationV2OlderSettledCursor | null,
153): boolean {
154 return frontier === null || atOrAboveCursor(thread, frontier);
155}
156
157/**
158 * Older settled threads in scope that this client does not hold: the server's counts minus the
159 * older rows it has. Add it to the local rows for the Settled label or a project's thread count.
160 */
161export function unloadedOlderSettledCount(input: {
162 readonly olderSettled: (environmentId: EnvironmentId) => OlderSettled | null;
163 readonly scopes: ReadonlyArray<OlderSettledScope>;
164 /** Every row the client holds in scope, in any section. */
165 readonly threads: ReadonlyArray<{
166 readonly environmentId: EnvironmentId;
167 readonly source: OlderSettledRow;
168 }>;
169}): number {
170 let unloaded = 0;
171 for (const scope of input.scopes) {
172 const older = input.olderSettled(scope.environmentId);
173 if (older === null) continue;
174 if (scope.projectIds === null) {
175 for (const count of older.counts.values()) unloaded += count;
176 } else {
177 for (const projectId of scope.projectIds) unloaded += older.counts.get(projectId) ?? 0;
178 }
179 for (const thread of input.threads) {
180 if (
181 thread.environmentId === scope.environmentId &&
182 (scope.projectIds === null || scope.projectIds.includes(thread.source.projectId)) &&
183 isOlderSettledThreadShell(thread.source, older.beforeMs)
184 ) {
185 unloaded -= 1;
186 }
187 }
188 }
189 return Math.max(0, unloaded);
190}
191
192/**
193 * The pages each environment should read: while the shelf lists fewer than `visibleCount` rows,
194 * one more for each environment whose frontier is the shelf's, once its pages have loaded. A
195 * short page ends an environment's chain; counts can lag a backdated settle, so they do not gate.
196 */
197export function wantedOlderSettledPages(input: {
198 readonly requested: ReadonlyMap<EnvironmentId, number>;
199 readonly environments: ReadonlyMap<EnvironmentId, OlderSettledShelfEnvironment>;
200 readonly listedCount: number;
201 readonly visibleCount: number;
202}): ReadonlyMap<EnvironmentId, number> {
203 if (input.listedCount >= input.visibleCount) return input.requested;
204 const frontier = settledShelfFrontier([...input.environments.values()]);
205 if (frontier === null) return input.requested;
206 let next: Map<EnvironmentId, number> | null = null;
207 for (const [environmentId, { older, pages }] of input.environments) {
208 if (older === null || pages.status !== "ready" || pages.frontier === null) continue;
209 if (
210 pages.frontier.settledAt !== frontier.settledAt ||
211 pages.frontier.threadId !== frontier.threadId
212 ) {
213 continue;
214 }
215 next ??= new Map(input.requested);
216 next.set(environmentId, (input.requested.get(environmentId) ?? 0) + 1);
217 }
218 return next ?? input.requested;
219}
220
221/**
222 * Shells that omit older settled threads (`omitOlderSettled`) name the cut and count what they left
223 * out. These atoms keep those counts current and read older threads in pages for the Settled shelf.
224 */
225export function createOlderSettledThreadAtoms<R, ER, E>(
226 runtime: Atom.AtomRuntime<EnvironmentRegistry.EnvironmentRegistry | R, ER>,
227 shellStateAtom: (
228 environmentId: EnvironmentId,
229 ) => Atom.Atom<AsyncResult.AsyncResult<EnvironmentShellState, E>>,
230) {
231 const shellStateValue = (state: AsyncResult.AsyncResult<EnvironmentShellState, E>) =>
232 AsyncResult.getOrElse(state, () => null);
233
234 const countReadAtom = createEnvironmentQueryAtomFamily(runtime, {
235 label: "environment-data:orchestration:older-settled-counts",
236 staleTimeMs: Number.POSITIVE_INFINITY,
237 // A newer key unmounts this read; with no idle time its wait is interrupted before the RPC.
238 idleTtlMs: 0,
239 execute: Effect.fn("OlderSettledThreads.readCounts")(function* (input: {
240 readonly generation: number;
241 readonly before: string;
242 readonly key: string;
243 }) {
244 yield* Effect.sleep(COUNT_REREAD_DELAY);
245 return yield* request(ORCHESTRATION_V2_WS_METHODS.getOlderSettledThreads, {
246 before: { settledAt: input.before },
247 limit: 0,
248 });
249 }),
250 });
251
252 const pageReadAtom = createEnvironmentQueryAtomFamily(runtime, {
253 label: "environment-data:orchestration:older-settled-page",
254 staleTimeMs: Number.POSITIVE_INFINITY,
255 // A collapsed shelf keeps no read alive, as an idle one runs again when a reconnect replaces
256 // the session.
257 idleTtlMs: 0,
258 execute: Effect.fn("OlderSettledThreads.readPage")(function* (input: {
259 readonly generation: number;
260 readonly projectIds: ReadonlyArray<ProjectId> | null;
261 readonly before: OrchestrationV2OlderSettledCursor;
262 }) {
263 return yield* request(ORCHESTRATION_V2_WS_METHODS.getOlderSettledThreads, {
264 before: input.before,
265 ...(input.projectIds === null ? {} : { projectIds: input.projectIds }),
266 limit: OLDER_SETTLED_PAGE_SIZE,
267 });
268 }),
269 });
270
271 const countsOf = (source: {
272 readonly counts: OrchestrationV2OlderSettledThreadsPage["counts"];
273 }) => new Map(source.counts.map(({ projectId, count }) => [projectId, count] as const));
274
275 const olderSettledAtom = Atom.family((environmentId: EnvironmentId) => {
276 let generation = -1;
277 // The key of the first state seen this generation, when that state is the snapshot itself.
278 let snapshotKey: string | null = null;
279 let keyInputs: { readonly threads: unknown; readonly removedAt: unknown } | null = null;
280 let key = "";
281 let fromSnapshot: OlderSettled | null = null;
282 let lastRead: OlderSettled | null = null;
283 let lastReadPage: OrchestrationV2OlderSettledThreadsPage | null = null;
284 return Atom.make((get): OlderSettled | null => {
285 const state = shellStateValue(get(shellStateAtom(environmentId)));
286 const snapshot = state === null ? null : Option.getOrNull(state.snapshot);
287 const cut = snapshot?.olderSettled;
288 if (state === null || snapshot === null || cut === undefined) return null;
289 if (fromSnapshot?.before !== cut.before || state.generation !== generation) {
290 fromSnapshot = {
291 before: cut.before,
292 beforeMs: DateTime.toEpochMillis(DateTime.makeUnsafe(cut.before)),
293 counts: countsOf(cut),
294 };
295 }
296 // Most shell updates move only the cursor; the key needs new rows or removals.
297 if (keyInputs?.threads !== snapshot.threads || keyInputs.removedAt !== state.removedAt) {
298 keyInputs = { threads: snapshot.threads, removedAt: state.removedAt };
299 key = olderSettledCountKey(snapshot.threads, fromSnapshot.beforeMs, state.removedAt);
300 }
301 if (state.generation !== generation) {
302 generation = state.generation;
303 snapshotKey = snapshot.snapshotSequence === cut.asOfSequence ? key : null;
304 lastRead = null;
305 }
306 // Generation 0 is a cached shell: no reads until this session's first snapshot.
307 if (key === snapshotKey || state.generation === 0) return fromSnapshot;
308 const read = selectEnvironmentQueryData(
309 get(
310 countReadAtom({
311 environmentId,
312 input: { generation: state.generation, before: cut.before, key },
313 }),
314 ),
315 );
316 if (read !== null && read !== lastReadPage) {
317 lastReadPage = read;
318 lastRead = { ...fromSnapshot, counts: countsOf(read) };
319 }
320 return lastRead ?? fromSnapshot;
321 }).pipe(Atom.withLabel(`environment-older-settled:${environmentId}`));
322 });
323
324 const pagesKey = (scope: OlderSettledScope, pages: number) =>
325 JSON.stringify([
326 scope.environmentId,
327 scope.projectIds === null ? null : [...scope.projectIds].sort(),
328 pages,
329 ]);
330
331 const pagesAtom = Atom.family((key: string) => {
332 const [environmentId, projectIds, pages] = JSON.parse(key) as [
333 EnvironmentId,
334 ReadonlyArray<ProjectId> | null,
335 number,
336 ];
337 let previous = NO_PAGES;
338 const keep = (next: OlderSettledPages): OlderSettledPages => {
339 const same =
340 next.status === previous.status &&
341 next.frontier?.settledAt === previous.frontier?.settledAt &&
342 next.frontier?.threadId === previous.frontier?.threadId &&
343 (next.frontier === null) === (previous.frontier === null) &&
344 next.reads.length === previous.reads.length &&
345 next.reads.every((read, index) => read === previous.reads[index]);
346 if (!same) previous = next;
347 return previous;
348 };
349 return Atom.make((get) => {
350 const state = shellStateValue(get(shellStateAtom(environmentId)));
351 const cut = state === null ? undefined : Option.getOrNull(state.snapshot)?.olderSettled;
352 // A cached shell (an environment not connected yet, or offline) must not hold the shelf
353 // back for other environments, so it lists what it holds until its first snapshot.
354 if (state === null || cut === undefined || state.generation === 0) return keep(NO_PAGES);
355 let frontier: OrchestrationV2OlderSettledCursor = { settledAt: cut.before };
356 if (pages === 0) return keep({ reads: [], frontier, status: "ready" });
357 const reads: OrchestrationV2ThreadFamily[] = [];
358 for (let index = 0; index < pages; index += 1) {
359 const result = get(
360 pageReadAtom({
361 environmentId,
362 input: { generation: state.generation, projectIds, before: frontier },
363 }),
364 );
365 const page = selectEnvironmentQueryData(result);
366 if (page === null) {
367 return keep({
368 reads,
369 frontier,
370 status: threadFamilyRead(result) === "failed" ? "failed" : "pending",
371 });
372 }
373 reads.push(page);
374 const last = page.threads.at(-1);
375 if (page.threads.length < OLDER_SETTLED_PAGE_SIZE || last?.settledAt == null) {
376 return keep({ reads, frontier: null, status: "ready" });
377 }
378 frontier = { settledAt: DateTime.formatIso(last.settledAt), threadId: last.id };
379 }
380 return keep({ reads, frontier, status: "ready" });
381 }).pipe(Atom.withLabel(`environment-older-settled-pages:${key}`));
382 });
383
384 const pageRowsAtom = Atom.family((key: string) =>
385 Atom.map(pagesAtom(key), (pages) => pages.reads).pipe(
386 Atom.withLabel(`environment-older-settled-page-rows:${key}`),
387 ),
388 );
389
390 const searchGenerationAtom = Atom.family((environmentId: EnvironmentId) =>
391 Atom.make((get): number | "waiting" | "none" => {
392 const state = shellStateValue(get(shellStateAtom(environmentId)));
393 if (state !== null && state.generation > 0) return state.generation;
394 return state?.status === "synchronizing" ? "waiting" : "none";
395 }),
396 );
397 const searchReadAtom = createEnvironmentQueryAtomFamily(runtime, {
398 label: "environment-data:orchestration:older-settled-search",
399 idleTtlMs: 60_000,
400 execute: Effect.fn("OlderSettledThreads.search")(function* (input: {
401 readonly generation: number;
402 readonly search: OrchestrationV2SearchThreadsInput;
403 }) {
404 return yield* request(ORCHESTRATION_V2_WS_METHODS.searchThreads, input.search);
405 }),
406 });
407 // Keyed by shell generation like pages: rows read before a new snapshot do not merge, so the
408 // search reads them again.
409 const searchAtom = Atom.family((key: string) => {
410 const [environmentId, search] = JSON.parse(key) as [
411 EnvironmentId,
412 OrchestrationV2SearchThreadsInput,
413 ];
414 return Atom.make((get) => {
415 const generation = get(searchGenerationAtom(environmentId));
416 // Nothing is read before this session's first snapshot, which would drop the rows. It shows
417 // as loading only while that snapshot is on its way, so an offline environment never spins.
418 return typeof generation === "number"
419 ? get(searchReadAtom({ environmentId, input: { generation, search } }))
420 : AsyncResult.initial(generation === "waiting");
421 }).pipe(Atom.withLabel(`environment-older-settled-search:${key}`));
422 });
423
424 // One read for a shelf that spans environments: each one's cut, counts and pages.
425 const shelfAtom = Atom.family((key: string) => {
426 const entries = JSON.parse(key) as ReadonlyArray<readonly [string, string]>;
427 let previous: ReadonlyMap<EnvironmentId, OlderSettledShelfEnvironment> = new Map();
428 return Atom.make((get): ReadonlyMap<EnvironmentId, OlderSettledShelfEnvironment> => {
429 const shelf = new Map<EnvironmentId, OlderSettledShelfEnvironment>();
430 let changed = previous.size !== entries.length;
431 for (const [environmentId, pagesKey] of entries) {
432 const older = get(olderSettledAtom(environmentId as EnvironmentId));
433 const pages = get(pagesAtom(pagesKey));
434 const prior = previous.get(environmentId as EnvironmentId);
435 changed ||= prior?.older !== older || prior.pages !== pages;
436 shelf.set(environmentId as EnvironmentId, { older, pages });
437 }
438 if (changed) previous = shelf;
439 return previous;
440 }).pipe(Atom.withLabel(`environment-older-settled-shelf:${key}`));
441 });
442
443 /** Reads a failed page again (a no-op when none failed); the pages after it follow. */
444 const retryPages = (
445 registry: AtomRegistry.AtomRegistry,
446 scope: OlderSettledScope,
447 pages: number,
448 ) => {
449 const state = shellStateValue(registry.get(shellStateAtom(scope.environmentId)));
450 const result = registry.get(pagesAtom(pagesKey(scope, pages)));
451 if (state === null || result.status !== "failed" || result.frontier === null) return;
452 registry.refresh(
453 pageReadAtom({
454 environmentId: scope.environmentId,
455 input: {
456 generation: state.generation,
457 projectIds: scope.projectIds === null ? null : [...scope.projectIds].sort(),
458 before: result.frontier,
459 },
460 }),
461 );
462 };
463
464 return {
465 /** The environment's cut and current counts, or null when its shell keeps every settled thread. */
466 olderSettledAtom,
467 /** Each scope's cut, counts and first `pages` pages. */
468 shelfAtom: (
469 scopes: ReadonlyArray<{ readonly scope: OlderSettledScope; readonly pages: number }>,
470 ) =>
471 shelfAtom(
472 JSON.stringify(
473 scopes.map(({ scope, pages }) => [scope.environmentId, pagesKey(scope, pages)] as const),
474 ),
475 ),
476 retryPages,
477 /** The first `pages` pages below the cut for one scope, read in order. */
478 pagesAtom: (scope: OlderSettledScope, pages: number) => pagesAtom(pagesKey(scope, pages)),
479 /** Their rows, for `createThreadFamilyAtoms(...).mountRows`, under `rowsKey`. */
480 pageRowsAtom: (scope: OlderSettledScope, pages: number) => pageRowsAtom(pagesKey(scope, pages)),
481 rowsKey: (scope: OlderSettledScope, pages: number) => `older-settled:${pagesKey(scope, pages)}`,
482 /** `orchestration.searchThreads` for a search that asks for older settled rows. */
483 searchAtom: (environmentId: EnvironmentId, search: OrchestrationV2SearchThreadsInput) =>
484 searchAtom(JSON.stringify([environmentId, search])),
485 };
486}