esports/streamer/obs.ts

/**
 * Just enough of obs-websocket 5 (built into OBS 28+) to set up and drive a
 * broadcast: https://github.com/obsproject/obs-websocket/blob/master/docs/generated/protocol.md
 *
 * `connected` means identified (authenticated), not merely an open socket.
 * Every connect and request has a timeout, so a hung OBS cannot stall a worker.
 */
import { createHash, randomUUID } from 'node:crypto';

const sha256 = (text: string) => createHash('sha256').update(text).digest('base64');

/** The challenge-response from the protocol's Authentication section. */
export function obsAuthentication(password: string, salt: string, challenge: string) {
	return sha256(sha256(password + salt) + challenge);
}

/** obs-websocket close codes worth explaining. */
const CLOSE_REASONS: Record<number, string> = {
	4009: 'OBS rejected the websocket password',
	4010: 'OBS does not support obs-websocket RPC version 1'
};

export class ObsRequestError extends Error {
	constructor(
		message: string,
		readonly requestType: string,
		readonly code: number | null
	) {
		super(message);
	}
}

interface Waiter {
	resolve(data: unknown): void;
	reject(err: Error): void;
	timer: ReturnType<typeof setTimeout>;
}

export class Obs {
	private socket: WebSocket | null = null;
	private pending = new Map<string, Waiter>();
	private identified = false;
	/** Called once when an identified connection closes. */
	onClose: ((reason: string) => void) | null = null;

	constructor(
		private url: string,
		private password: string,
		private timeoutMs = 15_000
	) {}

	get connected() {
		return this.identified && this.socket?.readyState === WebSocket.OPEN;
	}

	connect(): Promise<void> {
		this.close();
		// Each connection owns its requests, so a late close of an old socket cannot fail new ones.
		const pending = new Map<string, Waiter>();
		this.pending = pending;
		return new Promise((resolve, reject) => {
			let settled = false;
			const fail = (message: string) => {
				if (settled) return;
				settled = true;
				clearTimeout(timer);
				try {
					socket.close();
				} catch {
					/* already closed */
				}
				reject(new Error(message));
			};
			const timer = setTimeout(
				() => fail(`OBS at ${this.url} did not finish the websocket handshake in time`),
				this.timeoutMs
			);
			const socket = new WebSocket(this.url);
			this.socket = socket;
			socket.onerror = () =>
				fail(`Could not reach OBS at ${this.url}. Is its WebSocket server enabled?`);
			socket.onclose = (event) => {
				const wasIdentified = this.identified && this.socket === socket;
				if (this.socket === socket) {
					this.identified = false;
					this.socket = null;
				}
				const reason = CLOSE_REASONS[event.code] ?? `OBS closed the websocket (${event.code})`;
				for (const waiter of pending.values()) {
					clearTimeout(waiter.timer);
					waiter.reject(new Error(reason));
				}
				pending.clear();
				fail(reason);
				if (wasIdentified) this.onClose?.(reason);
			};
			socket.onmessage = (event) => {
				let message: { op: number; d: Record<string, unknown> };
				try {
					message = JSON.parse(String(event.data));
				} catch {
					return;
				}
				const { op, d } = message;
				if (op === 0) {
					const auth = d.authentication as { salt: string; challenge: string } | undefined;
					socket.send(
						JSON.stringify({
							op: 1,
							d: {
								rpcVersion: 1,
								eventSubscriptions: 0,
								...(auth
									? { authentication: obsAuthentication(this.password, auth.salt, auth.challenge) }
									: {})
							}
						})
					);
				} else if (op === 2) {
					if (settled) return;
					settled = true;
					clearTimeout(timer);
					this.identified = true;
					resolve();
				} else if (op === 7) {
					const requestId = d.requestId as string;
					const waiter = pending.get(requestId);
					if (!waiter) return;
					pending.delete(requestId);
					clearTimeout(waiter.timer);
					const status = d.requestStatus as
						{ result?: boolean; code?: number; comment?: string } | undefined;
					if (status?.result) waiter.resolve(d.responseData ?? {});
					else
						waiter.reject(
							new ObsRequestError(
								`OBS ${d.requestType}: ${status?.comment ?? status?.code ?? 'failed'}`,
								String(d.requestType),
								status?.code ?? null
							)
						);
				}
			};
		});
	}

	request<T = unknown>(
		requestType: string,
		requestData: Record<string, unknown> = {},
		timeoutMs = this.timeoutMs
	): Promise<T> {
		const socket = this.socket;
		if (!socket || !this.connected) return Promise.reject(new Error('OBS is not connected'));
		const requestId = randomUUID();
		const pending = this.pending;
		return new Promise<T>((resolve, reject) => {
			const timer = setTimeout(() => {
				pending.delete(requestId);
				reject(new Error(`OBS ${requestType} timed out`));
			}, timeoutMs);
			pending.set(requestId, { resolve: resolve as (data: unknown) => void, reject, timer });
			socket.send(JSON.stringify({ op: 6, d: { requestType, requestId, requestData } }));
		});
	}

	close() {
		const socket = this.socket;
		this.socket = null;
		this.identified = false;
		socket?.close();
	}
}