watchparty/server/src/realtime/SocketGateway.ts

'use strict';

import { Logger } from '../Logger';
import { MusicQueueService } from '../services/MusicQueueService';
import { PartyService } from '../services/PartyService';
import type { AppServer, LeaveResult, LobbyPartyDto, Party, SocketMemberDto } from '../types';

/**
 * Realtime side of the app: one party's own room, plus a shared "lobby"
 * room that powers the home page's live "watch parties en cours" list.
 *
 * Auth comes from the session cookie sent during the initial HTTP
 * handshake - exposed here via socket.request.headers.cookie.
 *
 * Events (client -> server unless noted) - see types.ts's
 * ClientToServerEvents/ServerToClientEvents for the exact payload shapes:
 *   "subscribeLobby"                        joins the "lobby" room
 *   "parties" (server -> client)            full snapshot, sent on subscribe
 *                                            and again whenever any party changes
 *   "join" {partyId, deviceId, force?}      joins that party's own room -
 *                                            deviceId identifies which physical
 *                                            device/browser this is (see
 *                                            client/src/deviceId.ts); force:true
 *                                            is a confirmed takeover, re-sent
 *                                            after "deviceConflict" below
 *   "deviceConflict" (server -> client, to the joining socket only)
 *                                            this account is already connected
 *                                            from a *different* deviceId - the
 *                                            socket stays connected but inert
 *                                            until the client re-sends "join"
 *                                            with force:true or gives up
 *   "members" (server -> client)            that party's member list
 *
 *   -- mediaType: 'video' parties --
 *   "playstate" {isPaused, positionTicks}   (host only - ignored otherwise)
 *   "playstate" (server -> client, to everyone in the party but the sender)
 *   "next"                                  (host only - advance the queue)
 *   "itemChanged" (server -> client)
 *   "setBurnInSubtitle" {subtitleIndex, audioIndex?}  (host only - picks
 *     or clears the party's *shared* self-hosted-ffmpeg burn-in stream,
 *     for a PGS/image subtitle a native <track> can't render - see
 *     TranscodeService. Setting a real one clears "setTextSubtitle"'s
 *     own state below - mutually exclusive, only one subtitle shows at once.)
 *   "burnInChanged" (server -> client)
 *   "setTextSubtitle" {subtitleIndex}  (host only - picks or clears a
 *     *text* subtitle for the whole party; every member still renders
 *     it themselves as a native <track>, this only syncs *which* one -
 *     no transcoding involved, unlike burn-in above. Setting a real one
 *     clears "setBurnInSubtitle"'s own state - mutually exclusive.)
 *   "textSubtitleChanged" (server -> client)
 *   "setStreamOptions" {audioStreamIndex?, quality?}  (host only - picks
 *     the audio track/quality of the party's *shared* normal Emby stream -
 *     see PartyStreamService. There's only ever one stream per party now
 *     (not one per viewer), so this is necessarily party-wide, not each
 *     viewer's own local pick the way it was before streams were shared.)
 *   "streamOptionsChanged" (server -> client)
 *

 *   -- mediaType: 'music' parties (see MusicQueueService for the model) --
 *   "voteSkip"                              (anyone - see MusicQueueService for the threshold)
 *   "musicPlaystate" {isPaused}             (anyone - no host check, unlike video's "playstate")
 *   "queueUpdated" (server -> client) - see MusicQueueService.getState() for the shape
 *     (adding a track is REST, not a socket event - POST
 *     /api/parties/:partyId/tracks in index.ts - since it needs to work
 *     from the album-browsing page too, which has no live connection to
 *     the party it might be adding into; MusicQueueService's own
 *     "queueUpdated" broadcast fires regardless of which path called it)
 *
 *   -- both --
 *   "hostChanged" (server -> client, to the new host only, no payload)
 *   "requestPause"/"requestResume"          (each throttled per member,
 *                                            independently - see below;
 *                                            video only - a music party has no
 *                                            host to ask, everyone already has
 *                                            direct pause/vote-skip access)
 *   "pauseRequested"/"resumeRequested" (server -> client, everyone in the
 *     party) - a plain signal, same as each other: broadcast and done, no
 *     accept/deny state tracked server-side. The host sees the same
 *     banner/avatar highlight as everyone else and acts (or doesn't) with
 *     their own existing playstate controls - see docs/architecture.md.
 *   "leave"
 *   "errorMessage" (server -> client)
 *     ("error" is a reserved Socket.IO event name - using our own to avoid
 *     colliding with its internal error handling.)
 */
