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 readonly threadShellAtom: (ref: ScopedThreadRef) => Atom.Atom<VisitedShell | null>;
36 readonly visit: AtomCommand<
37 { readonly environmentId: EnvironmentId; readonly input: VisitThreadInput },
38 unknown,
39 unknown
40 >;
41 /**
42 * Servers without visit tracking leave `lastVisitedAt` undefined; a client
43 * that keeps its own watermark for them records the visit here instead.
44 */
45 readonly markLocalVisit?: (ref: ScopedThreadRef, visitedAt: string) => void;
46}
47
48/**
49 * While mounted, a tracker counts as the user viewing the thread: mounting it
50 * records one open (which also feeds search ranking), and every shell update
51 * after that advances the visit watermark, throttled while a run streams.
52 */
53export function createThreadVisitTracker(options: ThreadVisitTrackerOptions) {
54 const family = Atom.family((key: string) => {
55 const ref = parseThreadKey(key);
56 return Atom.make((get) => {
57 const registry = get.registry;
58 // Showing the thread is the open, whether or not its shell has arrived;
59 // the shell only decides when the server hears about it.
60 // Recorded even when the server cannot take opens, so recents follow
61 // it. A palette pick already sent the open with its query.
62 let openPending = claimThreadOpen(ref.environmentId, ref.threadId, Date.now());
63 // Dedupe per watermark: the shell updates again before the command's echo
64 // lands. It also keeps a mark-unread on the open thread sticky, since the
65 // rewind leaves updatedAt untouched.
66 let lastDispatchedUpdatedAt: string | null = null;
67 let lastDispatchAt = 0;
68 let timer: ReturnType<typeof setTimeout> | undefined;
69 const clearTimer = () => {
70 if (timer === undefined) return;
71 clearTimeout(timer);
72 timer = undefined;
73 };
74 get.addFinalizer(clearTimer);
75
76 const dispatch = (shell: VisitedShell, visitedAt: string, opened: boolean) => {
77 clearTimer();
78 lastDispatchedUpdatedAt = visitedAt;
79 lastDispatchAt = Date.now();
80 void runAtomCommand(
81 registry,
82 options.visit,
83 {
84 environmentId: shell.environmentId,
85 input: {
86 threadId: shell.id,
87 visitedAt,
88 ...(opened ? { opened: true } : {}),
89 },
90 },
91 { reportFailure: false },
92 );
93 };
94
95 const onShell = (shell: VisitedShell | null) => {
96 clearTimer();
97 if (shell === null) return;
98
99 const shellUpdatedAtMs = Date.parse(shell.updatedAt);
100 if (Number.isNaN(shellUpdatedAtMs)) return;
101 const wokeAtMs = shell.wokeAt ? Date.parse(shell.wokeAt) : NaN;
102 const wakeIsLatest = wokeAtMs > shellUpdatedAtMs;
103 const visitedAt = wakeIsLatest && shell.wokeAt ? shell.wokeAt : shell.updatedAt;
104 const updatedAtMs = wakeIsLatest ? wokeAtMs : shellUpdatedAtMs;
105 if (shell.lastVisitedAt === undefined) {
106 options.markLocalVisit?.(ref, visitedAt);
107 return;
108 }
109 // The open rides on the first visit the server hears, however late
110 // the shell arrives, so it still counts toward search ranking.
111 if (openPending) {
112 openPending = false;
113 dispatch(shell, visitedAt, true);
114 return;
115 }
116 const lastVisitedAtMs = shell.lastVisitedAt ? Date.parse(shell.lastVisitedAt) : NaN;
117 if (!Number.isNaN(lastVisitedAtMs) && lastVisitedAtMs >= updatedAtMs) {
118 // Already seen, so a later mark-unread stays sticky even when no open was sent.
119 if (lastDispatchedUpdatedAt === null || Date.parse(lastDispatchedUpdatedAt) < updatedAtMs)
120 lastDispatchedUpdatedAt = visitedAt;
121 return;
122 }
123 if (lastDispatchedUpdatedAt !== null && Date.parse(lastDispatchedUpdatedAt) >= updatedAtMs)
124 return;
125
126 const completedAtMs = shell.latestRun?.completedAt
127 ? Date.parse(shell.latestRun.completedAt)
128 : NaN;
129 const hasUnseenCompletion =
130 !Number.isNaN(completedAtMs) &&
131 (Number.isNaN(lastVisitedAtMs) || completedAtMs > lastVisitedAtMs);
132 const elapsed = Date.now() - lastDispatchAt;
133 if (hasUnseenCompletion || wakeIsLatest || elapsed >= VISIT_DISPATCH_THROTTLE_MS) {
134 dispatch(shell, visitedAt, false);
135 return;
136 }
137 timer = setTimeout(
138 () => dispatch(shell, visitedAt, false),
139 VISIT_DISPATCH_THROTTLE_MS - elapsed,
140 );
141 };
142
143 // `subscribe`, not `get`: a rebuild would reset the open and throttle
144 // state on every streaming tick.
145 get.subscribe(options.threadShellAtom(ref), onShell, { immediate: true });
146 }).pipe(Atom.withLabel(`thread-visit-tracker:${key}`));
147 });
148 return (ref: ScopedThreadRef | null) => (ref === null ? IDLE_TRACKER : family(threadKey(ref)));
149}