mod common;

use std::io::{BufRead, BufReader, Read, Write};
use std::net::TcpStream;
use std::process::{Child, Command, Stdio};
use std::time::Duration;

struct Hosted {
    _dir: tempfile::TempDir,
    child: Option<Child>,
    address: String,
}

impl Hosted {
    fn start(source: &str, options: &[&str]) -> Self {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("main.js");
        std::fs::write(&path, source).unwrap();
        let child = Command::new(env!("CARGO_BIN_EXE_avril"))
            .args(["serve", "-0"])
            .args(options)
            .arg(path)
            .stdin(Stdio::null())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .spawn()
            .unwrap();
        let mut hosted = Self {
            _dir: dir,
            child: Some(child),
            address: String::new(),
        };
        let child = hosted.child.as_mut().unwrap();
        let mut stderr = BufReader::new(child.stderr.take().unwrap());
        let mut line = String::new();
        stderr.read_line(&mut line).unwrap();
        hosted.address = line
            .trim()
            .strip_prefix("Listening on ")
            .expect(&line)
            .to_owned();
        child.stderr = Some(stderr.into_inner());
        hosted
    }

    #[cfg(unix)]
    fn stop(self) {
        self.stop_with(libc::SIGINT);
    }

    #[cfg(unix)]
    fn stop_with(mut self, signal: i32) {
        let child = self.child.take().unwrap();
        assert_eq!(unsafe { libc::kill(child.id() as i32, signal) }, 0);
        let output = common::wait(child);
        assert!(
            output.status.success(),
            "{}",
            String::from_utf8_lossy(&output.stderr)
        );
    }
}

impl Drop for Hosted {
    fn drop(&mut self) {
        if let Some(mut child) = self.child.take() {
            let _ = child.kill();
            let _ = child.wait();
        }
    }
}

fn headers(reader: &mut impl BufRead) -> String {
    let mut result = String::new();
    loop {
        let mut line = String::new();
        assert_ne!(reader.read_line(&mut line).unwrap(), 0, "unexpected EOF");
        result.push_str(&line);
        if line == "\r\n" {
            return result;
        }
    }
}

fn masked_frame(socket: &mut impl Write, fin: bool, opcode: u8, payload: &[u8]) {
    assert!(payload.len() <= 125);
    let mask = [1, 7, 3, 9];
    let mut frame = vec![
        (if fin { 0x80 } else { 0 }) | opcode,
        0x80 | payload.len() as u8,
    ];
    frame.extend_from_slice(&mask);
    for (index, byte) in payload.iter().enumerate() {
        frame.push(byte ^ mask[index % 4]);
    }
    socket.write_all(&frame).unwrap();
}

fn server_frame(socket: &mut impl Read) -> (u8, Vec<u8>) {
    let mut header = [0; 2];
    socket.read_exact(&mut header).unwrap();
    assert_eq!(header[1] & 0x80, 0, "server frames must not be masked");
    assert_ne!(header[0] & 0x80, 0);
    let mut bytes = vec![0; (header[1] & 0x7f) as usize];
    socket.read_exact(&mut bytes).unwrap();
    (header[0] & 0x0f, bytes)
}

#[test]
#[cfg(unix)]
fn cli_file_responses_require_guest_read_authority() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("file.txt");
    std::fs::write(&path, b"file body").unwrap();
    let source = format!(
        "import {{serveFile}} from 'avril:std'; export default {{fetch(request) {{return serveFile(request, {});}}}};",
        serde_json::to_string(path.to_str().unwrap()).unwrap()
    );
    let grant = format!("--allow-read={}", path.display());
    for allowed in [false, true] {
        let mut options = vec!["--port=0"];
        if allowed {
            options.push(&grant);
        }
        let hosted = Hosted::start(&source, &options);
        let url = url::Url::parse(&hosted.address).unwrap();
        let mut socket =
            TcpStream::connect((url.host_str().unwrap(), url.port().unwrap())).unwrap();
        socket
            .set_read_timeout(Some(Duration::from_secs(3)))
            .unwrap();
        socket
            .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n")
            .unwrap();
        let mut reader = BufReader::new(socket);
        let head = headers(&mut reader);
        let mut body = String::new();
        reader.read_to_string(&mut body).unwrap();
        if allowed {
            assert!(head.starts_with("HTTP/1.1 200"), "{head}");
            assert_eq!(body, "file body");
        } else {
            assert!(head.starts_with("HTTP/1.1 500"), "{head}");
            assert_eq!(body, "Internal Server Error");
        }
        hosted.stop();
    }
}