class SocketGatewayClass {
  attach(io: AppServer, parseCookies: (header: string | undefined) => Record<string, string>, getSession: (id: string) => { userId: string; userName: string; embyToken: string } | null): void {
    io.on('connection', (socket) => {
      const cookies = parseCookies(socket.request.headers.cookie);
      const session = cookies.wp_session ? getSession(cookies.wp_session) : null;

      if (!session) {
        socket.emit('errorMessage', { message: 'Session expirée - reconnecte-toi.' });
        socket.disconnect(true);
        return;
      }

      const userId = session.userId;
      let party: Party | null = null;

      socket.on('subscribeLobby', this.safe(() => {
        socket.join('lobby');
        socket.emit('parties', this.lobbySnapshot());
      }));

      socket.on('join', this.safe((msg: { partyId: string; deviceId: string; force?: boolean }) => {
        const p = PartyService.getById(msg?.partyId);
        const member = p?.members.get(userId);
        if (!p || !member) {
          socket.emit('errorMessage', { message: "Tu n'es pas membre de cette party." });
          socket.disconnect(true);
          return;
        }
        const deviceId = msg?.deviceId || null;
        // Same account, a DIFFERENT device already actively connected as
        // this member - ask the new socket to confirm a takeover instead of
        // silently booting the other device. member.deviceId is null until
        // the first real socket "join" ever succeeds (a member just created
        // by the REST join/create routes has no live connection yet), so
        // there's nothing to conflict with the very first time.
        if (member.connection && member.deviceId && deviceId && member.deviceId !== deviceId && !msg?.force) {
          socket.emit('deviceConflict', { partyId: p.id });
          // This socket stays connected but inert: not joined to the party
          // room, member.connection untouched - until the client re-sends
          // "join" with force:true or gives up.
          return;
        }
        // Either a confirmed takeover (force:true) or the same device
        // reconnecting (reload, dropped connection) - boot whatever
        // connection currently holds the slot before taking its place, so
        // ITS eventual "disconnect" can't be mistaken for a real leave (see
        // the connection===socket guard on "leave"/"disconnect" below).
        if (member.connection && member.connection !== socket) {
          member.connection.emit('errorMessage', { message: 'Tu as rejoint depuis un autre appareil.' });
          member.connection.disconnect(true);
        }
        party = p;
        member.connection = socket;
        member.deviceId = deviceId;
        // Re-confirmed on every successful join, not just set once - the
        // periodic Emby-playstate sweep (index.ts) reads this to report
        // this member's own progress under their own Emby account (see
        // PlaystateService).
        member.embyToken = session.embyToken;
        socket.join(party.id);
        this.broadcastMembers(io, party);
      }));

      socket.on('playstate', this.safe((msg: { isPaused: boolean; positionTicks: number }) => {
        if (!party || userId !== party.hostUserId) {
          return; // only the host drives playback - see docs/architecture.md
        }
        // Not just defensive dead code: confirmed as a real crash. Two
        // connections for the same userId (two tabs/devices, no per-device
        // identity yet - see docs/architecture.md) share the one
        // userId-keyed map slot; if the connection currently occupying it
        // is the last member and leaves, the party is destroyed and this
        // slot is gone, but party.hostUserId is never reassigned (there's
        // no successor to hand off to) - so the OTHER, still-connected
        // socket's next 3s heartbeat still passes the host check above
        // (its stale userId still equals the stale hostUserId) and used
        // to crash the whole process on the `!` right here.
        const member = party.members.get(userId);
        if (!member) return;
        const wasPaused = member.isPaused;
        member.isPaused = !!msg?.isPaused;
        member.positionTicks = Number(msg?.positionTicks) || 0;
        member.lastUpdatedUtc = Date.now();
        party.lastActivityUtc = Date.now();

        // Logged (and the lobby refreshed) only on an actual play/pause
        // transition, not every one of these - the host's player also
        // re-sends the current state on a plain 3s heartbeat (see player),
        // which would otherwise spam this line - and every home-page
        // viewer's socket - for every active party every 3 seconds
        // regardless of whether anything actually changed. Between
        // transitions, the home page just ticks its own "elapsed" label
        // locally from the last broadcast position - see Home.ts.
        if (member.isPaused !== wasPaused) {
          const positionSeconds = Math.round(member.positionTicks / 10000000);
          Logger.party(`playstate partyId=${party.id} host=${Logger.who(userId, member.userName)} -> ${member.isPaused ? 'paused' : 'playing'} at ${positionSeconds}s`);
          this.broadcastLobby(io);
        }

        socket.to(party.id).emit('playstate', { isPaused: member.isPaused, positionTicks: member.positionTicks });
      }));

      socket.on('next', this.safe(() => {
        if (!party || userId !== party.hostUserId) {
          return; // only the host can advance the queue
        }
        if (PartyService.advanceQueue(party)) {
          io.to(party.id).emit('itemChanged', {
            itemId: party.itemId,
            itemName: party.itemName,
            itemImageUrl: party.itemImageUrl,
            queueIndex: party.queueIndex,
            queueLength: party.queue.length,
            hasNext: party.queueIndex < party.queue.length - 1
          });
          // advanceQueue() already reset every member's position/pause to
          // (0, false) for the new item - refresh the lobby now so the
          // home page's card picks up the new title/poster AND that reset
          // immediately, rather than only whenever some other event (a
          // join/leave) happens to refresh it next.
          this.broadcastLobby(io);
        }
      }));

      // Host-only, video-only: picks (or clears) the party's *shared*
      // burn-in stream - see TranscodeService and Party.burnInSubtitleIndex's
      // own comment. Deliberately doesn't touch TranscodeService directly
      // here - this only updates party state and broadcasts it, the same
      // way "next" only updates queueIndex and broadcasts "itemChanged"
      // without reaching into CatalogService itself. The actual ffmpeg
      // session starts/restarts lazily, whenever a client's <video> next
      // requests /api/catalog/burn/:partyId/master.m3u8 - see index.ts.
      socket.on('setBurnInSubtitle', this.safe((msg: { subtitleIndex: string | null; audioIndex?: string }) => {
        if (!party || party.mediaType !== 'video' || userId !== party.hostUserId) return;
        party.burnInSubtitleIndex = msg?.subtitleIndex || null;
        party.burnInAudioIndex = party.burnInSubtitleIndex ? (msg?.audioIndex || null) : null;
        // Mutually exclusive with a text pick - only one subtitle can be
        // showing at once, see "setTextSubtitle" below.
        if (party.burnInSubtitleIndex) party.textSubtitleIndex = null;
        party.lastActivityUtc = Date.now();
        Logger.party(`burnInChanged partyId=${party.id} subtitleIndex=${party.burnInSubtitleIndex ?? '(none)'} audioIndex=${party.burnInAudioIndex ?? '(default)'}`);
        io.to(party.id).emit('burnInChanged', { subtitleIndex: party.burnInSubtitleIndex, audioIndex: party.burnInAudioIndex });
      }));

      // Host-only, video-only: picks or clears a *text* subtitle for the
      // whole party - subtitle choice in general is party-wide now, not
      // each viewer's own local pick (see docs/architecture.md - that
      // was the original design, but turned out to be the wrong call
      // once people actually watched together). Every member still
      // renders it themselves as a native <track> - only *which* one is
      // synced here, no transcoding/stream change involved, unlike
      // "setBurnInSubtitle" above.
      socket.on('setTextSubtitle', this.safe((msg: { subtitleIndex: string | null }) => {
        if (!party || party.mediaType !== 'video' || userId !== party.hostUserId) return;
        party.textSubtitleIndex = msg?.subtitleIndex || null;
        // Mutually exclusive with a burn-in pick - clearing it here is
        // what actually drops the party back onto its normal shared
        // stream (PartyStreamService) once a real text index is picked.
        if (party.textSubtitleIndex) {
          party.burnInSubtitleIndex = null;
          party.burnInAudioIndex = null;
        }
        party.lastActivityUtc = Date.now();
        Logger.party(`textSubtitleChanged partyId=${party.id} subtitleIndex=${party.textSubtitleIndex ?? '(none)'}`);
        io.to(party.id).emit('textSubtitleChanged', { subtitleIndex: party.textSubtitleIndex });
      }));

      // Host-only, video-only: picks the audio track/quality of the
      // party's *shared* normal Emby stream - see PartyStreamService.
      // Deliberately doesn't touch PartyStreamService directly here -
      // same reasoning as "setBurnInSubtitle" above not touching
      // TranscodeService: this only updates party state and broadcasts
      // it; PartyStreamService.getOrNegotiate already compares against
      // the party's *current* values on every request, so the next one
      // naturally renegotiates once these differ from whatever's cached.
      socket.on('setStreamOptions', this.safe((msg: { audioStreamIndex?: string | null; quality?: string | null }) => {
        if (!party || party.mediaType !== 'video' || userId !== party.hostUserId) return;
        party.streamAudioIndex = msg?.audioStreamIndex || null;
        party.streamQuality = msg?.quality || null;
        party.lastActivityUtc = Date.now();
        Logger.party(`streamOptionsChanged partyId=${party.id} audioStreamIndex=${party.streamAudioIndex ?? '(default)'} quality=${party.streamQuality ?? '(auto)'}`);
        io.to(party.id).emit('streamOptionsChanged', { audioStreamIndex: party.streamAudioIndex, quality: party.streamQuality });
      }));

      // Majority of currently CONNECTED members, not everyone who's ever
      // joined - see MusicQueueService.
      socket.on('voteSkip', this.safe(() => {
        const member = party?.mediaType === 'music' ? party.members.get(userId) : undefined;
        if (!party || !member) return;
        const connectedCount = Array.from(party.members.values()).filter((m) => m.connection).length;
        MusicQueueService.voteSkip(party, userId, member.userName, connectedCount);
        this.broadcastLobby(io);
      }));

      // No host check, unlike video's "playstate" - see MusicQueueService.
      socket.on('musicPlaystate', this.safe((msg: { isPaused: boolean }) => {
        if (!party || party.mediaType !== 'music' || !party.members.get(userId)) return;
        MusicQueueService.setPaused(party, !!msg?.isPaused);
      }));

      // Anyone (not just followers - harmless either way) can ask the host
      // to pause; the presence overlay (the video player) is what triggers
      // this. Broadcast to the whole party, not just the host, so everyone
      // sees why playback is about to pause. Throttled per member
      // server-side (not just the client's own cooldown, which a stray
      // extra click - or anyone poking the socket directly - would
      // bypass). Video only: a music party has no host to ask - everyone
      // already has direct pause access via "musicPlaystate".
      socket.on('requestPause', this.safe(() => {
        if (!party || party.mediaType === 'music') return;
        const member = party.members.get(userId);
        if (!member) return;
        const now = Date.now();
        if (member.lastPauseRequestUtc && now - member.lastPauseRequestUtc < 5000) return;
        member.lastPauseRequestUtc = now;
        Logger.party(`pauseRequested partyId=${party.id} user=${Logger.who(userId, member.userName)}`);
        io.to(party.id).emit('pauseRequested', { userId, userName: member.userName });
      }));

      // Same shape as "requestPause" above, independently throttled (its
      // own lastResumeRequestUtc, not shared) - the mini "request..."
      // toolbar in the player (views/player/Player.ts) offers both.
      socket.on('requestResume', this.safe(() => {
        if (!party || party.mediaType === 'music') return;
        const member = party.members.get(userId);
        if (!member) return;
        const now = Date.now();
        if (member.lastResumeRequestUtc && now - member.lastResumeRequestUtc < 5000) return;
        member.lastResumeRequestUtc = now;
        Logger.party(`resumeRequested partyId=${party.id} user=${Logger.who(userId, member.userName)}`);
        io.to(party.id).emit('resumeRequested', { userId, userName: member.userName });
      }));

      socket.on('leave', this.safe(() => {
        // Only actually leave if THIS socket is still the connection the
        // member map has on file. A device takeover (see "join" above)
        // reassigns member.connection to the new socket and force-
        // disconnects the old one; without this guard the old socket's own
        // "leave"/"disconnect" would incorrectly end the NEW connection's
        // membership (wrong host handoff, party destroyed under a still-
        // present member, etc).
        if (party && party.members.get(userId)?.connection === socket) {
          this.applyLeave(io, party, userId);
        }
        party = null;
      }));

      // Fires on a clean "leave", a dead-connection timeout (Socket.IO's
      // own heartbeat - see docs/architecture.md), a device takeover's
      // forced disconnect of the OLD connection, or any other way the
      // socket goes away - the one cleanup path that always runs, unlike
      // relying on the client to always get a chance to emit "leave" first.
      // Same stale-connection guard as "leave" above.
      socket.on('disconnect', this.safe(() => {
        if (party && party.members.get(userId)?.connection === socket) {
          this.applyLeave(io, party, userId);
        }
      }));
    });
  }

