watchparty/server/src/services/PartyService.ts

'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 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<string, Party>();

  /**
   * `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 };
  }

  /**
   * 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
      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;
    }
    const outgoingItemId = party.itemId; // whatever was just playing - see the markPlayed loop below
    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();
    }
    // The party moving on from `outgoingItemId` is this app's own signal
    // that it's "done" - see PlaystateService.markPlayed's own comment
    // for why that'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 to the next item.
    for (const member of party.members.values()) {
      if (!member.connection || !member.embyToken) continue;
      PlaystateService.markPlayed(member.embyToken, member.userId, outgoingItemId);
    }
    // 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;
  }

  /** 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);
        // 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();