'use strict';

const fs = require('fs');
const fsp = require('fs/promises');
const path = require('path');
const crypto = require('crypto');
const { EventEmitter } = require('events');
const { open, probe, NetError, classify, sanitize } = require('./net');

const MIN_SPLIT = 1 << 20;   // 1 MiB — the smallest range worth opening a lane for up front
const SPLIT_FLOOR = 128 * 1024;  // never hand over a tail smaller than this
const SPLIT_SETUP_MS = 250;      // rough cost of opening a fresh ranged request
const STALL_MS = 3000;       // no bytes for this long: the lane reads as stalled
const ABORT_MS = 12000;      // ...this long: kill the socket and retry the remainder
const TICK_MS = 250;
const RATE_SAMPLES = 4;      // ~1s of ticks behind each displayed rate
const MAX_ATTEMPTS = 6;
const SAVE_MS = 2000;
const HISTORY = 60;          // seconds kept for the throughput graph

const S = {
  QUEUED: 'queued', PROBING: 'probing', ACTIVE: 'active', PAUSED: 'paused',
  VERIFYING: 'verifying', DONE: 'done', FAILED: 'failed', CANCELLED: 'cancelled',
};

const delay = (ms, signal) => new Promise((resolve, reject) => {
  const t = setTimeout(resolve, ms);
  if (!signal) return;
  if (signal.aborted) { clearTimeout(t); return reject(new NetError('abort', 'Stopped')); }
  signal.addEventListener('abort', () => { clearTimeout(t); reject(new NetError('abort', 'Stopped')); }, { once: true });
});

const backoff = (attempt) => Math.min(15000, 500 * 2 ** attempt);

function writeAll(fd, buf, position) {
  return new Promise((resolve, reject) => {
    let written = 0;
    const step = () => {
      fs.write(fd, buf, written, buf.length - written, position + written, (err, n) => {
        if (err) return reject(err);
        written += n;
        if (written >= buf.length) return resolve();
        step();
      });
    };
    step();
  });
}

/** One connection working one byte range. The lane strip in the UI is these. */
class Lane {
  constructor(id, start, end, dl) {
    this.id = id;
    this.start = start;
    this.end = end;              // inclusive, HTTP Range semantics
    this.received = 0;
    this.dl = dl;
    this.state = 'idle';         // idle|active|stalled|retrying|paused|done|failed
    this.routeId = null;
    this.attempts = 0;
    this.error = null;
    this.rate = 0;
    this.tickBytes = 0;
    this.samples = [];
    this.lastByteAt = 0;
    this.ctrl = null;
  }

  // end is mutable: a finished lane steals the tail of the slowest one, which
  // shortens that lane live. These stay getters so the running pump sees it.
  get length() { return this.end === Infinity ? Infinity : this.end - this.start + 1; }
  get remaining() { return this.length - this.received; }
  get pos() { return this.start + this.received; }

  async run() {
    while (this.received < this.length) {
      if (this.dl.stopping) { this.state = 'paused'; return; }
      try {
        this.state = 'active';
        this.error = null;
        this.lastByteAt = Date.now();
        await this.attempt();
        break;
      } catch (err) {
        const e = classify(err);
        if (this.dl.stopping) { this.state = 'paused'; return; }
        if (this.received >= this.length) break;       // the abort was us hitting our end

        this.attempts += 1;
        this.error = e.message;
        this.dl.noteRouteError(this.routeId, e);

        if (this.attempts >= MAX_ATTEMPTS) {
          this.state = 'failed';
          throw e;
        }
        if (e.routeFault) this.routeId = this.dl.pickRoute(this.routeId);
        this.state = 'retrying';
        await delay(backoff(this.attempts), this.dl.abort.signal);
      }
    }
    this.state = 'done';
    this.rate = 0;
    this.dl.onLaneDone(this);
  }

  async attempt() {
    this.ctrl = new AbortController();
    const headers = {};
    if (this.dl.rangeable) headers.range = `bytes=${this.pos}-${this.end}`;

    const { res } = await open(this.dl.url, {
      route: this.dl.routeById(this.routeId),
      signal: this.ctrl.signal,
      headers,
    });

    if (this.dl.rangeable && res.statusCode !== 206) {
      res.destroy();
      throw new NetError('http', 'Server ignored the range request', res.statusCode);
    }
    await this.pump(res);
  }

