script/fetch/

body.rs

/* This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this
 * file, You can obtain one at https://mozilla.org/MPL/2.0/. */

use std::io::Cursor;
use std::rc::Rc;
use std::{fs, ptr, str};

use encoding_rs::{Encoding, UTF_8};
use http::HeaderMap;
use http::header::{CONTENT_DISPOSITION, CONTENT_TYPE};
use ipc_channel::ipc::{self, IpcReceiver, IpcSender};
use ipc_channel::router::ROUTER;
use js::context::JSContext;
use js::jsapi::{Heap, JSObject, Value as JSValue};
use js::jsval::{JSVal, UndefinedValue};
use js::realm::CurrentRealm;
use js::rust::HandleValue;
use js::rust::wrappers2::{JS_ClearPendingException, JS_GetPendingException, JS_ParseJSON};
use js::typedarray::{ArrayBufferU8, Uint8};
use mime::{self, Mime};
use net_traits::request::{
    BodyChunkRequest, BodyChunkResponse, BodySource as NetBodySource, RequestBody,
};
use script_bindings::reflector::DomObject;
use servo_base::generic_channel::GenericSharedMemory;
use servo_constellation_traits::BlobImpl;
use url::form_urlencoded;

use crate::dom::RootedPromise;
use crate::dom::bindings::buffer_source::{create_buffer_source, get_buffer_source_copy};
use crate::dom::bindings::codegen::Bindings::BlobBinding::Blob_Binding::BlobMethods;
use crate::dom::bindings::codegen::Bindings::FormDataBinding::FormDataMethods;
use crate::dom::bindings::codegen::Bindings::XMLHttpRequestBinding::BodyInit;
use crate::dom::bindings::error::{Error, Fallible};
use crate::dom::bindings::inheritance::Castable;
use crate::dom::bindings::refcounted::Trusted;
use crate::dom::bindings::reflector::DomGlobal;
use crate::dom::bindings::root::{Dom, DomRoot, MutNullableDom};
use crate::dom::bindings::str::{DOMString, USVString};
use crate::dom::bindings::trace::RootedTraceableBox;
use crate::dom::blob::{Blob, normalize_type_string};
use crate::dom::encoding::textdecoderstream::TextDecoderStream;
use crate::dom::file::File;
use crate::dom::formdata::FormData;
use crate::dom::globalscope::GlobalScope;
use crate::dom::html::htmlformelement::{encode_multipart_form_data, generate_boundary};
use crate::dom::promise::Promise;
use crate::dom::promisenativehandler::{Callback, PromiseNativeHandler};
use crate::dom::readablestream::{
    ReadableStream, get_read_promise_bytes, get_read_promise_done, pipe_through,
};
use crate::dom::urlsearchparams::URLSearchParams;
use crate::fetch::mime_multipart::{Node, read_multipart_body};
use crate::realms::enter_auto_realm;
use crate::tasks::task_source::SendableTaskSource;

/// <https://fetch.spec.whatwg.org/#concept-body-clone>
pub(crate) fn clone_body_stream_for_dom_body(
    cx: &mut js::context::JSContext,
    original_body_stream: &MutNullableDom<ReadableStream>,
    cloned_body_stream: &MutNullableDom<ReadableStream>,
) -> Fallible<()> {
    // To clone a body *body*, run these steps:

    let Some(stream) = original_body_stream.get() else {
        return Ok(());
    };

    // step 1. Let « out1, out2 » be the result of teeing body’s stream.
    let branches = stream.tee(cx, true)?;
    let out1 = &*branches[0];
    let out2 = &*branches[1];

    // step 2. Set body’s stream to out1.
    // step 3. Return a body whose stream is out2 and other members are copied from body.
    original_body_stream.set(Some(out1));
    cloned_body_stream.set(Some(out2));

    Ok(())
}

/// The Dom object, or ReadableStream, that is the source of a body.
/// <https://fetch.spec.whatwg.org/#concept-body-source>
#[derive(Clone, Debug, PartialEq)]
pub(crate) enum BodySource {
    /// A ReadableStream comes with a null-source.
    Null,
    /// Another Dom object as source,
    /// TODO: store the actual object
    /// and re-extract a stream on re-direct.
    Object,
}

/// A body sent via [`TransmitBodyConnectHandler`]. Either in-memory or streamed.
#[derive(Clone)]
enum SendableBody {
    InMemory {
        data: GenericSharedMemory,
        done: bool,
    },
    Stream {
        stream: Trusted<ReadableStream>,
        task_source: SendableTaskSource,
        source: BodySource,
    },
}

/// The IPC route handler
/// for <https://fetch.spec.whatwg.org/#concept-request-transmit-body>.
/// This route runs in the script process,
/// and will queue tasks to perform operations
/// on the stream and transmit body chunks over IPC.
#[derive(Clone)]
struct TransmitBodyConnectHandler {
    body: SendableBody,
    bytes_sender: Option<IpcSender<BodyChunkResponse>>,
}

impl TransmitBodyConnectHandler {
    fn new(body: SendableBody) -> TransmitBodyConnectHandler {
        TransmitBodyConnectHandler {
            body,
            bytes_sender: None,
        }
    }

    /// Reset the `done` flag when this is an in-memory body, called when a stream is
    /// re-extracted from the source to support a re-direct.
    fn reset_in_memory_done(&mut self) {
        if let SendableBody::InMemory { done, .. } = &mut self.body {
            *done = false;
        }
    }

