use std::fs;
use std::io::Read;
use std::path::Path;
use std::process::{Command, Output, Stdio};
use std::time::{Duration, Instant};

const DEADLINE: Duration = Duration::from_secs(30);

fn run(dir: &Path, script: &str, source: &str, flags: &[&str]) -> Output {
    fs::write(dir.join(script), source).unwrap();
    let mut child = Command::new(env!("CARGO_BIN_EXE_avril"))
        .arg("run")
        .current_dir(dir)
        .args(flags)
        .arg(script)
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()
        .unwrap();
    let mut stdout = child.stdout.take().unwrap();
    let mut stderr = child.stderr.take().unwrap();
    let threads = std::thread::scope(|scope| {
        let out = scope.spawn(|| {
            let mut bytes = Vec::new();
            stdout.read_to_end(&mut bytes).unwrap();
            bytes
        });
        let err = scope.spawn(|| {
            let mut bytes = Vec::new();
            stderr.read_to_end(&mut bytes).unwrap();
            bytes
        });
        let started = Instant::now();
        let status = loop {
            if let Some(status) = child.try_wait().unwrap() {
                break status;
            }
            if started.elapsed() >= DEADLINE {
                child.kill().unwrap();
                panic!("avril run did not exit within {DEADLINE:?}");
            }
            std::thread::sleep(Duration::from_millis(10));
        };
        (out.join().unwrap(), err.join().unwrap(), status)
    });
    Output {
        status: threads.2,
        stdout: threads.0,
        stderr: threads.1,
    }
}

fn assert_marker(output: &Output, marker: &str) {
    assert!(
        output.status.success(),
        "exit={}\nstdout={}\nstderr={}",
        output.status,
        String::from_utf8_lossy(&output.stdout),
        String::from_utf8_lossy(&output.stderr)
    );
    assert!(
        String::from_utf8_lossy(&output.stdout).contains(marker),
        "missing {marker}\nstdout={}\nstderr={}",
        String::from_utf8_lossy(&output.stdout),
        String::from_utf8_lossy(&output.stderr)
    );
}

const PRELUDE: &str = r#"
const failures = [];
function check(cond, message) {
  if (!cond) failures[failures.length] = message;
}
function done(name) {
  if (failures.length !== 0) {
    for (let i = 0; i < failures.length; i++) console.error("FAIL: " + failures[i]);
    throw new Error(failures.length + " failure(s) in " + name);
  }
  console.log("KV-OK " + name);
}
"#;

