mod common;

use std::io::{BufRead, BufReader};

#[test]
fn tcp_streams_support_large_transfers_byob_and_half_close() {
    let output = common::run(
        r#"
        import {listen,connect,Conn,Listener} from 'avril:std';
        const listener=listen({port:0});
        if(!(listener instanceof Listener)||!Object.isFrozen(listener.addr))throw Error('listener');
        const accepting=listener.accept();
        const client=await connect({port:listener.addr.port});
        const server=await accepting;
        if(!(client instanceof Conn)||client.remoteAddr.port!==listener.addr.port||server.remoteAddr.port!==client.localAddr.port)throw Error('addresses');
        const payload=new Uint8Array(1024*1024+7);payload.fill(42);
        const consume=(async()=>{
            let count=0;const reader=server.readable.getReader({mode:'byob'});
            while(true){const {value,done}=await reader.read(new Uint8Array(8192));if(done)break;for(const byte of value)if(byte!==42)throw Error('corrupt');count+=value.length;}
            reader.releaseLock();if(count!==payload.length)throw Error('truncated');
            const writer=server.writable.getWriter();await writer.write(new Uint8Array([1,2,3]));await writer.close();writer.releaseLock();
        })();
        const writer=client.writable.getWriter();await writer.write(payload);await writer.close();writer.releaseLock();
        const reader=client.readable.getReader();let total=0;
        while(true){const {value,done}=await reader.read();if(done)break;for(const byte of value){if(byte!==++total)throw Error('reply');}}
        reader.releaseLock();if(total!==3)throw Error('reply length');
        await consume;client.close();server.close();listener.close();console.log('streams');
    "#,
        &[],
    );
    assert_eq!(output.stdout, b"streams\n");
}

#[test]
fn eof_does_not_close_the_write_half_and_writes_snapshot_buffers() {
    let output = common::run(
        r#"
        import {listen,connect,errors} from 'avril:std';
        const listener=listen({port:0}),accept=listener.accept();
        const client=await connect({port:listener.addr.port}),server=await accept;
        await server.closeWrite();await server.closeWrite();
        if(await client.read(new Uint8Array(4))!==null)throw Error('EOF');
        if(await client.read(new Uint8Array())!==0)throw Error('empty read');
        const bytes=new Uint8Array([7,8,9]);const written=client.write(bytes);bytes.fill(0);
        if(await written!==3)throw Error('write');await client.closeWrite();
        const buffer=new Uint8Array(10);let offset=0,count;
        while((count=await server.read(buffer.subarray(offset)))!==null)offset+=count;
        if(buffer.subarray(0,offset).join()!=='7,8,9')throw Error('snapshot');
        try{await client.write(new Uint8Array([1]));throw Error('write after FIN');}catch(e){if(!(e instanceof errors.BadResource))throw e;}
        client.close();server.close();listener.close();console.log('half');
    "#,
        &[],
    );
    assert_eq!(output.stdout, b"half\n");
}

#[test]
fn closing_cancels_accept_read_and_connect_and_disposal_is_idempotent() {
    let output = common::run(
        r#"
        import {listen,connect,errors} from 'avril:std';
        const listener=listen({port:0});const pending=listener.accept().catch(e=>e);listener.close();
        if(!((await pending) instanceof errors.BadResource))throw Error('accept cancellation');
        listener[Symbol.dispose]();
        const next=listen({port:0}),accept=next.accept(),client=await connect({port:next.addr.port}),server=await accept;
        const reading=client.read(new Uint8Array(1)).catch(e=>e);client.close();
        if(!((await reading) instanceof errors.BadResource))throw Error('read cancellation');
        client[Symbol.dispose]();server.close();next.close();
        const controller=new AbortController();controller.abort('stopped');
        try{await connect({port:1,signal:controller.signal});throw Error('aborted connect');}catch(e){if(e!=='stopped')throw e;}
        const active=new AbortController();const connecting=connect({port:1,signal:active.signal}).catch(e=>e);active.abort('cancel');
        if(await connecting!=='cancel')throw Error('pending connect');
        console.log('canceled');
    "#,
        &[],
    );
    assert_eq!(output.stdout, b"canceled\n");
}

#[test]
fn listener_iteration_closes_on_break_without_closing_accepted_connections() {
    let output = common::run(
        r#"
        import {listen,connect} from 'avril:std';
        const listener=listen({port:0});let server;
        const consuming=(async()=>{for await(const value of listener){server=value;break;}})();
        const client=await connect({port:listener.addr.port});await consuming;
        const receive=server.read(new Uint8Array(1));await client.write(new Uint8Array([8]));if(await receive!==1)throw Error('accepted connection closed');
        client.close();server.close();
        const iterator=listener[Symbol.asyncIterator]();if(!(await iterator.next()).done)throw Error('closed iteration');
        console.log('iterator');
    "#,
        &[],
    );
    assert_eq!(output.stdout, b"iterator\n");
}

