vyndr/src/services/statcastAggregateService.js

'use strict';

/**
 * STATCAST AGGREGATE SERVICE (Layer 1) — derivation + join + store.
 *
 * The AGGREGATE DERIVATION and IDENTITY BRIDGE boxes of the mechanism-data
 * pattern (docs/MECHANISM-DATA.md). Pure where it can be, injectable
 * everywhere, so tests never touch the network or the database.
 *
 * BACKFILL and REFRESH are the SAME operation: a full re-pull of the ~1,300-row
 * aggregate set, upserted on (sport, season, source_id). That makes the nightly
 * job idempotent and self-healing — a missed night self-corrects on the next
 * run, with no incremental "who played today" bookkeeping to drift out of sync.
 * At this volume the simple thing is also the robust thing.
 *
 * HONESTY RULES:
 *   - A metric the feed did not carry is ABSENT (null), never 0.
 *   - A player below the minimum sample keeps his row but is marked
 *     `sufficient: false` — thin is not the same as missing, and Layer 2 must
 *     be able to tell them apart.
 *   - A player we cannot join to our own roster still stores (player_key null)
 *     and joins later. Storing him is not a claim about him.
 *   - `updated_at` is written every run. Freshness is a truth property; the
 *     staleness alarm reads it.
 */

const { nameKey } = require('../utils/playerName');
const statcast = require('./adapters/statcastAdapter');

/** Minimum samples for a mechanism claim. Below these the row is stored but
 *  flagged insufficient — the explicit gate that prevents a 12-PA sample from
 *  being classified as confidently as a 400-PA one (the FLEX-at-weight-1.0
 *  failure, structurally prevented). */
const MIN_PA = Number(process.env.STATCAST_MIN_PA) || 50;
const MIN_IP = Number(process.env.STATCAST_MIN_IP) || 10;

const num = (v) => {
  if (v == null || v === '') return null;
  const n = typeof v === 'number' ? v : Number(v);
  return Number.isFinite(n) ? n : null;
};

/**
 * buildRows(season, feeds, opts) — PURE. Merges the five feed indexes into one
 * row per player per role. No I/O.
 */
function buildRows(season, feeds, opts = {}) {
  const sport = opts.sport || 'mlb';
  const now = opts.now || new Date().toISOString();
  const rows = [];

  // ---- BATTERS -----------------------------------------------------------
  const batterIds = new Set([...feeds.batterDiscipline.keys(), ...feeds.batterBattedBall.keys()]);
  for (const id of batterIds) {
    const d = feeds.batterDiscipline.get(id);
    const b = feeds.batterBattedBall.get(id);
    const m = { ...(b ? b.metrics : {}), ...(d ? d.metrics : {}) };
    const name = (d && d.name) || (b && b.name) || null;
    const pa = num(m.pa);
    rows.push({
      sport,
      season,
      source_id: id,
      player_key: name ? nameKey(name) : null,
      player_name: name,
      role: 'batter',
      throws: null,
      bats: null, // roster join fills this; absent otherwise (never guessed)
      sample_pa: pa,
      sample_ip: null,
      sample_bip: num(m.bip),
      k_pct: num(m.k_pct),
      bb_pct: num(m.bb_pct),
      whiff_pct: num(m.whiff_pct),
      swing_pct: num(m.swing_pct),
      chase_pct: num(m.chase_pct),
      barrel_pct: num(m.barrel_pct) ?? num(m.barrel_pct_bb),
      hard_hit_pct: num(m.hard_hit_pct),
      avg_exit_velo: num(m.avg_exit_velo),
      max_exit_velo: num(m.max_exit_velo),
      avg_launch_angle: num(m.avg_launch_angle),
      sweet_spot_pct: num(m.sweet_spot_pct) ?? num(m.sweet_spot_pct_bb),
      ev95_pct: num(m.ev95_pct),
      arm_angle: null,
      gb_pct: null,
      fb_pct: null,
      ld_pct: null,
      pitch_mix: null,
      metrics: m,
      updated_at: now,
      _sufficient: pa != null && pa >= MIN_PA,
    });
  }

  // ---- PITCHERS ----------------------------------------------------------
  const pitcherIds = new Set([
    ...feeds.pitcherDiscipline.keys(),
    ...feeds.pitcherBattedBall.keys(),
    ...feeds.pitchMix.keys(),
  ]);
  for (const id of pitcherIds) {
    const d = feeds.pitcherDiscipline.get(id);
    const b = feeds.pitcherBattedBall.get(id);
    const mix = feeds.pitchMix.get(id);
    const m = { ...(b ? b.metrics : {}), ...(d ? d.metrics : {}) };
    const name = (d && d.name) || (mix && mix.name) || (b && b.name) || null;
    const ip = num(m.ip);
    rows.push({
      sport,
      season,
      source_id: id,
      player_key: name ? nameKey(name) : null,
      player_name: name,
      role: 'pitcher',
      // Handedness comes free on the movement feed — the ONLY feed that carries
      // it. Absent when a pitcher has thrown too few tracked pitches to appear.
      throws: (mix && mix.throws) || null,
      bats: null,
      sample_pa: null,
      sample_ip: ip,
      sample_bip: num(m.bip),
      k_pct: num(m.k_pct),
      bb_pct: num(m.bb_pct),
      whiff_pct: num(m.whiff_pct),
      swing_pct: num(m.swing_pct),
      chase_pct: num(m.chase_pct),
      barrel_pct: num(m.barrel_pct_allowed),
      hard_hit_pct: num(m.hard_hit_pct_allowed),
      avg_exit_velo: num(m.avg_exit_velo),
      max_exit_velo: num(m.max_exit_velo),
      avg_launch_angle: num(m.avg_launch_angle),
      sweet_spot_pct: num(m.sweet_spot_pct_bb),
      ev95_pct: num(m.ev95_pct),
      arm_angle: num(m.arm_angle),
      gb_pct: num(m.gb_pct),
      fb_pct: num(m.fb_pct),
      ld_pct: num(m.ld_pct),
      pitch_mix: mix && mix.pitches && mix.pitches.length ? mix.pitches : null,
      metrics: m,
      updated_at: now,
      _sufficient: ip != null && ip >= MIN_IP,
    });
  }

  return rows;
}