#[test]
fn roundtrip_and_persistence() {
    let dir = tempfile::tempdir().unwrap();
    let write = format!(
        r#"{PRELUDE}
import {{ openKv }} from "avril:kv";
const db = await openKv("state.kv", {{ create: true }});
const ks = db.keyspace("v");
// Strings and equal byte sequences are the same key and value.
await ks.set("str", "hello");
await ks.set(new Uint8Array([98, 121, 116, 101, 115]), new Uint8Array([1, 2, 3]));
await ks.set("abc", "as-string");
await ks.set(new Uint8Array([120]), "json-value");
const first = await ks.set("rewrite", "one");
const second = await ks.set("rewrite", "two");
check(first.versionstamp !== second.versionstamp, "versionstamp changes on rewrite");
check((await db.keyspaces()).join(",") === "v", "keyspaces lists v");
db.close();
const reopened = await openKv("state.kv", {{ create: false }});
const again = reopened.keyspace("v");
const entry = await again.entry("str");
check(entry !== null && entry.keyspace === "v", "entry keyspace");
check(entry !== null && Array.from(entry.key).join(",") === "115,116,114", "entry key is canonical bytes");
check(entry !== null && typeof entry.versionstamp === "bigint", "entry carries a versionstamp");
const str = await again.get("str");
check(str !== null && Array.from(str).join(",") === "104,101,108,108,111", "get returns raw bytes");
const text = await again.getText("str");
check(text === "hello", "getText decodes strings");
const asBytes = await again.get(new Uint8Array([115, 116, 114]));
check(asBytes !== null && asBytes.length === 5, "byte key hits string key");
const overwritten = await again.get("abc");
check(overwritten !== null && overwritten instanceof Uint8Array, "overwritten value is bytes");
await again.set("abc", "as-string-again");
const back = await again.get(new Uint8Array([97, 98, 99]));
check(back !== null && (await again.getText(new Uint8Array([97, 98, 99]))) === "as-string-again", "bytes and strings are one key");
const json = await again.getText("json-holder");
check(json === null, "missing key reads null");
await again.set("json-holder", JSON.stringify({{ list: [1, "two"], map: {{ a: 1 }} }}));
const parsed = JSON.parse((await again.getText("json-holder")) ?? "null");
check(parsed.list[1] === "two" && parsed.map.a === 1, "JSON roundtrips through strings");
// Non-UTF-8 bytes: get works byte-exact, getText rejects.
await again.set("raw", new Uint8Array([0xff, 0x00, 0xfe]));
const raw = await again.get("raw");
check(raw !== null && Array.from(raw).join(",") === "255,0,254", "non-UTF8 roundtrips as bytes");
try {{
  await again.getText("raw");
  check(false, "getText of non-UTF8 must reject");
}} catch (e) {{
  check(e instanceof TypeError, "getText rejection is a TypeError, got: " + e);
}}
done("roundtrip");
"#
    );
    assert_marker(&run(dir.path(), "write.js", &write, &[]), "KV-OK roundtrip");
    let read = r#"import { openKv } from "avril:kv";
const failures = [];
function check(cond, message) {
  if (!cond) failures[failures.length] = message;
}
const db = await openKv("state.kv", { create: false });
const ks = db.keyspace("v");
const str = await ks.getText("str");
check(str === "hello", "string persists across processes");
const bytes = await ks.get("bytes");
check(bytes !== null && Array.from(bytes).join(",") === "1,2,3", "bytes persist across processes");
const missing = await ks.get("nope");
check(missing === null, "missing reads null");
if (failures.length !== 0) {
  for (let i = 0; i < failures.length; i++) console.error("FAIL: " + failures[i]);
  throw new Error(failures.length + " failure(s)");
}
console.log("KV-OK reread");
"#;
    assert_marker(&run(dir.path(), "read.js", read, &[]), "KV-OK reread");
}