#[test]
fn revocation_blocks_existing_connections_queued_writes_and_new_operations() {
    let output = common::run(
        r#"
        import {listen,connect,pledge,errors} from 'avril:std';
        const listener=listen({port:0}),accept=listener.accept(),client=await connect({port:listener.addr.port}),server=await accept;
        const written=client.write(new Uint8Array([1]));pledge({net:false});
        for(const operation of [()=>written,()=>client.read(new Uint8Array(1)),()=>client.closeWrite(),()=>listener.accept(),()=>connect({port:listener.addr.port}),()=>listen({port:0})]){
            try{await operation();throw Error('authority escaped');}catch(e){if(!(e instanceof errors.PermissionDenied))throw e;}
        }
        client.close();server.close();listener.close();console.log('denied');
    "#,
        &[],
    );
    assert_eq!(output.stdout, b"denied\n");
}

#[test]
fn raw_sockets_use_captured_intrinsics_and_are_not_cloneable() {
    let output = common::run(
        r#"
        import {listen,connect} from 'avril:std';
        const then=Promise.prototype.then,PromiseCtor=Promise;
        function observe(p){Object.defineProperty(p,'constructor',{value:undefined});const result=Reflect.apply(then,p,[v=>v,e=>{throw e;}]);Object.defineProperty(result,'constructor',{value:PromiseCtor});return result;}
        const chunk=new Uint8Array([9]);
        const poison=()=>{throw Error('guest hook');};
        Object.defineProperty(Promise.prototype,'constructor',{get:poison,configurable:true});Object.defineProperty(Promise,Symbol.species,{get:poison});Promise.prototype.then=poison;Array.prototype[Symbol.iterator]=poison;
        const listener=listen({port:0}),accept=listener.accept(),client=await observe(connect({port:listener.addr.port})),server=await observe(accept);
        for(let i=0;i<2;i++){try{structuredClone(i?client:listener);throw Error('cloned');}catch(e){if(e.name!=='DataCloneError')throw e;}}
        const read=server.read(new Uint8Array(1));await observe(client.write(chunk));if(await observe(read)!==1)throw Error('read');
        if(await observe(client.closeWrite())!==undefined)throw Error('closeWrite result');client.close();server.close();listener.close();console.log('safe');
    "#,
        &[],
    );
    assert_eq!(output.stdout, b"safe\n");
}

#[test]
fn validation_and_illegal_receivers_fail_without_opening_resources() {
    let output = common::run(
        r#"
        import {listen,connect,Conn,Listener} from 'avril:std';
        for(const port of [-1,65536,NaN,1.5]){try{listen({port});throw Error('invalid port');}catch(e){if(!(e instanceof RangeError))throw e;}}
        for(const ctor of [Conn,Listener]){try{new ctor();throw Error('constructed');}catch(e){if(!(e instanceof TypeError))throw e;}}
        try{listen({transport:'udp',port:0});throw Error('transport');}catch(e){if(!(e instanceof TypeError))throw e;}
        try{await connect({port:0});throw Error('connect port');}catch(e){if(!(e instanceof RangeError))throw e;}
        console.log('validated');
    "#,
        &[],
    );
    assert_eq!(output.stdout, b"validated\n");
}

#[test]
fn worker_termination_releases_a_pending_accept() {
    let directory = tempfile::tempdir().unwrap();
    let path = directory.path().join("worker.js");
    std::fs::write(&path,"import {listen} from 'avril:std';const listener=listen({port:0});postMessage(listener.addr.port);await listener.accept();").unwrap();
    let url = url::Url::from_file_path(&path).unwrap();
    let output = common::run(
        &format!(
            r#"
        import {{listen}} from 'avril:std';
        const worker=new Worker('{url}',{{type:'module'}});const port=await new Promise(resolve=>worker.onmessage=e=>resolve(e.data));worker.terminate();
        let listener;for(let i=0;i<100&&!listener;i++){{try{{listener=listen({{port}});}}catch{{await new Promise(resolve=>setTimeout(resolve,5));}}}}
        if(!listener)throw Error('leaked listener');listener.close();console.log('terminated');
    "#
        ),
        &[],
    );
    assert_eq!(output.stdout, b"terminated\n");
}

