utils/jellyfinSocket.ts

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

import type { SessionsResponse } from "../types/media";
import type { JellyfinSession } from "./jellyfinApi";
import {
  activeJellyfinSessions,
  enrichJellyfinSessions,
  fetchJellyfinSessions,
  jellyfinAuthHeader,
  SessionFeed,
} from "./sessionSources";
import { openWebSocket, reconnectDelay, webSocketUrl } from "./upstreamSocket";

// "initialDelayMs,intervalMs": the server pushes the session list on session
// events, at most this often
const SESSIONS_SUBSCRIPTION = "0,1500";
const DEFAULT_KEEPALIVE_TIMEOUT_S = 60;

interface SocketMessage {
  MessageType: string;
  Data?: unknown;
}

/**
 * Jellyfin pushes the full session list over its WebSocket whenever playback
 * changes, but sends nothing on subscribe, so each connection is seeded with
 * one REST fetch.
 */
export const jellyfinSessionFeed: SessionFeed = ({ onSessions, onDisconnect }) => {
  const JELLYFIN_URL = process.env.JELLYFIN_URL!;
  const JELLYFIN_API_KEY = process.env.JELLYFIN_API_KEY!;

  let stopped = false;
  let socket: WebSocket | undefined;
  let failures = 0;
  let reconnectTimer: ReturnType<typeof setTimeout> | undefined;
  let keepAliveTimer: ReturnType<typeof setInterval> | undefined;
  let watchdogTimer: ReturnType<typeof setTimeout> | undefined;
  let watchdogMs = DEFAULT_KEEPALIVE_TIMEOUT_S * 1500;
  // Only the newest update is published; enrichment can finish out of order
  let sequence = 0;

  const clearTimers = () => {
    clearInterval(keepAliveTimer);
    clearTimeout(watchdogTimer);
  };

  const connect = () => {
    let ws: WebSocket;
    try {
      // Jellyfin 12 rejects the api_key query parameter; auth must be a header
      ws = openWebSocket(webSocketUrl(JELLYFIN_URL, "socket"), {
        Authorization: jellyfinAuthHeader(JELLYFIN_API_KEY),
      });
    } catch (err) {
      // Runs from a timer too, where a throw would take down the server
      console.error("Jellyfin socket could not be created:", err);
      failures++;
      reconnectTimer = setTimeout(connect, reconnectDelay(failures));
      return;
    }
    socket = ws;

    const isCurrent = () => !stopped && socket === ws;

    const publish = async (load: () => Promise<SessionsResponse>) => {
      const id = ++sequence;
      try {
        const data = await load();
        if (isCurrent() && id === sequence) onSessions(data);
      } catch (err) {
        console.error("Jellyfin session update failed:", err);
      }
    };

    // A half-open connection never closes on its own; silence past the
    // keepalive window means it's gone
    const resetWatchdog = () => {
      clearTimeout(watchdogTimer);
      watchdogTimer = setTimeout(() => {
        console.warn("Jellyfin socket went silent; reconnecting");
        ws.close();
        handleClose();
      }, watchdogMs);
    };

    let closed = false;
    const handleClose = () => {
      if (closed || !isCurrent()) return;
      closed = true;
      clearTimers();
      socket = undefined;
      failures++;
      onDisconnect();
      reconnectTimer = setTimeout(connect, reconnectDelay(failures));
    };

    ws.onopen = () => {
      if (!isCurrent()) return;
      failures = 0;
      ws.send(JSON.stringify({ MessageType: "SessionsStart", Data: SESSIONS_SUBSCRIPTION }));
      resetWatchdog();
      publish(fetchJellyfinSessions);
    };

    ws.onmessage = (event: MessageEvent) => {
      if (!isCurrent()) return;
      resetWatchdog();

      let message: SocketMessage;
      try {
        message = JSON.parse(String(event.data));
      } catch {
        return;
      }

      if (message.MessageType === "ForceKeepAlive") {
        const timeoutS = Number(message.Data) || DEFAULT_KEEPALIVE_TIMEOUT_S;
        watchdogMs = timeoutS * 1500;
        resetWatchdog();
        clearInterval(keepAliveTimer);
        keepAliveTimer = setInterval(() => {
          if (ws.readyState === WebSocket.OPEN) {
            ws.send(JSON.stringify({ MessageType: "KeepAlive" }));
          }
        }, (timeoutS * 1000) / 2);
      } else if (message.MessageType === "Sessions" && Array.isArray(message.Data)) {
        const sessions = message.Data as JellyfinSession[];
        publish(() => enrichJellyfinSessions(activeJellyfinSessions(sessions)));
      }
    };

    ws.onerror = () => {
      if (isCurrent() && !closed) console.warn("Jellyfin socket error");
    };
    ws.onclose = handleClose;
  };

  connect();

  return () => {
    stopped = true;
    clearTimers();
    clearTimeout(reconnectTimer);
    socket?.close();
    socket = undefined;
  };
};