#[test]
#[cfg(unix)]
fn cli_serves_without_net_authority_preserves_this_and_waits_for_top_level_await() {
    let hosted = Hosted::start(
        r#"
        import {serve} from 'avril:std';
        await new Promise(resolve=>setTimeout(resolve,10));
        setInterval(()=>{},1000);
        export default {
            prefix:'host:',
            async fetch(request,info) {
                try {serve({port:0},()=>new Response()); throw Error('programmatic listener escaped');}
                catch(e) {if(!e.message.includes('PermissionDenied')) throw e;}
                try {await fetch(request.url); throw Error('outbound fetch escaped');}
                catch(e) {if(!e.message.includes('PermissionDenied')) throw e;}
                try {new WebSocket(request.url); throw Error('outbound WebSocket escaped');}
                catch(e) {if(!e.message.includes('PermissionDenied')) throw e;}
                if(info.remoteAddr.transport!=='tcp') throw Error('peer');
                return new Response(this.prefix+await request.text(),{status:202});
            }
        };
    "#,
        &["--port=0"],
    );
    let url = url::Url::parse(&hosted.address).unwrap();
    let mut socket = TcpStream::connect((url.host_str().unwrap(), url.port().unwrap())).unwrap();
    socket
        .set_read_timeout(Some(Duration::from_secs(3)))
        .unwrap();
    socket.write_all(b"POST / HTTP/1.1\r\nHost: example.test\r\nContent-Length: 4\r\nConnection: close\r\n\r\ntest").unwrap();
    let mut reader = BufReader::new(socket);
    assert!(headers(&mut reader).starts_with("HTTP/1.1 202"));
    let mut body = String::new();
    reader.read_to_string(&mut body).unwrap();
    assert_eq!(body, "host:test");
    hosted.stop();
}

#[test]
fn cli_rejects_invalid_default_exports_and_listener_options() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("main.js");
    for source in [
        "",
        "export default () => new Response();",
        "export default null;",
        "export default {};",
        "export default {fetch:1};",
    ] {
        std::fs::write(&path, source).unwrap();
        let child = Command::new(env!("CARGO_BIN_EXE_avril"))
            .args(["serve", "-0", "--port=0"])
            .arg(&path)
            .stdout(Stdio::piped())
            .stderr(Stdio::piped())
            .spawn()
            .unwrap();
        let output = common::wait(child);
        assert!(!output.status.success());
        let error = String::from_utf8_lossy(&output.stderr);
        assert!(
            error.contains("default export object with a fetch function"),
            "{error}"
        );
        assert!(!error.contains("Listening on"));
    }
    for options in [
        vec!["--unix=/tmp/unused-avril.sock", "--port=0"],
        vec!["--port=65536"],
        vec!["--port=-1"],
    ] {
        let output = Command::new(env!("CARGO_BIN_EXE_avril"))
            .arg("serve")
            .args(options)
            .arg(&path)
            .output()
            .unwrap();
        assert!(!output.status.success());
        assert!(!String::from_utf8_lossy(&output.stderr).contains("Listening on"));
    }
    let output = Command::new(env!("CARGO_BIN_EXE_avril"))
        .args(["serve", "--help"])
        .output()
        .unwrap();
    let help = String::from_utf8_lossy(&output.stdout);
    assert!(
        help.contains("--unix=PATH")
            && help.contains("--hostname=HOST")
            && help.contains("--port=PORT")
    );
}

