praxis-proxy-filter 0.7.2

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2024 Praxis Contributors

//! The [`HttpFilter`] trait definition.
//!
//! Every HTTP filter implements this trait.

#[cfg(feature = "upstream-binding")]
use std::sync::Arc;

use async_trait::async_trait;
use bytes::Bytes;
use praxis_core::config::InsecureOptions;

pub(crate) use crate::context::HttpFilterContext;
#[cfg(feature = "upstream-binding")]
use crate::pipeline::catalog::{ClusterApplicationCatalog, ClusterMetadataDeclaration};
use crate::{
    actions::{FilterAction, SelectedUpstreamBodyOutcome},
    body::{BodyAccess, BodyMode},
    builtins::http::payload_processing::compression_config::CompressionConfig,
    pipeline::FilterPipeline,
};

// -----------------------------------------------------------------------------
// Backward-compatible Aliases
// -----------------------------------------------------------------------------

/// Backward-compatible alias for [`HttpFilter`].
pub type Filter = dyn HttpFilter;

/// Backward-compatible alias for [`HttpFilterContext`].
///
/// [`HttpFilterContext`]: crate::context::HttpFilterContext
pub type FilterContext<'a> = HttpFilterContext<'a>;

// -----------------------------------------------------------------------------
// HttpFilter Trait
// -----------------------------------------------------------------------------

/// A filter that participates in HTTP request/response processing.
///
/// # Example
///
/// ```
/// use async_trait::async_trait;
/// use praxis_filter::{FilterAction, FilterError, HttpFilter, HttpFilterContext};
///
/// struct NoopFilter;
///
/// #[async_trait]
/// impl HttpFilter for NoopFilter {
///     fn name(&self) -> &'static str {
///         "noop"
///     }
///
///     async fn on_request(
///         &self,
///         _ctx: &mut HttpFilterContext<'_>,
///     ) -> Result<FilterAction, FilterError> {
///         Ok(FilterAction::Continue)
///     }
/// }
///
/// let filter = NoopFilter;
/// assert_eq!(filter.name(), "noop");
/// ```
#[async_trait]
pub trait HttpFilter: Send + Sync {
    /// Unique name identifying this filter type (e.g. `"router"`, `"rate_limit"`).
    fn name(&self) -> &'static str;

