utils/sessionHub.ts

Rama main · réplica de lectura del repositorio original.

import type {
  BaseMedia,
  ServiceHealth,
  SessionItem,
  SessionsResponse,
  SessionStreamMessage,
  SourceHealth,
} from "../types/media";
import type { NowPlayingStatus, SourceMode, SourceStatus, StatusSnapshot } from "../types/status";
import {
  configuredSources,
  SESSION_FETCHERS,
  SessionFeed,
  SessionSource,
} from "./sessionSources";
import { jellyfinSessionFeed } from "./jellyfinSocket";
import { plexSessionFeed } from "./plexSocket";
import { monitor } from "./monitor";
import { getPlexConfig, PlexAuthError, setupState } from "./plexConfig";

// Live upstream connections; a source without one is polled
const FEEDS: Partial<Record<SessionSource, SessionFeed>> = {
  plex: plexSessionFeed,
  jellyfin: jellyfinSessionFeed,
};

const POLL_PLAYING_MS = 10_000;
const POLL_IDLE_MS = 30_000;
// A rejected token won't start working by itself, so check back rarely
const POLL_UNAUTHORIZED_MS = 5 * 60_000;
// A feed that hasn't delivered by then is backed up by polling
const FEED_CONNECT_GRACE_MS = 5_000;
// Upstream work stops this long after the last viewer leaves; a page reload
// shouldn't tear everything down
const IDLE_SHUTDOWN_MS = 2 * 60_000;
// Coalesces near-simultaneous source updates into one push
const BROADCAST_DELAY_MS = 100;
// A manual refresh within this window of the last fetch reuses it
const REFRESH_MIN_GAP_MS = 2_000;
// Viewers interpolate playing progress, so a fresh offset within this much of
// the interpolated one isn't worth pushing
const PROGRESS_TOLERANCE_MS = 2_000;

type Listener = (message: SessionStreamMessage) => void;
type Item = SessionItem<BaseMedia>;
type Category = keyof SessionsResponse;

const CATEGORIES: Category[] = ["tracks", "movies", "episodes"];

const SOURCE_LABELS: Record<SessionSource, string> = { plex: "Plex", jellyfin: "Jellyfin" };

interface SourceState {
  // What viewers were last sent, so change detection compares against what
  // they're actually interpolating from
  data: SessionsResponse | null;
  // When data's viewOffsets were current
  observedAt: number;
  error: Error | null;
  // Has completed at least one fetch since the hub started
  settled: boolean;
  fetching: boolean;
  lastFetch: number;
  failures: number;
  timer?: ReturnType<typeof setTimeout>;
  // The feed is connected and delivering, so polling is paused
  live: boolean;
  stopFeed?: () => void;
  // For the status page
  mode: SourceMode;
  modeSince: number;
  lastUpdate: number | null;
}

function hasPlaying(data: SessionsResponse | null): boolean {
  return !!data && CATEGORIES.some((c) => data[c].some((item) => item.state === "playing"));
}

function modeOf(state: SourceState): SourceMode {
  if (state.live) return "live";
  if (state.error instanceof PlexAuthError) return "unauthorized";
  if (state.error) return "error";
  return state.settled ? "polling" : "connecting";
}

// What viewers are told; "connecting" never reaches them, since nothing is
// sent until every source has settled
function healthOf(mode: SourceMode): SourceHealth {
  if (mode === "unauthorized") return "unauthorized";
  return mode === "error" ? "unreachable" : "connected";
}

// Upstream errors can echo request URLs, which may carry credentials
function redact(message: string): string {
  let result = message;
  for (const secret of [getPlexConfig()?.token, process.env.PLEX_TOKEN, process.env.JELLYFIN_API_KEY]) {
    if (secret) result = result.split(secret).join("***");
  }
  return result;
}

