mod common;

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

#[test]
fn request_response_streaming_and_concurrent_handlers() {
    let output = common::run(
        r#"
        import {serve} from 'avril:std';
        let unblock;
        const blocked = new Promise(resolve => unblock = resolve);
        const server = serve({port:0}, async (request, info) => {
            if (info.remoteAddr.transport !== 'tcp' || info.remoteAddr.hostname !== '127.0.0.1') throw Error('remote address');
            if (request.url.endsWith('/release')) { unblock(); return new Response('released'); }
            await blocked;
            if (request.method !== 'POST' || request.headers.get('x-test') !== 'request') throw Error('request metadata');
            try { request.headers.set('x-test','changed'); throw Error('mutable'); }
            catch (error) { if (error.message === 'mutable') throw error; }
            const clone = request.clone();
            if (await clone.text() !== 'firstsecond') throw Error('request clone');
            return new Response(request.body, {status:201, headers:{'x-test':'response'}});
        });
        const url = `http://127.0.0.1:${server.addr.port}`;
        const upload = new ReadableStream({async start(c) {
            c.enqueue(new TextEncoder().encode('first'));
            await new Promise(resolve => setTimeout(resolve,10));
            c.enqueue(new TextEncoder().encode('second')); c.close();
        }});
        const pending = fetch(url+'/blocked', {method:'POST', headers:{'x-test':'request'}, body:upload, duplex:'half'});
        if (await (await fetch(url+'/release')).text() !== 'released') throw Error('concurrency');
        const response = await pending;
        if (response.status !== 201 || response.headers.get('x-test') !== 'response' || await response.text() !== 'firstsecond') throw Error('streaming response');
        await server.shutdown(); await server.finished;
        console.log('streaming');
        "#,
        &[],
    );
    assert_eq!(output.stdout, b"streaming\n");
}

#[test]
fn full_duplex_does_not_buffer_the_request_or_response() {
    let output = common::run(
        r#"
        import {serve} from 'avril:std';
        const server = serve({port:0}, request => new Response(request.body));
        let upload;
        const body = new ReadableStream({start(c) {upload=c;}});
        const pending = fetch(`http://127.0.0.1:${server.addr.port}`, {method:'POST',body,duplex:'half'});
        upload.enqueue(new Uint8Array([1,2,3]));
        const response = await pending, reader = response.body.getReader();
        const first = await reader.read();
        if (first.done || first.value.join() !== '1,2,3') throw Error('first chunk');
        upload.enqueue(new Uint8Array([4,5])); upload.close();
        const second = await reader.read();
        if (second.done || second.value.join() !== '4,5' || !(await reader.read()).done) throw Error('second chunk');
        reader.releaseLock(); await server.shutdown(); console.log('duplex');
        "#,
        &[],
    );
    assert_eq!(output.stdout, b"duplex\n");
}

#[test]
fn graceful_shutdown_waits_for_active_responses() {
    let output = common::run(
        r#"
        import {serve} from 'avril:std';
        let release, started;
        const entered = new Promise(resolve => started=resolve);
        const gate = new Promise(resolve => release=resolve);
        const server = serve({port:0}, async request => {
            started(); await gate;
            if (request.signal.aborted) throw Error('shutdown aborted request');
            return new Response('finished');
        });
        const pending = fetch(`http://127.0.0.1:${server.addr.port}`);
        await entered;
        let done = false;
        const closing = server.shutdown(); closing.then(() => done=true);
        if (server.shutdown() !== closing) throw Error('shutdown is not idempotent');
        await new Promise(resolve => setTimeout(resolve,20));
        if (done) throw Error('shutdown finished before response');
        release(); if (await (await pending).text() !== 'finished') throw Error('lost response');
        await closing;
        try { await fetch(`http://127.0.0.1:${server.addr.port}`); throw Error('listener still open'); }
        catch (error) { if (error.message === 'listener still open') throw error; }
        console.log('shutdown');
        "#,
        &[],
    );
    assert_eq!(output.stdout, b"shutdown\n");
}