  pump(res) {
    return new Promise((resolve, reject) => {
      let settled = false;
      let pending = Promise.resolve();   // the write currently in flight, if any

      const finish = (err) => {
        if (settled) return;
        settled = true;
        res.destroy();
        // Queued writes still have to land before we can say whether this range
        // finished. Settling first sends a lane that already holds every byte
        // into a retry backoff it does not need.
        pending.then(() => {
          if (!err || this.remaining <= 0) resolve();
          else reject(err);
        });
      };

      res.on('data', (chunk) => {
        res.pause();
        // Never write past our end: a rebalance can shorten this lane mid-flight,
        // and some servers over-send on a range request.
        const room = this.remaining;
        if (room <= 0) return finish();
        const buf = chunk.length > room ? chunk.subarray(0, room) : chunk;
        const at = this.pos;

        pending = writeAll(this.dl.fd, buf, at)
          .then(async () => {
            this.received += buf.length;
            this.tickBytes += buf.length;
            this.lastByteAt = Date.now();
            this.dl.received += buf.length;
            if (this.remaining <= 0) return finish();
            const wait = this.dl.limiter.take(buf.length);
            if (wait) await wait;
            if (!settled) res.resume();
          })
          .catch((err) => finish(classify(err)));
      });

      res.on('end', () => {
        if (this.dl.rangeable && this.remaining > 0) {
          return finish(new NetError('reset', 'Server closed the range early'));
        }
        finish();
      });
      res.on('error', (err) => finish(classify(err)));
      this.ctrl.signal.addEventListener('abort', () => {
        finish(new NetError('timeout', `No bytes for ${ABORT_MS / 1000}s, restarting the range`));
      }, { once: true });
    });
  }

  stop() { try { this.ctrl && this.ctrl.abort(); } catch { /* already closed */ } }

  snapshot() {
    const len = this.length === Infinity ? null : this.length;
    return {
      id: this.id,
      start: this.start,
      end: this.end === Infinity ? null : this.end,
      length: len,
      received: this.received,
      pct: len ? Math.min(100, (this.received / len) * 100) : null,
      rate: this.rate,
      state: this.state,
      routeId: this.routeId,
      attempts: this.attempts,
      error: this.error,
      stalledMs: this.state === 'stalled' && this.lastByteAt ? Date.now() - this.lastByteAt : 0,
    };
  }
}

class Download extends EventEmitter {
  constructor(opts, ctx) {
    super();
    this.id = opts.id || crypto.randomUUID();
    this.url = opts.url;
    this.originUrl = opts.url;
    this.dir = opts.dir;
    this.name = opts.name || null;
    this.wantConnections = Math.max(1, Math.min(32, opts.connections || 8));
    this.routeMode = opts.routeMode || 'spread';   // spread | direct | <routeId>
    this.verifyAfter = opts.verifyAfter !== false;
    this.expectedChecksum = opts.checksum || null;
    this.addedAt = opts.addedAt || Date.now();

    this.ctx = ctx;                 // { routes(), limiter, onChange }
    this.limiter = ctx.limiter;

    this.state = S.QUEUED;
    this.size = null;
    this.received = 0;
    this.rangeable = false;
    this.etag = null;
    this.lastModified = null;
    this.filePath = null;
    this.fd = null;
    this.lanes = [];
    this.error = null;
    this.startedAt = null;
    this.finishedAt = null;
    this.retries = 0;
    this.rebalances = 0;
    this.rate = 0;
    this.history = [];
    this.verify = { state: 'idle', progress: 0, hash: null, match: null };
    this.abort = new AbortController();
    this.stopping = false;
    this.ticker = null;
    this.lastSave = 0;
    this.rrCursor = 0;
  }

  // ---- routes -------------------------------------------------------------

  healthyRoutes() {
    const all = this.ctx.routes().filter((r) => r.enabled !== false && r.health !== 'off');
    if (this.routeMode === 'direct') return all.filter((r) => r.type === 'direct');
    if (this.routeMode !== 'spread') {
      const one = all.find((r) => r.id === this.routeMode);
      return one ? [one] : all.filter((r) => r.type === 'direct');
    }
    const up = all.filter((r) => r.health !== 'degraded');
    return up.length ? up : all;
  }