#[test]
#[cfg(unix)]
fn cli_unix_listener_needs_no_guest_filesystem_authority_and_cleans_up() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("server.sock");
    let option = format!("--unix={}", path.display());
    let hosted = Hosted::start(
        r#"
        export default {fetch(request,info){
            if(info.remoteAddr.transport!=='unix'||info.remoteAddr.path!==null) throw Error('peer');
            return new Response(request.url);
        }};
    "#,
        &[&option],
    );
    assert_eq!(hosted.address, format!("unix:{}", path.display()));
    let mut socket = std::os::unix::net::UnixStream::connect(&path).unwrap();
    socket
        .set_read_timeout(Some(Duration::from_secs(3)))
        .unwrap();
    socket
        .write_all(b"GET /unix?q=1 HTTP/1.1\r\nHost: unix.test\r\nConnection: close\r\n\r\n")
        .unwrap();
    let mut reader = BufReader::new(socket);
    assert!(headers(&mut reader).starts_with("HTTP/1.1 200"));
    let mut body = String::new();
    reader.read_to_string(&mut body).unwrap();
    assert_eq!(body, "http://unix.test/unix?q=1");
    hosted.stop();
    assert!(!path.exists());
}

#[test]
#[cfg(unix)]
fn programmatic_unix_listener_requires_only_write_and_preserves_replacements() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("server.sock");
    let label = serde_json::to_string(path.to_str().unwrap()).unwrap();
    let (_source, mut child) = common::spawn(&format!(
        r#"
        import {{serve,pledge}} from 'avril:std';
        pledge({{net:false,write:{{allow:[{label}]}}}});
        const server=serve({{path:{label}}},()=>{{server.shutdown();return new Response('unix');}});
        if(server.addr.transport!=='unix'||server.addr.path!=={label}) throw Error('address');
        console.log('ready'); await server.finished;
    "#
    ));
    let mut stdout = BufReader::new(child.stdout.take().unwrap());
    let mut line = String::new();
    stdout.read_line(&mut line).unwrap();
    assert_eq!(line, "ready\n");
    child.stdout = Some(stdout.into_inner());
    let mut socket = std::os::unix::net::UnixStream::connect(&path).unwrap();
    socket
        .set_read_timeout(Some(Duration::from_secs(3)))
        .unwrap();
    std::fs::remove_file(&path).unwrap();
    std::fs::write(&path, "replacement").unwrap();
    socket
        .write_all(b"GET / HTTP/1.1\r\nHost: unix.test\r\nConnection: close\r\n\r\n")
        .unwrap();
    let mut reader = BufReader::new(socket);
    assert!(headers(&mut reader).starts_with("HTTP/1.1 200"));
    let mut body = String::new();
    reader.read_to_string(&mut body).unwrap();
    assert_eq!(body, "unix");
    let output = common::wait(child);
    assert!(
        output.status.success(),
        "{}",
        String::from_utf8_lossy(&output.stderr)
    );
    assert_eq!(std::fs::read_to_string(&path).unwrap(), "replacement");
    let output = common::run(
        &format!(
            r#"
        import {{serve,pledge}} from 'avril:std';
        try {{serve({{path:{label}}},()=>new Response());throw Error('replaced existing entry');}}
        catch(e) {{if(e.message==='replaced existing entry')throw e;}}
        pledge({{write:false}});
        try {{serve({{path:{label}}},()=>new Response());throw Error('no write authority');}}
        catch(e) {{if(!e.message.includes('PermissionDenied'))throw e;}}
        console.log('denied');
    "#
        ),
        &[],
    );
    assert_eq!(output.stdout, b"denied\n");
    assert_eq!(std::fs::read_to_string(&path).unwrap(), "replacement");
}

#[test]
fn websockets_upgrade_echo_text_binary_and_complete_clean_close() {
    let output = common::run(
        r#"
        import {serve,upgradeWebSocket} from 'avril:std';
        const server=serve({port:0},request=>{
            const {socket,response}=upgradeWebSocket(request,{protocol:'chat'});
            if(!(socket instanceof WebSocket)||response.status!==101)throw Error('upgrade objects');
            socket.onmessage=e=>socket.send(e.data);
            return response;
        });
        const client=new WebSocket(`ws://127.0.0.1:${server.addr.port}`,'chat');
        client.binaryType='arraybuffer';
        const messages=[];
        client.onopen=()=>{if(client.protocol!=='chat')throw Error('protocol');client.send('echo');client.send(new Uint8Array([4,2]));};
        client.onmessage=e=>{messages.push(typeof e.data==='string'?e.data:new Uint8Array(e.data).join());if(messages.length===2)client.close(3001,'done');};
        const close=await new Promise(resolve=>client.onclose=resolve);
        if(messages.join('|')!=='echo|4,2'||!close.wasClean||close.code!==3001||close.reason!=='done')throw Error('echo or close');
        await server.shutdown();console.log('upgraded');
    "#,
        &[],
    );
    assert_eq!(output.stdout, b"upgraded\n");
}

