vyndr/src/services/snapshotService.js

/**
 * snapshotService — scheduled grade pipeline (Session 45).
 *
 * Orchestrates ONE snapshot cycle for one sport. The on-demand "Read" model is
 * retired: a snapshot pre-grades the full slate, LOCKS each grade to the line at
 * snapshot time (`gradedAt`), classifies each player's archetype, computes line
 * deltas vs the previous snapshot, and emits ticker events.
 *
 * Everything is orchestration of EXISTING services (oddsService,
 * gradeSlateService, archetypeService, playerIntelService). All I/O is injectable
 * so the whole cycle is unit-testable with zero network.
 *
 * Redis keys written:
 *   snapshot:{sport}:latest    — current locked snapshot { sport, updated_at, grades, deltas }
 *   snapshot:{sport}:previous  — prior snapshot (for the next delta computation)
 *   grades:{sport}             — { grades, updated_at, source } (GameCard / Explore / leaders)
 *   ticker:items               — capped array of ticker events (newest first)
 */

// Session 60 (AUTONOMY P0) — 24h, NOT 6h. The overnight cron gap is 11 hours
// (03:00 UTC → 14:00 UTC); with a 6h TTL the morning settle pass read an
// EXPIRED snapshot and the accuracy loop silently settled nothing. The
// snapshot must survive until the morning-after settle reads it.
const SNAP_TTL = 24 * 3600;
const TICKER_TTL = 24 * 3600;
const TICKER_CAP = 50;
const DELTA_NOISE = 0.5;        // ignore movements smaller than this
const DELTA_MOVE = 1.0;         // ticker MOVE threshold
const STATS_CONCURRENCY = 5;

const { nameKey, normalizeName } = require('../utils/playerName');
// Session 46 — group/dedupe by the normalized name key so "A.J. Ewing" and
// "AJ Ewing" (or "Jazz Chisholm" / "Jazz Chisholm Jr.") collapse to one player.
const norm = (s) => nameKey(s);
const lastName = (full) => {
  const parts = String(full || '').trim().split(/\s+/);
  return parts.length > 1 ? parts[parts.length - 1] : (parts[0] || '');
};
const sideChar = (dir) => (String(dir || 'over').toLowerCase() === 'under' ? 'U' : 'O');
const propKey = (g) => `${norm(g.player || g.player_name)}|${String(g.stat_type || g.stat || '').toLowerCase()}|${String(g.direction || '').toLowerCase()}`;

async function mapLimit(items, concurrency, fn) {
  const out = new Array(items.length);
  let i = 0;
  async function worker() {
    while (i < items.length) {
      const idx = i++;
      out[idx] = await fn(items[idx], idx);
    }
  }
  await Promise.all(Array.from({ length: Math.min(concurrency, items.length || 1) }, worker));
  return out;
}

/** Index original odds props by propKey-ish (player|stat) for odds lookup.
 *  Session 61 — prefer a book row with BOTH sides priced: the first-seen row
 *  sometimes carried only one side (e.g. betmgm SB unders with no juice),
 *  which locked a NULL odds even though another book priced it. Still a
 *  REAL book row, never synthesized. */
function indexOdds(props) {
  const map = {};
  const bothSides = (p) => p && p.over_odds != null && p.under_odds != null;
  for (const p of props || []) {
    const k = `${norm(p.player)}|${String(p.stat_type || '').toLowerCase()}`;
    if (!map[k] || (!bothSides(map[k]) && bothSides(p))) map[k] = p;
  }
  return map;
}

function gradedAtFor(g, oddsByKey, ts) {
  const k = `${norm(g.player || g.player_name)}|${String(g.stat_type || g.stat || '').toLowerCase()}`;
  const o = oddsByKey[k];
  let odds = null;
  if (o) {
    odds = String(g.direction || '').toLowerCase() === 'under'
      ? (o.under_odds ?? o.under ?? o.odds ?? null)
      : (o.over_odds ?? o.over ?? o.odds ?? null);
  }
  return { line: g.line, odds, timestamp: ts };
}

/**
 * Compare current grades to the previous snapshot's locked lines. A delta is
 * emitted only when |movement| >= DELTA_NOISE. `direction`:
 *   'toward' = market moving in the direction of our graded side (confirming)
 *   'away'   = market moving against it.
 * For an OVER, a rising line confirms (toward); for an UNDER, a falling line.
 */
