/**
 * earth-now.co
 * Author:  Simon-Pierre Boucher
 * Contact: contact@spboucher.ai
 * File:    apps/ingest/src/index.ts
 * Purpose: Ingestion orchestrator — BullMQ repeatable jobs when REDIS_URL is set, plain setInterval scheduler otherwise; never crashes on a failing source
 */

import { SOURCES, runSource } from "./runners.js";

const log = {
  info: (msg: string): void => console.log(`${new Date().toISOString()} [ingest] ${msg}`),
  warn: (msg: string): void => console.warn(`${new Date().toISOString()} [ingest] WARN ${msg}`),
  error: (msg: string): void => console.error(`${new Date().toISOString()} [ingest] ERROR ${msg}`),
};

interface SourceState {
  lastSuccessMs: number | null;
  failureCount: number;
  startedMs: number;
}

const states = new Map<string, SourceState>(
  SOURCES.map((s) => [s.id, { lastSuccessMs: null, failureCount: 0, startedMs: Date.now() }]),
);

/**
 * Production guardrail: a source failing for longer than 2× its cadence must raise the
 * "stale" badge + alert. Stub for now — wired to the API alerting later.
 */
function checkStale(sourceId: string, cadenceMs: number): void {
  const state = states.get(sourceId);
  if (!state) return;
  const reference = state.lastSuccessMs ?? state.startedMs;
  if (Date.now() - reference > 2 * cadenceMs) {
    log.warn(
      `[stale] ${sourceId} has had no successful ingestion for > 2× its cadence ` +
        `(${state.failureCount} consecutive failure(s)) — would trigger the stale badge + alert (stub)`,
    );
  }
}

/** Run one source, absorbing every failure: log + count, never crash the scheduler. */
async function tick(sourceId: string, cadenceMs: number): Promise<void> {
  const state = states.get(sourceId);
  try {
    const summary = await runSource(sourceId);
    if (state) {
      state.lastSuccessMs = Date.now();
      state.failureCount = 0;
    }
    log.info(
      `${sourceId}: ok — ${summary.observationCount} observation(s), ` +
        `last=${summary.last ? `${summary.last.time} → ${summary.last.value}` : "n/a"}, raw=${summary.rawPath}`,
    );
  } catch (err) {
    if (state) state.failureCount += 1;
    log.error(`${sourceId}: run failed — ${err instanceof Error ? err.message : String(err)}`);
    checkStale(sourceId, cadenceMs);
  }
}

async function startWithRedis(redisUrl: string): Promise<void> {
  // bullmq/ioredis are imported ONLY when REDIS_URL is set — dev stays dependency-light.
  const { Queue, Worker } = await import("bullmq");
  const { Redis } = await import("ioredis");
  const connection = new Redis(redisUrl, { maxRetriesPerRequest: null });

  const queue = new Queue("ingest", { connection });
  for (const source of SOURCES) {
    await queue.add(
      source.id,
      {},
      {
        repeat: { every: source.cadenceMs },
        jobId: `repeat:${source.id}`,
        removeOnComplete: 100,
        removeOnFail: 500,
      },
    );
    log.info(`scheduled '${source.id}' every ${source.cadenceMs / 1000}s (BullMQ repeatable)`);
  }

  const worker = new Worker(
    "ingest",
    async (job) => {
      const source = SOURCES.find((s) => s.id === job.name);
      await tick(job.name, source?.cadenceMs ?? 0);
    },
    { connection },
  );
  worker.on("error", (err) => log.error(`worker error: ${err.message}`));
  log.info(`BullMQ mode: queue "ingest" on ${redisUrl}`);
}

function startRedisLess(): void {
  log.info("REDIS_URL not set — Redis-less dev mode: plain setInterval scheduler");
  for (const source of SOURCES) {
    void tick(source.id, source.cadenceMs);
    setInterval(() => void tick(source.id, source.cadenceMs), source.cadenceMs);
    log.info(`scheduled '${source.id}' every ${source.cadenceMs / 1000}s (setInterval)`);
  }
}

async function main(): Promise<void> {
  const redisUrl = process.env.REDIS_URL;
  if (redisUrl) await startWithRedis(redisUrl);
  else startRedisLess();
}

main().catch((err: unknown) => {
  log.error(`fatal startup error: ${err instanceof Error ? err.message : String(err)}`);
  process.exit(1);
});