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 function wrap<Input extends { readonly threadId: ThreadId }, E>(
52 command: AtomCommand<
53 { readonly environmentId: EnvironmentId; readonly input: Input },
54 { readonly sequence: number },
55 E
56 >,
57 apply: (
58 thread: OrchestrationV2ThreadShell,
59 input: Input,
60 now: DateTime.Utc,
61 accepted: boolean,
62 ) => OrchestrationV2ThreadShell,
63 ): typeof command {
64 return {
65 label: command.label,
66 run: async (registry, target) => {
67 const now = DateTime.nowUnsafe();
68 const pending = pendingAtom(target.environmentId);
69 const source = sourceSnapshotAtom(target.environmentId);
70 const update: PendingThreadUpdate = {
71 threadId: target.input.threadId,
72 apply: (thread) => apply(thread, target.input, now, update.sequence !== undefined),
73 };
74 const remove = () =>
75 registry.update(pending, (current) => current.filter((item) => item !== update));
76 registry.update(pending, (current) => [...current, update]);
77 let confirmed = false;
78 try {
79 const result = await command.run(registry, target);
80 if (result._tag === "Success") {
81 update.sequence = result.value.sequence;
82 registry.update(pending, (current) => [...current]);
83 const reconcile = (snapshot: OrchestrationV2ShellSnapshot | null) => {
84 if (snapshot === null || snapshot.snapshotSequence >= result.value.sequence) {
85 remove();
86 unsubscribe();
87 }
88 };
89 const unsubscribe = registry.subscribe(source, reconcile);
90 reconcile(registry.get(source));
91 confirmed = true;
92 }
93 return result;
94 } finally {
95 if (!confirmed) remove();
96 }
97 },
98 };
99 }
100
101 return { snapshotAtom, wrap };
102}