#[test]
fn keys_and_ordering() {
    let dir = tempfile::tempdir().unwrap();
    let source = r#"import { openKv } from "avril:kv";
const failures = [];
function check(cond, message) {
  if (!cond) failures[failures.length] = message;
}
const db = await openKv("state.kv", { create: true });
const ks = db.keyspace("o");
// "b" as a string and as bytes [98] is one key; the second write wins.
await ks.set("b", "string-b");
await ks.set(new Uint8Array([98]), "bytes-b");
await ks.set("", "empty");
await ks.set(new Uint8Array([0]), "zero");
await ks.set("a", "a");
await ks.set("a\x00", "a-nul");
await ks.set("a\x00b", "a-nul-b");
await ks.set("aa", "aa");
await ks.set(new Uint8Array([255]), "max");
const tag = (key) => Array.from(key).join(",");
const listed = [];
for await (const e of ks.list()) listed[listed.length] = tag(e.key);
check(JSON.stringify(listed) === JSON.stringify(["", "0", "97", "97,0", "97,0,98", "97,97", "98", "255"]), "byte ordering:\n" + JSON.stringify(listed));
const b = await ks.getText("b");
check(b === "bytes-b", "string and bytes are one key");
const prefixed = [];
for await (const e of ks.list({ prefix: "user:" })) prefixed[prefixed.length] = tag(e.key);
const emptyBefore = prefixed.length;
await ks.set("user:1", "u1");
await ks.set("user:2", "u2");
await ks.set("admin:1", "a1");
const prefixedAfter = [];
for await (const e of ks.list({ prefix: "user:" })) prefixedAfter[prefixedAfter.length] = tag(e.key);
check(emptyBefore === 0, "no user: keys before insert");
check(JSON.stringify(prefixedAfter) === JSON.stringify(["117,115,101,114,58,49", "117,115,101,114,58,50"]), "string prefix listing");
const bytePrefixed = [];
for await (const e of ks.list({ prefix: new Uint8Array([0x89, 0x50]) })) bytePrefixed[bytePrefixed.length] = tag(e.key);
check(bytePrefixed.length === 0, "binary prefix with no match");
await ks.set(new Uint8Array([0x89, 0x50, 0x01]), "bin1");
await ks.set(new Uint8Array([0x89, 0x50, 0x02]), "bin2");
await ks.set(new Uint8Array([0x89, 0x51]), "other");
const bytePrefixedAfter = [];
for await (const e of ks.list({ prefix: new Uint8Array([0x89, 0x50]) })) bytePrefixedAfter[bytePrefixedAfter.length] = tag(e.key);
check(JSON.stringify(bytePrefixedAfter) === JSON.stringify(["137,80,1", "137,80,2"]), "binary prefix listing");
const bounded = [];
for await (const e of ks.list({ start: "a", end: "b" })) bounded[bounded.length] = tag(e.key);
check(JSON.stringify(bounded) === JSON.stringify(["97", "97,0", "97,0,98", "97,97", "97,100,109,105,110,58,49"]), "start/end bounds");
const limited = [];
for await (const e of ks.list({ limit: 2 })) limited[limited.length] = tag(e.key);
check(limited.length === 2, "limit");
const reversed = [];
for await (const e of ks.list({ reverse: true })) reversed[reversed.length] = tag(e.key);
const forward = [];
for await (const e of ks.list()) forward[forward.length] = tag(e.key);
check(JSON.stringify(reversed) === JSON.stringify(forward.slice().reverse()), "reverse");
const all = [];
for await (const e of ks.list({ prefix: "" })) all[all.length] = tag(e.key);
check(all.length === forward.length, "empty prefix scans all");
if (failures.length !== 0) {
  for (let i = 0; i < failures.length; i++) console.error("FAIL: " + failures[i]);
  throw new Error(failures.length + " failure(s)");
}
console.log("KV-OK ordering");
"#;
    assert_marker(&run(dir.path(), "main.js", source, &[]), "KV-OK ordering");
}

#[test]
fn keyspaces_isolation_clear_and_get_many() {
    let dir = tempfile::tempdir().unwrap();
    let source = r#"import { openKv } from "avril:kv";
const failures = [];
function check(cond, message) {
  if (!cond) failures[failures.length] = message;
}
const db = await openKv("state.kv", { create: true });
const a = db.keyspace("a");
const b = db.keyspace("b");
check(a.name === "a" && b.name === "b", "keyspace names");
check(a === db.keyspace("a"), "keyspace identity is stable");
await a.set("k", "from-a");
await b.set("k", "from-b");
const ra = await a.get("k");
const rb = await b.get("k");
check(ra !== null && Array.from(ra).join(",") === "102,114,111,109,45,97", "space a isolated");
check(rb !== null && (await b.getText("k")) === "from-b", "space b isolated");
check(JSON.stringify(await db.keyspaces()) === '["a","b"]', "keyspaces listed");
const many = await b.getMany(["k", "missing"]);
check(many.length === 2 && many[0] !== null && many[1] === null, "getMany");
await a.clear();
check(await a.get("k") === null, "clear empties its space");
const still = await b.getText("k");
check(still === "from-b", "clear scoped to its space");
if (failures.length !== 0) {
  for (let i = 0; i < failures.length; i++) console.error("FAIL: " + failures[i]);
  throw new Error(failures.length + " failure(s)");
}
console.log("KV-OK spaces");
"#;
    assert_marker(&run(dir.path(), "main.js", source, &[]), "KV-OK spaces");
}