#[test]
fn abandoning_an_upgrade_releases_the_socket_and_server() {
    let output = common::run(
        r#"
        import {serve,upgradeWebSocket} from 'avril:std';
        let closed;
        const gone=new Promise(resolve=>closed=resolve);
        const server=serve({port:0},request=>{
            const {socket}=upgradeWebSocket(request);
            socket.onclose=closed;
            return new Response('not upgraded');
        });
        const client=new WebSocket(`ws://127.0.0.1:${server.addr.port}`);
        const clientClosed=new Promise(resolve=>client.onclose=resolve);
        await gone; await clientClosed; await server.shutdown(); console.log('abandoned');
    "#,
        &[],
    );
    assert_eq!(output.stdout, b"abandoned\n");
}

#[test]
fn forced_shutdown_closes_upgraded_sockets_even_after_graceful_shutdown() {
    let output = common::run(
        r#"
        import {serve,upgradeWebSocket} from 'avril:std';
        const controller=new AbortController();let serverClosed;
        const gone=new Promise(resolve=>serverClosed=resolve);
        const server=serve({port:0,signal:controller.signal},request=>{
            const {socket,response}=upgradeWebSocket(request);socket.onclose=serverClosed;return response;
        });
        const client=new WebSocket(`ws://127.0.0.1:${server.addr.port}`);
        await new Promise(resolve=>client.onopen=resolve);
        const closed=new Promise(resolve=>client.onclose=resolve);
        let finished=false;server.shutdown().then(()=>finished=true);
        await new Promise(resolve=>setTimeout(resolve,10));if(finished)throw Error('lost active WebSocket');
        controller.abort(); await gone; await closed; await server.finished;console.log('forced');
    "#,
        &[],
    );
    assert_eq!(output.stdout, b"forced\n");
}

fn handshake<S: Read + Write>(mut socket: S) -> BufReader<S> {
    socket.write_all(b"GET / HTTP/1.1\r\nHost: example.test\r\nConnection: keep-alive, Upgrade\r\nUpgrade: websocket\r\nSec-WebSocket-Version: 13\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n\r\n").unwrap();
    let mut reader = BufReader::new(socket);
    let response = headers(&mut reader);
    assert!(response.starts_with("HTTP/1.1 101"), "{response}");
    assert!(
        response
            .to_ascii_lowercase()
            .contains("sec-websocket-accept: s3pplmbitxaq9kygzzhzrbk+xoo=")
    );
    reader
}

#[test]
#[cfg(unix)]
fn cli_websocket_upgrades_require_no_guest_net_authority_and_obey_masking() {
    let hosted = Hosted::start(
        r#"
        import {upgradeWebSocket} from 'avril:std';
        export default {fetch(request){const {socket,response}=upgradeWebSocket(request);socket.onmessage=e=>socket.send(e.data);return response;}};
    "#,
        &["--port=0"],
    );
    let url = url::Url::parse(&hosted.address).unwrap();
    let socket = TcpStream::connect((url.host_str().unwrap(), url.port().unwrap())).unwrap();
    socket
        .set_read_timeout(Some(Duration::from_secs(3)))
        .unwrap();
    let mut reader = handshake(socket);
    masked_frame(reader.get_mut(), false, 1, b"frag");
    masked_frame(reader.get_mut(), true, 9, b"ping");
    assert_eq!(server_frame(&mut reader), (10, b"ping".to_vec()));
    masked_frame(reader.get_mut(), true, 0, b"mented");
    assert_eq!(server_frame(&mut reader), (1, b"fragmented".to_vec()));
    reader.get_mut().write_all(b"\x81\x01x").unwrap();
    let (opcode, payload) = server_frame(&mut reader);
    assert_eq!(opcode, 8);
    assert_eq!(u16::from_be_bytes([payload[0], payload[1]]), 1002);
    drop(reader);
    hosted.stop();
}

