cratestack_client_rust/

lib.rs

use std::fs;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};

use chrono::{DateTime, Utc};
pub use cratestack_codec_cbor::CborCodec;
pub use cratestack_codec_json::JsonCodec;
use cratestack_core::{
    CoolCodec, CoolError, CoolErrorResponse, Page, SelectionQuery, canonical_request_string,
};
use reqwest::header::{ACCEPT, CONTENT_TYPE, HeaderMap, HeaderName, HeaderValue};
use reqwest::{Method, StatusCode, Url};
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use serde_json::Value as JsonValue;

const BRIDGE_CONTENT_TYPE: &str = "application/json";
const CBOR_SEQUENCE_CONTENT_TYPE: &str = "application/cbor-seq";

pub trait Projection {
    type Output;

    fn selection_query(&self) -> SelectionQuery;

    fn decode_one(&self, value: JsonValue) -> Result<Self::Output, CoolError>;

    fn decode_many(&self, value: JsonValue) -> Result<Vec<Self::Output>, CoolError> {
        match value {
            JsonValue::Array(values) => values
                .into_iter()
                .map(|value| self.decode_one(value))
                .collect(),
            other => Err(CoolError::Internal(format!(
                "projected list payload must be an array, got {other:?}"
            ))),
        }
    }

    fn decode_page(&self, value: JsonValue) -> Result<Page<Self::Output>, CoolError> {
        let page = serde_json::from_value::<Page<JsonValue>>(value).map_err(|error| {
            CoolError::Codec(format!("failed to decode projected page payload: {error}"))
        })?;
        let items = page
            .items
            .into_iter()
            .map(|value| self.decode_one(value))
            .collect::<Result<Vec<_>, _>>()?;
        Ok(Page::new(items, page.page_info).with_total_count(page.total_count))
    }
}

impl Projection for SelectionQuery {
    type Output = JsonValue;

    fn selection_query(&self) -> SelectionQuery {
        self.clone()
    }

    fn decode_one(&self, value: JsonValue) -> Result<Self::Output, CoolError> {
        Ok(value)
    }
}

#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
pub enum RuntimeCodecConfig {
    #[default]
    Cbor,
    Json,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
pub enum RuntimeEnvelopeConfig {
    #[default]
    None,
    CoseSign1,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, Default)]
pub struct RuntimeTransportConfig {
    pub codec: RuntimeCodecConfig,
    pub envelope: RuntimeEnvelopeConfig,
}

pub trait HttpClientCodec: CoolCodec {
    fn accept_header_value(&self) -> &'static str;

    fn sequence_accept_header_value(&self) -> &'static str;

    fn decode_response<T>(&self, content_type: &str, body: &[u8]) -> Result<T, CoolError>
    where
        T: DeserializeOwned;

    fn decode_sequence_response<T>(
        &self,
        content_type: &str,
        body: &[u8],
    ) -> Result<Vec<T>, CoolError>
    where
        T: DeserializeOwned;
}

impl HttpClientCodec for CborCodec {
    fn accept_header_value(&self) -> &'static str {
        "application/cbor, application/json"
    }

    fn sequence_accept_header_value(&self) -> &'static str {
        "application/cbor-seq, application/cbor, application/json"
    }

    fn decode_response<T>(&self, content_type: &str, body: &[u8]) -> Result<T, CoolError>
    where
        T: DeserializeOwned,
    {
        if media_type_matches(content_type, CborCodec::CONTENT_TYPE) {
            self.decode(body)
        } else if media_type_matches(content_type, JsonCodec::CONTENT_TYPE) {
            JsonCodec.decode(body)
        } else {
            Err(CoolError::Codec(format!(
                "unsupported response Content-Type {content_type}"
            )))
        }
    }

    fn decode_sequence_response<T>(
        &self,
        content_type: &str,
        body: &[u8],
    ) -> Result<Vec<T>, CoolError>
    where
        T: DeserializeOwned,
    {
        if media_type_matches(content_type, CBOR_SEQUENCE_CONTENT_TYPE) {
            decode_cbor_sequence(body)
        } else if media_type_matches(content_type, CborCodec::CONTENT_TYPE) {
            self.decode(body)
        } else if media_type_matches(content_type, JsonCodec::CONTENT_TYPE) {
            JsonCodec.decode(body)
        } else {
            Err(CoolError::Codec(format!(
                "unsupported response Content-Type {content_type}"
            )))
        }
    }
}