  /**
   * Wraps a socket event handler so an unexpected exception logs and gets
   * swallowed instead of crashing the whole process. Confirmed as a real
   * failure mode, not theoretical: Socket.IO's own internals have no
   * equivalent of Express's error-handling middleware (which asyncRoute()
   * in index.ts already routes every REST handler's rejection/throw
   * through) - a synchronous throw from a listener here propagates
   * straight out of `Socket.emit()` and takes the entire Node process
   * down with it, every active party, not just the one connection at
   * fault. See docs/architecture.md for the specific bug this was added
   * after (a stale connection's next heartbeat reading a since-deleted
   * party.members entry).
   */
  private safe<A extends unknown[]>(handler: (...args: A) => void): (...args: A) => void {
    return (...args: A) => {
      try {
        handler(...args);
      } catch (err) {
        console.error('[watchparty] socket handler crashed', err);
      }
    };
  }

  /**
   * Removes `userId` from `party` and handles everything that follows from
   * it: if they were the host, the new host (if any) gets a personal
   * "hostChanged" so their player can promote itself without a page
   * reload; either way, the party's own room and the lobby both get
   * refreshed. Shared by the socket "leave"/"disconnect" handlers above
   * and index.ts's REST leave endpoint so this logic - especially the
   * host handoff - only has to be right in one place.
   */
  applyLeave(io: AppServer, party: Party, userId: string): LeaveResult {
    const result = PartyService.leave(party, userId);
    if (!result.destroyed) {
      this.broadcastMembers(io, party);
      if (result.hostChanged && result.newHostUserId) {
        const newHost = party.members.get(result.newHostUserId);
        newHost?.connection?.emit('hostChanged');
      }
    }
    this.broadcastLobby(io);
    return result;
  }