#[test]
#[cfg(unix)]
fn websocket_upgrades_work_over_unix_sockets() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("websocket.sock");
    let option = format!("--unix={}", path.display());
    let hosted = Hosted::start(
        r#"
        import {upgradeWebSocket} from 'avril:std';
        export default {fetch(request){const {socket,response}=upgradeWebSocket(request);socket.onmessage=e=>socket.send(e.data);return response;}};
    "#,
        &[&option],
    );
    let socket = std::os::unix::net::UnixStream::connect(&path).unwrap();
    socket
        .set_read_timeout(Some(Duration::from_secs(3)))
        .unwrap();
    let mut reader = handshake(socket);
    masked_frame(reader.get_mut(), true, 1, b"unix");
    assert_eq!(server_frame(&mut reader), (1, b"unix".to_vec()));
    masked_frame(reader.get_mut(), true, 8, &1000u16.to_be_bytes());
    assert_eq!(
        server_frame(&mut reader),
        (8, 1000u16.to_be_bytes().to_vec())
    );
    drop(reader);
    hosted.stop();
    assert!(!path.exists());
}

#[test]
#[cfg(unix)]
fn malformed_websocket_handshakes_are_rejected_without_terminating_the_server() {
    let hosted = Hosted::start(
        r#"
        import {upgradeWebSocket} from 'avril:std';
        export default {fetch(request){
            try { return upgradeWebSocket(request,new URL(request.url).pathname==='/protocol'?{protocol:'missing'}:{}).response; }
            catch(error){ return new Response(error.message,{status:400}); }
        }};
    "#,
        &["--port=0"],
    );
    let url = url::Url::parse(&hosted.address).unwrap();
    for (method, path, version, key, extra, connection) in [
        ("GET", "/", "14", "dGhlIHNhbXBsZSBub25jZQ==", "", "Upgrade"),
        ("GET", "/", "13", "eA==", "", "Upgrade"),
        (
            "GET",
            "/",
            "13",
            "dGhlIHNhbXBsZSBub25jZQ==",
            "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n",
            "Upgrade",
        ),
        ("POST", "/", "13", "dGhlIHNhbXBsZSBub25jZQ==", "", "Upgrade"),
        ("GET", "/", "13", "dGhlIHNhbXBsZSBub25jZQ==", "", "close"),
        (
            "GET",
            "/protocol",
            "13",
            "dGhlIHNhbXBsZSBub25jZQ==",
            "Sec-WebSocket-Protocol: chat\r\n",
            "Upgrade",
        ),
    ] {
        let mut socket =
            TcpStream::connect((url.host_str().unwrap(), url.port().unwrap())).unwrap();
        socket
            .set_read_timeout(Some(Duration::from_secs(3)))
            .unwrap();
        write!(socket,"{method} {path} HTTP/1.1\r\nHost: example.test\r\nConnection: {connection}, close\r\nUpgrade: websocket\r\nSec-WebSocket-Version: {version}\r\nSec-WebSocket-Key: {key}\r\n{extra}\r\n").unwrap();
        let mut reader = BufReader::new(socket);
        let response = headers(&mut reader);
        assert!(response.starts_with("HTTP/1.1 400"), "{response}");
        let mut error = String::new();
        reader.read_to_string(&mut error).unwrap();
        assert!(error.contains("WebSocket"), "{error}");
    }
    hosted.stop();
}

#[test]
fn constructed_requests_clones_and_responses_from_other_requests_cannot_upgrade() {
    let output = common::run(
        r#"
        import {serve,upgradeWebSocket} from 'avril:std';
        try {upgradeWebSocket(new Request('http://example.test'));throw Error('constructed request escaped');}
        catch(error){if(!(error instanceof TypeError))throw error;}
        let saved;
        const server=serve({port:0},request=>{
            if(new URL(request.url).pathname==='/reuse') return saved;
            try {upgradeWebSocket(request.clone());throw Error('cloned request escaped');}
            catch(error){if(!(error instanceof TypeError))throw error;}
            saved=upgradeWebSocket(request).response;
            try {upgradeWebSocket(request);throw Error('duplicate upgrade');}
            catch(error){if(error.message==='duplicate upgrade')throw error;}
            return new Response('abandoned');
        });
        const client=new WebSocket(`ws://127.0.0.1:${server.addr.port}`);
        await new Promise(resolve=>client.onclose=resolve);
        const response=await fetch(`http://127.0.0.1:${server.addr.port}/reuse`);
        if(response.status!==500)throw Error('unrelated request upgraded');
        await response.text();await server.shutdown();console.log('guarded');
    "#,
        &[],
    );
    assert_eq!(output.stdout, b"guarded\n");
}