#[test]
fn abort_closes_connections_and_signals_handlers() {
    let output = common::run(
        r#"
        import {serve} from 'avril:std';
        const control = new AbortController();
        let entered, aborted;
        const start = new Promise(resolve => entered=resolve);
        const abort = new Promise(resolve => aborted=resolve);
        const server = serve({port:0,signal:control.signal}, async request => {
            request.signal.addEventListener('abort', () => aborted(request.signal.reason.name), {once:true});
            entered(); await abort; request.signal.throwIfAborted(); return new Response('too late');
        });
        const pending = fetch(`http://127.0.0.1:${server.addr.port}`);
        const result = pending.then(() => {throw Error('response after abort');}, () => {});
        await start; control.abort();
        if (await abort !== 'AbortError') throw Error('request signal');
        await result; await server.finished; await server.shutdown();
        console.log('abort');
        "#,
        &[],
    );
    assert_eq!(output.stdout, b"abort\n");
    assert!(
        output.stderr.is_empty(),
        "{}",
        String::from_utf8_lossy(&output.stderr)
    );
}

#[test]
fn handler_errors_invalid_responses_and_null_body_statuses() {
    let output = common::run(
        r#"
        import {serve} from 'avril:std';
        let canceled=false;
        const server = serve({port:0}, request => {
            const path = new URL(request.url).pathname;
            if (path === '/throw') throw Error('broken');
            if (path === '/invalid') return 'not a Response';
            if (path === '/error') return Response.error();
            if (path === '/used') { const r = new Response('used'); r.text(); return r; }
            if (path === '/head') return new Response(new ReadableStream({cancel(){canceled=true;}}));
            if (path === '/204') return new Response(null,{status:204});
            if (path === '/205') return new Response(null,{status:205});
            if (path === '/304') return new Response(null,{status:304});
            return new Response('healthy');
        });
        const url=`http://127.0.0.1:${server.addr.port}`;
        for (const path of ['/throw','/invalid','/error','/used']) {
            const r=await fetch(url+path);
            if (r.status!==500 || await r.text()!=='Internal Server Error') throw Error(path);
        }
        for (const status of [204,205,304]) {
            const r=await fetch(url+'/'+status,{redirect:'manual'});
            if (r.status!==status || r.body!==null) throw Error('null body');
        }
        const head=await fetch(url+'/head',{method:'HEAD'});
        if (head.body!==null || !canceled) throw Error('HEAD');
        if (await (await fetch(url)).text()!=='healthy') throw Error('server did not recover');
        await server.shutdown();
        const custom=serve({port:0,onError: error => new Response(error.message,{status:503})}, () => {throw Error('custom');});
        const r=await fetch(`http://127.0.0.1:${custom.addr.port}`);
        if (r.status!==503 || await r.text()!=='custom') throw Error('onError');
        await custom.shutdown();
        const broken=serve({port:0,onError: () => {throw Error('also broken');}}, () => {throw Error('broken');});
        if ((await fetch(`http://127.0.0.1:${broken.addr.port}`)).status!==500) throw Error('onError fallback');
        await broken.shutdown(); console.log('responses');
        "#,
        &[],
    );
    assert_eq!(output.stdout, b"responses\n");
}

#[test]
fn listen_validation_permissions_and_callback_cleanup() {
    let output = common::run(
        r#"
        import {serve,pledge} from 'avril:std';
        for (const port of [-1,65536,1.5,NaN,Infinity]) {
            try { serve({port}, () => new Response()); throw Error('accepted invalid port'); }
            catch(e) {if (!(e instanceof RangeError)) throw e;}
        }
        for (const options of [{onError:1},{onListen:1},{signal:{}}]) {
            try { serve(options, () => new Response()); throw Error('accepted invalid options'); }
            catch(e) {if (!(e instanceof TypeError)) throw e;}
        }
        const stopped = new AbortController(); stopped.abort('already aborted');
        try {serve({port:0,signal:stopped.signal}, () => new Response()); throw Error('already aborted');}
        catch(e) {if(e!=='already aborted') throw e;}
        let addr;
        try {serve({port:0,onListen:a=>{addr=a;throw Error('onListen');}}, () => new Response());}
        catch(e) {if(e.message!=='onListen') throw e;}
        await new Promise(resolve => setTimeout(resolve,20));
        const rebound=serve({port:addr.port}, () => new Response());
        if (!Object.isFrozen(rebound.addr) || rebound.addr.hostname!=='127.0.0.1') throw Error('address');
        await rebound.shutdown();
        pledge({net:false});
        try {serve({port:0}, () => new Response()); throw Error('missing permission');}
        catch(e) {if(!e.message.includes('PermissionDenied')) throw e;}
        console.log('permissions');
        "#,
        &[],
    );
    assert_eq!(output.stdout, b"permissions\n");
}