    /// Called for each incoming request, in pipeline order.
    async fn on_request(&self, ctx: &mut HttpFilterContext<'_>) -> Result<FilterAction, FilterError>;

    // -------------------------------------------------------------------------
    // Pipeline Capability Declarations
    // -------------------------------------------------------------------------

    /// Whether this filter can assign [`HttpFilterContext::cluster`].
    ///
    /// Pipeline validation uses this to detect `load_balancer`
    /// configurations that cannot receive a cluster selection and to
    /// reject multiple competing cluster selectors before the same
    /// load balancer. Filters that route to named clusters should
    /// override this.
    ///
    /// [`HttpFilterContext::cluster`]: crate::HttpFilterContext::cluster
    fn selects_cluster(&self) -> bool {
        false
    }

    /// The cluster names this filter may select at runtime.
    ///
    /// Return every configured cluster target so pipeline validation
    /// can verify that downstream load balancers define matching
    /// clusters. Filters without static cluster targets can leave the
    /// default empty list.
    fn selected_clusters(&self) -> Vec<String> {
        Vec::new()
    }

    /// Emit this filter's end-of-request record from the logging phase.
    ///
    /// The access log normally emits from its own completion hooks, but
    /// some responses never reach them: a request rejected before the
    /// upstream, a stream aborted mid-body, or a gRPC response whose
    /// body ends with trailers rather than an end-of-stream body chunk.
    /// The protocol layer calls this from the logging phase for those,
    /// where late-arriving facts — the `grpc-status` trailer among them
    /// — are finally available.
    ///
    /// Returns whether a record was emitted. Filters that log nothing
    /// leave the default.
    fn emit_deferred_record(&self, _ctx: &HttpFilterContext<'_>, _status: u16) -> bool {
        false
    }

    /// The cluster names this filter can load balance.
    ///
    /// Load-balancing filters override this so validation can compare
    /// selector outputs against the configured upstream cluster set.
    fn load_balancer_clusters(&self) -> Vec<String> {
        Vec::new()
    }

    /// Application metadata for the clusters this filter declares.
    ///
    /// The pipeline folds every filter's declarations into a single
    /// metadata catalog so a binding `router` can resolve a matched
    /// cluster's opaque application protocol and provider without owning
    /// endpoint state. Load-balancing filters override this and declare the
    /// same names they report from [`load_balancer_clusters`]; a name declared
    /// here but not served there would tag a binding nothing can dispatch.
    /// Filters that declare no clusters leave the default empty list.
    ///
    /// [`load_balancer_clusters`]: HttpFilter::load_balancer_clusters
    #[cfg(feature = "upstream-binding")]
    fn declared_cluster_metadata(&self) -> Vec<ClusterMetadataDeclaration> {
        Vec::new()
    }

    /// Whether this filter publishes the request's logical upstream binding.
    ///
    /// Framework-internal: only the built-in `router` can publish, because the
    /// context API that writes the binding is crate-private, and it reports
    /// `true` only after [`enable_upstream_binding`] handed it a catalog.
    /// Pipeline validation treats the first filter that reports `true` as the
    /// binding router, so an out-of-tree filter must leave the default.
    ///
    /// [`enable_upstream_binding`]: HttpFilter::enable_upstream_binding
    #[doc(hidden)]
    #[cfg(feature = "upstream-binding")]
    fn binds_upstream(&self) -> bool {
        false
    }

    /// Enable logical-upstream publication for this filter.
    ///
    /// Pipeline construction calls this only when the resolved pipeline
    /// contains a bound-upstream observer or consumer, passing the pipeline's
    /// cluster `catalog` so a published binding carries each cluster's
    /// application metadata. The built-in `router` uses the hook to keep
    /// ordinary routing pipelines on their pre-binding fast path; other
    /// filters leave the default no-op implementation.
    #[doc(hidden)]
    #[cfg(feature = "upstream-binding")]
    fn enable_upstream_binding(&mut self, _catalog: Arc<ClusterApplicationCatalog>) {}

    /// Whether this filter selects its cluster from the frozen logical
    /// binding (`BoundUpstream`) rather than a preceding `router`'s
    /// exchange-local [`HttpFilterContext::cluster`].
    ///
    /// A load balancer configured with `cluster_source: bound_upstream`
    /// overrides this. Pipeline validation uses it to reason about direct
    /// branches and IRR-owned pipelines: a bound-consuming load balancer
    /// needs a binding guaranteed on every reachable path and must declare
    /// the bound cluster among its own [`load_balancer_clusters`].
    ///
    /// [`HttpFilterContext::cluster`]: crate::HttpFilterContext::cluster
    /// [`load_balancer_clusters`]: HttpFilter::load_balancer_clusters
    #[cfg(feature = "upstream-binding")]
    fn consumes_bound_upstream(&self) -> bool {
        false
    }

    /// Names of the nested steps this composite filter runs that read the
    /// logical binding, sorted.
    ///
    /// Framework filters that own nested step pipelines (the IRR) override
    /// this. A non-empty list means the filter needs a binding to
    /// exist when its own execution begins, and pipeline validation names these
    /// pipelines when no binding is guaranteed. This is broader than
    /// [`HttpFilter::consumes_bound_upstream`]: a nested condition or body
    /// participant observes the binding without selecting a cluster from it.
    #[cfg(feature = "iterative-request-router")]
    fn nested_bound_upstream_readers(&self) -> Vec<String> {
        Vec::new()
    }

    /// The `when: bound_upstream` matchers inside this composite filter's
    /// nested steps, each with its step name, sorted by step.
    ///
    /// Framework filters that own nested step pipelines (the IRR) override
    /// this so pipeline validation can judge a step's matchers against the
    /// clusters the parent's router can bind, exactly as it judges the parent's
    /// own conditions.
    #[cfg(feature = "iterative-request-router")]
    fn nested_bound_upstream_matchers(&self) -> Vec<(String, praxis_core::config::ApplicationMatch)> {
        Vec::new()
    }

    /// Cluster names this filter can select from the frozen logical binding.
    ///
    /// A load balancer with `cluster_source: bound_upstream` reports the
    /// cluster names it can resolve from `BoundUpstream`; framework filters
    /// that own nested pipelines (the IRR) fold their steps' bound-consuming
    /// declarations up through this hook. Pipeline validation unions these
    /// across every reachable path so a cluster the binding router may bind
    /// must be served by some bound-consuming load balancer. Filters that do
    /// not consume the binding leave the default empty list.
    #[cfg(feature = "upstream-binding")]
    fn bound_upstream_clusters(&self) -> Vec<String> {
        Vec::new()
    }

    /// Whether this filter may select a streaming sub-request response.
    ///
    /// Pipeline validation uses this declaration to reject response
    /// pipelines that require [`BodyMode::StreamBuffer`], which is
    /// incompatible with incremental terminal delivery. Filters that can
    /// call [`HttpFilterContext::set_subrequest_response_mode`] with
    /// `Streaming` must override this method.
    ///
    /// A runtime guard still validates an actual streaming terminal action,
    /// protecting dynamically registered or incorrectly declared filters.
    ///
    /// [`HttpFilterContext::set_subrequest_response_mode`]: crate::HttpFilterContext::set_subrequest_response_mode
    fn may_select_streaming_subrequest_response(&self) -> bool {
        false
    }

    /// Whether this filter may return a terminal [`FilterAction`] — one that
    /// short-circuits the request phase with a response and no upstream
    /// (`TerminalResponse` / `StreamingTerminalResponse`).
    ///
    /// Pipeline validation uses this declaration to reject terminal filters from
    /// outbound chains bound into a filtered sub-request: that executor forwards
    /// to a resolved upstream and cannot surface such a response, so it would
    /// drop the action and then error that no upstream resolved. Any filter that
    /// can return a terminal action must override this, so detection is not
    /// limited to the hard-coded builtin terminal names in [`TERMINAL_FILTERS`].
    ///
    /// Binding validation reads this as "may answer": it keeps the path past
    /// the filter alive and requires that path to be served too. Only the
    /// built-in answering filters (`redirect`, `static_response`, and the IRR)
    /// are treated as always ending the request.
    ///
    /// [`FilterAction`]: crate::FilterAction
    /// [`TERMINAL_FILTERS`]: praxis_core::config::TERMINAL_FILTERS
    fn produces_terminal_response(&self) -> bool {
        false
    }

    /// Visit pipelines owned by this filter.
    ///
    /// Framework filters that embed nested pipelines override this hook so
    /// server-provided runtime resources are propagated into those pipelines.
    /// Ordinary filters should use the default no-op implementation.
    #[doc(hidden)]
    fn visit_nested_pipelines(&mut self, _visitor: &mut dyn FnMut(&mut FilterPipeline)) {}

    /// Called for each response, in reverse pipeline order.
    async fn on_response(&self, ctx: &mut HttpFilterContext<'_>) -> Result<FilterAction, FilterError> {
        let _ = ctx;
        Ok(FilterAction::Continue)
    }

    // -------------------------------------------------------------------------
    // Body Access Declarations
    // -------------------------------------------------------------------------

    /// Declares what access this filter needs to request bodies.
    ///
    /// Return [`BodyAccess::None`] (the default) when the filter
    /// does not inspect or modify request bodies. Return
    /// [`BodyAccess::ReadOnly`] to receive body chunks in
    /// [`on_request_body`] without modification rights, or
    /// [`BodyAccess::ReadWrite`] to mutate body bytes in place.
    ///
    /// This declares the *pre-read* request-body hook, which runs before the
    /// request phase selects an upstream. A filter whose body operation can run
    /// either before routing or only after a logical binding declares this hook
    /// **and** `bound_upstream_request_body_access`; core then schedules the
    /// single hook in exactly one phase based on the filter's conditions (see
    /// that method's dual-access contract).
    ///
    /// [`on_request_body`]: HttpFilter::on_request_body
    fn request_body_access(&self) -> BodyAccess {
        BodyAccess::None
    }

    /// Declares what access this filter needs to response bodies.
    ///
    /// Return [`BodyAccess::None`] (the default) when the filter
    /// does not inspect or modify response bodies. Return
    /// [`BodyAccess::ReadOnly`] to receive body chunks in
    /// [`on_response_body`] without modification rights, or
    /// [`BodyAccess::ReadWrite`] to mutate body bytes in place.
    ///
    /// [`on_response_body`]: HttpFilter::on_response_body
    fn response_body_access(&self) -> BodyAccess {
        BodyAccess::None
    }

    /// Declares what access this filter needs to the request body during
    /// the selected-upstream phase.
    ///
    /// The selected-upstream request-body phase runs after the request
    /// phase has chosen an upstream cluster, giving a filter a final
    /// opportunity to inspect or rewrite the request body for the
    /// specific upstream it is about to be forwarded to. Return
    /// [`BodyAccess::None`] (the default) to opt out,
    /// [`BodyAccess::ReadOnly`] to observe the body in
    /// [`on_selected_upstream_request_body`], or
    /// [`BodyAccess::ReadWrite`] to mutate it.
    ///
    /// A participating filter must declare a bounded
    /// [`BodyMode::StreamBuffer`] via [`request_body_mode`] so the
    /// complete body is buffered before this phase runs; the phase reuses
    /// the same request-body delivery mode rather than defining its own.
    /// Pipeline validation rejects a participant whose
    /// [`request_body_mode`] is not a bounded `StreamBuffer`.
    ///
    /// [`on_selected_upstream_request_body`]: HttpFilter::on_selected_upstream_request_body
    /// [`request_body_mode`]: HttpFilter::request_body_mode
    /// [`BodyMode::StreamBuffer`]: crate::BodyMode::StreamBuffer
    fn selected_upstream_request_body_access(&self) -> BodyAccess {
        BodyAccess::None
    }

    /// Declares what access this filter needs to the request body during
    /// the bound-upstream phase.
    ///
    /// Requires the experimental `bound-upstream-request-body` build feature.
    ///
    /// The bound-upstream request-body phase runs at most once per downstream
    /// request, at the barrier immediately after the `router` binds a logical
    /// upstream (`BoundUpstream`) and before any later request filters
    /// or IRR run. It is skipped when routing stops before a binding is
    /// published. It gives a filter a
    /// chance to inspect or rewrite the request body against the frozen
    /// logical binding, before an endpoint is selected. Return
    /// [`BodyAccess::None`] (the default) to opt out,
    /// [`BodyAccess::ReadOnly`] to observe the body in
    /// [`on_bound_upstream_request_body`], or [`BodyAccess::ReadWrite`] to
    /// mutate it.
    ///
    /// A participating filter must declare a bounded
    /// [`BodyMode::StreamBuffer`] via [`request_body_mode`] so the complete
    /// body is buffered before this phase runs; the phase reuses the same
    /// request-body delivery mode rather than defining its own. Pipeline
    /// validation rejects a participant whose [`request_body_mode`] is not a
    /// bounded `StreamBuffer`, and rejects a participant that no binding
    /// filter can precede.
    ///
    /// # Dual-access contract
    ///
    /// A filter may declare both [`request_body_access`] and this method.
    /// Declaring both means "this body operation can be *deferred* to the
    /// binding barrier when its conditions require the binding" — it never means
    /// "run the hook twice." Core resolves a single effective phase per filter
    /// and runs the hook exactly once:
    ///
    /// | Declared access | Has `bound_upstream` condition | Effective phase |
    /// |---|---|---|
    /// | Pre-read only | No | pre-read ([`on_request_body`]) |
    /// | Pre-read only | Yes | validation error |
    /// | Bound only | Either | bound-upstream ([`on_bound_upstream_request_body`]) |
    /// | Both | No | pre-read ([`on_request_body`]) |
    /// | Both | Yes | bound-upstream ([`on_bound_upstream_request_body`]) |
    ///
    /// So a filter that adapts to the binding when required implements both
    /// hooks, and the operator writes only the policy — e.g. a
    /// `bound_upstream` condition — to defer it. Because a pre-read-only hook
    /// runs before any binding exists, validation rejects a `bound_upstream`
    /// condition on a filter that leaves this method at [`BodyAccess::None`].
    ///
    /// [`on_bound_upstream_request_body`]: HttpFilter::on_bound_upstream_request_body
    /// [`on_request_body`]: HttpFilter::on_request_body
    /// [`request_body_access`]: HttpFilter::request_body_access
    /// [`request_body_mode`]: HttpFilter::request_body_mode
    /// [`BodyMode::StreamBuffer`]: crate::BodyMode::StreamBuffer
    #[cfg(feature = "bound-upstream-request-body")]
    fn bound_upstream_request_body_access(&self) -> BodyAccess {
        BodyAccess::None
    }

    /// Declares the delivery mode for request body chunks.
    ///
    /// [`BodyMode::Stream`] (the default) delivers chunks as they
    /// arrive with minimal latency. [`BodyMode::StreamBuffer`]
    /// accumulates chunks and defers forwarding until the filter
    /// calls [`FilterAction::Release`] or end-of-stream; use this
    /// when the filter needs the full body before making a decision
    /// (e.g. JSON parsing for routing).
    ///
    /// [`FilterAction::Release`]: crate::FilterAction::Release
    fn request_body_mode(&self) -> BodyMode {
        BodyMode::Stream
    }

    /// Declares the delivery mode for response body chunks.
    ///
    /// [`BodyMode::Stream`] (the default) delivers chunks as they
    /// arrive with minimal latency. [`BodyMode::StreamBuffer`]
    /// accumulates chunks and defers forwarding until the filter
    /// calls [`FilterAction::Release`] or end-of-stream; use this
    /// when the filter needs the complete response body (e.g.
    /// response persistence).
    ///
    /// [`FilterAction::Release`]: crate::FilterAction::Release
    fn response_body_mode(&self) -> BodyMode {
        BodyMode::Stream
    }

    /// Whether this filter needs the original request context
    /// during body phases.
    ///
    /// When `true`, the pipeline preserves the original request
    /// headers and URI so they are available in [`on_request_body`]
    /// and [`on_response_body`]. Enable this when body-phase
    /// decisions depend on request metadata (method, path, headers).
    ///
    /// [`on_request_body`]: HttpFilter::on_request_body
    /// [`on_response_body`]: HttpFilter::on_response_body
    fn needs_request_context(&self) -> bool {
        false
    }

    /// Filesystem paths this filter reads its configuration from, beyond the
    /// main Praxis config.
    ///
    /// The config watcher uses this to decide which files should trigger a
    /// reload. A filter whose configuration lives entirely inside the Praxis
    /// config has nothing to declare.
    ///
    /// Paths are returned as configured, not canonicalized; the watcher
    /// resolves them itself.
    fn referenced_files(&self) -> Vec<std::path::PathBuf> {
        Vec::new()
    }

    /// Apply global [`InsecureOptions`] to this filter.
    ///
    /// Filters that support insecure overrides (e.g. CSRF
    /// log-only mode) override this.
    ///
    /// [`InsecureOptions`]: praxis_core::config::InsecureOptions
    fn apply_insecure_options(&self, _options: &InsecureOptions) {}

    /// Returns the compression configuration if this filter enables
    /// response compression.
    ///
    /// Return `Some` to activate response compression with the
    /// given algorithm and level settings. Return `None` (the
    /// default) to leave compression unmodified. Only one filter
    /// in a pipeline should return `Some`.
    fn compression_config(&self) -> Option<&CompressionConfig> {
        None
    }

    // -------------------------------------------------------------------------
    // Body Hooks
    // -------------------------------------------------------------------------

    /// Called for each chunk of request body data, in pipeline order.
    ///
    /// `body` contains the current chunk (`None` when empty).
    /// `end_of_stream` is `true` on the final chunk. Filters may
    /// safely modify `body` before `end_of_stream` when using
    /// [`BodyAccess::ReadWrite`]; buffered modes guarantee all
    /// bytes arrive in a single call with `end_of_stream = true`.
    /// Return [`FilterAction::Reject`] to abort with an error
    /// response.
    ///
    /// # Errors
    ///
    /// Returns [`FilterError`] if body processing fails. The
    /// pipeline converts errors into 500 responses.
    ///
    /// [`FilterAction::Reject`]: crate::FilterAction::Reject
    async fn on_request_body(
        &self,
        ctx: &mut HttpFilterContext<'_>,
        body: &mut Option<Bytes>,
        end_of_stream: bool,
    ) -> Result<FilterAction, FilterError> {
        let _ = (ctx, body, end_of_stream);
        Ok(FilterAction::Continue)
    }

    /// Called for each chunk of response body data, in reverse
    /// pipeline order.
    ///
    /// `body` contains the current chunk (`None` when empty).
    /// `end_of_stream` is `true` on the final chunk. Filters may
    /// safely modify `body` before `end_of_stream` when using
    /// [`BodyAccess::ReadWrite`]; buffered modes guarantee all
    /// bytes arrive in a single call with `end_of_stream = true`.
    /// Return [`FilterAction::Reject`] to abort with an error
    /// response.
    ///
    /// # Errors
    ///
    /// Returns [`FilterError`] if body processing fails. The
    /// pipeline converts errors into 502 responses.
    ///
    /// [`FilterAction::Reject`]: crate::FilterAction::Reject
    fn on_response_body(
        &self,
        ctx: &mut HttpFilterContext<'_>,
        body: &mut Option<Bytes>,
        end_of_stream: bool,
    ) -> Result<FilterAction, FilterError> {
        let _ = (ctx, body, end_of_stream);
        Ok(FilterAction::Continue)
    }

    /// Called once with the fully buffered request body after the request
    /// phase has selected an upstream cluster.
    ///
    /// Runs only for filters that declare
    /// [`selected_upstream_request_body_access`] other than
    /// [`BodyAccess::None`], in pipeline order, and only if the filter
    /// executed during the request phase. `body` holds the complete
    /// request body (`None` when the request had no body). Filters that
    /// declared [`BodyAccess::ReadWrite`] may mutate `body` in place;
    /// this is a working value distinct from any canonical body the
    /// pipeline retains. Return [`SelectedUpstreamBodyOutcome::Reject`] to
    /// abort with an error response; the pipeline stops and does not call
    /// later selected-upstream body filters.
    ///
    /// # Errors
    ///
    /// Returns [`FilterError`] if body processing fails. The pipeline
    /// honors the filter's `failure_mode`: a closed filter's error aborts
    /// the request, an open filter's error is logged and treated as
    /// [`SelectedUpstreamBodyOutcome::Continue`].
    ///
    /// [`selected_upstream_request_body_access`]: HttpFilter::selected_upstream_request_body_access
    /// [`SelectedUpstreamBodyOutcome::Reject`]: crate::SelectedUpstreamBodyOutcome::Reject
    /// [`SelectedUpstreamBodyOutcome::Continue`]: crate::SelectedUpstreamBodyOutcome::Continue
    async fn on_selected_upstream_request_body(
        &self,
        ctx: &mut HttpFilterContext<'_>,
        body: &mut Option<Bytes>,
    ) -> Result<SelectedUpstreamBodyOutcome, FilterError> {
        let _ = (ctx, body);
        Ok(SelectedUpstreamBodyOutcome::Continue)
    }

    /// Called at most once with the fully buffered request body at the
    /// bound-upstream barrier, immediately after the `router` binds a logical
    /// upstream and before any later request filters or the IRR run. It is
    /// not called when routing stops before a binding is published.
    ///
    /// Requires the experimental `bound-upstream-request-body` build feature.
    ///
    /// Runs only for filters that declare
    /// [`bound_upstream_request_body_access`] other than
    /// [`BodyAccess::None`], in pipeline order, and only when the filter's
    /// request conditions (including any `bound_upstream` predicate) match
    /// the frozen binding. Conditions are evaluated at the barrier, not at the
    /// participant's later header-phase position, so they must not depend on
    /// headers or results produced by later request filters. `body` holds the complete request body (`None`
    /// when the request had no body). Filters that declared
    /// [`BodyAccess::ReadWrite`] may mutate `body` in place; the rewritten
    /// buffer becomes the canonical body observed by the remaining pipeline,
    /// direct dispatch, IRR, and retries. Return
    /// [`BoundUpstreamBodyOutcome::Reject`] to abort with an error response;
    /// the pipeline stops and does not call later bound-upstream body
    /// filters.
    ///
    /// # Errors
    ///
    /// Returns [`FilterError`] if body processing fails. The pipeline honors
    /// the filter's `failure_mode`: a closed filter's error aborts the
    /// request, an open filter's error is logged and treated as
    /// [`BoundUpstreamBodyOutcome::Continue`].
    ///
    /// [`bound_upstream_request_body_access`]: HttpFilter::bound_upstream_request_body_access
    /// [`BoundUpstreamBodyOutcome::Reject`]: crate::BoundUpstreamBodyOutcome::Reject
    /// [`BoundUpstreamBodyOutcome::Continue`]: crate::BoundUpstreamBodyOutcome::Continue
    #[cfg(feature = "bound-upstream-request-body")]
    async fn on_bound_upstream_request_body(
        &self,
        ctx: &mut HttpFilterContext<'_>,
        body: &mut Option<Bytes>,
    ) -> Result<crate::BoundUpstreamBodyOutcome, FilterError> {
        let _ = (ctx, body);
        Ok(crate::BoundUpstreamBodyOutcome::Continue)
    }

    /// Whether this filter rewrites upstream response trailers.
    ///
    /// Declaring `false` (the default) lets the protocol layer skip the
    /// trailer hook entirely for pipelines that do not need it.
    fn response_trailer_access(&self) -> bool {
        false
    }

    /// Called once with the upstream response trailers, before they are
    /// forwarded downstream.
    ///
    /// Returning `Some(bytes)` replaces the downstream trailers with
    /// those bytes, appended to the response body as its final chunk.
    /// That is how a gRPC call's status reaches a client that cannot
    /// read HTTP trailers at all.
    ///
    /// Trailers exist only on an HTTP/2 upstream leg, so this never
    /// fires for a cluster left on the default HTTP/1.1.
    ///
    /// # Errors
    ///
    /// Returns [`FilterError`] if the trailers cannot be processed.
    fn on_response_trailers(
        &self,
        ctx: &mut HttpFilterContext<'_>,
        trailers: &mut http::HeaderMap,
    ) -> Result<Option<Bytes>, FilterError> {
        let _ = (ctx, trailers);
        Ok(None)
    }
}

/// Boxed error type for filter results.
pub type FilterError = Box<dyn std::error::Error + Send + Sync>;

// -----------------------------------------------------------------------------
// Tests
// -----------------------------------------------------------------------------

#[cfg(test)]
#[expect(clippy::allow_attributes, reason = "blanket test suppressions")]
#[allow(
    clippy::unwrap_used,
    clippy::expect_used,
    clippy::indexing_slicing,
    clippy::panic,
    reason = "tests"
)]
mod tests {
    use async_trait::async_trait;

