#!/usr/bin/env node
// The nightly ingest: one UTC day of adsb.lol's open ADS-B data (ODbL 1.0) in, that day's observed airport movements
// out, into RW_ADSB_DB. Run by radarworks-adsb.timer at 04:20 UTC, after adsb.lol publishes the day (~03:25 UTC).
//
//   node src/server/adsb/ingest.mjs                       yesterday (UTC), unless it is already stored
//   node src/server/adsb/ingest.mjs --date 2026-10-25     one day (again with --force)
//   node src/server/adsb/ingest.mjs --catch-up [N]        every missing day of the last N (RW_ADSB_CATCHUP_DAYS, 8)
//   node src/server/adsb/ingest.mjs --date D --tar a b    read local copies of the parts instead (development)
//
// It streams the day's tar parts from GitHub (never written to disk), unpacks and parses one aircraft trace at a
// time, keeps the movements, and writes the day in one transaction. Built to be gentle on a shared server: one core,
// lowest CPU priority, a bounded read-ahead and well under 1.2 GB of memory.
import { closeSync, mkdirSync, openSync, readFileSync, realpathSync, unlinkSync, writeSync } from "node:fs";
import os from "node:os";
import path from "node:path";
import { setTimeout as sleep } from "node:timers/promises";
import { fileURLToPath } from "node:url";
import { gunzipSync } from "node:zlib";
import { fileBytes, findRelease, releaseBytes } from "./download.mjs";
import { mayHaveAirline, movementsOf, newStats, settle } from "./movements.mjs";
import { loadReference, otherFromRoute, routesFor } from "./reference.mjs";
import { addDays, adsbSettings, openForWriting, pairAcrossMidnight, purgeOld, writeDay } from "./store.mjs";
import { ByteReader, tarEntries } from "./tar.mjs";

const DATE = /^\d{4}-\d{2}-\d{2}$/;
const TRACE = /(?:^|\/)traces\/[0-9a-f~]{2}\/trace_full_[0-9a-f~]+\.json$/;
const utcToday = () => new Date().toISOString().slice(0, 10);
const stamp = () => new Date().toISOString().replace(/\.\d{3}Z$/, "Z");

function defaultLog(line) {
  console.log(`${stamp()} ${line}`);
}

// One ingest at a time on a machine (the timer and an admin's command could otherwise overlap).
function lock(workDir) {
  mkdirSync(workDir, { recursive: true });
  const file = path.join(workDir, "ingest.lock");
  for (let attempt = 0; attempt < 2; attempt++) {
    try {
      const fd = openSync(file, "wx");
      writeSync(fd, String(process.pid));
      closeSync(fd);
      return () => {
        try {
          unlinkSync(file);
        } catch {
          // already gone
        }
      };
    } catch (error) {
      if (error.code !== "EEXIST") throw error;
      const pid = Number(readFileSync(file, "utf8"));
      let alive = false;
      try {
        process.kill(pid, 0);
        alive = pid > 0;
      } catch {
        alive = false;
      }
      if (alive) throw new Error(`Another ingest is running (pid ${pid}).`);
      unlinkSync(file);
    }
  }
  throw new Error("Could not take the ingest lock.");
}

const mb = (n) => (n / 1e6).toFixed(0);

// Same aircraft, same airport, same direction within ten minutes: one movement seen twice.
function dedupe(movements) {
  movements.sort((a, b) => (a.airport < b.airport ? -1 : a.airport > b.airport ? 1 : a.hex < b.hex ? -1 : a.hex > b.hex ? 1 : a.kind < b.kind ? -1 : a.kind > b.kind ? 1 : a.utc - b.utc));
  const out = [];
  for (const m of movements) {
    const last = out[out.length - 1];
    if (last && last.airport === m.airport && last.hex === m.hex && last.kind === m.kind && m.utc - last.utc < 600) {
      if (!last.other && m.other) Object.assign(last, { other: m.other, via: m.via });
      continue;
    }
    out.push(m);
  }
  return out;
}

