yeti_types/plugins/

request.rs

//! HTTP-shaped per-request work and the in-pipeline `Context` service contract.
//!
//! Two request types coexist here, intentionally:
//!
//!  * [`YetiRequest`] / [`YetiResponse`] — the **protocol-edge** shape
//!    (`http::Request<Bytes>` / `http::Response<Bytes>`). Every transport
//!    (HTTP, MQTT, MCP) translates its inbound payload into this, runs
//!    edge-level Services (e.g. `tower-http` CORS / compression / tracing),
//!    and translates the response back. Plugins targeting raw HTTP
//!    edge concerns implement `Service<YetiRequest>`.
//!
//!  * [`ContextService`] — the **internal pipeline** shape. Once the
//!    router has resolved app/resource/path-id and constructed a
//!    [`Context`], every downstream layer (auth resolution, rate
//!    limiting, table dispatch) is a
//!    `Service<Context, Response = Context, Error = YetiError>`.
//!
//! `Context` is `Clone`: every field is `Bytes` / `Arc<str>` /
//! `Arc<dyn Trait>` / small map, so cloning is cheap. With `Clone`,
//! `Context` flows through `tower::Service<YetiRequest>` chains via
//! `http::Request::extensions` (the standard tower-http pattern).

use bytes::Bytes;
use http::{Request, Response};
use tower::util::BoxCloneSyncService;

use crate::error::{Result, YetiError};
use crate::resource::Context;

/// The canonical Yeti request — `http::Request<Bytes>` — for protocol-edge
/// services. The router peels this apart into a [`Context`] before invoking
/// the middleware pipeline.
pub type YetiRequest = Request<Bytes>;

/// The canonical Yeti response — `http::Response<Bytes>`.
pub type YetiResponse = Response<Bytes>;

/// The middleware-pipeline service type. Each registered service consumes
/// a [`Context`] and returns a (possibly mutated) [`Context`], or fails
/// with [`YetiError`].
///
/// This is the Tower-native replacement for the legacy
/// `RequestMiddleware` trait. Build one with `tower::service_fn`:
///
/// ```ignore
/// use yeti_types::plugins::{ContextService, service_fn};
/// use tower::util::BoxCloneSyncService;
///
/// let svc: ContextService = BoxCloneSyncService::new(service_fn(|mut ctx: Context| async move {
///     ctx.set_access(/* ... */);
///     Ok::<_, YetiError>(ctx)
/// }));
/// ```
///
/// Uses `BoxCloneSyncService` (rather than plain `BoxCloneService`)
/// so the type is `Send + Sync` — the router stores its frozen
/// snapshot inside `Arc<[ContextService]>` and the surrounding
/// `AutoRouter` is shared across threads.
pub type ContextService = BoxCloneSyncService<Context, Context, YetiError>;

/// Sequential runner for a chain of [`ContextService`]s. The pipeline
/// owns the services and threads a `Context` through them one after
/// another, short-circuiting on the first error.
///
/// This replaces the legacy `MiddlewareRegistry::process` loop.
pub struct RequestPipeline {
    services: Vec<ContextService>,
}

impl Default for RequestPipeline {
    fn default() -> Self {
        Self::new()
    }
}

impl std::fmt::Debug for RequestPipeline {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("RequestPipeline")
            .field("len", &self.services.len())
            .finish_non_exhaustive()
    }
}

impl RequestPipeline {
    /// Create an empty pipeline.
    #[must_use]
    pub const fn new() -> Self {
        Self {
            services: Vec::new(),
        }
    }

    /// Build a pipeline from a vec of services.
    #[must_use]
    pub const fn from_services(services: Vec<ContextService>) -> Self {
        Self { services }
    }

    /// Push a service onto the end of the pipeline.
    pub fn push(&mut self, svc: ContextService) {
        self.services.push(svc);
    }

    /// Number of registered services.
    #[must_use]
    pub const fn len(&self) -> usize {
        self.services.len()
    }

    /// Whether the pipeline has no services registered.
    #[must_use]
    pub const fn is_empty(&self) -> bool {
        self.services.is_empty()
    }

    /// Run every service in registration order, threading the `Context`
    /// through. Returns the final `Context` once the chain completes,
    /// or the first `YetiError`.
    ///
    /// # Errors
    ///
    /// Surfaces the first `Err` returned by any registered service.
    /// Subsequent services are skipped.
    pub async fn run(&mut self, mut ctx: Context) -> Result<Context> {
        use tower::{Service, ServiceExt};
        for svc in &mut self.services {
            ctx = svc.ready().await?.call(ctx).await?;
        }
        Ok(ctx)
    }

    /// Borrow the underlying service vector (for snapshots / cloning).
    #[must_use]
    pub fn services(&self) -> &[ContextService] {
        &self.services
    }

    /// Compose every registered service into a single `ContextService`.
    /// Per-request the host clones the composed service (one `Arc`
    /// bump) and calls it once — versus the unfrozen path which clones
    /// every service per request and allocates a `Vec`. Used by the
    /// router's `freeze_middleware` to build the hot-path snapshot.
    #[must_use]
    pub fn into_composed(self) -> ContextService {
        compose(self.services)
    }
}

/// Compose a slice of `ContextService`s into a single `ContextService`
/// that runs them in registration order. The composed service is
/// `BoxCloneSyncService`, so cloning it is one `Arc` bump regardless
/// of chain length. Each underlying service still requires a per-call
/// `clone` (to satisfy `&mut self`), but the outer container is shared
/// — so the per-request work scales with chain length only inside the
/// `Future`, not at the boundary.
#[must_use]
pub fn compose(services: Vec<ContextService>) -> ContextService {
    let services: std::sync::Arc<[ContextService]> = services.into();
    let services_for_fn = std::sync::Arc::clone(&services);
    BoxCloneSyncService::new(tower::service_fn(move |ctx: Context| {
        let services = std::sync::Arc::clone(&services_for_fn);
        async move {
            use tower::{Service, ServiceExt};
            let mut current = ctx;
            for svc in services.iter() {
                let mut svc = svc.clone();
                current = svc.ready().await?.call(current).await?;
            }
            Ok::<Context, YetiError>(current)
        }
    }))
}