1import { scopeThreadRef, scopedThreadKey } from "@t3tools/client-runtime/environment";
2import type { EnvironmentThreadShell } from "@t3tools/client-runtime/state/shell";
3import { runAtomCommand } from "@t3tools/client-runtime/state/runtime";
4import type { AtomRegistry } from "effect/reactivity";
5
6import { environmentThreadShells, threadEnvironment } from "../state/threads";
7import { useUiStateStore } from "../uiStateStore";
8
9/**
10 * Hands visit times recorded locally (before servers tracked visits) to each
11 * server once. Subscribed outside render: it only acts on threads it has not
12 * seen, so a shell update must not re-render anything.
13 */
14export function startThreadVisitedMigration(registry: AtomRegistry.AtomRegistry): () => void {
15 const migratedThreadKeys = new Set<string>();
16 const migrate = (threads: ReadonlyArray<EnvironmentThreadShell>) => {
17 for (const thread of threads) {
18 // An established server watermark may be deliberately older after Mark unread.
19 if (thread.lastVisitedAt !== null) continue;
20 const threadKey = scopedThreadKey(scopeThreadRef(thread.environmentId, thread.id));
21 if (migratedThreadKeys.has(threadKey)) continue;
22 migratedThreadKeys.add(threadKey);
23 const local = useUiStateStore.getState().threadLastVisitedAtById[threadKey];
24 if (!local) continue;
25 const localMs = Date.parse(local);
26 if (!Number.isFinite(localMs)) continue;
27 void runAtomCommand(
28 registry,
29 threadEnvironment.visit,
30 { environmentId: thread.environmentId, input: { threadId: thread.id, visitedAt: local } },
31 { reportFailure: false },
32 );
33 }
34 };
35 return registry.subscribe(environmentThreadShells.threadShellsAtom, migrate, {
36 immediate: true,
37 });
38}