owltide/composables/event-source.ts

/*
	SPDX-FileCopyrightText: © 2025 Hornwitser <code@hornwitser.no>
	SPDX-License-Identifier: AGPL-3.0-or-later
*/
import type { ApiEvent } from "~/shared/types/api";

interface AppEventMap {
	"open": Event,
	"message": MessageEvent<string>,
	"update": MessageEvent<ApiEvent>,
	"error": Event,
	"close": Event,
}

class AppEventSource extends EventTarget {
	#source: EventSource | null = null;
	#sourceSessionId: number | undefined = undefined;

	#forwardEvent(type: string) {
		this.#source!.addEventListener(type, event => {
			if (type === "open" || type === "message" || type === "error") {
				console.log("AppEventSource", event.type, event.data);
				this.dispatchEvent(new Event(event.type));
			} else {
				const data = event.data ? JSON.parse(event.data) as ApiEvent : undefined;
				console.log("AppEventSource", event.type, data);
				if (data?.type === "connected") {
					this.#sourceSessionId = data.session?.id;
				}
				this.dispatchEvent(new MessageEvent(event.type, {
					data,
					origin: event.origin,
					lastEventId: event.lastEventId,
					source: event.source,
					ports: [...event.ports],
				}));
			}
		});
	}

	open(sessionId: number | undefined) {
		console.log("Opening event source sid:", sessionId);
		this.#sourceSessionId = sessionId;
		this.#source = new EventSource("/api/events");
		this.#forwardEvent("open");
		this.#forwardEvent("message");
		this.#forwardEvent("update");
		this.#forwardEvent("error");
	}

	close() {
		console.log("Closing event source sid:", this.#sourceSessionId);
		this.#source!.close();
		this.#source = null;
		this.#sourceSessionId = undefined;
		console.log("AppEventSource", "close");
		this.dispatchEvent(new Event("close"));
	}

	#connectRefs = 0;
	connect(sessionId: number | undefined) {
		this.#connectRefs += 1;
		if (this.#source && this.#sourceSessionId !== sessionId) {
			this.close();
		}
		if (!this.#source) {
			this.open(sessionId);
		}
	}

	reconnect(sessionId: number | undefined) {
		if (this.#source && this.#sourceSessionId !== sessionId) {
			this.close();
			this.open(sessionId);
		}
	}

	disconnect() {
		if (this.#connectRefs === 0) {
			throw Error("Connection reference count already zero");
		}
		this.#connectRefs -= 1;
		if (this.#connectRefs === 0 && this.#source) {
			this.close();
		}
	}

	override addEventListener<K extends keyof AppEventMap>(
		type: K,
		listener: (this: AppEventSource, ev: AppEventMap[K]) => any,
		options?: boolean | AddEventListenerOptions
	) {
		super.addEventListener(type, listener as (ev: Event) => void, options);
	}

	override dispatchEvent<K extends keyof AppEventMap>(
		event: AppEventMap[K],
	) {
		return super.dispatchEvent(event);
	}

	override removeEventListener<K extends keyof AppEventMap>(
		type: K,
		listener: (this: AppEventSource, ev: AppEventMap[K]) => any,
		options?: boolean | EventListenerOptions
	) {
		return super.removeEventListener(type, listener as (ev: Event) => void, options);
	}
}

// The event source exists only on the client.
export const appEventSource = import.meta.client ? new AppEventSource() : null;

export function useEventSource() {
	const sessionStore = useSessionStore();
	onMounted(() => {
		console.log("useEventSource onMounted", sessionStore.id);
		appEventSource!.connect(sessionStore.id);
	})

	watch(() => sessionStore.id, () => {
		console.log("useEventSource sessionStore.id change", sessionStore.id);
		appEventSource!.reconnect(sessionStore.id);
	})

	onUnmounted(() => {
		console.log("useEventSource onUnmounted");
		appEventSource!.disconnect();
	});

	return appEventSource;
}