mod common;

use std::io::{BufRead, Write};
use std::net::TcpListener;
use std::thread;
use std::time::Duration;

use common::run;

#[test]
fn wasm_streaming_compiles_and_instantiates_incrementally() {
    let output = run(
        r#"
        const bytes = new Uint8Array([0, 97, 115, 109, 1, 0, 0, 0]);
        function response(body = bytes, headers = {'content-type': 'application/wasm'}, status = 200) {
            return new Response(body, {headers, status});
        }
        const chunks = new ReadableStream({
            start(controller) { this.controller = controller; this.index = 0; },
            pull(controller) {
                if (this.index === bytes.length) controller.close();
                else controller.enqueue(bytes.slice(this.index, ++this.index));
            },
        });
        const streamed = response(chunks, {'content-type': 'application/wasm; charset=binary'});
        const module = await WebAssembly.compileStreaming(Promise.resolve(streamed), {});
        if (!(module instanceof WebAssembly.Module) || !streamed.bodyUsed) throw Error('Streaming compile');
        let controller, settled = false;
        const pendingResponse = response(new ReadableStream({start(c) { controller = c; }}));
        const pending = WebAssembly.compileStreaming(pendingResponse).then(value => { settled = true; return value; });
        controller.enqueue(bytes);
        for (let i = 0; i < 10 && !pendingResponse.bodyUsed; i++) await Promise.resolve();
        if (settled || !pendingResponse.bodyUsed) throw Error('Streaming waited for body aggregation');
        controller.close();
        if (!(await pending instanceof WebAssembly.Module)) throw Error('Incremental completion');
        const instantiatedResponse = response();
        const result = await WebAssembly.instantiateStreaming(instantiatedResponse, {}, {});
        if (!(result.module instanceof WebAssembly.Module) || !(result.instance instanceof WebAssembly.Instance)) {
            throw Error('Streaming instantiate result');
        }
        if (WebAssembly.compileStreaming.length !== 1 || WebAssembly.instantiateStreaming.length !== 2) {
            throw Error('Streaming function lengths');
        }
        for (const fn of [WebAssembly.compileStreaming, WebAssembly.instantiateStreaming]) {
            try { new fn(); throw Error('Constructible streaming method'); }
            catch (error) { if (!(error instanceof TypeError)) throw error; }
        }
        if (WebAssembly.compileStreaming.name !== 'compileStreaming' || WebAssembly.instantiateStreaming.name !== 'instantiateStreaming') {
            throw Error('Streaming function names');
        }
        for (const storage of [
            new ArrayBuffer(12),
            new ArrayBuffer(12, {maxByteLength: 24}),
            new SharedArrayBuffer(12),
            new SharedArrayBuffer(12, {maxByteLength: 24}),
        ]) {
            const view = new Uint8Array(storage, 2, bytes.length);
            view.set(bytes);
            const body = new ReadableStream({
                start(c) {c.enqueue(view); c.close();}
            });
            if (!(await WebAssembly.compileStreaming(response(body)) instanceof WebAssembly.Module)) {
                throw Error('Buffer storage and view bounds');
            }
        }
        console.log('wasm stream success');
        "#,
        &[],
    );
    assert_eq!(output.stdout, b"wasm stream success\n");
}

#[test]
fn wasm_streaming_consumes_chunked_http_responses_incrementally() {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let url = format!("http://{}", listener.local_addr().unwrap());
    let server = thread::spawn(move || {
        let (mut socket, _) = listener.accept().unwrap();
        let mut request = std::io::BufReader::new(socket.try_clone().unwrap());
        let mut line = String::new();
        request.read_line(&mut line).unwrap();
        loop {
            line.clear();
            request.read_line(&mut line).unwrap();
            if line == "\r\n" {
                break;
            }
        }
        socket.write_all(b"HTTP/1.1 200 OK\r\nContent-Type: application/wasm\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n4\r\n\x00asm\r\n").unwrap();
        socket.flush().unwrap();
        thread::sleep(Duration::from_millis(100));
        socket
            .write_all(b"4\r\n\x01\x00\x00\x00\r\n0\r\n\r\n")
            .unwrap();
    });
    let source = format!(
        r#"
        const response = await fetch('{url}');
        let settled = false;
        const compiling = WebAssembly.compileStreaming(response).then(module => {{ settled = true; return module; }});
        for (let i = 0; i < 100 && !response.bodyUsed; i++) await new Promise(resolve => setTimeout(resolve, 2));
        if (!response.bodyUsed || settled) throw Error('HTTP body was buffered before streaming');
        if (!(await compiling instanceof WebAssembly.Module)) throw Error('Chunked HTTP module');
        console.log('wasm http stream');
        "#
    );
    assert_eq!(run(&source, &[]).stdout, b"wasm http stream\n");
    server.join().unwrap();
}

