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[] = []; const server: Server = 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[] = []; const api = { report: async (body: Record) => { 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 = 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('x', { 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( 'x' ); const socket = new WebSocket(`ws://127.0.0.1:${proxy.port}/ws`); const reply = await new Promise((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('

no head

')).toStartWith('