#[test]
#[cfg(unix)]
fn workers_do_not_inherit_cli_listener_authority_or_private_bootstrap_access() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("worker.js");
    std::fs::write(
        &path,
        r#"
        import {serve} from 'avril:std';
        try {serve({port:0},()=>new Response());postMessage('escaped');}
        catch(error){postMessage(error.message);}
        close();
    "#,
    )
    .unwrap();
    let worker = url::Url::from_file_path(&path).unwrap();
    let grant = format!("--allow-js-exec={}", path.display());
    let hosted = Hosted::start(
        &format!(
            r#"
        const std=await import('avril:std');
        if('startCLI' in std)throw Error('bootstrap exported');
        try {{await import('ext:http/main.js');throw Error('private module escaped');}}
        catch(error){{if(!error.message.includes('private'))throw error;}}
        const worker=new Worker('{worker}',{{type:'module'}});
        const result=await new Promise(resolve=>worker.onmessage=e=>resolve(e.data));
        worker.terminate();
        export default {{fetch(){{return new Response(result);}}}};
    "#
        ),
        &["--port=0", &grant],
    );
    let url = url::Url::parse(&hosted.address).unwrap();
    let mut socket = TcpStream::connect((url.host_str().unwrap(), url.port().unwrap())).unwrap();
    socket
        .set_read_timeout(Some(Duration::from_secs(3)))
        .unwrap();
    socket
        .write_all(b"GET / HTTP/1.1\r\nHost: example.test\r\nConnection: close\r\n\r\n")
        .unwrap();
    let mut reader = BufReader::new(socket);
    assert!(headers(&mut reader).starts_with("HTTP/1.1 200"));
    let mut body = String::new();
    reader.read_to_string(&mut body).unwrap();
    assert!(body.contains("PermissionDenied"), "{body}");
    hosted.stop();
}

#[test]
fn aborting_during_option_conversion_prevents_binding() {
    let output = common::run(
        r#"
        import {serve,pledge} from 'avril:std';
        pledge({net:false});
        const controller=new AbortController();
        try {
            serve({port:0,signal:controller.signal,get path(){controller.abort('late');return undefined;}},()=>new Response());
            throw Error('aborted listener bound');
        } catch(error){if(error!=='late')throw error;}
        console.log('aborted');
    "#,
        &[],
    );
    assert_eq!(output.stdout, b"aborted\n");
}

#[test]
#[cfg(unix)]
fn sigterm_cleans_up_cli_unix_listeners() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("terminate.sock");
    let option = format!("--unix={}", path.display());
    let hosted = Hosted::start(
        "export default {fetch(){return new Response('ok');}};",
        &[&option],
    );
    assert!(path.exists());
    hosted.stop_with(libc::SIGTERM);
    assert!(!path.exists());
}

#[test]
#[cfg(unix)]
fn a_second_shutdown_signal_interrupts_cpu_bound_handlers() {
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("busy.sock");
    let option = format!("--unix={}", path.display());
    let mut hosted = Hosted::start(
        "export default {fetch(){console.log('busy');while(true){}}};",
        &[&option],
    );
    let mut socket = std::os::unix::net::UnixStream::connect(&path).unwrap();
    socket
        .write_all(b"GET / HTTP/1.1\r\nHost: example.test\r\n\r\n")
        .unwrap();
    let child = hosted.child.as_mut().unwrap();
    let mut stdout = BufReader::new(child.stdout.take().unwrap());
    let mut line = String::new();
    stdout.read_line(&mut line).unwrap();
    assert_eq!(line, "busy\n");
    child.stdout = Some(stdout.into_inner());
    assert_eq!(unsafe { libc::kill(child.id() as i32, libc::SIGINT) }, 0);
    let deadline = std::time::Instant::now() + Duration::from_secs(2);
    while path.exists() && std::time::Instant::now() < deadline {
        std::thread::sleep(Duration::from_millis(5));
    }
    assert!(!path.exists());
    assert_eq!(unsafe { libc::kill(child.id() as i32, libc::SIGINT) }, 0);
    let output = common::wait(hosted.child.take().unwrap());
    assert!(
        output.status.code().is_some(),
        "forced shutdown needed an external kill"
    );
}