#[test]
fn wasm_streaming_validates_responses_and_preserves_body_errors() {
    let output = run(
        r#"
        const bytes = new Uint8Array([0, 97, 115, 109, 1, 0, 0, 0]);
        const importingModule = new Uint8Array([0, 97, 115, 109, 1, 0, 0, 0, 1, 4, 1, 96, 0, 0, 2, 7, 1, 1, 109, 1, 102, 0, 0]);
        async function rejects(promise, Type, expected) {
            try { await promise; throw Error('Expected rejection'); }
            catch (error) {
                if (Type !== null && !(error instanceof Type)) throw error;
                if (expected !== undefined && error !== expected) throw Error('Rejection identity changed');
            }
        }
        async function rejectsBeforeSource(promise, expected) {
            let settled = false, error;
            promise.then(() => { settled = true; }, reason => { settled = true; error = reason; });
            for (let i = 0; i < 20 && !settled; i++) await Promise.resolve();
            if (!settled) throw Error('Native argument validation waited for the source');
            if (expected === undefined ? !(error instanceof TypeError) : error !== expected) throw error;
        }
        const never = new Promise(() => {});
        await rejectsBeforeSource(WebAssembly.instantiateStreaming(never, 42));
        await rejectsBeforeSource(WebAssembly.instantiateStreaming(Promise.reject('source failure'), 42));
        const optionsReason = {badOptions: true};
        await rejectsBeforeSource(WebAssembly.compileStreaming(never, {get builtins() { throw optionsReason; }}), optionsReason);
        await rejects(WebAssembly.compileStreaming({}), TypeError);
        await rejects(WebAssembly.compileStreaming(Promise.reject('source failure')), null, 'source failure');
        const wrongMime = new Response(bytes);
        await rejects(WebAssembly.compileStreaming(wrongMime), TypeError);
        if (wrongMime.bodyUsed) throw Error('Rejected MIME response body was consumed');
        await rejects(WebAssembly.compileStreaming(new Response(bytes, {headers: {'content-type': 'application/wasm'}, status: 404})), TypeError);
        const locked = new Response(bytes, {headers: {'content-type': 'application/wasm'}});
        locked.body.getReader();
        await rejects(WebAssembly.compileStreaming(locked), TypeError);
        const used = new Response(bytes, {headers: {'content-type': 'application/wasm'}});
        await used.arrayBuffer();
        await rejects(WebAssembly.compileStreaming(used), TypeError);
        const reason = {failure: true};
        const errored = new Response(new ReadableStream({start(controller) { controller.error(reason); }}), {
            headers: {'content-type': 'application/wasm'},
        });
        await rejects(WebAssembly.compileStreaming(errored), Object, reason);
        const undefinedError = new Response(new ReadableStream({start(controller) { controller.error(undefined); }}), {
            headers: {'content-type': 'application/wasm'},
        });
        let gotUndefined = false;
        await WebAssembly.compileStreaming(undefinedError).then(
            () => { throw Error('Accepted errored response'); },
            error => { gotUndefined = error === undefined; },
        );
        if (!gotUndefined) throw Error('Undefined stream error was not preserved');
        const invalidChunk = new Response(new ReadableStream({start(controller) { controller.enqueue('not bytes'); }}), {
            headers: {'content-type': 'application/wasm'},
        });
        await rejects(WebAssembly.compileStreaming(invalidChunk), TypeError);
        const truncated = new Response(bytes.subarray(0, 6), {headers: {'content-type': 'application/wasm'}});
        await rejects(WebAssembly.compileStreaming(truncated), WebAssembly.CompileError);
        const missingImport = new Response(importingModule, {headers: {'content-type': 'application/wasm'}});
        await rejects(WebAssembly.instantiateStreaming(missingImport, {m: {}}), WebAssembly.LinkError);
        const providedImport = new Response(importingModule, {headers: {'content-type': 'application/wasm'}});
        const imported = await WebAssembly.instantiateStreaming(providedImport, {m: {f() {}}});
        if (!(imported.instance instanceof WebAssembly.Instance)) throw Error('Streaming imports');
        console.log('wasm stream errors');
        "#,
        &[],
    );
    assert_eq!(output.stdout, b"wasm stream errors\n");
}

#[test]
fn ignored_streaming_compilation_failures_remain_unhandled() {
    let (_dir, child) = common::spawn(
        "WebAssembly.compileStreaming(new Response(new Uint8Array([0]),{headers:{'content-type':'application/wasm'}}));",
    );
    let out = common::wait(child);
    assert!(!out.status.success());
    assert!(String::from_utf8_lossy(&out.stderr).contains("CompileError"));
}