#[test]
fn incoming_data_is_not_delivered_after_authority_is_revoked() {
    let (_directory, mut child) = common::spawn(
        r#"
        import {listen,pledge,errors} from 'avril:std';
        const listener=listen({port:0});console.log(listener.addr.port);const connection=await listener.accept();
        const reading=connection.read(new Uint8Array(1));await new Promise(resolve=>setTimeout(resolve,10));pledge({net:false});console.log('narrowed');
        try{await reading;throw Error('received without authority');}catch(e){if(!(e instanceof errors.PermissionDenied))throw e;}
        connection.close();listener.close();console.log('revoked');
    "#,
    );
    let mut stdout = BufReader::new(child.stdout.take().unwrap());
    let mut port = String::new();
    stdout.read_line(&mut port).unwrap();
    let mut socket =
        std::net::TcpStream::connect(("127.0.0.1", port.trim().parse::<u16>().unwrap())).unwrap();
    let mut marker = String::new();
    stdout.read_line(&mut marker).unwrap();
    assert_eq!(marker, "narrowed\n");
    std::io::Write::write_all(&mut socket, b"x").unwrap();
    marker.clear();
    stdout.read_line(&mut marker).unwrap();
    assert_eq!(marker, "revoked\n");
    drop(stdout);
    let output = common::wait(child);
    assert!(
        output.status.success(),
        "{}",
        String::from_utf8_lossy(&output.stderr)
    );
    assert!(output.stdout.is_empty());
}

#[test]
fn socket_iterators_return_promises_and_finish_pending_reads() {
    let output = common::run(
        r#"
        import {listen,listenDatagram} from 'avril:std';
        for(const open of [listen,listenDatagram]) {
            const socket=open({port:0}),iterator=socket[Symbol.asyncIterator]();
            const pending=iterator.next();
            const result=await iterator.return().then(value=>value);
            if(!result.done||result.value!==undefined)throw Error('return result');
            if(!(await pending).done)throw Error('pending iteration');
            if(!(await iterator.return().then(value=>value)).done)throw Error('repeated return');
            if(!(await iterator.next()).done)throw Error('closed iteration');
        }
        console.log('iterator promises');
    "#,
        &[],
    );
    assert_eq!(output.stdout, b"iterator promises\n");
}

#[test]
fn tcp_writev_completes_vectors_snapshots_and_orders_shutdown() {
    let output = common::run(
        r#"
        import {listen,connect,errors} from 'avril:std';
        const listener=listen({port:0});
        async function pair(){const accept=listener.accept(),client=await connect({port:listener.addr.port}),server=await accept;return [client,server];}
        async function readAll(server,length){
            const buf=new Uint8Array(length);let offset=0,count;
            while(offset<length&&(count=await server.read(buf.subarray(offset)))!==null)offset+=count;
            if(offset!==length)throw Error('Truncated '+offset+'/'+length);
            return buf;
        }
        let [client,server]=await pair();
        if(await client.writev([new Uint8Array([10,20]),new Uint8Array(0),new Uint8Array([30])])!==3)throw Error('Count');
        if(await client.writev([])!==0)throw Error('Empty');
        await client.write(new Uint8Array([40]));
        await client.closeWrite();
        if((await readAll(server,4)).join()!=='10,20,30,40')throw Error('Order');
        try{await client.writev([new Uint8Array([1])]);throw Error('FIN');}catch(e){if(!(e instanceof errors.BadResource))throw e;}
        try{await client.writev([]);throw Error('Empty FIN');}catch(e){if(!(e instanceof errors.BadResource))throw e;}
        client.close();server.close();
        [client,server]=await pair();
        const big=[];for(let i=0;i<40;i++){const view=new Uint8Array(5000);view.fill(i);big.push(view);}
        const reading=readAll(server,200000);
        const sending=client.writev(big);
        for(const view of big)view.fill(255);
        big[big.length-1]=new Uint8Array([0]);
        const shutdown=client.closeWrite();
        if(await sending!==200000)throw Error('Total');
        await shutdown;
        const got=await reading;
        for(let i=0;i<40;i++)for(let j=0;j<5000;j++)if(got[i*5000+j]!==i)throw Error('Bytes '+i);
        client.close();server.close();
        [client,server]=await pair();
        const small=[];for(let i=0;i<3000;i++)small.push(new Uint8Array([i&255]));
        const gathering=readAll(server,3000);
        if(await client.writev(small)!==3000)throw Error('Small total');
        await client.closeWrite();
        const back=await gathering;
        for(let i=0;i<3000;i++)if(back[i]!==(i&255))throw Error('Small order '+i);
        client.close();server.close();
        [client,server]=await pair();
        const wide=[];for(let i=0;i<20;i++){const view=new Uint8Array(5000);view.fill(200+i);wide.push(view);}
        const following=readAll(server,100003);
        const vector=client.writev(wide);
        if(await client.write(new Uint8Array([7,7,7]))!==3)throw Error('Later write');
        await client.closeWrite();
        if(await vector!==100000)throw Error('Wide');
        const ordered=await following;
        for(let i=0;i<20;i++)for(let j=0;j<5000;j++)if(ordered[i*5000+j]!==200+i)throw Error('Wide order '+i);
        if(ordered[100000]!==7||ordered[100001]!==7||ordered[100002]!==7)throw Error('Later order');
        client.close();server.close();
        [client,server]=await pair();
        const first=new Uint8Array(100000);first.fill(7);
        const one=client.write(first),parts=[new Uint8Array([1,2]),new Uint8Array([3])];
        const two=client.writev(parts);
        parts[0].fill(9);parts[1]=new Uint8Array([4]);parts.push(new Uint8Array([0]));
        const n1=await one;
        if(!(n1>0&&n1<=65536))throw Error('Queued count');
        if(await two!==3)throw Error('Vector total');
        await client.closeWrite();
        const tail=await readAll(server,n1+3);
        if(tail[n1]!==1||tail[n1+1]!==2||tail[n1+2]!==3)throw Error('Snapshot');
        client.close();server.close();listener.close();console.log('writev');
    "#,
        &[],
    );
    assert_eq!(output.stdout, b"writev\n");
}