  routeById(id) { return this.ctx.routes().find((r) => r.id === id) || null; }

  /** Round-robin across healthy routes, preferring one that is not `avoidId`. */
  pickRoute(avoidId = null) {
    const pool = this.healthyRoutes();
    if (!pool.length) return null;
    const usable = pool.length > 1 && avoidId ? pool.filter((r) => r.id !== avoidId) : pool;
    const chosen = usable[this.rrCursor % usable.length];
    this.rrCursor += 1;
    return chosen.id;
  }

  noteRouteError(routeId, err) {
    this.retries += 1;
    if (routeId) this.ctx.onRouteError && this.ctx.onRouteError(routeId, err);
  }

  // ---- lifecycle ----------------------------------------------------------

  async start() {
    if (this.state === S.ACTIVE || this.state === S.PROBING) return;
    this.stopping = false;
    this.abort = new AbortController();
    this.error = null;
    this.state = S.PROBING;
    this.emitChange();

    try {
      const route = this.routeById(this.pickRoute());
      const info = await probe(this.url, { route, signal: this.abort.signal });

      this.url = info.finalUrl;
      this.size = info.size;
      this.rangeable = info.acceptsRanges && !!info.size;
      this.etag = info.etag;
      this.lastModified = info.lastModified;
      this.name = this.name || info.filename;
      this.filePath = await this.reserveName(path.join(this.dir, sanitize(this.name)));
      this.name = path.basename(this.filePath);

      await fsp.mkdir(this.dir, { recursive: true });
      const restored = await this.restore();
      if (!restored) this.plan();

      this.fd = await this.openFile();
      this.startedAt = this.startedAt || Date.now();
      this.state = S.ACTIVE;
      this.emitChange();
      this.startTicker();
      this.runLanes();
    } catch (err) {
      const e = classify(err);
      this.state = S.FAILED;
      this.error = e.message;
      this.emitChange();
    }
  }

  /** Split the file into `wantConnections` contiguous ranges. */
  plan() {
    const n = this.rangeable ? Math.min(this.wantConnections, Math.max(1, Math.floor(this.size / MIN_SPLIT))) : 1;
    this.lanes = [];
    if (!this.rangeable) {
      const lane = new Lane(1, 0, this.size ? this.size - 1 : Infinity, this);
      lane.routeId = this.pickRoute();
      this.lanes.push(lane);
      return;
    }
    const chunk = Math.floor(this.size / n);
    for (let i = 0; i < n; i++) {
      const start = i * chunk;
      const end = i === n - 1 ? this.size - 1 : start + chunk - 1;
      const lane = new Lane(i + 1, start, end, this);
      lane.routeId = this.pickRoute();
      this.lanes.push(lane);
    }
  }

  runLanes() {
    for (const lane of this.lanes) {
      if (lane.state === 'done' || lane.remaining <= 0) { lane.state = 'done'; continue; }
      this.spawn(lane);
    }
    if (this.lanes.every((l) => l.state === 'done')) this.complete();
  }

  spawn(lane) {
    lane.run().catch((err) => {
      if (this.stopping) return;
      this.state = S.FAILED;
      this.error = `Connection ${lane.id}: ${classify(err).message}`;
      this.stopLanes();
      this.emitChange();
    });
  }

  /**
   * A finished lane takes the back half of whatever range is furthest from done.
   * This is what keeps one slow route from setting the finish time, and what the
   * "Rebalance connections" action triggers by hand.
   */
  onLaneDone(free) {
    if (this.stopping || !this.rangeable) return this.maybeComplete();
    if (!this.steal(free)) this.maybeComplete();
  }

  /**
   * How long this lane still needs. Bytes left is the wrong measure once rates
   * differ: a stalled lane holding 2 MB hurts more than a fast one holding 50.
   */
  laneEta(lane) {
    if (lane.remaining <= 0) return -1;
    if (lane.state === 'stalled' || lane.state === 'retrying') return Infinity;
    const active = this.lanes.filter((l) => l.state === 'active').length || 1;
    const rate = lane.rate || this.rate / active || 1;
    return lane.remaining / rate;
  }