#[test]
fn atomic_batches() {
    let dir = tempfile::tempdir().unwrap();
    let source = r#"import { openKv } from "avril:kv";
const failures = [];
function check(cond, message) {
  if (!cond) failures[failures.length] = message;
}
const db = await openKv("state.kv", { create: true });
const other = await openKv("other.kv", { create: true });
const a = db.keyspace("a");
const b = db.keyspace("b");
const stamp = (await a.set("k1", "v1")).versionstamp;
const entryBefore = await a.entry("k1");
check(entryBefore !== null && entryBefore.versionstamp === stamp, "entry() exposes the commit stamp");
const ok = await db.atomic((tx) => {
  tx.check(a, "k1", stamp);
  tx.set(a, "k1", "v2");
  tx.set(b, "k2", "w2");
  tx.delete(a, "ghost");
});
check(ok.ok === true, "commit succeeds");
const ra = await a.get("k1");
const rb = await b.get("k2");
check(ra !== null && rb !== null && (await a.entry("k1"))?.versionstamp === (await b.entry("k2"))?.versionstamp, "same stamp across keyspaces");
check(ra !== null && (await a.getText("k1")) === "v2" && (await b.getText("k2")) === "w2", "both writes visible");
const bad = await db.atomic((tx) => {
  tx.check(a, "k1", 255n);
  tx.set(a, "k1", "lost");
  tx.set(b, "k2", "lost");
});
check(bad.ok === false, "bad check aborts");
check((await a.getText("k1")) === "v2", "aborted set invisible");
check(await b.get("missing") === null, "aborted second-space write invisible");
const fresh = await db.atomic((tx) => {
  tx.check(b, "fresh", null);
  tx.set(b, "fresh", "1");
});
check(fresh.ok === true, "null check passes when absent");
const stale = await db.atomic((tx) => {
  tx.check(b, "fresh", null);
  tx.set(b, "fresh", "2");
});
check(stale.ok === false, "null check fails when present");
const only = await db.atomic((tx) => {
  tx.check(b, "definitely-absent", null);
});
check(only.ok === true && only.versionstamp === null, "check-only batch has null versionstamp");
const manyText = await b.getTextMany(["fresh", "missing", "fresh"]);
check(JSON.stringify(manyText) === JSON.stringify(["1", null, "1"]), "getTextMany preserves order and nulls");
const stamped = await a.entries(["k1", "missing"]);
check(stamped.length === 2 && stamped[0] !== null && stamped[1] === null, "entries preserves order and nulls");
const accepted = await db.atomic((tx) => {
  tx.check(a, "k1", stamped[0].versionstamp);
  tx.set(a, "k1", "v3");
});
check(accepted.ok === true, "entries() stamps are accepted by check");
check((await a.getText("k1")) === "v3", "check-then-set via entry stamp commits");
try {
  await db.atomic(() => {
    throw new Error("boom");
  });
  check(false, "throwing callback must reject");
} catch (e) {
  check(e.message === "boom", "callback error propagates");
}
check((await b.getText("fresh")) === "1", "thrown callback commits nothing");
try {
  await db.atomic(async (tx) => {
    tx.set(a, "async", "v");
  });
  check(false, "async callback must reject");
} catch (e) {
  check(e instanceof TypeError, "async callback is a TypeError");
}
check(await a.get("async") === null, "async callback commits nothing");
let stolen = null;
await db.atomic((tx) => {
  stolen = tx;
});
try {
  stolen.set(a, "late", "v");
  check(false, "retained tx must throw");
} catch (e) {
  check(e instanceof TypeError, "retained tx is a TypeError");
}
try {
  await db.atomic((tx) => {
    tx.set(other.keyspace("x"), "k", "v");
  });
  check(false, "foreign keyspace must throw");
} catch (e) {
  check(e instanceof TypeError, "foreign keyspace is a TypeError");
}
if (failures.length !== 0) {
  for (let i = 0; i < failures.length; i++) console.error("FAIL: " + failures[i]);
  throw new Error(failures.length + " failure(s)");
}
console.log("KV-OK atomic");
"#;
    assert_marker(&run(dir.path(), "main.js", source, &[]), "KV-OK atomic");
}