function nowPlaying(source: SessionSource, data: SessionsResponse): NowPlayingStatus[] {
  const entry = (item: Item, title: string): NowPlayingStatus => ({
    source,
    user: item.userId,
    title,
    player: item.player,
    state: item.state,
  });
  return [
    ...data.tracks.map((t) => entry(t, `${t.artist} – ${t.title}`)),
    ...data.movies.map((m) => entry(m, m.year ? `${m.title} (${m.year})` : m.title)),
    ...data.episodes.map((e) => entry(e, `${e.showTitle} S${e.season}E${e.episode} – ${e.title}`)),
  ];
}

function retryDelay(failures: number): number {
  return Math.min(5000 * Math.pow(1.5, failures - 1), 60_000);
}

function withoutOffset(item: Item): string {
  return JSON.stringify({ ...item, viewOffset: 0 });
}

function isMeaningfulChange(
  prev: SessionsResponse,
  prevAt: number,
  next: SessionsResponse,
  nextAt: number
): boolean {
  return CATEGORIES.some((category) => {
    const prevItems: Item[] = prev[category];
    const nextItems: Item[] = next[category];
    if (prevItems.length !== nextItems.length) return true;

    const prevBySession = new Map(prevItems.map((item) => [item.sessionId, item]));
    return nextItems.some((item) => {
      const old = prevBySession.get(item.sessionId);
      if (!old || withoutOffset(old) !== withoutOffset(item)) return true;

      const oldOffset = old.viewOffset ?? 0;
      const offset = item.viewOffset ?? 0;
      if (item.state !== "playing") return offset !== oldOffset;

      const projected = oldOffset + (nextAt - prevAt);
      const expected = item.duration ? Math.min(projected, item.duration) : projected;
      return Math.abs(offset - expected) > PROGRESS_TOLERANCE_MS;
    });
  });
}

// Moves playing items' offsets forward to `now`, so a viewer joining between
// fetches starts from the current position
function advance<T extends Item>(items: T[], elapsed: number): T[] {
  if (elapsed <= 0) return items;
  return items.map((item) => {
    if (item.state !== "playing") return item;
    const projected = (item.viewOffset ?? 0) + elapsed;
    return { ...item, viewOffset: item.duration ? Math.min(projected, item.duration) : projected };
  });
}

/**
 * Single owner of upstream session tracking. However many browsers are
 * watching, each media server has one live connection (or, failing that, one
 * polling loop) here, and every change is pushed to all subscribers. Upstream
 * work starts with the first subscriber and stops a while after the last one
 * leaves.
 */
class SessionHub {
  private listeners = new Set<Listener>();
  private sources = new Map<SessionSource, SourceState>();
  private running = false;
  // Bumped on stop so fetches from a previous run are ignored
  private generation = 0;
  private shutdownTimer?: ReturnType<typeof setTimeout>;
  private stopsAt: number | null = null;
  private broadcastTimer?: ReturnType<typeof setTimeout>;

  subscribe(listener: Listener): () => void {
    this.listeners.add(listener);
    clearTimeout(this.shutdownTimer);
    this.stopsAt = null;

    if (!this.running) this.start();
    else if (this.isReady()) listener(this.snapshot());

    return () => {
      if (!this.listeners.delete(listener) || this.listeners.size > 0) return;
      clearTimeout(this.shutdownTimer);
      this.stopsAt = Date.now() + IDLE_SHUTDOWN_MS;
      this.shutdownTimer = setTimeout(() => this.stop(), IDLE_SHUTDOWN_MS);
    };
  }

  /** Fetches every polled source now, unless one just did. Live feeds are already current. */
  refresh(): void {
    if (!this.running) return;
    const now = Date.now();
    for (const [source, state] of this.sources) {
      if (!state.live && !state.fetching && now - state.lastFetch > REFRESH_MIN_GAP_MS) {
        this.poll(source, this.generation);
      }
    }
  }

  /**
   * Picks up a source whose connection was added, changed or removed (e.g. by
   * signing in on /setup) without disturbing the others.
   */
  reconfigure(source: SessionSource): void {
    if (!this.running) return;
    const old = this.sources.get(source);
    if (old) {
      clearTimeout(old.timer);
      old.stopFeed?.();
      this.sources.delete(source);
    }
    if (configuredSources().includes(source)) this.startSource(source, this.generation);
    this.scheduleBroadcast();
  }

