spb/earth-now Public License

1/**2 * earth-now.co3 * Author:  Simon-Pierre Boucher4 * Contact: contact@spboucher.ai5 * File:    apps/ingest/src/index.ts6 * Purpose: Ingestion orchestrator — BullMQ repeatable jobs when REDIS_URL is set, plain setInterval scheduler otherwise; never crashes on a failing source7 */89import { SOURCES, runSource } from "./runners.js";1011const log = {12  info: (msg: string): void => console.log(`${new Date().toISOString()} [ingest] ${msg}`),13  warn: (msg: string): void => console.warn(`${new Date().toISOString()} [ingest] WARN ${msg}`),14  error: (msg: string): void => console.error(`${new Date().toISOString()} [ingest] ERROR ${msg}`),15};1617interface SourceState {18  lastSuccessMs: number | null;19  failureCount: number;20  startedMs: number;21}2223const states = new Map<string, SourceState>(24  SOURCES.map((s) => [s.id, { lastSuccessMs: null, failureCount: 0, startedMs: Date.now() }]),25);2627/**28 * Production guardrail: a source failing for longer than 2× its cadence must raise the29 * "stale" badge + alert. Stub for now — wired to the API alerting later.30 */31function checkStale(sourceId: string, cadenceMs: number): void {32  const state = states.get(sourceId);33  if (!state) return;34  const reference = state.lastSuccessMs ?? state.startedMs;35  if (Date.now() - reference > 2 * cadenceMs) {36    log.warn(37      `[stale] ${sourceId} has had no successful ingestion for > 2× its cadence ` +38        `(${state.failureCount} consecutive failure(s)) — would trigger the stale badge + alert (stub)`,39    );40  }41}4243/** Run one source, absorbing every failure: log + count, never crash the scheduler. */44async function tick(sourceId: string, cadenceMs: number): Promise<void> {45  const state = states.get(sourceId);46  try {47    const summary = await runSource(sourceId);48    if (state) {49      state.lastSuccessMs = Date.now();50      state.failureCount = 0;51    }52    log.info(53      `${sourceId}: ok — ${summary.observationCount} observation(s), ` +54        `last=${summary.last ? `${summary.last.time} → ${summary.last.value}` : "n/a"}, raw=${summary.rawPath}`,55    );56  } catch (err) {57    if (state) state.failureCount += 1;58    log.error(`${sourceId}: run failed — ${err instanceof Error ? err.message : String(err)}`);59    checkStale(sourceId, cadenceMs);60  }61}6263async function startWithRedis(redisUrl: string): Promise<void> {64  // bullmq/ioredis are imported ONLY when REDIS_URL is set — dev stays dependency-light.65  const { Queue, Worker } = await import("bullmq");66  const { Redis } = await import("ioredis");67  const connection = new Redis(redisUrl, { maxRetriesPerRequest: null });6869  const queue = new Queue("ingest", { connection });70  for (const source of SOURCES) {71    await queue.add(72      source.id,73      {},74      {75        repeat: { every: source.cadenceMs },76        jobId: `repeat:${source.id}`,77        removeOnComplete: 100,78        removeOnFail: 500,79      },80    );81    log.info(`scheduled '${source.id}' every ${source.cadenceMs / 1000}s (BullMQ repeatable)`);82  }8384  const worker = new Worker(85    "ingest",86    async (job) => {87      const source = SOURCES.find((s) => s.id === job.name);88      await tick(job.name, source?.cadenceMs ?? 0);89    },90    { connection },91  );92  worker.on("error", (err) => log.error(`worker error: ${err.message}`));93  log.info(`BullMQ mode: queue "ingest" on ${redisUrl}`);94}9596function startRedisLess(): void {97  log.info("REDIS_URL not set — Redis-less dev mode: plain setInterval scheduler");98  for (const source of SOURCES) {99    void tick(source.id, source.cadenceMs);100    setInterval(() => void tick(source.id, source.cadenceMs), source.cadenceMs);101    log.info(`scheduled '${source.id}' every ${source.cadenceMs / 1000}s (setInterval)`);102  }103}104105async function main(): Promise<void> {106  const redisUrl = process.env.REDIS_URL;107  if (redisUrl) await startWithRedis(redisUrl);108  else startRedisLess();109}110111main().catch((err: unknown) => {112  log.error(`fatal startup error: ${err instanceof Error ? err.message : String(err)}`);113  process.exit(1);114});115