#[test]
fn tcp_writev_validates_vectors_and_honors_close_and_revocation() {
    let output = common::run(
        r#"
        import {listen,connect,pledge,errors} from 'avril:std';
        const listener=listen({port:0}),accept=listener.accept();
        const client=await connect({port:listener.addr.port}),server=await accept;
        async function rejects(vector){
            try{await client.writev(vector);}catch(e){if(e instanceof TypeError)return;throw e;}
            throw Error('Accepted vector');
        }
        for(const bad of ['bytes',42,null,undefined,{},new Uint8Array([1]),new Set([new Uint8Array([1])]),(function*(){yield new Uint8Array([1]);})()])await rejects(bad);
        for(const bad of [[new Uint8Array([1]),'x'],[[1,2]],[new DataView(new ArrayBuffer(1))],[new Uint16Array([1])],[new Uint8Array([1]),undefined],[null]])await rejects(bad);
        const detached=new Uint8Array([1]);
        structuredClone(detached.buffer,{transfer:[detached.buffer]});
        await rejects([detached]);
        const resizable=new ArrayBuffer(4,{maxByteLength:8}),outOfBounds=new Uint8Array(resizable,1,2);
        resizable.resize(0);
        await rejects([outOfBounds]);
        // An invalid later element rejects before anything is sent.
        await rejects([new Uint8Array([5,5]),'x']);
        const late=[];for(let i=0;i<20;i++)late.push(new Uint8Array(5000));late.push('x');
        await rejects(late);
        const marker=client.writev([new Uint8Array([9])]);
        await client.closeWrite();
        const first=new Uint8Array(2);let offset=0,count;
        while(offset<1&&(count=await server.read(first.subarray(offset)))!==null)offset+=count;
        if(await marker!==1||offset!==1||first[0]!==9)throw Error('Prefix sent');
        if(await server.read(new Uint8Array(1))!==null)throw Error('EOF');
        try{await client.writev([new Uint8Array([1])]);throw Error('FIN');}catch(e){if(!(e instanceof errors.BadResource))throw e;}
        client.close();server.close();
        const accept3=listener.accept(),client3=await connect({port:listener.addr.port}),server3=await accept3;
        const ok=new Uint8Array([1,2]),into=new Uint8Array(2);
        const RealArray=Uint8Array,realPush=Array.prototype.push;
        globalThis.Uint8Array=()=>{throw Error('Mutable intrinsic');};
        Array.prototype.push=()=>{throw Error('Mutable intrinsic');};
        if(await client3.writev([ok])!==2)throw Error('Captured');
        let done=0,part;
        while(done<2&&(part=await server3.read(into.subarray(done)))!==null)done+=part;
        if(done!==2||into.join()!=='1,2')throw Error('Poisoned bytes');
        client3.close();server3.close();
        globalThis.Uint8Array=RealArray;Array.prototype.push=realPush;
        const accept2=listener.accept(),client2=await connect({port:listener.addr.port}),server2=await accept2;
        const queued=client2.writev([new Uint8Array([1,2,3])]);
        pledge({net:false});
        try{await queued;throw Error('authority escaped');}catch(e){if(!(e instanceof errors.PermissionDenied))throw e;}
        client2.close();server2.close();listener.close();console.log('writev-guards');
    "#,
        &[],
    );
    assert_eq!(output.stdout, b"writev-guards\n");
}