#[test]
fn raw_http_keep_alive_preserves_host_headers_and_cookies() {
    let (_dir, mut child) = common::spawn(
        r#"
        import {serve} from 'avril:std';
        let count=0;
        const server=serve({port:0,onListen:addr=>console.log(addr.port)}, request => {
            if (request.url!=='http://example.test:4321/raw?q=1' || request.headers.get('host')!=='example.test:4321' || request.headers.get('x-test')!=='one, two') throw Error('wire request');
            if (++count===2) server.shutdown();
            return new Response('wire',{headers:[['set-cookie','a=1'],['set-cookie','b=2'],['x-latin','\u00e9']]});
        });
        await server.finished;
        "#,
    );
    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().unwrap();
    child.stdout = Some(stdout.into_inner());
    let mut socket = TcpStream::connect(("127.0.0.1", port)).unwrap();
    socket
        .set_read_timeout(Some(Duration::from_secs(3)))
        .unwrap();
    let request =
        b"GET /raw?q=1 HTTP/1.1\r\nHost: example.test:4321\r\nX-Test: one\r\nX-Test: two\r\n\r\n";
    let mut reader = BufReader::new(socket.try_clone().unwrap());
    for _ in 0..2 {
        socket.write_all(request).unwrap();
        let mut response = Vec::new();
        loop {
            let mut line = Vec::new();
            reader.read_until(b'\n', &mut line).unwrap();
            assert!(!line.is_empty(), "unexpected EOF");
            response.extend_from_slice(&line);
            if line == b"\r\n" {
                break;
            }
        }
        let headers = String::from_utf8_lossy(&response).to_ascii_lowercase();
        assert!(headers.starts_with("http/1.1 200 ok\r\n"), "{headers}");
        assert!(headers.contains("set-cookie: a=1\r\n"));
        assert!(headers.contains("set-cookie: b=2\r\n"));
        assert!(
            response
                .windows(12)
                .any(|bytes| bytes == b"x-latin: \xe9\r\n")
        );
        let mut body = [0; 4];
        reader.read_exact(&mut body).unwrap();
        assert_eq!(&body, b"wire");
    }
    let output = common::wait(child);
    assert!(
        output.status.success(),
        "{}",
        String::from_utf8_lossy(&output.stderr)
    );
}

#[test]
fn request_signal_follows_client_disconnect() {
    let output = common::run(
        r#"
        import {serve} from 'avril:std';
        let entered, disconnected;
        const start=new Promise(resolve=>entered=resolve);
        const gone=new Promise(resolve=>disconnected=resolve);
        const server=serve({port:0}, async request=>{
            request.signal.addEventListener('abort',disconnected,{once:true});
            entered(); await gone; request.signal.throwIfAborted(); return new Response('late');
        });
        const client=new AbortController();
        const pending=fetch(`http://127.0.0.1:${server.addr.port}`,{signal:client.signal});
        const settled=pending.catch(()=>{});
        await start; client.abort(); await gone; await settled;
        await server.shutdown(); console.log('disconnect');
        "#,
        &[],
    );
    assert_eq!(output.stdout, b"disconnect\n");
    assert!(
        output.stderr.is_empty(),
        "{}",
        String::from_utf8_lossy(&output.stderr)
    );
}

#[test]
fn server_in_worker_uses_worker_capabilities() {
    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(e) {postMessage(e.message);}
        close();
    "#,
    )
    .unwrap();
    let url = url::Url::from_file_path(path).unwrap();
    let output = common::run(
        &format!(
            r#"
        import {{avril}} from 'avril:std';
        const worker=new Worker('{url}',{{type:'module',[avril]:{{caps:{{net:false}}}}}});
        const result=await new Promise(resolve=>worker.onmessage=e=>resolve(e.data));
        if(!result.includes('PermissionDenied')) throw Error(result);
        worker.terminate(); console.log('worker');
        "#
        ),
        &[],
    );
    assert_eq!(output.stdout, b"worker\n");
}