    /// Re-extract the source to support streaming it again for a re-direct.
    /// TODO: actually re-extract the source, instead of just cloning data, to support Blob.
    fn re_extract(&mut self, chunk_request_receiver: IpcReceiver<BodyChunkRequest>) {
        let mut body_handler = self.clone();
        body_handler.reset_in_memory_done();

        ROUTER.add_typed_route(
            chunk_request_receiver,
            Box::new(move |message| {
                let request = message.unwrap();
                match request {
                    BodyChunkRequest::Connect(sender) => {
                        body_handler.start_reading(sender);
                    },
                    BodyChunkRequest::Extract(receiver) => {
                        body_handler.re_extract(receiver);
                    },
                    BodyChunkRequest::Chunk => body_handler.transmit_source(),
                }
            }),
        );
    }

    /// In case of re-direct, and of a source available in memory,
    /// send it all in one chunk.
    ///
    /// TODO: this method should be deprecated
    /// in favor of making `re_extract` actually re-extract a stream from the source.
    /// See #26686
    fn transmit_source(&mut self) {
        match &mut self.body {
            SendableBody::InMemory { done: true, .. } => {
                // Step 5.1.3
                self.stop_reading();
            },
            SendableBody::InMemory { data, done } => {
                // The memoized bytes are sent so we mark it as done again
                *done = true;
                let _ = self
                    .bytes_sender
                    .as_ref()
                    .expect("No bytes sender to transmit source.")
                    .send(BodyChunkResponse::Chunk(data.clone()));
            },
            SendableBody::Stream { source, .. } => {
                assert_ne!(
                    *source,
                    BodySource::Null,
                    "ReadableStream(Null) sources should not re-direct."
                );
                warn!("Re-directs for file-based Blobs not supported yet.");
            },
        }
    }

    /// Take the IPC sender sent by `net`, so we can send body chunks with it.
    /// Also the entry point to <https://fetch.spec.whatwg.org/#concept-request-transmit-body>
    fn start_reading(&mut self, sender: IpcSender<BodyChunkResponse>) {
        self.bytes_sender = Some(sender);

        // If we're using an actual ReadableStream, acquire a reader for it.
        if let SendableBody::Stream {
            source: BodySource::Null,
            stream,
            task_source,
        } = &self.body
        {
            let stream = stream.clone();
            task_source.queue(task!(start_reading_request_body_stream: move |cx| {
                // Step 1, Let body be request’s body.
                let rooted_stream = stream.root();

                // TODO: Step 2, If body is null.

                // Step 3, get a reader for stream.
                rooted_stream.acquire_default_reader(cx)
                    .expect("Couldn't acquire a reader for the body stream.");

                // Note: this algorithm continues when the first chunk is requested by `net`.
            }));
        }
    }

    /// Take and drop [`Self::bytes_sender`].
    fn stop_reading(&mut self) {
        let bytes_sender = self
            .bytes_sender
            .take()
            .expect("Stop reading called multiple times on TransmitBodyConnectHandler.");
        let _ = bytes_sender.send(BodyChunkResponse::Done);
    }

    /// Step 4 and following of <https://fetch.spec.whatwg.org/#concept-request-transmit-body>
    fn transmit_body_chunk(&mut self) {
        let bytes_sender = self
            .bytes_sender
            .clone()
            .expect("No bytes sender to transmit chunk.");

        match &mut self.body {
            SendableBody::InMemory { done: true, .. } => {
                // Step 5.1.3
                self.stop_reading();
            },
            SendableBody::InMemory { data, done } => {
                // In case of the data being in-memory, send everything in one chunk,
                // by-passing SpiderMonkey.
                let _ = bytes_sender.send(BodyChunkResponse::Chunk(data.clone()));
                // Mark this body as `done` so that we can stop reading in the next tick,
                // matching the behavior of the promise-based flow
                *done = true;
            },
            SendableBody::Stream {
                stream,
                task_source,
                ..
            } => {
                let stream = stream.clone();
                task_source.queue(task!(setup_native_body_promise_handler: move |cx| {
                    let rooted_stream = stream.root();
                    let global = rooted_stream.global();

                    // Step 4, the result of reading a chunk from body’s stream with reader.
                    let promise = rooted_stream.read_a_chunk(cx);

                    // Step 5, the parallel steps waiting for and handling the result of the read promise,
                    // are a combination of the promise native handler here,
                    // and the corresponding IPC route in `component::net::http_loader`.
                    rooted!(&in(cx) let mut promise_handler = Some(TransmitBodyPromiseHandler {
                        bytes_sender: bytes_sender.clone(),
                        stream: Dom::from_ref(&rooted_stream),
                    }));

                    rooted!(&in(cx) let mut rejection_handler =
                        Some(TransmitBodyPromiseRejectionHandler {
                            bytes_sender,
                            stream: Dom::from_ref(&rooted_stream),
                        })
                    );

                    let handler = PromiseNativeHandler::new(
                        cx,
                        &global,
                        promise_handler.take().map(|handler| Box::new(handler) as Box<_>),
                        rejection_handler.take().map(|handler| Box::new(handler) as Box<_>),
                    );

                    let mut realm = enter_auto_realm(cx, &*global);
                    let realm = &mut realm.current_realm();
                    promise.append_native_handler(realm, &handler);
                }));
            },
        }
    }
}