#[test]
fn modes_exclusive_blocks_second_process_and_read_only() {
    let dir = tempfile::tempdir().unwrap();
    fs::write(
        dir.path().join("hold.js"),
        r#"import { openKv } from "avril:kv";
import { stdin } from "avril:std";
const db = await openKv("held.kv", { access: "exclusive", create: true });
await db.keyspace("a").set("k", "held");
console.log("READY");
const reader = stdin.getReader();
while (!(await reader.read()).done) {}
"#,
    )
    .unwrap();
    let mut holder = Command::new(env!("CARGO_BIN_EXE_avril"))
        .arg("run")
        .current_dir(dir.path())
        .arg("hold.js")
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()
        .unwrap();
    let (ready_tx, ready_rx) = std::sync::mpsc::channel();
    let mut holder_stdout = holder.stdout.take().unwrap();
    std::thread::spawn(move || {
        let mut prefix = [0u8; 6];
        let _ = holder_stdout.read_exact(&mut prefix);
        let _ = ready_tx.send(prefix);
    });
    let started = Instant::now();
    loop {
        if let Ok(prefix) = ready_rx.try_recv() {
            assert_eq!(&prefix, b"READY\n", "holder handshake");
            break;
        }
        if started.elapsed() >= DEADLINE {
            holder.kill().unwrap();
            panic!("holder did not print READY");
        }
        std::thread::sleep(Duration::from_millis(10));
    }
    let attempt = run(
        dir.path(),
        "take.js",
        r#"import { openKv } from "avril:kv";
await openKv("held.kv", { access: "exclusive", create: false });
console.log("OPENED-UNEXPECTEDLY");
"#,
        &[],
    );
    assert!(!attempt.status.success(), "second exclusive open must fail");
    assert!(
        String::from_utf8_lossy(&attempt.stderr).contains("already open"),
        "already-open error, stderr={}",
        String::from_utf8_lossy(&attempt.stderr)
    );
    drop(holder.stdin.take());
    let status = loop {
        if let Some(status) = holder.try_wait().unwrap() {
            break status;
        }
        if started.elapsed() >= DEADLINE {
            panic!("holder did not exit");
        }
        std::thread::sleep(Duration::from_millis(10));
    };
    assert!(status.success(), "holder exits cleanly");
    let modes = r#"import { openKv } from "avril:kv";
const failures = [];
function check(cond, message) {
  if (!cond) failures[failures.length] = message;
}
try {
  await openKv("missing.kv", { create: false });
  check(false, "create:false on missing must reject");
} catch (e) {
  check(String(e.message).includes("does not exist"), "missing message: " + e.message);
}
const ro = await openKv("missing.kv", { access: "read-only" });
check(await ro.keyspace("a").get("k") === null, "read-only missing reads null");
try {
  await ro.keyspace("a").set("k", "v");
  check(false, "read-only set must reject");
} catch (e) {
  check(String(e.message).includes("read-only"), "read-only set rejects: " + e.message);
}
const existing = await openKv("held.kv", { access: "read-only" });
const held = await existing.keyspace("a").getText("k");
check(held === "held", "read-only reads existing");
if (failures.length !== 0) {
  for (let i = 0; i < failures.length; i++) console.error("FAIL: " + failures[i]);
  throw new Error(failures.length + " failure(s)");
}
console.log("KV-OK modes");
"#;
    assert_marker(&run(dir.path(), "modes.js", modes, &[]), "KV-OK modes");
}