export async function ingestDay(date, { force = false, tarFiles = null, log = defaultLog, fetchImpl = fetch, waitHours = 0, settings = adsbSettings(), reference = null } = {}) {
  if (!DATE.test(date) || Number.isNaN(Date.parse(`${date}T00:00:00Z`))) throw new Error("Give the date as YYYY-MM-DD.");
  if (date >= utcToday() && !tarFiles) throw new Error(`${date} is not over yet in UTC; adsb.lol publishes a day the morning after.`);
  const db = openForWriting(settings.db);
  if (!force && db.prepare("SELECT 1 FROM ingested WHERE day = ?").get(date)) {
    log(`${date}: already ingested (use --force to replace it).`);
    return { date, skipped: true };
  }
  const unlock = lock(settings.work);
  try {
    try {
      os.setPriority(19);
    } catch {
      // not allowed here; systemd's Nice= does it on the server
    }
    const started = Date.now();
    const startedAt = stamp();
    const ref = reference || (await loadReference(settings.work, { log, fetchImpl }));

    let tag;
    let total;
    let source;
    let downloaded = 0;
    const onBytes = (n) => {
      downloaded += n;
    };
    if (tarFiles) {
      tag = `local:${path.basename(tarFiles[0])}`;
      const { statSync } = await import("node:fs");
      total = tarFiles.reduce((n, f) => n + statSync(f).size, 0);
      source = fileBytes(tarFiles, { onBytes });
    } else {
      let release = await findRelease(date, { fetchImpl, log });
      const deadline = Date.now() + waitHours * 3600000;
      while (!release && Date.now() < deadline) {
        log(`${date}: not published yet; looking again in 20 minutes.`);
        await sleep(20 * 60000);
        release = await findRelease(date, { fetchImpl, log });
      }
      if (!release) throw new Error(`adsb.lol has not published ${date} (no ${date.replaceAll("-", ".")} release in adsblol/globe_history_${date.slice(0, 4)}).`);
      tag = release.tag;
      total = release.size;
      log(`${date}: ${tag}, ${release.parts.length} parts, ${(total / 1e9).toFixed(2)} GB.`);
      source = releaseBytes(release.parts, { fetchImpl, log, onBytes });
    }

    const day0 = Date.parse(`${date}T00:00:00Z`) / 1000;
    const stats = newStats();
    let movements = [];
    let peakRss = 0;
    let lastYield = performance.now();
    let lastLog = Date.now();
    const progress = () => {
      const rss = process.memoryUsage.rss();
      peakRss = Math.max(peakRss, rss);
      const secs = (Date.now() - started) / 1000;
      const pct = total ? ((downloaded * 100) / total).toFixed(1) : "?";
      log(
        `${date}: ${pct}% (${mb(downloaded)} of ${mb(total)} MB, ${(downloaded / 1e6 / secs).toFixed(1)} MB/s), ${stats.traces} traces, ${stats.legs} legs, ${movements.length} movements, RSS ${mb(rss)} MB`,
      );
    };
    const reader = new ByteReader(source);
    try {
      for await (const entry of tarEntries(reader, (name) => TRACE.test(name))) {
        stats.traces++;
        let json;
        try {
          const raw = entry.data[0] === 0x1f && entry.data[1] === 0x8b ? gunzipSync(entry.data) : entry.data;
          // latin1 is faster than utf8 and the fields read here (callsign, registration, type) are ASCII.
          json = raw.toString("latin1");
        } catch {
          stats.tracesUnreadable++;
          continue;
        }
        if (!mayHaveAirline(json)) {
          stats.tracesNoAirlineCallsign++;
          continue;
        }
        let doc;
        try {
          doc = JSON.parse(json);
        } catch {
          stats.tracesUnreadable++;
          continue;
        }
        json = null;
        stats.tracesParsed++;
        for (const m of movementsOf(doc, ref, stats, day0)) movements.push(m);
        // Let the download's read-ahead run between traces.
        if (performance.now() - lastYield > 25) {
          await new Promise((resolve) => setImmediate(resolve));
          lastYield = performance.now();
        }
        if (Date.now() - lastLog > 30000) {
          lastLog = Date.now();
          progress();
        }
      }
    } finally {
      await reader.close();
    }
    progress();
    if (!stats.traces) throw new Error(`${tag} held no aircraft traces; its layout may have changed.`);

    // Route data (CC0) for every flight with an end still open: it settles ends seen only from afar, then names the
    // other end of one-ended movements when the observed airport is on the route.
    const wanted = new Set(movements.filter((m) => !m.other).map((m) => m.callsign));
    const routes = await routesFor(ref.routesFile, wanted);
    movements = settle(movements, routes, ref, stats);
    let fromRoutes = 0;
    for (const m of movements) {
      if (m.other) continue;
      const other = otherFromRoute(routes.get(m.callsign), m.airport, m.kind);
      if (other && other !== m.airport && ref.index.get(other)) {
        m.other = other;
        m.via = "R";
        fromRoutes++;
      }
    }
    const before = movements.length;
    movements = dedupe(movements);
    stats.duplicates = before - movements.length;
    stats.otherFromRoutes = fromRoutes;
    stats.movements = movements.length;
    stats.departures = movements.filter((m) => m.kind === "D").length;
    stats.arrivals = movements.length - stats.departures;
    stats.withOther = movements.filter((m) => m.other).length;
    stats.archiveBytes = downloaded;

    const seconds = (Date.now() - started) / 1000;
    peakRss = Math.max(peakRss, process.resourceUsage().maxRSS * 1024);
    stats.seconds = Math.round(seconds);
    stats.peakRssMb = Math.round(peakRss / 1e6);
    stats.mbPerSecond = Number((downloaded / 1e6 / seconds).toFixed(1));
    const written = writeDay(db, {
      day: date,
      tag,
      startedAt,
      seconds,
      stats,
      movements,
      airports: [...ref.index.byIcao.values()],
      airlines: ref.airlines,
    });
    const paired = pairAcrossMidnight(db, addDays(date, -1)) + pairAcrossMidnight(db, date);
    const purged = purgeOld(db, utcToday(), settings.retentionDays);
    log(
      `${date}: stored ${movements.length} movements (${stats.departures} departures, ${stats.arrivals} arrivals) at ${written.airports} airports from ${stats.traces} traces in ${Math.round(seconds)} s` +
        ` (${stats.mbPerSecond} MB/s, peak RSS ${stats.peakRssMb} MB); other end known for ${movements.length ? Math.round((stats.withOther * 100) / movements.length) : 0}%` +
        ` (${fromRoutes} from route data, ${paired} paired across midnight).${purged.length ? ` Purged ${purged.join(", ")}.` : ""}`,
    );
    log(`${date}: dropped legs ${JSON.stringify(stats.dropped)}.`);
    return { date, tag, seconds, stats, airports: written.airports, paired, purged };
  } finally {
    unlock();
  }
}

