watchparty/server/test/socketGateway.test.ts
Nicolas 11f003d269
All checks were successful
Publier l'image Docker / test (push) Successful in 19s
Publier l'image Docker / publish (push) Successful in 16s
Sync watch-party progress back to Emby (PlaystateService)
This app has been read/proxy-only toward Emby so far. Queued right
after the playlist feature, asked for directly: while someone's in a
party, push their progress to Emby's own PlaystateService, so their own
Emby library (resume position, last-played state) stays in sync with
the watch/listen party, under their own account - not just the host's,
and not just this app's own memory of it.

Scoped to periodic progress reporting only for v1 - no explicit
start/stop calls (Emby's own "now playing" list won't show this app's
streams, and "mark fully watched" isn't guaranteed the way a clean Stop
near 100% would trigger it). Steady Progress calls alone already cover
the actual ask; wiring clean start/stop into every transition (item
change, leave, party destroy, a device takeover) is real complexity for
a secondary payoff - left as a documented follow-up.

- server/src/types.ts: PartyMember gains embyToken (null until known,
  same shape as this session's own deviceId addition) - never sent to
  the client, toPublicState's member mapping doesn't grow it.
- server/src/emby/EmbyClient.ts: fetch() now optionally takes
  method/body (JSON-encoded once, here) - every existing GET-only call
  site unaffected, both are optional.
- server/src/services/PlaystateService.ts (new): reportProgress(token,
  userId, itemId, positionTicks, isPaused) - POSTs to
  /Users/{userId}/PlayingItems/{itemId}/Progress. Catches and logs its
  own failures (new Logger.playstate) rather than throwing - this runs
  on a timer with nothing to catch a rejection, and one bad token or a
  transient Emby hiccup must never touch real playback.
- server/src/services/PartyService.ts: new setMemberEmbyToken (records
  a member's own token) and getCurrentPlaybackState (video: the host's
  own tracked position - there's no per-follower tracking today, a
  reasonable approximation given the player's own drift correction;
  music: the one shared, server-authoritative position every member
  already sees the same way). The one real piece of logic here is pure
  and unit-tested on its own, same shape as this session's
  absoluteToRelativeIndex/splitPlaylistByMediaType.
- server/src/realtime/SocketGateway.ts: attach()'s getSession callback
  widened to expose embyToken - it was already being passed the full
  EmbySession (AuthService.getSession), just typed down to {userId,
  userName} before anything downstream could read the rest. The "join"
  handler now also records member.embyToken alongside deviceId.
- server/src/index.ts: setMemberEmbyToken called after each of the 4
  REST create/join call sites, so a member has a valid token from the
  moment they're a member, not only once their socket first connects.
  New ~20s setInterval (same shape as the existing removeIdleParties/
  TranscodeService.reapIdleSessions timers): for every active party,
  getCurrentPlaybackState once, then reportProgress for every connected
  member with a known token.
- server/test/embyClient.test.ts, partyService.test.ts,
  playstateService.test.ts (new): the new body-encoding path (and that
  a plain GET stays untouched), getCurrentPlaybackState's video/music/
  empty branches, setMemberEmbyToken, and that reportProgress's request
  shape is right and a thrown failure never propagates.
- docs/architecture.md, README.md: design + the deliberate v1 scope and
  the follower-position approximation spelled out.

No client changes - this is a pure server-side background process with
nothing for the browser to render or react to.

Verified: full typecheck+test+build on both projects (71/71 server
tests, 9 new). No Emby server is reachable from this dev environment,
so the actual endpoint/body shape is unverified end-to-end - flagged
explicitly in the docs, the third such Emby-API assumption this session
(after the ffmpeg direct-stream URL and the Playlists endpoint, both
fine once actually tried).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-09-14 19:21:16 +02:00

421 lines
21 KiB
TypeScript

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