mod common;

use std::fs;
use std::io::Write;
#[cfg(unix)]
use std::io::{BufRead, Read};
use std::process::{Command, Stdio};
#[cfg(unix)]
use std::time::Duration;

#[test]
fn blocking_binary_io_preserves_views_and_does_not_run_callbacks() {
    let input: Vec<u8> = (0..160_003).map(|index| (index % 251) as u8).collect();
    let output = common::run(
        r#"
        import { stdin, stdout } from 'avril:std/sync';
        const storage = new Uint8Array(1033).fill(255);
        const buffer = new Uint8Array(storage.buffer, 7, 1024);
        let ran = false;
        queueMicrotask(() => ran = true);
        if (stdin.read(new Uint8Array()) !== 0) throw Error('Empty read');
        for (;;) {
            const count = stdin.read(buffer);
            if (count === null) break;
            if (count <= 0 || count > buffer.length) throw Error('Read count');
            if (stdout.write(buffer.subarray(0, count)) !== count) throw Error('Write count');
            if (storage[0] !== 255 || storage[6] !== 255 || storage[1032] !== 255) throw Error('View boundary');
            if (ran) throw Error('Sync IO ran JS');
        }
        if (stdin.read(new Uint8Array(1)) !== null || stdin.read(new Uint8Array()) !== 0) throw Error('EOF');
        if (stdout.write(new Uint8Array()) !== 0) throw Error('Empty write');
    "#,
        &input,
    );
    assert_eq!(output.stdout, input);
    assert!(output.stderr.is_empty());
}

#[test]
fn blocking_output_shares_ordering_with_console_and_submitted_stream_writes() {
    let output = common::run(
        r#"
        import { stdout as stream, stderr as errorStream } from 'avril:std';
        import { stdout, stderr } from 'avril:std/sync';
        const encode = text => new TextEncoder().encode(text);
        const writer = stream.getWriter(), errors = errorStream.getWriter();
        await writer.ready; await errors.ready;
        const first = writer.write(encode('A'));
        let settled = false;
        first.then(() => settled = true);
        if (stdout.write(encode('B')) !== 1 || settled) throw Error('Output order or reentrancy');
        console.log('C');
        await first;
        await writer.write(encode('D'));
        stdout.write(encode('E'));
        await writer.close();
        stdout.write(encode('F'));
        const pending = errors.write(encode('G'));
        stderr.write(encode('H')); console.error('I');
        await pending; await errors.close();
        stderr.write(encode('J'));
    "#,
        &[],
    );
    assert_eq!(output.stdout, b"ABC\nDEF");
    assert_eq!(output.stderr, b"GHI\nJ");
}

#[test]
fn sync_input_refuses_to_steal_an_outstanding_async_read() {
    let (_dir, mut child) = common::spawn(
        r#"
        import { stdin as stream } from 'avril:std';
        import { stdin, errors } from 'avril:std/sync';
        const reader = stream.getReader();
        await Promise.resolve();
        const pending = reader.read();
        let busy = false;
        try { stdin.read(new Uint8Array(1)); }
        catch (error) { if (!(error instanceof errors.Busy)) throw error; busy = true; }
        if (!busy) throw Error('Overlapping read accepted');
        if (stdin.read(new Uint8Array()) !== 0) throw Error('Empty read should not reserve stdin');
        await reader.cancel();
        if (!(await pending).done || stdin.read(new Uint8Array(1)) !== null) throw Error('Cancellation');
        console.log('canceled');
    "#,
    );
    let stdin = child.stdin.take().unwrap();
    let output = common::wait(child);
    drop(stdin);
    assert!(
        output.status.success(),
        "{}",
        String::from_utf8_lossy(&output.stderr)
    );
    assert_eq!(output.stdout, b"canceled\n");
}

