Watch party: guests stream instead of waiting on a device copy, one load per song, host seat and guest seats can't be taken over, host reload restores the song, real taps only detach, reconnect after background, keep invite code and chat draft

server/party.js

@@ -26,6 +26,7 @@ const ALPHABET = 'ABCDEFGHJKLMNPQRSTUVWXYZ23456789'; // no 0/O, 1/I
 const PARTY_GRACE_MS = 5 * 60_000;
 const CHAT_KEEP = 60;
 const MAX_MSG = 64 * 1024;
+const MSG_BURST = 40, MSG_PER_SEC = 20; // per member: everything except chat, which has its own guard
 export const PARTY_COMMANDS = new Set(['toggle', 'play', 'pause', 'seek', 'next', 'prev']);
 
 const clean = (v, max, fallback = '') =>
@@ -73,22 +74,24 @@ export function createPartyHub({ now = () => Date.now() } = {}) {
       if (!parties.has(code)) {
         fails.push(t); joinFails.set(ip, fails);
         if (joinFails.size > 5000) joinFails.clear();
-        return new Response('no such party', { status: 404 });
       }
     } else if (role !== 'host') {
       return new Response('unknown role', { status: 400 });
     }
-    const ok = server.upgrade(req, { data: { hub: 'party', role, pid, name, code, secret: q('secret') } });
+    // Unknown codes still upgrade so open() can answer with close code 4004 —
+    // a plain HTTP 404 made the client retry forever.
+    const ok = server.upgrade(req, { data: { hub: 'party', role, pid, name, code, secret: q('secret'), key: q('key').slice(0, 64), ip } });
     return ok ? undefined : new Response('upgrade failed', { status: 400 });
   }
 
   function open(ws) {
     const d = ws.data;
-    let p;
+    let p, lost = false;
     if (d.role === 'host') {
       p = d.code && parties.get(d.code);
       if (p && !same(p.secret, d.secret)) { ws.close(4003, 'not the host of this party'); return; }
       if (!p) {
+        lost = !!d.code; // resuming a room the server no longer has (restart / expiry)
         const code = newCode();
         p = { direct: createDirectRelay(), code, secret: randomBytes(18).toString('base64url'), hostPid: d.pid, members: new Map(), state: null, chat: [], dj: createDJQueue(), allowControl: false, hostOnline: true, hostLeftAt: 0 };
         parties.set(code, p);
@@ -99,12 +102,16 @@ export function createPartyHub({ now = () => Date.now() } = {}) {
     } else {
       p = parties.get(d.code);
       if (!p) { ws.close(4004, 'no such party'); return; }
+      // Participant ids are public (members list); a guest may not claim the host's.
+      if (d.pid === p.hostPid) { ws.close(4003, 'not the host of this party'); return; }
     }
     const prev = p.members.get(d.pid);
+    // Only the same device (same private key) may take over a participant's seat.
+    if (prev && prev.ws !== ws && prev.key && d.role !== 'host' && !same(prev.key, d.key || '')) { ws.close(4005, 'participant id in use'); return; }
     if (prev && prev.ws !== ws) { try { prev.ws.close(4000, 'replaced'); } catch { /* gone */ } }
-    p.members.set(d.pid, { ws, pid: d.pid, name: d.name, voice: false, lastChat: 0 });
+    p.members.set(d.pid, { ws, pid: d.pid, name: d.name, voice: false, lastChat: 0, key: d.key || '', host: d.role === 'host', tokens: MSG_BURST, refill: now() });
     send(ws, {
-      type: 'hello', code: p.code, secret: d.role === 'host' ? p.secret : undefined, you: d.pid,
+      type: 'hello', code: p.code, secret: d.role === 'host' ? p.secret : undefined, you: d.pid, lost,
       now: now(), members: members(p), state: p.state, chat: p.chat, dj: queueSnapshot(p.dj), allowControl: p.allowControl,
     });
     broadcast(p, { type: 'members', members: members(p), joined: d.name }, ws);
@@ -118,7 +125,13 @@ export function createPartyHub({ now = () => Date.now() } = {}) {
     const p = parties.get(d.code);
     const me = p && p.members.get(d.pid);
     if (!me || me.ws !== ws || !m || typeof m !== 'object') return;
-    const isHost = d.pid === p.hostPid;
+    const isHost = me.host && d.pid === p.hostPid;
+    if (m.type !== 'chat') {
+      const t = now();
+      me.tokens = Math.min(MSG_BURST, me.tokens + ((t - me.refill) / 1000) * MSG_PER_SEC); me.refill = t;
+      if (me.tokens < 1) return; // flood: drop
+      me.tokens -= 1;
+    }
     if (m.type === 'direct') {
       p.direct.handle(me.pid, m, new Map([...p.members].map(([pid, member]) => [pid, member.ws])), send);
     } else if (m.type === 'dj') {
@@ -141,6 +154,7 @@ export function createPartyHub({ now = () => Date.now() } = {}) {
       if (p.chat.length > CHAT_KEEP) p.chat.shift();
       broadcast(p, { type: 'chat', msg });
     } else if (m.type === 'voice') {
+      if (me.voice === !!m.on) return;
       me.voice = !!m.on;
       broadcast(p, { type: 'members', members: members(p) });
     } else if (m.type === 'rtc' && typeof m.to === 'string') {

server/party.test.js

@@ -126,3 +126,45 @@ test('DJ suggestions and votes are broadcast, host consumes approved songs', asy
   h.send({ type: 'dj', action: 'take' }); expect((await h.next('dj-play')).video.id).toBe('0gfX0dFLaBc'); expect((await g.next('dj')).entries).toEqual([]);
   h.ws.close(); g.ws.close();
 });
+
+describe('watch party seat rules', () => {
+  test('unknown code closes with 4004 (no reconnect loop)', async () => {
+    const bad = client({ role: 'guest', pid: 'guestpid7777', name: 'X', code: 'YYYYYY' });
+    expect(await bad.closed).toBe(4004);
+  });
+
+  test('a guest cannot take the host seat or another guest seat; the same device can', async () => {
+    const host = client({ role: 'host', pid: 'hostpid0002', name: 'H' });
+    const { code } = await host.next('hello');
+    const fake = client({ role: 'guest', pid: 'hostpid0002', name: 'Evil', code });
+    expect(await fake.closed).toBe(4003);
+    const g = client({ role: 'guest', pid: 'guestpid0003', key: 'k'.repeat(36), name: 'G', code });
+    await g.next('hello');
+    const thief = client({ role: 'guest', pid: 'guestpid0003', key: 'x'.repeat(36), name: 'T', code });
+    expect(await thief.closed).toBe(4005);
+    const again = client({ role: 'guest', pid: 'guestpid0003', key: 'k'.repeat(36), name: 'G', code });
+    await again.next('hello');
+    expect(await g.closed).toBe(4000);
+    host.send({ type: 'end' }); await host.closed;
+  });
+
+  test('a host resuming a room the server lost is told so', async () => {
+    const host = client({ role: 'host', pid: 'hostpid0004', name: 'H', code: 'QQQQQQ', secret: 'whatever-secret' });
+    const hello = await host.next('hello');
+    expect(hello.lost).toBe(true);
+    expect(hello.code).not.toBe('QQQQQQ');
+    host.send({ type: 'end' }); await host.closed;
+  });
+
+  test('voice presence floods are dropped', async () => {
+    const host = client({ role: 'host', pid: 'hostpid0005', name: 'H' });
+    const { code } = await host.next('hello');
+    const g = client({ role: 'guest', pid: 'guestpid0006', key: 'k'.repeat(36), name: 'G', code });
+    await g.next('hello'); host.drain('members');
+    for (let i = 0; i < 400; i++) g.send({ type: 'voice', on: i % 2 === 0 });
+    await Bun.sleep(300);
+    let n = 0; try { for (;;) { await host.next('members', 50); n++; } } catch { /* drained */ }
+    expect(n).toBeLessThanOrEqual(41);
+    host.send({ type: 'end' }); await host.closed;
+  });
+});