    use super::*;

    #[tokio::test]
    async fn default_on_response_returns_continue() {
        let filter = MinimalFilter;
        let req = crate::test_utils::make_request(http::Method::GET, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);

        let action = filter.on_response(&mut ctx).await.unwrap();

        assert!(
            matches!(action, FilterAction::Continue),
            "default on_response should return Continue"
        );
    }

    #[test]
    fn default_body_access_is_none() {
        let filter = MinimalFilter;
        assert_eq!(
            filter.request_body_access(),
            BodyAccess::None,
            "default request body access should be None"
        );
        assert_eq!(
            filter.response_body_access(),
            BodyAccess::None,
            "default response body access should be None"
        );
        assert_eq!(
            filter.selected_upstream_request_body_access(),
            BodyAccess::None,
            "default selected-upstream request body access should be None"
        );
        assert_eq!(
            filter.request_body_mode(),
            BodyMode::Stream,
            "default request body mode should be Stream"
        );
        assert_eq!(
            filter.response_body_mode(),
            BodyMode::Stream,
            "default response body mode should be Stream"
        );
        assert!(
            !filter.needs_request_context(),
            "default needs_request_context should be false"
        );
    }

    #[test]
    fn default_referenced_files_is_empty() {
        let filter = MinimalFilter;
        assert!(
            filter.referenced_files().is_empty(),
            "a filter with no external config must declare nothing"
        );
    }