  /** Upstream health and current sessions, for the owner's status page. */
  getStatus(): Pick<StatusSnapshot, "hub" | "sources" | "nowPlaying"> {
    const sources: SourceStatus[] = [];
    const playing: NowPlayingStatus[] = [];
    for (const [source, state] of this.sources) {
      const sessions = state.data ? nowPlaying(source, state.data) : [];
      playing.push(...sessions);
      sources.push({
        source,
        mode: state.mode,
        since: state.modeSince,
        lastUpdate: state.lastUpdate,
        failures: state.failures,
        lastError: state.error ? redact(state.error.message) : null,
        sessions: sessions.length,
      });
    }
    return {
      hub: { running: this.running, subscribers: this.listeners.size, stopsAt: this.stopsAt },
      sources,
      nowPlaying: playing,
    };
  }

  private start(): void {
    this.running = true;
    this.sources.clear();
    const generation = this.generation;
    monitor.log("info", "Session hub started");

    for (const source of configuredSources()) this.startSource(source, generation);
    // Nothing configured: viewers get an empty snapshot rather than waiting forever
    if (this.sources.size === 0) this.scheduleBroadcast();
  }

  private startSource(source: SessionSource, generation: number): void {
    const state: SourceState = {
      data: null,
      observedAt: 0,
      error: null,
      settled: false,
      fetching: false,
      lastFetch: 0,
      failures: 0,
      live: false,
      mode: "connecting",
      modeSince: Date.now(),
      lastUpdate: null,
    };
    this.sources.set(source, state);

    const feed = FEEDS[source];
    // Node < 22 has no built-in WebSocket; those servers just poll
    if (!feed || typeof WebSocket === "undefined") {
      this.poll(source, generation);
      return;
    }

    try {
      state.stopFeed = feed({
        onSessions: (data) => this.onFeedSessions(source, data, generation),
        onDisconnect: () => this.onFeedDisconnect(source, generation),
      });
      state.timer = setTimeout(() => this.poll(source, generation), FEED_CONNECT_GRACE_MS);
    } catch (err) {
      console.error(`Could not start ${source} feed; polling instead:`, err);
      this.poll(source, generation);
    }
  }

  private stop(): void {
    monitor.log("info", "Session hub stopped (no viewers)");
    this.running = false;
    this.stopsAt = null;
    this.generation++;
    clearTimeout(this.broadcastTimer);
    this.broadcastTimer = undefined;
    for (const state of this.sources.values()) {
      clearTimeout(state.timer);
      state.stopFeed?.();
    }
    // Drop data too: offsets can't be projected across a gap with no fetches
    this.sources.clear();
  }

  private onFeedSessions(source: SessionSource, data: SessionsResponse, generation: number): void {
    const state = this.sources.get(source);
    if (!state || generation !== this.generation) return;
    if (!state.live) console.info(`${source} sessions: live`);
    state.live = true;
    clearTimeout(state.timer);
    this.settle(state, this.ingest(state, data));
    this.noteMode(source, state);
  }

  private onFeedDisconnect(source: SessionSource, generation: number): void {
    const state = this.sources.get(source);
    if (!state || generation !== this.generation || !state.live) return;
    console.warn(`${source} sessions: feed lost, polling until it reconnects`);
    state.live = false;
    this.noteMode(source, state);
    if (!state.fetching) this.poll(source, generation);
  }

  // Returns whether viewers need to hear about it
  private ingest(state: SourceState, data: SessionsResponse): boolean {
    let changed = state.error !== null;
    const observedAt = Date.now();
    state.lastUpdate = observedAt;
    if (!state.data || isMeaningfulChange(state.data, state.observedAt, data, observedAt)) {
      state.data = data;
      state.observedAt = observedAt;
      changed = true;
    }
    state.error = null;
    state.failures = 0;
    return changed;
  }

  private settle(state: SourceState, changed: boolean): void {
    if (!state.settled) {
      state.settled = true;
      changed = true;
    }
    if (changed) this.scheduleBroadcast();
  }

