1import * as Option from "effect/Option";
2import { Atom } from "effect/reactivity";
3import { scopedThreadKey } from "../lib/scopedEntities";
4import { appAtomRegistry } from "./atom-registry";
5import { threadHasCaughtUpTo } from "@t3tools/client-runtime/state/threads";
6import type { AcknowledgedThreadMessage, QueuedThreadMessage } from "./thread-outbox-model";
7import { environmentThreadDetails } from "./threads";
8
9// Keep acknowledged rows until the thread catches up; command receipts can precede messages.
10export const acknowledgedThreadMessagesAtom = Atom.make<ReadonlyArray<AcknowledgedThreadMessage>>(
11 [],
12).pipe(Atom.keepAlive);
13
14// Peeks at a thread only when something already has it in the registry: reading
15// it through the registry would open a snapshot load and a live stream of its own.
16function isEchoedInMountedThread(message: AcknowledgedThreadMessage): boolean {
17 const stateAtom = environmentThreadDetails.stateAtom({
18 environmentId: message.environmentId,
19 threadId: message.threadId,
20 });
21 const node = appAtomRegistry.getNodes().get(stateAtom);
22 if (node === undefined) return false;
23 const loaded: Option.Option<Atom.Type<typeof stateAtom>> = node.valueOption();
24 return Option.match(loaded, {
25 onNone: () => false,
26 onSome: (thread) =>
27 threadHasCaughtUpTo(message, {
28 sequence: thread.sequence,
29 messages: Option.getOrUndefined(thread.data)?.messages,
30 }),
31 });
32}
33
34function writeAcknowledgedThreadMessages(
35 update: (
36 current: ReadonlyArray<AcknowledgedThreadMessage>,
37 ) => ReadonlyArray<AcknowledgedThreadMessage>,
38) {
39 const current = appAtomRegistry.get(acknowledgedThreadMessagesAtom);
40 const next = update(current).filter((entry) => !isEchoedInMountedThread(entry));
41 if (next.length === current.length && next.every((entry, index) => entry === current[index])) {
42 return;
43 }
44 appAtomRegistry.set(acknowledgedThreadMessagesAtom, next);
45}
46
47export function retainAcknowledgedThreadMessage(message: AcknowledgedThreadMessage) {
48 writeAcknowledgedThreadMessages((current) =>
49 current.some((entry) => entry.messageId === message.messageId)
50 ? current
51 : [...current, message],
52 );
53}
54
55// A bounded window later pages the message out and a remount restarts at sequence 0, so
56// rows must leave as soon as their thread has caught up, not at the next write.
57export function pruneCaughtUpAcknowledgedThreadMessages(
58 threadKey: string,
59 thread: Parameters<typeof threadHasCaughtUpTo>[1],
60) {
61 writeAcknowledgedThreadMessages((current) =>
62 current.filter(
63 (entry) =>
64 scopedThreadKey(entry.environmentId, entry.threadId) !== threadKey ||
65 !threadHasCaughtUpTo(entry, thread),
66 ),
67 );
68}
69
70export function forgetAcknowledgedThreadMessage(message: QueuedThreadMessage) {
71 writeAcknowledgedThreadMessages((current) =>
72 current.filter((entry) => entry.messageId !== message.messageId),
73 );
74}