71 lines
2.7 KiB
TypeScript
71 lines
2.7 KiB
TypeScript
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,
|
|
}));
|
|
},
|
|
};
|
|
};
|