  /**
   * Smallest tail worth handing to another connection. Splitting costs one fresh
   * ranged request; it pays off when it saves more time than that setup costs,
   * which depends on how fast the lane is actually moving — not on a fixed byte
   * count. A stalled lane (rate 0) falls to the floor, so its work is taken over.
   */
  splitFloor(victim) {
    const active = this.lanes.filter((l) => l.state === 'active').length || 1;
    const rate = victim.state === 'stalled' || victim.state === 'retrying'
      ? 0
      : victim.rate || this.rate / active || 0;
    return Math.max(SPLIT_FLOOR, Math.round(rate * (SPLIT_SETUP_MS / 1000) * 2));
  }

  steal(free) {
    let victim = null;
    let worst = -1;
    for (const lane of this.lanes) {
      if (lane === free) continue;
      if (lane.state === 'done' || lane.state === 'failed') continue;
      const eta = this.laneEta(lane);
      if (eta > worst || (eta === worst && victim && lane.remaining > victim.remaining)) {
        worst = eta;
        victim = lane;
      }
    }
    if (!victim) return false;

    // Hand over the back half, provided both halves are worth a connection.
    const half = Math.floor(victim.remaining / 2);
    if (half < this.splitFloor(victim)) {
      // Too small to split. If that lane is wedged its connection is not coming
      // back, and waiting out its timeout would hold up a finished download, so
      // take the whole remainder instead and retire it.
      const stuck = victim.state === 'stalled' || victim.state === 'retrying';
      if (!stuck || victim.remaining <= 0) return false;
      return this.takeOver(free, victim);
    }
    const splitAt = victim.end - half;
    if (splitAt <= victim.pos) return false;

    free.start = splitAt + 1;
    free.end = victim.end;
    free.received = 0;
    free.attempts = 0;
    free.error = null;
    free.state = 'idle';
    free.routeId = this.pickRoute(victim.routeId);
    victim.end = splitAt;           // the victim's live pump sees this immediately
    this.rebalances += 1;

    this.spawn(free);
    return true;
  }

  /** Move a wedged lane's entire remainder to a fresh connection on another route. */
  takeOver(free, victim) {
    free.start = victim.pos;
    free.end = victim.end;
    free.received = 0;
    free.attempts = 0;
    free.error = null;
    free.state = 'idle';
    free.routeId = this.pickRoute(victim.routeId);

    victim.end = victim.pos - 1;   // nothing left for it; its pump settles immediately
    victim.stop();
    this.rebalances += 1;

    this.spawn(free);
    return true;
  }

  /**
   * Put finished connections back to work. Running this every tick is what stops
   * throughput collapsing at the tail: without it a lane that had nothing worth
   * stealing the moment it finished would stay idle for the rest of the download,
   * and a lane that stalls after the others are done would have nobody left to
   * take its work.
   */
  topUp() {
    if (this.stopping || !this.rangeable || this.state !== S.ACTIVE || this.fd == null) return;
    for (const lane of this.lanes) {
      if (lane.state !== 'done' || lane.remaining > 0) continue;
      if (!this.steal(lane)) break;   // nothing splittable now, so not for the next one either
    }
  }

  rebalance() {
    // Re-point every stalled or failed lane at a different route, and restart it.
    for (const lane of this.lanes) {
      if (lane.state === 'stalled' || lane.state === 'retrying') {
        lane.routeId = this.pickRoute(lane.routeId);
        lane.stop();
      }
    }
    this.emitChange();
  }

  maybeComplete() {
    if (this.lanes.every((l) => l.state === 'done')) this.complete();
  }

  async complete() {
    if (this.state !== S.ACTIVE) return;
    this.stopTicker();
    await this.closeFile();

    if (this.verifyAfter) {
      this.state = S.VERIFYING;
      this.verify = { state: 'running', progress: 0, hash: null, match: null };
      this.emitChange();
      try {
        const hash = await this.sha256();
        this.verify.hash = hash;
        this.verify.state = 'done';
        this.verify.match = this.expectedChecksum ? hash === this.expectedChecksum.toLowerCase() : null;
        if (this.verify.match === false) {
          this.state = S.FAILED;
          this.error = 'Checksum did not match the expected value';
          this.emitChange();
          return;
        }
      } catch (err) {
        this.verify.state = 'failed';
        this.verify.error = classify(err).message;
      }
    }

    await this.clearSidecar();
    this.state = S.DONE;
    this.finishedAt = Date.now();
    this.rate = 0;
    this.emitChange();
    this.emit('done', this);
  }

