spb/earth-now Public License

1/**2 * earth-now.co3 * Author:  Simon-Pierre Boucher4 * Contact: contact@spboucher.ai5 * File:    apps/ingest/src/runners.ts6 * Purpose: Per-source run functions (fetch → archive raw → parse → summary) shared by the CLI and the queue worker7 */89import { archiveRaw } from "./archive.js";10import { type FetchLike, defaultFetch } from "./fetch-like.js";11import { NOAA_CO2_URL, parseNoaaMonthlyCo2 } from "./sources/noaa-co2.js";12import { parseUsgsCount, usgsCountUrl } from "./sources/usgs.js";1314/** USGS live counter parameters: quakes M ≥ 4.5 over the trailing 24 h. */15export const USGS_WINDOW_HOURS = 24;16// Must match the registry definition of earthquakes_24h (catalog: M ≥ 2.5).17export const USGS_MIN_MAGNITUDE = 2.5;1819export interface ObservationPoint {20  time: string;21  value: number;22}2324export interface RunSummary {25  sourceId: string;26  observationCount: number;27  first: ObservationPoint | null;28  last: ObservationPoint | null;29  rawPath: string;30  sha256: string;31}3233async function fetchText(url: string, fetchImpl: FetchLike): Promise<string> {34  const res = await fetchImpl(url);35  if (!res.ok) throw new Error(`fetch failed: HTTP ${res.status} for ${url}`);36  return res.text();37}3839/** USGS FDSN: one observation — the M≥4.5 quake count over the trailing 24 h window. */40export async function runUsgsFdsn(fetchImpl: FetchLike = defaultFetch): Promise<RunSummary> {41  const url = usgsCountUrl(USGS_WINDOW_HOURS, USGS_MIN_MAGNITUDE);42  const text = await fetchText(url, fetchImpl);43  const archived = await archiveRaw("usgs_fdsn", text);44  const count = parseUsgsCount(JSON.parse(text));45  const point: ObservationPoint = { time: new Date().toISOString(), value: count };46  return {47    sourceId: "usgs_fdsn",48    observationCount: 1,49    first: point,50    last: point,51    rawPath: archived.rawPath,52    sha256: archived.sha256,53  };54}5556/** NOAA GML Mauna Loa: full monthly CO₂ series (Keeling curve). */57export async function runNoaaGmlMlo(fetchImpl: FetchLike = defaultFetch): Promise<RunSummary> {58  const text = await fetchText(NOAA_CO2_URL, fetchImpl);59  const archived = await archiveRaw("noaa_gml_mlo", text);60  const observations = parseNoaaMonthlyCo2(text);61  return {62    sourceId: "noaa_gml_mlo",63    observationCount: observations.length,64    first: observations[0] ?? null,65    last: observations[observations.length - 1] ?? null,66    rawPath: archived.rawPath,67    sha256: archived.sha256,68  };69}7071export interface SourceSpec {72  id: string;73  /** Scheduling cadence in ms (usgs 5 min, noaa daily). */74  cadenceMs: number;75  run: (fetchImpl?: FetchLike) => Promise<RunSummary>;76}7778/** Every source the ingest service knows how to run, with its scheduling cadence. */79export const SOURCES: readonly SourceSpec[] = [80  { id: "usgs_fdsn", cadenceMs: 5 * 60_000, run: runUsgsFdsn },81  { id: "noaa_gml_mlo", cadenceMs: 24 * 3_600_000, run: runNoaaGmlMlo },82];8384export const KNOWN_SOURCE_IDS: readonly string[] = SOURCES.map((s) => s.id);8586/** Run one source by id. Throws on unknown id or on any fetch/parse failure. */87export async function runSource(88  sourceId: string,89  fetchImpl: FetchLike = defaultFetch,90): Promise<RunSummary> {91  const spec = SOURCES.find((s) => s.id === sourceId);92  if (!spec) throw new Error(`Unknown source '${sourceId}'. Known: ${KNOWN_SOURCE_IDS.join(", ")}`);93  return spec.run(fetchImpl);94}95