mod common;

use common::run;

#[test]
fn stream_transfers_preserve_backpressure_chunks_and_failure_propagation() {
    let output = run(
        r#"
        let canceled;
        const source = new ReadableStream({
            start(controller) { controller.enqueue({value: 1}); controller.close(); },
            cancel(reason) { canceled = reason; },
        });
        const movedReadable = structuredClone(source, {transfer: [source]});
        if (!source.locked || movedReadable === source) throw Error('Readable transfer lock');
        try { movedReadable.getReader({mode: 'byob'}); throw Error('Transferred stream retained byte mode'); }
        catch (error) { if (!(error instanceof TypeError)) throw error; }
        let pulls = 0;
        const lazySource = new ReadableStream({pull(controller) { pulls++; controller.enqueue(1); }});
        const lazyClone = structuredClone(lazySource, {transfer: [lazySource]});
        if (pulls !== 0) throw Error('Readable transfer pulled before demand');
        await lazyClone.getReader().read();
        if (pulls === 0) throw Error('Readable transfer did not pull on demand');
        const chunk = (await movedReadable.getReader().read()).value;
        if (chunk.value !== 1) throw Error('Readable transfer chunk');
        const cancelSource = new ReadableStream({cancel(reason) { canceled = reason; }});
        const movedCancel = structuredClone(cancelSource, {transfer: [cancelSource]});
        const cancelReason = {cancel: true};
        await movedCancel.cancel(cancelReason);
        if (canceled === cancelReason || !canceled.cancel) throw Error('Readable transfer cancel');
        const cancelFailure = {cancelFailure: true};
        const failedCancelSource = new ReadableStream({cancel() { throw cancelFailure; }});
        const failedCancelClone = structuredClone(failedCancelSource, {transfer: [failedCancelSource]});
        try { await failedCancelClone.cancel(); throw Error('Readable cancel rejection'); }
        catch (error) { if (error === cancelFailure || !error.cancelFailure) throw error; }

        const written = [];
        const writable = new WritableStream({write(value) { written.push(value); }});

        const movedWritable = structuredClone(writable, {transfer: [writable]});
        if (!writable.locked || movedWritable === writable) throw Error('Writable transfer lock');
        const writer = movedWritable.getWriter();
        await writer.write({value: 2});
        await writer.close();
        if (written.length !== 1 || written[0].value !== 2) throw Error('Writable transfer writes');
        const closedSource = new WritableStream(), closedWriter = closedSource.getWriter();
        await closedWriter.close();
        closedWriter.releaseLock();
        const closedClone = structuredClone(closedSource, {transfer: [closedSource]});
        try { await closedClone.getWriter().write(1); throw Error('Closed writable transfer'); }
        catch (error) { if (!(error instanceof TypeError)) throw error; }
        const writeReason = {writeFailure: true};
        const failedWritable = new WritableStream({write() { throw writeReason; }});
        const movedFailedWritable = structuredClone(failedWritable, {transfer: [failedWritable]});
        try { await movedFailedWritable.getWriter().write(1); throw Error('Writable transfer error'); }
        catch (error) { if (error === writeReason || !error.writeFailure) throw error; }
        let cloneWriteAbort;
        const uncloneableWriteSource = new WritableStream({abort(reason) { cloneWriteAbort = reason; }});
        const uncloneableWriteClone = structuredClone(uncloneableWriteSource, {transfer: [uncloneableWriteSource]});
        try { await uncloneableWriteClone.getWriter().write(() => {}); throw Error('Uncloneable writable chunk'); }
        catch (error) { if (!(error instanceof DOMException) || error.name !== 'DataCloneError') throw error; }
        if (!(cloneWriteAbort instanceof DOMException) || cloneWriteAbort.name !== 'DataCloneError') {
            throw Error('Writable clone failure did not abort the source');
        }

        const uncommitted = new ReadableStream(), transferBuffer = new ArrayBuffer(1);
        try {
            structuredClone({stream: uncommitted, fail() {}}, {transfer: [uncommitted, transferBuffer]});
            throw Error('Transfer serialization failure');
        } catch (error) { if (!(error instanceof DOMException) || error.name !== 'DataCloneError') throw error; }
        if (transferBuffer.detached || uncommitted.locked) throw Error('Uncommitted stream transfer changed source');
        const recovered = structuredClone(uncommitted, {transfer: [uncommitted]});
        if (recovered === uncommitted || !uncommitted.locked) throw Error('Failed transfer poisoned later clone');

        const transform = new TransformStream({transform(value, controller) { controller.enqueue(value * 2); }});
        const movedTransform = structuredClone(transform, {transfer: [transform]});
        if (Object.hasOwn(movedTransform, 'readable') || Object.hasOwn(movedTransform, 'writable')) {
            throw Error('Transform transfer replaced public accessors');
        }
        const reader = movedTransform.readable.getReader(), transformWriter = movedTransform.writable.getWriter();
        const firstWrite = transformWriter.write(3);
        if ((await reader.read()).value !== 6) throw Error('Transform transfer');
        await firstWrite;
        reader.releaseLock();
        transformWriter.releaseLock();
        const movedAgain = structuredClone(movedTransform, {transfer: [movedTransform]});
        const secondReader = movedAgain.readable.getReader(), secondWriter = movedAgain.writable.getWriter();
        const secondWrite = secondWriter.write(4);
        if ((await secondReader.read()).value !== 8) throw Error('Retransferred transform');
        await secondWrite;
        await secondWriter.close();
        if (!(await secondReader.read()).done) throw Error('Retransferred transform close');

        const reason = {failure: true};
        const failedSource = new ReadableStream({pull() {throw reason;}});
        const failed = structuredClone(failedSource, {transfer: [failedSource]});
        try { await failed.getReader().read(); throw Error('Readable transfer error'); }
        catch (error) { if (error === reason || error.failure !== true) throw error; }
        let cloneFailureCancellation;
        const uncloneableChunkSource = new ReadableStream({
            pull(controller) { controller.enqueue(() => {}); },
            cancel(error) { cloneFailureCancellation = error; },
        });
        const uncloneableChunkClone = structuredClone(uncloneableChunkSource, {transfer: [uncloneableChunkSource]});
        try { await uncloneableChunkClone.getReader().read(); throw Error('Uncloneable stream chunk'); }
        catch (error) { if (!(error instanceof DOMException) || error.name !== 'DataCloneError') throw error; }
        if (!(cloneFailureCancellation instanceof DOMException) || cloneFailureCancellation.name !== 'DataCloneError') {
            throw Error('Chunk clone failure did not cancel the source');
        }

        const byteSource = new ReadableStream({
            type: 'bytes',
            start(controller) { controller.enqueue(new Uint8Array([1, 2, 3])); controller.close(); },
        });
        const byteClone = structuredClone(byteSource, {transfer: [byteSource]});
        const byteReader = byteClone.getReader();
        const byteResult = await byteReader.read();
        if (byteResult.done || byteResult.value.join() !== '1,2,3') throw Error('Transferred byte stream read');
        if (!(await byteReader.read()).done) throw Error('Transferred byte stream EOF');
        console.log('stream transfer');
        "#,
        &[],
    );
    assert_eq!(output.stdout, b"stream transfer\n");
}