#[test]
fn closed_handles_reject() {
    let dir = tempfile::tempdir().unwrap();
    let source = r#"import { openKv } from "avril:kv";
const failures = [];
function check(cond, message) {
  if (!cond) failures[failures.length] = message;
}
const db = await openKv("state.kv", { create: true });
const ks = db.keyspace("a");
await ks.set("k", "v");
db.close();
db.close();
try {
  await ks.get("k");
  check(false, "get after close must reject");
} catch (e) {
  check(e instanceof TypeError && e.message.includes("closed"), "get rejects: " + e.message);
}
try {
  db.keyspace("b");
  check(false, "keyspace after close must throw");
} catch (e) {
  check(e instanceof TypeError, "keyspace throws TypeError");
}
await db[Symbol.asyncDispose]();
let retained = null;
{
  await using live = await openKv("state.kv", { create: true });
  retained = live.keyspace("a");
}
try {
  await retained.get("k");
  check(false, "use after await using must reject");
} catch (e) {
  check(e instanceof TypeError, "await using disposes");
}
if (failures.length !== 0) {
  for (let i = 0; i < failures.length; i++) console.error("FAIL: " + failures[i]);
  throw new Error(failures.length + " failure(s)");
}
console.log("KV-OK closed");
"#;
    assert_marker(&run(dir.path(), "main.js", source, &[]), "KV-OK closed");
}

#[test]
fn capability_denial_at_open_and_after_pledge() {
    let dir = tempfile::tempdir().unwrap();
    let denied = run(
        dir.path(),
        "main.js",
        r#"import { openKv } from "avril:kv";
try {
  await openKv("state.kv", { create: true });
  console.log("OPENED-UNEXPECTEDLY");
} catch (e) {
  console.log("KV-OK denied: " + e.message);
}
"#,
        &["--deny-write=."],
    );
    assert_marker(&denied, "KV-OK denied: write state.kv");
    assert!(
        fs::metadata(dir.path().join("state.kv")).is_err(),
        "denied open must not create the file"
    );
    let pledged = run(
        dir.path(),
        "pledge.js",
        r#"import { openKv } from "avril:kv";
import { pledge } from "avril:std";
const failures = [];
function check(cond, message) {
  if (!cond) failures[failures.length] = message;
}
const db = await openKv("state.kv", { create: true });
await db.keyspace("a").set("k", "v");
pledge({ read: false, write: false });
try {
  await db.keyspace("a").get("k");
  check(false, "get after pledge must reject");
} catch (e) {
  check(String(e.message).includes("read state.kv"), "get denial: " + e.message);
}
try {
  await db.keyspace("a").set("k", "v2");
  check(false, "set after pledge must reject");
} catch (e) {
  // Reads are re-authorized before writes, so the read denial surfaces first.
  check(String(e.message).includes("read state.kv"), "set denial: " + e.message);
}
if (failures.length !== 0) {
  for (let i = 0; i < failures.length; i++) console.error("FAIL: " + failures[i]);
  throw new Error(failures.length + " failure(s)");
}
console.log("KV-OK pledge");
"#,
        &[],
    );
    assert_marker(&pledged, "KV-OK pledge");
}