#[test]
fn completed_sync_and_byob_reads_share_the_input_without_losing_bytes() {
    let output = common::run(
        r#"
        import { stdin as stream } from 'avril:std';
        import { stdin, stdout } from 'avril:std/sync';
        const first = new Uint8Array(2);
        if (stdin.read(first) !== 2) throw Error('Initial read');
        stdout.write(first);
        const reader = stream.getReader({mode: 'byob'});
        const {value, done} = await reader.read(new Uint8Array(3));
        if (done || value.length !== 3) throw Error('BYOB read');
        stdout.write(value); reader.releaseLock();
        const last = new Uint8Array(16), count = stdin.read(last);
        if (count !== 1) throw Error('Trailing read');
        stdout.write(last.subarray(0, count));
        if (stdin.read(last) !== null) throw Error('EOF');
    "#,
        b"abcdef",
    );
    assert_eq!(output.stdout, b"abcdef");
}

#[test]
fn blocking_stdio_validates_buffers_and_exposes_only_frozen_interfaces() {
    let output = common::run(
        r#"
        import { stdin, stdout, stderr } from 'avril:std/sync';
        import { stdin as stream } from 'avril:std';
        for (const value of [stdin, stdout, stderr]) {
            if (!Object.isFrozen(value) || 'close' in value || Symbol.dispose in value) throw Error('Ownership');
        }
        if (stdin === stream || 'readable' in stdin || 'writable' in stdout) throw Error('Stream API replaced');
        const bytes = new Uint8Array(2);
        structuredClone(bytes.buffer, {transfer: [bytes.buffer]});
        const resizable = new ArrayBuffer(4, {maxByteLength: 8});
        const outOfBounds = new Uint8Array(resizable, 1, 2); resizable.resize(0);
        for (const invalid of [undefined, null, [], 'text', new DataView(new ArrayBuffer(2)), new Uint16Array(2), bytes, outOfBounds]) {
            for (const method of [stdin.read, stdout.write, stderr.write]) {
                let threw = false;
                try { method(invalid); }
                catch (error) { if (!(error instanceof TypeError)) throw error; threw = true; }
                if (!threw) throw Error('Accepted invalid buffer');
            }
        }
        console.log('validated');
    "#,
        &[],
    );
    assert_eq!(output.stdout, b"validated\n");
}

#[test]
fn blocking_stdio_supports_shared_resizable_views_and_captured_intrinsics() {
    let output = common::run(
        r#"
        import { stdin, stdout } from 'avril:std/sync';
        const shared = new Uint8Array(new SharedArrayBuffer(4));
        const a = new Uint8Array(shared.buffer, 1, 2);
        const resizable = new ArrayBuffer(4, {maxByteLength: 8});
        const b = new Uint8Array(resizable, 1, 2);
        const c = new Uint8Array(resizable, 3); resizable.resize(7);
        const growable = new SharedArrayBuffer(1, {maxByteLength: 4});
        const d = new Uint8Array(growable); growable.grow(2);
        const fail = () => { throw Error('Mutable intrinsic'); };
        globalThis.Uint8Array = fail;
        Object.getPrototypeOf(a).set = fail;
        Object.prototype.toJSON = fail;
        Array.prototype.toJSON = fail;
        for (const view of [a, b, c, d]) {
            if (stdin.read(view) !== view.length || stdout.write(view) !== view.length) throw Error('Transfer count');
        }
    "#,
        b"ABCDEFGHIJ",
    );
    assert_eq!(output.stdout, b"ABCDEFGHIJ");
}

#[test]
fn stdio_remains_available_without_filesystem_authority() {
    let output = common::run(
        r#"
        import { pledge } from 'avril:std';
        import { stdin, stdout, stderr } from 'avril:std/sync';
        pledge({read: false, write: false, env: false, sys: false});
        const bytes = new Uint8Array(32), count = stdin.read(bytes);
        stdout.write(bytes.subarray(0, count)); stderr.write(new TextEncoder().encode('diagnostic'));
    "#,
        b"stdio",
    );
    assert_eq!(output.stdout, b"stdio");
    assert_eq!(output.stderr, b"diagnostic");
}

