bitvid/js/gridHealth.js

import { createCardObserver } from "./dom/cardObserver.js";
import { infoHashFromMagnet } from "./magnets.js";
import {
  updateVideoCardSourceVisibility,
  cardNeedsEagerLivenessProbe,
} from "./utils/cardSourceVisibility.js";
import { TorrentClient, torrentClient } from "./webtorrent.js";
import { userLogger } from "./utils/logger.js";
import {
  readTorrentProbeFromStorage,
  removeTorrentProbeFromStorage,
  writeTorrentProbeToStorage,
} from "./utils/storage.js";
import {
  prioritizeEntries,
  getViewportCenter,
  PRIORITY_BASELINE,
} from "./gridHealthLogic.js";
import {
  FIVE_MINUTES_MS,
  getLivenessProbePrefetchMargin,
} from "./constants.js";

const badgeUpdateListeners = new Map();

function notifyBadgeUpdate(payload) {
  if (!payload) {
    return;
  }

  badgeUpdateListeners.forEach((_owners, listener) => {
    if (typeof listener !== "function") {
      return;
    }
    try {
      listener(payload);
    } catch (err) {
      userLogger.warn("[gridHealth] badge listener failed", err);
    }
  });
}

// Configurable prefetch margin so cards probe ahead of the scroll (WebTorrent
// probes stay concurrency-capped + priority-ordered, so a larger margin only
// reorders work). Falls back to the existing 200px vertical prefetch.
const ROOT_MARGIN = getLivenessProbePrefetchMargin() || "200px 0px";

function now() {
  return Date.now();
}

function getCacheEntry(key) {
  if (!key) {
    return null;
  }
  let entry = probeCache.get(key);

  if (!entry) {
    // Try storage if memory miss
    entry = readTorrentProbeFromStorage(key);
  }

  if (!entry) {
    return null;
  }

  // Verify freshness
  if (now() - entry.ts > PROBE_CACHE_TTL_MS) {
    probeCache.delete(key);
    removeTorrentProbeFromStorage(key);
    return null;
  }

  // Populate memory if it came from storage
  if (!probeCache.has(key)) {
    probeCache.set(key, entry);
  }

  return entry.value;
}

function setCacheEntry(key, value) {
  if (!key) {
    return;
  }
  const entry = { ts: now(), value };
  probeCache.set(key, entry);
  writeTorrentProbeToStorage(key, entry);
}

function formatTime(ts) {
  if (!Number.isFinite(ts)) {
    return "";
  }
  try {
    return new Date(ts).toLocaleTimeString([], { hour12: false });
  } catch (err) {
    return new Date(ts).toLocaleTimeString();
  }
}

function buildTooltip({ peers = 0, checkedAt, reason, webseedOnly } = {}) {
  const parts = [];
  if (webseedOnly) {
    parts.push("Webseed only");
  } else if (Number.isFinite(peers)) {
    parts.push(`Peers: ${Math.max(0, peers)}`);
  }
  if (Number.isFinite(checkedAt)) {
    const formatted = formatTime(checkedAt);
    if (formatted) {
      parts.push(`Checked ${formatted}`);
    }
  }
  if (typeof reason === "string" && reason && reason !== "peer") {
    let normalizedReason;
    if (reason === "timeout") {
      normalizedReason = "Timed out";
    } else if (reason === "no-trackers") {
      normalizedReason = "No WSS trackers";
    } else if (reason === "invalid") {
      normalizedReason = "Invalid magnet";
    } else {
      normalizedReason = reason.charAt(0).toUpperCase() + reason.slice(1);
    }
    parts.push(normalizedReason);
  }
  if (!parts.length) {
    return "WebTorrent status unknown";
  }
  return `WebTorrent • ${parts.join(" • ")}`;
}

function normalizeResult(result) {
  const fallback = {
    healthy: false,
    peers: 0,
    reason: "error",
    webseedOnly: false,
    appendedTrackers: false,
    hasProbeTrackers: false,
    usedTrackers: Array.isArray(TorrentClient.PROBE_TRACKERS)
      ? [...TorrentClient.PROBE_TRACKERS]
      : [],
    durationMs: 0,
  };
  if (!result || typeof result !== "object") {
    return { ...fallback, checkedAt: now() };
  }
  const peers = Number.isFinite(result.peers)
    ? Math.max(0, Number(result.peers))
    : 0;

  /**
   * CRITICAL: Health determination relies SOLELY on `peers > 0`.
   * Because we pass webseeds to the client, a connected webseed counts as a peer.
   *
   * Regression warning:
   * Do not attempt to add `|| webseedOnly` here. If `peers` is 0, the webseed
   * did not connect, and the stream is NOT healthy.
   */
  const healthy = Boolean(result.healthy) && peers > 0;

  const reason = typeof result.reason === "string" ? result.reason : "error";
  const webseedOnly = Boolean(result.webseedOnly) && peers === 0;
  return {
    healthy,
    peers: healthy ? Math.max(1, peers) : peers,
    reason,
    webseedOnly,
    appendedTrackers: Boolean(result.appendedTrackers),
    hasProbeTrackers:
      typeof result.hasProbeTrackers === "boolean"
        ? result.hasProbeTrackers
        : healthy || Boolean(result.appendedTrackers),
    usedTrackers: Array.isArray(result.usedTrackers)
      ? result.usedTrackers.slice()
      : fallback.usedTrackers,
    durationMs: Number.isFinite(result.durationMs)
      ? Math.max(0, Number(result.durationMs))
      : 0,
    checkedAt: now(),
  };
}

