Boundaries: HTTP, Retries, Async, Threads, FFI and no_std

use anyhow::Context;
use axum::{
    extract::{Path, State},
    http::{header, StatusCode},
    response::{IntoResponse, Response},
    routing::get,
    Json, Router,
};
use serde_json::json;

#[derive(Debug, thiserror::Error)]
pub enum ApiError {
    #[error("{resource} {id} not found")]
    NotFound { resource: &'static str, id: String },

    #[error("invalid {field}: {reason}")]
    Validation { field: &'static str, reason: String },

    #[error("{0}")]
    Conflict(String),

    #[error("rate limited")]
    RateLimited { retry_after_secs: u64 },

    #[error("upstream {service} unavailable")]
    Unavailable { service: &'static str, #[source] source: anyhow::Error },

    #[error(transparent)]
    Internal(#[from] anyhow::Error),
}

impl ApiError {
    fn status(&self) -> StatusCode {
        match self {
            Self::NotFound { .. } => StatusCode::NOT_FOUND,
            Self::Validation { .. } => StatusCode::UNPROCESSABLE_ENTITY,
            Self::Conflict(_) => StatusCode::CONFLICT,
            Self::RateLimited { .. } => StatusCode::TOO_MANY_REQUESTS,
            Self::Unavailable { .. } => StatusCode::SERVICE_UNAVAILABLE,
            Self::Internal(_) => StatusCode::INTERNAL_SERVER_ERROR,
        }
    }

    fn code(&self) -> &'static str {
        match self {
            Self::NotFound { .. } => "not_found",
            Self::Validation { .. } => "validation_failed",
            Self::Conflict(_) => "conflict",
            Self::RateLimited { .. } => "rate_limited",
            Self::Unavailable { .. } => "unavailable",
            Self::Internal(_) => "internal",
        }
    }
}

impl IntoResponse for ApiError {
    fn into_response(self) -> Response {
        let status = self.status();
        let message = if status.is_server_error() {
            // The only place this error is logged. {:#} on the anyhow source prints the whole chain.
            match &self {
                Self::Internal(err) | Self::Unavailable { source: err, .. } => {
                    tracing::error!(code = self.code(), error = %format!("{err:#}"), "request failed")
                }
                _ => tracing::error!(code = self.code(), error = %self, "request failed"),
            }
            "internal server error".to_owned()
        } else {
            self.to_string()
        };
        let body = Json(json!({ "error": { "code": self.code(), "message": message } }));
        match self {
            Self::RateLimited { retry_after_secs } => {
                (status, [(header::RETRY_AFTER, retry_after_secs.to_string())], body).into_response()
            }
            _ => (status, body).into_response(),
        }
    }
}

async fn get_order(State(db): State<Db>, Path(id): Path<u64>) -> Result<Json<Order>, ApiError> {
    let order = db
        .find_order(id)
        .await
        .context("querying orders")?                       // sqlx::Error -> anyhow -> ApiError::Internal
        .ok_or_else(|| ApiError::NotFound { resource: "order", id: id.to_string() })?;
    Ok(Json(order))
}

pub fn router(db: Db) -> Router {
    Router::new().route("/orders/{id}", get(get_order)).with_state(db)
}

use axum::{extract::{rejection::JsonRejection, State}, http::StatusCode, Json};
use anyhow::Context;

impl From<JsonRejection> for ApiError {
    fn from(rejection: JsonRejection) -> Self {
        ApiError::Validation { field: "body", reason: rejection.body_text() }
    }
}

async fn create_order(
    State(db): State<Db>,
    payload: Result<Json<NewOrder>, JsonRejection>,
) -> Result<(StatusCode, Json<Order>), ApiError> {
    let Json(new) = payload?;                                   // rejection -> ApiError::Validation
    if new.quantity == 0 {
        return Err(ApiError::Validation { field: "quantity", reason: "must be at least 1".into() });
    }
    let order = db.insert_order(&new).await.context("inserting order")?;
    Ok((StatusCode::CREATED, Json(order)))
}

impl From<ApiError> for tonic::Status {
    fn from(err: ApiError) -> Self {
        match err {
            ApiError::NotFound { .. } => tonic::Status::not_found(err.to_string()),
            ApiError::Validation { .. } => tonic::Status::invalid_argument(err.to_string()),
            ApiError::Conflict(_) => tonic::Status::aborted(err.to_string()),
            ApiError::RateLimited { .. } => tonic::Status::resource_exhausted("rate limited"),
            ApiError::Unavailable { .. } => tonic::Status::unavailable("dependency unavailable"),
            ApiError::Internal(e) => {
                tracing::error!(error = %format!("{e:#}"), "rpc failed");
                tonic::Status::internal("internal error")
            }
        }
    }
}

impl From<reqwest::Error> for FetchError {
    fn from(err: reqwest::Error) -> Self {
        let retryable = err.is_timeout()
            || err.is_connect()
            || err.status().is_some_and(|s| s.as_u16() == 429 || s.is_server_error());
        FetchError::Http { retryable, source: err }
    }
}

fn classify_db(err: sqlx::Error) -> StoreError {
    match &err {
        sqlx::Error::RowNotFound => StoreError::NotFound,
        sqlx::Error::Database(db) if db.code().as_deref() == Some("23505") => StoreError::Duplicate,
        sqlx::Error::Database(db) if matches!(db.code().as_deref(), Some("40001" | "40P01")) => {
            StoreError::Transient { source: err }
        }
        sqlx::Error::PoolTimedOut | sqlx::Error::Io(_) => StoreError::Transient { source: err },
        _ => StoreError::Backend { source: err },
    }
}

use std::{future::Future, time::Duration};

pub async fn retry<T, F, Fut>(mut op: F, max_attempts: u32) -> Result<T, StoreError>
where
    F: FnMut() -> Fut,
    Fut: Future<Output = Result<T, StoreError>>,
{
    let mut delay = Duration::from_millis(100);
    let mut attempt = 1;
    loop {
        match op().await {
            Err(err) if err.is_retryable() && attempt < max_attempts => {
                tracing::warn!(attempt, error = %err, "transient failure, retrying");
                // add jitter in production (e.g. delay * random 0.5..1.5) to avoid synchronized retries
                tokio::time::sleep(delay).await;
                delay = (delay * 2).min(Duration::from_secs(10));
                attempt += 1;
            }
            result => return result,
        }
    }
}

// usage
let user = retry(|| store.get_user(id), 5).await?;

// does not compile when spawned:
async fn sync_once() -> Result<(), Box<dyn std::error::Error>> {
    let data = fetch().await?;
    store(data).await?;
    Ok(())
}
// error: future cannot be sent between threads safely
//        ... `dyn std::error::Error` cannot be sent between threads safely

#[test]
fn error_is_send_sync_static() {
    fn assert_bounds<T: std::error::Error + Send + Sync + 'static>() {}
    assert_bounds::<StoreError>();
}

let handle = tokio::spawn(async move { index_batch(batch).await });
match handle.await {
    Ok(Ok(count)) => tracing::info!(count, "batch indexed"),
    Ok(Err(err)) => return Err(err).context("indexing batch"),          // task returned an error
    Err(join_err) if join_err.is_panic() => {
        return Err(anyhow::anyhow!("indexer task panicked"));            // bug: report loudly
    }
    Err(join_err) => return Err(join_err).context("indexer task cancelled"),
}

let handle = std::thread::spawn(move || compress(&input));
let output = match handle.join() {
    Ok(result) => result?,                       // thread's own Result
    Err(payload) => {
        let msg = payload
            .downcast_ref::<&str>()
            .map(|s| s.to_string())
            .or_else(|| payload.downcast_ref::<String>().cloned())
            .unwrap_or_else(|| "non-string panic payload".into());
        anyhow::bail!("compression thread panicked: {msg}");
    }
};

use std::ffi::{c_char, c_int, CStr};
use std::panic::{catch_unwind, AssertUnwindSafe};

pub const KV_OK: c_int = 0;
pub const KV_NOT_FOUND: c_int = 1;
pub const KV_INVALID_ARG: c_int = 2;
pub const KV_INTERNAL: c_int = -1;

/// # Safety
/// `key` must be a valid NUL-terminated string; `out` must be valid for writes.
#[unsafe(no_mangle)] // edition 2024 syntax; plain #[no_mangle] on edition 2021
pub unsafe extern "C" fn kv_get(key: *const c_char, out: *mut u64) -> c_int {
    let result = catch_unwind(AssertUnwindSafe(|| {
        if key.is_null() || out.is_null() {
            return KV_INVALID_ARG;
        }
        // SAFETY: checked non-null above; caller guarantees NUL termination and validity.
        let key = match unsafe { CStr::from_ptr(key) }.to_str() {
            Ok(k) => k,
            Err(_) => return KV_INVALID_ARG,
        };
        match store().get(key) {
            Ok(v) => {
                // SAFETY: checked non-null; caller guarantees `out` is writable.
                unsafe { *out = v };
                KV_OK
            }
            Err(StoreError::NotFound { .. }) => KV_NOT_FOUND,
            Err(err) => {
                set_last_error(format!("{err}"));   // thread-local message for kv_last_error()
                KV_INTERNAL
            }
        }
    }));
    result.unwrap_or(KV_INTERNAL)
}

#![no_std]

use core::fmt;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum FrameError {
    TooShort { len: usize },
    BadChecksum { expected: u16, actual: u16 },
}

impl fmt::Display for FrameError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::TooShort { len } => write!(f, "frame too short ({len} bytes)"),
            Self::BadChecksum { expected, actual } => {
                write!(f, "checksum mismatch: expected {expected:#06x}, got {actual:#06x}")
            }
        }
    }
}

impl core::error::Error for FrameError {}

[dependencies]
thiserror = { version = "2", default-features = false }   # no_std; needs Rust 1.81+

#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
#[non_exhaustive]
pub enum FrameError {
    #[error("frame too short ({len} bytes)")]
    TooShort { len: usize },
    #[error("checksum mismatch: expected {expected:#06x}, got {actual:#06x}")]
    BadChecksum { expected: u16, actual: u16 },
}