bug fixes, and performance improvements, and drawer

server/utils/topic-snapshot-stream.ts

@@ -0,0 +1,70 @@
+import { createHash } from 'node:crypto';
+import { Err, Ok } from '../../types/result';
+import { serializeTopicSnapshot } from '../../utils/topic-sync';
+import { topicEvents } from './events';
+
+interface TopicSnapshot {
+    messages: { id: string; parts: any[] }[];
+}
+
+export const createTopicSnapshotStream = (topicId: string, offeredFingerprint: unknown) => {
+    let controller: ReadableStreamDefaultController | undefined;
+    let closed = false;
+    const buffered: string[] = [];
+    const sink = {
+        enqueue(frame: string) {
+            if (closed) return;
+            if (controller) controller.enqueue(frame);
+            else buffered.push(frame);
+        },
+    };
+    const close = () => {
+        closed = true;
+        topicEvents.unsubscribe(topicId, sink);
+        buffered.length = 0;
+    };
+
+    return {
+        close,
+        async open(readSnapshot: () => Promise<TopicSnapshot | undefined>) {
+            if (closed) return Err('closed' as const);
+            // Subscribe before reading: changes during the query must not be lost.
+            topicEvents.subscribe(topicId, sink);
+            let topic;
+            try {
+                topic = await readSnapshot();
+            } catch (error) {
+                close();
+                console.error('Failed to read topic snapshot', error);
+                return Err('read-failed' as const);
+            }
+            if (closed || !topic) {
+                close();
+                return Err('not-found' as const);
+            }
+            // Include deltas already emitted but not yet flushed to PostgreSQL.
+            topicEvents.applyLiveText(topicId, topic.messages);
+            const fingerprint = createHash('sha256').update(serializeTopicSnapshot(topic)).digest('hex');
+            const matches = offeredFingerprint === fingerprint;
+
+            return Ok(new ReadableStream<string>({
+                start(streamController) {
+                    controller = streamController;
+                    controller.enqueue(`data: ${JSON.stringify(matches
+                        ? { type: 'synced', payload: { id: topicId } }
+                        : { type: 'initial_state', payload: topic })}\n\n`);
+                    for (const frame of buffered) {
+                        if (frame.startsWith('data: ')) {
+                            const replay = { ...JSON.parse(frame.slice(6)), replay: true };
+                            controller.enqueue(`data: ${JSON.stringify(replay)}\n\n`);
+                        } else {
+                            controller.enqueue(frame);
+                        }
+                    }
+                    buffered.length = 0;
+                },
+                cancel: close,
+            }));
+        },
+    };
+};