veridian/server/utils/events.ts

interface StreamConnection {
    controller: ReadableStreamDefaultController;
    lastPing: number;
    subscribedAt: number;
}

const userChannels = new Map<string, Set<StreamConnection>>();

const PING_INTERVAL = 15_000;

function sendPing(conn: StreamConnection, onError: (e: any) => void) {
    try {
        conn.controller.enqueue(':ping\n\n');
    } catch (e) {
        onError(e);
    }
}

export const userEvents = {
    subscribe(userId: string, controller: ReadableStreamDefaultController) {
        if (!userChannels.has(userId)) {
            userChannels.set(userId, new Set());
        }
        const conn: StreamConnection = {
            controller,
            lastPing: Date.now(),
            subscribedAt: Date.now(),
        };
        userChannels.get(userId)!.add(conn);

        try {
            controller.enqueue(':connected\n\n');
        } catch (e) {
            userChannels.get(userId)?.delete(conn);
        }

        const pingInterval = setInterval(() => {
            const connections = userChannels.get(userId);
            if (!connections?.has(conn)) {
                clearInterval(pingInterval);
                return;
            }
            sendPing(conn, (e) => {
                console.log('Failed to send ping, cleaning up');
                clearInterval(pingInterval);
                this.unsubscribe(userId, controller);
            });
        }, PING_INTERVAL);

        (conn as any).pingInterval = pingInterval;
    },

    unsubscribe(userId: string, controller: ReadableStreamDefaultController) {
        const connections = userChannels.get(userId);
        if (!connections) return;

        for (const conn of connections) {
            if (conn.controller === controller) {
                clearInterval((conn as any).pingInterval);
                connections.delete(conn);
                break;
            }
        }
        if (connections.size === 0) {
            userChannels.delete(userId);
        }
    },

    async emit(userId: string, entity: string, event: { op: string; payload: any; timestamp?: number }) {
        const connections = userChannels.get(userId);
        if (!connections || connections.size === 0) return;

        const timestamp = event.timestamp ?? Date.now();
        const message = `data: ${JSON.stringify({
            entity,
            op: event.op,
            payload: event.payload,
            timestamp,
        })}\n\n`;

        for (const conn of Array.from(connections)) {
            try {
                conn.controller.enqueue(message);
            } catch (e) {
                console.log('Failed to emit to connection, cleaning up');
                clearInterval((conn as any).pingInterval);
                connections.delete(conn);
            }
        }
    }
};

const topicSubscribers = new Map<string, Set<StreamConnection>>();

export const topicEvents = {
    subscribe(topicId: string, controller: ReadableStreamDefaultController) {
        if (!topicSubscribers.has(topicId)) {
            topicSubscribers.set(topicId, new Set());
        }
        const conn: StreamConnection = {
            controller,
            lastPing: Date.now(),
            subscribedAt: Date.now(),
        };
        topicSubscribers.get(topicId)!.add(conn);

        const pingInterval = setInterval(() => {
            const connections = topicSubscribers.get(topicId);
            if (!connections?.has(conn)) {
                clearInterval(pingInterval);
                return;
            }
            sendPing(conn, (e) => {
                console.log('Failed to send ping, cleaning up');
                clearInterval(pingInterval);
                this.unsubscribe(topicId, controller);
            });
        }, PING_INTERVAL);

        (conn as any).pingInterval = pingInterval;
    },

    unsubscribe(topicId: string, controller: ReadableStreamDefaultController) {
        const connections = topicSubscribers.get(topicId);
        if (!connections) return;

        for (const conn of connections) {
            if (conn.controller === controller) {
                clearInterval((conn as any).pingInterval);
                connections.delete(conn);
                break;
            }
        }
        if (connections.size === 0) {
            topicSubscribers.delete(topicId);
        }
    },

    async emit(topicId: string, events: Array<{ type: string; payload: any; timestamp?: number }> | { type: string; payload: any; timestamp?: number }) {
        const connections = topicSubscribers.get(topicId);
        if (!connections || connections.size === 0) return;

        if (!Array.isArray(events)) {
            events = [events];
        }

        for (const event of events) {
            const timestamp = event.timestamp ?? Date.now();
            const data = `data: ${JSON.stringify({ ...event, timestamp })}\n\n`;

            for (const conn of Array.from(connections)) {
                try {
                    conn.controller.enqueue(data);
                } catch (e) {
                    clearInterval((conn as any).pingInterval);
                    connections.delete(conn);
                }
            }
        }
    }
};