watchparty/server/test/socketGateway.test.ts

// End-to-end checks of the realtime protocol through the REAL
// SocketGateway + PartyService + MusicQueueService together (not just one
// service in isolation) - a fake io/socket pair standing in for Socket.IO
// itself, since a real client<->server round trip needs an actual HTTP
// upgrade.
import { describe, it } from 'node:test';
import assert from 'node:assert/strict';
import { PartyService } from '../src/services/PartyService';
import { MusicQueueService } from '../src/services/MusicQueueService';
import { SocketGateway } from '../src/realtime/SocketGateway';
import type { AppServer, AppSocket, LobbyPartyDto } from '../src/types';

type Handler = (payload?: unknown) => void;

function fakeSocket(cookie: string) {
  const handlers = new Map<string, Handler[]>();
  const emitted: [string, unknown][] = [];
  return {
    request: { headers: { cookie } },
    emitted,
    on(event: string, cb: Handler) {
      const list = handlers.get(event) || [];
      list.push(cb);
      handlers.set(event, list);
    },
    emit(event: string, payload?: unknown) { emitted.push([event, payload]); },
    to() { return { emit() {} }; },
    join() {},
    disconnect() {},
    trigger(event: string, payload?: unknown) { (handlers.get(event) || []).forEach((cb) => cb(payload)); }
  };
}

function fakeIo() {
  const handlers = new Map<string, ((...args: unknown[]) => void)[]>();
  const roomEmits: { room: string; event: string; payload: unknown }[] = [];
  return {
    roomEmits,
    on(event: string, cb: (...args: unknown[]) => void) {
      const list = handlers.get(event) || [];
      list.push(cb);
      handlers.set(event, list);
    },
    to(room: string) { return { emit: (event: string, payload: unknown) => roomEmits.push({ room, event, payload }) }; },
    trigger(event: string, ...args: unknown[]) { (handlers.get(event) || []).forEach((cb) => cb(...args)); }
  };
}

function attach(io: ReturnType<typeof fakeIo>) {
  SocketGateway.attach(
    io as unknown as AppServer,
    (raw) => ({ wp_session: raw as string }),
    (sessionId) => ({ userId: sessionId, userName: sessionId })
  );
  MusicQueueService.attachIo(io as unknown as AppServer, () => {});
}