  sha256() {
    return new Promise((resolve, reject) => {
      const hash = crypto.createHash('sha256');
      const stream = fs.createReadStream(this.filePath, { highWaterMark: 1 << 20 });
      let read = 0;
      stream.on('data', (chunk) => {
        hash.update(chunk);
        read += chunk.length;
        if (this.size) {
          this.verify.progress = (read / this.size) * 100;
          this.emitChange();
        }
      });
      stream.on('end', () => resolve(hash.digest('hex')));
      stream.on('error', reject);
    });
  }

  async pause() {
    if (this.state !== S.ACTIVE && this.state !== S.PROBING) return;
    this.stopping = true;
    this.abort.abort();
    this.stopLanes();
    this.stopTicker();
    await this.save();
    await this.closeFile();
    this.state = S.PAUSED;
    this.rate = 0;
    for (const lane of this.lanes) if (lane.state !== 'done') lane.state = 'paused';
    this.emitChange();
  }

  async resume() {
    if (this.state !== S.PAUSED && this.state !== S.FAILED) return;
    for (const lane of this.lanes) { if (lane.state !== 'done') { lane.state = 'idle'; lane.attempts = 0; } }
    if (!this.lanes.length) return this.start();
    this.stopping = false;
    this.abort = new AbortController();
    this.error = null;
    this.fd = await this.openFile();
    this.state = S.ACTIVE;
    this.emitChange();
    this.startTicker();
    this.runLanes();
  }

  async cancel({ deleteFile = true } = {}) {
    this.stopping = true;
    this.abort.abort();
    this.stopLanes();
    this.stopTicker();
    await this.closeFile();
    this.state = S.CANCELLED;
    if (deleteFile && this.filePath) {
      await fsp.rm(this.filePath, { force: true }).catch(() => {});
      await this.clearSidecar();
    }
    this.emitChange();
  }

  stopLanes() { for (const lane of this.lanes) lane.stop(); }

  // ---- file ---------------------------------------------------------------

  async openFile() {
    if (this.fd != null) return this.fd;
    let handle;
    try {
      handle = await fsp.open(this.filePath, 'r+');
    } catch {
      handle = await fsp.open(this.filePath, 'w+');
    }
    if (this.size) await handle.truncate(this.size);   // preallocate so offsets are valid
    this.handle = handle;
    return handle.fd;
  }

  async closeFile() {
    if (!this.handle) return;
    try { await this.handle.close(); } catch { /* already closed */ }
    this.handle = null;
    this.fd = null;
  }

  /** Never silently overwrite: file.iso -> file (2).iso */
  async reserveName(target) {
    const dir = path.dirname(target);
    const ext = path.extname(target);
    const base = path.basename(target, ext);
    let candidate = target;
    for (let i = 2; i < 500; i++) {
      const sidecar = `${candidate}.fdpart`;
      const hasPart = await fsp.access(sidecar).then(() => true, () => false);
      const hasFile = await fsp.access(candidate).then(() => true, () => false);
      if (!hasFile || hasPart) return candidate;       // resumable, or free
      candidate = path.join(dir, `${base} (${i})${ext}`);
    }
    return candidate;
  }

  get sidecarPath() { return `${this.filePath}.fdpart`; }

  async save() {
    if (!this.filePath || !this.rangeable) return;
    this.lastSave = Date.now();
    const body = {
      v: 1,
      url: this.originUrl,
      size: this.size,
      etag: this.etag,
      lastModified: this.lastModified,
      lanes: this.lanes.map((l) => ({ id: l.id, start: l.start, end: l.end, received: l.received })),
    };
    await fsp.writeFile(this.sidecarPath, JSON.stringify(body)).catch(() => {});
  }

