// Finding and streaming one day of adsb.lol's open data (github.com/adsblol/globe_history_<year>, ODbL 1.0).
//
// Each UTC day is one GitHub release, tagged v<YYYY.MM.DD>-planes-readsb-<prod|staging>-<n>, published around 03:25
// UTC the next day. Its assets are one tar split into ~2 GB parts (.tar.aa, .tar.ab, sometimes .tar.ac). The parts are
// streamed in order as one byte stream and never written to disk; a broken connection is resumed with a Range request
// from the byte it stopped at. Each day costs one or two API calls and one download.
import { setTimeout as sleep } from "node:timers/promises";

export const USER_AGENT = "radarworks-adsb-ingest/1 (+https://radarworks.net/open-data)";
const API = "https://api.github.com";

function apiHeaders() {
  const headers = { "User-Agent": USER_AGENT, Accept: "application/vnd.github+json", "X-GitHub-Api-Version": "2022-11-28" };
  if (process.env.GITHUB_TOKEN) headers.Authorization = `Bearer ${process.env.GITHUB_TOKEN}`;
  return headers;
}

export function tagFor(date, feed = "prod", n = 0) {
  return `v${date.replaceAll("-", ".")}-planes-readsb-${feed}-${n}`;
}

async function getRelease(repo, tag, fetchImpl) {
  const response = await fetchImpl(`${API}/repos/adsblol/${repo}/releases/tags/${tag}`, { headers: apiHeaders(), signal: AbortSignal.timeout(30000) });
  if (response.status === 404) return null;
  if (!response.ok) {
    const reset = response.headers.get("x-ratelimit-reset");
    const hint = response.status === 403 || response.status === 429 ? ` (rate limited${reset ? ` until ${new Date(Number(reset) * 1000).toISOString()}` : ""}; set GITHUB_TOKEN)` : "";
    throw new Error(`GitHub answered ${response.status} for ${tag}${hint}.`);
  }
  return response.json();
}

function partsOf(release) {
  // Only the split tar parts, in suffix order (.tar.aa, .tar.ab, ...); a release with a single .tar is also accepted.
  const parts = (release.assets || [])
    .filter((a) => /\.tar(\.[a-z]{2})?$/.test(a.name))
    .map((a) => ({ name: a.name, size: a.size, url: a.browser_download_url }))
    .sort((a, b) => a.name.localeCompare(b.name));
  return parts;
}

// The release to read for a day: prod, unless it is missing or clearly smaller than staging (adsb.lol's README says
// to prefer whichever of the two is complete).
export async function findRelease(date, { fetchImpl = fetch, log = () => {} } = {}) {
  const repo = `globe_history_${date.slice(0, 4)}`;
  const prod = await getRelease(repo, tagFor(date, "prod"), fetchImpl);
  const prodParts = prod ? partsOf(prod) : [];
  const prodSize = prodParts.reduce((n, p) => n + p.size, 0);
  let chosen = prod && prodParts.length ? { tag: prod.tag_name, parts: prodParts, size: prodSize, publishedAt: prod.published_at } : null;
  // Staging can still be uploading while yesterday's prod is on its way (adsb.lol publishes prod about 03:25 UTC), so it
  // only stands in for days at least two days old, when prod is missing or clearly incomplete.
  const ageDays = Math.floor((Date.now() - Date.parse(`${date}T00:00:00Z`)) / 86400000);
  if ((!chosen || prodSize < 2.5e9) && ageDays >= 2) {
    const staging = await getRelease(repo, tagFor(date, "staging"), fetchImpl);
    const stagingParts = staging ? partsOf(staging) : [];
    const stagingSize = stagingParts.reduce((n, p) => n + p.size, 0);
    if (stagingParts.length && stagingSize > prodSize * 1.25) {
      log(`prod for ${date} is ${(prodSize / 1e9).toFixed(2)} GB, staging ${(stagingSize / 1e9).toFixed(2)} GB: using staging.`);
      chosen = { tag: staging.tag_name, parts: stagingParts, size: stagingSize, publishedAt: staging.published_at };
    }
  }
  return chosen;
}

// One read from the body, or an error when nothing arrives for `idleMs`. The timer only runs while waiting on the
// network, never while the caller is busy with the previous chunk.
async function readWithin(reader, idleMs) {
  let timer;
  try {
    return await Promise.race([
      reader.read(),
      new Promise((_, reject) => {
        timer = setTimeout(() => reject(new Error(`no data for ${idleMs / 1000} s`)), idleMs);
      }),
    ]);
  } finally {
    clearTimeout(timer);
  }
}

// The parts as one stream of Buffers. Each part is requested once; after a network error it is requested again from
// the byte it stopped at (GitHub's asset storage honours Range), with growing pauses, up to `retries` times in a row.
export async function* releaseBytes(parts, { fetchImpl = fetch, log = () => {}, retries = 8, idleMs = 120000, onBytes = () => {} } = {}) {
  for (const part of parts) {
    let got = 0;
    let failures = 0;
    while (got < part.size) {
      let reader = null;
      try {
        const headers = { "User-Agent": USER_AGENT };
        if (got > 0) headers.Range = `bytes=${got}-`;
        const response = await fetchImpl(part.url, { headers, redirect: "follow", signal: AbortSignal.timeout(60000 * 60 * 3) });
        if (got > 0 && response.status !== 206) throw new Error(`resume of ${part.name} at ${got} answered ${response.status}`);
        if (got === 0 && response.status !== 200) throw new Error(`${part.name} answered ${response.status}`);
        reader = response.body.getReader();
        for (;;) {
          const { value, done } = await readWithin(reader, idleMs);
          if (done) break;
          const room = part.size - got;
          const chunk = value.length > room ? value.subarray(0, room) : value;
          got += chunk.length;
          failures = 0;
          onBytes(chunk.length);
          yield Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength);
          if (got >= part.size) break;
        }
        if (got < part.size) throw new Error(`${part.name} ended at ${got} of ${part.size} bytes`);
      } catch (error) {
        failures++;
        if (failures > retries) throw new Error(`Download failed after ${retries} retries: ${error?.message || error}`);
        const pause = Math.min(300, 5 * 2 ** (failures - 1));
        log(`network: ${error?.message || error}; resuming ${part.name} at ${got} in ${pause} s (try ${failures}/${retries}).`);
        await sleep(pause * 1000);
      } finally {
        try {
          await reader?.cancel();
        } catch {
          // already closed
        }
      }
    }
  }
}

// Local copies of the parts (development and tests): the same stream, read from disk.
export async function* fileBytes(files, { onBytes = () => {} } = {}) {
  const { createReadStream } = await import("node:fs");
  for (const file of files) {
    for await (const chunk of createReadStream(file, { highWaterMark: 1 << 20 })) {
      onBytes(chunk.length);
      yield chunk;
    }
  }
}