function computeLineDeltas(current, previous) {
  const prevMap = {};
  for (const p of previous || []) prevMap[propKey(p)] = p;
  // Session 52 — opt-in verification log (SNAPSHOT_DEBUG=1). Confirms the delta
  // pipeline has a previous snapshot to diff against; off by default (hot path).
  if (process.env.SNAPSHOT_DEBUG === '1') {
    console.log(`[deltas] diffing ${(current || []).length} current vs ${(previous || []).length} previous locked lines`);
  }
  const out = [];
  for (const c of current || []) {
    const prev = prevMap[propKey(c)];
    if (!prev) continue;
    const gradedLine = prev.gradedAt ? prev.gradedAt.line : prev.line;
    const currentLine = c.line;
    if (gradedLine == null || currentLine == null) continue;
    const delta = +(Number(currentLine) - Number(gradedLine)).toFixed(2);
    if (Math.abs(delta) < DELTA_NOISE) continue;
    const side = String(c.direction || 'over').toLowerCase();
    const toward = side === 'over' ? delta > 0 : delta < 0;
    out.push({
      player: c.player || c.player_name,
      stat: c.stat_type || c.stat,
      side: sideChar(side),
      gradedLine: Number(gradedLine),
      currentLine: Number(currentLine),
      delta,
      direction: toward ? 'toward' : 'away',
      grade: c.grade,
    });
  }
  return out;
}

const isTopGrade = (g) => g === 'A+' || g === 'A';

/**
 * Build ticker events from a snapshot: a SCAN summary, GRADE events for the top
 * grades, and MOVE events for significant deltas. Newest-relevant first.
 */
function generateTickerEvents(sport, grades, deltas, ts) {
  const events = [];
  events.push({
    tag: 'SCAN', color: 'var(--g-a)', ts, sport, // sport tag → dedupe one SCAN per sport
    text: `${sport.toUpperCase()} slate scanned · ${grades.length} props graded`,
  });
  for (const g of grades.filter((x) => isTopGrade(x.grade)).slice(0, 6)) {
    const arch = g.archetype ? `${g.archetype} ` : '';
    events.push({
      tag: g.grade, color: g.grade === 'A+' ? 'var(--g-ap)' : 'var(--g-a)', ts,
      text: `${arch}${lastName(g.player || g.player_name)} ${g.stat_type || g.stat} ${sideChar(g.direction)}${g.line} graded ${g.grade}`,
    });
  }
  for (const d of deltas.filter((x) => Math.abs(x.delta) >= DELTA_MOVE).slice(0, 6)) {
    const arrow = d.delta > 0 ? '▲' : '▼';
    const s = d.side === 'U' ? 'u' : 'o';
    events.push({
      tag: 'MOVE', color: 'var(--amber)', ts,
      text: `${lastName(d.player)} ${s}${d.gradedLine} → ${s}${d.currentLine} ${arrow}${d.delta > 0 ? '+' : ''}${d.delta}`,
    });
  }
  return events;
}

// Session 47 — a SCAN event's sport, from the event field or its text prefix
// (defends ticker items written before the `sport` field existed).
function scanSportOf(e) {
  if (e.tag !== 'SCAN') return null;
  if (e.sport) return String(e.sport).toLowerCase();
  const m = String(e.text || '').match(/^([a-z]+)\s+slate scanned/i);
  return m ? m[1].toLowerCase() : null;
}

async function pushTickerItems(events, deps) {
  if (!events || events.length === 0) return;
  const existing = await deps.cacheGet('ticker:items');
  const arr = Array.isArray(existing) ? existing : [];
  // Keep only the LATEST SCAN per sport: drop existing SCAN events for any sport
  // that has a fresh SCAN in this batch. MOVE/GRADE events are time-specific and
  // preserved.
  const freshScanSports = new Set(events.map(scanSportOf).filter(Boolean));
  const pruned = arr.filter((e) => {
    const sp = scanSportOf(e);
    return !(sp && freshScanSports.has(sp));
  });
  const merged = [...events, ...pruned].slice(0, TICKER_CAP);
  await deps.cacheSet('ticker:items', merged, TICKER_TTL);
}