    #[test]
    fn default_cluster_capabilities_are_empty() {
        let filter = MinimalFilter;
        assert!(!filter.selects_cluster(), "default selects_cluster should be false");
        assert!(
            filter.selected_clusters().is_empty(),
            "default selected_clusters should be empty"
        );
        assert!(
            filter.load_balancer_clusters().is_empty(),
            "default load_balancer_clusters should be empty"
        );
    }

    // -------------------------------------------------------------------------
    // Test Utilities
    // -------------------------------------------------------------------------

    #[tokio::test]
    async fn default_body_hooks_return_continue() {
        let filter = MinimalFilter;
        let req = crate::test_utils::make_request(http::Method::POST, "/");
        let mut ctx = crate::test_utils::make_filter_context(&req);
        let mut body = Some(Bytes::from_static(b"chunk"));

        let request_action = filter.on_request_body(&mut ctx, &mut body, true).await.unwrap();
        assert!(
            matches!(request_action, FilterAction::Continue),
            "default on_request_body should return Continue"
        );

        let response_action = filter.on_response_body(&mut ctx, &mut body, true).unwrap();
        assert!(
            matches!(response_action, FilterAction::Continue),
            "default on_response_body should return Continue"
        );

        let selected_action = filter
            .on_selected_upstream_request_body(&mut ctx, &mut body)
            .await
            .unwrap();
        assert!(
            matches!(selected_action, SelectedUpstreamBodyOutcome::Continue),
            "default on_selected_upstream_request_body should return Continue"
        );
        assert_eq!(
            body.as_deref(),
            Some(b"chunk".as_slice()),
            "defaults must not touch the body"
        );
    }

    #[test]
    fn default_streaming_selection_is_disabled() {
        let filter = MinimalFilter;
        assert!(
            !filter.may_select_streaming_subrequest_response(),
            "filters must opt in to streaming selection"
        );
    }

    /// Minimal filter for verifying trait defaults.
    struct MinimalFilter;

    #[async_trait]
    impl HttpFilter for MinimalFilter {
        fn name(&self) -> &'static str {
            "minimal"
        }

        async fn on_request(&self, _ctx: &mut HttpFilterContext<'_>) -> Result<FilterAction, FilterError> {
            Ok(FilterAction::Continue)
        }
    }
}