/// The handler of read promises of body streams used in
/// <https://fetch.spec.whatwg.org/#concept-request-transmit-body>.
#[derive(Clone, JSTraceable, MallocSizeOf)]
#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
struct TransmitBodyPromiseHandler {
    #[no_trace]
    bytes_sender: IpcSender<BodyChunkResponse>,
    stream: Dom<ReadableStream>,
}

impl js::gc::Rootable for TransmitBodyPromiseHandler {}

impl Callback for TransmitBodyPromiseHandler {
    /// Step 5 of <https://fetch.spec.whatwg.org/#concept-request-transmit-body>
    fn callback(&self, cx: &mut CurrentRealm, v: HandleValue) {
        let is_done = match get_read_promise_done(cx, &v) {
            Ok(is_done) => is_done,
            Err(_) => {
                // Step 5.5, the "otherwise" steps.
                // TODO: terminate fetch.
                let _ = self.bytes_sender.send(BodyChunkResponse::Done);
                return self.stream.stop_reading(cx);
            },
        };

        if is_done {
            // Step 5.3, the "done" steps.
            // TODO: queue a fetch task on request to process request end-of-body.
            let _ = self.bytes_sender.send(BodyChunkResponse::Done);
            return self.stream.stop_reading(cx);
        }

        let chunk = match get_read_promise_bytes(cx, &v) {
            Ok(chunk) => chunk,
            Err(_) => {
                // Step 5.5, the "otherwise" steps.
                let _ = self.bytes_sender.send(BodyChunkResponse::Error);
                return self.stream.stop_reading(cx);
            },
        };

        // Step 5.1 and 5.2, transmit chunk.
        // Send the chunk to the body transmitter in net::http_loader::obtain_response.
        // TODO: queue a fetch task on request to process request body for request.
        let _ = self
            .bytes_sender
            .send(BodyChunkResponse::Chunk(GenericSharedMemory::from_vec(
                chunk,
            )));
    }
}

/// The handler of read promises rejection of body streams used in
/// <https://fetch.spec.whatwg.org/#concept-request-transmit-body>.
#[derive(Clone, JSTraceable, MallocSizeOf)]
#[cfg_attr(crown, crown::unrooted_must_root_lint::must_root)]
struct TransmitBodyPromiseRejectionHandler {
    #[no_trace]
    bytes_sender: IpcSender<BodyChunkResponse>,
    stream: Dom<ReadableStream>,
}

impl js::gc::Rootable for TransmitBodyPromiseRejectionHandler {}

impl Callback for TransmitBodyPromiseRejectionHandler {
    /// <https://fetch.spec.whatwg.org/#concept-request-transmit-body>
    fn callback(&self, cx: &mut CurrentRealm, _v: HandleValue) {
        // Step 5.4, the "rejection" steps.
        let _ = self.bytes_sender.send(BodyChunkResponse::Error);
        self.stream.stop_reading(cx);
    }
}

/// <https://fetch.spec.whatwg.org/#body-with-type>
pub(crate) struct ExtractedBody {
    /// <https://fetch.spec.whatwg.org/#concept-body-stream>
    pub(crate) stream: DomRoot<ReadableStream>,
    /// <https://fetch.spec.whatwg.org/#concept-body-source>
    pub(crate) source: BodySource,
    /// <https://fetch.spec.whatwg.org/#concept-body-total-bytes>
    pub(crate) total_bytes: Option<usize>,
    /// <https://fetch.spec.whatwg.org/#body-with-type-type>
    pub(crate) content_type: Option<DOMString>,
}

impl ExtractedBody {
    /// Build a request body from the extracted body,
    /// to be sent over IPC to net to use with `concept-request-transmit-body`,
    /// see <https://fetch.spec.whatwg.org/#concept-request-transmit-body>.
    ///
    /// Also returning the corresponding readable stream,
    /// to be stored on the request in script,
    /// and potentially used as part of `consume_body`,
    /// see <https://fetch.spec.whatwg.org/#concept-body-consume-body>
    ///
    /// Transmitting a body over fetch, and consuming it in script,
    /// are mutually exclusive operations, since each will lock the stream to a reader.
    pub(crate) fn into_net_request_body(
        self,
        cx: &mut JSContext,
    ) -> (RequestBody, DomRoot<ReadableStream>) {
        let ExtractedBody {
            stream,
            total_bytes,
            content_type: _,
            source,
        } = self;

        // First, setup some infra to be used to transmit body
        //  from `components::script` to `components::net`.
        let (chunk_request_sender, chunk_request_receiver) = ipc::channel().unwrap();

        // In case of the data being in-memory, send everything in one chunk, by-passing SM.
        // Empty extracted bodies are always representable as an in-memory empty payload.
        let in_memory = stream.get_in_memory_bytes(cx).or_else(|| {
            if total_bytes == Some(0) {
                Some(GenericSharedMemory::from_bytes(&[]))
            } else {
                None
            }
        });

        let net_source = match source {
            BodySource::Null => NetBodySource::Null,
            BodySource::Object => NetBodySource::Object,
        };

        let body = match in_memory {
            Some(data) => SendableBody::InMemory { data, done: false },
            None => {
                let global = stream.global();
                let task_manager = global.task_manager();
                let task_source = task_manager.networking_task_source().into();
                SendableBody::Stream {
                    stream: Trusted::new(&*stream),
                    task_source,
                    source,
                }
            },
        };

        let mut body_handler = TransmitBodyConnectHandler::new(body);
        ROUTER.add_typed_route(
            chunk_request_receiver,
            Box::new(move |message| match message.unwrap() {
                BodyChunkRequest::Connect(sender) => {
                    body_handler.start_reading(sender);
                },
                BodyChunkRequest::Extract(receiver) => {
                    body_handler.re_extract(receiver);
                },
                BodyChunkRequest::Chunk => body_handler.transmit_body_chunk(),
            }),
        );

        // Return `components::net` view into this request body,
        // which can be used by `net` to transmit it over the network.
        let request_body = RequestBody::new(chunk_request_sender, net_source, total_bytes);

        // Also return the stream for this body, which can be used by script to consume it.
        (request_body, stream)
    }

