standard-plugin-sdk 0.1.1

//! The `wasi:http` transport of [`super::http`]: a request goes out through
//! the host's outgoing handler and its response is read on the daemon
//! executor, both bounded by the request's deadline.

use alloc::string::String;
use alloc::vec::Vec;

use wasip2::http::outgoing_handler;
use wasip2::http::types::{
    ErrorCode, Fields, IncomingBody, Method, OutgoingBody, OutgoingRequest, RequestOptions, Scheme,
};
use wasip2::io::streams::StreamError;

use super::http::{Request, Response, Target, timeout, too_long};
use super::io::wait_any;
use crate::error::{Error, Result};

/// Sends `request` over `wasi:http` and reads the response.
pub(crate) async fn send(spec: &Request, target: &Target<'_>) -> Result<Response> {
    let deadline =
        wasip2::clocks::monotonic_clock::subscribe_duration(u64::from(spec.timeout_ms) * 1_000_000);
    let fields = Fields::new();
    for (name, value) in &spec.headers {
        fields
            .append(&name.to_ascii_lowercase(), &value.as_bytes().to_vec())
            .map_err(|error| Error::Invalid(alloc::format!("header {name}: {error:?}")))?;
    }
    // A known body goes with its length, not chunked.
    if !spec.body.is_empty()
        && !spec
            .headers
            .iter()
            .any(|(name, _)| name.eq_ignore_ascii_case("content-length"))
    {
        let length = alloc::format!("{}", spec.body.len());
        fields
            .append("content-length", &length.into_bytes())
            .map_err(|error| Error::Invalid(alloc::format!("content-length: {error:?}")))?;
    }
    let request = OutgoingRequest::new(fields);
    let invalid = |what: &str| Error::Invalid(alloc::format!("{what} of {}", spec.url));
    request
        .set_method(&method(&spec.method))
        .map_err(|()| invalid("the method"))?;
    request
        .set_scheme(Some(&scheme(target.scheme)))
        .map_err(|()| invalid("the scheme"))?;
    request
        .set_authority(Some(target.authority))
        .map_err(|()| invalid("the host"))?;
    request
        .set_path_with_query(Some(target.path))
        .map_err(|()| invalid("the path"))?;
    let outgoing = request.body().map_err(|()| invalid("the body"))?;
    let options = RequestOptions::new();
    let nanos = u64::from(spec.timeout_ms) * 1_000_000;
    // The host may not support every timeout; the deadline above holds
    // regardless.
    let _ = options.set_connect_timeout(Some(nanos));
    let _ = options.set_first_byte_timeout(Some(nanos));
    let _ = options.set_between_bytes_timeout(Some(nanos));
    let future = outgoing_handler::handle(request, Some(options)).map_err(failed)?;
    write_body(&outgoing, &spec.body, &deadline).await?;
    OutgoingBody::finish(outgoing, None).map_err(failed)?;
    let response = loop {
        match future.get() {
            Some(Ok(Ok(response))) => break response,
            Some(Ok(Err(ErrorCode::HttpRequestDenied))) => return Err(target.denied()),
            Some(Ok(Err(error))) => return Err(failed(error)),
            Some(Err(())) => return Err(Error::Unavailable("response taken".into())),
            None => {
                if wait_any(&[&future.subscribe(), &deadline]).await == 1 {
                    return Err(timeout());
                }
            }
        }
    };
    let status = response.status();
    let headers: Vec<(String, String)> = response
        .headers()
        .entries()
        .into_iter()
        .map(|(name, value)| (name, String::from_utf8_lossy(&value).into_owned()))
        .collect();
    if let Some(length) = headers
        .iter()
        .find(|(name, _)| name.eq_ignore_ascii_case("content-length"))
        .and_then(|(_, value)| value.trim().parse::<usize>().ok())
        && length > spec.max_body
    {
        return Err(too_long(spec.max_body));
    }
    let body = response
        .consume()
        .map_err(|()| Error::Unavailable("body taken".into()))?;
    let bytes = read_body(&body, spec.max_body, &deadline).await?;
    drop(body);
    Ok(Response {
        status,
        headers,
        body: bytes,
    })
}

fn failed(error: ErrorCode) -> Error {
    match error {
        ErrorCode::ConnectionTimeout | ErrorCode::ConnectionReadTimeout => timeout(),
        other => Error::Unavailable(alloc::format!("{other:?}")),
    }
}

fn scheme(name: &str) -> Scheme {
    match name {
        "http" => Scheme::Http,
        _ => Scheme::Https,
    }
}

fn method(name: &str) -> Method {
    match name.to_ascii_uppercase().as_str() {
        "GET" => Method::Get,
        "HEAD" => Method::Head,
        "POST" => Method::Post,
        "PUT" => Method::Put,
        "DELETE" => Method::Delete,
        "CONNECT" => Method::Connect,
        "OPTIONS" => Method::Options,
        "TRACE" => Method::Trace,
        "PATCH" => Method::Patch,
        other => Method::Other(String::from(other)),
    }
}

async fn write_body(
    body: &OutgoingBody,
    bytes: &[u8],
    deadline: &wasip2::io::poll::Pollable,
) -> Result<()> {
    if bytes.is_empty() {
        return Ok(());
    }
    let stream = body
        .write()
        .map_err(|()| Error::Unavailable("request body taken".into()))?;
    let mut rest = bytes;
    while !rest.is_empty() {
        let room = match stream.check_write() {
            Ok(0) => {
                if wait_any(&[&stream.subscribe(), deadline]).await == 1 {
                    return Err(timeout());
                }
                continue;
            }
            Ok(room) => room as usize,
            Err(error) => return Err(stream_error(error)),
        };
        let chunk = rest.len().min(room);
        stream.write(&rest[..chunk]).map_err(stream_error)?;
        rest = &rest[chunk..];
    }
    // Every byte is the host's now. A server may answer and close before
    // the flush settles: the response still reads, so a closed stream here
    // is not a failure.
    match stream.flush() {
        Ok(()) | Err(StreamError::Closed) => {}
        Err(error) => return Err(stream_error(error)),
    }
    if wait_any(&[&stream.subscribe(), deadline]).await == 1 {
        return Err(timeout());
    }
    match stream.check_write() {
        Ok(_) | Err(StreamError::Closed) => {}
        Err(error) => return Err(stream_error(error)),
    }
    drop(stream);
    Ok(())
}

fn stream_error(error: StreamError) -> Error {
    match error {
        StreamError::Closed => Error::Unavailable("the connection closed".into()),
        StreamError::LastOperationFailed(error) => Error::Unavailable(error.to_debug_string()),
    }
}

async fn read_body(
    body: &IncomingBody,
    max: usize,
    deadline: &wasip2::io::poll::Pollable,
) -> Result<Vec<u8>> {
    let stream = body
        .stream()
        .map_err(|()| Error::Unavailable("body stream taken".into()))?;
    let mut bytes = Vec::new();
    loop {
        match stream.read(64 * 1024) {
            Ok(chunk) if chunk.is_empty() => {
                if wait_any(&[&stream.subscribe(), deadline]).await == 1 {
                    return Err(timeout());
                }
            }
            Ok(chunk) => {
                if bytes.len() + chunk.len() > max {
                    return Err(too_long(max));
                }
                bytes.extend_from_slice(&chunk);
            }
            Err(StreamError::Closed) => break,
            Err(StreamError::LastOperationFailed(error)) => {
                return Err(Error::Unavailable(error.to_debug_string()));
            }
        }
    }
    Ok(bytes)
}