function queueProbe(magnet, cacheKey, priority = 0, webSeeds = [], owner) {
  if (!magnet) {
    return { promise: Promise.resolve(null), owners: new Set() };
  }

  const cached = getCacheEntry(cacheKey);
  if (cached) {
    return { promise: Promise.resolve(cached), owners: new Set() };
  }

  if (probeInflight.has(cacheKey)) {
    const record = probeInflight.get(cacheKey);
    if (!record.cancelled) {
      record.owners.add(owner);
      owner.release = () => releaseProbe(record, owner, cacheKey);
      return record;
    }
  }

  const record = { owners: new Set([owner]), cancelled: false };
  owner.release = () => releaseProbe(record, owner, cacheKey);
  // Publish ownership and the result before starting IO: client.add() can
  // synchronously unregister a card or register another owner of this source.
  let resolveJob;
  let rejectJob;
  record.promise = new Promise((resolve, reject) => {
    resolveJob = resolve;
    rejectJob = reject;
  })
    .then((result) => {
      if (record.cancelled) return null;
      const normalized = normalizeResult(result);
      if (cacheKey) setCacheEntry(cacheKey, normalized);
      return normalized;
    })
    .finally(() => {
      if (cacheKey && probeInflight.get(cacheKey) === record) {
        probeInflight.delete(cacheKey);
      }
    });
  if (cacheKey) probeInflight.set(cacheKey, record);
  record.job = probeQueue.run(
    () =>
      torrentClient
        .probePeers(
          magnet,
          {
            timeoutMs: PROBE_TIMEOUT_MS,
            maxWebConns: 2,
            polls: PROBE_POLL_COUNT,
            urlList: webSeeds,
          },
          record
        )
        .catch((err) => ({
          healthy: false,
          peers: 0,
          reason: "error",
          error: err,
          appendedTrackers: false,
          hasProbeTrackers: false,
        })),
    priority
  );
  record.job.promise.then(resolveJob, rejectJob);

  return record;
}

function releaseProbe(record, owner, cacheKey) {
  owner.active = false;
  record.owners.delete(owner);
  if (!record.owners.size && !record.cancelled) {
    record.cancelled = true;
    if (probeInflight.get(cacheKey) === record) {
      probeInflight.delete(cacheKey);
    }
    record.job?.cancel();
    record.cancelActive?.();
  }
}

const PROBE_CACHE_TTL_MS = FIVE_MINUTES_MS;
const PROBE_TIMEOUT_MS = 20000;
// Each probe is a real WebTorrent swarm join (tracker announces + peer
// connections), NOT a cheap ping. At 96 concurrent, loading/scrolling the feed
// opened hundreds-to-thousands of simultaneous WebSocket/WebRTC connections,
// pegging the main thread and exhausting memory (freeze + tab crash). Keep this
// small — badges still populate, just a few at a time.
const PROBE_CONCURRENCY = 4;
const PROBE_POLL_COUNT = 3;

class ProbeQueue {
  constructor(max = 2) {
    this.max = Math.max(1, Number(max) || 1);
    this.running = 0;
    this.queue = [];
  }

  run(task, priority = 0) {
    let job;
    const promise = new Promise((resolve, reject) => {
      const normalizedPriority = Number.isFinite(priority) ? priority : 0;
      job = {
        task,
        resolve,
        reject,
        priority: normalizedPriority,
      };
      this.enqueue(job);
      this.drain();
    });
    return {
      promise,
      cancel: () => {
        const index = this.queue.indexOf(job);
        if (index < 0) {
          return false;
        }
        this.queue.splice(index, 1);
        job.resolve(null);
        return true;
      },
    };
  }

  enqueue(job) {
    if (!this.queue.length) {
      this.queue.push(job);
      return;
    }
    const index = this.queue.findIndex(
      (existing) => existing.priority < job.priority
    );
    if (index === -1) {
      this.queue.push(job);
    } else {
      this.queue.splice(index, 0, job);
    }
  }

