// A small streaming tar reader: enough for the GNU tar that adsb.lol's daily releases are (regular files, directories,
// GNU long names and pax headers). Entries are read from a ByteReader one at a time and never all held in memory, and
// the ones the caller does not want are skipped without being kept.

const BLOCK = 512;

// Pulls bytes from an async iterable of Buffers (a download, a file) with a bounded read-ahead, so the network keeps
// filling the buffer while the caller works on the previous entry.
export class ByteReader {
  constructor(source, { readAhead = 32 * 1024 * 1024 } = {}) {
    this.iterator = source[Symbol.asyncIterator]();
    this.readAhead = readAhead;
    this.chunks = [];
    this.head = 0; // offset into chunks[0]
    this.buffered = 0;
    this.done = false;
    this.error = null;
    this.position = 0; // bytes handed to the caller (or skipped)
    this.waiter = null; // resolves when more data, the end or an error arrives
    this.space = null; // resolves when the caller has consumed enough to read ahead again
    this.need = 0; // bytes the caller is waiting for, which may be more than the read-ahead
    this.closed = false;
    this.pump = this.#pump();
  }

  async #pump() {
    try {
      for (;;) {
        while (!this.closed && this.buffered >= this.readAhead && this.buffered >= this.need) await new Promise((resolve) => (this.space = resolve));
        if (this.closed) break;
        const { value, done } = await this.iterator.next();
        if (done || this.closed) break;
        if (!value || value.length === 0) continue;
        this.chunks.push(Buffer.isBuffer(value) ? value : Buffer.from(value.buffer, value.byteOffset, value.byteLength));
        this.buffered += value.length;
        this.#wake();
      }
    } catch (error) {
      if (!this.closed) this.error = error;
    }
    this.done = true;
    this.#wake();
  }

  #wake() {
    const w = this.waiter;
    this.waiter = null;
    if (w) w();
  }

  async #fill(n) {
    if (this.buffered < n && !this.done) {
      this.need = n;
      this.#consumed(true);
      while (this.buffered < n && !this.done) await new Promise((resolve) => (this.waiter = resolve));
      this.need = 0;
    }
    if (this.error) throw this.error;
  }

  #consumed(force = false) {
    if (this.space && (force || this.buffered < this.readAhead)) {
      const s = this.space;
      this.space = null;
      s();
    }
  }

  // Exactly n bytes, or null at a clean end of stream (nothing left). A stream that ends part-way through is an error.
  async read(n) {
    if (n === 0) return Buffer.alloc(0);
    await this.#fill(n);
    if (this.buffered === 0 && this.done) return null;
    if (this.buffered < n) throw new Error(`The archive ended early: wanted ${n} bytes at ${this.position}, ${this.buffered} left.`);
    let out;
    const first = this.chunks[0];
    if (first.length - this.head >= n) {
      out = first.subarray(this.head, this.head + n);
      this.head += n;
      if (this.head === first.length) {
        this.chunks.shift();
        this.head = 0;
      }
    } else {
      out = Buffer.allocUnsafe(n);
      let at = 0;
      while (at < n) {
        const c = this.chunks[0];
        const take = Math.min(c.length - this.head, n - at);
        c.copy(out, at, this.head, this.head + take);
        at += take;
        this.head += take;
        if (this.head === c.length) {
          this.chunks.shift();
          this.head = 0;
        }
      }
    }
    this.buffered -= n;
    this.position += n;
    this.#consumed();
    return out;
  }

  async skip(n) {
    let left = n;
    while (left > 0) {
      await this.#fill(1);
      if (this.buffered === 0) throw new Error(`The archive ended early while skipping ${n} bytes at ${this.position}.`);
      const c = this.chunks[0];
      const take = Math.min(c.length - this.head, left);
      this.head += take;
      if (this.head === c.length) {
        this.chunks.shift();
        this.head = 0;
      }
      this.buffered -= take;
      this.position += take;
      left -= take;
      this.#consumed();
    }
  }

  // Stops the read-ahead (and with it the download), also when the caller is finished early or failed.
  async close() {
    this.closed = true;
    this.chunks = [];
    this.buffered = 0;
    if (this.space) {
      const s = this.space;
      this.space = null;
      s();
    }
    // A read in flight finishes first (a stalled download gives up on its own idle timeout).
    await this.pump;
    try {
      await this.iterator.return?.();
    } catch {
      // closing a stream that already failed
    }
  }
}

