1import type {
2 EnvironmentId,
3 OrchestrationV2ShellSnapshot,
4 OrchestrationV2ThreadShell,
5 ThreadId,
6} from "@t3tools/contracts";
7import * as DateTime from "effect/DateTime";
8import { Atom } from "effect/reactivity";
9
10import type { AtomCommand } from "./runtime.ts";
11
12interface PendingThreadUpdate {
13 readonly threadId: ThreadId;
14 readonly apply: (thread: OrchestrationV2ThreadShell) => OrchestrationV2ThreadShell;
15 sequence?: number;
16}
17
18export function createOptimisticThreadLifecycle(
19 sourceSnapshotAtom: (
20 environmentId: EnvironmentId,
21 ) => Atom.Atom<OrchestrationV2ShellSnapshot | null>,
22) {
23 const pendingAtom = Atom.family((_environmentId: EnvironmentId) =>
24 Atom.make<ReadonlyArray<PendingThreadUpdate>>([]).pipe(Atom.keepAlive),
25 );
26 const snapshotAtom = Atom.family((environmentId: EnvironmentId) =>
27 Atom.make((get) => {
28 const snapshot = get(sourceSnapshotAtom(environmentId));
29 const pending = get(pendingAtom(environmentId));
30 if (snapshot === null || pending.length === 0) return snapshot;
31 const byThread = new Map<ThreadId, PendingThreadUpdate[]>();
32 for (const update of pending) {
33 if (update.sequence !== undefined && update.sequence <= snapshot.snapshotSequence) continue;
34 const updates = byThread.get(update.threadId) ?? [];
35 updates.push(update);
36 byThread.set(update.threadId, updates);
37 }
38 if (byThread.size === 0) return snapshot;
39 return {
40 ...snapshot,
41 threads: snapshot.threads.map((thread) =>
42 (byThread.get(thread.id) ?? []).reduce(
43 (current, update) => update.apply(current),
44 thread,
45 ),
46 ),
47 };
48 }),
49 );
50
51 // Until the durable snapshot catches up, an older pending update can still replace what shows.
52 const pendingThreadIdsAtom = Atom.family((environmentId: EnvironmentId) =>
53 Atom.make(
54 (get): ReadonlySet<ThreadId> =>
55 new Set(get(pendingAtom(environmentId)).map((update) => update.threadId)),
56 ),
57 );
58
59 function wrap<Input extends { readonly threadId: ThreadId }, E>(
60 command: AtomCommand<
61 { readonly environmentId: EnvironmentId; readonly input: Input },
62 { readonly sequence: number },
63 E
64 >,
65 apply: (
66 thread: OrchestrationV2ThreadShell,
67 input: Input,
68 now: DateTime.Utc,
69 accepted: boolean,
70 ) => OrchestrationV2ThreadShell,
71 ): typeof command {
72 return {
73 label: command.label,
74 run: async (registry, target) => {
75 const now = DateTime.nowUnsafe();
76 const pending = pendingAtom(target.environmentId);
77 const source = sourceSnapshotAtom(target.environmentId);
78 const update: PendingThreadUpdate = {
79 threadId: target.input.threadId,
80 apply: (thread) => apply(thread, target.input, now, update.sequence !== undefined),
81 };
82 const remove = () =>
83 registry.update(pending, (current) => current.filter((item) => item !== update));
84 registry.update(pending, (current) => [...current, update]);
85 let confirmed = false;
86 try {
87 const result = await command.run(registry, target);
88 if (result._tag === "Success") {
89 update.sequence = result.value.sequence;
90 registry.update(pending, (current) => [...current]);
91 const reconcile = (snapshot: OrchestrationV2ShellSnapshot | null) => {
92 if (snapshot === null || snapshot.snapshotSequence >= result.value.sequence) {
93 remove();
94 unsubscribe();
95 }
96 };
97 const unsubscribe = registry.subscribe(source, reconcile);
98 reconcile(registry.get(source));
99 confirmed = true;
100 }
101 return result;
102 } finally {
103 if (!confirmed) remove();
104 }
105 },
106 };
107 }
108
109 return { snapshotAtom, pendingThreadIdsAtom, wrap };
110}