    /// Is the data of the stream of this extracted body available in memory?
    pub(crate) fn in_memory(&self) -> bool {
        self.stream.in_memory()
    }
}

/// <https://fetch.spec.whatwg.org/#concept-bodyinit-extract>
pub(crate) trait Extractable {
    fn extract(
        &self,
        cx: &mut js::context::JSContext,
        global: &GlobalScope,
        keep_alive: bool,
    ) -> Fallible<ExtractedBody>;
}

/// Part of <https://fetch.spec.whatwg.org/#concept-bodyinit-extract>
fn stream_from_body_init_bytes(
    cx: &mut js::context::JSContext,
    global: &GlobalScope,
    bytes: Vec<u8>,
) -> Fallible<DomRoot<ReadableStream>> {
    // Step 4: "Otherwise, set stream to a new ReadableStream object, and set up stream with byte reading support."
    // Step 11: "If source is a byte sequence, then set action to a step that returns source and length to source’s length."
    // Step 12.1: "Whenever one or more bytes are available and stream is not errored, enqueue the result of creating a Uint8Array from the available bytes into stream."
    // Step 12.1: "When running action is done, close stream."
    ReadableStream::new_from_bytes_with_byte_reading_support(cx, global, bytes)
}

impl Extractable for BodyInit {
    /// <https://fetch.spec.whatwg.org/#concept-bodyinit-extract>
    fn extract(
        &self,
        cx: &mut js::context::JSContext,
        global: &GlobalScope,
        keep_alive: bool,
    ) -> Fallible<ExtractedBody> {
        match self {
            BodyInit::String(s) => s.extract(cx, global, keep_alive),
            BodyInit::URLSearchParams(usp) => usp.extract(cx, global, keep_alive),
            BodyInit::Blob(b) => b.extract(cx, global, keep_alive),
            BodyInit::FormData(formdata) => formdata.extract(cx, global, keep_alive),
            BodyInit::ArrayBuffer(typedarray) => {
                // Set source to a copy of the bytes held by object.
                let bytes = get_buffer_source_copy(typedarray.into());
                let total_bytes = bytes.len();
                let stream = stream_from_body_init_bytes(cx, global, bytes)?;
                Ok(ExtractedBody {
                    stream,
                    total_bytes: Some(total_bytes),
                    content_type: None,
                    source: BodySource::Object,
                })
            },
            BodyInit::ArrayBufferView(typedarray) => {
                // Set source to a copy of the bytes held by object.
                let bytes = get_buffer_source_copy(typedarray.into());
                let total_bytes = bytes.len();
                let stream = stream_from_body_init_bytes(cx, global, bytes)?;
                Ok(ExtractedBody {
                    stream,
                    total_bytes: Some(total_bytes),
                    content_type: None,
                    source: BodySource::Object,
                })
            },
            BodyInit::ReadableStream(stream) => {
                // If keepalive is true, then throw a TypeError.
                if keep_alive {
                    return Err(Error::Type(
                        c"The body's stream is for a keepalive request".to_owned(),
                    ));
                }
                // If object is disturbed or locked, then throw a TypeError.
                if stream.is_locked() || stream.is_disturbed() {
                    return Err(Error::Type(
                        c"The body's stream is disturbed or locked".to_owned(),
                    ));
                }

                Ok(ExtractedBody {
                    stream: stream.clone(),
                    total_bytes: None,
                    content_type: None,
                    source: BodySource::Null,
                })
            },
        }
    }
}

impl Extractable for Vec<u8> {
    fn extract(
        &self,
        cx: &mut js::context::JSContext,
        global: &GlobalScope,
        _keep_alive: bool,
    ) -> Fallible<ExtractedBody> {
        let bytes = self.clone();
        let total_bytes = self.len();
        let stream = stream_from_body_init_bytes(cx, global, bytes)?;
        Ok(ExtractedBody {
            stream,
            total_bytes: Some(total_bytes),
            content_type: None,
            // A vec is used only in `submit_entity_body`.
            source: BodySource::Object,
        })
    }
}

impl Extractable for Blob {
    fn extract(
        &self,
        cx: &mut js::context::JSContext,
        _global: &GlobalScope,
        _keep_alive: bool,
    ) -> Fallible<ExtractedBody> {
        let blob_type = self.Type();
        let content_type = if blob_type.is_empty() {
            None
        } else {
            Some(blob_type)
        };
        let total_bytes = self.Size() as usize;
        let stream = self.get_stream(cx)?;
        Ok(ExtractedBody {
            stream,
            total_bytes: Some(total_bytes),
            content_type,
            source: BodySource::Object,
        })
    }
}

