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, embyToken: `tok-${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 a device takeover destroys the party 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, deviceId: 'device-A' });

    // A confirmed device takeover (see the "join" handler below) hands
    // socketA's map slot to socketB - the exact "two connections for the
    // same userId share one slot" shape the original crash needed, now
    // reached through the real takeover flow instead of a raw
    // PartyService.join() call (that method is idempotent now - see its
    // own test in partyService.test.ts - so a second call to it would no
    // longer reproduce this).
    const socketB = fakeSocket('u-alice');
    io.trigger('connection', socketB as unknown as AppSocket);
    socketB.trigger('join', { partyId: party.id, deviceId: 'device-B', force: true });

    // 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');
  });

  it('a second device for the same account gets a deviceConflict instead of silently taking over', () => {
    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, deviceId: 'device-A' });

    const socketB = fakeSocket('u-alice');
    io.trigger('connection', socketB as unknown as AppSocket);
    const roomEmitsBefore = io.roomEmits.length;
    socketB.trigger('join', { partyId: party.id, deviceId: 'device-B' });

    assert.ok(socketB.emitted.some(([event]) => event === 'deviceConflict'), 'the new device should be told about the conflict');
    assert.equal(party.members.get('u-alice')?.connection, socketA, 'the existing connection must stay in place until the conflict is confirmed');
    assert.equal(io.roomEmits.length, roomEmitsBefore, 'a conflicting join must not broadcast "members" - the socket never actually joined');
    assert.ok(!socketA.emitted.some(([event]) => event === 'errorMessage'), 'the existing device must not be kicked yet');
  });

  it('re-joining with force:true confirms the takeover: kicks the old device, attaches the new one', () => {
    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, deviceId: 'device-A' });

    const socketB = fakeSocket('u-alice');
    io.trigger('connection', socketB as unknown as AppSocket);
    socketB.trigger('join', { partyId: party.id, deviceId: 'device-B' }); // conflict - the client would show a confirm() here
    socketB.trigger('join', { partyId: party.id, deviceId: 'device-B', force: true }); // user confirmed

    assert.ok(socketA.emitted.some(([event]) => event === 'errorMessage'), 'the old device should be told it got disconnected');
    assert.equal(party.members.get('u-alice')?.connection, socketB, 'the new device now holds the slot');
    assert.equal(party.members.get('u-alice')?.deviceId, 'device-B');

    // The old connection's own eventual "disconnect" - Socket.IO fires this
    // automatically after a server-side .disconnect(true), simulated here
    // manually like this file's other tests - must NOT be mistaken for a
    // real leave: it no longer holds the member's connection slot (see the
    // "join" handler's own comment on why, and the guard on "leave"/
    // "disconnect" below).
    socketA.trigger('disconnect');
    assert.equal(PartyService.getById(party.id), party, 'the party must survive the superseded connection\'s disconnect');
    assert.ok(party.members.has('u-alice'), 'the member must still be present, still attached to socketB');
  });

  it('the same device reconnecting (matching deviceId) takes over silently - no deviceConflict', () => {
    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, deviceId: 'device-A' });

    // e.g. a page reload on the same browser: a fresh socket, but the same
    // deviceId persisted in localStorage (see client/src/deviceId.ts).
    const socketA2 = fakeSocket('u-alice');
    io.trigger('connection', socketA2 as unknown as AppSocket);
    socketA2.trigger('join', { partyId: party.id, deviceId: 'device-A' });

    assert.ok(!socketA2.emitted.some(([event]) => event === 'deviceConflict'), 'the same device should never conflict with itself');
    assert.equal(party.members.get('u-alice')?.connection, socketA2);
    assert.equal(party.members.get('u-alice')?.deviceId, 'device-A');
  });

  it('setBurnInSubtitle is host-only and a no-op for anyone else (same pattern as "next")', () => {
    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, deviceId: 'device-bob' });

    bobSocket.trigger('setBurnInSubtitle', { subtitleIndex: '4', audioIndex: '2' });

    assert.equal(party.burnInSubtitleIndex, null, 'a non-host request must not change party state');
    assert.equal(party.burnInAudioIndex, null);
    assert.ok(!io.roomEmits.some((e) => e.event === 'burnInChanged'), 'and must not broadcast anything');
  });

  it('setBurnInSubtitle from the host updates the party and broadcasts burnInChanged to the room', () => {
    const io = fakeIo();
    attach(io);
    const party = PartyService.create('u-alice', 'Alice', 'm1', 'Un Film', '/img/m1', null, 'video');
    const aliceSocket = fakeSocket('u-alice');
    io.trigger('connection', aliceSocket as unknown as AppSocket);
    aliceSocket.trigger('join', { partyId: party.id, deviceId: 'device-alice' });

    aliceSocket.trigger('setBurnInSubtitle', { subtitleIndex: '4', audioIndex: '2' });

    assert.equal(party.burnInSubtitleIndex, '4');
    assert.equal(party.burnInAudioIndex, '2');
    const broadcast = io.roomEmits.find((e) => e.room === party.id && e.event === 'burnInChanged');
    assert.deepEqual(broadcast?.payload, { subtitleIndex: '4', audioIndex: '2' });

    // Clearing it (a text subtitle/"Aucun" pick) resets both fields
    // together, never leaving a stray audioIndex behind with no subtitle.
    aliceSocket.trigger('setBurnInSubtitle', { subtitleIndex: null });
    assert.equal(party.burnInSubtitleIndex, null);
    assert.equal(party.burnInAudioIndex, null);
  });

  it('setBurnInSubtitle is a no-op on a music party (no host to drive it, no video stream either)', () => {
    const io = fakeIo();
    attach(io);
    const party = PartyService.create('u-alice', 'Alice', 't1', 'Track One', '/img/t1', null, 'music');
    const aliceSocket = fakeSocket('u-alice');
    io.trigger('connection', aliceSocket as unknown as AppSocket);
    aliceSocket.trigger('join', { partyId: party.id, deviceId: 'device-alice' });

    aliceSocket.trigger('setBurnInSubtitle', { subtitleIndex: '4' });

    assert.equal(party.burnInSubtitleIndex, null);
    assert.ok(!io.roomEmits.some((e) => e.event === 'burnInChanged'));
  });

  it('setStreamOptions is host-only and a no-op for anyone else (same pattern as "next")', () => {
    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, deviceId: 'device-bob' });

    bobSocket.trigger('setStreamOptions', { audioStreamIndex: '2', quality: '4000000' });

    assert.equal(party.streamAudioIndex, null, 'a non-host request must not change party state');
    assert.equal(party.streamQuality, null);
    assert.ok(!io.roomEmits.some((e) => e.event === 'streamOptionsChanged'), 'and must not broadcast anything');
  });

  it('setStreamOptions from the host updates the party and broadcasts streamOptionsChanged to the room', () => {
    const io = fakeIo();
    attach(io);
    const party = PartyService.create('u-alice', 'Alice', 'm1', 'Un Film', '/img/m1', null, 'video');
    const aliceSocket = fakeSocket('u-alice');
    io.trigger('connection', aliceSocket as unknown as AppSocket);
    aliceSocket.trigger('join', { partyId: party.id, deviceId: 'device-alice' });

    aliceSocket.trigger('setStreamOptions', { audioStreamIndex: '2', quality: '4000000' });

    assert.equal(party.streamAudioIndex, '2');
    assert.equal(party.streamQuality, '4000000');
    const broadcast = io.roomEmits.find((e) => e.room === party.id && e.event === 'streamOptionsChanged');
    assert.deepEqual(broadcast?.payload, { audioStreamIndex: '2', quality: '4000000' });

    // Clearing a pick (back to "Emby's own default") resets it to null,
    // independently of the other field.
    aliceSocket.trigger('setStreamOptions', { audioStreamIndex: null, quality: '4000000' });
    assert.equal(party.streamAudioIndex, null);
    assert.equal(party.streamQuality, '4000000');
  });

  it('setStreamOptions is a no-op on a music party (no picker exists there at all)', () => {
    const io = fakeIo();
    attach(io);
    const party = PartyService.create('u-alice', 'Alice', 't1', 'Track One', '/img/t1', null, 'music');
    const aliceSocket = fakeSocket('u-alice');
    io.trigger('connection', aliceSocket as unknown as AppSocket);
    aliceSocket.trigger('join', { partyId: party.id, deviceId: 'device-alice' });

    aliceSocket.trigger('setStreamOptions', { audioStreamIndex: '2' });

    assert.equal(party.streamAudioIndex, null);
    assert.ok(!io.roomEmits.some((e) => e.event === 'streamOptionsChanged'));
  });
});