// Many streams run at once, each from its own pooled engine instance (src/core.js's
// per-slot pool, which keeps released instances until a timer trims it). Interleaves
// dozens of streams one chunk at a time (string and byte chunks, every chunk size 1-23),
// opens and closes streams mid-flight so instances get released and reused, and grows
// every memory region (input, output, frame stack, errors) while other
// streams hold instances of their own. Each stream's output and errors must equal its own
// whole-string result, and a repeated burst must not keep instantiating more engines than
// the pool already settled on. Also proves a reused instance actually comes back clean:
// streams that finish at staggered times hand their instance to a newcomer while siblings
// are still open, and a stream abandoned mid-namespace/class/expression (frame stack,
// anchors, and errors all non-empty) must leave its instance fit for the very next acquire.
// Usage: node test/concurrent-streams.mjs
import assert from "node:assert/strict";
import fs from "node:fs";

const { transformSync, engine } = await import("../src/index.js");

const lines = (f) => fs.readFileSync(f, "utf8").split("\n").filter(Boolean).map((l) => JSON.parse(l));
const small = [
  ...lines("corpus/adversarial.jsonl"),
  ...lines("corpus/regressions.jsonl"),
  ...lines("corpus/downlevel.jsonl").map((c) => c.src),
].filter((s) => s.length < 3000);
const dom = fs.readFileSync("node_modules/typescript/lib/lib.dom.d.ts", "utf8");
// Inputs that grow one region each: frames (deep nesting), errors (many non-erasable constructs),
// input retained under an anchor (a long namespace), and plain input/output volume.
const big = [
  "let a = " + "(".repeat(6000) + "1" + ")".repeat(6000) + ";\n",
  "enum E { A }\nlet x: number = 1;\n".repeat(3000),
  "namespace N {\n" + "  export const x: number = 1;\n".repeat(8000) + "}\n",
  dom.slice(0, 400000),
];

function whole(mode, src) {
  const errors = [];
  try {
    return { out: transformSync(src, { mode, onError: (e) => errors.push(e.start, e.end) }), errors };
  } catch (e) {
    return { threw: e.message };
  }
}

const enc = new TextEncoder();
// A stream over src in chunks of `size` characters, sent as UTF-8 bytes when `bytes` (except for a leading BOM,
// which the byte path drops by design).
function stream(mode, src, size, bytes) {
  if (src.startsWith("")) bytes = false;
  const parts = [];
  for (let i = 0; i < src.length; i += size) parts.push(bytes ? enc.encode(src.slice(i, i + size)) : src.slice(i, i + size));
  const errors = [];
  const gen = transformSync(parts, { mode, onError: (e) => errors.push(e.start, e.end) });
  return { gen, out: "", errors, src, threw: undefined, done: false };
}

// Advances every live stream by one chunk, round-robin, until all finish. `spawn` may add streams
// mid-flight; `cut` may abandon one early (its generator's return() closes its instance).
function interleave(live, spawn) {
  let round = 0;
  while (live.some((s) => !s.done)) {
    for (const s of live) {
      if (s.done) continue;
      try {
        const r = s.gen.next();
        if (r.done) s.done = true;
        else s.out += r.value;
      } catch (e) {
        s.threw = e.message;
        s.done = true;
      }
    }
    spawn?.(live, round++);
  }
}

function check(mode, live, label) {
  for (const s of live) {
    if (s.cut) continue;
    const want = whole(mode, s.src);
    if (want.threw) {
      assert.ok(s.threw, `${label}: stream should throw like the whole-string call`);
      continue;
    }
    assert.equal(s.threw, undefined, `${label}: stream threw ${s.threw}`);
    assert.equal(s.out, want.out, `${label}: interleaved output differs from the whole-string result`);
    assert.deepEqual(s.errors, want.errors, `${label}: interleaved errors differ from the whole-string result`);
  }
}

// Counts fresh WebAssembly.Instance creations, so the burst section below can check that
// the pool settles instead of re-instantiating on every repeat.
let instances = 0;
const OrigInstance = WebAssembly.Instance;
WebAssembly.Instance = class extends OrigInstance {
  constructor(...args) {
    super(...args);
    instances++;
  }
};