#[test]
fn release_and_internal_errors_do_not_create_unhandled_rejections() {
    let output = run(
        r#"
        for (const make of [
            () => new ReadableStream(),
            () => new ReadableStream({type:'bytes'}),
            () => new ReadableStream({start(c) {c.error('source failure');}})
        ]) {
            const reader = make().getReader();
            reader.releaseLock();
        }
        const byob = new ReadableStream({type:'bytes'}).getReader({mode:'byob'});
        byob.releaseLock();
        for (const make of [
            () => new WritableStream(),
            () => new WritableStream({start(c) {c.error('sink failure');}})
        ]) {
            const writer = make().getWriter();
            writer.releaseLock();
        }
        await new Promise(resolve => setTimeout(resolve));
        console.log('handled');
        "#,
        &[],
    );
    assert_eq!(output.stdout, b"handled\n");
}

#[test]
fn pull_and_write_failures_preserve_reasons_and_observable_promises() {
    let output = run(
        r#"
        const reason = {failure:true};
        async function rejects(promise) {
            try {await promise; throw Error('Expected rejection');}
            catch (error) {if (error !== reason) throw error;}
        }
        for (const failure of [
            () => {throw reason;},
            () => Promise.reject(reason),
            () => ({then(resolve, reject) {reject(reason);}})
        ]) {
            const reader = new ReadableStream({pull:failure}).getReader();
            await Promise.all([rejects(reader.read()), rejects(reader.closed)]);
            reader.releaseLock();
            const writer = new WritableStream({write:failure}).getWriter();
            await Promise.all([rejects(writer.write(1)), rejects(writer.closed)]);
            writer.releaseLock();
        }
        console.log('reasons');
        "#,
        &[],
    );
    assert_eq!(output.stdout, b"reasons\n");
}

#[test]
fn piping_propagates_source_and_sink_failures_without_leaking_internal_rejections() {
    let output = run(
        r#"
        const reason = {failure:true};
        let aborted, canceled;
        const failedSource = new ReadableStream({pull() {throw reason;}});
        try {
            await failedSource.pipeTo(new WritableStream({abort(error) {aborted = error;}}));
            throw Error('Successful failed source');
        } catch (error) {if (error !== reason) throw error;}
        if (aborted !== reason || failedSource.locked) throw Error('Source failure propagation');
        const source = new ReadableStream({
            pull(c) {c.enqueue(1);},
            cancel(error) {canceled = error;}
        });
        const failedSink = new WritableStream({write() {return Promise.reject(reason);}});
        try {await source.pipeTo(failedSink); throw Error('Successful failed sink');}
        catch (error) {if (error !== reason) throw error;}
        if (canceled !== reason || source.locked || failedSink.locked) throw Error('Sink failure propagation');
        console.log('piping');
        "#,
        &[],
    );
    assert_eq!(output.stdout, b"piping\n");
}