function text(buf, start, length) {
  const end = buf.indexOf(0, start);
  return buf.toString("utf8", start, end === -1 || end > start + length ? start + length : end);
}

function octal(buf, start, length) {
  // GNU base-256 for sizes of 8 GiB and more: the top bit of the first byte is set.
  if (buf[start] & 0x80) {
    let value = 0;
    for (let i = start + 1; i < start + length; i++) value = value * 256 + buf[i];
    return value;
  }
  const s = text(buf, start, length).trim();
  return s ? parseInt(s, 8) : 0;
}

function checksumOk(header) {
  let sum = 0;
  for (let i = 0; i < BLOCK; i++) sum += i >= 148 && i < 156 ? 32 : header[i];
  return sum === octal(header, 148, 8);
}

function paxPath(data) {
  // "<length> key=value\n" records
  const s = data.toString("utf8");
  let at = 0;
  let path = null;
  while (at < s.length) {
    const space = s.indexOf(" ", at);
    if (space < 0) break;
    const len = parseInt(s.slice(at, space), 10);
    if (!len) break;
    const record = s.slice(space + 1, at + len - 1);
    const eq = record.indexOf("=");
    if (record.slice(0, eq) === "path") path = record.slice(eq + 1);
    at += len;
  }
  return path;
}

const padding = (size) => (BLOCK - (size % BLOCK)) % BLOCK;

// Yields { name, size, data } for each regular file `want(name, size)` accepts; the rest are skipped. The data buffer
// may be a view into the reader's own chunk, so a caller that keeps it must copy it.
export async function* tarEntries(reader, want = () => true) {
  let longName = null;
  let zeros = 0;
  for (;;) {
    const header = await reader.read(BLOCK);
    if (header === null) return;
    if (header.every((b) => b === 0)) {
      if (++zeros === 2) return;
      continue;
    }
    zeros = 0;
    if (!checksumOk(header)) throw new Error(`Not a tar header at byte ${reader.position - BLOCK}.`);
    const type = header[156] === 0 ? "0" : String.fromCharCode(header[156]);
    const size = octal(header, 124, 12);
    const magic = header.toString("latin1", 257, 263);
    let name = text(header, 0, 100);
    // POSIX ustar keeps a path prefix at 345; old GNU headers use that space for times instead.
    if (magic === "ustar\0") {
      const prefix = text(header, 345, 155);
      if (prefix) name = `${prefix}/${name}`;
    }
    if (type === "L" || type === "x") {
      const data = await reader.read(size);
      await reader.skip(padding(size));
      const value = type === "L" ? text(data, 0, size) : paxPath(data);
      if (value) longName = value;
      continue;
    }
    if (longName) {
      name = longName;
      longName = null;
    }
    if ((type === "0" || type === "7") && want(name, size)) {
      const data = await reader.read(size);
      await reader.skip(padding(size));
      yield { name, size, data };
    } else {
      await reader.skip(size + padding(size));
    }
  }
}

// Builds a tar archive in memory: for tests, which need a small archive shaped like adsb.lol's.
export function buildTar(files) {
  const blocks = [];
  for (const { name, data } of files) {
    const body = Buffer.isBuffer(data) ? data : Buffer.from(data);
    const header = Buffer.alloc(BLOCK);
    header.write(name, 0, 100, "utf8");
    header.write("0000644\0", 100, "latin1");
    header.write("0000000\0", 108, "latin1");
    header.write("0000000\0", 116, "latin1");
    header.write(body.length.toString(8).padStart(11, "0") + "\0", 124, "latin1");
    header.write("00000000000\0", 136, "latin1");
    header.write(name.endsWith("/") ? "5" : "0", 156, "latin1");
    header.write("ustar  \0", 257, "latin1");
    header.fill(32, 148, 156);
    let sum = 0;
    for (let i = 0; i < BLOCK; i++) sum += header[i];
    header.write(sum.toString(8).padStart(6, "0") + "\0 ", 148, "latin1");
    blocks.push(header, body, Buffer.alloc(padding(body.length)));
  }
  blocks.push(Buffer.alloc(BLOCK * 2));
  return Buffer.concat(blocks);
}