  async restore() {
    try {
      const body = JSON.parse(await fsp.readFile(this.sidecarPath, 'utf8'));
      if (body.size !== this.size) return false;
      if (body.etag && this.etag && body.etag !== this.etag) return false;  // file changed under us
      this.lanes = body.lanes.map((l) => {
        const lane = new Lane(l.id, l.start, l.end, this);
        lane.received = l.received;
        lane.routeId = this.pickRoute();
        if (lane.remaining <= 0) lane.state = 'done';
        return lane;
      });
      this.received = this.lanes.reduce((sum, l) => sum + l.received, 0);
      return this.lanes.length > 0;
    } catch {
      return false;
    }
  }

  async clearSidecar() {
    if (this.filePath) await fsp.rm(this.sidecarPath, { force: true }).catch(() => {});
  }

  // ---- ticking ------------------------------------------------------------

  startTicker() {
    if (this.ticker) return;
    this.ticker = setInterval(() => this.tick(), TICK_MS);
    if (this.ticker.unref) this.ticker.unref();
  }

  stopTicker() {
    if (this.ticker) clearInterval(this.ticker);
    this.ticker = null;
  }

  tick() {
    const now = Date.now();
    let total = 0;

    for (const lane of this.lanes) {
      lane.samples.push(lane.tickBytes);
      if (lane.samples.length > RATE_SAMPLES) lane.samples.shift();
      lane.tickBytes = 0;
      const sum = lane.samples.reduce((a, b) => a + b, 0);
      lane.rate = sum / (lane.samples.length * (TICK_MS / 1000));
      total += lane.rate;

      if (lane.state === 'active' && lane.lastByteAt) {
        const idle = now - lane.lastByteAt;
        if (idle > ABORT_MS) lane.stop();              // run() will retry it
        else if (idle > STALL_MS) lane.state = 'stalled';
      } else if (lane.state === 'stalled' && now - lane.lastByteAt < STALL_MS) {
        lane.state = 'active';
      }
    }

    this.rate = total;
    this.topUp();
    this.history.push(total);
    if (this.history.length > HISTORY * (1000 / TICK_MS)) this.history.shift();

    if (now - this.lastSave > SAVE_MS) this.save();
    this.emitChange();
  }

  emitChange() { this.ctx.onChange && this.ctx.onChange(this); }

  // ---- view model ---------------------------------------------------------

  snapshot() {
    const lanes = this.lanes.map((l) => l.snapshot()).sort((a, b) => a.start - b.start);
    const active = lanes.filter((l) => l.state !== 'done' && l.pct != null);
    const slowest = active.length
      ? active.reduce((a, b) => (a.pct <= b.pct ? a : b))
      : null;
    const pcts = active.map((l) => l.pct);
    const spread = pcts.length > 1 ? (Math.max(...pcts) - Math.min(...pcts)) / 2 : 0;
    // Overlap from a rebalance can push received past size; never report negative work left.
    const remaining = this.size ? Math.max(0, this.size - this.received) : null;

    // Downsample the 250ms history to one point per second for the graph.
    const graph = [];
    for (let i = 0; i < this.history.length; i += 4) {
      const slice = this.history.slice(i, i + 4);
      graph.push(slice.reduce((a, b) => a + b, 0) / slice.length);
    }

    return {
      id: this.id,
      name: this.name,
      url: this.originUrl,
      finalUrl: this.url,
      filePath: this.filePath,
      state: this.state,
      size: this.size,
      received: this.received,
      pct: this.size ? Math.min(100, (this.received / this.size) * 100) : null,
      rate: this.rate,
      eta: this.rate > 0 && remaining != null ? remaining / this.rate : null,
      rangeable: this.rangeable,
      etag: this.etag,
      connections: this.lanes.length,
      routeMode: this.routeMode,
      routesInUse: [...new Set(lanes.map((l) => l.routeId).filter(Boolean))],
      error: this.error,
      retries: this.retries,
      rebalances: this.rebalances,
      verify: this.verify,
      addedAt: this.addedAt,
      startedAt: this.startedAt,
      finishedAt: this.finishedAt,
      lanes,
      slowest: slowest ? { id: slowest.id, state: slowest.state, stalledMs: slowest.stalledMs } : null,
      spread,
      graph,
    };
  }
}

module.exports = { Download, Lane, S, MIN_SPLIT };