Open History to see the last commit
1/**
2 * A queued-message edit ends on save, cancel, or when its run leaves the queue.
3 * The store owns one watcher per edit, so "run left the queue" is handled exactly
4 * once, when the projection reports it, wherever the user is in the app. It also
5 * owns the edit's save, because a save outlives the view that started it.
6 */
7import type { RunId } from "@t3tools/contracts";
8import { Atom, type AtomRegistry } from "effect/reactivity";
9
10export interface QueuedRunStatus {
11 readonly id: RunId;
12 readonly status: string;
13}
14
15/** The thread's runs, or null while they are unknown (detail not loaded yet). */
16export type QueuedRunsAtom = Atom.Atom<ReadonlyArray<QueuedRunStatus> | null>;
17
18interface ShellThreadPresence {
19 readonly id: string;
20 readonly deletedAt: unknown;
21}
22
23/**
24 * The edited thread's runs. A deleted thread (shell `deletedAt` or detail status
25 * `deleted`) reads as no runs, so its edit ends. Absence from the shell is not
26 * deletion: archived threads are dropped there but keep an editable queue, so
27 * everything else reads the detail's runs, unknown until it loads.
28 */
29export function makeQueuedRunsAtom(input: {
30 readonly threadId: string;
31 readonly snapshot: Atom.Atom<{ readonly threads: ReadonlyArray<ShellThreadPresence> } | null>;
32 readonly detailStatus: Atom.Atom<string>;
33 readonly runs: Atom.Atom<ReadonlyArray<QueuedRunStatus> | null>;
34}): QueuedRunsAtom {
35 return Atom.make((get) => {
36 const shellThread = get(input.snapshot)?.threads.find(
37 (candidate) => candidate.id === input.threadId,
38 );
39 if (shellThread !== undefined && shellThread.deletedAt !== null) return [];
40 if (get(input.detailStatus) === "deleted") return [];
41 return get(input.runs);
42 });
43}
44
45function runLeftQueue(runs: ReadonlyArray<QueuedRunStatus> | null, runId: RunId): boolean {
46 if (runs === null) return false;
47 return !runs.some((run) => run.id === runId && run.status === "queued");
48}
49
50/**
51 * What happens to an open edit when its run leaves the queue: an unchanged
52 * edit just closes; a changed one moves into the thread's own draft when that
53 * draft is empty, and is dropped otherwise so it never overwrites the user's
54 * other work.
55 */
56export function resolveQueuedEditExit(input: {
57 readonly editChanged: boolean;
58 readonly threadDraftEmpty: boolean;
59}): "kept" | "discarded" | "clean" {
60 if (!input.editChanged) return "clean";
61 return input.threadDraftEmpty ? "kept" : "discarded";
62}
63
64/**
65 * Why a queued message can't be opened in the composer now, or null when it
66 * can. An open question takes the composer over for its answer, so an edit
67 * begun then would happen out of sight and could be sent before it is seen.
68 */
69export function queuedRunEditBlockedReason(input: {
70 readonly openQuestionCount: number;
71}): string | null {
72 return input.openQuestionCount > 0 ? "Answer the question to edit queued messages" : null;
73}
74
75interface ExitWatch<E> {
76 readonly runs: QueuedRunsAtom;
77 readonly runId: RunId;
78 readonly onRunLeftQueue: (threadKey: string, edit: E) => void;
79 cancel: (() => void) | null;
80}
81
82/**
83 * Settles a save from `startSave`. False, changing nothing, when the edit was
84 * ended or begun again meanwhile. Success ends the edit; failure resumes its
85 * exit watch and returns whether the edit survived it.
86 */
87export type FinishQueuedRunEditSave = (succeeded: boolean) => boolean;
88
89export function makeQueuedRunEditStore<E extends { readonly runId: RunId }>(options: {
90 readonly registry: AtomRegistry.AtomRegistry;
91 readonly label: string;
92}) {
93 const { registry } = options;
94 const atom = Atom.make<Readonly<Record<string, E>>>({}).pipe(
95 Atom.keepAlive,
96 Atom.withLabel(options.label),
97 );
98 const savingAtom = Atom.make<Readonly<Record<string, true>>>({}).pipe(
99 Atom.keepAlive,
100 Atom.withLabel(`${options.label}-saving`),
101 );
102 // Each begin creates a new watch, so the watch is the edit's identity.
103 const watches = new Map<string, ExitWatch<E>>();
104
105 const setSaving = (threadKey: string, saving: boolean) => {
106 const current = registry.get(savingAtom);
107 if (threadKey in current === saving) return;
108 const next = { ...current };
109 if (saving) next[threadKey] = true;
110 else delete next[threadKey];
111 registry.set(savingAtom, next);
112 };
113
114 const get = (threadKey: string | null): E | null =>
115 threadKey === null ? null : (registry.get(atom)[threadKey] ?? null);
116
117 const write = (threadKey: string, edit: E | null) => {
118 const current = registry.get(atom);
119 if (edit === null) {
120 if (!(threadKey in current)) return;
121 const next = { ...current };
122 delete next[threadKey];
123 registry.set(atom, next);
124 return;
125 }
126 registry.set(atom, { ...current, [threadKey]: edit });
127 };
128
129 const unsubscribe = (watch: ExitWatch<E>) => {
130 watch.cancel?.();
131 watch.cancel = null;
132 };
133
134 const subscribe = (threadKey: string, watch: ExitWatch<E>) => {
135 const check = (runs: ReadonlyArray<QueuedRunStatus> | null) => {
136 if (watches.get(threadKey) !== watch || watch.cancel === null) return;
137 if (!runLeftQueue(runs, watch.runId)) return;
138 unsubscribe(watch);
139 watches.delete(threadKey);
140 const edit = get(threadKey);
141 if (edit === null || edit.runId !== watch.runId) return;
142 write(threadKey, null);
143 watch.onRunLeftQueue(threadKey, edit);
144 };
145 watch.cancel = registry.subscribe(watch.runs, check);
146 check(registry.get(watch.runs));
147 };
148
149 const stopWatch = (threadKey: string) => {
150 const watch = watches.get(threadKey);
151 if (watch === undefined) return;
152 unsubscribe(watch);
153 watches.delete(threadKey);
154 };
155
156 return {
157 atom,
158 /** Thread keys whose edit has a save in flight. */
159 savingAtom,
160 get,
161 /**
162 * Starts an edit, replacing any edit already open on the thread. When the
163 * run leaves the queue the edit is removed and `onRunLeftQueue` receives
164 * it; if the run already left, that happens immediately.
165 */
166 begin(
167 threadKey: string,
168 edit: E,
169 watch: {
170 readonly runs: QueuedRunsAtom;
171 readonly onRunLeftQueue: (threadKey: string, edit: E) => void;
172 },
173 ) {
174 stopWatch(threadKey);
175 write(threadKey, edit);
176 const next: ExitWatch<E> = { ...watch, runId: edit.runId, cancel: null };
177 watches.set(threadKey, next);
178 subscribe(threadKey, next);
179 },
180 update(threadKey: string, f: (edit: E) => E) {
181 const edit = get(threadKey);
182 if (edit !== null) write(threadKey, f(edit));
183 },
184 end(threadKey: string) {
185 stopWatch(threadKey);
186 write(threadKey, null);
187 },
188 /**
189 * Claims the open edit for a save, or returns null when there is no edit
190 * or a save is in flight. The exit watch pauses so the save's result decides
191 * the edit's fate. The claim outlives the edit: the server orders saves by
192 * arrival, so an overlapping older save could overwrite a newer one.
193 */
194 startSave(threadKey: string): FinishQueuedRunEditSave | null {
195 const watch = watches.get(threadKey);
196 if (watch === undefined || threadKey in registry.get(savingAtom)) return null;
197 setSaving(threadKey, true);
198 unsubscribe(watch);
199 let settled = false;
200 return (succeeded) => {
201 if (!settled) {
202 settled = true;
203 setSaving(threadKey, false);
204 }
205 if (watches.get(threadKey) !== watch) return false;
206 if (succeeded) {
207 stopWatch(threadKey);
208 write(threadKey, null);
209 return true;
210 }
211 subscribe(threadKey, watch);
212 return watches.get(threadKey) === watch;
213 };
214 },
215 };
216}