#[test]
fn handler_failures_report_uncaught_stacks_without_stopping_the_server() {
    let output = common::run(
        r#"
        import {serve} from 'avril:std';
        const errors=[],stacks=[];
        addEventListener('error',event=>{
            if(event.error!==errors[stacks.length]||event.message!==event.error.message||!event.cancelable)throw Error('error event');
            stacks.push(event.error.stack);
        });
        console.error=()=>{throw Error('guest console');};
        function throwingHandler(request) {
            const path=new URL(request.url).pathname;
            if(path==='/healthy')return new Response('healthy');
            const error=new Error(path==='/throw'?'handler failed':'handler rejected');
            errors.push(error);
            if(path==='/throw')throw error;
            return Promise.reject(error);
        }
        const server=serve({port:0},throwingHandler),url=`http://127.0.0.1:${server.addr.port}`;
        for(const path of ['/throw','/reject']) {
            const response=await fetch(url+path);
            if(response.status!==500||await response.text()!=='Internal Server Error')throw Error('fallback');
            if(await (await fetch(url+'/healthy')).text()!=='healthy')throw Error('server stopped');
        }
        if(stacks.length!==2)throw Error('duplicate or missing reports');
        await server.shutdown();console.log(JSON.stringify(stacks));
    "#,
        &[],
    );
    let stacks: Vec<String> = serde_json::from_slice(&output.stdout).unwrap();
    assert!(stacks[0].starts_with("Error: handler failed\n    at throwingHandler"));
    assert!(stacks[1].starts_with("Error: handler rejected\n    at throwingHandler"));
    assert_eq!(
        String::from_utf8(output.stderr).unwrap(),
        format!("{}\n", stacks.join("\n"))
    );
}

#[test]
fn canceled_http_error_events_suppress_stderr_and_on_error_handles_original_failures() {
    let output = common::run(
        r#"
        import {serve} from 'avril:std';
        const errors=[],failure=Error('handler failure');
        addEventListener('error',event=>{errors.push(event.error);event.preventDefault();});
        const server=serve({port:0},async request=>{
            const path=new URL(request.url).pathname;
            if(path==='/throw')throw failure;
            if(path==='/invalid')return null;
            const used=new Response('used');await used.text();return used;
        });
        for(const path of ['/throw','/invalid','/used']) {
            const response=await fetch(`http://127.0.0.1:${server.addr.port}${path}`);
            if(response.status!==500||await response.text()!=='Internal Server Error')throw Error('fallback');
        }
        if(errors.length!==3||errors[0]!==failure||!(errors[1] instanceof TypeError)||!(errors[2] instanceof TypeError))throw Error('handler reports');
        await server.shutdown();
        const recovered=serve({port:0,onError:error=>new Response(error.message,{status:503})},()=>{throw failure;});
        const custom=await fetch(`http://127.0.0.1:${recovered.addr.port}`);
        if(custom.status!==503||await custom.text()!==failure.message||errors.length!==3)throw Error('handled failure reported');
        await recovered.shutdown();
        for(const onError of [()=>{throw failure;},()=>null]) {
            const broken=serve({port:0,onError},()=>{throw failure;});
            const response=await fetch(`http://127.0.0.1:${broken.addr.port}`);
            if(response.status!==500||await response.text()!=='Internal Server Error')throw Error('onError fallback');
            await broken.shutdown();
        }
        if(errors.length!==5||errors[3]!==failure||!(errors[4] instanceof TypeError))throw Error('onError reports');
        console.log('canceled');
    "#,
        &[],
    );
    assert_eq!(output.stdout, b"canceled\n");
    assert!(
        output.stderr.is_empty(),
        "{}",
        String::from_utf8_lossy(&output.stderr)
    );
}

#[test]
fn on_error_and_response_stream_failures_are_reported_once_nonfatally() {
    let output = common::run(
        r#"
        import {serve} from 'avril:std';
        const reports=[];
        addEventListener('error',event=>reports.push(event.message));
        const broken=serve({port:0,onError:()=>{throw Error('onError failed');}},()=>{throw Error('handled original');});
        const fallback=await fetch(`http://127.0.0.1:${broken.addr.port}`);
        if(fallback.status!==500||await fallback.text()!=='Internal Server Error')throw Error('onError fallback');
        await broken.shutdown();
        let fail;
        const server=serve({port:0},request=>{
            if(new URL(request.url).pathname==='/healthy')return new Response('healthy');
            const gate=new Promise(resolve=>fail=resolve);
            return new Response(new ReadableStream({async start(c){
                c.enqueue(new Uint8Array([7]));await gate;c.error(Error('response stream failed'));
            }}));
        });
        const url=`http://127.0.0.1:${server.addr.port}`,response=await fetch(url),reader=response.body.getReader();
        if((await reader.read()).value[0]!==7)throw Error('first chunk');
        fail();
        try{await reader.read();throw Error('stream succeeded');}
        catch(error){if(error.message==='stream succeeded')throw error;}
        reader.releaseLock();
        if(await (await fetch(url+'/healthy')).text()!=='healthy')throw Error('server stopped');
        await server.shutdown();
        if(reports.join('|')!=='onError failed|response stream failed')throw Error('reports');
        console.log('reported');
    "#,
        &[],
    );
    assert_eq!(output.stdout, b"reported\n");
    let stderr = String::from_utf8(output.stderr).unwrap();
    assert_eq!(
        stderr.matches("Error: onError failed\n").count(),
        1,
        "{stderr}"
    );
    assert_eq!(
        stderr.matches("Error: response stream failed\n").count(),
        1,
        "{stderr}"
    );
    assert!(stderr.contains("\n    at "), "{stderr}");
    assert!(!stderr.contains("handled original"), "{stderr}");
}

