esports/streamer/link.test.ts

import { afterEach, describe, expect, test } from 'bun:test';
import type { Server } from 'bun';
import { VashLink, type Follow } from './link';
import { Broadcaster, parsePlaylist, type Source } from './relay';
import type { StreamConfig, VashApi } from './vash-api';
import { startAccessProxy, withoutReferrer } from './windows/proxy';

const servers: { stop(force?: boolean): unknown }[] = [];
afterEach(() => {
	for (const server of servers.splice(0)) server.stop(true);
});

const until = async (check: () => boolean, ms = 2000) => {
	for (let waited = 0; waited < ms && !check(); waited += 10) await Bun.sleep(10);
	expect(check()).toBe(true);
};

describe('VashLink', () => {
	function fakeVash(token: string) {
		const received: Record<string, unknown>[] = [];
		const server: Server<undefined> = Bun.serve({
			port: 0,
			fetch: (request, server) => (server.upgrade(request) ? undefined : new Response('no')),
			websocket: {
				open(ws) {
					ws.subscribe('agents');
				},
				message(ws, raw) {
					const message = JSON.parse(String(raw));
					received.push(message);
					if (message.type === 'tosu')
						ws.send(
							JSON.stringify({
								type: 'scores',
								matchId: message.matchId,
								accepted: message.clients.length
							})
						);
					if (message.type !== 'hello') return;
					if (message.token !== token) return ws.close(4001, 'Unknown channel or wrong token');
					ws.send(JSON.stringify({ type: 'welcome', channel: 'Main' }));
					ws.send(
						JSON.stringify({
							type: 'follow',
							scene: 'gameplay',
							matchId: 'm1',
							lobbyId: 118000042,
							roomName: 'VASH: (A) vs (B)',
							teamSize: 2,
							title: 'A vs B'
						})
					);
				}
			}
		});
		servers.push(server);
		return { server, received, url: `http://127.0.0.1:${server.port}` };
	}

	test('signs in, follows, throttles scores and reports status only when it changes', async () => {
		const vash = fakeVash('good');
		const follows: Follow[] = [];
		const link = new VashLink({
			server: vash.url,
			channel: 'c',
			token: 'good',
			onFollow: (follow) => follows.push(follow)
		});
		link.start();
		await until(() => link.connected && follows.length === 1);
		expect(follows[0]).toEqual({
			scene: 'gameplay',
			matchId: 'm1',
			lobbyId: 118000042,
			roomName: 'VASH: (A) vs (B)',
			teamSize: 2,
			title: 'A vs B'
		});
		const frame = (score: number) => ({
			ipcState: 3,
			beatmapId: 1,
			progress: 0.5,
			teams: { left: 'A', right: 'B' },
			clients: [{ osuId: 2, name: 'p', score, accuracy: 99, combo: 1, failed: false }]
		});
		for (let score = 1; score <= 10; score++) link.sendScores(frame(score));
		await Bun.sleep(300);
		const scores = vash.received.filter((message) => message.type === 'tosu');
		expect(scores.length).toBeLessThanOrEqual(2);
		expect((scores.at(-1)!.clients as { score: number }[])[0].score).toBe(10);

		const status = {
			state: 'spectating',
			detail: null,
			onAir: true,
			encoder: 'AMD AMF (hardware)',
			instance: 'prod',
			version: '0.1.0'
		};
		link.report(status);
		link.report(status);
		link.report({ ...status, state: 'selecting' });
		await Bun.sleep(50);
		expect(
			vash.received.filter((message) => message.type === 'status').map((message) => message.state)
		).toEqual(['spectating', 'selecting']);
		link.stop();
	});

	test('sends accuracy, combo and progress changes, heartbeats fresh scores, and drops queued scores on a director cut', async () => {
		const vash = fakeVash('good');
		const link = new VashLink({ server: vash.url, channel: 'c', token: 'good' });
		link.start();
		try {
			await until(() => link.follow?.matchId === 'm1');
			const frame = {
				ipcState: 3,
				beatmapId: 75,
				progress: 0.1,
				teams: { left: 'A', right: 'B' },
				clients: [{ osuId: 2, name: 'p', score: 100, accuracy: 99, combo: 10, failed: false }]
			};
			const messages = () => vash.received.filter((message) => message.type === 'tosu');
			link.sendScores(frame);
			await until(() => link.scores.players === 1);
			expect(messages()[0]).toMatchObject({ matchId: 'm1', beatmapId: 75 });
			expect(link.scores.acceptedAt).toBeGreaterThan(0);
			const changed = {
				...frame,
				progress: 0.2,
				clients: [{ ...frame.clients[0], accuracy: 97, combo: 0 }]
			};
			link.sendScores(changed);
			await until(() => messages().length === 2);
			expect(messages()[1]).toMatchObject({
				progress: 0.2,
				clients: [{ score: 100, accuracy: 97, combo: 0 }]
			});
			await Bun.sleep(1050);
			link.sendScores(changed);
			await until(() => messages().length === 3);
			link.sendScores({ ...changed, progress: 0.3 });
			vash.server.publish(
				'agents',
				JSON.stringify({ type: 'follow', matchId: 'm2', lobbyId: 2, scene: 'gameplay' })
			);
			await until(() => link.follow?.matchId === 'm2');
			await Bun.sleep(300);
			expect(messages()).toHaveLength(3);
			expect(link.scores.players).toBe(0);
			link.sendScores(changed);
			await until(() => messages().length === 4);
			expect(messages()[3]).toMatchObject({ matchId: 'm2', beatmapId: 75 });
		} finally {
			link.stop();
		}
	});

	test('a refused token is reported, not retried in a tight loop', async () => {
		const vash = fakeVash('good');
		let rejected = 0;
		const link = new VashLink({
			server: vash.url,
			channel: 'c',
			token: 'bad',
			onRejected: () => rejected++
		});
		link.start();
		await until(() => rejected === 1);
		expect(link.rejected).toBe(true);
		await Bun.sleep(200);
		expect(vash.received.filter((message) => message.type === 'hello')).toHaveLength(1);
		link.stop();
	});
});