impl Extractable for DOMString {
    fn extract(
        &self,
        cx: &mut js::context::JSContext,
        global: &GlobalScope,
        _keep_alive: bool,
    ) -> Fallible<ExtractedBody> {
        let bytes = self.as_bytes(cx.no_gc()).to_owned();
        let total_bytes = bytes.len();
        let content_type = Some(DOMString::from_static("text/plain;charset=UTF-8"));
        let stream = stream_from_body_init_bytes(cx, global, bytes)?;
        Ok(ExtractedBody {
            stream,
            total_bytes: Some(total_bytes),
            content_type,
            source: BodySource::Object,
        })
    }
}

impl Extractable for FormData {
    fn extract(
        &self,
        cx: &mut js::context::JSContext,
        global: &GlobalScope,
        _keep_alive: bool,
    ) -> Fallible<ExtractedBody> {
        let boundary = generate_boundary();
        let bytes = encode_multipart_form_data(&mut self.datums(), boundary.clone(), UTF_8);
        let total_bytes = bytes.len();
        let content_type = Some(DOMString::from(format!(
            "multipart/form-data; boundary={}",
            boundary
        )));
        let stream = stream_from_body_init_bytes(cx, global, bytes)?;
        Ok(ExtractedBody {
            stream,
            total_bytes: Some(total_bytes),
            content_type,
            source: BodySource::Object,
        })
    }
}

impl Extractable for URLSearchParams {
    fn extract(
        &self,
        cx: &mut js::context::JSContext,
        global: &GlobalScope,
        _keep_alive: bool,
    ) -> Fallible<ExtractedBody> {
        let bytes = self.serialize_utf8().into_bytes();
        let total_bytes = bytes.len();
        let content_type = Some(DOMString::from_static(
            "application/x-www-form-urlencoded;charset=UTF-8",
        ));
        let stream = stream_from_body_init_bytes(cx, global, bytes)?;
        Ok(ExtractedBody {
            stream,
            total_bytes: Some(total_bytes),
            content_type,
            source: BodySource::Object,
        })
    }
}

#[derive(Clone, Copy, JSTraceable, MallocSizeOf)]
pub(crate) enum BodyType {
    Blob,
    Bytes,
    FormData,
    Json,
    Text,
    ArrayBuffer,
}

pub(crate) enum FetchedData {
    Text(String),
    Json(RootedTraceableBox<Heap<JSValue>>),
    BlobData(DomRoot<Blob>),
    Bytes(RootedTraceableBox<Heap<*mut JSObject>>),
    FormData(DomRoot<FormData>),
    ArrayBuffer(RootedTraceableBox<Heap<*mut JSObject>>),
    JSException(RootedTraceableBox<Heap<JSVal>>),
}

/// <https://fetch.spec.whatwg.org/#concept-body-consume-body>
/// <https://fetch.spec.whatwg.org/#body-fully-read>
/// A combination of parts of both algorithms,
/// `body-fully-read` can be fully implemented, and separated, later,
/// see #36049.
pub(crate) fn consume_body<T: BodyMixin + DomObject>(
    cx: &mut js::context::JSContext,
    object: &T,
    body_type: BodyType,
) -> RootedPromise {
    let global = object.global();

    // Enter the realm of the object whose body is being consumed.
    let mut realm = enter_auto_realm(cx, &*global);
    let cx: &mut _ = &mut realm.current_realm();

    // Let promise be a new promise.
    // Note: re-ordered so we can return the promise below.
    let promise = Promise::new_in_realm(cx);

    // If object is unusable, then return a promise rejected with a TypeError.
    if object.is_unusable() {
        promise.reject_error(
            cx,
            Error::Type(c"The body's stream is disturbed or locked".to_owned()),
        );
        return promise;
    }

    let stream = match object.body() {
        Some(stream) => stream,
        None => {
            // If object’s body is null, then run successSteps with an empty byte sequence.
            let mime_type = object.get_mime_type(cx);
            resolve_result_promise(cx, body_type, &promise, mime_type, &[]);
            return promise;
        },
    };

    // <https://fetch.spec.whatwg.org/#concept-body-consume-body>
    // Otherwise, fully read object’s body given successSteps, errorSteps, and object’s relevant global object.
    //
    // <https://fetch.spec.whatwg.org/#body-fully-read>
    // Let reader be the result of getting a reader for body’s stream.
    // Read all bytes from reader, given successSteps and errorSteps.
    //
    // <https://streams.spec.whatwg.org/#readable-stream-default-reader-read>
    // Set stream.[[disturbed]] to true.
    // Otherwise, if stream.[[state]] is "errored", perform readRequest’s error steps given stream.[[storedError]].
    //
    // If the body stream is already errored (for example, the fetch was aborted after the Response exists),
    // the normal fully read path would reject with [[storedError]] but would also mark the stream disturbed.
    // Once the stream is disturbed, later calls reject with TypeError ("disturbed or locked") instead of the
    // original AbortError. This early return rejects with the same [[storedError]] without disturbing the
    // stream, so repeated calls (for example, calling text() twice) keep rejecting with AbortError.
    if stream.is_errored() {
        rooted!(&in(cx) let mut stored_error = UndefinedValue());
        stream.get_stored_error(stored_error.handle_mut());
        promise.reject(cx, stored_error.handle());
        return promise;
    }

    // Note: from `fully_read`.
    // Let reader be the result of getting a reader for body’s stream.
    // If that threw an exception,
    // then run errorSteps with that exception and return.
    let reader = match stream.acquire_default_reader(cx) {
        Ok(r) => r,
        Err(e) => {
            promise.reject_error(cx, e);
            return promise;
        },
    };

    // Let errorSteps given error be to reject promise with error.
    let error_promise = promise.clone();

    // Let successSteps given a byte sequence data be to resolve promise
    // with the result of running convertBytesToJSValue with data.
    // If that threw an exception, then run errorSteps with that exception.
    let mime_type = object.get_mime_type(cx);
    let success_promise = promise.clone();

    // Read all bytes from reader, given successSteps and errorSteps.
    // Note: spec uses an intermediary concept of `fully_read`,
    // which seems useful when invoking fetch from other places.
    // TODO: #36049
    reader.read_all_bytes(
        cx,
        Rc::new(move |cx, bytes| {
            resolve_result_promise(cx, body_type, &success_promise, mime_type.clone(), &bytes);
        }),
        Rc::new(move |cx, v| {
            error_promise.reject(cx, v);
        }),
    );

    promise
}