// Every day of the last `days` (yesterday first) that is not stored yet. A failed day does not stop the others.
export async function catchUp(days = adsbSettings().catchUpDays, { log = defaultLog, fetchImpl = fetch, settings = adsbSettings() } = {}) {
  const db = openForWriting(settings.db);
  const yesterday = addDays(utcToday(), -1);
  const have = new Set(db.prepare("SELECT day FROM ingested").all().map((r) => r.day));
  const todo = [];
  for (let i = 0; i < days; i++) {
    const day = addDays(yesterday, -i);
    if (!have.has(day)) todo.push(day);
  }
  if (!todo.length) {
    log(`Nothing to do: the last ${days} days are stored.`);
    return { done: [], failed: [] };
  }
  log(`To ingest: ${todo.join(", ")}.`);
  const done = [];
  const failed = [];
  let reference = null;
  for (const day of todo) {
    try {
      reference = reference || (await loadReference(settings.work, { log, fetchImpl }));
      await ingestDay(day, { log, fetchImpl, settings, reference, waitHours: day === yesterday ? settings.waitHours : 0 });
      done.push(day);
    } catch (error) {
      log(`${day}: FAILED: ${error?.message || error}`);
      failed.push(day);
    }
  }
  return { done, failed };
}

async function main(argv) {
  const args = { tar: [] };
  for (let i = 0; i < argv.length; i++) {
    const a = argv[i];
    if (a === "--date") args.date = argv[++i];
    else if (a === "--force") args.force = true;
    else if (a === "--catch-up") args.catchUp = /^\d+$/.test(argv[i + 1] || "") ? Number(argv[++i]) : adsbSettings().catchUpDays;
    else if (a === "--tar") while (argv[i + 1] && !argv[i + 1].startsWith("--")) args.tar.push(argv[++i]);
    else if (a === "--help" || a === "-h") {
      console.log(readFileSync(fileURLToPath(import.meta.url), "utf8").split("\n").slice(1, 9).map((l) => l.slice(3)).join("\n"));
      return 0;
    } else throw new Error(`Unknown argument ${a}`);
  }
  if (args.catchUp) {
    const r = await catchUp(args.catchUp);
    return r.failed.length ? 1 : 0;
  }
  const date = args.date || addDays(utcToday(), -1);
  await ingestDay(date, { force: args.force, tarFiles: args.tar.length ? args.tar : null, waitHours: args.date ? 0 : adsbSettings().waitHours });
  return 0;
}

// Run as a script (the service starts it through /opt/radarworks/current, a symlink, while Node loads the module from
// its real path, so compare real paths).
const invokedAsScript = (() => {
  try {
    return Boolean(process.argv[1]) && realpathSync(process.argv[1]) === realpathSync(fileURLToPath(import.meta.url));
  } catch {
    return false;
  }
})();
if (invokedAsScript) {
  main(process.argv.slice(2)).then(
    (code) => process.exit(code),
    (error) => {
      defaultLog(`FAILED: ${error?.message || error}`);
      process.exit(1);
    },
  );
}