spb/earth-now Public License

1/**2 * earth-now.co3 * Author:  Simon-Pierre Boucher4 * Contact: contact@spboucher.ai5 * File:    apps/api/src/store.ts6 * Purpose: In-memory model store — boots from registry+fixtures, tracks blocked/stale metrics, emits "model" events for SSE7 */89import { EventEmitter } from "node:events";10import type { CounterModel, DisplayHints } from "@earth-now/counter";11import type { MetricConstraints, ValidationIssue } from "@earth-now/models";12import {13  type MetricEntry,14  type Registry,15  fitAllModels,16  loadFixtures,17  loadRegistry,18} from "@earth-now/registry";1920export interface BlockedMetric {21  metricId: string;22  issues: ValidationIssue[];23}2425export interface QuakesSnapshot {26  count24h: number;27  lastMajor: { mag: number; place: string; timeIso: string } | null;28  fetchedAt: string;29}3031export type ModelListener = (model: CounterModel) => void;3233/** Registry display block → runtime DisplayHints (same mapping the fitter uses). */34export function displayHintsOf(entry: MetricEntry, locale: "fr" | "en" = "en"): DisplayHints {35  const hints: DisplayHints = {36    decimals: entry.display.decimals,37    unit: entry.display.unit[locale],38  };39  if (entry.display.sigFigs !== undefined) hints.sigFigs = entry.display.sigFigs;40  if (entry.display.scale !== undefined) hints.scale = entry.display.scale;41  return hints;42}4344/** Registry constraints block → validation MetricConstraints. */45export function metricConstraints(entry: MetricEntry): MetricConstraints {46  const c: MetricConstraints = {47    kind: entry.kind === "cumulative" ? "cumulative" : "stock",48  };49  if (entry.constraints.maxAbsRatePerSec !== undefined)50    c.maxAbsRatePerSec = entry.constraints.maxAbsRatePerSec;51  if (entry.constraints.maxJumpOnRefit !== undefined)52    c.maxJumpOnRefit = entry.constraints.maxJumpOnRefit;53  return c;54}5556/** Fit/validation horizon: Jan 1 of the current UTC year → Dec 31 of the next year. */57export function fitHorizon(nowMs: number): { fromMs: number; toMs: number } {58  const year = new Date(nowMs).getUTCFullYear();59  return { fromMs: Date.UTC(year, 0, 1), toMs: Date.UTC(year + 2, 0, 1) };60}6162/**63 * Holds the currently deployable CounterModel of every metric. The API serves64 * models, never per-tick values; SSE subscribers listen to the "model" event65 * and only receive a payload when a NEW model is deployed.66 */67export class ModelStore extends EventEmitter {68  private readonly modelMap = new Map<string, CounterModel>();69  private blockedList: BlockedMetric[] = [];70  private readonly staleSinceMap = new Map<string, string>();71  private loadedRegistry: Registry | null = null;72  private quakesSnapshot: QuakesSnapshot = {73    count24h: 0,74    lastMajor: null,75    fetchedAt: new Date(0).toISOString(),76  };7778  /** Load registry + fixtures and fit every model. Blocked metrics are served as stale. */79  boot(nowMs: number = Date.now()): void {80    const registry = loadRegistry();81    const fixtures = loadFixtures();82    const { fromMs, toMs } = fitHorizon(nowMs);83    const { models, blocked } = fitAllModels(registry.metrics, fixtures, {84      validateFromMs: fromMs,85      validateToMs: toMs,86    });8788    this.loadedRegistry = registry;89    this.modelMap.clear();90    for (const [id, model] of models) this.modelMap.set(id, model);91    this.blockedList = blocked;9293    // Loud by design: a blocked metric must never fail silently (stale badge instead).94    for (const b of blocked) {95      console.error(`[store] BLOCKED metric '${b.metricId}' — no model served (stale):`);96      for (const issue of b.issues) console.error(`[store]   ${issue.code}: ${issue.message}`);97    }9899    // Fixture-backed defaults for /v1/rt/quakes until the USGS poller first succeeds.100    const quakesModel = models.get("earthquakes_24h");101    if (quakesModel) {102      this.quakesSnapshot = {103        count24h: quakesModel.anchorValue,104        lastMajor: null,105        fetchedAt: quakesModel.observedAt,106      };107    }108  }109110  get registry(): Registry {111    if (!this.loadedRegistry) throw new Error("ModelStore used before boot()");112    return this.loadedRegistry;113  }114115  get models(): ReadonlyMap<string, CounterModel> {116    return this.modelMap;117  }118119  get blocked(): readonly BlockedMetric[] {120    return this.blockedList;121  }122123  get quakes(): QuakesSnapshot {124    return this.quakesSnapshot;125  }126127  getModel(metricId: string): CounterModel | undefined {128    return this.modelMap.get(metricId);129  }130131  isStale(metricId: string): boolean {132    return !this.modelMap.has(metricId) || this.staleSinceMap.has(metricId);133  }134135  staleSince(metricId: string): string | undefined {136    return this.staleSinceMap.get(metricId);137  }138139  /**140   * Swap the model of a true-RT (event) metric and notify SSE subscribers.141   * Event-driven jumps are legitimate (no interpolation, no anti-teleportation142   * diff); every other model family MUST go through fitAllModels + validation.143   */144  updateRtModel(metricId: string, model: CounterModel): void {145    const entry = this.registry.byId.get(metricId);146    if (!entry) throw new Error(`updateRtModel: unknown metric '${metricId}'`);147    if (entry.level !== "rt") {148      throw new Error(`updateRtModel: metric '${metricId}' is not event-driven (level ${entry.level})`);149    }150    this.modelMap.set(metricId, model);151    this.staleSinceMap.delete(metricId);152    this.emit("model", model);153  }154155  /** Record when a metric's ingestion started failing (previous model keeps serving). */156  markStale(metricId: string, atIso: string = new Date().toISOString()): void {157    if (!this.staleSinceMap.has(metricId)) this.staleSinceMap.set(metricId, atIso);158  }159160  setQuakesSnapshot(snapshot: QuakesSnapshot): void {161    this.quakesSnapshot = snapshot;162  }163164  onModel(listener: ModelListener): void {165    this.on("model", listener);166  }167168  offModel(listener: ModelListener): void {169    this.off("model", listener);170  }171}172