/// The success steps of
/// <https://fetch.spec.whatwg.org/#concept-body-consume-body>.
fn resolve_result_promise(
    cx: &mut js::context::JSContext,
    body_type: BodyType,
    promise: &Promise,
    mime_type: Vec<u8>,
    body: &[u8],
) {
    let pkg_data_results = run_package_data_algorithm(cx, body, body_type, mime_type);

    match pkg_data_results {
        Ok(results) => {
            match results {
                FetchedData::Text(s) => promise.resolve_native(cx, &USVString(s)),
                FetchedData::Json(j) => promise.resolve_native(cx, &j),
                FetchedData::BlobData(b) => promise.resolve_native(cx, &b),
                FetchedData::FormData(f) => promise.resolve_native(cx, &f),
                FetchedData::Bytes(b) => promise.resolve_native(cx, &b),
                FetchedData::ArrayBuffer(a) => promise.resolve_native(cx, &a),
                FetchedData::JSException(e) => promise.reject_native(cx, &e.handle()),
            };
        },
        Err(err) => promise.reject_error(cx, err),
    }
}

/// The algorithm that takes a byte sequence
/// and returns a JavaScript value or throws an exception of
/// <https://fetch.spec.whatwg.org/#concept-body-consume-body>.
fn run_package_data_algorithm(
    cx: &mut js::context::JSContext,
    bytes: &[u8],
    body_type: BodyType,
    mime_type: Vec<u8>,
) -> Fallible<FetchedData> {
    let mime = &*mime_type;
    let mut realm = CurrentRealm::assert(cx);
    let global = GlobalScope::from_current_realm(&mut realm);
    match body_type {
        BodyType::Text => run_text_data_algorithm(bytes),
        BodyType::Json => run_json_data_algorithm(cx, bytes),
        BodyType::Blob => run_blob_data_algorithm(cx, &global, bytes.to_vec(), mime),
        BodyType::FormData => run_form_data_algorithm(cx, &global, bytes, mime),
        BodyType::ArrayBuffer => run_array_buffer_data_algorithm(cx, bytes),
        BodyType::Bytes => run_bytes_data_algorithm(cx, bytes),
    }
}

/// <https://fetch.spec.whatwg.org/#ref-for-concept-body-consume-body%E2%91%A4>
fn run_text_data_algorithm(bytes: &[u8]) -> Fallible<FetchedData> {
    // This implements the Encoding standard's "decode UTF-8", which removes the
    // BOM if present.
    let no_bom_bytes = if bytes.starts_with(b"\xEF\xBB\xBF") {
        &bytes[3..]
    } else {
        bytes
    };
    Ok(FetchedData::Text(
        String::from_utf8_lossy(no_bom_bytes).into_owned(),
    ))
}

#[expect(unsafe_code)]
/// <https://fetch.spec.whatwg.org/#ref-for-concept-body-consume-body%E2%91%A3>
fn run_json_data_algorithm(cx: &mut js::context::JSContext, bytes: &[u8]) -> Fallible<FetchedData> {
    // The JSON spec allows implementations to either ignore UTF-8 BOM or treat it as an error.
    // `JS_ParseJSON` treats this as an error, so it is necessary for us to strip it if present.
    //
    // https://datatracker.ietf.org/doc/html/rfc8259#section-8.1
    let json_text = decode_to_utf16_with_bom_removal(bytes, UTF_8);
    rooted!(&in(cx) let mut rval = UndefinedValue());
    unsafe {
        if !JS_ParseJSON(
            cx,
            json_text.as_ptr(),
            json_text.len() as u32,
            rval.handle_mut(),
        ) {
            rooted!(&in(cx) let mut exception = UndefinedValue());
            assert!(JS_GetPendingException(cx, exception.handle_mut()));
            JS_ClearPendingException(cx);
            return Ok(FetchedData::JSException(RootedTraceableBox::from_box(
                Heap::boxed(exception.get()),
            )));
        }
        let rooted_heap = RootedTraceableBox::from_box(Heap::boxed(rval.get()));
        Ok(FetchedData::Json(rooted_heap))
    }
}

/// <https://fetch.spec.whatwg.org/#ref-for-concept-body-consume-body%E2%91%A0>
fn run_blob_data_algorithm(
    cx: &mut js::context::JSContext,
    root: &GlobalScope,
    bytes: Vec<u8>,
    mime: &[u8],
) -> Fallible<FetchedData> {
    let mime_string = String::from_utf8(mime.to_vec()).unwrap_or_default();
    let blob = Blob::new(
        cx,
        root,
        BlobImpl::new_from_bytes(bytes, normalize_type_string(&mime_string)),
    );
    Ok(FetchedData::BlobData(blob))
}