  drain() {
    while (this.running < this.max && this.queue.length > 0) {
      const job = this.queue.shift();
      if (!job) {
        return;
      }
      this.running += 1;
      let finished = false;
      const finalize = () => {
        if (finished) {
          return;
        }
        finished = true;
        this.running -= 1;
        this.drain();
      };

      let result;
      try {
        result = job.task();
      } catch (err) {
        job.reject(err);
        finalize();
        continue;
      }

      Promise.resolve(result)
        .then((value) => {
          job.resolve(value);
        })
        .catch((err) => {
          job.reject(err);
        })
        .finally(() => {
          finalize();
        });
    }
  }
}

const probeQueue = new ProbeQueue(PROBE_CONCURRENCY);
const probeCache = new Map();
const probeInflight = new Map();

function setBadge(card, state, details) {
  if (!(card instanceof HTMLElement)) {
    return;
  }

  const normalizedState =
    typeof state === "string" && state ? state : "unknown";
  const peersValue =
    details && Number.isFinite(details.peers)
      ? Math.max(0, Number(details.peers))
      : 0;
  const hasPeerCount = details ? Number.isFinite(details.peers) : false;
  const peersTextValue = hasPeerCount ? String(peersValue) : "";
  const webseedOnly = Boolean(details?.webseedOnly);

  card.dataset.streamHealthState = normalizedState;
  if (hasPeerCount) {
    card.dataset.streamHealthPeers = peersTextValue;
  } else if (card.dataset.streamHealthPeers) {
    delete card.dataset.streamHealthPeers;
  }

  if (details && typeof details.reason === "string" && details.reason) {
    card.dataset.streamHealthReason = details.reason;
  } else if (card.dataset.streamHealthReason) {
    delete card.dataset.streamHealthReason;
  }

  updateVideoCardSourceVisibility(card);

  const badge = card.querySelector(".torrent-health-badge");
  if (!badge) {
    return;
  }

  const hadCompactMargin =
    badge.classList.contains("mt-sm") || badge.classList.contains("mt-3");

  const classes = ["badge", "torrent-health-badge"];
  if (hadCompactMargin) {
    classes.push("mt-sm");
  }
  badge.className = classes.join(" ");

  const map = {
    healthy: {
      icon: "🟢",
      aria: "WebTorrent peers available",
      variant: "success",
      role: "status",
    },
    unhealthy: {
      icon: "🔴",
      aria: "WebTorrent peers unavailable",
      variant: "critical",
      role: "alert",
    },
    checking: {
      icon: "⏳",
      aria: "Checking WebTorrent peers",
      variant: "neutral",
      role: "status",
    },
    unknown: {
      icon: "⚪",
      aria: "WebTorrent status unknown",
      variant: "neutral",
      role: "status",
    },
  };

  const entry = map[normalizedState] || map.unknown;
  if (entry.variant) {
    badge.dataset.variant = entry.variant;
  } else if (badge.dataset.variant) {
    delete badge.dataset.variant;
  }

  const iconPrefix = entry.icon ? `${entry.icon} ` : "";
  badge.textContent = `${iconPrefix}WebTorrent`;
  const tooltip =
    normalizedState === "checking" || normalizedState === "unknown"
      ? entry.aria
      : buildTooltip({
          peers: peersValue,
          checkedAt: details?.checkedAt,
          reason: details?.reason,
          webseedOnly,
        });
  badge.setAttribute("aria-label", tooltip);
  badge.setAttribute("title", tooltip);
  const ariaLive = entry.role === "alert" ? "assertive" : "polite";
  badge.setAttribute("aria-live", ariaLive);
  badge.setAttribute("role", entry.role);
  badge.dataset.streamHealthState = normalizedState;
  if (hasPeerCount) {
    badge.dataset.streamHealthPeers = peersTextValue;
  } else if (badge.dataset.streamHealthPeers) {
    delete badge.dataset.streamHealthPeers;
  }

  const payload = {
    card,
    state: normalizedState,
    peers: hasPeerCount ? peersValue : null,
    reason:
      details && typeof details.reason === "string" ? details.reason : null,
    webseedOnly,
    checkedAt:
      details && Number.isFinite(details.checkedAt)
        ? Number(details.checkedAt)
        : null,
    text: badge.textContent,
    tooltip,
    role: entry.role,
    ariaLive,
  };
  notifyBadgeUpdate(payload);
}

