import { DatabaseSync } from 'node:sqlite'; import { copyFileSync, mkdirSync, readdirSync, rmSync } from 'node:fs'; import path from 'node:path'; // Persistence via Node's built-in SQLite (no native deps). One file on a Docker // volume holds the response cache, source health, fixture↔provider id mapping, // per-fixture enrichment, historical player goals, and ingest logs. const DATA_DIR = path.resolve(process.env.DATA_DIR ?? path.join(process.cwd(), 'data')); const SCHEMA = ` CREATE TABLE IF NOT EXISTS kv_cache ( key TEXT PRIMARY KEY, value TEXT NOT NULL, fetched_at INTEGER NOT NULL, ttl INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS source_health ( source TEXT PRIMARY KEY, last_ok INTEGER, last_err INTEGER, consec_fail INTEGER NOT NULL DEFAULT 0, open_until INTEGER NOT NULL DEFAULT 0, note TEXT ); CREATE TABLE IF NOT EXISTS source_map ( fixture_num INTEGER NOT NULL, source TEXT NOT NULL, event_id TEXT NOT NULL, PRIMARY KEY (fixture_num, source) ); CREATE TABLE IF NOT EXISTS match_ext ( fixture_num INTEGER PRIMARY KEY, json TEXT NOT NULL, updated_at INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS injuries ( team TEXT NOT NULL, player TEXT NOT NULL, status TEXT, reason TEXT, return_date TEXT, updated_at INTEGER NOT NULL, PRIMARY KEY (team, player) ); CREATE TABLE IF NOT EXISTS squad ( team TEXT NOT NULL, player TEXT NOT NULL, position TEXT, number INTEGER, market_value INTEGER, age INTEGER, updated_at INTEGER NOT NULL, PRIMARY KEY (team, player) ); CREATE TABLE IF NOT EXISTS goalscorers ( date TEXT NOT NULL, home TEXT NOT NULL, away TEXT NOT NULL, team TEXT NOT NULL, scorer TEXT, minute INTEGER, own_goal INTEGER NOT NULL DEFAULT 0, penalty INTEGER NOT NULL DEFAULT 0 ); CREATE INDEX IF NOT EXISTS idx_goalscorers_scorer ON goalscorers(scorer); CREATE INDEX IF NOT EXISTS idx_goalscorers_team ON goalscorers(team); CREATE TABLE IF NOT EXISTS odds_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, fixture_num INTEGER NOT NULL, provider TEXT NOT NULL, captured_at INTEGER NOT NULL, home_ml REAL, draw_ml REAL, away_ml REAL, over_under REAL, spread REAL ); CREATE INDEX IF NOT EXISTS idx_odds_fixture ON odds_history(fixture_num, captured_at); CREATE TABLE IF NOT EXISTS prediction_snapshots ( fixture_num INTEGER PRIMARY KEY, taken_at INTEGER NOT NULL, late INTEGER NOT NULL DEFAULT 0, model_json TEXT NOT NULL, market_json TEXT ); CREATE TABLE IF NOT EXISTS push_subs ( endpoint TEXT PRIMARY KEY, json TEXT NOT NULL, team TEXT, created_at INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS inplay_history ( fixture_num INTEGER NOT NULL, minute INTEGER NOT NULL, home_score INTEGER NOT NULL, away_score INTEGER NOT NULL, p_home REAL NOT NULL, p_draw REAL NOT NULL, p_away REAL NOT NULL, at INTEGER NOT NULL, PRIMARY KEY (fixture_num, minute, home_score, away_score) ); CREATE TABLE IF NOT EXISTS ingest_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, source TEXT NOT NULL, job TEXT NOT NULL, ok INTEGER NOT NULL, ms INTEGER, note TEXT, at INTEGER NOT NULL ); CREATE INDEX IF NOT EXISTS idx_ingest_log_at ON ingest_log(at); `; let _db: DatabaseSync | null = null; export function db(): DatabaseSync { if (_db) return _db; mkdirSync(DATA_DIR, { recursive: true }); const file = path.join(DATA_DIR, 'cup26.db'); const d = new DatabaseSync(file); d.exec('PRAGMA journal_mode = WAL;'); d.exec('PRAGMA busy_timeout = 4000;'); d.exec(SCHEMA); _db = d; console.log(`[db] sqlite ready → ${file}`); return d; } // ---- response cache ---- export function cacheGet(key: string): { value: string; fresh: boolean } | null { const row = db().prepare('SELECT value, fetched_at, ttl FROM kv_cache WHERE key = ?').get(key) as | { value: string; fetched_at: number; ttl: number } | undefined; if (!row) return null; return { value: row.value, fresh: Date.now() - row.fetched_at < row.ttl }; } export function cacheSet(key: string, value: string, ttl: number): void { db() .prepare('INSERT OR REPLACE INTO kv_cache (key, value, fetched_at, ttl) VALUES (?, ?, ?, ?)') .run(key, value, Date.now(), ttl); } // ---- Web Push subscriptions (goal/kickoff notifications, opt-in per team) ---- export interface PushSub { endpoint: string; json: string; team: string | null; } export function addPushSub(endpoint: string, json: string, team: string | null): void { db() .prepare('INSERT OR REPLACE INTO push_subs (endpoint, json, team, created_at) VALUES (?, ?, ?, ?)') .run(endpoint, json, team, Date.now()); } export function removePushSub(endpoint: string): void { db().prepare('DELETE FROM push_subs WHERE endpoint = ?').run(endpoint); } export function pushSubsForTeams(teams: string[]): PushSub[] { if (teams.length === 0) return []; const marks = teams.map(() => '?').join(','); return db() .prepare(`SELECT endpoint, json, team FROM push_subs WHERE team IN (${marks})`) .all(...teams) as unknown as PushSub[]; } // ---- in-play win-probability history (the momentum curve on match pages) ---- export interface InplayPoint { minute: number; home_score: number; away_score: number; p_home: number; p_draw: number; p_away: number; } /** Insert-if-new keyed on (minute, score) — one point per game state. */ export function recordInplay(num: number, p: InplayPoint): void { db() .prepare(`INSERT OR IGNORE INTO inplay_history (fixture_num, minute, home_score, away_score, p_home, p_draw, p_away, at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`) .run(num, p.minute, p.home_score, p.away_score, p.p_home, p.p_draw, p.p_away, Date.now()); } export function inplayHistory(num: number): InplayPoint[] { return db() .prepare(`SELECT minute, home_score, away_score, p_home, p_draw, p_away FROM inplay_history WHERE fixture_num = ? ORDER BY minute, home_score + away_score`) .all(num) as unknown as InplayPoint[]; } /** Nightly consistent snapshot: VACUUM INTO a LOCAL temp file (SQLite needs * real locking, which the CIFS-mounted archive cannot offer), then a plain * sequential copy into the archive dir. Keeps the newest `keep` dated copies. */ export function backupDb(keep = 14): string | null { const dir = process.env.ARCHIVE_DIR ?? path.join(DATA_DIR, 'archive'); const tmp = path.join(DATA_DIR, 'backup-tmp.db'); try { mkdirSync(dir, { recursive: true }); rmSync(tmp, { force: true }); // VACUUM INTO refuses to overwrite db().exec(`VACUUM INTO '${tmp.replaceAll("'", "''")}'`); const stamp = new Date().toISOString().slice(0, 10); const dest = path.join(dir, `cup26-${stamp}.db`); copyFileSync(tmp, dest); rmSync(tmp, { force: true }); rmSync(`${dest}-journal`, { force: true }); // leftover from any failed run const files = readdirSync(dir).filter((f) => /^cup26-\d{4}-\d{2}-\d{2}\.db$/.test(f)).sort(); for (const f of files.slice(0, Math.max(0, files.length - keep))) { rmSync(path.join(dir, f), { force: true }); } console.log(`[db] snapshot → ${dest}`); return dest; } catch (e) { rmSync(tmp, { force: true }); console.error('[db] backup failed:', e instanceof Error ? e.message : e); return null; } } /** Bound DB growth: drop cache entries stale for >7 days (the breaker only ever * serves recent staleness) and ingest-log rows older than 14 days. Returns the * number of rows removed. Runs daily from the scheduler. */ export function prune(now = Date.now()): number { const cache = db() .prepare('DELETE FROM kv_cache WHERE ? - fetched_at > ttl + 7 * 86400000') .run(now); const log = db() .prepare('DELETE FROM ingest_log WHERE at < ?') .run(now - 14 * 86400000); return Number(cache.changes) + Number(log.changes); } // ---- source health / circuit breaker ---- export interface SourceHealthRow { source: string; last_ok: number | null; last_err: number | null; consec_fail: number; open_until: number; note: string | null; } export function healthGet(source: string): SourceHealthRow | null { return (db().prepare('SELECT * FROM source_health WHERE source = ?').get(source) as SourceHealthRow | undefined) ?? null; } export function healthAll(): SourceHealthRow[] { return db().prepare('SELECT * FROM source_health ORDER BY source').all() as unknown as SourceHealthRow[]; } export function healthRecord(source: string, ok: boolean, note?: string, openMs?: number): void { const now = Date.now(); const cur = healthGet(source); const consec = ok ? 0 : (cur?.consec_fail ?? 0) + 1; const openUntil = ok ? 0 : openMs ? now + openMs : (cur?.open_until ?? 0); db() .prepare( `INSERT INTO source_health (source, last_ok, last_err, consec_fail, open_until, note) VALUES (@source, @last_ok, @last_err, @consec, @open_until, @note) ON CONFLICT(source) DO UPDATE SET last_ok=COALESCE(@last_ok, source_health.last_ok), last_err=COALESCE(@last_err, source_health.last_err), consec_fail=@consec, open_until=@open_until, note=@note`, ) .run({ source, last_ok: ok ? now : null, last_err: ok ? null : now, consec: consec, open_until: openUntil, note: note ?? null, }); } /** Row counts for the /data page — everything the live layer has collected. */ export function dbCounts(): Record { const c = (sql: string) => (db().prepare(sql).get() as { n: number }).n; return { oddsLines: c('SELECT COUNT(*) n FROM odds_history'), oddsFixtures: c('SELECT COUNT(DISTINCT fixture_num) n FROM odds_history'), frozenForecasts: c('SELECT COUNT(*) n FROM prediction_snapshots'), enrichedMatches: c('SELECT COUNT(*) n FROM match_ext'), mappedFixtures: c('SELECT COUNT(*) n FROM source_map'), cachedResponses: c('SELECT COUNT(*) n FROM kv_cache'), ingestEvents: c('SELECT COUNT(*) n FROM ingest_log'), }; } export function logIngest(source: string, job: string, ok: boolean, ms: number, note?: string): void { db().prepare('INSERT INTO ingest_log (source, job, ok, ms, note, at) VALUES (?, ?, ?, ?, ?, ?)') .run(source, job, ok ? 1 : 0, Math.round(ms), note ?? null, Date.now()); } // ---- fixture ↔ provider event-id mapping ---- export function setSourceMap(fixtureNum: number, source: string, eventId: string): void { db().prepare('INSERT OR REPLACE INTO source_map (fixture_num, source, event_id) VALUES (?, ?, ?)') .run(fixtureNum, source, eventId); } export function getSourceMap(fixtureNum: number, source: string): string | null { const row = db().prepare('SELECT event_id FROM source_map WHERE fixture_num = ? AND source = ?').get(fixtureNum, source) as | { event_id: string } | undefined; return row?.event_id ?? null; } // ---- bookmaker odds history (benchmark only, never a model input) ---- export interface OddsRow { fixture_num: number; provider: string; captured_at: number; home_ml: number | null; draw_ml: number | null; away_ml: number | null; over_under: number | null; spread: number | null; } /** Insert a snapshot only if the lines moved since the last capture. */ export function recordOdds(o: Omit): boolean { const last = db() .prepare('SELECT home_ml, draw_ml, away_ml, over_under, spread FROM odds_history WHERE fixture_num = ? AND provider = ? ORDER BY captured_at DESC LIMIT 1') .get(o.fixture_num, o.provider) as Omit | undefined; if ( last && last.home_ml === o.home_ml && last.draw_ml === o.draw_ml && last.away_ml === o.away_ml && last.over_under === o.over_under && last.spread === o.spread ) return false; db() .prepare('INSERT INTO odds_history (fixture_num, provider, captured_at, home_ml, draw_ml, away_ml, over_under, spread) VALUES (?, ?, ?, ?, ?, ?, ?, ?)') .run(o.fixture_num, o.provider, Date.now(), o.home_ml, o.draw_ml, o.away_ml, o.over_under, o.spread); return true; } export function oddsFor(fixtureNum: number): OddsRow[] { return db() .prepare('SELECT fixture_num, provider, captured_at, home_ml, draw_ml, away_ml, over_under, spread FROM odds_history WHERE fixture_num = ? ORDER BY captured_at') .all(fixtureNum) as unknown as OddsRow[]; } export function latestOddsAll(): OddsRow[] { return db() .prepare(`SELECT o.* FROM odds_history o JOIN (SELECT fixture_num, provider, MAX(captured_at) m FROM odds_history GROUP BY fixture_num, provider) t ON o.fixture_num = t.fixture_num AND o.provider = t.provider AND o.captured_at = t.m ORDER BY o.fixture_num`) .all() as unknown as OddsRow[]; } // ---- frozen pre-kickoff forecasts (Model-vs-Market scoreboard) ---- export interface SnapshotRow { fixture_num: number; taken_at: number; late: number; model_json: string; market_json: string | null; } export function hasSnapshot(fixtureNum: number): boolean { return !!db().prepare('SELECT 1 FROM prediction_snapshots WHERE fixture_num = ?').get(fixtureNum); } export function saveSnapshot(fixtureNum: number, late: boolean, model: unknown, market: unknown | null): void { db() .prepare('INSERT OR IGNORE INTO prediction_snapshots (fixture_num, taken_at, late, model_json, market_json) VALUES (?, ?, ?, ?, ?)') .run(fixtureNum, Date.now(), late ? 1 : 0, JSON.stringify(model), market ? JSON.stringify(market) : null); } export function allSnapshots(): SnapshotRow[] { return db() .prepare('SELECT fixture_num, taken_at, late, model_json, market_json FROM prediction_snapshots ORDER BY fixture_num') .all() as unknown as SnapshotRow[]; } // ---- per-fixture enrichment blob ---- export function setMatchExt(fixtureNum: number, json: unknown): void { db().prepare('INSERT OR REPLACE INTO match_ext (fixture_num, json, updated_at) VALUES (?, ?, ?)') .run(fixtureNum, JSON.stringify(json), Date.now()); } export function getMatchExt(fixtureNum: number): { data: T; updatedAt: number } | null { const row = db().prepare('SELECT json, updated_at FROM match_ext WHERE fixture_num = ?').get(fixtureNum) as | { json: string; updated_at: number } | undefined; if (!row) return null; return { data: JSON.parse(row.json) as T, updatedAt: row.updated_at }; }