impl HttpClientCodec for JsonCodec {
    fn accept_header_value(&self) -> &'static str {
        "application/json, application/cbor"
    }

    fn sequence_accept_header_value(&self) -> &'static str {
        "application/cbor-seq, application/json, application/cbor"
    }

    fn decode_response<T>(&self, content_type: &str, body: &[u8]) -> Result<T, CoolError>
    where
        T: DeserializeOwned,
    {
        if media_type_matches(content_type, JsonCodec::CONTENT_TYPE) {
            self.decode(body)
        } else if media_type_matches(content_type, CborCodec::CONTENT_TYPE) {
            CborCodec.decode(body)
        } else {
            Err(CoolError::Codec(format!(
                "unsupported response Content-Type {content_type}"
            )))
        }
    }

    fn decode_sequence_response<T>(
        &self,
        content_type: &str,
        body: &[u8],
    ) -> Result<Vec<T>, CoolError>
    where
        T: DeserializeOwned,
    {
        if media_type_matches(content_type, CBOR_SEQUENCE_CONTENT_TYPE) {
            decode_cbor_sequence(body)
        } else if media_type_matches(content_type, JsonCodec::CONTENT_TYPE) {
            self.decode(body)
        } else if media_type_matches(content_type, CborCodec::CONTENT_TYPE) {
            CborCodec.decode(body)
        } else {
            Err(CoolError::Codec(format!(
                "unsupported response Content-Type {content_type}"
            )))
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct RequestJournalEntry {
    pub method: String,
    pub path: String,
    pub status_code: u16,
    pub content_type: Option<String>,
    pub recorded_at: DateTime<Utc>,
}

fn default_schema_version() -> u32 {
    1
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct PersistedClientState {
    #[serde(default = "default_schema_version")]
    pub schema_version: u32,
    #[serde(default)]
    pub state_version: u64,
    #[serde(default)]
    pub request_journal: Vec<RequestJournalEntry>,
}

impl Default for PersistedClientState {
    fn default() -> Self {
        Self {
            schema_version: default_schema_version(),
            state_version: 0,
            request_journal: Vec::new(),
        }
    }
}

pub trait ClientStateStore: Send + Sync {
    fn load(&self) -> Result<PersistedClientState, ClientError>;
    fn save(&self, state: &PersistedClientState) -> Result<(), ClientError>;

    fn append_request_journal(&self, entry: &RequestJournalEntry) -> Result<(), ClientError> {
        let mut state = self.load()?;
        state.request_journal.push(entry.clone());
        state.state_version = state.state_version.saturating_add(1);
        self.save(&state)
    }
}

#[derive(Debug, Default)]
pub struct InMemoryStateStore {
    state: Mutex<PersistedClientState>,
}

impl ClientStateStore for InMemoryStateStore {
    fn load(&self) -> Result<PersistedClientState, ClientError> {
        self.state
            .lock()
            .map_err(|error| ClientError::State(format!("failed to lock state store: {error}")))
            .map(|state| state.clone())
    }

    fn save(&self, state: &PersistedClientState) -> Result<(), ClientError> {
        let mut guard = self
            .state
            .lock()
            .map_err(|error| ClientError::State(format!("failed to lock state store: {error}")))?;
        *guard = state.clone();
        Ok(())
    }
}

#[derive(Debug, Clone)]
pub struct JsonFileStateStore {
    path: PathBuf,
}

impl JsonFileStateStore {
    pub fn new(path: impl Into<PathBuf>) -> Self {
        Self { path: path.into() }
    }

    pub fn path(&self) -> &Path {
        &self.path
    }
}

impl ClientStateStore for JsonFileStateStore {
    fn load(&self) -> Result<PersistedClientState, ClientError> {
        match fs::read(&self.path) {
            Ok(bytes) => serde_json::from_slice(&bytes).map_err(|error| {
                ClientError::State(format!(
                    "failed to decode state file {}: {error}",
                    self.path.display()
                ))
            }),
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
                Ok(PersistedClientState::default())
            }
            Err(error) => Err(ClientError::State(format!(
                "failed to read state file {}: {error}",
                self.path.display()
            ))),
        }
    }

    fn save(&self, state: &PersistedClientState) -> Result<(), ClientError> {
        if let Some(parent) = self.path.parent() {
            fs::create_dir_all(parent).map_err(|error| {
                ClientError::State(format!(
                    "failed to create state directory {}: {error}",
                    parent.display()
                ))
            })?;
        }
        let bytes = serde_json::to_vec_pretty(state).map_err(|error| {
            ClientError::State(format!(
                "failed to encode state file {}: {error}",
                self.path.display()
            ))
        })?;
        fs::write(&self.path, bytes).map_err(|error| {
            ClientError::State(format!(
                "failed to write state file {}: {error}",
                self.path.display()
            ))
        })
    }
}

#[derive(Debug, Clone)]
pub struct ClientConfig {
    pub base_url: Url,
}

impl ClientConfig {
    pub fn new(base_url: Url) -> Self {
        Self { base_url }
    }
}

#[derive(Debug, thiserror::Error)]
pub enum ClientError {
    #[error("transport error: {0}")]
    Transport(#[from] reqwest::Error),
    #[error("codec error: {0}")]
    Codec(#[from] CoolError),
    #[error("state error: {0}")]
    State(String),
    #[error("invalid response: {0}")]
    InvalidResponse(String),
    #[error("bad input: {0}")]
    BadInput(String),
    #[error("remote call failed with status {status}: {message}")]
    Remote {
        status: StatusCode,
        error: Option<CoolErrorResponse>,
        message: String,
    },
}

pub type HeaderPair<'a> = (&'a str, &'a str);
pub type QueryPair<'a> = (&'a str, &'a str);

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AuthorizationRequest {
    pub method: String,
    pub path: String,
    pub canonical_query: Option<String>,
    pub content_type: Option<String>,
    pub body: Vec<u8>,
    pub canonical_request: String,
}

pub trait RequestAuthorizer: Send + Sync {
    fn authorize(
        &self,
        request: &AuthorizationRequest,
    ) -> Result<Vec<(String, String)>, ClientError>;
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct RuntimeHeader {
    pub name: String,
    pub value: String,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct RuntimeRequestWire {
    pub method: String,
    pub path: String,
    pub canonical_query: Option<String>,
    pub headers: Vec<RuntimeHeader>,
    pub body: Vec<u8>,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct RuntimeResponseWire {
    pub status_code: u16,
    pub headers: Vec<RuntimeHeader>,
    pub body: Vec<u8>,
}

#[repr(u32)]
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
pub enum RuntimeErrorCode {
    Transport = 1,
    Codec = 2,
    State = 3,
    InvalidResponse = 4,
    Remote = 5,
    BadInput = 6,
}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct RuntimeErrorWire {
    pub code: RuntimeErrorCode,
    pub http_status: Option<u16>,
    pub message: String,
    pub remote_code: Option<String>,
    pub remote_body: Option<Vec<u8>>,
}

impl From<ClientError> for RuntimeErrorWire {
    fn from(value: ClientError) -> Self {
        match value {
            ClientError::Transport(error) => Self {
                code: RuntimeErrorCode::Transport,
                http_status: None,
                message: error.to_string(),
                remote_code: None,
                remote_body: None,
            },
            ClientError::Codec(error) => Self {
                code: RuntimeErrorCode::Codec,
                http_status: Some(error.status_code().as_u16()),
                message: error.to_string(),
                remote_code: Some(error.code().to_owned()),
                remote_body: None,
            },
            ClientError::State(message) => Self {
                code: RuntimeErrorCode::State,
                http_status: None,
                message,
                remote_code: None,
                remote_body: None,
            },
            ClientError::InvalidResponse(message) => Self {
                code: RuntimeErrorCode::InvalidResponse,
                http_status: None,
                message,
                remote_code: None,
                remote_body: None,
            },
            ClientError::BadInput(message) => Self {
                code: RuntimeErrorCode::BadInput,
                http_status: None,
                message,
                remote_code: None,
                remote_body: None,
            },
            ClientError::Remote {
                status,
                error,
                message,
            } => Self {
                code: RuntimeErrorCode::Remote,
                http_status: Some(status.as_u16()),
                remote_code: error.as_ref().map(|value| value.code.clone()),
                remote_body: error
                    .as_ref()
                    .and_then(|value| serde_json::to_vec(value).ok()),
                message,
            },
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RuntimeStateStoreConfig {
    InMemory,
    JsonFile { path: PathBuf },
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RuntimeConfigWire {
    pub base_url: String,
    pub state_store: RuntimeStateStoreConfig,
    pub transport: RuntimeTransportConfig,
}

pub struct RuntimeHandle {
    runtime: tokio::runtime::Runtime,
    client: RuntimeTransportClient,
}

enum RuntimeTransportClient {
    Cbor(CratestackClient<CborCodec>),
    Json(CratestackClient<JsonCodec>),
}

impl RuntimeHandle {
    pub fn new(config: RuntimeConfigWire) -> Result<Self, RuntimeErrorWire> {
        let base_url = Url::parse(&config.base_url).map_err(|error| RuntimeErrorWire {
            code: RuntimeErrorCode::BadInput,
            http_status: None,
            message: format!("invalid base URL '{}': {error}", config.base_url),
            remote_code: None,
            remote_body: None,
        })?;
        let state_store: Arc<dyn ClientStateStore> = match config.state_store {
            RuntimeStateStoreConfig::InMemory => Arc::new(InMemoryStateStore::default()),
            RuntimeStateStoreConfig::JsonFile { path } => Arc::new(JsonFileStateStore::new(path)),
        };
        if config.transport.envelope != RuntimeEnvelopeConfig::None {
            return Err(RuntimeErrorWire {
                code: RuntimeErrorCode::BadInput,
                http_status: None,
                message: "COSE envelope support is not implemented yet".to_owned(),
                remote_code: None,
                remote_body: None,
            });
        }
        let client = match config.transport.codec {
            RuntimeCodecConfig::Cbor => RuntimeTransportClient::Cbor(
                CratestackClient::new(ClientConfig::new(base_url.clone()), CborCodec)
                    .with_state_store(state_store.clone()),
            ),
            RuntimeCodecConfig::Json => RuntimeTransportClient::Json(
                CratestackClient::new(ClientConfig::new(base_url), JsonCodec)
                    .with_state_store(state_store),
            ),
        };
        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .map_err(|error| RuntimeErrorWire {
                code: RuntimeErrorCode::State,
                http_status: None,
                message: format!("failed to build runtime: {error}"),
                remote_code: None,
                remote_body: None,
            })?;

        Ok(Self { runtime, client })
    }

    pub fn execute(
        &self,
        request: RuntimeRequestWire,
    ) -> Result<RuntimeResponseWire, RuntimeErrorWire> {
        self.runtime
            .block_on(self.client.execute_raw(request))
            .map_err(RuntimeErrorWire::from)
    }

    pub fn state(&self) -> Result<PersistedClientState, ClientError> {
        self.client.state()
    }
}

fn replace_bridge_content_type(headers: &mut Vec<RuntimeHeader>) {
    headers.retain(|header| !header.name.eq_ignore_ascii_case("content-type"));
    headers.push(RuntimeHeader {
        name: "content-type".to_owned(),
        value: BRIDGE_CONTENT_TYPE.to_owned(),
    });
}

impl RuntimeTransportClient {
    async fn execute_raw(
        &self,
        request: RuntimeRequestWire,
    ) -> Result<RuntimeResponseWire, ClientError> {
        let request = self.bridge_request_to_transport(request)?;
        match self {
            Self::Cbor(client) => client.execute_raw_transport(request).await,
            Self::Json(client) => client.execute_raw_transport(request).await,
        }
        .and_then(|response| self.transport_response_to_bridge(response))
    }

    fn bridge_request_to_transport(
        &self,
        request: RuntimeRequestWire,
    ) -> Result<RuntimeRequestWire, ClientError> {
        if request.body.is_empty() {
            return Ok(request);
        }

        let value: JsonValue = serde_json::from_slice(&request.body).map_err(|error| {
            ClientError::BadInput(format!("invalid bridge payload JSON: {error}"))
        })?;
        let body = match self {
            Self::Cbor(_) => CborCodec.encode(&value)?,
            Self::Json(_) => JsonCodec.encode(&value)?,
        };

        Ok(RuntimeRequestWire { body, ..request })
    }

    fn transport_response_to_bridge(
        &self,
        mut response: RuntimeResponseWire,
    ) -> Result<RuntimeResponseWire, ClientError> {
        if response.body.is_empty() {
            replace_bridge_content_type(&mut response.headers);
            return Ok(response);
        }

        let value = match self {
            Self::Cbor(_) => CborCodec.decode::<JsonValue>(&response.body)?,
            Self::Json(_) => JsonCodec.decode::<JsonValue>(&response.body)?,
        };

        response.body = serde_json::to_vec(&value).map_err(|error| {
            ClientError::InvalidResponse(format!("failed to encode bridge payload JSON: {error}"))
        })?;
        replace_bridge_content_type(&mut response.headers);
        Ok(response)
    }

    fn state(&self) -> Result<PersistedClientState, ClientError> {
        match self {
            Self::Cbor(client) => client.state(),
            Self::Json(client) => client.state(),
        }
    }
}

#[derive(Clone)]
pub struct CratestackClient<C = CborCodec> {
    http: reqwest::Client,
    config: ClientConfig,
    codec: C,
    state_store: Arc<dyn ClientStateStore>,
    request_authorizer: Option<Arc<dyn RequestAuthorizer>>,
}

impl CratestackClient<CborCodec> {
    pub fn cbor(config: ClientConfig) -> Self {
        Self::new(config, CborCodec)
    }
}

impl<C> CratestackClient<C>
where
    C: HttpClientCodec,
{
    pub fn new(config: ClientConfig, codec: C) -> Self {
        Self {
            http: reqwest::Client::new(),
            config,
            codec,
            state_store: Arc::new(InMemoryStateStore::default()),
            request_authorizer: None,
        }
    }

    pub fn with_http_client(config: ClientConfig, codec: C, http: reqwest::Client) -> Self {
        Self {
            http,
            config,
            codec,
            state_store: Arc::new(InMemoryStateStore::default()),
            request_authorizer: None,
        }
    }

    pub fn with_state_store(mut self, state_store: Arc<dyn ClientStateStore>) -> Self {
        self.state_store = state_store;
        self
    }

    pub fn with_optional_state_store(self, state_store: Option<Arc<dyn ClientStateStore>>) -> Self {
        match state_store {
            Some(state_store) => self.with_state_store(state_store),
            None => self,
        }
    }

    pub fn with_request_authorizer(
        mut self,
        request_authorizer: Arc<dyn RequestAuthorizer>,
    ) -> Self {
        self.request_authorizer = Some(request_authorizer);
        self
    }

    pub fn state(&self) -> Result<PersistedClientState, ClientError> {
        self.state_store.load()
    }

    pub async fn get<Output>(
        &self,
        path: &str,
        query: &[QueryPair<'_>],
        headers: &[HeaderPair<'_>],
    ) -> Result<Output, ClientError>
    where
        Output: DeserializeOwned,
    {
        let response = self
            .request_raw(Method::GET, path, None, query, headers)
            .await?;
        decode_typed_response(&self.codec, &response)
    }

    pub async fn get_view<P>(
        &self,
        path: &str,
        projection: &P,
        headers: &[HeaderPair<'_>],
    ) -> Result<P::Output, ClientError>
    where
        P: Projection,
    {
        let selection = projection.selection_query();
        let canonical_query = canonical_query_from_selection(&selection, &[])?;
        let response = self
            .request_raw_with_query_and_accept(
                Method::GET,
                path,
                None,
                canonical_query.as_deref(),
                headers,
                Some(JsonCodec::CONTENT_TYPE),
            )
            .await?;
        let value = decode_json_value_response(&JsonCodec, &response)?;
        projection.decode_one(value).map_err(ClientError::from)
    }

    pub async fn list_view<P>(
        &self,
        path: &str,
        projection: &P,
        extra_query: &[QueryPair<'_>],
        headers: &[HeaderPair<'_>],
    ) -> Result<Vec<P::Output>, ClientError>
    where
        P: Projection,
    {
        let selection = projection.selection_query();
        let canonical_query = canonical_query_from_selection(&selection, extra_query)?;
        let response = self
            .request_raw_with_query_and_accept(
                Method::GET,
                path,
                None,
                canonical_query.as_deref(),
                headers,
                Some(JsonCodec::CONTENT_TYPE),
            )
            .await?;
        let value = decode_json_value_response(&JsonCodec, &response)?;
        projection.decode_many(value).map_err(ClientError::from)
    }

    pub async fn list_view_paged<P>(
        &self,
        path: &str,
        projection: &P,
        extra_query: &[QueryPair<'_>],
        headers: &[HeaderPair<'_>],
    ) -> Result<Page<P::Output>, ClientError>
    where
        P: Projection,
    {
        let selection = projection.selection_query();
        let canonical_query = canonical_query_from_selection(&selection, extra_query)?;
        let response = self
            .request_raw_with_query_and_accept(
                Method::GET,
                path,
                None,
                canonical_query.as_deref(),
                headers,
                Some(JsonCodec::CONTENT_TYPE),
            )
            .await?;
        let value = decode_json_value_response(&JsonCodec, &response)?;
        projection.decode_page(value).map_err(ClientError::from)
    }

    pub async fn post<Input, Output>(
        &self,
        path: &str,
        input: &Input,
        headers: &[HeaderPair<'_>],
    ) -> Result<Output, ClientError>
    where
        Input: Serialize,
        Output: DeserializeOwned,
    {
        let body = self.codec.encode(input)?;
        let response = self
            .request_raw(Method::POST, path, Some(body), &[], headers)
            .await?;
        decode_typed_response(&self.codec, &response)
    }

    pub async fn post_list<Input, Output>(
        &self,
        path: &str,
        input: &Input,
        headers: &[HeaderPair<'_>],
    ) -> Result<Vec<Output>, ClientError>
    where
        Input: Serialize,
        Output: DeserializeOwned,
    {
        let body = self.codec.encode(input)?;
        let response = self
            .request_raw_with_query_and_accept(
                Method::POST,
                path,
                Some(body),
                None,
                headers,
                Some(self.codec.sequence_accept_header_value()),
            )
            .await?;
        decode_sequence_response(&self.codec, &response)
    }

    pub async fn patch<Input, Output>(
        &self,
        path: &str,
        input: &Input,
        headers: &[HeaderPair<'_>],
    ) -> Result<Output, ClientError>
    where
        Input: Serialize,
        Output: DeserializeOwned,
    {
        let body = self.codec.encode(input)?;
        let response = self
            .request_raw(Method::PATCH, path, Some(body), &[], headers)
            .await?;
        decode_typed_response(&self.codec, &response)
    }

    pub async fn delete<Output>(
        &self,
        path: &str,
        headers: &[HeaderPair<'_>],
    ) -> Result<Output, ClientError>
    where
        Output: DeserializeOwned,
    {
        let response = self
            .request_raw(Method::DELETE, path, None, &[], headers)
            .await?;
        decode_typed_response(&self.codec, &response)
    }

    pub async fn execute_raw_transport(
        &self,
        request: RuntimeRequestWire,
    ) -> Result<RuntimeResponseWire, ClientError> {
        let method = Method::from_bytes(request.method.as_bytes()).map_err(|error| {
            ClientError::BadInput(format!("invalid HTTP method '{}': {error}", request.method))
        })?;
        let header_pairs = request
            .headers
            .iter()
            .map(|header| (header.name.as_str(), header.value.as_str()))
            .collect::<Vec<_>>();
        self.request_raw_with_query(
            method,
            &request.path,
            if request.body.is_empty() {
                None
            } else {
                Some(request.body)
            },
            request.canonical_query.as_deref(),
            &header_pairs,
        )
        .await
    }

    async fn request_raw(
        &self,
        method: Method,
        path: &str,
        body: Option<Vec<u8>>,
        query: &[QueryPair<'_>],
        headers: &[HeaderPair<'_>],
    ) -> Result<RuntimeResponseWire, ClientError> {
        let canonical_query =
            if query.is_empty() {
                None
            } else {
                Some(serde_urlencoded::to_string(query).map_err(|error| {
                    ClientError::BadInput(format!("invalid query pairs: {error}"))
                })?)
            };
        self.request_raw_with_query(method, path, body, canonical_query.as_deref(), headers)
            .await
    }

    async fn request_raw_with_query_and_accept(
        &self,
        method: Method,
        path: &str,
        body: Option<Vec<u8>>,
        canonical_query: Option<&str>,
        headers: &[HeaderPair<'_>],
        accept_override: Option<&'static str>,
    ) -> Result<RuntimeResponseWire, ClientError> {
        let url = build_url(&self.config.base_url, path, canonical_query)?;
        let mut header_map = HeaderMap::new();
        header_map.insert(
            ACCEPT,
            HeaderValue::from_static(
                accept_override.unwrap_or_else(|| self.codec.accept_header_value()),
            ),
        );
        let content_type = if body.is_some() {
            header_map.insert(CONTENT_TYPE, HeaderValue::from_static(C::CONTENT_TYPE));
            Some(C::CONTENT_TYPE.to_owned())
        } else {
            None
        };
        if let Some(authorizer) = &self.request_authorizer {
            let canonical_request = canonical_request_string(
                method.as_str(),
                path,
                canonical_query,
                content_type.as_deref(),
                body.as_deref().unwrap_or(&[]),
            );
            let authorization_request = AuthorizationRequest {
                method: method.as_str().to_owned(),
                path: path.to_owned(),
                canonical_query: canonical_query.map(str::to_owned),
                content_type: content_type.clone(),
                body: body.clone().unwrap_or_default(),
                canonical_request,
            };
            for (name, value) in authorizer.authorize(&authorization_request)? {
                header_map.insert(
                    HeaderName::from_bytes(name.as_bytes()).map_err(|error| {
                        ClientError::BadInput(format!("invalid header name '{name}': {error}"))
                    })?,
                    HeaderValue::from_str(&value).map_err(|error| {
                        ClientError::BadInput(format!("invalid header value for '{name}': {error}"))
                    })?,
                );
            }
        }
        for (name, value) in headers {
            header_map.insert(
                HeaderName::from_bytes(name.as_bytes()).map_err(|error| {
                    ClientError::BadInput(format!("invalid header name '{name}': {error}"))
                })?,
                HeaderValue::from_str(value).map_err(|error| {
                    ClientError::BadInput(format!("invalid header value for '{name}': {error}"))
                })?,
            );
        }

        let mut request = self.http.request(method.clone(), url).headers(header_map);
        if let Some(body) = body {
            request = request.body(body);
        }

        let response = request.send().await?;
        let status = response.status();
        let headers = response.headers().clone();
        let bytes = response.bytes().await?;
        let response_wire = RuntimeResponseWire {
            status_code: status.as_u16(),
            headers: headers_to_runtime(&headers),
            body: bytes.to_vec(),
        };

        self.record_request(method.as_str(), path, status, &headers)?;

        Ok(response_wire)
    }

    async fn request_raw_with_query(
        &self,
        method: Method,
        path: &str,
        body: Option<Vec<u8>>,
        canonical_query: Option<&str>,
        headers: &[HeaderPair<'_>],
    ) -> Result<RuntimeResponseWire, ClientError> {
        self.request_raw_with_query_and_accept(method, path, body, canonical_query, headers, None)
            .await
    }

    fn record_request(
        &self,
        method: &str,
        path: &str,
        status: StatusCode,
        headers: &HeaderMap,
    ) -> Result<(), ClientError> {
        self.state_store
            .append_request_journal(&RequestJournalEntry {
                method: method.to_owned(),
                path: path.to_owned(),
                status_code: status.as_u16(),
                content_type: headers
                    .get(CONTENT_TYPE)
                    .and_then(|value| value.to_str().ok())
                    .map(ToOwned::to_owned),
                recorded_at: Utc::now(),
            })
    }
}

fn decode_typed_response<C, Output>(
    codec: &C,
    response: &RuntimeResponseWire,
) -> Result<Output, ClientError>
where
    C: HttpClientCodec,
    Output: DeserializeOwned,
{
    let content_type = response
        .headers
        .iter()
        .find(|header| header.name.eq_ignore_ascii_case("content-type"))
        .map(|header| header.value.as_str())
        .ok_or_else(|| {
            ClientError::InvalidResponse("response is missing Content-Type header".to_owned())
        })?;

    if (200..=299).contains(&response.status_code) {
        codec
            .decode_response::<Output>(content_type, &response.body)
            .map_err(ClientError::from)
    } else {
        let error = codec
            .decode_response::<CoolErrorResponse>(content_type, &response.body)
            .ok();
        let message = error
            .as_ref()
            .map(|value| value.message.clone())
            .unwrap_or_else(|| {
                format!("unexpected error body for status {}", response.status_code)
            });
        Err(ClientError::Remote {
            status: StatusCode::from_u16(response.status_code)
                .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR),
            error,
            message,
        })
    }
}

fn decode_json_value_response<C>(
    codec: &C,
    response: &RuntimeResponseWire,
) -> Result<JsonValue, ClientError>
where
    C: HttpClientCodec,
{
    decode_typed_response(codec, response)
}

fn decode_sequence_response<C, Output>(
    codec: &C,
    response: &RuntimeResponseWire,
) -> Result<Vec<Output>, ClientError>
where
    C: HttpClientCodec,
    Output: DeserializeOwned,
{
    let content_type = response
        .headers
        .iter()
        .find(|header| header.name.eq_ignore_ascii_case("content-type"))
        .map(|header| header.value.as_str())
        .ok_or_else(|| {
            ClientError::InvalidResponse("response is missing Content-Type header".to_owned())
        })?;

    if (200..=299).contains(&response.status_code) {
        codec
            .decode_sequence_response::<Output>(content_type, &response.body)
            .map_err(ClientError::from)
    } else {
        let error = if media_type_matches(content_type, CBOR_SEQUENCE_CONTENT_TYPE) {
            decode_cbor_sequence::<CoolErrorResponse>(&response.body)
                .ok()
                .and_then(|mut values| {
                    if values.len() == 1 {
                        values.pop()
                    } else {
                        None
                    }
                })
        } else {
            codec
                .decode_response::<CoolErrorResponse>(content_type, &response.body)
                .ok()
        };
        let message = error
            .as_ref()
            .map(|value| value.message.clone())
            .unwrap_or_else(|| {
                format!("unexpected error body for status {}", response.status_code)
            });
        Err(ClientError::Remote {
            status: StatusCode::from_u16(response.status_code)
                .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR),
            error,
            message,
        })
    }
}

fn canonical_query_from_selection(
    selection: &SelectionQuery,
    extra_query: &[QueryPair<'_>],
) -> Result<Option<String>, ClientError> {
    let mut query: Vec<(String, String)> = Vec::new();
    if !selection.fields.is_empty() {
        query.push(("fields".to_owned(), selection.fields.join(",")));
    }
    if !selection.includes.is_empty() {
        query.push(("include".to_owned(), selection.includes.join(",")));
    }
    for (include, fields) in &selection.include_fields {
        if !fields.is_empty() {
            query.push((format!("includeFields[{include}]"), fields.join(",")));
        }
    }
    for (key, value) in extra_query {
        if *key == "fields" || *key == "include" || key.starts_with("includeFields[") {
            return Err(ClientError::BadInput(format!(
                "projection query parameter '{key}' must come from SelectionQuery, not extra_query"
            )));
        }
        query.push(((*key).to_owned(), (*value).to_owned()));
    }
    if query.is_empty() {
        return Ok(None);
    }
    serde_urlencoded::to_string(&query)
        .map(Some)
        .map_err(|error| ClientError::BadInput(format!("invalid selection query: {error}")))
}

fn headers_to_runtime(headers: &HeaderMap) -> Vec<RuntimeHeader> {
    headers
        .iter()
        .filter_map(|(name, value)| {
            value.to_str().ok().map(|value| RuntimeHeader {
                name: name.as_str().to_owned(),
                value: value.to_owned(),
            })
        })
        .collect()
}

fn build_url(
    base_url: &Url,
    path: &str,
    canonical_query: Option<&str>,
) -> Result<Url, ClientError> {
    let mut base = base_url.clone();
    if !base.path().ends_with('/') {
        let next_path = format!("{}/", base.path());
        base.set_path(&next_path);
    }
    let mut url = base.join(path.trim_start_matches('/')).map_err(|error| {
        ClientError::InvalidResponse(format!(
            "failed to resolve path '{path}' against {}: {error}",
            base
        ))
    })?;
    match canonical_query {
        Some(query) if !query.is_empty() => url.set_query(Some(query)),
        _ => url.set_query(None),
    }
    Ok(url)
}

fn media_type_matches(candidate: &str, expected: &str) -> bool {
    candidate.split(';').next().unwrap_or(candidate).trim() == expected
}

fn decode_cbor_sequence<T>(bytes: &[u8]) -> Result<Vec<T>, CoolError>
where
    T: DeserializeOwned,
{
    let mut values = Vec::new();
    let mut offset = 0usize;
    while offset < bytes.len() {
        let mut deserializer = minicbor_serde::Deserializer::new(&bytes[offset..]);
        values.push(T::deserialize(&mut deserializer).map_err(|error| {
            CoolError::Codec(format!("failed to decode CBOR sequence body: {error}"))
        })?);
        let consumed = deserializer.decoder().position();
        if consumed == 0 {
            return Err(CoolError::Codec(
                "failed to decode CBOR sequence body: decoder made no progress".to_owned(),
            ));
        }
        offset += consumed;
    }
    Ok(values)
}

#[cfg(test)]
mod tests {
    use std::path::PathBuf;

    use super::{
        ClientStateStore, JsonFileStateStore, PersistedClientState, RequestJournalEntry,
        RuntimeCodecConfig, RuntimeConfigWire, RuntimeEnvelopeConfig, RuntimeErrorCode,
        RuntimeHandle, RuntimeRequestWire, RuntimeStateStoreConfig, RuntimeTransportConfig,
    };

    #[test]
    fn json_file_store_round_trips_state_under_project_tmp() {
        let path = project_tmp_path("state-store-unit.json");
        if path.exists() {
            std::fs::remove_file(&path).expect("existing tmp file should be removable");
        }

        let store = JsonFileStateStore::new(&path);
        store
            .append_request_journal(&RequestJournalEntry {
                method: "GET".to_owned(),
                path: "/posts".to_owned(),
                status_code: 200,
                content_type: Some("application/cbor".to_owned()),
                recorded_at: chrono::Utc::now(),
            })
            .expect("journal entry should append");

        let loaded = store.load().expect("state should load");
        assert_eq!(loaded.schema_version, 1);
        assert_eq!(loaded.state_version, 1);
        assert_eq!(loaded.request_journal.len(), 1);

        std::fs::remove_file(&path).expect("tmp file should be removable");
    }

    #[test]
    fn runtime_handle_rejects_invalid_method_without_running_http() {
        let handle = RuntimeHandle::new(RuntimeConfigWire {
            base_url: "http://127.0.0.1:1/".to_owned(),
            state_store: RuntimeStateStoreConfig::InMemory,
            transport: RuntimeTransportConfig::default(),
        })
        .expect("runtime handle should build");

        let error = handle
            .execute(RuntimeRequestWire {
                method: "BAD METHOD".to_owned(),
                path: "/posts".to_owned(),
                canonical_query: None,
                headers: Vec::new(),
                body: Vec::new(),
            })
            .expect_err("invalid method should fail before transport");

        assert_eq!(error.code as u32, super::RuntimeErrorCode::BadInput as u32);
    }

    #[test]
    fn persisted_state_defaults_missing_state_version() {
        let state: PersistedClientState =
            serde_json::from_str(r#"{"schema_version":1,"request_journal":[]}"#)
                .expect("legacy state should decode");

        assert_eq!(state.state_version, 0);
    }

    #[test]
    fn runtime_handle_rejects_unsupported_envelope_config() {
        let result = RuntimeHandle::new(RuntimeConfigWire {
            base_url: "http://127.0.0.1:1/".to_owned(),
            state_store: RuntimeStateStoreConfig::InMemory,
            transport: RuntimeTransportConfig {
                codec: RuntimeCodecConfig::Cbor,
                envelope: RuntimeEnvelopeConfig::CoseSign1,
            },
        });

        let error = match result {
            Ok(_) => panic!("unsupported envelope should fail"),
            Err(error) => error,
        };

        assert_eq!(error.code, RuntimeErrorCode::BadInput);
    }

    fn project_tmp_path(file_name: &str) -> PathBuf {
        PathBuf::from(env!("CARGO_MANIFEST_DIR"))
            .join("../../tmp/client-rust-tests")
            .join(file_name)
    }
}