sofarr/server/utils/poller.js

// Copyright (c) 2025 Gordon Bolton. MIT License.
const axios = require('axios');
const cache = require('./cache');
const { getTorrents } = require('./qbittorrent');
const {
  getSABnzbdInstances,
  getSonarrInstances,
  getRadarrInstances
} = require('./config');

const rawPollInterval = (process.env.POLL_INTERVAL || '').toLowerCase();
const POLL_INTERVAL = (rawPollInterval === 'off' || rawPollInterval === 'false' || rawPollInterval === 'disabled')
  ? 0
  : (parseInt(process.env.POLL_INTERVAL, 10) || 5000);
const POLLING_ENABLED = POLL_INTERVAL > 0;

let polling = false;
let lastPollTimings = null;

// SSE subscribers: Set of () => void callbacks, each registered by an open /stream connection
const pollSubscribers = new Set();

function onPollComplete(cb) { pollSubscribers.add(cb); }
function offPollComplete(cb) { pollSubscribers.delete(cb); }

// Timed fetch helper: runs a fetch and records how long it took
async function timed(label, fn) {
  const t0 = Date.now();
  const result = await fn();
  return { label, result, ms: Date.now() - t0 };
}

async function pollAllServices() {
  if (polling) {
    console.log('[Poller] Previous poll still running, skipping');
    return;
  }
  polling = true;
  const start = Date.now();

  try {
    const sabInstances = getSABnzbdInstances();
    const sonarrInstances = getSonarrInstances();
    const radarrInstances = getRadarrInstances();

    // All fetches in parallel, each individually timed
    const results = await Promise.all([
      timed('SABnzbd Queue', () => Promise.all(sabInstances.map(inst =>
        axios.get(`${inst.url}/api`, {
          params: { mode: 'queue', apikey: inst.apiKey, output: 'json' }
        }).then(res => ({ instance: inst.id, data: res.data })).catch(err => {
          console.error(`[Poller] SABnzbd ${inst.id} queue error:`, err.message);
          return { instance: inst.id, data: { queue: { slots: [] } } };
        })
      ))),
      timed('SABnzbd History', () => Promise.all(sabInstances.map(inst =>
        axios.get(`${inst.url}/api`, {
          params: { mode: 'history', apikey: inst.apiKey, output: 'json', limit: 10 }
        }).then(res => ({ instance: inst.id, data: res.data })).catch(err => {
          console.error(`[Poller] SABnzbd ${inst.id} history error:`, err.message);
          return { instance: inst.id, data: { history: { slots: [] } } };
        })
      ))),
      timed('Sonarr Tags', () => Promise.all(sonarrInstances.map(inst =>
        axios.get(`${inst.url}/api/v3/tag`, {
          headers: { 'X-Api-Key': inst.apiKey }
        }).then(res => ({ instance: inst.id, data: res.data })).catch(err => {
          console.error(`[Poller] Sonarr ${inst.id} tags error:`, err.message);
          return { instance: inst.id, data: [] };
        })
      ))),
      timed('Sonarr Queue', () => Promise.all(sonarrInstances.map(inst =>
        axios.get(`${inst.url}/api/v3/queue`, {
          headers: { 'X-Api-Key': inst.apiKey },
          params: { includeSeries: true, includeEpisode: true }
        }).then(res => ({ instance: inst.id, data: res.data })).catch(err => {
          console.error(`[Poller] Sonarr ${inst.id} queue error:`, err.message);
          return { instance: inst.id, data: { records: [] } };
        })
      ))),
      timed('Sonarr History', () => Promise.all(sonarrInstances.map(inst =>
        axios.get(`${inst.url}/api/v3/history`, {
          headers: { 'X-Api-Key': inst.apiKey },
          params: { pageSize: 10, includeEpisode: true }
        }).then(res => ({ instance: inst.id, data: res.data })).catch(err => {
          console.error(`[Poller] Sonarr ${inst.id} history error:`, err.message);
          return { instance: inst.id, data: { records: [] } };
        })
      ))),
      timed('Radarr Queue', () => Promise.all(radarrInstances.map(inst =>
        axios.get(`${inst.url}/api/v3/queue`, {
          headers: { 'X-Api-Key': inst.apiKey },
          params: { includeMovie: true }
        }).then(res => ({ instance: inst.id, data: res.data })).catch(err => {
          console.error(`[Poller] Radarr ${inst.id} queue error:`, err.message);
          return { instance: inst.id, data: { records: [] } };
        })
      ))),
      timed('Radarr History', () => Promise.all(radarrInstances.map(inst =>
        axios.get(`${inst.url}/api/v3/history`, {
          headers: { 'X-Api-Key': inst.apiKey },
          params: { pageSize: 10 }
        }).then(res => ({ instance: inst.id, data: res.data })).catch(err => {
          console.error(`[Poller] Radarr ${inst.id} history error:`, err.message);
          return { instance: inst.id, data: { records: [] } };
        })
      ))),
      timed('Radarr Tags', () => Promise.all(radarrInstances.map(inst =>
        axios.get(`${inst.url}/api/v3/tag`, {
          headers: { 'X-Api-Key': inst.apiKey }
        }).then(res => ({ instance: inst.id, data: res.data })).catch(err => {
          console.error(`[Poller] Radarr ${inst.id} tags error:`, err.message);
          return { instance: inst.id, data: [] };
        })
      ))),
      timed('qBittorrent', () => getTorrents().catch(err => {
        console.error(`[Poller] qBittorrent error:`, err.message);
        return [];
      }))
    ]);

    const [
      { result: sabQueues }, { result: sabHistories },
      { result: sonarrTagsResults }, { result: sonarrQueues },
      { result: sonarrHistories },
      { result: radarrQueues }, { result: radarrHistories },
      { result: radarrTagsResults },
      { result: qbittorrentTorrents }
    ] = results;

    // Store per-task timings
    const totalMs = Date.now() - start;
    lastPollTimings = {
      totalMs,
      timestamp: new Date().toISOString(),
      tasks: results.map(r => ({ label: r.label, ms: r.ms }))
    };

    // When polling is active, TTL is 3x interval to avoid gaps between polls
    // When polling is disabled (on-demand), use 30s so data refreshes on next request after expiry
    const cacheTTL = POLLING_ENABLED ? POLL_INTERVAL * 3 : 30000;

    // SABnzbd
    const firstSabQueue = sabQueues[0] && sabQueues[0].data && sabQueues[0].data.queue;
    cache.set('poll:sab-queue', {
      slots: sabQueues.flatMap(q => (q.data.queue && q.data.queue.slots) || []),
      status: firstSabQueue && firstSabQueue.status,
      speed: firstSabQueue && firstSabQueue.speed,
      kbpersec: firstSabQueue && firstSabQueue.kbpersec
    }, cacheTTL);

    cache.set('poll:sab-history', {
      slots: sabHistories.flatMap(h => (h.data.history && h.data.history.slots) || [])
    }, cacheTTL);

    // Sonarr
    cache.set('poll:sonarr-tags', sonarrTagsResults, cacheTTL);
    // Tag queue/history records with _instanceUrl so embedded series/movie objects can build links
    cache.set('poll:sonarr-queue', {
      records: sonarrQueues.flatMap(q => {
        const inst = sonarrInstances.find(i => i.id === q.instance);
        const url = inst ? inst.url : null;
        return (q.data.records || []).map(r => {
          if (r.series) r.series._instanceUrl = url;
          return r;
        });
      })
    }, cacheTTL);
    cache.set('poll:sonarr-history', {
      records: sonarrHistories.flatMap(h => h.data.records || [])
    }, cacheTTL);

    // Radarr
    cache.set('poll:radarr-queue', {
      records: radarrQueues.flatMap(q => {
        const inst = radarrInstances.find(i => i.id === q.instance);
        const url = inst ? inst.url : null;
        return (q.data.records || []).map(r => {
          if (r.movie) r.movie._instanceUrl = url;
          return r;
        });
      })
    }, cacheTTL);
    cache.set('poll:radarr-history', {
      records: radarrHistories.flatMap(h => h.data.records || [])
    }, cacheTTL);
    cache.set('poll:radarr-tags', radarrTagsResults.flatMap(t => t.data || []), cacheTTL);

    // qBittorrent
    cache.set('poll:qbittorrent', qbittorrentTorrents, cacheTTL);

    const elapsed = Date.now() - start;
    console.log(`[Poller] Poll complete in ${elapsed}ms`);

    // Notify all SSE stream connections so they push fresh data immediately
    for (const cb of pollSubscribers) {
      try { cb(); } catch { /* subscriber already disconnected */ }
    }
  } catch (err) {
    console.error(`[Poller] Poll error:`, err.message);
  } finally {
    polling = false;
  }
}

let intervalHandle = null;

function startPoller() {
  if (!POLLING_ENABLED) {
    console.log(`[Poller] Background polling disabled (POLL_INTERVAL=${process.env.POLL_INTERVAL || 'not set'}). Data will be fetched on-demand.`);
    return;
  }
  console.log(`[Poller] Starting background poller (interval: ${POLL_INTERVAL}ms)`);
  // Run immediately, then on interval
  pollAllServices();
  intervalHandle = setInterval(pollAllServices, POLL_INTERVAL);
}

function stopPoller() {
  if (intervalHandle) {
    clearInterval(intervalHandle);
    intervalHandle = null;
    console.log('[Poller] Stopped');
  }
}

function getLastPollTimings() {
  return lastPollTimings;
}

module.exports = { startPoller, stopPoller, pollAllServices, getLastPollTimings, onPollComplete, offPollComplete, POLL_INTERVAL, POLLING_ENABLED };