323 lines
12 KiB
JavaScript
323 lines
12 KiB
JavaScript
// Copyright (c) 2026 Gordon Bolton. MIT License.
|
|
const axios = require('axios');
|
|
const cache = require('./cache');
|
|
const { initializeClients, getAllDownloads, getDownloadsByClientType } = require('./downloadClients');
|
|
const arrRetrieverRegistry = require('./arrRetrievers');
|
|
const {
|
|
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;
|
|
|
|
// Webhook fallback timeout in minutes (default 10)
|
|
const WEBHOOK_FALLBACK_TIMEOUT_MINUTES = parseInt(process.env.WEBHOOK_FALLBACK_TIMEOUT, 10) || 10;
|
|
const WEBHOOK_FALLBACK_TIMEOUT_MS = WEBHOOK_FALLBACK_TIMEOUT_MINUTES * 60 * 1000;
|
|
|
|
// Webhook poll interval multiplier when webhooks are active (default 3x)
|
|
const WEBHOOK_POLL_INTERVAL_MULTIPLIER = parseInt(process.env.WEBHOOK_POLL_INTERVAL_MULTIPLIER, 10) || 3;
|
|
|
|
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 };
|
|
}
|
|
|
|
// Helper function to determine if instance polling should be skipped
|
|
function shouldSkipInstancePolling(instances, instanceType) {
|
|
if (!instances || instances.length === 0) {
|
|
return false;
|
|
}
|
|
|
|
const now = Date.now();
|
|
let allInstancesHaveRecentWebhooks = true;
|
|
let skippedCount = 0;
|
|
|
|
for (const instance of instances) {
|
|
const metrics = cache.getWebhookMetrics(instance.url);
|
|
|
|
// Skip polling if:
|
|
// 1. Webhook events have been received (eventsReceived > 0)
|
|
// 2. Last webhook was recent (within fallback timeout)
|
|
// 3. Webhook has been enabled (we have metrics)
|
|
const hasWebhookActivity = metrics && metrics.eventsReceived > 0;
|
|
const isRecent = metrics && metrics.lastWebhookTimestamp && (now - metrics.lastWebhookTimestamp) < WEBHOOK_FALLBACK_TIMEOUT_MS;
|
|
|
|
if (hasWebhookActivity && isRecent) {
|
|
skippedCount++;
|
|
cache.incrementPollsSkipped(instance.url);
|
|
} else {
|
|
allInstancesHaveRecentWebhooks = false;
|
|
}
|
|
}
|
|
|
|
if (allInstancesHaveRecentWebhooks && skippedCount > 0) {
|
|
console.log(`[Poller] Skipping ${instanceType} polling for ${skippedCount} instance(s) with active webhooks`);
|
|
return true;
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
async function pollAllServices() {
|
|
if (polling) {
|
|
console.log('[Poller] Previous poll still running, skipping');
|
|
return;
|
|
}
|
|
polling = true;
|
|
const start = Date.now();
|
|
|
|
try {
|
|
// Ensure download clients and *arr retrievers are initialized
|
|
await initializeClients();
|
|
await arrRetrieverRegistry.initialize();
|
|
|
|
const sonarrInstances = getSonarrInstances();
|
|
const radarrInstances = getRadarrInstances();
|
|
|
|
// Check webhook fallback: if no webhook events for WEBHOOK_FALLBACK_TIMEOUT, force full poll
|
|
const globalMetrics = cache.getGlobalWebhookMetrics();
|
|
const now = Date.now();
|
|
const lastWebhookTime = globalMetrics.lastGlobalWebhookTimestamp;
|
|
const fallbackTriggered = lastWebhookTime && (now - lastWebhookTime) > WEBHOOK_FALLBACK_TIMEOUT_MS;
|
|
|
|
if (fallbackTriggered) {
|
|
console.log(`[Poller] Webhook fallback triggered: no webhook events for ${WEBHOOK_FALLBACK_TIMEOUT_MINUTES} minutes, forcing full poll`);
|
|
}
|
|
|
|
// Determine which instances should be polled based on webhook activity
|
|
const shouldPollSonarr = fallbackTriggered || !shouldSkipInstancePolling(sonarrInstances, 'sonarr');
|
|
const shouldPollRadarr = fallbackTriggered || !shouldSkipInstancePolling(radarrInstances, 'radarr');
|
|
|
|
// All fetches in parallel, each individually timed
|
|
const results = await Promise.all([
|
|
timed('Download Clients', async () => {
|
|
const downloadsByType = await getDownloadsByClientType();
|
|
return downloadsByType;
|
|
}),
|
|
shouldPollSonarr ? timed('Sonarr Tags', async () => {
|
|
const tagsByType = await arrRetrieverRegistry.getTagsByType();
|
|
return tagsByType.sonarr || [];
|
|
}) : timed('Sonarr Tags', async () => []),
|
|
shouldPollSonarr ? timed('Sonarr Queue', async () => {
|
|
const queuesByType = await arrRetrieverRegistry.getQueuesByType();
|
|
return queuesByType.sonarr || [];
|
|
}) : timed('Sonarr Queue', async () => []),
|
|
shouldPollSonarr ? timed('Sonarr History', async () => {
|
|
const historyByType = await arrRetrieverRegistry.getHistoryByType({ pageSize: 10 });
|
|
return historyByType.sonarr || [];
|
|
}) : timed('Sonarr History', async () => []),
|
|
shouldPollRadarr ? timed('Radarr Queue', async () => {
|
|
const queuesByType = await arrRetrieverRegistry.getQueuesByType();
|
|
return queuesByType.radarr || [];
|
|
}) : timed('Radarr Queue', async () => []),
|
|
shouldPollRadarr ? timed('Radarr History', async () => {
|
|
const historyByType = await arrRetrieverRegistry.getHistoryByType({ pageSize: 10 });
|
|
return historyByType.radarr || [];
|
|
}) : timed('Radarr History', async () => []),
|
|
shouldPollRadarr ? timed('Radarr Tags', async () => {
|
|
const tagsByType = await arrRetrieverRegistry.getTagsByType();
|
|
return tagsByType.radarr || [];
|
|
}) : timed('Radarr Tags', async () => []),
|
|
]);
|
|
|
|
const [
|
|
{ result: downloadsByType },
|
|
{ result: sonarrTagsResults }, { result: sonarrQueues },
|
|
{ result: sonarrHistories },
|
|
{ result: radarrQueues }, { result: radarrHistories },
|
|
{ result: radarrTagsResults }
|
|
] = 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;
|
|
|
|
// Download Clients (SABnzbd, qBittorrent, Transmission)
|
|
// Preserve backward compatibility with existing cache keys
|
|
const sabnzbdDownloads = downloadsByType.sabnzbd || [];
|
|
const qbittorrentDownloads = downloadsByType.qbittorrent || [];
|
|
|
|
// SABnzbd - separate queue and history based on source
|
|
const sabQueue = sabnzbdDownloads.filter(d => d.raw && d.raw.source === 'queue');
|
|
const sabHistory = sabnzbdDownloads.filter(d => d.raw && d.raw.source === 'history');
|
|
|
|
// Transform SABnzbd downloads to legacy format for cache
|
|
const sabQueueLegacy = {
|
|
slots: sabQueue.map(d => ({
|
|
nzo_id: d.id,
|
|
filename: d.title,
|
|
status: d.status,
|
|
progress: d.progress / 100,
|
|
mb: d.size / (1024 * 1024),
|
|
mbleft: (d.size - d.downloaded) / (1024 * 1024),
|
|
kbpersec: d.speed / 1024,
|
|
timeleft: d.eta ? `${Math.floor(d.eta / 60)}:${String(Math.floor(d.eta % 60)).padStart(2, '0')}` : 'unknown',
|
|
cat: d.category,
|
|
labels: d.tags.join(','),
|
|
added: d.addedOn ? Math.floor(new Date(d.addedOn).getTime() / 1000) : null,
|
|
raw: d.raw
|
|
}))
|
|
};
|
|
|
|
const sabHistoryLegacy = {
|
|
slots: sabHistory.map(d => ({
|
|
nzo_id: d.id,
|
|
filename: d.title,
|
|
status: d.status,
|
|
mb: d.size / (1024 * 1024),
|
|
cat: d.category,
|
|
labels: d.tags.join(','),
|
|
added: d.addedOn ? Math.floor(new Date(d.addedOn).getTime() / 1000) : null,
|
|
raw: d.raw
|
|
}))
|
|
};
|
|
|
|
// Extract status from first SABnzbd download if available
|
|
const firstSabDownload = sabQueue[0];
|
|
const sabStatus = firstSabDownload ? {
|
|
status: 'Active',
|
|
speed: firstSabDownload.speed,
|
|
kbpersec: firstSabDownload.speed / 1024
|
|
} : { status: 'Idle', speed: 0, kbpersec: 0 };
|
|
|
|
cache.set('poll:sab-queue', {
|
|
...sabQueueLegacy,
|
|
...sabStatus
|
|
}, cacheTTL);
|
|
|
|
cache.set('poll:sab-history', sabHistoryLegacy, cacheTTL);
|
|
|
|
// qBittorrent - transform to legacy format
|
|
const qbittorrentLegacy = qbittorrentDownloads.map(d => ({
|
|
...d.raw,
|
|
instanceId: d.instanceId,
|
|
instanceName: d.instanceName
|
|
}));
|
|
|
|
cache.set('poll:qbittorrent', qbittorrentLegacy, cacheTTL);
|
|
|
|
// Sonarr
|
|
if (shouldPollSonarr) {
|
|
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;
|
|
const key = inst ? inst.apiKey : null;
|
|
return (q.data.records || []).map(r => {
|
|
if (r.series) r.series._instanceUrl = url;
|
|
r._instanceUrl = url;
|
|
r._instanceKey = key;
|
|
return r;
|
|
});
|
|
})
|
|
}, cacheTTL);
|
|
cache.set('poll:sonarr-history', {
|
|
records: sonarrHistories.flatMap(h => h.data.records || [])
|
|
}, cacheTTL);
|
|
} else {
|
|
// Extend TTL of existing cached data when polling is skipped
|
|
const existingSonarrTags = cache.get('poll:sonarr-tags');
|
|
const existingSonarrQueue = cache.get('poll:sonarr-queue');
|
|
const existingSonarrHistory = cache.get('poll:sonarr-history');
|
|
if (existingSonarrTags) cache.set('poll:sonarr-tags', existingSonarrTags, cacheTTL);
|
|
if (existingSonarrQueue) cache.set('poll:sonarr-queue', existingSonarrQueue, cacheTTL);
|
|
if (existingSonarrHistory) cache.set('poll:sonarr-history', existingSonarrHistory, cacheTTL);
|
|
}
|
|
|
|
// Radarr
|
|
if (shouldPollRadarr) {
|
|
cache.set('poll:radarr-queue', {
|
|
records: radarrQueues.flatMap(q => {
|
|
const inst = radarrInstances.find(i => i.id === q.instance);
|
|
const url = inst ? inst.url : null;
|
|
const key = inst ? inst.apiKey : null;
|
|
return (q.data.records || []).map(r => {
|
|
if (r.movie) r.movie._instanceUrl = url;
|
|
r._instanceUrl = url;
|
|
r._instanceKey = key;
|
|
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);
|
|
} else {
|
|
// Extend TTL of existing cached data when polling is skipped
|
|
const existingRadarrQueue = cache.get('poll:radarr-queue');
|
|
const existingRadarrHistory = cache.get('poll:radarr-history');
|
|
const existingRadarrTags = cache.get('poll:radarr-tags');
|
|
if (existingRadarrQueue) cache.set('poll:radarr-queue', existingRadarrQueue, cacheTTL);
|
|
if (existingRadarrHistory) cache.set('poll:radarr-history', existingRadarrHistory, cacheTTL);
|
|
if (existingRadarrTags) cache.set('poll:radarr-tags', existingRadarrTags, cacheTTL);
|
|
}
|
|
|
|
// qBittorrent (already set above in download clients section)
|
|
|
|
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 };
|