vyndr/src/snapshotScheduler.js

'use strict';

/**
 * snapshotScheduler — in-process snapshot cron (Session 45).
 *
 * No new dependency: a 1-minute unref'd interval that fires `runAllSnapshots`
 * at the configured UTC hours (default 14,19,22,1,3 = 10AM/3PM/6PM/9PM/11PM ET,
 * matching the sports cycle — morning research, afternoon news, pre-game lock,
 * in-game). Gated on SNAPSHOT_CRON=1 so it never runs in dev/test or on a
 * container that shouldn't own the schedule. Prefer an EXTERNAL cron (n8n) hitting
 * POST /api/internal/snapshot/all when running multiple API replicas — this
 * in-process variant assumes a single scheduler instance.
 */

const HOURS_UTC = (process.env.SNAPSHOT_HOURS_UTC || '14,19,22,1,3')
  .split(',')
  .map((n) => parseInt(n, 10))
  .filter((n) => Number.isInteger(n) && n >= 0 && n <= 23);

// Job 1 — per-sport cadence. The scheduler still FIRES at HOURS_UTC (and the
// missed-cron watchdog still references MLB, which runs at every grid hour), but
// each slot grades only the sports whose lines actually post/move then, and only
// PropLine-backed sports get the 20-min intraday refresh (soccer is odds-api
// quota-gated). All of that lives in the config table, not here.
const cadence = require('./config/sportCadence');

// Session 56 — missed-cron detection (pure, testable).
// The latest scheduled hour:00 (UTC) at or before `date`. null if none in 48h.
function mostRecentExpectedSlot(date, hours = HOURS_UTC) {
  const d = new Date(date.getTime());
  d.setUTCMinutes(0, 0, 0);
  for (let i = 0; i < 48; i += 1) {
    if (hours.includes(d.getUTCHours())) return new Date(d.getTime());
    d.setUTCHours(d.getUTCHours() - 1);
  }
  return null;
}

// True when we're >graceMin past the most recent expected slot AND the last
// recorded snapshot predates that slot (i.e. the run was missed). Never fires on
// cold boot (no lastSnapshotIso) — we don't cry wolf before the first snapshot.
function isSnapshotOverdue(lastSnapshotIso, now = new Date(), hours = HOURS_UTC, graceMin = 30) {
  const slot = mostRecentExpectedSlot(now, hours);
  if (!slot) return false;
  if (now.getTime() - slot.getTime() < graceMin * 60_000) return false;
  if (!lastSnapshotIso) return false;
  const last = new Date(lastSnapshotIso).getTime();
  return Number.isFinite(last) && last < slot.getTime();
}

