1// @effect-diagnostics globalTimers:off globalDate:off - The tracker runs in atom subscriptions, outside an Effect runtime.
2import type { EnvironmentId, ScopedThreadRef, ThreadId } from "@t3tools/contracts";
3import { Atom } from "effect/reactivity";
4
5import { parseThreadKey, threadKey } from "./entities.ts";
6import { type AtomCommand, runAtomCommand } from "./runtime.ts";
7import { claimThreadOpen } from "./threadSearch.ts";
8import type { VisitThreadInput } from "../operations/commands.ts";
9
10// During an active turn the thread's updatedAt advances several times per
11// second, and every server-side visit is a full command dispatch plus a
12// broadcast to all shell subscribers.
13export const VISIT_DISPATCH_THROTTLE_MS = 10_000;
14
15const IDLE_TRACKER: Atom.Atom<void> = Atom.make(undefined).pipe(
16 Atom.withLabel("thread-visit-tracker:idle"),
17);
18
19interface VisitedShell {
20 readonly environmentId: EnvironmentId;
21 readonly id: ThreadId;
22 readonly updatedAt: string;
23 readonly lastVisitedAt?: string | null | undefined;
24 readonly latestRun: { readonly completedAt: string | null } | null;
25 /**
26 * When a scheduled wake the user is looking at took effect. A wake later
27 * than `updatedAt` is an update the server never sent, so the watermark
28 * advances to it, and at once: leaving within the throttle would cancel
29 * the dispatch and keep the thread marked Woke.
30 */
31 readonly wokeAt?: string | null | undefined;
32}
33
34export interface ThreadVisitTrackerOptions {
35 /** Persist dispatched watermarks when a layout tab unmounts its visible thread. */
36 readonly dispatchedVisits?: Map<string, string>;
37 readonly threadShellAtom: (ref: ScopedThreadRef) => Atom.Atom<VisitedShell | null>;
38 readonly visit: AtomCommand<
39 { readonly environmentId: EnvironmentId; readonly input: VisitThreadInput },
40 unknown,
41 unknown
42 >;
43 /**
44 * Servers without visit tracking leave `lastVisitedAt` undefined; a client
45 * that keeps its own watermark for them records the visit here instead.
46 */
47 readonly markLocalVisit?: (ref: ScopedThreadRef, visitedAt: string) => void;
48}
49
50/**
51 * While mounted, a tracker counts as the user viewing the thread: mounting it
52 * records one open (which also feeds search ranking), and every shell update
53 * after that advances the visit watermark, throttled while a run streams.
54 */
55export function createThreadVisitTracker(options: ThreadVisitTrackerOptions) {
56 const family = Atom.family((key: string) => {
57 const ref = parseThreadKey(key);
58 return Atom.make((get) => {
59 const registry = get.registry;
60 // Showing the thread is the open, whether or not its shell has arrived;
61 // the shell only decides when the server hears about it.
62 // Recorded even when the server cannot take opens, so recents follow
63 // it. A palette pick already sent the open with its query.
64 let openPending = claimThreadOpen(ref.environmentId, ref.threadId, Date.now());
65 // Dedupe per watermark: the shell updates again before the command's echo
66 // lands. It also keeps a mark-unread on the open thread sticky, since the
67 // rewind leaves updatedAt untouched.
68 let lastDispatchedUpdatedAt: string | null = options.dispatchedVisits?.get(key) ?? null;
69 let lastDispatchAt = 0;
70 let timer: ReturnType<typeof setTimeout> | undefined;
71 const clearTimer = () => {
72 if (timer === undefined) return;
73 clearTimeout(timer);
74 timer = undefined;
75 };
76 get.addFinalizer(clearTimer);
77
78 const dispatch = (shell: VisitedShell, visitedAt: string, opened: boolean) => {
79 clearTimer();
80 lastDispatchedUpdatedAt = visitedAt;
81 options.dispatchedVisits?.clear();
82 options.dispatchedVisits?.set(key, visitedAt);
83 lastDispatchAt = Date.now();
84 void runAtomCommand(
85 registry,
86 options.visit,
87 {
88 environmentId: shell.environmentId,
89 input: {
90 threadId: shell.id,
91 visitedAt,
92 ...(opened ? { opened: true } : {}),
93 },
94 },
95 { reportFailure: false },
96 );
97 };
98
99 const onShell = (shell: VisitedShell | null) => {
100 clearTimer();
101 if (shell === null) return;
102
103 const shellUpdatedAtMs = Date.parse(shell.updatedAt);
104 if (Number.isNaN(shellUpdatedAtMs)) return;
105 const wokeAtMs = shell.wokeAt ? Date.parse(shell.wokeAt) : NaN;
106 const wakeIsLatest = wokeAtMs > shellUpdatedAtMs;
107 const visitedAt = wakeIsLatest && shell.wokeAt ? shell.wokeAt : shell.updatedAt;
108 const updatedAtMs = wakeIsLatest ? wokeAtMs : shellUpdatedAtMs;
109 if (shell.lastVisitedAt === undefined) {
110 options.markLocalVisit?.(ref, visitedAt);
111 return;
112 }
113 // The open rides on the first visit the server hears, however late
114 // the shell arrives, so it still counts toward search ranking.
115 if (openPending) {
116 openPending = false;
117 if (lastDispatchedUpdatedAt !== visitedAt) {
118 dispatch(shell, visitedAt, true);
119 return;
120 }
121 }
122 const lastVisitedAtMs = shell.lastVisitedAt ? Date.parse(shell.lastVisitedAt) : NaN;
123 if (!Number.isNaN(lastVisitedAtMs) && lastVisitedAtMs >= updatedAtMs) {
124 // Already seen, so a later mark-unread stays sticky even when no open was sent.
125 if (lastDispatchedUpdatedAt === null || Date.parse(lastDispatchedUpdatedAt) < updatedAtMs)
126 lastDispatchedUpdatedAt = visitedAt;
127 return;
128 }
129 if (lastDispatchedUpdatedAt !== null && Date.parse(lastDispatchedUpdatedAt) >= updatedAtMs)
130 return;
131
132 const completedAtMs = shell.latestRun?.completedAt
133 ? Date.parse(shell.latestRun.completedAt)
134 : NaN;
135 const hasUnseenCompletion =
136 !Number.isNaN(completedAtMs) &&
137 (Number.isNaN(lastVisitedAtMs) || completedAtMs > lastVisitedAtMs);
138 const elapsed = Date.now() - lastDispatchAt;
139 if (hasUnseenCompletion || wakeIsLatest || elapsed >= VISIT_DISPATCH_THROTTLE_MS) {
140 dispatch(shell, visitedAt, false);
141 return;
142 }
143 timer = setTimeout(
144 () => dispatch(shell, visitedAt, false),
145 VISIT_DISPATCH_THROTTLE_MS - elapsed,
146 );
147 };
148
149 // `subscribe`, not `get`: a rebuild would reset the open and throttle
150 // state on every streaming tick.
151 get.subscribe(options.threadShellAtom(ref), onShell, { immediate: true });
152 }).pipe(Atom.withLabel(`thread-visit-tracker:${key}`));
153 });
154 return (ref: ScopedThreadRef | null) => (ref === null ? IDLE_TRACKER : family(threadKey(ref)));
155}