1import * as progress from "@clo/lib/progress";
2import { attachToDocument } from "@clo/lib/progress/browser";
3import type {
4 EnvironmentId,
5 ScriptProgressEvent,
6 ScriptProgressStream,
7 ThreadId,
8} from "@t3tools/contracts";
9import { AsyncResult } from "effect/reactivity";
10import { useEffect, useRef } from "react";
11
12import { appAtomRegistry } from "../../rpc/atomRegistry";
13import { scriptProgressEnvironment } from "../../state/scriptProgress";
14
15export function ScriptProgress({
16 environmentId,
17 threadId,
18}: {
19 environmentId: EnvironmentId;
20 threadId: ThreadId;
21}) {
22 const container = useRef<HTMLElement>(null);
23
24 useEffect(() => {
25 const host = container.current;
26 if (!host) return;
27
28 const streams = new Map<
29 string,
30 {
31 root: progress.Root;
32 input: ReadableStreamDefaultController<progress.StreamEvent>;
33 element: HTMLElement;
34 status: HTMLElement;
35 dispose: () => void;
36 }
37 >();
38
39 const updateVisibility = () => {
40 let visible = false;
41 for (const stream of streams.values()) {
42 stream.element.hidden = !stream.root.active.some(progress.isVisible);
43 visible ||= !stream.element.hidden;
44 }
45 host.hidden = !visible;
46 };
47
48 const remove = (id: string) => {
49 const stream = streams.get(id);
50 if (!stream) return;
51 streams.delete(id);
52 stream.dispose();
53 stream.element.remove();
54 updateVisibility();
55 };
56
57 const upsert = (descriptor: ScriptProgressStream) => {
58 let stream = streams.get(descriptor.id);
59 if (!stream) {
60 const root = new progress.Root();
61 const element = document.createElement("div");
62 element.hidden = true;
63 const status = document.createElement("div");
64 status.className = "text-xs text-muted-foreground";
65 const tree = document.createElement("div");
66 tree.dataset.progressMotion = "static";
67 element.append(status, tree);
68 host.append(element);
69
70 const detach = attachToDocument(root, tree);
71 const off = root.on("change", updateVisibility);
72 let input: ReadableStreamDefaultController<progress.StreamEvent> | undefined;
73 const readable = new ReadableStream<progress.StreamEvent>({
74 start(controller) {
75 input = controller;
76 },
77 });
78 if (!input) throw new Error("Progress stream did not initialize.");
79
80 const decoder = progress.decodeEventStream(readable, root);
81 void decoder.catch(() => {
82 status.textContent = "Progress unavailable";
83 status.hidden = false;
84 updateVisibility();
85 });
86 stream = {
87 root,
88 input,
89 element,
90 status,
91 dispose: () => {
92 off();
93 decoder.cancel();
94 detach();
95 },
96 };
97 streams.set(descriptor.id, stream);
98 }
99
100 stream.status.textContent =
101 descriptor.status === "disconnected" ? `${descriptor.label} · Disconnected` : "";
102 stream.status.hidden = descriptor.status === "connected";
103 return stream;
104 };
105
106 const subscription = scriptProgressEnvironment.watch({
107 environmentId,
108 input: { threadId },
109 });
110 appAtomRegistry.refresh(subscription);
111 let previousEvents: ReadonlyArray<ScriptProgressEvent> | undefined;
112 const unsubscribe = appAtomRegistry.subscribe(
113 subscription,
114 (result) => {
115 if (!AsyncResult.isSuccess(result)) {
116 previousEvents = undefined;
117 for (const id of streams.keys()) remove(id);
118 return;
119 }
120
121 const { events } = result.value;
122 if (events === previousEvents) return;
123 previousEvents = events;
124 for (const event of events) {
125 switch (event.type) {
126 case "snapshot":
127 for (const id of streams.keys()) remove(id);
128 for (const descriptor of event.streams) upsert(descriptor);
129 break;
130 case "upsert":
131 upsert(event.stream);
132 break;
133 case "progress": {
134 const stream = streams.get(event.id);
135 // The shared decoder owns the stable wire format; RPC validates JSON at its boundary.
136 stream?.input.enqueue(event.event as progress.StreamEvent);
137 break;
138 }
139 case "remove":
140 remove(event.id);
141 break;
142 }
143 }
144 },
145 { immediate: true },
146 );
147
148 return () => {
149 unsubscribe();
150 for (const id of streams.keys()) remove(id);
151 };
152 }, [environmentId, threadId]);
153
154 return (
155 <section
156 ref={container}
157 hidden
158 aria-label="Script progress"
159 className="mb-2 max-h-56 space-y-2 overflow-auto rounded-xl border border-border bg-card p-3 [--progress-accent:var(--primary)] [--progress-background:var(--card)] [--progress-border:var(--border)] [--progress-foreground:var(--foreground)] [--progress-muted:var(--muted-foreground)] [--progress-secondary:var(--secondary)]"
160 />
161 );
162}