fn extract_name_from_content_disposition(headers: &HeaderMap) -> Option<String> {
    let cd = headers.get(CONTENT_DISPOSITION)?.to_str().ok()?;

    for part in cd.split(';').map(|s| s.trim()) {
        if let Some(rest) = part.strip_prefix("name=") {
            let v = rest.trim();
            let v = v.strip_prefix('"').unwrap_or(v);
            let v = v.strip_suffix('"').unwrap_or(v);
            return Some(v.to_string());
        }
    }
    None
}

fn extract_filename_from_content_disposition(headers: &HeaderMap) -> Option<String> {
    let cd = headers.get(CONTENT_DISPOSITION)?.to_str().ok()?;
    if let Some(index) = cd.find("filename=") {
        let start = index + "filename=".len();
        return Some(
            cd.get(start..)
                .unwrap_or_default()
                .trim_matches('"')
                .to_owned(),
        );
    }
    if let Some(index) = cd.find("filename*=UTF-8''") {
        let start = index + "filename*=UTF-8''".len();
        return Some(
            cd.get(start..)
                .unwrap_or_default()
                .trim_matches('"')
                .to_owned(),
        );
    }
    None
}

fn content_type_from_headers(headers: &HeaderMap) -> Result<String, Error> {
    match headers.get(CONTENT_TYPE) {
        Some(value) => Ok(value
            .to_str()
            .map_err(|_| Error::Type(c"Inappropriate MIME-type for Body".to_owned()))?
            .to_string()),
        None => Ok("text/plain".to_string()),
    }
}

fn append_form_data_entry_from_part(
    cx: &mut js::context::JSContext,
    root: &GlobalScope,
    formdata: &FormData,
    headers: &HeaderMap,
    body: Vec<u8>,
) -> Fallible<()> {
    let Some(name) = extract_name_from_content_disposition(headers) else {
        return Ok(());
    };
    // A part whose `Content-Disposition` header contains a `name` parameter whose value is `_charset_` is parsed like any other part. It does not change the encoding.
    let filename = extract_filename_from_content_disposition(headers);
    if let Some(filename) = filename {
        // Each part whose `Content-Disposition` header contains a `filename` parameter must be parsed into an entry whose value is a File object whose contents are the contents of the part.
        //
        // The name attribute of the File object must have the value of the `filename` parameter of the part.
        //
        // The type attribute of the File object must have the value of the `Content-Type` header of the part if the part has such header, and `text/plain` (the default defined by [RFC7578] section 4.4) otherwise.
        let content_type = content_type_from_headers(headers)?;
        let file = File::new(
            cx,
            root,
            BlobImpl::new_from_bytes(body, normalize_type_string(&content_type)),
            DOMString::from(filename),
            None,
        );
        let blob = file.upcast::<Blob>();
        formdata.Append_(cx, USVString(name), blob, None);
    } else {
        // Each part whose `Content-Disposition` header does not contain a `filename` parameter must be parsed into an entry whose value is the UTF-8 decoded without BOM content of the part. This is done regardless of the presence or the value of a `Content-Type` header and regardless of the presence or the value of a `charset` parameter.

        let (value, _) = UTF_8.decode_without_bom_handling(&body);
        formdata.Append(cx, USVString(name), USVString(value.to_string()));
    }
    Ok(())
}

fn append_multipart_nodes(
    cx: &mut js::context::JSContext,
    root: &GlobalScope,
    formdata: &FormData,
    nodes: Vec<Node>,
) -> Fallible<()> {
    for node in nodes {
        match node {
            Node::Part(part) => {
                append_form_data_entry_from_part(cx, root, formdata, &part.headers, part.body)?;
            },
            Node::File(file_part) => {
                let body = fs::read(&file_part.path)
                    .map_err(|_| Error::Type(c"file part could not be read".to_owned()))?;
                append_form_data_entry_from_part(cx, root, formdata, &file_part.headers, body)?;
            },
            Node::Multipart((_, inner)) => {
                append_multipart_nodes(cx, root, formdata, inner)?;
            },
        }
    }
    Ok(())
}