#[cfg(unix)]
#[test]
fn broken_pipes_throw_synchronously_and_remain_output_failures() {
    let (_dir, mut child) = common::spawn(
        r#"
        import { stdout, stderr, errors } from 'avril:std/sync';
        let failed = false;
        try { stdout.write(new Uint8Array([1])); }
        catch (error) { if (!(error instanceof errors.BrokenPipe)) throw error; failed = true; }
        if (!failed) throw Error('Broken pipe accepted');
        stderr.write(new TextEncoder().encode('caught BrokenPipe\n'));
    "#,
    );
    drop(child.stdout.take());
    let output = common::wait(child);
    assert!(!output.status.success());
    assert!(String::from_utf8_lossy(&output.stderr).contains("caught BrokenPipe"));
}

#[test]
fn the_repl_reserves_input_but_allows_blocking_output_in_callbacks() {
    let dir = tempfile::tempdir().unwrap();
    let mut child = Command::new(env!("CARGO_BIN_EXE_avril"))
        .args(["repl", "-0"])
        .current_dir(dir.path())
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()
        .unwrap();
    child.stdin.take().unwrap().write_all(br#"import { stdin, stderr, errors } from 'avril:std/sync'
try { stdin.read(new Uint8Array(1)) } catch (error) { error instanceof errors.NotSupported }
await new Promise(resolve => setTimeout(() => { stderr.write(new TextEncoder().encode('callback')); resolve(); }, 0))
.exit
"#).unwrap();
    let output = common::wait(child);
    assert!(
        output.status.success(),
        "{}",
        String::from_utf8_lossy(&output.stderr)
    );
    assert!(String::from_utf8_lossy(&output.stdout).contains("true\n"));
    assert_eq!(output.stderr, b"callback");
}

fn worker(main: &str, source: &str) -> (tempfile::TempDir, std::process::Child) {
    let dir = tempfile::tempdir().unwrap();
    fs::write(dir.path().join("main.js"), main).unwrap();
    fs::write(dir.path().join("worker.js"), source).unwrap();
    fs::write(dir.path().join("lock"), "").unwrap();
    let child = Command::new(env!("CARGO_BIN_EXE_avril"))
        .args(["run", "main.js"])
        .current_dir(dir.path())
        .stdin(Stdio::null())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()
        .unwrap();
    (dir, child)
}

#[test]
fn workers_reserve_input_but_allow_blocking_output_from_callbacks() {
    let (_dir, child) = worker(
        r#"
        const worker = new Worker('./worker.js', {type: 'module'});
        const done = new Promise((resolve, reject) => { worker.onmessage = resolve; worker.onerror = reject; });
        worker.postMessage('go'); await done; worker.terminate();
    "#,
        r#"
        import { stdin, stdout, errors } from 'avril:std/sync';
        onmessage = () => {
            let refused = false;
            try { stdin.read(new Uint8Array(1)); }
            catch (error) { if (!(error instanceof errors.NotSupported)) throw error; refused = true; }
            if (!refused) throw Error('Worker consumed stdin');
            stdout.write(new TextEncoder().encode('worker output'));
            postMessage('done');
        };
    "#,
    );
    let output = common::wait(child);
    assert!(
        output.status.success(),
        "{}",
        String::from_utf8_lossy(&output.stderr)
    );
    assert_eq!(output.stdout, b"worker output");
}

#[cfg(unix)]
fn stderr_lines(
    stderr: std::process::ChildStderr,
) -> (
    std::sync::mpsc::Receiver<String>,
    std::thread::JoinHandle<Vec<u8>>,
) {
    let (send, receive) = std::sync::mpsc::channel();
    let reader = std::thread::spawn(move || {
        let mut stderr = std::io::BufReader::new(stderr);
        let mut output = Vec::new();
        loop {
            let mut line = String::new();
            if stderr.read_line(&mut line).unwrap() == 0 {
                break;
            }
            output.extend_from_slice(line.as_bytes());
            let _ = send.send(line);
        }
        output
    });
    (receive, reader)
}

#[cfg(unix)]
#[test]
fn terminating_a_worker_interrupts_blocked_output_and_its_shutdown_flush() {
    let (_dir, mut child) = worker(
        r#"
        import { open, stderr } from 'avril:std/sync';
        const worker = new Worker('./worker.js', {type: 'module'});
        await new Promise((resolve, reject) => { worker.onmessage = resolve; worker.onerror = reject; });
        await new Promise(resolve => setTimeout(resolve, 30));
        worker.terminate();
        const file = open('lock', {read: true, write: true});
        file.flockSync(); file.funlockSync(); file.close();
        stderr.write(new TextEncoder().encode('terminated\n'));
    "#,
        r#"
        import { open, stdout } from 'avril:std/sync';
        const file = open('lock', {read: true, write: true});
        file.flockSync();
        postMessage('writing');
        stdout.write(new Uint8Array(4 * 1024 * 1024).fill(65));
        throw Error('Blocked write returned before termination');
    "#,
    );
    let (receive, reader) = stderr_lines(child.stderr.take().unwrap());
    let marker = receive.recv_timeout(Duration::from_secs(3));
    if marker.as_deref() != Ok("terminated\n") {
        let _ = child.kill();
        let output = common::wait(child);
        let stderr = reader.join().unwrap();
        panic!(
            "Worker retained its file lock during output: {marker:?}; {:?}; {stderr:?}",
            output.status
        );
    }
    let output = common::wait(child);
    let stderr = reader.join().unwrap();
    assert!(
        output.status.success(),
        "{}",
        String::from_utf8_lossy(&stderr)
    );
    assert_eq!(output.stdout, vec![65; 4 * 1024 * 1024]);
}

#[cfg(unix)]
#[test]
fn repl_interrupts_a_blocked_sync_write_without_waiting_for_stdout_to_drain() {
    use std::os::unix::process::CommandExt;

    let dir = tempfile::tempdir().unwrap();
    let mut command = Command::new(env!("CARGO_BIN_EXE_avril"));
    // SAFETY: signal is async-signal-safe; no inherited SIGINT ignore may mask the test.
    unsafe {
        command.pre_exec(|| {
            libc::signal(libc::SIGINT, libc::SIG_DFL);
            Ok(())
        });
    }
    let mut child = command
        .args(["repl", "-0"])
        .current_dir(dir.path())
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()
        .unwrap();
    let mut stdin = child.stdin.take().unwrap();
    let (messages, stderr) = stderr_lines(child.stderr.take().unwrap());
    let mut stdout = child.stdout.take().unwrap();
    let (send, receive) = std::sync::mpsc::channel();
    let first = std::thread::spawn(move || {
        let mut byte = [0];
        let result = stdout.read_exact(&mut byte).map(|()| byte[0]);
        let _ = send.send(result);
        stdout
    });
    stdin.write_all(b"(await import('avril:std/sync')).stdout.write(new Uint8Array(4 * 1024 * 1024).fill(65))\n").unwrap();
    let started = receive.recv_timeout(Duration::from_secs(3));
    if !matches!(started, Ok(Ok(65))) {
        let _ = child.kill();
        child.stdout = Some(first.join().unwrap());
        let output = common::wait(child);
        panic!(
            "Sync write did not start: {started:?}; {:?}; {:?}",
            output.status,
            stderr.join().unwrap()
        );
    }
    child.stdout = Some(first.join().unwrap());
    assert_eq!(unsafe { libc::kill(child.id() as i32, libc::SIGINT) }, 0);
    let interrupted = messages.recv_timeout(Duration::from_secs(3));
    if interrupted.as_deref() != Ok("Evaluation interrupted\n") {
        let _ = child.kill();
        let output = common::wait(child);
        panic!(
            "Sync write was not interrupted: {interrupted:?}; {:?}; {:?}",
            output.status,
            stderr.join().unwrap()
        );
    }
    stdin.write_all(b".exit\n").unwrap();
    drop(stdin);
    let output = common::wait(child);
    let stderr = stderr.join().unwrap();
    assert!(
        output.status.success(),
        "{}",
        String::from_utf8_lossy(&stderr)
    );
    assert_eq!(output.stdout, vec![65; 4 * 1024 * 1024 - 1]);
}