spb/market-atlas

1import { spawn } from "node:child_process";2import { createWriteStream, mkdirSync } from "node:fs";3import { stat } from "node:fs/promises";4import { join } from "node:path";5import { createGzip } from "node:zlib";6import { pipeline as streamPipeline } from "node:stream/promises";7import { Readable } from "node:stream";8import { config } from "../config.js";9import { pool } from "../db/pool.js";10import { logger } from "../logger.js";11import { rollupBars } from "./bars.js";12import { bus } from "./bus.js";13import { calendar } from "./calendar.js";14import { eventEngine } from "./events.js";15import { health } from "./health.js";16import { instruments } from "./instruments.js";17import { pipeline } from "./pipeline.js";18import { quoteStore } from "./quotes.js";19import { telemetry } from "./telemetry.js";20import { connectorManager } from "./connector-manager.js";21import { lineage } from "./lineage.js";2223/** Periodic maintenance: health snapshots, market open/close events, rollups, retention/archive, backups. */24export class Scheduler {25  private timers: NodeJS.Timeout[] = [];26  private lastState = new Map<string, string>();27  private lastDaily: string | null = null;2829  start(): void {30    this.every(60_000, () => health.tick((id) => staleAfter(id)));31    this.every(5_000, () => this.marketStates());32    this.every(10_000, () => pipeline.sweep());33    this.every(5 * 60_000, () => rollupBars(3));34    this.every(5 * 60_000, () => lineage.recompute());35    this.every(60_000, () => this.gauges());36    this.every(15 * 60_000, () => this.daily());37    this.marketStates();38    this.gauges();39  }4041  stop(): void {42    for (const t of this.timers) clearInterval(t);43    this.timers = [];44  }4546  private every(ms: number, fn: () => unknown) {47    const t = setInterval(() => {48      Promise.resolve()49        .then(fn)50        .catch((err) => logger.error({ err }, "scheduled job failed"));51    }, ms);52    this.timers.push(t);53  }5455  private gauges() {56    telemetry.gauge("quotes_cached", quoteStore.size());57    telemetry.gauge("instruments_total", instruments.count());58    telemetry.gauge("connectors_healthy", connectorManager.list().filter((r) => health.state(r.def.metadata.id) === "HEALTHY").length);59    telemetry.gauge("connectors_total", connectorManager.list().length);60  }6162  /** Emit MARKET_OPEN / MARKET_CLOSE events and reset session stats at session start. */63  private marketStates() {64    const now = Date.now();65    for (const ex of calendar.list()) {66      if (ex.sessions.continuous) continue;67      const st = calendar.status(ex.id, now);68      const prev = this.lastState.get(ex.id);69      this.lastState.set(ex.id, st.state);70      if (prev === undefined || prev === st.state) continue;71      bus.publish("market.state", { exchangeId: ex.id, state: st.state, at: now });72      if (st.state === "OPEN") {73        eventEngine.system("MARKET_OPEN", `${ex.id}:${new Date(now).toISOString().slice(0, 10)}`, now, "NOTICE", `${ex.name} market open`, { exchangeId: ex.id, localTime: st.localTime }, ["market-atlas-calendar"]);74        for (const inst of instruments.all()) {75          if (inst.exchangeId === ex.id) {76            pipeline.consensus.resetSession(inst.id, now);77            eventEngine.resetSession(inst.id);78          }79        }80      } else if (st.state === "CLOSED" && (prev === "OPEN" || prev === "POST")) {81        eventEngine.system("MARKET_CLOSE", `${ex.id}:${new Date(now).toISOString().slice(0, 10)}`, now, "NOTICE", `${ex.name} market close`, { exchangeId: ex.id, localTime: st.localTime }, ["market-atlas-calendar"]);82      }83    }84  }8586  /** Once per day (UTC 04:40-ish): archive old observation partitions, backup metadata. */87  private async daily() {88    const now = new Date();89    const key = now.toISOString().slice(0, 10);90    if (this.lastDaily === key) return;91    if (now.getUTCHours() < 8) return; // ~04:00 America/Toronto92    this.lastDaily = key;93    await archiveOldPartitions().catch((err) => logger.error({ err }, "archive failed"));94    await backupMetadata().catch((err) => logger.error({ err }, "backup failed"));95    await pool.query("delete from connector_health where ts < now() - interval '90 days'");96  }97}9899function staleAfter(connectorId: string): number {100  const rt = connectorManager.get(connectorId);101  const m = rt?.def.metadata;102  if (!m) return 10 * 60_000;103  if (m.sourceType === "WEBSOCKET") return 5 * 60_000;104  const s = rt?.def.schedule;105  if (s && "intervalMs" in s) return s.intervalMs * 3 + 60_000;106  if (s) return s.weekendMs * 2 + 60_000;107  return 24 * 3_600_000;108}109110/** Export observation partitions older than the retention window to gzip NDJSON, then drop them. */111export async function archiveOldPartitions(): Promise<string[]> {112  const cutoff = new Date(Date.now() - config.observationRetentionDays * 86_400_000).toISOString().slice(0, 10).replace(/-/g, "");113  const parts = await pool.query<{ relname: string }>(114    `select c.relname from pg_inherits i join pg_class c on c.oid = i.inhrelid join pg_class p on p.oid = i.inhparent where p.relname = 'observations' order by 1`,115  );116  const archived: string[] = [];117  for (const { relname } of parts.rows) {118    const day = relname.replace("observations_", "");119    if (day >= cutoff) continue;120    const dir = join(config.dataDir, "archive", day.slice(0, 4));121    mkdirSync(dir, { recursive: true });122    const path = join(dir, `${relname}.ndjson.gz`);123    const rows = await pool.query(`select * from ${relname}`);124    await streamPipeline(Readable.from(rows.rows.map((r) => JSON.stringify(r) + "\n")), createGzip({ level: 9 }), createWriteStream(path));125    const { size } = await stat(path);126    await pool.query(`insert into observation_archives (partition_name, day, rows_archived, path, bytes) values ($1,$2,$3,$4,$5) on conflict (partition_name) do update set rows_archived = excluded.rows_archived, bytes = excluded.bytes, archived_at = now()`, [127      relname,128      `${day.slice(0, 4)}-${day.slice(4, 6)}-${day.slice(6, 8)}`,129      rows.rowCount ?? 0,130      path,131      size,132    ]);133    await pool.query(`drop table ${relname}`);134    archived.push(relname);135    logger.info({ partition: relname, rows: rows.rowCount, bytes: size }, "partition archived");136  }137  return archived;138}139140/** pg_dump of metadata tables (schema + data, excluding observations/bars) into data/backups. */141export function backupMetadata(): Promise<string> {142  return new Promise((resolve, reject) => {143    const dir = join(config.dataDir, "backups");144    mkdirSync(dir, { recursive: true });145    const file = join(dir, `market-atlas-meta-${new Date().toISOString().slice(0, 10)}.sql.gz`);146    const pgDump = process.env.PG_DUMP ?? "pg_dump";147    const child = spawn(pgDump, ["--no-owner", "--exclude-table-data=observations*", "--exclude-table-data=bars", "--exclude-table-data=connector_health", config.databaseUrl], { stdio: ["ignore", "pipe", "pipe"] });148    const out = createWriteStream(file);149    child.stdout.pipe(createGzip()).pipe(out);150    let stderr = "";151    child.stderr.on("data", (d) => (stderr += d));152    child.on("error", reject);153    child.on("close", (code) => (code === 0 ? resolve(file) : reject(new Error(`pg_dump exited ${code}: ${stderr.slice(0, 300)}`))));154  });155}156157export const scheduler = new Scheduler();158