/// <https://fetch.spec.whatwg.org/#ref-for-concept-body-consume-body%E2%91%A2>
fn run_form_data_algorithm(
    cx: &mut js::context::JSContext,
    root: &GlobalScope,
    bytes: &[u8],
    mime: &[u8],
) -> Fallible<FetchedData> {
    // The formData() method steps are to return the result of running consume body
    // with this and the following steps given a byte sequence bytes:
    let mime_str = str::from_utf8(mime).unwrap_or_default();
    let mime: Mime = mime_str
        .parse()
        .map_err(|_| Error::Type(c"Inappropriate MIME-type for Body".to_owned()))?;

    // Let mimeType be the result of get the MIME type with this.
    //
    // If mimeType is non-null, then switch on mimeType’s essence and run the corresponding steps:
    if mime.type_() == mime::MULTIPART && mime.subtype() == mime::FORM_DATA {
        // "multipart/form-data"
        // Parse bytes, using the value of the `boundary` parameter from mimeType,
        // per the rules set forth in Returning Values from Forms: multipart/form-data. [RFC7578]
        let mut headers = HeaderMap::new();
        headers.insert(
            CONTENT_TYPE,
            mime_str
                .parse()
                .map_err(|_| Error::Type(c"Inappropriate MIME-type for Body".to_owned()))?,
        );

        if let Some(boundary) = mime.get_param(mime::BOUNDARY) {
            let closing_boundary = format!("--{}--", boundary.as_str()).into_bytes();
            let trimmed_bytes = bytes.strip_suffix(b"\r\n").unwrap_or(bytes);
            if trimmed_bytes == closing_boundary {
                let formdata = FormData::new(cx, None, root);
                return Ok(FetchedData::FormData(formdata));
            }
        }

        let mut cursor = Cursor::new(bytes);
        // If that fails for some reason, then throw a TypeError.
        let nodes = read_multipart_body(&mut cursor, &headers, false)
            .map_err(|_| Error::Type(c"Inappropriate MIME-type for Body".to_owned()))?;
        // The above is a rough approximation of what is needed for `multipart/form-data`,
        // a more detailed parsing specification is to be written. Volunteers welcome.

        // Return a new FormData object, appending each entry, resulting from the parsing operation, to its entry list.
        let formdata = FormData::new(cx, None, root);

        append_multipart_nodes(cx, root, &formdata, nodes)?;

        return Ok(FetchedData::FormData(formdata));
    }

    if mime.type_() == mime::APPLICATION && mime.subtype() == mime::WWW_FORM_URLENCODED {
        // "application/x-www-form-urlencoded"
        // Let entries be the result of parsing bytes.
        //
        // Return a new FormData object whose entry list is entries.
        let entries = form_urlencoded::parse(bytes);
        let formdata = FormData::new(cx, None, root);
        for (k, e) in entries {
            formdata.Append(cx, USVString(k.into_owned()), USVString(e.into_owned()));
        }
        return Ok(FetchedData::FormData(formdata));
    }

    // Throw a TypeError.
    Err(Error::Type(c"Inappropriate MIME-type for Body".to_owned()))
}

/// <https://fetch.spec.whatwg.org/#ref-for-concept-body-consume-body%E2%91%A1>
fn run_bytes_data_algorithm(
    cx: &mut js::context::JSContext,
    bytes: &[u8],
) -> Fallible<FetchedData> {
    rooted!(&in(cx) let mut array_buffer_ptr = ptr::null_mut::<JSObject>());

    create_buffer_source::<Uint8>(cx, bytes, array_buffer_ptr.handle_mut())
        .map_err(|_| Error::JSFailed)?;

    let rooted_heap = RootedTraceableBox::from_box(Heap::boxed(array_buffer_ptr.get()));
    Ok(FetchedData::Bytes(rooted_heap))
}

/// <https://fetch.spec.whatwg.org/#ref-for-concept-body-consume-body>
pub(crate) fn run_array_buffer_data_algorithm(
    cx: &mut js::context::JSContext,
    bytes: &[u8],
) -> Fallible<FetchedData> {
    rooted!(&in(cx) let mut array_buffer_ptr = ptr::null_mut::<JSObject>());

    create_buffer_source::<ArrayBufferU8>(cx, bytes, array_buffer_ptr.handle_mut())
        .map_err(|_| Error::JSFailed)?;

    let rooted_heap = RootedTraceableBox::from_box(Heap::boxed(array_buffer_ptr.get()));
    Ok(FetchedData::ArrayBuffer(rooted_heap))
}

pub(crate) fn decode_to_utf16_with_bom_removal(
    bytes: &[u8],
    encoding: &'static Encoding,
) -> Vec<u16> {
    let mut decoder = encoding.new_decoder_with_bom_removal();
    let capacity = decoder
        .max_utf16_buffer_length(bytes.len())
        .expect("Overflow");
    let mut utf16 = vec![0u16; capacity];
    let (_, read, written, _) = decoder.decode_to_utf16(bytes, &mut utf16, true);
    assert_eq!(read, bytes.len());
    utf16.truncate(written);
    utf16
}

/// <https://fetch.spec.whatwg.org/#body>
pub(crate) trait BodyMixin {
    /// <https://fetch.spec.whatwg.org/#dom-body-bodyused>
    fn is_body_used(&self) -> bool;
    /// <https://fetch.spec.whatwg.org/#body-unusable>
    fn is_unusable(&self) -> bool;
    /// <https://fetch.spec.whatwg.org/#dom-body-body>
    fn body(&self) -> Option<DomRoot<ReadableStream>>;
    /// <https://fetch.spec.whatwg.org/#concept-body-mime-type>
    fn get_mime_type(&self, cx: &mut js::context::JSContext) -> Vec<u8>;
}

/// <https://fetch.spec.whatwg.org/#dom-body-textstream>
pub(crate) fn body_text_stream<T: BodyMixin + DomObject>(
    cx: &mut js::context::JSContext,
    object: &T,
) -> Fallible<DomRoot<ReadableStream>> {
    // Step 1: If this is unusable, then throw a TypeError.
    if object.is_unusable() {
        return Err(Error::Type(
            c"The body's stream is disturbed or locked".to_owned(),
        ));
    }

    // Step 3: Let stream be this’s body’s stream.
    let Some(stream) = object.body() else {
        // Step 2: If this's body is null:
        // set up a ReadableStream emptyStream, close it, and return it.
        return ReadableStream::new_empty(cx, &object.global());
    };

    // Step 4: Let decoder be a new TextDecoderStream object in this’s relevant realm.
    // Step 5: Set up decoder with UTF-8.
    let decoder =
        TextDecoderStream::new_with_proto(cx, &object.global(), None, UTF_8, false, false)?;

    // Step 6. Return the result of stream, piped through decoder.
    Ok(pipe_through(&stream, cx, &object.global(), &decoder))
}