// Exercises the island's WHATWG-streams subset differentially: under Node
// these are the REAL streams, so every line pins the island implementation
// against the spec behavior the AI-SDK/eventsource-parser paths rely on —
// reader protocol, pull timing, async iteration, pipeThrough chains,
// TransformStream subclassing (the eventsource-parser shape), flush
// ordering, error propagation both directions, and cancel-on-break. The
// async work is left PENDING when run() returns: the island's job queue
// drains at loop quiescence, like Node's microtasks.
const out = (line) => process.stdout.write(line + "\n");

export function run() {
  (async () => {
    // push source + async iteration
    {
      const rs = new ReadableStream({
        start(c) {
          c.enqueue("a");
          c.enqueue("b");
          c.close();
        },
      });
      const got = [];
      for await (const v of rs) got.push(v);
      out("push-iter " + got.join(","));
    }
    // pull source: pull count observes the spec's proactive HWM-1 pulling
    {
      let n = 0;
      const rs = new ReadableStream({
        pull(c) {
          n++;
          if (n <= 3) c.enqueue("p" + n);
          else c.close();
        },
      });
      const got = [];
      for await (const v of rs) got.push(v);
      out("pull-iter " + got.join(",") + " pulls=" + n);
    }
    // reader protocol + locking
    {
      const rs = new ReadableStream({
        start(c) {
          c.enqueue(1);
          c.close();
        },
      });
      const r = rs.getReader();
      const a = await r.read();
      const b = await r.read();
      out(`reader ${a.value} ${a.done} ${b.value} ${b.done} locked=${rs.locked}`);
      r.releaseLock();
      out("released " + rs.locked);
    }
    // enqueue AFTER a read is already waiting
    {
      let ctrl;
      const rs = new ReadableStream({ start(c) { ctrl = c; } });
      const r = rs.getReader();
      const p = r.read();
      ctrl.enqueue("late");
      const a = await p;
      ctrl.close();
      const b = await r.read();
      await r.closed;
      out(`deferred ${a.value} ${a.done} ${b.done} closed-resolved`);
    }
    // controller.error rejects waiting and future reads
    {
      let ctrl;
      const rs = new ReadableStream({ start(c) { ctrl = c; } });
      const r = rs.getReader();
      const p = r.read();
      ctrl.error(new Error("boom"));
      const m1 = await p.then(() => "no-throw", (e) => e.message);
      const m2 = await r.read().then(() => "no-throw", (e) => e.message);
      out(`errored ${m1} ${m2}`);
    }
    // pipeThrough + TransformStream SUBCLASS (the eventsource-parser shape)
    {
      class Upper extends TransformStream {
        constructor() {
          super({
            transform(chunk, c) {
              c.enqueue(String(chunk).toUpperCase());
            },
          });
        }
      }
      const src = new ReadableStream({
        start(c) {
          c.enqueue("x");
          c.enqueue("y");
          c.close();
        },
      });
      const got = [];
      for await (const v of src.pipeThrough(new Upper())) got.push(v);
      out("pipe-subclass " + got.join(","));
    }
    // start + transform + flush ordering
    {
      const t = new TransformStream({
        start(c) { c.enqueue("S"); },
        transform(chunk, c) { c.enqueue(chunk + "!"); },
        flush(c) { c.enqueue("F"); },
      });
      const src = new ReadableStream({
        start(c) {
          c.enqueue("m");
          c.close();
        },
      });
      const got = [];
      for await (const v of src.pipeThrough(t)) got.push(v);
      out("flush " + got.join(","));
    }
    // chained pipeThrough
    {
      const dup = () =>
        new TransformStream({
          transform(chunk, c) {
            c.enqueue(chunk);
            c.enqueue(chunk);
          },
        });
      const src = new ReadableStream({
        start(c) {
          c.enqueue("z");
          c.close();
        },
      });
      const got = [];
      for await (const v of src.pipeThrough(dup()).pipeThrough(dup())) got.push(v);
      out("chain " + got.join(","));
    }
    // a throwing transform errors the consumer
    {
      const t = new TransformStream({
        transform() {
          throw new Error("t-boom");
        },
      });
      const src = new ReadableStream({
        start(c) {
          c.enqueue("q");
          c.close();
        },
      });
      let msg = "no-throw";
      try {
        for await (const v of src.pipeThrough(t)) out("unreachable " + v);
      } catch (e) {
        msg = e.message;
      }
      out("transform-error " + msg);
    }
    // a source error propagates THROUGH a transform to the consumer
    {
      let ctrl;
      const t = new TransformStream({
        transform(chunk, c) {
          c.enqueue(chunk);
        },
      });
      const src = new ReadableStream({ start(c) { ctrl = c; } });
      const it = src.pipeThrough(t)[Symbol.asyncIterator]();
      ctrl.enqueue("one");
      const a = await it.next();
      ctrl.error(new Error("src-boom"));
      const m = await it.next().then(() => "no-throw", (e) => e.message);
      out(`source-error ${a.value} ${m}`);
    }
    // breaking out of for-await cancels the source and releases the lock
    {
      let cancelled = "no";
      const rs = new ReadableStream({
        start(c) {
          c.enqueue(1);
          c.enqueue(2);
          c.enqueue(3);
        },
        cancel() {
          cancelled = "yes";
        },
      });
      for await (const v of rs) {
        if (v === 1) break;
      }
      out(`break-cancel ${cancelled} locked=${rs.locked}`);
    }
    // writing a TransformStream's writable directly (writer protocol)
    {
      const t = new TransformStream();
      const w = t.writable.getWriter();
      const r = t.readable.getReader();
      const wp = w.write("idw");
      const a = await r.read();
      await wp;
      const cp = w.close();
      const b = await r.read();
      await cp;
      out(`identity-writer ${a.value} ${b.done}`);
    }
    // desiredSize under the default HWM-1 strategy
    {
      let ctrl;
      const rs = new ReadableStream({ start(c) { ctrl = c; } });
      const d0 = ctrl.desiredSize;
      ctrl.enqueue("a");
      const d1 = ctrl.desiredSize;
      const rd = rs.getReader();
      await rd.read();
      const d2 = ctrl.desiredSize;
      ctrl.close();
      out(`desired ${d0} ${d1} ${d2}`);
    }
    // stream.cancel(): source sees the reason, readers see done
    {
      let reason = "";
      const rs = new ReadableStream({
        start(c) {
          c.enqueue("x");
        },
        cancel(r) {
          reason = String(r);
        },
      });
      await rs.cancel("bye");
      const a = await rs.getReader().read();
      out(`cancel ${reason} ${a.done}`);
    }
    // misuse throws TypeErrors: enqueue-after-close, double getReader
    {
      let enq = "no-throw";
      const rs = new ReadableStream({
        start(c) {
          c.close();
          try {
            c.enqueue("x");
          } catch (e) {
            enq = String(e instanceof TypeError);
          }
        },
      });
      rs.getReader();
      let lock = "no-throw";
      try {
        rs.getReader();
      } catch (e) {
        lock = String(e instanceof TypeError);
      }
      out(`misuse enqueue-after-close=${enq} double-reader=${lock}`);
    }
    out("done");
  })();
}