#[test]
#[cfg(target_os = "linux")]
fn server_recovers_after_accept_exhausts_file_descriptors() {
    use std::os::unix::process::CommandExt;
    use std::time::Instant;

    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("main.js");
    std::fs::write(
        &path,
        "import {serve} from 'avril:std'; const server=serve({port:0},()=>{server.shutdown();return new Response('recovered');}); console.log(server.addr.port); await server.finished;",
    )
    .unwrap();
    let mut command = Command::new(env!("CARGO_BIN_EXE_avril"));
    command
        .arg("run")
        .arg(path)
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    unsafe {
        command.pre_exec(|| {
            let limit = libc::rlimit { rlim_cur: 64, rlim_max: 64 };
            if libc::setrlimit(libc::RLIMIT_NOFILE, &limit) != 0 {
                return Err(std::io::Error::last_os_error());
            }
            Ok(())
        });
    }
    let child = command.spawn().unwrap();
    let mut hosted = Hosted {
        _dir: dir,
        child: Some(child),
        address: String::new(),
    };
    let child = hosted.child.as_mut().unwrap();
    let mut stdout = BufReader::new(child.stdout.take().unwrap());
    let mut line = String::new();
    stdout.read_line(&mut line).unwrap();
    let port: u16 = line.trim().parse().expect(&line);
    child.stdout = Some(stdout.into_inner());
    let fd_path = format!("/proc/{}/fd", child.id());
    let mut sockets = Vec::new();
    for _ in 0..80 {
        sockets.push(
            TcpStream::connect_timeout(&([127, 0, 0, 1], port).into(), Duration::from_secs(2))
                .unwrap(),
        );
    }
    let deadline = Instant::now() + Duration::from_secs(2);
    while std::fs::read_dir(&fd_path).unwrap().count() < 64 {
        assert!(
            Instant::now() < deadline,
            "server did not exhaust descriptors"
        );
        std::thread::sleep(Duration::from_millis(10));
    }
    // Leave an accept pending while the process has no free descriptors.
    std::thread::sleep(Duration::from_millis(200));
    drop(sockets);
    let mut socket = TcpStream::connect(("127.0.0.1", port)).unwrap();
    socket
        .set_read_timeout(Some(Duration::from_secs(3)))
        .unwrap();
    socket
        .write_all(b"GET / HTTP/1.1\r\nHost: example.test\r\nConnection: close\r\n\r\n")
        .unwrap();
    let mut response = String::new();
    socket.read_to_string(&mut response).unwrap();
    assert!(response.starts_with("HTTP/1.1 200"), "{response}");
    assert!(response.ends_with("recovered"), "{response}");
    let output = common::wait(hosted.child.take().unwrap());
    assert!(
        output.status.success(),
        "{}",
        String::from_utf8_lossy(&output.stderr)
    );
}

#[test]
fn programmatic_ipv6_loopback_address_can_be_reused() {
    let Ok(_probe) = std::net::TcpListener::bind("[::1]:0") else {
        return;
    };
    let output = common::run(
        r#"
        import {serve,pledge} from 'avril:std';
        pledge({net:{allow:['[::1]']}});
        let server=serve({hostname:'::1',port:0},()=>new Response('ipv6'));
        if(server.addr.hostname!=='::1')throw Error('hostname');
        const addr=server.addr;
        await server.shutdown();
        server=serve(addr,()=>new Response('reused'));
        const response=await fetch(`http://[${addr.hostname}]:${addr.port}/`);
        if(await response.text()!=='reused')throw Error('response');
        await server.shutdown();console.log('ipv6');
    "#,
        &[],
    );
    assert_eq!(output.stdout, b"ipv6\n");
}