xtream bundler

Dockers/xtream-ppv-bundler/src/scheduler.js

@@ -0,0 +1,230 @@
+'use strict';
+
+const { settings } = require('./settings');
+const logger = require('./logger');
+const { fetchLiveStreams } = require('./xtream-client');
+const { buildEvents, isEligible, compareEvents } = require('./events');
+
+function createState(maxChannels = settings.maxChannels) {
+  return {
+    slots: Array.from({ length: maxChannels }, () => ({ eventKey: null, assignedAt: null })),
+    events: new Map(),
+    upcoming: [],
+    nextSeq: 0,
+    warnings: [],
+    lastRefreshAt: null,
+    lastRefreshOk: false,
+    consecutiveFailures: 0,
+    lastError: null,
+  };
+}
+
+function ensureSlots(state, maxChannels) {
+  while (state.slots.length < maxChannels) {
+    state.slots.push({ eventKey: null, assignedAt: null });
+  }
+  while (state.slots.length > maxChannels) {
+    state.slots.pop();
+  }
+}
+
+function releaseSlot(state, eventKey) {
+  for (const slot of state.slots) {
+    if (slot.eventKey === eventKey) {
+      slot.eventKey = null;
+      slot.assignedAt = null;
+    }
+  }
+}
+
+function mergeSources(existing, fresh) {
+  const result = existing ? existing.slice() : [];
+  const seen = new Set(result.map((source) => source.streamId));
+  for (const source of fresh || []) {
+    if (seen.has(source.streamId)) continue;
+    seen.add(source.streamId);
+    result.push(source);
+  }
+  return result;
+}
+
+function reconcile(state, currentEvents, now, options) {
+  const { maxChannels, startGraceMs } = options;
+  ensureSlots(state, maxChannels);
+
+  const assignedNow = new Set(
+    state.slots.filter((slot) => slot.eventKey).map((slot) => slot.eventKey)
+  );
+
+  for (const [key, event] of [...state.events]) {
+    const fresh = currentEvents.get(key);
+    if (fresh) {
+      event.sources = mergeSources(event.sources, fresh.sources);
+      event.title = fresh.title;
+      event.league = fresh.league;
+      event.endAt = fresh.endAt;
+      event.missingCount = 0;
+      if (assignedNow.has(key)) event.status = 'assigned';
+    } else {
+      event.missingCount = (event.missingCount || 0) + 1;
+    }
+
+    if (now >= event.endAt.getTime()) {
+      releaseSlot(state, key);
+      assignedNow.delete(key);
+      state.events.delete(key);
+      continue;
+    }
+
+    if (!assignedNow.has(key) && event.missingCount >= 2) {
+      state.events.delete(key);
+    }
+  }
+
+  for (const [key, fresh] of currentEvents) {
+    if (state.events.has(key)) continue;
+    if (!isEligible(fresh, now, { startGraceMs })) continue;
+    if (!fresh.sources.length) continue;
+    state.events.set(key, {
+      ...fresh,
+      firstSeenSeq: state.nextSeq++,
+      missingCount: 0,
+      status: 'queued',
+    });
+  }
+
+  const assigned = new Set(
+    state.slots.filter((slot) => slot.eventKey).map((slot) => slot.eventKey)
+  );
+  const waiting = [...state.events.values()]
+    .filter((event) => !assigned.has(event.key) && isEligible(event, now, { startGraceMs }))
+    .sort(compareEvents);
+
+  for (const slot of state.slots) {
+    if (slot.eventKey) continue;
+    const next = waiting.shift();
+    if (!next) break;
+    slot.eventKey = next.key;
+    slot.assignedAt = now;
+    next.status = 'assigned';
+  }
+
+  return state;
+}
+
+function summarize(state) {
+  const assignedKeys = new Set(
+    state.slots.filter((slot) => slot.eventKey).map((slot) => slot.eventKey)
+  );
+  const eventView = (event) => ({
+    key: event.key,
+    title: event.title,
+    league: event.league,
+    language: event.language,
+    startAt: event.startAt.toISOString(),
+    endAt: event.endAt.toISOString(),
+    status: event.status,
+    sourceCount: event.sources.length,
+    sourceNames: event.sources.map((source) => source.name),
+  });
+  return {
+    slots: state.slots.map((slot, index) => ({
+      slot: index + 1,
+      assigned: Boolean(slot.eventKey),
+      event: slot.eventKey && state.events.has(slot.eventKey)
+        ? eventView(state.events.get(slot.eventKey))
+        : null,
+    })),
+    queue: [...state.events.values()]
+      .filter((event) => !assignedKeys.has(event.key))
+      .sort(compareEvents)
+      .map(eventView),
+    warnings: state.warnings,
+    lastRefreshAt: state.lastRefreshAt,
+    lastRefreshOk: state.lastRefreshOk,
+    consecutiveFailures: state.consecutiveFailures,
+    lastError: state.lastError,
+  };
+}
+
+function createScheduler() {
+  const state = createState();
+  let refreshTimer = null;
+  let refreshing = false;
+
+  async function refreshOnce() {
+    if (refreshing) return { ok: false, skipped: true };
+    refreshing = true;
+    try {
+      const streams = await fetchLiveStreams();
+      const { events, warnings } = buildEvents(streams, {
+        searchTerm: settings.searchTerm,
+        language: settings.language,
+        timezone: settings.timezone,
+        eventDurationMs: settings.eventDurationMs,
+      });
+      const now = Date.now();
+      reconcile(state, events, now, {
+        maxChannels: settings.maxChannels,
+        startGraceMs: settings.startGraceMs,
+      });
+      state.warnings = warnings.slice(0, 50);
+      state.upcoming = [...events.values()].sort(compareEvents);
+      state.lastRefreshAt = new Date(now).toISOString();
+      state.lastRefreshOk = true;
+      state.consecutiveFailures = 0;
+      state.lastError = null;
+      const filled = state.slots.filter((slot) => slot.eventKey).length;
+      logger.info(
+        `Refresh ok: ${events.size} matching event(s), ${filled}/${state.slots.length} slot(s) filled`
+      );
+      for (const warning of state.warnings) {
+        logger.debug('Unparseable matching stream:', warning.name);
+      }
+      return { ok: true };
+    } catch (error) {
+      state.consecutiveFailures += 1;
+      state.lastRefreshOk = false;
+      state.lastError = logger.redact(error.message || String(error));
+      logger.error('Refresh failed:', state.lastError);
+      return { ok: false, error };
+    } finally {
+      refreshing = false;
+    }
+  }
+
+  function startRefreshLoop() {
+    stopRefreshLoop();
+    refreshTimer = setInterval(() => {
+      refreshOnce().catch(() => {});
+    }, settings.refreshIntervalSec * 1000);
+    if (refreshTimer.unref) refreshTimer.unref();
+  }
+
+  function stopRefreshLoop() {
+    if (refreshTimer) {
+      clearInterval(refreshTimer);
+      refreshTimer = null;
+    }
+  }
+
+  return {
+    state,
+    refreshOnce,
+    startRefreshLoop,
+    stopRefreshLoop,
+    getSlotEvent(index) {
+      const slot = state.slots[index];
+      if (!slot || !slot.eventKey) return null;
+      return state.events.get(slot.eventKey) || null;
+    },
+    getSlotCount() {
+      return state.slots.length;
+    },
+    summarize() {
+      return summarize(state);
+    },
+  };
+}
+
+module.exports = { createState, reconcile, mergeSources, summarize, createScheduler };