// Session 60 (night2/B) — accumulate slate players' game logs into the
// roster blob the streaks/hot-list engines read. Merge by nameKey (newer
// entry wins), cap the blob, 72h TTL (a player off the slate for 3 days
// ages out — honest churn, not a leak).
const ROSTERLOGS_TTL = 72 * 3600;
const ROSTERLOGS_CAP = 300;

async function mergeRosterLogs(sport, entries, deps) {
  if (!entries || entries.length === 0) return;
  try {
    const key = `rosterlogs:${sport}`;
    const existing = await deps.cacheGet(key);
    const byKey = new Map();
    for (const e of Array.isArray(existing) ? existing : []) byKey.set(nameKey(e.name), e);
    for (const e of entries) byKey.set(nameKey(e.name), e); // fresh resolve wins
    const merged = [...byKey.values()].slice(-ROSTERLOGS_CAP);
    await deps.cacheSet(key, merged, ROSTERLOGS_TTL);
  } catch (e) {
    console.warn(`[snapshot] rosterlogs merge failed for ${sport}:`, e.message);
  }
}

const ACTIVE_SPORTS = ['mlb', 'nba', 'wnba', 'soccer'];

/**
 * Run one snapshot cycle for `sport`. Returns a summary; never throws.
 * opts (all injectable): getOdds, gradeAndCacheSlate, resolveStats, classify,
 * cacheGet, cacheSet, now, nowMs.
 */
