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,
            }));
        },
    };
};