/** Strip the internal flag before the row reaches Postgres. */
function toDbRow(r) {
  const { _sufficient, ...rest } = r;
  return rest;
}

/**
 * refreshSeason(opts) — THE JOB. Backfill and nightly refresh are the same call.
 *
 * opts: { season, sport, supabase, fetchSeason, now, chunkSize }
 * Returns a summary — never throws on a partial failure; a broken feed degrades
 * to fewer rows, which is honest, rather than to a wrong number.
 */
async function refreshSeason(opts = {}) {
  const season = opts.season || statcast.DEFAULT_SEASON;
  const sport = opts.sport || 'mlb';
  const started = opts.now || new Date().toISOString();

  const fetchSeason = opts.fetchSeason || statcast.fetchSeason;
  const feeds = await fetchSeason(season, opts);
  const rows = buildRows(season, feeds, { sport, now: started });

  const summary = {
    ok: true,
    sport,
    season,
    started_at: started,
    feed_counts: feeds.counts || null,
    rows: rows.length,
    batters: rows.filter((r) => r.role === 'batter').length,
    pitchers: rows.filter((r) => r.role === 'pitcher').length,
    sufficient: rows.filter((r) => r._sufficient).length,
    thin: rows.filter((r) => !r._sufficient).length,
    joined: rows.filter((r) => r.player_key).length,
    unjoined: rows.filter((r) => !r.player_key).length,
    with_handedness: rows.filter((r) => r.throws).length,
    written: 0,
  };

  if (!rows.length) {
    // Every feed empty = the source failed, NOT "there is no mechanism data".
    // Refuse to write, so a bad night can never blank a good table.
    summary.ok = false;
    summary.reason = 'no rows from source — refusing to write';
    return summary;
  }

  const sb = opts.supabase || getServiceClient();
  if (!sb) {
    summary.ok = false;
    summary.reason = 'no supabase client (env not configured)';
    return summary;
  }

  // Chunked upsert — idempotent on the natural key, so re-running is a no-op
  // beyond refreshing values and `updated_at`.
  const chunk = opts.chunkSize || 500;
  for (let i = 0; i < rows.length; i += chunk) {
    const batch = rows.slice(i, i + chunk).map(toDbRow);
    const { error } = await sb
      .from('statcast_aggregates')
      .upsert(batch, { onConflict: 'sport,season,source_id' });
    if (error) {
      summary.ok = false;
      summary.reason = `upsert failed: ${error.message}`;
      return summary;
    }
    summary.written += batch.length;
  }

  return summary;
}

function getServiceClient() {
  try {
    return require('../utils/supabase').getSupabaseServiceClient();
  } catch {
    return null;
  }
}

/**
 * getFreshness(opts) — how current is the mechanism tier?
 * Reads the newest `updated_at`. Used by the staleness alarm and the status
 * probe. Returns { updated_at, age_hours, rows } or nulls — never a guess.
 */
async function getFreshness(opts = {}) {
  const sb = opts.supabase || getServiceClient();
  if (!sb) return { updated_at: null, age_hours: null, rows: null };
  const sport = opts.sport || 'mlb';
  const { data, error } = await sb
    .from('statcast_aggregates')
    .select('updated_at')
    .eq('sport', sport)
    .order('updated_at', { ascending: false })
    .limit(1);
  if (error || !data || !data.length) return { updated_at: null, age_hours: null, rows: null };
  const updated = data[0].updated_at;
  const nowMs = opts.nowMs || Date.now();
  const ageH = (nowMs - new Date(updated).getTime()) / 3_600_000;
  return { updated_at: updated, age_hours: Math.round(ageH * 10) / 10 };
}

/**
 * isStale(freshness, maxAgeHours) — PURE. The alarm predicate.
 *
 * NEVER-REFRESHED (updated_at null) is NOT stale — it is "not built yet", a
 * different condition with a different fix. Conflating them would page on a
 * fresh install and teach the operator to ignore the alarm.
 */
const MAX_AGE_HOURS = Number(process.env.STATCAST_MAX_AGE_HOURS) || 48;
function isStale(freshness, maxAgeHours = MAX_AGE_HOURS) {
  if (!freshness || freshness.updated_at == null) return false;
  return Number.isFinite(freshness.age_hours) && freshness.age_hours > maxAgeHours;
}

module.exports = {
  refreshSeason,
  buildRows,
  getFreshness,
  isStale,
  MIN_PA,
  MIN_IP,
  MAX_AGE_HOURS,
  __internals: { toDbRow, num, getServiceClient },
};