  private async poll(source: SessionSource, generation: number): Promise<void> {
    const state = this.sources.get(source);
    if (!state || state.live) return;
    clearTimeout(state.timer);
    state.fetching = true;

    let changed = false;
    try {
      const data = await SESSION_FETCHERS[source]();
      // The feed took over mid-fetch and is at least as fresh, or the source
      // was reconfigured and this fetch used the old connection
      if (generation !== this.generation || state.live || this.sources.get(source) !== state) return;
      changed = this.ingest(state, data);
    } catch (err) {
      if (generation !== this.generation || state.live || this.sources.get(source) !== state) return;

      console.error(`Error fetching ${source} sessions:`, err);
      // A failed source contributes nothing rather than going stale
      changed = state.data !== null || state.error === null;
      state.data = null;
      state.error = err instanceof Error ? err : new Error(String(err));
      state.failures++;
    } finally {
      if (generation === this.generation) {
        state.fetching = false;
        state.lastFetch = Date.now();
      }
    }

    this.settle(state, changed);
    this.noteMode(source, state);

    const delay = state.error instanceof PlexAuthError
      ? POLL_UNAUTHORIZED_MS
      : state.error
      ? retryDelay(state.failures)
      : hasPlaying(state.data)
        ? POLL_PLAYING_MS
        : POLL_IDLE_MS;
    state.timer = setTimeout(() => this.poll(source, generation), delay);
  }

  private noteMode(source: SessionSource, state: SourceState): void {
    const mode = modeOf(state);
    if (mode === state.mode) return;
    // e.g. unreachable -> unauthorized changes no data but changes the banner
    if (healthOf(mode) !== healthOf(state.mode)) this.scheduleBroadcast();
    const label = SOURCE_LABELS[source];
    if (mode === "live") monitor.log("info", `${label}: live connection established`);
    else if (mode === "unauthorized") monitor.log("warn", `${label}: token rejected; sign in again to reconnect`);
    else if (mode === "error") {
      monitor.log("error", `${label}: unreachable — ${redact(state.error?.message ?? "unknown error")}`);
    } else if (state.mode === "live") monitor.log("warn", `${label}: live connection lost, polling`);
    else monitor.log("info", `${label}: polling`);
    state.mode = mode;
    state.modeSince = Date.now();
  }

  private isReady(): boolean {
    for (const state of this.sources.values()) {
      if (!state.settled) return false;
    }
    return true;
  }

  private scheduleBroadcast(): void {
    if (this.broadcastTimer || !this.isReady()) return;
    this.broadcastTimer = setTimeout(() => {
      this.broadcastTimer = undefined;
      if (!this.running || !this.isReady()) return;
      const message = this.snapshot();
      for (const listener of this.listeners) listener(message);
    }, BROADCAST_DELAY_MS);
  }

  private snapshot(): SessionStreamMessage {
    const now = Date.now();
    const sessions: SessionsResponse = { tracks: [], movies: [], episodes: [] };
    // Labels only: upstream error text can include server URLs and tokens
    const failed: string[] = [];
    const health: ServiceHealth = {};

    for (const [source, state] of this.sources) {
      health[source] = healthOf(modeOf(state));
      if (state.error) failed.push(SOURCE_LABELS[source]);
      if (!state.data) continue;
      const elapsed = now - state.observedAt;
      sessions.tracks.push(...advance(state.data.tracks, elapsed));
      sessions.movies.push(...advance(state.data.movies, elapsed));
      sessions.episodes.push(...advance(state.data.episodes, elapsed));
    }

    const allFailed = this.sources.size > 0 && failed.length === this.sources.size;
    const message: SessionStreamMessage = { sessions, sources: health, setup: setupState() };
    return allFailed ? { ...message, error: `Unable to reach ${failed.join(" or ")}.` } : message;
  }
}

// One hub per server process, surviving dev hot reloads and duplicated module
// instances so there's never more than one upstream loop per media server
const globalForHub = globalThis as typeof globalThis & { __sessionHub?: SessionHub };

export const sessionHub = (globalForHub.__sessionHub ??= new SessionHub());