describe('SocketGateway', () => {
  it('voteSkip/musicPlaystate are no-ops on a video party', () => {
    const io = fakeIo();
    attach(io);
    const party = PartyService.create('u-host', 'Host', 'm1', 'Un Film', '/img/m1', null, 'video');
    const socket = fakeSocket('u-host');
    io.trigger('connection', socket as unknown as AppSocket);
    socket.trigger('join', { partyId: party.id });

    const before = io.roomEmits.length;
    socket.trigger('voteSkip');
    socket.trigger('musicPlaystate', { isPaused: true });
    assert.equal(io.roomEmits.length, before, 'neither event should produce any broadcast on a video party');
  });

  it('a non-host member can pause and resume a music party directly (no host check)', () => {
    const io = fakeIo();
    attach(io);
    const party = PartyService.create('u-alice', 'Alice', 't1', 'Track One', '/img/t1', null, 'music');
    MusicQueueService.init(party, { itemId: 't1', itemName: 'Track One', itemImageUrl: '/img/t1', durationTicks: 10_000_000 }, 'u-alice', 'Alice');
    PartyService.join(party, 'u-bob', 'Bob');

    const bobSocket = fakeSocket('u-bob');
    io.trigger('connection', bobSocket as unknown as AppSocket);
    bobSocket.trigger('join', { partyId: party.id });

    assert.notEqual(party.hostUserId, 'u-bob');
    bobSocket.trigger('musicPlaystate', { isPaused: true });
    assert.equal(party.music?.current?.isPaused, true);
    bobSocket.trigger('musicPlaystate', { isPaused: false });
    assert.equal(party.music?.current?.isPaused, false);
  });

  it('voteSkip needs a majority of connected members and broadcasts queueUpdated on the party room', () => {
    const io = fakeIo();
    attach(io);
    const party = PartyService.create('u-alice', 'Alice', 't1', 'Track One', '/img/t1', null, 'music');
    MusicQueueService.init(party, { itemId: 't1', itemName: 'Track One', itemImageUrl: '/img/t1', durationTicks: 10_000_000 }, 'u-alice', 'Alice');
    PartyService.join(party, 'u-bob', 'Bob');

    const aliceSocket = fakeSocket('u-alice');
    io.trigger('connection', aliceSocket as unknown as AppSocket);
    aliceSocket.trigger('join', { partyId: party.id });
    const bobSocket = fakeSocket('u-bob');
    io.trigger('connection', bobSocket as unknown as AppSocket);
    bobSocket.trigger('join', { partyId: party.id });

    bobSocket.trigger('voteSkip');
    assert.equal(party.music?.current?.track.itemId, 't1', 'one of two connected should not skip yet');

    io.roomEmits.length = 0;
    aliceSocket.trigger('voteSkip');
    assert.equal(party.music?.current, null, 'the second vote reaches the majority (no second track queued)');
    assert.ok(io.roomEmits.some((e) => e.room === party.id && e.event === 'queueUpdated'));
  });

  it('leaving hands the host off live and refreshes the lobby', () => {
    const io = fakeIo();
    attach(io);
    const party = PartyService.create('u-alice', 'Alice', 'm1', 'Un Film', '/img/m1', null, 'video');
    PartyService.join(party, 'u-bob', 'Bob');
    const bobSocket = fakeSocket('u-bob');
    io.trigger('connection', bobSocket as unknown as AppSocket);
    bobSocket.trigger('join', { partyId: party.id });

    const aliceSocket = fakeSocket('u-alice');
    io.trigger('connection', aliceSocket as unknown as AppSocket);
    aliceSocket.trigger('join', { partyId: party.id });
    aliceSocket.trigger('disconnect');

    assert.equal(party.hostUserId, 'u-bob');
    assert.ok(bobSocket.emitted.some(([event]) => event === 'hostChanged'), 'the new host should get a personal notification');
  });

  it('playstate logs only on an actual pause/play transition, never on an unchanged heartbeat resend', (t) => {
    const logged: string[] = [];
    t.mock.method(console, 'log', (line: string) => { logged.push(line); });

    const io = fakeIo();
    attach(io);
    const party = PartyService.create('u-alice', 'Alice', 'm1', 'Un Film', '/img/m1', null, 'video');
    const socket = fakeSocket('u-alice');
    io.trigger('connection', socket as unknown as AppSocket);
    socket.trigger('join', { partyId: party.id });

    const playstateLines = () => logged.filter((l) => l.includes(' playstate '));

    // The initial state is a real transition (a new member defaults to
    // isPaused:true - see PartyService's newMember) - this is "the party
    // started playing", and should log.
    socket.trigger('playstate', { isPaused: false, positionTicks: 10_000_000 });
    assert.equal(playstateLines().length, 1);

    // 5 heartbeats, same isPaused, only position moving - none of these
    // should add a line (the player re-sends its current state every 3s
    // regardless of whether anything changed - see docs/architecture.md).
    for (let i = 0; i < 5; i++) socket.trigger('playstate', { isPaused: false, positionTicks: 10_000_000 + i * 30_000_000 });
    assert.equal(playstateLines().length, 1);

    socket.trigger('playstate', { isPaused: true, positionTicks: 160_000_000 });
    assert.equal(playstateLines().length, 2);

    for (let i = 0; i < 3; i++) socket.trigger('playstate', { isPaused: true, positionTicks: 160_000_000 });
    assert.equal(playstateLines().length, 2);

    socket.trigger('playstate', { isPaused: false, positionTicks: 160_000_000 });
    assert.equal(playstateLines().length, 3);
  });

  it('requestPause is throttled per member (5s) even against a raw repeat', (t) => {
    const logged: string[] = [];
    t.mock.method(console, 'log', (line: string) => { logged.push(line); });

    const io = fakeIo();
    attach(io);
    const party = PartyService.create('u-alice', 'Alice', 'm1', 'Un Film', '/img/m1', null, 'video');
    PartyService.join(party, 'u-bob', 'Bob');
    const bobSocket = fakeSocket('u-bob');
    io.trigger('connection', bobSocket as unknown as AppSocket);
    bobSocket.trigger('join', { partyId: party.id });

    bobSocket.trigger('requestPause');
    bobSocket.trigger('requestPause');
    const pauseRequestLines = logged.filter((l) => l.includes(' pauseRequested '));
    assert.equal(pauseRequestLines.length, 1, 'an immediate repeat must not produce a second line');
  });

  it('requestResume is throttled independently from requestPause (its own 5s window, own counter)', (t) => {
    const logged: string[] = [];
    t.mock.method(console, 'log', (line: string) => { logged.push(line); });

    const io = fakeIo();
    attach(io);
    const party = PartyService.create('u-alice', 'Alice', 'm1', 'Un Film', '/img/m1', null, 'video');
    PartyService.join(party, 'u-bob', 'Bob');
    const bobSocket = fakeSocket('u-bob');
    io.trigger('connection', bobSocket as unknown as AppSocket);
    bobSocket.trigger('join', { partyId: party.id });

    bobSocket.trigger('requestPause');
    bobSocket.trigger('requestResume');
    bobSocket.trigger('requestResume');
    assert.equal(logged.filter((l) => l.includes(' pauseRequested ')).length, 1);
    assert.equal(logged.filter((l) => l.includes(' resumeRequested ')).length, 1, 'an immediate repeat must not produce a second line, and requestPause just above must not have consumed this budget');
  });

  it('the lobby snapshot exposes the host\'s live position for a video party, and null for a music party', () => {
    const io = fakeIo();
    attach(io);
    const videoParty = PartyService.create('u-alice', 'Alice', 'm1', 'Un Film', '/img/m1', null, 'video');
    const musicParty = PartyService.create('u-bob', 'Bob', 't1', 'Track One', '/img/t1', null, 'music');
    MusicQueueService.init(musicParty, { itemId: 't1', itemName: 'Track One', itemImageUrl: '/img/t1', durationTicks: 10_000_000 }, 'u-bob', 'Bob');

    const aliceSocket = fakeSocket('u-alice');
    io.trigger('connection', aliceSocket as unknown as AppSocket);
    aliceSocket.trigger('join', { partyId: videoParty.id });
    aliceSocket.trigger('playstate', { isPaused: false, positionTicks: 50_000_000 }); // 5s in, playing

    const subscriber = fakeSocket('u-eve');
    io.trigger('connection', subscriber as unknown as AppSocket);
    subscriber.trigger('subscribeLobby');

    const parties = subscriber.emitted.find(([event]) => event === 'parties')![1] as LobbyPartyDto[];
    const video = parties.find((p) => p.partyId === videoParty.id)!;
    const music = parties.find((p) => p.partyId === musicParty.id)!;

    assert.equal(video.positionTicks, 50_000_000);
    assert.equal(video.isPaused, false);
    assert.ok(video.positionUpdatedUtc, 'should carry a timestamp so the client can tick from it');

    assert.equal(music.positionTicks, null, 'no single "elapsed" makes sense across a rotating music queue');
    assert.equal(music.isPaused, null);
    assert.equal(music.positionUpdatedUtc, null);
  });

  it('a play/pause transition refreshes the lobby (an unchanged heartbeat resend does not); advancing the queue does too', () => {
    const io = fakeIo();
    attach(io);
    const party = PartyService.create('u-alice', 'Alice', 'ep1', 'S01E01', '/img/ep1', [
      { itemId: 'ep1', itemName: 'S01E01', itemImageUrl: '/img/ep1' },
      { itemId: 'ep2', itemName: 'S01E02', itemImageUrl: '/img/ep2' }
    ], 'video');
    const socket = fakeSocket('u-alice');
    io.trigger('connection', socket as unknown as AppSocket);
    socket.trigger('join', { partyId: party.id });

    const lobbyRefreshCount = () => io.roomEmits.filter((e) => e.room === 'lobby' && e.event === 'parties').length;

    // The default new-member state is isPaused:true (see PartyService's
    // newMember) - becoming "playing" is a real transition, so this
    // should refresh the lobby once.
    socket.trigger('playstate', { isPaused: false, positionTicks: 10_000_000 });
    assert.equal(lobbyRefreshCount(), 1);

    // Same isPaused, only position moving - the player's 3s heartbeat
    // resend (see Player.ts) - must NOT refresh the lobby again, same
    // discipline as the "playstate logs only on a transition" test above.
    socket.trigger('playstate', { isPaused: false, positionTicks: 40_000_000 });
    assert.equal(lobbyRefreshCount(), 1);

    socket.trigger('next');
    assert.equal(lobbyRefreshCount(), 2, 'advancing the queue resets position/pause for the new item - the home page needs that immediately, not just whenever some other event next refreshes it');
  });

  it('a stale connection\'s next heartbeat after the map slot it shared gets deleted does not crash the process (regression - see docs/architecture.md)', (t) => {
    const crashed: unknown[] = [];
    t.mock.method(console, 'error', (..._args: unknown[]) => { crashed.push(_args); });

    const io = fakeIo();
    attach(io);
    const party = PartyService.create('u-alice', 'Alice', 'm1', 'Un Film', '/img/m1', null, 'video');

    const socketA = fakeSocket('u-alice');
    io.trigger('connection', socketA as unknown as AppSocket);
    socketA.trigger('join', { partyId: party.id });

    // A second join for the same userId (a second device/tab, before any
    // real per-device identity - see docs/architecture.md) re-runs
    // PartyService.join, which overwrites the one map slot this userId
    // occupies - orphaning socketA's own member record from the map even
    // though socketA itself is still very much "connected".
    PartyService.join(party, 'u-alice', 'Alice');
    const socketB = fakeSocket('u-alice');
    io.trigger('connection', socketB as unknown as AppSocket);
    socketB.trigger('join', { partyId: party.id });

    // socketB (currently occupying the slot) leaves - the only member, so
    // the party is destroyed; hostUserId is never reassigned since
    // there's no successor to hand off to (see PartyService.leave).
    socketB.trigger('leave');
    assert.equal(PartyService.getById(party.id), null, 'sanity check: the party really is gone');

    // socketA's regular 3s heartbeat fires next, completely unaware any
    // of this happened: its own closure still references the same
    // (now-destroyed) Party object, and its stale userId still equals
    // the stale hostUserId, so the host check alone doesn't catch this -
    // party.members.get(userId) is what has to.
    assert.doesNotThrow(() => socketA.trigger('playstate', { isPaused: true, positionTicks: 0 }));
    assert.equal(crashed.length, 0, 'should be a clean early return (the specific guard), not the generic safe() net silently swallowing an actual throw');
  });
});