watchparty/client/src/api/party.ts

// Party REST + realtime - same origin as everything else, auth is the
// session cookie (see embyApi.ts), so no headers/tokens to pass around here.
// Realtime is Socket.IO (the `io` global, loaded via <script> in
// index.html - see global.d.ts) - gives us automatic reconnection and, on
// the server side, heartbeat-based dead-connection detection for free (see
// server/src/realtime/SocketGateway.ts).
//
// No join codes: joining happens straight from the home page's live list
// (connectLobby below) by party id - see docs/architecture.md.
import { getDeviceId } from '../deviceId';
import type { BurnInChangedMsg, ItemChangedMsg, LobbyPartyDto, MusicPublicState, MusicTrackInput, PartyPublicState, PauseRequestedMsg, ResumeRequestedMsg, SocketMember, StreamOptionsChangedMsg, TextSubtitleChangedMsg } from '../types';

async function apiFetch<T>(path: string, options: RequestInit = {}): Promise<T> {
  const res = await fetch(path, {
    credentials: 'same-origin',
    ...options,
    headers: { 'Content-Type': 'application/json', ...(options.headers || {}) }
  });
  if (!res.ok) {
    const body = await res.json().catch(() => ({}));
    throw new Error(body.error || `Request failed: ${res.status}`);
  }
  return res.json() as Promise<T>;
}

export function createParty(itemId: string): Promise<PartyPublicState> {
  return apiFetch('/api/parties', { method: 'POST', body: JSON.stringify({ itemId }) });
}

export function joinParty(partyId: string): Promise<PartyPublicState> {
  return apiFetch(`/api/parties/${encodeURIComponent(partyId)}/join`, { method: 'POST' });
}

export function leaveParty(partyId: string): Promise<{ ok: true }> {
  return apiFetch(`/api/parties/${partyId}/leave`, { method: 'POST' });
}

export function getParty(partyId: string): Promise<PartyPublicState> {
  return apiFetch(`/api/parties/${partyId}`);
}

/**
 * Music only: adds a track to a party's shared queue. REST, not a socket
 * event, deliberately - unlike voteSkip/setMusicPaused below, this needs
 * to work from the album-browsing page (views/album/Album.ts), which
 * holds no live connection to the party at all - only the persistent
 * musicBar/MusicBar.ts does (see its header comment for why browsing and
 * the live party are two separate things now). `track` is whatever
 * ItemDto shape the catalog endpoints already returned - see
 * server/src/services/MusicQueueService.ts.
 */
export function addTrackToQueue(partyId: string, track: MusicTrackInput): Promise<PartyPublicState> {
  return apiFetch(`/api/parties/${partyId}/tracks`, { method: 'POST', body: JSON.stringify(track) });
}

/**
 * Mints a short-lived, party-scoped token a Chromecast can use to reach
 * this party's own shared stream routes, which otherwise only accept the
 * sender's httpOnly session cookie - something a Chromecast can never
 * carry (see server/src/services/CastTokenService.ts). Called right
 * before starting a cast - see cast.ts/Player.ts.
 */
export function issueCastToken(partyId: string): Promise<{ token: string }> {
  return apiFetch(`/api/parties/${partyId}/cast-token`, { method: 'POST' });
}

/** Invalidates whatever cast token this member currently holds for this
 *  party - called when a cast session ends, from either side (the user
 *  stopping it here, or the receiver itself ending it - see
 *  Player.ts/cast.ts's onCastSessionEnded). Harmless if nothing was cast. */
export function revokeCastToken(partyId: string): Promise<{ ok: true }> {
  return apiFetch(`/api/parties/${partyId}/cast-token`, { method: 'DELETE' });
}

/**
 * Live "watch parties en cours" feed for the home page. `onParties` gets
 * called with the full current list on subscribe, and again every time any
 * party anywhere changes (created, someone joined/left, host handed off).
 * Caller MUST call .disconnect() when navigating away (see home's router
 * cleanup) - same leak-prevention pattern as connectParty below.
 */
export function connectLobby(onParties: (parties: LobbyPartyDto[]) => void): { disconnect(): void } {
  const socket = io();
  socket.on('connect', () => socket.emit('subscribeLobby'));
  socket.on('parties', (parties) => onParties(parties as LobbyPartyDto[]));
  return {
    disconnect() {
      socket.disconnect();
    }
  };
}

export interface PartyConnectionCallbacks {
  onMembers?: (members: SocketMember[]) => void;
  onPlaystate?: (msg: { isPaused: boolean; positionTicks: number }) => void;
  onItemChanged?: (msg: ItemChangedMsg) => void;
  onHostChanged?: () => void;
  onPauseRequested?: (msg: PauseRequestedMsg) => void;
  onResumeRequested?: (msg: ResumeRequestedMsg) => void;
  onQueueUpdated?: (state: MusicPublicState) => void;
  /** This account is already actively connected to this party from a
   *  DIFFERENT device (see client/src/deviceId.ts) - the socket is still
   *  connected but inert (not joined to the party) until the caller either
   *  calls confirmDeviceTakeover() or gives up (e.g. navigates away). See
   *  SocketGateway's "join" handler. */
  onDeviceConflict?: () => void;
  /** The host picked (or cleared) the party's shared burn-in stream - see
   *  setBurnInSubtitle below and Player.ts. */
  onBurnInChanged?: (msg: BurnInChangedMsg) => void;
  /** The host picked (or cleared) a *text* subtitle for the whole party -
   *  see setTextSubtitle below and Player.ts. */
  onTextSubtitleChanged?: (msg: TextSubtitleChangedMsg) => void;
  /** The host changed the party's shared stream's audio track/quality -
   *  see setStreamOptions below and Player.ts. */
  onStreamOptionsChanged?: (msg: StreamOptionsChangedMsg) => void;
  onError?: (message: string) => void;
  onClose?: () => void;
}