#[test]
fn hostile_stack_getters_do_not_turn_http_reporting_into_a_failure() {
    let output = common::run(
        r#"
        import {serve} from 'avril:std';
        const failure=Error('original failure');
        Object.defineProperty(failure,'stack',{get(){throw Error('stack getter failed');}});
        const server=serve({port:0},request=>{
            if(new URL(request.url).pathname==='/healthy')return new Response('healthy');
            throw failure;
        });
        const url=`http://127.0.0.1:${server.addr.port}`,response=await fetch(url);
        if(response.status!==500||await response.text()!=='Internal Server Error')throw Error('fallback');
        if(await (await fetch(url+'/healthy')).text()!=='healthy')throw Error('server stopped');
        await server.shutdown();console.log('safe reporting');
    "#,
        &[],
    );
    assert_eq!(output.stdout, b"safe reporting\n");
    assert_eq!(output.stderr, b"Error: original failure\n");
}

#[test]
fn option_conversion_callback_receivers_and_frozen_server_metadata() {
    let output = common::run(
        r#"
        import {serve, serveFile, upgradeWebSocket} from 'avril:std';
        if (serve.length !== 1 || serveFile.length !== 2 || upgradeWebSocket.length !== 1) throw Error('arity');
        const order = [], failure = {};
        let hostnameReads = 0, portReads = 0, listened;
        const options = {
            get hostname() {
                order.push('hostname');
                if (++hostnameReads === 1) return Symbol('presence only');
                return {toString() {order.push('hostname:string'); return '127.0.0.1';}};
            },
            get port() {
                order.push('port');
                if (++portReads === 1) return Symbol('presence only');
                return {valueOf() {order.push('port:number'); return 0;}};
            },
            get signal() {order.push('signal'); return undefined;},
            get onError() {
                order.push('onError');
                return function(error) {
                    if (this !== undefined || error !== failure) throw Error('onError receiver or reason');
                    order.push('recover');
                    return {then(resolve) {order.push('recover:then'); resolve(new Response('recovered', {status:202}));}};
                };
            },
            get onListen() {
                order.push('onListen');
                return function(addr) {
                    if (this !== undefined) throw Error('onListen receiver');
                    order.push('listened'); listened = addr;
                };
            },
            get path() {order.push('path'); return undefined;}
        };
        const server = serve(options, function(request, info) {
            if (this !== undefined || request.method !== 'GET') throw Error('handler receiver');
            for (const value of [info, info.remoteAddr]) {
                if (!Object.isFrozen(value) || Object.getPrototypeOf(value) !== null) throw Error('mutable metadata');
            }
            if (info.remoteAddr.transport !== 'tcp' || info.remoteAddr.hostname !== '127.0.0.1') throw Error('peer');
            order.push('handler');
            return {then(resolve, reject) {order.push('handler:then'); reject(failure);}};
        });
        const prefix = 'hostname|hostname|hostname:string|port|port|port:number|signal|onError|onListen|path|listened';
        if (order.join('|') !== prefix) throw Error('conversion order: '+order);
        if (server.addr !== listened) throw Error('listen address identity');
        for (const value of [server, server.addr]) {
            if (!Object.isFrozen(value) || Object.getPrototypeOf(value) !== null) throw Error('mutable server');
        }
        const response = await fetch(`http://127.0.0.1:${server.addr.port}`);
        if (response.status !== 202 || await response.text() !== 'recovered') throw Error('thenable recovery');
        if (order.join('|') !== prefix+'|handler|handler:then|recover|recover:then') throw Error('callback order');
        if (server.shutdown() !== server.finished) throw Error('completion identity');
        await server.finished;
        console.log('server shape and conversion');
        "#,
        &[],
    );
    assert_eq!(output.stdout, b"server shape and conversion\n");
    assert!(
        output.stderr.is_empty(),
        "{}",
        String::from_utf8_lossy(&output.stderr)
    );
}