function startSnapshotScheduler(opts = {}) {
  if (process.env.SNAPSHOT_CRON !== '1') {
    // Session 52 — log the disarmed state so container logs make it unambiguous
    // that the in-process cron is intentionally off (vs. crashed/missing).
    if (process.env.NODE_ENV !== 'test') {
      console.log(`[snapshotScheduler] disarmed — SNAPSHOT_CRON=${process.env.SNAPSHOT_CRON ?? 'unset'} (in-process cron off; use external cron → POST /api/internal/snapshot/all)`);
    }
    return null;
  }
  const runAll = opts.runAllSnapshots || require('./services/snapshotService').runAllSnapshots;
  // Session 55 — self-learning loop. Settle the PRIOR snapshot's grades against
  // real (now-completed) results BEFORE grading the fresh slate, so the accuracy
  // record reflects yesterday's games each cycle.
  const settleAll = opts.settleAllOutcomes || require('./services/outcomeService').settleAllOutcomes;
  // Session 58 — Phase 1 truth infrastructure: settle the persistent ledger
  // (outcome + actual + CLV) in the same pre-grade settle pass.
  const settleLedgers = opts.settleAllLedgers || require('./services/ledgerService').settleAllLedgers;
  const notify = opts.notify || require('./utils/opsNotify').notify;
  const cacheGet = opts.cacheGet || require('./utils/redis').cacheGet;
  const cacheSet = opts.cacheSet || require('./utils/redis').cacheSet;
  const now = opts.now || (() => new Date());
  // Session 8 (A1 board) — ops: the product watches itself. Pure logic lives
  // in services/opsWatch + services/systemHealth; this file only wires it.
  const opsWatch = require('./services/opsWatch');
  const getSystemHealth = opts.getSystemHealth || require('./services/systemHealth').getSystemHealth;
  const healthIssues = opts.healthIssues || require('./services/systemHealth').healthIssues;
  const getQuotaStatus = opts.getQuotaStatus || require('./services/quotaTracker').getQuotaStatus;
  const countLedgerRows = opts.countLedgerRows || require('./services/ledgerService').countRowsForDate;
  // Wave 1 — real-finals probe for the zero-settle alarm (NBA/WNBA only page
  // when yesterday actually had games). Reuses the free, cached ESPN scoreboard.
  const getSchedule = opts.getSchedule || require('./services/scheduleService').getSchedule;
  const failureTracker = opts.failureTracker || opsWatch.createFailureTracker();
  let lastFiredSlot = null;
  let lastOverdueSlot = null;
  let lastPulseDate = null;

  // Session 56 — missed-cron watchdog. Runs every minute (independent of the
  // fire schedule): if a scheduled slot came and went without a snapshot, alert
  // ONCE per missed slot.
  const checkOverdue = async () => {
    try {
      const d = now();
      const slot = mostRecentExpectedSlot(d);
      if (!slot) return;
      const slotKey = slot.toISOString();
      if (slotKey === lastOverdueSlot) return;   // already alerted for this slot
      const latest = await cacheGet('snapshot:mlb:latest');
      const lastTs = latest && latest.updated_at;
      if (isSnapshotOverdue(lastTs, d)) {
        lastOverdueSlot = slotKey;
        await notify(`⚠️ Snapshot OVERDUE — expected ${slot.getUTCHours()}:00 UTC, last was ${lastTs || 'never'}`, { title: 'VYNDR pipeline', priority: 'high', tags: ['warning'] });
      }
    } catch { /* watchdog must never throw */ }
  };

  // Session 8 — daily 9 AM ET pulse (13:00 UTC), ONE notification. Dedupe is
  // belt-and-braces: in-process date + a Redis key that survives a restart
  // inside the same day. Runs on the per-minute cadence, independent of slots.
  const PULSE_HOUR_UTC = Number.parseInt(process.env.PULSE_HOUR_UTC || '13', 10);
  const pulseTick = async () => {
    try {
      const d = now();
      if (d.getUTCHours() !== PULSE_HOUR_UTC || d.getUTCMinutes() !== 0) return;
      const dayKey = d.toISOString().slice(0, 10);
      if (dayKey === lastPulseDate) return;
      lastPulseDate = dayKey;
      if (await cacheGet(`ops:pulse:${dayKey}`)) return;   // already sent today (pre-restart)
      await cacheSet(`ops:pulse:${dayKey}`, '1', 2 * 24 * 3600);

      const yesterdayEt = opsWatch.dateET(new Date(d.getTime() - 24 * 3600 * 1000));
      let ledgerRows = null;
      try { ledgerRows = await countLedgerRows(yesterdayEt); } catch { /* n/a */ }
      let settles24h = null;
      try {
        const logs = [];
        for (const sp of ['mlb', 'nba', 'wnba', 'soccer']) {
          const raw = await cacheGet(`outcomes:${sp}:log`);
          logs.push(Array.isArray(raw) ? raw : (raw && raw.log) || []);
        }
        settles24h = opsWatch.countRecentSettles(logs, d.getTime());
      } catch { /* n/a */ }
      let quota = null;
      try { quota = await getQuotaStatus('odds-api'); } catch { /* n/a */ }
      let health = null;
      try { health = await getSystemHealth(); } catch { /* n/a */ }

      await notify(
        opsWatch.buildPulseMessage({ dateEt: opsWatch.dateET(d), ledgerRows, settles24h, quota, health }),
        { title: 'VYNDR daily pulse', tags: ['newspaper'] },
      );
      // Box health pages separately at high priority — the pulse is a read,
      // the threshold breach is an action item.
      const issues = healthIssues(health);
      if (issues.length > 0) {
        await notify(`Box health: ${issues.join('; ')}. The record does not fit on a full disk.`, {
          title: 'VYNDR box', priority: 'high', tags: ['warning'],
        });
      }
    } catch { /* the pulse must never break the scheduler */ }
  };

  const tick = async () => {
    await checkOverdue();
    await pulseTick();
    const d = now();
    if (d.getUTCMinutes() !== 0) return;
    const h = d.getUTCHours();
    if (!HOURS_UTC.includes(h)) return;
    const slot = `${d.getUTCFullYear()}-${d.getUTCMonth()}-${d.getUTCDate()}-${h}`;
    if (slot === lastFiredSlot) return; // fire once per slot
    lastFiredSlot = slot;
    // Session 8 — a settle pass that THROWS is a paged event, not a log line.
    // Settlement failing quietly is how the record dies.
    try {
      const settled = await settleAll();
      const totalSettled = settled.reduce((n, r) => n + (r.settled || 0), 0);
      console.log(`[outcomes] cron fired ${h}:00 UTC — ${totalSettled} props settled vs real results`);
    } catch (e) {
      console.warn('[outcomes] settle run failed:', e.message);
      await notify(`Outcome settlement THREW at ${h}:00 UTC — ${e.message}. Yesterday's grades are unsettled until the next slot.`, {
        title: 'VYNDR settlement', priority: 'high', tags: ['rotating_light'],
      });
    }
    let ledgerResults = null;
    try {
      ledgerResults = await settleLedgers();
      const n = ledgerResults.reduce((t, r) => t + (r.settled || 0), 0);
      const nv = ledgerResults.reduce((t, r) => t + (r.voided || 0), 0);
      const nu = ledgerResults.reduce((t, r) => t + (r.unrecoverable || 0), 0);
      console.log(`[ledger] settle pass — ${n} settled, ${nv} voided, ${nu} unrecoverable (outcome + CLV)`);
      // Session 64 — SETTLEMENT-RATE ALARM. zeroSettleAlarm only catches a
      // TOTAL zero; the diagnostic found ~30% of a slate quietly failing while
      // the pass "succeeded". A persistently low resolution rate is a broken
      // pipe, not a quiet night.
      try {
        const sr = opsWatch.settlementRateAlarm(ledgerResults);
        if (sr.alarm) {
          await notify(`Settlement rate BELOW FLOOR at ${h}:00 UTC — ${sr.reason}. Rows that cannot resolve are pending, not scored.`, {
            title: 'VYNDR settlement', priority: 'high', tags: ['rotating_light'],
          });
        }
      } catch { /* alarm evaluation must never break the tick */ }
    } catch (e) {
      console.warn('[ledger] settle run failed:', e.message);
      await notify(`Ledger settlement THREW at ${h}:00 UTC — ${e.message}. The public record did not advance.`, {
        title: 'VYNDR settlement', priority: 'high', tags: ['rotating_light'],
      });
    }
    // Session 8 — zero-settle alarm, MORNING slot only (the book-closing pass).
    // Signal = the ledger settle's own Postgres-backed return values (see
    // opsWatch.zeroSettleAlarm for why not snapshot:{sport}:previous). Deduped
    // once per ET date; a genuinely empty yesterday (0 rows fetched) never fires.
    try {
      if (h === opsWatch.morningHourUtc(HOURS_UTC) && ledgerResults) {
        // Wave 1 — probe yesterday's ESPN scoreboard so an NBA/WNBA off-day
        // (0 finals) never false-pages "settled 0" on stale pending rows.
        const yEt = opsWatch.dateET(new Date(d.getTime() - 24 * 3600 * 1000));
        const finalsBySport = {};
        for (const fsp of opsWatch.FINALS_GATED_SPORTS) {
          try {
            const games = await getSchedule(fsp, yEt);
            finalsBySport[fsp] = Array.isArray(games) && games.some((g) => g && g.status === 'post');
          } catch { finalsBySport[fsp] = false; }
        }
        const z = opsWatch.zeroSettleAlarm(ledgerResults, { finalsBySport });
        if (z.alarm) {
          const dk = `ops:settle_zero:${opsWatch.dateET(d)}`;
          if (!(await cacheGet(dk))) {
            await cacheSet(dk, '1', 2 * 24 * 3600);
            await notify(`Morning settle pass closed with 0 settles — ${z.pending} ledger rows from before today are still pending. Yesterday had graded rows; the record did not advance.`, {
              title: 'VYNDR settlement', priority: 'high', tags: ['rotating_light'],
            });
          }
        }
      }
    } catch { /* alarm evaluation must never break the tick */ }
    try {
      // Job 1 — only the sports whose cadence includes this hour. MLB runs every
      // grid hour; WNBA skips 14/3 UTC (props not posted); soccer runs 14/19 only.
      const scheduled = cadence.sportsForHour(h);
      const results = await runAll({ sports: scheduled });
      const ok = results.filter((r) => r.status === 'ok');
      console.log(`[snapshot] cron fired ${h}:00 UTC — ${ok.length}/${results.length} sports graded (scheduled: ${scheduled.join(',') || 'none'})`);
      // Session 63 (A1-S4) — after the day's FIRST slot, tell Kev the media
      // pack is ready. One ping, only when something actually graded.
      if (h === HOURS_UTC[0] && ok.length > 0) {
        const total = results.reduce((n, r) => n + (r.gradeCount || 0), 0);
        await notify(`Desk pack ready — ${total} props graded across ${ok.length} sports. vyndr.app/desk`, { title: 'VYNDR desk', tags: ['newspaper'] });
      }
      // Session 64 — RETENTION ZERO-WRITE ALARM. Features are irreplaceable: a
      // night not captured cannot be reconstructed. Retention is best-effort so
      // it never breaks a snapshot, which means a broken write is silent —
      // this is the counterweight, at missed-snapshot severity.
      try {
        const written = {};
        for (const r of results) written[r.sport] = r.retentionRows;
        const rz = opsWatch.retentionZeroWriteAlarm(results, written);
        if (rz.alarm) {
          await notify(`RETENTION NOT CAPTURING at ${h}:00 UTC — ${rz.reason}. Feature vectors for this slate are irreplaceable and are being lost.`, {
            title: 'VYNDR retention', priority: 'high', tags: ['rotating_light'],
          });
        }
      } catch { /* alarm evaluation must never break the tick */ }

      // Session 8 — persistent-failure pager: 3+ CONSECUTIVE erroring slots for
      // a sport pages once (single-slot errors are normal before lines post).
      for (const r of results) {
        const t = failureTracker.record(r.sport, r);
        if (t.shouldPage) {
          await notify(`${String(r.sport).toUpperCase()} snapshot has produced nothing for ${t.count} consecutive slots (latest: ${r.status}${r.reason ? ` — ${r.reason}` : ''}). One empty slot is a quiet book; ${t.count} is a broken pipe.`, {
            title: 'VYNDR pipeline', priority: 'high', tags: ['rotating_light'],
          });
        }
      }
    } catch (e) {
      console.warn('[snapshot] cron run failed:', e.message);
    }
    // Session 64 — NIGHTLY BACKTEST HARNESS, on our own scheduler. Runs once
    // per day at HARNESS_HOUR_UTC and appends to harness_results, so the
    // calibration trend is visible as retention compounds. An
    // INSUFFICIENT_HISTORY verdict is expected and correct.
    try {
      const harnessHour = Number(process.env.HARNESS_HOUR_UTC || 14);
      if (h === harnessHour) {
        const runner = opts.harnessRunner || require('./services/harnessRunner');
        const res = await runner.runAndRecord();
        console.log(`[harness] nightly run — verdict=${res.verdict || 'n/a'} scored=${res.scored ?? 'n/a'}${res.error ? ` ERROR: ${res.error}` : ''}`);
        // A validator that stops running looks exactly like one that keeps
        // passing — so staleness pages.
        const stale = opsWatch.harnessStaleAlarm(await runner.lastRunAt(), now());
        if (stale.alarm) {
          await notify(`Backtest harness stale — ${stale.reason}. The metric gate is not running.`, {
            title: 'VYNDR harness', priority: 'high', tags: ['rotating_light'],
          });
        }
      }
    } catch (e) { console.warn('[harness] nightly run failed:', e.message); }

    // Session 74 — GAME-LEVEL CONTEXT. One schedule call per game DATE (not per
    // game, not per prop) capturing venue + final runs + HR totals for the day
    // that just completed. Forward-only: a day not captured is permanently
    // lost, which is why this runs beside the settle pass rather than waiting
    // until a self-derived park factor is wanted.
    try {
      const gc = opts.gameContext || require('./services/gameContext');
      const d = new Date(now().getTime() - 24 * 3600 * 1000);
      const ymd = d.toISOString().slice(0, 10);
      const r = await gc.captureDate(ymd, { sport: 'mlb' });
      if (r && r.games) console.log(`[gamectx] ${ymd} — ${r.written}/${r.games} games, ${r.venues} venues, ${r.with_hr} with HR totals`);
    } catch (e) { console.warn('[gamectx] capture failed:', e.message); }

    // Session 70 — THE MEASUREMENT INSTRUMENT. Attach the de-vigged CLOSING
    // probability to locked rows from the append-only closing_captures, so
    // p_win, the close, the archetype vector and the outcome all land on ONE
    // joinable record. Runs beside the settle pass; write-once, and a past game
    // with no usable capture is marked market-unavailable rather than imputed.
    try {
      const led = opts.ledgerService || require('./services/ledgerService');
      for (const sp of cadence.ALL_SPORTS) {
        const r = await led.attachClosingProb(sp, {});
        if (r && (r.updated || r.absent)) {
          console.log(`[instrument] ${sp} — closes attached ${r.updated}, market-unavailable ${r.absent}`);
        }
      }
    } catch (e) { console.warn('[instrument] closing attach failed:', e.message); }

    // Session 68 — LAYER 1 MECHANISM DATA. Nightly full re-pull of the
    // ~1,350-row Statcast aggregate set at STATCAST_HOUR_UTC. Backfill and
    // refresh are the SAME call, upserted on the natural key, so the job is
    // idempotent and self-healing: a missed night self-corrects on the next
    // run with no incremental bookkeeping to drift. Kill switch: STATCAST=0.
    try {
      const statcastHour = Number(process.env.STATCAST_HOUR_UTC || 11); // ~7 AM ET, after every game is final
      if (h === statcastHour && process.env.STATCAST !== '0') {
        const agg = opts.statcastService || require('./services/statcastAggregateService');
        const res = await agg.refreshSeason({});
        console.log(`[statcast] nightly refresh — ok=${res.ok} rows=${res.rows ?? 'n/a'} written=${res.written ?? 0} joined=${res.joined ?? 'n/a'} thin=${res.thin ?? 'n/a'}${res.reason ? ` reason: ${res.reason}` : ''}`);
        if (!res.ok) {
          await notify(`Statcast refresh failed — ${res.reason || 'unknown'}. Mechanism data is not updating.`, {
            title: 'VYNDR statcast', priority: 'high', tags: ['rotating_light'],
          });
        }
        // Serving a stale aggregate as current is a quiet fabrication of
        // currency, so staleness pages on its own — separate from a failed run,
        // because a silently-not-scheduled job never produces a failure.
        const fresh = await agg.getFreshness({});
        if (agg.isStale(fresh)) {
          await notify(`Statcast aggregates stale — last updated ${fresh.updated_at} (${fresh.age_hours}h). Mechanism data is being served as current.`, {
            title: 'VYNDR statcast', priority: 'high', tags: ['rotating_light'],
          });
        }
      }
    } catch (e) { console.warn('[statcast] nightly refresh failed:', e.message); }

    // Session 8 — quota check after each snapshot run: odds-api >= 80% alerts
    // once per day (Redis-deduped). Never throws (guarded inside checkQuotaDaily).
    await opsWatch.checkQuotaDaily({ getStatus: getQuotaStatus, cacheGet, cacheSet, notify, now: () => now() });
  };

  // Session 60 (night2/D) — Phase 2.5 intraday refresh. Every
  // REFRESH_MINUTES during slate hours (noon–midnight ET), odds-only:
  // STEAM/VALUE badges, ≥1.0-against re-grades with public revisions,
  // refresh-fidelity closing capture, higher-frequency ticker MOVEs.
  // Skips the top of scheduled snapshot hours (the full run owns those).
  // Kill switch: INTRADAY_REFRESH=0.
  const REFRESH_MINUTES = Math.max(5, parseInt(process.env.INTRADAY_REFRESH_MINUTES || '20', 10) || 20);
  const refreshAll = opts.runAllIntradayRefreshes || require('./services/intradayRefreshService').runAllIntradayRefreshes;
  const inSlateHours = opts.inSlateHours || require('./services/intradayRefreshService').inSlateHours;
  let lastRefreshSlot = null;
  const refreshTick = async () => {
    if (process.env.INTRADAY_REFRESH === '0') return;
    const d = now();
    if (!inSlateHours(d)) return;
    if (d.getUTCMinutes() % REFRESH_MINUTES !== 0) return;
    if (d.getUTCMinutes() === 0 && HOURS_UTC.includes(d.getUTCHours())) return; // full snapshot owns this slot
    const slot = `${d.toISOString().slice(0, 13)}-${d.getUTCMinutes()}`;
    if (slot === lastRefreshSlot) return;
    lastRefreshSlot = slot;
    try {
      // Job 1 — only PropLine-backed sports refresh intraday (9k/day headroom).
      // Soccer is excluded: a 20-min odds-api refresh would drain the 500/mo key.
      const results = await refreshAll({ sports: cadence.intradaySports() });
      const touched = results.filter((r) => r.status === 'ok');
      const revised = results.reduce((n, r) => n + (r.revised || 0), 0);
      if (touched.length > 0) console.log(`[intraday] refresh — ${touched.length} sports, ${revised} public revisions`);
    } catch (e) {
      console.warn('[intraday] refresh failed:', e.message);
    }
  };

  const interval = setInterval(() => { void tick(); void refreshTick(); }, 60_000);
  if (interval.unref) interval.unref();
  console.log(`[snapshotScheduler] armed — SNAPSHOT_CRON=${process.env.SNAPSHOT_CRON}, hours=${HOURS_UTC.join(',')} UTC, intraday=${process.env.INTRADAY_REFRESH === '0' ? 'off' : `${REFRESH_MINUTES}m (slate hours)`}, statcast=${process.env.STATCAST === '0' ? 'off' : `${process.env.STATCAST_HOUR_UTC || 11}h UTC`}`);
  // Job 1 — state each sport's cadence at boot so "when does WNBA grade?" is
  // answerable from logs, and warn if a sport is scheduled at an hour the
  // scheduler never fires (it would silently never run).
  const cadenceSummary = cadence.ALL_SPORTS.map((s) => `${s}=${cadence.hoursFor(s).join('/') || '—'}${cadence.SPORT_CADENCE[s].intraday ? '' : ' (no intraday)'}`).join('  ');
  console.log(`[snapshotScheduler] per-sport cadence — ${cadenceSummary}`);
  const orphanHours = cadence.ALL_HOURS.filter((h) => !HOURS_UTC.includes(h));
  if (orphanHours.length > 0) {
    console.warn(`[snapshotScheduler] WARNING — sports scheduled at ${orphanHours.join(',')} UTC but the scheduler never fires then (not in hours=${HOURS_UTC.join(',')}); those slots will silently never run. Add them to SNAPSHOT_HOURS_UTC.`);
  }
  // Session 61 — settlement is scheduled IN THIS tick (settleAllOutcomes +
  // settleAllLedgers run FIRST at every snapshot slot, before grading). It
  // was invisible at boot, which made "is settlement scheduled?" unanswerable
  // from logs. This line makes it verifiable forever.
  // Wave 1 — announce settlement PER settleable sport so "does NBA settle?" is
  // answerable from boot logs (mlb=statsapi, nba/wnba=ESPN game logs).
  const settleTags = require('./services/opsWatch').SETTLEABLE_SPORTS.map((s) => `[settle:${s}]`).join(' ');
  console.log(`[settlement] armed — ${settleTags} outcomes + ledger settle pass runs FIRST at each snapshot slot (${HOURS_UTC.join(',')} UTC), idempotent re-runs`);
  // Session 8 — same verifiability rule: every watchdog states itself at boot.
  console.log(`[opsWatch] armed — settle alarms (throw + morning zero-settle), failure pager (${failureTracker.threshold} consecutive), quota daily check, pulse ${PULSE_HOUR_UTC}:00 UTC`);
  return { interval, tick, refreshTick, pulseTick };
}

module.exports = { startSnapshotScheduler, HOURS_UTC, mostRecentExpectedSlot, isSnapshotOverdue };