for (const mode of ["erase", "downlevel"]) {
  engine(mode);

  // 1. Every small input at once, each with its own chunk size (1-23) and chunk kind
  // (string or bytes), all interleaved concurrently: each one must hold its own pooled
  // instance rather than clobbering another stream's parse state.
  let live = small.map((src, i) => stream(mode, src, 1 + (i % 23), i % 2 === 1));
  interleave(live);
  check(mode, live, `${mode} small x${live.length}`);

  // 2. Big inputs that grow the frame stack, error list, and input/output regions,
  // interleaved with small ones; streams join and leave mid-flight (including one cut off
  // early via the generator's return(), which must release its instance rather than leak it).
  live = big.map((src, i) => stream(mode, src, 4096 + i * 1000, i % 2 === 1));
  let next = 0;
  interleave(live, (all, round) => {
    if (round % 3 === 0 && next < small.length) all.push(stream(mode, small[next++], 7, false));
    if (round % 5 === 4) {
      const victim = all.find((s) => !s.done && !s.cut && s.src.length < 3000);
      if (victim) {
        victim.gen.return();
        victim.cut = victim.done = true;
      }
    }
  });
  check(mode, live, `${mode} big+churn x${live.length}`);

  // 3. A burst of concurrent streams past the pool's idle minimum must settle: once the pool
  // has kept what a burst released, a repeated, identically-shaped burst must not keep
  // instantiating more and more engines. (A pool that trims could still
  // instantiate some per burst — that's the pool working as designed, not a leak; what must
  // not happen is the count climbing burst over burst.)
  const burst = () => {
    const l = big.concat(small.slice(0, 60)).map((src, i) => stream(mode, src, 5000, i % 2 === 1));
    interleave(l);
    check(mode, l, `${mode} burst`);
  };
  burst();
  const afterFirst = instances;
  burst();
  const createdSecond = instances - afterFirst;
  burst();
  const createdThird = instances - afterFirst - createdSecond;
  assert.equal(createdThird, createdSecond, `${mode}: a repeated burst must settle, not keep instantiating more instances each time`);

  // 4. Dirty reuse: an instance release() takes back must come back genuinely clean to
  // whoever acquire()s it next, even while sibling streams are still open. Two ways an
  // instance gets released dirty:
  //
  // (a) Staggered reuse: streams with small, varied chunk sizes finish at staggered times
  // while others remain mid-flight; a new stream opened right after one finishes is very
  // likely handed the instance that sibling just released, with others still live around it.
  {
    const group = small.slice(0, 50).map((src, i) => stream(mode, src, 1 + (i % 5), i % 3 === 0));
    let joined = 50;
    interleave(group, (all) => {
      // Every time at least one sibling has finished this round, throw a fresh stream at the
      // pool immediately, so its acquire() lands on a just-freed instance while others are
      // still open around it.
      if (joined < small.length && all.some((s) => s.done)) {
        group.push(stream(mode, small[joined], 1 + (joined % 5), joined % 3 === 1));
        joined++;
      }
    });
    check(mode, group, `${mode} staggered reuse`);
  }

  // (b) Abandon mid-construct: a stream cut off with `.return()` deep inside a namespace, a
  // class, and an ambiguous expression leaves its frame stack, anchors, and reported errors
  // all non-empty when release() takes the instance back. `held` streams stay open
  // throughout, so the dirtied instance is the *only* thing released in between, and the
  // pool hands the most recently released instance to the very next acquire().
  {
    const held = Array.from({ length: 6 }, () => stream(mode, small[0], 4096, false));
    for (const s of held) {
      const r = s.gen.next();
      if (r.done) s.done = true;
      else s.out += r.value;
    }

    const dirty =
      "namespace Deep {\n  export class C {\n    static v = " +
      "(".repeat(200) +
      "1" +
      ")".repeat(80) +
      ";\n    static w = a ?? b as any as unknown && 2;\n";
    const d = stream(mode, dirty, 11, false);
    for (let i = 0; i < 10 && !d.done; i++) {
      const r = d.gen.next();
      if (r.done) d.done = true;
    }
    assert.equal(d.done, false, `${mode}: the dirty stream should still be mid-construct when abandoned`);
    d.gen.return(); // abandoned inside the namespace/class/expression; the instance releases dirty

    const fresh = stream(mode, small[1], 9, false);
    interleave([fresh]);
    check(mode, [fresh], `${mode} dirty-reuse`);

    for (const s of held) interleave([s]);
    check(mode, held, `${mode} dirty-reuse holders`);
  }
}
WebAssembly.Instance = OrigInstance;
console.log(`concurrent-streams: ok (${small.length} small + ${big.length} large streams interleaved per mode across the pool; bursts settle)`);