export interface PartyConnection {
  sendPlaystate(isPaused: boolean, positionTicks: number): void;
  /** Host-only: advance the party's queue (e.g. to the next episode). */
  next(): void;
  /** Asks the host (and tells everyone else) "please pause". Video only -
   *  a music party has no host to ask, see voteSkip/setMusicPaused. */
  requestPause(): void;
  /** Same idea, the other direction - "please resume". Independently
   *  throttled server-side from requestPause (see SocketGateway). */
  requestResume(): void;
  /** Music only: vote to skip the current track - see MusicQueueService
   *  on the server for the majority threshold. */
  voteSkip(): void;
  /** Music only: no host check server-side, unlike sendPlaystate above. */
  setMusicPaused(isPaused: boolean): void;
  /** Re-sends "join" with force:true after the user confirms they want to
   *  disconnect their other device and take over here - call this from
   *  onDeviceConflict's handler once the user says yes. */
  confirmDeviceTakeover(): void;
  /** Host-only: picks the party's shared self-hosted-ffmpeg burn-in
   *  stream for a PGS/image subtitle (null clears it) - see
   *  TranscodeService and Player.ts's subtitle <select>. Not for a text
   *  subtitle - see setTextSubtitle below for those. */
  setBurnInSubtitle(subtitleIndex: string | null, audioIndex?: string): void;
  /** 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. Mutually exclusive with setBurnInSubtitle -
   *  the server clears one when the other is set for real. */
  setTextSubtitle(subtitleIndex: string | null): void;
  /** Host-only, video-only: picks the audio track/quality of the party's
   *  shared normal stream (PartyStreamService) - there's only ever one
   *  stream now, so this is necessarily party-wide, not each viewer's
   *  own local pick the way it was before streams were shared. Either
   *  argument null/omitted means "Emby's own default". */
  setStreamOptions(audioStreamIndex?: string | null, quality?: string | null): void;
  leave(): void;
}

/**
 * Opens the realtime connection for a party. Returns an object the caller
 * MUST eventually call .leave() on (e.g. from a router cleanup function -
 * see router.ts/player) - otherwise the socket, and whatever it's keeping
 * alive via closures, never gets released.
 */
export function connectParty(partyId: string, callbacks: PartyConnectionCallbacks = {}): PartyConnection {
  const { onMembers, onPlaystate, onItemChanged, onHostChanged, onPauseRequested, onResumeRequested, onQueueUpdated, onDeviceConflict, onBurnInChanged, onTextSubtitleChanged, onStreamOptionsChanged, onError, onClose } = callbacks;
  const socket = io();
  const deviceId = getDeviceId();

  // Reconnects (Socket.IO retries automatically) re-fire "connect", so this
  // also covers rejoining after a transient network drop, not just the
  // first connection. force is deliberately NOT set here even on a
  // reconnect - a drop-and-retry re-sends the same deviceId as before, so
  // it never conflicts with itself; force only ever comes from
  // confirmDeviceTakeover() below, after a real "deviceConflict".
  socket.on('connect', () => {
    socket.emit('join', { partyId, deviceId });
  });

  if (onMembers) socket.on('members', (m) => onMembers(m as SocketMember[]));
  if (onPlaystate) socket.on('playstate', (m) => onPlaystate(m as { isPaused: boolean; positionTicks: number }));
  if (onItemChanged) socket.on('itemChanged', (m) => onItemChanged(m as ItemChangedMsg));
  if (onHostChanged) socket.on('hostChanged', () => onHostChanged());
  if (onPauseRequested) socket.on('pauseRequested', (m) => onPauseRequested(m as PauseRequestedMsg));
  if (onResumeRequested) socket.on('resumeRequested', (m) => onResumeRequested(m as ResumeRequestedMsg));
  if (onQueueUpdated) socket.on('queueUpdated', (m) => onQueueUpdated(m as MusicPublicState)); // music parties only - see musicBar/MusicBar.ts
  if (onDeviceConflict) socket.on('deviceConflict', () => onDeviceConflict());
  if (onBurnInChanged) socket.on('burnInChanged', (m) => onBurnInChanged(m as BurnInChangedMsg));
  if (onTextSubtitleChanged) socket.on('textSubtitleChanged', (m) => onTextSubtitleChanged(m as TextSubtitleChangedMsg));
  if (onStreamOptionsChanged) socket.on('streamOptionsChanged', (m) => onStreamOptionsChanged(m as StreamOptionsChangedMsg));
  if (onError) socket.on('errorMessage', (m) => onError((m as { message: string }).message));
  if (onClose) socket.on('disconnect', () => onClose());

  return {
    sendPlaystate(isPaused, positionTicks) {
      socket.emit('playstate', { isPaused, positionTicks });
    },
    next() {
      socket.emit('next');
    },
    requestPause() {
      socket.emit('requestPause');
    },
    requestResume() {
      socket.emit('requestResume');
    },
    voteSkip() {
      socket.emit('voteSkip');
    },
    setMusicPaused(isPaused) {
      socket.emit('musicPlaystate', { isPaused });
    },
    confirmDeviceTakeover() {
      socket.emit('join', { partyId, deviceId, force: true });
    },
    setBurnInSubtitle(subtitleIndex, audioIndex) {
      socket.emit('setBurnInSubtitle', { subtitleIndex, audioIndex });
    },
    setTextSubtitle(subtitleIndex) {
      socket.emit('setTextSubtitle', { subtitleIndex });
    },
    setStreamOptions(audioStreamIndex, quality) {
      socket.emit('setStreamOptions', { audioStreamIndex, quality });
    },
    leave() {
      socket.emit('leave');
      socket.disconnect();
    }
  };
}