#[test]
fn public_operation_promises_and_observer_promises_keep_their_identities() {
    let output = run(
        r#"
        const reader = new ReadableStream({start(c) {c.close();}}).getReader();
        const reads = [reader.read(), reader.read()];
        if (reads[0] === reads[1]) throw Error('Shared read results');
        if ((await Promise.all(reads)).some(result => !result.done)) throw Error('Closed reads');
        const cancels = [reader.cancel(), reader.cancel()];
        if (cancels[0] === cancels[1]) throw Error('Shared closed cancel results');
        await Promise.all(cancels);
        reader.releaseLock();

        const a = new ReadableStream().getReader(), b = new ReadableStream().getReader();
        if (a.closed !== a.closed || a.closed === b.closed) throw Error('Reader closed identity');
        const cancellations = [a.cancel(), b.cancel()];
        if (cancellations[0] === cancellations[1]) throw Error('Shared cancel results');
        await Promise.all(cancellations);
        a.releaseLock(); b.releaseLock();

        const writer = new WritableStream().getWriter();
        const other = new WritableStream().getWriter();
        if (writer.ready !== writer.ready || writer.ready === other.ready ||
            writer.closed !== writer.closed || writer.closed === other.closed) {
            throw Error('Writer observer identity');
        }
        const writes = [writer.write(1), writer.write(2)];
        if (writes[0] === writes[1]) throw Error('Shared write results');
        await Promise.all(writes);
        await writer.close();
        const aborts = [writer.abort(), writer.abort()];
        if (aborts[0] === aborts[1]) throw Error('Shared closed abort results');
        await Promise.all(aborts);
        await other.close();
        writer.releaseLock(); other.releaseLock();
        console.log('identity');
        "#,
        &[],
    );
    assert_eq!(output.stdout, b"identity\n");
}

#[test]
fn default_transforms_keep_backpressure_independent_between_streams() {
    let output = run(
        r#"
        const a = new TransformStream(), b = new TransformStream();
        const ar = a.readable.getReader(), br = b.readable.getReader();
        const aw = a.writable.getWriter(), bw = b.writable.getWriter();
        let aWritten = false, bWritten = false, bReady = false;
        const aWrite = aw.write('a').then(() => {aWritten = true;});
        const bWrite = bw.write('b').then(() => {bWritten = true;});
        const aReady = aw.ready, bReadyPromise = bw.ready;
        if (aReady === bReadyPromise) throw Error('Shared backpressure promise');
        bReadyPromise.then(() => {bReady = true;});
        await new Promise(resolve => setTimeout(resolve));
        if (aWritten || bWritten || bReady) throw Error('Missing initial backpressure');
        if ((await ar.read()).value !== 'a') throw Error('First transform');
        await aWrite; await aReady;
        if (!aWritten || bWritten || bReady) throw Error('Cross-stream backpressure');
        if ((await br.read()).value !== 'b') throw Error('Second transform');
        await bWrite; await bReadyPromise;
        await Promise.all([aw.close(), bw.close()]);
        if (!(await ar.read()).done || !(await br.read()).done) throw Error('Transform closure');
        ar.releaseLock(); br.releaseLock(); aw.releaseLock(); bw.releaseLock();
        console.log('backpressure');
        "#,
        &[],
    );
    assert_eq!(output.stdout, b"backpressure\n");
}

#[test]
fn user_callback_thenables_are_assimilated_once_per_invocation() {
    let output = run(
        r#"
        let accesses = 0, calls = 0;
        const writer = new WritableStream({
            write(value) {
                calls++;
                return {get then() {
                    accesses++;
                    return resolve => queueMicrotask(resolve);
                }};
            }
        }).getWriter();
        await Promise.all([writer.write(1), writer.write(2), writer.write(3)]);
        await writer.close();
        if (calls !== 3 || accesses !== 3) throw Error('Callback assimilation');
        writer.releaseLock();
        console.log('thenables');
        "#,
        &[],
    );
    assert_eq!(output.stdout, b"thenables\n");
}

#[test]
fn internal_handling_does_not_suppress_unobserved_public_write_failures() {
    let (_dir, child) = common::spawn(
        "new WritableStream({write() {throw Error('public write failure');}}).getWriter().write(1);",
    );
    let output = common::wait(child);
    assert!(!output.status.success());
    assert!(String::from_utf8_lossy(&output.stderr).contains("public write failure"));
}

#[test]
fn transferred_writable_ready_resolves_before_the_first_write() {
    let out = run(
        r#"
        let value;
        const source=new WritableStream({write(chunk){value=chunk;}});
        const moved=structuredClone(source,{transfer:[source]});
        const writer=moved.getWriter();
        await writer.ready;
        await writer.write('ready');
        await writer.close();
        if(value!=='ready')throw Error('Writable transfer readiness');
        console.log('transfer ready');
    "#,
        &[],
    );
    assert_eq!(out.stdout, b"transfer ready\n");
}