  broadcastLobby(io: AppServer): void {
    io.to('lobby').emit('parties', this.lobbySnapshot());
  }

  private broadcastMembers(io: AppServer, party: Party): void {
    const members: SocketMemberDto[] = Array.from(party.members.values()).map((m) => ({
      userId: m.userId,
      userName: m.userName,
      isHost: m.userId === party.hostUserId,
      connected: !!m.connection
    }));
    io.to(party.id).emit('members', members);
  }

  private lobbySnapshot(): LobbyPartyDto[] {
    return PartyService.listActive().map((party) => {
      const host = party.members.get(party.hostUserId);
      // Video only - see LobbyPartyDto's own comment for why music parties
      // get null here and fall back to createdUtc client-side instead.
      // The host is authoritative for position/pause the same way they are
      // inside the party's own room (see "playstate" below) - no separate
      // "party-level" position is tracked anywhere.
      const isVideo = party.mediaType === 'video';
      return {
        partyId: party.id,
        mediaType: party.mediaType,
        itemId: party.itemId,
        itemName: party.itemName,
        itemImageUrl: party.itemImageUrl,
        hostUserName: host ? host.userName : null,
        memberCount: party.members.size,
        createdUtc: party.createdUtc, // client computes/ticks elapsed time itself from this
        positionTicks: isVideo && host ? host.positionTicks : null,
        isPaused: isVideo && host ? host.isPaused : null,
        positionUpdatedUtc: isVideo && host ? host.lastUpdatedUtc : null
      };
    });
  }
}

export const SocketGateway = new SocketGatewayClass();