#[test]
fn invalid_keys_and_values_reject() {
    let dir = tempfile::tempdir().unwrap();
    let source = r#"import { openKv } from "avril:kv";
const failures = [];
function check(cond, message) {
  if (!cond) failures[failures.length] = message;
}
async function rejectsTypeError(fn, label) {
  try {
    await fn();
  } catch (e) {
    check(e instanceof TypeError, label + " is a TypeError, got: " + e);
    return;
  }
  check(false, label + " must reject");
}
const db = await openKv("state.kv", { create: true });
const ks = db.keyspace("a");
await rejectsTypeError(() => ks.set("k", 42), "number value");
await rejectsTypeError(() => ks.set("k", { nested: true }), "object value");
await rejectsTypeError(() => ks.set("k", () => {}), "function value");
await rejectsTypeError(() => ks.set("k", Symbol("s")), "symbol value");
await rejectsTypeError(() => ks.set("k", undefined), "top-level undefined");
await rejectsTypeError(() => ks.get(42), "number key");
await rejectsTypeError(() => ks.get(Object.create(Uint8Array.prototype)), "forged Uint8Array key");
class Bytes extends Uint8Array {}
await ks.set(new Bytes([7]), "subclass");
check((await ks.getText(new Uint8Array([7]))) === "subclass", "Uint8Array subclass key works");
await rejectsTypeError(() => ks.set("k", "\ud800"), "lone-surrogate value");
await rejectsTypeError(() => ks.get("\ud800"), "ill-formed string key");
await rejectsTypeError(() => db.atomic((tx) => tx.check(ks, "k", 42)), "number stamp");
await rejectsTypeError(() => db.atomic((tx) => tx.check(ks, "k", "1")), "string stamp");
try {
  db.keyspace("");
  check(false, "empty keyspace name must throw");
} catch (e) {
  check(e instanceof TypeError, "empty keyspace is a TypeError");
}
// Empty key and empty value are valid: bytes, including empty, are keys.
await ks.set("", "empty-key");
await ks.set("empty-value", new Uint8Array([]));
const ek = await ks.getText("");
check(ek === "empty-key", "empty string key works");
const ev = await ks.get("empty-value");
check(ev !== null && ev instanceof Uint8Array && ev.length === 0, "empty bytes value works");
// Non-UTF-8 bytes: get works, getText rejects.
await ks.set("raw", new Uint8Array([0xff, 0x00]));
const gotRaw = await ks.get("raw");
check(gotRaw !== null && Array.from(gotRaw).join(",") === "255,0", "non-UTF8 set works and reads bytes");
try {
  await ks.getText("raw");
  check(false, "getText of non-UTF8 must reject");
} catch (e) {
  check(e instanceof TypeError, "getText rejection is a TypeError, got: " + e);
}
// Oversized batches: the offending tx call itself throws RangeError, so the
// callback aborts and nothing commits.
const huge = new Uint8Array(4 * 1024 * 1024);
try {
  await db.atomic((tx) => {
    tx.set(ks, "huge", huge);
  });
  check(false, "oversized atomic batch must throw");
} catch (e) {
  check(e instanceof RangeError, "oversized atomic batch is a RangeError, got: " + e);
}
check(await ks.get("huge") === null, "oversized atomic batch commits nothing");
let opCount = 0;
try {
  await db.atomic((tx) => {
    for (; opCount <= 1000; opCount++) tx.delete(ks, "k" + opCount);
  });
  check(false, "too many atomic ops must throw");
} catch (e) {
  check(e instanceof RangeError && opCount === 1000, "op 1001 is a RangeError, got: " + e + " at " + opCount);
}
if (failures.length !== 0) {
  for (let i = 0; i < failures.length; i++) console.error("FAIL: " + failures[i]);
  throw new Error(failures.length + " failure(s)");
}
console.log("KV-OK invalid");
"#;
    assert_marker(&run(dir.path(), "main.js", source, &[]), "KV-OK invalid");
}

#[test]
fn poisoned_builtins_do_not_affect_roundtrip() {
    let dir = tempfile::tempdir().unwrap();
    let source = r#"import { openKv } from "avril:kv";
const failures = [];
function check(cond, message) {
  if (!cond) failures[failures.length] = message;
}
const db = await openKv("state.kv", { create: true });
const ks = db.keyspace("p");
await ks.set("s", "hello");
await ks.set(new Uint8Array([1, 2]), new Uint8Array([3, 4]));
Array.prototype.push = function () { throw new Error("poisoned"); };
Map.prototype.set = function () { throw new Error("poisoned"); };
Object.prototype.evil = 1;
const text = await ks.getText("s");
check(text === "hello", "string roundtrip under poisoning");
const bytes = await ks.get(new Uint8Array([1, 2]));
check(bytes !== null && Array.from(bytes).join(",") === "3,4", "bytes roundtrip under poisoning");
const listed = [];
for await (const e of ks.list()) listed[listed.length] = e.key.length;
check(listed.length === 2, "list works under poisoning");
if (failures.length !== 0) {
  for (let i = 0; i < failures.length; i++) console.error("FAIL: " + failures[i]);
  throw new Error(failures.length + " failure(s)");
}
console.log("KV-OK poison");
"#;
    assert_marker(&run(dir.path(), "main.js", source, &[]), "KV-OK poison");
}