describe('Broadcaster', () => {
	class FakeSource implements Source {
		starts = 0;
		stops = 0;
		running = false;
		start() {
			this.starts++;
			this.running = true;
		}
		stop() {
			this.stops++;
			this.running = false;
		}
	}

	test('follows Go live: one session per broadcast, restarting a dropped source', async () => {
		const reports: Record<string, unknown>[] = [];
		const api = {
			report: async (body: Record<string, unknown>) => {
				reports.push(body);
				return body.event === 'start' ? { id: 's1', storagePrefix: null } : { ok: true };
			}
		} as unknown as VashApi;
		const source = new FakeSource();
		const broadcaster = new Broadcaster({
			api,
			ffmpeg: 'ffmpeg',
			hlsDir: '/nonexistent',
			source,
			log: () => {}
		});
		const config: StreamConfig = {
			live: true,
			width: 1920,
			height: 1080,
			fps: 60,
			bitrateKbps: 6000,
			overlayUrl: '',
			record: false,
			destinations: []
		};
		await broadcaster.apply(config);
		expect(broadcaster.onAir).toBe(true);
		expect(reports).toEqual([{ event: 'start' }]);
		await broadcaster.apply(config);
		expect(source.starts).toBe(1);
		source.running = false; // OBS dropped
		await broadcaster.apply(config);
		expect(source.starts).toBe(2);
		await broadcaster.apply({ ...config, live: false });
		expect(broadcaster.onAir).toBe(false);
		expect(reports.at(-1)).toMatchObject({ event: 'end', sessionId: 's1' });
		expect(source.running).toBe(false);
	}, 10_000);

	test('reads HLS playlists with CRLF line endings', () => {
		expect(
			parsePlaylist('#EXTM3U\r\n#EXTINF:2.000,\r\nseg_00001.ts\r\n#EXTINF:1.5,\r\nseg_00002.ts\r\n')
		).toEqual([
			{ name: 'seg_00001.ts', duration: 2 },
			{ name: 'seg_00002.ts', duration: 1.5 }
		]);
	});
});

describe('Access proxy', () => {
	test('adds the service token to page loads and websockets, and drops the referrer', async () => {
		const seen: string[] = [];
		const upstream: Server<undefined> = Bun.serve({
			port: 0,
			fetch(request, server) {
				seen.push(request.headers.get('cf-access-client-id') ?? 'none');
				if (request.headers.get('upgrade'))
					return server.upgrade(request) ? undefined : new Response('no');
				return new Response('<html><head><title>x</title></head></html>', {
					headers: { 'content-type': 'text/html' }
				});
			},
			websocket: {
				message: (ws, message) => {
					ws.send(`echo ${message}`);
				}
			}
		});
		servers.push(upstream);
		const proxy = startAccessProxy({
			port: 0,
			upstream: `http://127.0.0.1:${upstream.port}`,
			headers: { 'CF-Access-Client-Id': 'id', 'CF-Access-Client-Secret': 'secret' },
			log: () => {}
		});
		servers.push(proxy);
		const html = await (await fetch(`http://127.0.0.1:${proxy.port}/broadcast/x/overlay`)).text();
		expect(html).toBe(
			'<html><head><meta name="referrer" content="no-referrer"><title>x</title></head></html>'
		);
		const socket = new WebSocket(`ws://127.0.0.1:${proxy.port}/ws`);
		const reply = await new Promise<string>((resolve) => {
			socket.onopen = () => socket.send('hi');
			socket.onmessage = (event) => resolve(String(event.data));
		});
		socket.close();
		expect(reply).toBe('echo hi');
		expect(seen).toEqual(['id', 'id']);
		expect(withoutReferrer('<p>no head</p>')).toStartWith('<meta name="referrer"');
	});
});