async function runSnapshot(sport, opts = {}) {
  const sp = String(sport || '').toLowerCase();
  const deps = {
    getOdds: opts.getOdds || require('./oddsService').getOdds,
    gradeAndCacheSlate: opts.gradeAndCacheSlate || require('./gradeSlateService').gradeAndCacheSlate,
    resolveStats: opts.resolveStats || require('./playerIntelService').resolvePlayerStats,
    classify: opts.classify || require('./archetypeService').classify,
    cacheGet: opts.cacheGet || require('../utils/redis').cacheGet,
    cacheSet: opts.cacheSet || require('../utils/redis').cacheSet,
    now: opts.now || (() => new Date().toISOString()),
    nowMs: opts.nowMs || (() => Date.now()),
    // Session 56 — ops alerting (ntfy) + retry-once on a hard odds failure.
    notify: opts.notify || require('../utils/opsNotify').notify,
    sleep: opts.sleep || ((ms) => new Promise((r) => setTimeout(r, ms))),
    retryDelayMs: opts.retryDelayMs != null ? opts.retryDelayMs : 60_000,
    // Session 58 — Phase 1 truth infrastructure. ledger no-ops without
    // SUPABASE env, so tests / local dev never touch a database.
    ledger: opts.ledger || require('./ledgerService'),
  };
  const start = deps.nowMs();
  const ts = deps.now();

  // Session 56 — retry ONCE on a hard failure (thrown error / null response = a
  // transient PropLine/network blip). A successful-but-empty slate is NOT a
  // failure (off-hours), so it is not retried — that would waste quota + latency.
  let odds;
  try {
    odds = await deps.getOdds(sp);
    if (odds == null) throw new Error('null odds response');
  } catch (e) {
    try {
      await deps.sleep(deps.retryDelayMs);
      odds = await deps.getOdds(sp);
      if (odds == null) throw new Error('null odds response (retry)');
    } catch (e2) {
      await deps.notify(`❌ ${sp.toUpperCase()} snapshot FAILED: ${e2.message}`, { title: 'VYNDR pipeline', priority: 'high', tags: ['x'] });
      return { sport: sp, status: 'error', reason: e2.message, gradeCount: 0 };
    }
  }
  const props = (odds && Array.isArray(odds.props)) ? odds.props : [];
  if (props.length === 0) {
    await deps.notify(`⚠️ ${sp.toUpperCase()} snapshot: 0 props (odds unavailable)`, { title: 'VYNDR pipeline', priority: 'low', tags: ['warning'] });
    return { sport: sp, status: 'skipped', reason: 'no props', gradeCount: 0 };
  }

  // Grade the slate via the existing service; capture the envelope instead of
  // letting it write (we re-write an ENRICHED version below).
  let envelope = null;
  await deps.gradeAndCacheSlate(sp, props, {
    source: (odds && odds.provider) || 'odds-api',
    now: deps.now,
    cacheSet: async (_k, v) => { envelope = v; },
  });
  const rawGraded = (envelope && Array.isArray(envelope.grades)) ? envelope.grades : [];
  if (rawGraded.length === 0) return { sport: sp, status: 'skipped', reason: 'no grades', gradeCount: 0 };

  // Session 48 — normalize player display names + dedupe variant grades at the
  // SOURCE so every consumer (GameCard, Explore, leaders, profile) gets clean,
  // merged names. PropLine sends "Matt"/"Matthew", "A.J."/"AJ", "(STL)" tags as
  // separate players; collapse to ONE grade per normalized player + stat (keep
  // the highest-confidence; rawGraded is already confidence-desc).
  // Session 54 — also keep the RICHEST display per player (prefer the accented
  // variant: "José" over "Jose", then the longer string) so the prop rows match
  // the accented pitcher line. The GRADE picked is still the highest-confidence.
  const hasAccent = (s) => [...String(s)].some((c) => c.charCodeAt(0) > 127);
  const richerDisplay = (a, b) => {
    if (!b) return a;
    if (hasAccent(a) !== hasAccent(b)) return hasAccent(a) ? a : b;
    return a.length >= b.length ? a : b;
  };
  const dedup = new Map();
  const bestDisplay = new Map();
  for (const g of rawGraded) {
    const disp = normalizeName(g.player || g.player_name).display || g.player || g.player_name || '';
    const pk = nameKey(disp);
    bestDisplay.set(pk, richerDisplay(disp, bestDisplay.get(pk)));
    const k = `${pk}|${String(g.stat_type || g.stat || '').toLowerCase()}`;
    const cur = { ...g, player: disp, player_name: disp };
    const prev = dedup.get(k);
    if (!prev || (Number(g.confidence) || 0) > (Number(prev.confidence) || 0)) dedup.set(k, cur);
  }
  const graded = [...dedup.values()].map((g) => {
    const disp = bestDisplay.get(nameKey(g.player)) || g.player;
    return { ...g, player: disp, player_name: disp };
  });

  // Archetype per unique player (pure math once we have stats). Best-effort —
  // a missing stat line → no badge (not a fallback archetype).
  const oddsByKey = indexOdds(props);
  const players = [...new Set(graded.map((g) => g.player || g.player_name).filter(Boolean))];
  const archByPlayer = {};
  // Session 59 — capture the player's REAL team from the same stats resolve
  // (statsapi/ESPN), never guessed. Feeds the ledger team/opponent columns
  // and the slate join guard (a prop only attaches to its own game).
  const teamByPlayer = {};
  // Wave 1 (trust bug) — the prop's game participants become the resolve's
  // teamHint: it disambiguates namesake collisions (two "James Wood") and, when
  // the resolved player's real team isn't in the prop's game, the resolver drops
  // the team rather than tag a foreign one (the streaks/rosterlogs JOIN
  // INVARIANT, mirroring the S59 slate guard). Keyed by the normalized name.
  const teamHintByPlayer = {};
  for (const p of props || []) {
    const k = norm(p.player);
    if (!k || teamHintByPlayer[k]) continue;
    const hint = [p.home_team, p.away_team].filter(Boolean);
    if (hint.length) teamHintByPlayer[k] = hint;
  }
  // Session 60 (night2/B) — THE STREAKS PRODUCER. The aggregator (streaks +
  // hot lists) starved because its data producers were all external and
  // unarmed (tank01-prefetch via n8n, the offline Python grading flow).
  // The stats resolve above already fetched each slate player's game log —
  // accumulate it into the `rosterlogs:{sport}` blob rosterLogs.loadRosterLogs
  // reads FIRST. Zero extra API calls; the pipeline now feeds its own
  // free layer.
  const logEntries = [];
  await mapLimit(players, STATS_CONCURRENCY, async (player) => {
    try {
      const teamHint = teamHintByPlayer[norm(player)] || null;
      const stats = await deps.resolveStats(player, sp, teamHint ? { teamHint } : {});
      if (stats && stats.found) {
        const c = deps.classify(sp, stats.classifierInput || {});
        archByPlayer[player] = c.primary ? c.primary.name : null;
        if (stats.team) teamByPlayer[player] = stats.team;
        if (Array.isArray(stats.rawLog) && stats.rawLog.length > 0) {
          logEntries.push({
            name: normalizeName(player).display || player,
            playerId: stats.playerId ?? null,
            team: stats.team || null,
            group: stats.group || null,
            seasonRaw: stats.seasonRaw || null,
            games: stats.rawLog,
          });
        }
      }
    } catch { /* graceful — no badge */ }
  });
  await mergeRosterLogs(sp, logEntries, deps);

  const enriched = graded.map((g) => ({
    ...g,
    gradedAt: gradedAtFor(g, oddsByKey, ts),
    archetype: archByPlayer[g.player || g.player_name] || null,
    team: teamByPlayer[g.player || g.player_name] || g.team || null,
  }));

  // Line deltas vs the previous snapshot's locked lines.
  const prev = await deps.cacheGet(`snapshot:${sp}:latest`);
  const deltas = computeLineDeltas(enriched, prev && prev.grades);

  // Lock: previous = old latest, latest = new, grades = enriched.
  if (prev) await deps.cacheSet(`snapshot:${sp}:previous`, prev, SNAP_TTL);
  const snapshot = { sport: sp, updated_at: ts, grades: enriched, deltas, gradeCount: enriched.length };
  await deps.cacheSet(`snapshot:${sp}:latest`, snapshot, SNAP_TTL);
  // Session 59 — grades:{sport} must outlive the gap between cron runs (up to
  // 5h) or team rosters / Explore / leaders go dark mid-day. SNAP_TTL (6h),
  // NOT the legacy 2h gradeSlateService TTL — that gap was why /team showed
  // "No active props" for players who were on the slate (audit 2.4).
  await deps.cacheSet(`grades:${sp}`, { grades: enriched, updated_at: ts, source: (odds && odds.provider) || 'odds-api' }, SNAP_TTL);

  // Ticker exhaust.
  const events = generateTickerEvents(sp, enriched, deltas, ts);
  await pushTickerItems(events, deps);

  // Session 58 — Phase 1 truth infrastructure. (a) Upsert the public model
  // record (user_id NULL, idempotent — re-runs never duplicate or overwrite
  // the original lock). (b) Overwrite today's closing_line/odds from the
  // CURRENT feed — the last capture before game start IS the closing line.
  // Best-effort: the ledger must never break the snapshot.
  let ledgerWritten = 0;
  try {
    const rec = await deps.ledger.recordPipelineGrades(sp, enriched, props, { now: deps.now });
    ledgerWritten = rec.written || 0;
    await deps.ledger.captureClosing(sp, props);
  } catch (e) {
    console.warn(`[snapshot] ledger write failed for ${sp}:`, e.message);
  }

  // Session 56 — success alert, enriched with the rolling accuracy (if settled).
  let accPart = '';
  try {
    const acc = await deps.cacheGet(`accuracy:${sp}`);
    const pct = acc && acc.overall && acc.overall.pct;
    if (pct != null) accPart = `, ${pct}% accuracy (30d)`;
  } catch { /* accuracy is best-effort in the alert */ }
  await deps.notify(
    `✅ ${sp.toUpperCase()} snapshot: ${enriched.length} props graded, ${deltas.length} deltas${accPart}`,
    { title: 'VYNDR pipeline', tags: ['white_check_mark'] },
  );

  return {
    sport: sp,
    status: 'ok',
    gradeCount: enriched.length,
    ledgerWritten,
    topGrades: enriched.filter((g) => isTopGrade(g.grade)).slice(0, 5).map((g) => ({
      player: g.player || g.player_name, stat: g.stat_type || g.stat, grade: g.grade, archetype: g.archetype,
    })),
    deltas: deltas.length,
    duration: deps.nowMs() - start,
  };
}

/** Run snapshots for every active sport sequentially (cron entrypoint). */
async function runAllSnapshots(opts = {}) {
  const results = [];
  for (const sp of ACTIVE_SPORTS) {
    results.push(await runSnapshot(sp, opts));
  }
  return results;
}

module.exports = {
  runSnapshot,
  runAllSnapshots,
  computeLineDeltas,
  generateTickerEvents,
  pushTickerItems,
  ACTIVE_SPORTS,
  __internals: { propKey, gradedAtFor, indexOdds, lastName, isTopGrade, DELTA_NOISE, DELTA_MOVE, TICKER_CAP },
};