pondering-orb/web/health.js

// health.js — stream health prober + scoring (upgrade #4).
//
// Lazy + bounded: we probe the current channel and whatever guide rows are
// visible, never the whole catalog (respects upstreams, scales to ~39k). Results
// are cached in-memory (short TTL) and folded into the Oracle's Memory uptime
// history. Uses the relay's /probe endpoint when available; falls back to the
// provider's own healthCheck on a plain static server.

import { memory } from './memory.js';

const TTL_MS = 5 * 60 * 1000;     // re-probe a channel at most this often
const SLOW_MS = 4000;             // 2xx but slower than this → "degraded"
const POOL = 6;                   // max concurrent probes

/** @typedef {'live'|'degraded'|'dead'} HealthState */

export class Health {
  /**
   * @param {import('./providers/source-provider.js').ChannelAggregator} agg
   * @param {boolean} hasRelay  is the same-origin /probe endpoint available?
   */
  constructor(agg, hasRelay) {
    this.agg = agg;
    this.hasRelay = hasRelay;
    this._cache = new Map();       // id -> { state, ts }
    this._inflight = new Map();    // id -> Promise<HealthState>
    this._queue = [];
    this._active = 0;
    /** @type {(id:string, state:HealthState)=>void} */
    this.onResult = () => {};
  }

  /** Cached state without triggering a probe (for instant render / auto-skip). */
  cached(id) {
    const c = this._cache.get(id);
    if (c) return c.state;
    return memory.reliability(id)?.last ?? null;
  }

  isKnownDead(id) { return this.cached(id) === 'dead'; }

  /** Probe one channel (deduped, TTL-cached). */
  async probe(channel) {
    const id = channel.id;
    const c = this._cache.get(id);
    if (c && Date.now() - c.ts < TTL_MS) return c.state;
    if (this._inflight.has(id)) return this._inflight.get(id);

    const p = this._enqueue(channel).then((state) => {
      this._cache.set(id, { state, ts: Date.now() });
      this._inflight.delete(id);
      memory.recordHealth(id, state);
      this.onResult(id, state);
      return state;
    });
    this._inflight.set(id, p);
    return p;
  }

  /** Probe many with bounded concurrency (fire-and-forget UI updates). */
  probeMany(channels) {
    for (const ch of channels) {
      if (!ch) continue;
      const c = this._cache.get(ch.id);
      if (c && Date.now() - c.ts < TTL_MS) continue;
      this.probe(ch);
    }
  }

  // ---- internal pool ----
  _enqueue(channel) {
    return new Promise((resolve) => {
      this._queue.push({ channel, resolve });
      this._pump();
    });
  }
  _pump() {
    while (this._active < POOL && this._queue.length) {
      const { channel, resolve } = this._queue.shift();
      this._active++;
      this._runOne(channel).then((state) => {
        this._active--;
        resolve(state);
        this._pump();
      });
    }
  }

  /** @returns {Promise<HealthState>} */
  async _runOne(channel) {
    const raw = channel.probeUrls && channel.probeUrls[0];
    if (this.hasRelay && raw) {
      try {
        const t0 = performance.now();
        const r = await fetch('/probe?url=' + encodeURIComponent(raw), {
          credentials: 'omit', referrerPolicy: 'no-referrer',
        });
        if (r.ok) {
          const j = await r.json();
          if (!j.ok) return 'dead';
          const ms = j.latency_ms ?? (performance.now() - t0);
          return ms >= SLOW_MS ? 'degraded' : 'live';
        }
        // 4xx/5xx from /probe means policy block or upstream fail
        return 'dead';
      } catch {
        return 'dead';
      }
    }
    // fallback: provider-native check (mock / static server)
    const owner = this.agg.providers.get(channel.source);
    if (owner?.healthCheck) {
      try { const r = await owner.healthCheck(channel.id); return r.ok ? 'live' : 'dead'; }
      catch { return 'dead'; }
    }
    return 'live';
  }
}