'use strict'; import crypto from 'crypto'; import { Logger } from '../Logger'; import { MusicQueueService } from './MusicQueueService'; import { TranscodeService } from './TranscodeService'; import { PartyStreamService } from './PartyStreamService'; import { PlaystateService } from './PlaystateService'; import { CastTokenService } from './CastTokenService'; import { EmbySessionService } from './EmbySessionService'; import { StaticPartyService } from './StaticPartyService'; import type { LeaveResult, MediaType, Party, PartyMember, PartyPublicState, QueueItem } from '../types'; /** * In-memory party store - mirrors the Party/PartyMember design from the * earlier Emby-plugin attempt (see git history / docs/architecture.md), * adapted for a model where we own the player: for a video party, one host * drives playback and everyone else's client just applies whatever the * host reports (a music party has no host for playback at all - see * MusicQueueService). Nothing is persisted to disk; a server restart * clears every party, which is fine - a party is meant to live for one * watch/listen session, not persist for weeks. * * No join codes: the home page's live "watch parties en cours" list (see * SocketGateway's lobby broadcast) is the only way to find a party, and * since that's real-time, a code that has to be copied/typed out of band * doesn't add anything - see docs/architecture.md. * * Singleton (exported as an instance below), same shape as * MusicQueueService/AuthService. */ class PartyServiceClass { private partiesById = new Map(); /** * `queue` is the ordered list of {itemId, itemName, itemImageUrl} this * party will play through - for a movie/standalone episode it's just the * one item; for "start watching this series from here", CatalogService * builds the rest of the season/series into it. `itemId`/`itemName`/ * `itemImageUrl` always mirror queue[queueIndex] - kept as top-level * fields too since that's what most of the UI reads. Unused for * `mediaType: 'music'` parties (still populated with a one-item fallback * so those same top-level fields have *something* sane in them for the * lobby/home-page card) - a music party's real, mutable queue is * `party.music`, owned by MusicQueueService, not this fixed pre-built one. */ create( hostUserId: string, hostUserName: string, itemId: string, itemName: string, itemImageUrl: string, queue: QueueItem[] | null, mediaType: MediaType ): Party { const party: Party = { id: crypto.randomUUID(), mediaType: mediaType || 'video', itemId, itemName, itemImageUrl, queue: queue && queue.length ? queue : [{ itemId, itemName, itemImageUrl }], queueIndex: 0, hostUserId, createdUtc: Date.now(), lastActivityUtc: Date.now(), members: new Map(), burnInSubtitleIndex: null, burnInAudioIndex: null, textSubtitleIndex: null, streamAudioIndex: null, streamQuality: null }; party.members.set(hostUserId, this.newMember(hostUserId, hostUserName)); this.partiesById.set(party.id, party); Logger.party(`created partyId=${party.id} mediaType=${party.mediaType} host=${Logger.who(hostUserId, hostUserName)} item="${itemName}" queueLength=${party.queue.length}`); return party; } getById(partyId: string): Party | null { return this.partiesById.get(partyId) || null; } /** * Idempotent for a userId already in `party`: returns the *existing* * member as-is rather than resetting it. This matters because the REST * "Rejoindre" flow calls this every time regardless of whether the * caller is already a member - re-clicking it, a page reload, a second * device/tab, a stale retry. Blindly overwriting used to reset * connection/isPaused/deviceId on every one of those, which is exactly * what let two connections for one userId silently share (and lose * track of) one map slot - see docs/architecture.md's server-crash * section. A genuinely new member still goes through here fresh; only * the "already present" case is now a no-op. Device-aware conflict * handling for an *already-connected* member lives in SocketGateway's * "join" handler instead - this method has no socket to reason about. */ join(party: Party, userId: string, userName: string): PartyMember { const existing = party.members.get(userId); if (existing) return existing; const member = this.newMember(userId, userName); party.members.set(userId, member); party.lastActivityUtc = Date.now(); Logger.party(`joined partyId=${party.id} user=${Logger.who(userId, userName)} memberCount=${party.members.size}`); return member; } /** * Records `userId`'s own Emby access token on their member entry - * called right after every REST create/join call site (index.ts) and * re-confirmed on every socket "join" (SocketGateway), so the periodic * Emby-playstate sweep (index.ts/PlaystateService) can always report * *their own* progress under *their own* Emby account. A no-op if * they're not (or no longer) a member - defensive only, every real * call site has just created/found them. */ setMemberEmbyToken(party: Party, userId: string, embyToken: string): void { const member = party.members.get(userId); if (member) member.embyToken = embyToken; } /** * What's currently playing for this party, for the Emby-playstate * sweep (index.ts) - null if there's genuinely nothing to report * (an empty music queue, or - shouldn't happen - no resolvable host). * Video reads the host's own tracked position/pause state (there's no * per-*follower* tracking today - see docs/architecture.md's own * "Approximation" note on why every member's report uses this same * value); music reads the one shared, server-authoritative position * every member already sees the same way (MusicQueueService). */ getCurrentPlaybackState(party: Party): { itemId: string; positionTicks: number; isPaused: boolean } | null { if (party.mediaType === 'music') { const current = party.music?.current; if (!current) return null; return { itemId: current.track.itemId, positionTicks: MusicQueueService.getElapsedSeconds(party) * 10000000, isPaused: current.isPaused }; } const host = party.members.get(party.hostUserId); if (!host) return null; return { itemId: party.itemId, positionTicks: host.positionTicks, isPaused: host.isPaused }; } /** * Whether a request claiming to be `deviceId` is stale for `member` - * already replaced by a takeover on a *different* device (see * SocketGateway's "join" handler). Used by the REST /leave route * (index.ts), which - unlike the socket "leave"/"disconnect" handlers, * guarded by comparing `member.connection` against the live socket * itself - has no connection to compare against instead. Without this, * a device already superseded by a takeover elsewhere could still call * /leave and end the CURRENT device's membership by mistake - confirmed * as the real cause of a "both devices end up on the session, controls * never transfer" report: the stale device's own client didn't react to * being disconnected at all (see Player.ts/MusicBar.ts's onClose), so * its still-rendered "Quitter" button stayed clickable and reached here. * * Only ever returns true on positive evidence of a mismatch (both sides * known and different) - a member with no recorded deviceId yet, or a * caller that didn't send one, is never treated as stale, mirroring the * socket handlers' own "nothing to conflict with yet" reasoning for a * null deviceId. */ isStaleDevice(member: PartyMember | undefined, deviceId: string | null): boolean { return !!(member?.deviceId && deviceId && member.deviceId !== deviceId); } /** * Removes a member. If they were the host, promotes whoever's left * (arbitrary but deterministic-ish - good enough for v1); if nobody's * left, the party is destroyed outright. Returns what happened so * callers (SocketGateway's applyLeave) can notify the new host and * refresh the home page's live list - see docs/architecture.md. */ leave(party: Party, userId: string): LeaveResult { const leavingMember = party.members.get(userId); const wasHost = party.hostUserId === userId; party.members.delete(userId); party.lastActivityUtc = Date.now(); MusicQueueService.removeVote(party, userId); // no-op for a video party (party.music is unset) CastTokenService.revokeForMember(party.id, userId); // no-op if they never cast - see its own comment for why this runs unconditionally, not just on an explicit "stop casting" // No-op if they never had a live Emby Sessions-API entry (never // connected, or the periodic sweep hadn't reached them yet) - see // EmbySessionService's own comment. Fire-and-forget, same as every // other per-member Emby report in this app - it already swallows its // own failures, and leaving the party must never wait on Emby. if (leavingMember?.embyToken && leavingMember.embySession) { const s = leavingMember.embySession; EmbySessionService.stop(leavingMember.embyToken, s.itemId, s.mediaSourceId, s.playSessionId, leavingMember.positionTicks); } let newHostUserId: string | null = party.hostUserId; if (wasHost) { const next = party.members.keys().next(); newHostUserId = next.done ? null : next.value; // Only reassign when there's a real successor - if nobody's left, // the party is destroyed immediately below and this stale value is // never read again, so there's no reason to let it go null. if (newHostUserId) party.hostUserId = newHostUserId; } const destroyed = party.members.size === 0; const whoLeft = leavingMember ? Logger.who(userId, leavingMember.userName) : userId; if (destroyed) { this.partiesById.delete(party.id); MusicQueueService.destroy(party); // clears the auto-advance timer, if any - see its own comment TranscodeService.endSession(party.id); // no-op if this party never had a burn-in stream running PartyStreamService.endSession(party.id); // same - no-op if this party never negotiated a shared stream if (party.staticLink) StaticPartyService.onPartyDestroyed(party.staticLink, party.id); Logger.party(`destroyed partyId=${party.id} (last member ${whoLeft} left)`); } else if (wasHost && newHostUserId != null) { const newHost = party.members.get(newHostUserId)!; Logger.party(`left partyId=${party.id} user=${whoLeft} memberCount=${party.members.size} hostHandoff->${Logger.who(newHostUserId, newHost.userName)}`); } else { Logger.party(`left partyId=${party.id} user=${whoLeft} memberCount=${party.members.size}`); } return { hostChanged: wasHost && newHostUserId != null, newHostUserId, destroyed }; } /** * Moves to the next item in the (fixed, video) queue - e.g. next * episode. Returns false if there's nothing after the current position. * Resets every member's known position/pause state, since it's a new * item playing from zero. */ advanceQueue(party: Party): boolean { if (party.queueIndex >= party.queue.length - 1) { return false; } // Whatever was just playing is "done" - before the party's own state // moves on. See completeCurrentItem's own comment. this.completeCurrentItem(party); party.queueIndex += 1; const next = party.queue[party.queueIndex]!; party.itemId = next.itemId; party.itemName = next.itemName; party.itemImageUrl = next.itemImageUrl; for (const member of party.members.values()) { member.positionTicks = 0; member.isPaused = false; member.lastUpdatedUtc = Date.now(); } // A new item's own MediaStreams indices have nothing to do with the // old one's - carrying a burn-in or text subtitle pick across would // show whatever track happens to share that index on the new item, // not the same subtitle; same reasoning for streamAudioIndex (an // audio *track* index is just as item-specific). streamQuality is // left alone - that's a bandwidth/quality preference, not tied to // any one item, and already persisted across an item change even // before streams were shared (qualitySelect was never rebuilt // per-item either). TranscodeService/PartyStreamService.endSession // also release the old item's now-irrelevant ffmpeg process or // negotiated Emby target immediately rather than waiting for their // idle reapers - see index.ts's own comment on why those are a // backstop, not the normal path. party.burnInSubtitleIndex = null; party.burnInAudioIndex = null; party.textSubtitleIndex = null; party.streamAudioIndex = null; TranscodeService.endSession(party.id); PartyStreamService.endSession(party.id); party.lastActivityUtc = Date.now(); Logger.party(`advanced partyId=${party.id} to item="${next.itemName}" (${party.queueIndex + 1}/${party.queue.length})`); return true; } /** * Marks the party's current item as "done": the party moving on from it * (advanceQueue) or the last queued item playing to its end (the host's * "itemEnded" - see SocketGateway) is this app's own signal for that - * see PlaystateService.markPlayed's own comment for why it's a * heuristic, not a real "watched it all" check, and why every * currently-connected member gets their own account updated, not just * the host. Fire-and-forget, same as the periodic position sync in * index.ts - a member's bad token or a transient Emby hiccup here must * never block the party from actually moving on. * * For a session of a static watch party, also records the item as * completed there (StaticPartyService.onItemCompleted) for every * connected member. Idempotent per item (`lastCompletedItemId`), so an * "itemEnded" and a later "next" on the same item never count twice. */ completeCurrentItem(party: Party): void { const itemId = party.itemId; if (party.lastCompletedItemId === itemId) return; party.lastCompletedItemId = itemId; const connected = Array.from(party.members.values()).filter((m) => !!m.connection); for (const member of connected) { if (!member.embyToken) continue; PlaystateService.markPlayed(member.embyToken, member.userId, itemId); } if (party.staticLink) { StaticPartyService.onItemCompleted( party.staticLink, itemId, connected.map((m) => ({ userId: m.userId, userName: m.userName })) ); } } /** All active parties - powers the "watch parties en cours" home section. */ listActive(): Party[] { return Array.from(this.partiesById.values()); } removeIdleParties(idleTimeoutMs: number): void { const cutoff = Date.now() - idleTimeoutMs; for (const party of this.partiesById.values()) { if (party.lastActivityUtc < cutoff) { this.partiesById.delete(party.id); MusicQueueService.destroy(party); // same leak this class's own destroy path guards against TranscodeService.endSession(party.id); // same leak - see the "leave" destroy path above PartyStreamService.endSession(party.id); if (party.staticLink) StaticPartyService.onPartyDestroyed(party.staticLink, party.id); // Same reasoning, for any member whose own "leave" (a clean // disconnect) never fired for whatever reason - a real gap would // otherwise leave a stale entry sitting in Emby's own dashboard // indefinitely, not just until the next sweep tick the way a // still-connected member's stale Progress report would. for (const member of party.members.values()) { if (member.embyToken && member.embySession) { const s = member.embySession; EmbySessionService.stop(member.embyToken, s.itemId, s.mediaSourceId, s.playSessionId, member.positionTicks); } } const idleMinutes = Math.round((Date.now() - party.lastActivityUtc) / 60000); Logger.party(`destroyed partyId=${party.id} (idle ${idleMinutes}min, ${party.members.size} member(s) still connected)`); } } } /** * `viewerUserId` lets the response say "youAreHost" instead of the * client having to know its own Emby user id to figure that out itself - * now that auth is a server-side session (see AuthService), the client * doesn't hold an id to compare with at all. */ toPublicState(party: Party, viewerUserId: string | null): PartyPublicState { const host = party.members.get(party.hostUserId); return { partyId: party.id, mediaType: party.mediaType, itemId: party.itemId, itemName: party.itemName, itemImageUrl: party.itemImageUrl, createdUtc: party.createdUtc, // hasNext/queueLength/queueIndex describe the fixed video episode // queue - meaningless for a music party (see `music` below instead). hasNext: party.queueIndex < party.queue.length - 1, queueLength: party.queue.length, queueIndex: party.queueIndex, hostUserName: host ? host.userName : null, memberCount: party.members.size, youAreHost: viewerUserId != null && viewerUserId === party.hostUserId, members: Array.from(party.members.values()).map((m) => ({ userId: m.userId, userName: m.userName, isHost: m.userId === party.hostUserId, isPaused: m.isPaused, positionTicks: m.positionTicks, connected: !!m.connection })), music: party.mediaType === 'music' ? MusicQueueService.getState(party) : null, burnInSubtitleIndex: party.burnInSubtitleIndex, burnInAudioIndex: party.burnInAudioIndex, textSubtitleIndex: party.textSubtitleIndex, streamAudioIndex: party.streamAudioIndex, streamQuality: party.streamQuality }; } private newMember(userId: string, userName: string): PartyMember { return { userId, userName, connection: null, deviceId: null, embyToken: null, isPaused: true, positionTicks: 0, lastUpdatedUtc: Date.now(), embySession: null }; } } export const PartyService = new PartyServiceClass();