function handleCardVisible({ card, pendingByCard, liveCards, priority = 0 }) {
  if (!(card instanceof HTMLElement)) {
    return;
  }
  const magnet = card.dataset.magnet || "";
  if (!magnet) {
    setBadge(card, "unhealthy", { reason: "missing-source" });
    return;
  }

  const infoHash = infoHashFromMagnet(magnet);
  if (!infoHash) {
    setBadge(card, "unhealthy", { reason: "invalid" });
    return;
  }

  if (!(pendingByCard instanceof WeakMap)) {
    return;
  }

  if (pendingByCard.has(card)) {
    return;
  }

  const cached = getCacheEntry(infoHash);
  if (cached) {
    const cachedState = cached.healthy
      ? "healthy"
      : cached.hasProbeTrackers
        ? "unhealthy"
        : "unknown";
    setBadge(card, cachedState, cached);
    return;
  }

  setBadge(card, "checking");
  // Badge callbacks can synchronously dispose/rerender the feed. Do not create
  // a new request after its unregister hook has already run.
  if (!liveCards.has(card)) return;

  const encodedUrl = card.dataset.urlHealthUrl || "";
  const webSeed = encodedUrl ? decodeURIComponent(encodedUrl) : "";
  const webSeeds = webSeed ? [webSeed] : [];

  const owner = { active: true };
  pendingByCard.set(card, owner);
  const record = queueProbe(magnet, infoHash, priority, webSeeds, owner);

  record.promise
    .then((result) => {
      if (!owner.active || pendingByCard.get(card) !== owner) {
        return;
      }
      pendingByCard.delete(card);
      record.owners.delete(owner);
      if (!card.isConnected) {
        return;
      }
      if (!result) {
        setBadge(card, "unknown");
        return;
      }
      if (!result.hasProbeTrackers) {
        setBadge(card, "unknown", result);
        return;
      }
      if (result.healthy) {
        setBadge(card, "healthy", result);
        return;
      }
      setBadge(card, "unhealthy", result);
    })
    .catch((err) => {
      if (!owner.active || pendingByCard.get(card) !== owner) {
        return;
      }
      pendingByCard.delete(card);
      record.owners.delete(owner);
      if (!card.isConnected) {
        return;
      }
      userLogger.warn("[gridHealth] probe failed", err);
      setBadge(card, "unhealthy");
    });
}

const gridCardObserver = createCardObserver({
  rootMargin: ROOT_MARGIN,
  threshold: 0.01,
  createState: () => ({
    pendingByCard: new WeakMap(),
    liveCards: new WeakSet(),
  }),
  prepareEntries: (entries) => {
    const viewportCenter = getViewportCenter();
    return prioritizeEntries(entries, viewportCenter);
  },
  isCardVisible: (entry) => {
    if (!entry) {
      return false;
    }
    const isIntersecting = Boolean(entry.isIntersecting);
    const ratio =
      typeof entry.intersectionRatio === "number" ? entry.intersectionRatio : 0;
    return isIntersecting && ratio > 0;
  },
  onCardRegister: ({ card, state }) => {
    state.liveCards.add(card);
    if (!card.dataset.magnet) {
      setBadge(card, "unhealthy", { reason: "missing-source" });
      return;
    }
    // Hide-until-verified cards start hidden, so the viewport observer can never
    // fire for them — probe eagerly on register (still concurrency-capped) so
    // they can be verified and revealed instead of staying hidden forever.
    if (
      cardNeedsEagerLivenessProbe(card) &&
      state &&
      state.pendingByCard instanceof WeakMap
    ) {
      handleCardVisible({
        card,
        pendingByCard: state.pendingByCard,
        liveCards: state.liveCards,
        priority: PRIORITY_BASELINE,
      });
    }
  },
  onCardUnregister: ({ card, state }) => {
    state.liveCards.delete(card);
    const owner = state.pendingByCard.get(card);
    if (owner) {
      owner.release();
      state.pendingByCard.delete(card);
    }
  },
  onCardVisible: ({ card, meta, state }) => {
    if (!state?.liveCards.has(card)) {
      return;
    }
    const priority =
      meta && typeof meta === "object" && Number.isFinite(meta.priority)
        ? meta.priority
        : PRIORITY_BASELINE;
    handleCardVisible({
      card,
      pendingByCard: state.pendingByCard,
      liveCards: state.liveCards,
      priority,
    });
  },
});

export function attachHealthBadges(container, options = {}) {
  if (!(container instanceof HTMLElement)) {
    return () => {};
  }
  const listener =
    typeof options?.onUpdate === "function" ? options.onUpdate : null;
  if (listener) {
    badgeUpdateListeners.set(
      listener,
      (badgeUpdateListeners.get(listener) || 0) + 1
    );
  }
  gridCardObserver.observe(container);
  let disposed = false;
  return () => {
    if (disposed) return;
    disposed = true;
    if (listener) {
      const remaining = badgeUpdateListeners.get(listener) - 1;
      if (remaining > 0) badgeUpdateListeners.set(listener, remaining);
      else badgeUpdateListeners.delete(listener);
    }
    gridCardObserver.disconnect(container);
  };
}