hyperi_rustlib/transport/

http.rs

// Project:   hyperi-rustlib
// File:      src/transport/http.rs
// Purpose:   HTTP/HTTPS transport (send via POST, receive via embedded server)
// Language:  Rust
//
// License:   BUSL-1.1
// Copyright: (c) 2026 HYPERI PTY LIMITED

//! # HTTP Transport
//!
//! HTTP/HTTPS transport for webhook delivery and REST ingest.
//!
//! ## Send
//!
//! POSTs payload bytes to `{endpoint}/{key}` using reqwest.
//!
//! ## Receive (requires `http-server` feature)
//!
//! Starts an embedded axum HTTP server that accepts POST requests on a
//! configurable path. Incoming payloads are queued into a bounded
//! `tokio::sync::mpsc` channel. `recv()` drains from this channel.
//!
//! ## Example
//!
//! ```rust,ignore
//! use hyperi_rustlib::transport::http::{HttpTransport, HttpTransportConfig};
//!
//! // Send-only
//! let config = HttpTransportConfig {
//!     endpoint: Some("http://loader:8080/ingest".into()),
//!     ..Default::default()
//! };
//! let transport = HttpTransport::new(&config).await?;
//! transport.send("events", bytes::Bytes::from_static(b"{\"msg\":\"hello\"}")).await;
//! ```

use super::error::{TransportError, TransportResult};
#[cfg(feature = "http-server")]
use super::traits::RecvBatch;
use super::traits::{CommitToken, TransportBase, TransportReceiver, TransportSender};
#[cfg(feature = "http-server")]
use super::types::Message;
#[cfg(feature = "http-server")]
use super::types::PayloadFormat;
use super::types::SendResult;
use super::work_batch::WorkBatch;
use serde::{Deserialize, Serialize};
use std::sync::Arc;
#[cfg(feature = "http-server")]
use std::sync::atomic::AtomicU64;
use std::sync::atomic::{AtomicBool, Ordering};

/// Commit token for HTTP transport.
///
/// HTTP is fire-and-forget from the receiver's perspective, so commit
/// is a no-op. The token provides sequence tracking and optional
/// client address for observability.
#[derive(Debug, Clone)]
pub struct HttpToken {
    /// Local sequence number (monotonically increasing per transport instance).
    pub seq: u64,

    /// Source client address (if available from the HTTP request).
    pub source_addr: Option<String>,
}

impl HttpToken {
    /// Create a new token with sequence number.
    #[must_use]
    pub fn new(seq: u64) -> Self {
        Self {
            seq,
            source_addr: None,
        }
    }

    /// Create a new token with sequence number and source address.
    #[must_use]
    pub fn with_source(seq: u64, addr: String) -> Self {
        Self {
            seq,
            source_addr: Some(addr),
        }
    }
}

impl CommitToken for HttpToken {}

impl std::fmt::Display for HttpToken {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match &self.source_addr {
            Some(addr) => write!(f, "http:{}:{}", addr, self.seq),
            None => write!(f, "http:{}", self.seq),
        }
    }
}

fn default_recv_path() -> String {
    "/ingest".to_string()
}

fn default_buffer_size() -> usize {
    10_000
}

fn default_recv_timeout_ms() -> u64 {
    100
}

fn default_connect_timeout_ms() -> u64 {
    5_000
}

fn default_send_timeout_ms() -> u64 {
    30_000
}

fn default_max_body_bytes() -> usize {
    16 * 1024 * 1024
}

/// Configuration for HTTP transport.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HttpTransportConfig {
    /// Endpoint URL for sending (e.g., "http://loader:8080/ingest"). None = send disabled.
    #[serde(default)]
    pub endpoint: Option<String>,

    /// Listen address for receiving (e.g., "0.0.0.0:8080"). None = receive disabled.
    /// Requires the `http-server` feature.
    #[serde(default)]
    pub listen: Option<String>,

    /// Path to accept POSTs on for receive mode. Default: "/ingest".
    #[serde(default = "default_recv_path")]
    pub recv_path: String,

    /// Receive buffer size (bounded channel capacity). Default: 10000.
    #[serde(default = "default_buffer_size")]
    pub recv_buffer_size: usize,

    /// Receive timeout in milliseconds. Default: 100.
    #[serde(default = "default_recv_timeout_ms")]
    pub recv_timeout_ms: u64,

    /// Inbound message filters (applied on recv before caller sees messages).
    #[serde(default)]
    pub filters_in: Vec<super::filter::FilterRule>,

    /// Outbound message filters (applied on send before transport dispatches).
    #[serde(default)]
    pub filters_out: Vec<super::filter::FilterRule>,

    /// Connect timeout (ms) for the send-side HTTP client. Default 5000.
    #[serde(default = "default_connect_timeout_ms")]
    pub connect_timeout_ms: u64,

    /// Total request timeout (ms) per send. Default 30000. Prevents a hung
    /// send from consuming worker capacity indefinitely.
    #[serde(default = "default_send_timeout_ms")]
    pub send_timeout_ms: u64,

    /// Maximum accepted request body size in bytes (receive side). Default
    /// 16 MiB. Oversized POSTs are rejected with 413 before buffering.
    #[serde(default = "default_max_body_bytes")]
    pub max_body_bytes: usize,

    /// Private-CA PEM path for the send-side HTTPS client (needs the `tls`
    /// feature). Trusts this CA *in addition to* the OS native roots. None =
    /// native roots only (reqwest default). Maps to `TlsTrust { native_roots:
    /// true, extra_roots: [ca] }`.
    #[serde(default)]
    pub tls_ca_path: Option<String>,
}

impl Default for HttpTransportConfig {
    fn default() -> Self {
        Self {
            endpoint: None,
            listen: None,
            recv_path: default_recv_path(),
            recv_buffer_size: default_buffer_size(),
            recv_timeout_ms: default_recv_timeout_ms(),
            filters_in: Vec::new(),
            filters_out: Vec::new(),
            connect_timeout_ms: default_connect_timeout_ms(),
            send_timeout_ms: default_send_timeout_ms(),
            max_body_bytes: default_max_body_bytes(),
            tls_ca_path: None,
        }
    }
}

impl HttpTransportConfig {
    /// Load from the config cascade under the `transport.http` key.
    #[must_use]
    pub fn from_cascade() -> Self {
        <Self as super::traits::FromCascade>::from_cascade_key("transport.http")
    }

    /// Create a send-only config pointing at the given endpoint URL.
    #[must_use]
    pub fn sender(endpoint: &str) -> Self {
        Self {
            endpoint: Some(endpoint.to_string()),
            ..Default::default()
        }
    }

    /// Create a receive-only config listening on the given address.
    #[must_use]
    pub fn receiver(listen: &str) -> Self {
        Self {
            listen: Some(listen.to_string()),
            ..Default::default()
        }
    }
}

/// HTTP/HTTPS transport.
///
/// Supports send (POST to endpoint) and receive (embedded axum server).
/// The receive side requires the `http-server` feature for axum.
pub struct HttpTransport {
    /// reqwest client for sending (always available when transport-http is enabled).
    client: reqwest::Client,

    /// Base URL for sending (None = send disabled).
    endpoint: Option<String>,

    /// Receiver channel populated by the embedded HTTP server.
    /// Only available when `http-server` feature is enabled AND `listen` is configured.
    #[cfg(feature = "http-server")]
    receiver: Option<tokio::sync::Mutex<tokio::sync::mpsc::Receiver<Message<HttpToken>>>>,

    /// Shutdown signal for the server task.
    ///
    /// Behind a `parking_lot::Mutex` for interior mutability: `close(&self)`
    /// fires it to stop the embedded server promptly (graceful shutdown),
    /// rather than waiting for `Drop`. `take()` makes both paths idempotent.
    #[cfg(feature = "http-server")]
    shutdown_tx: parking_lot::Mutex<Option<tokio::sync::oneshot::Sender<()>>>,

    /// Server background task handle.
    #[cfg(feature = "http-server")]
    _server_handle: Option<tokio::task::JoinHandle<()>>,

    /// Whether the transport is closed.
    closed: Arc<AtomicBool>,

    /// Receive timeout in milliseconds (used by receive side).
    #[cfg(feature = "http-server")]
    recv_timeout_ms: u64,

    /// Transport-level message filter engine.
    filter_engine: super::filter::TransportFilterEngine,
}

impl HttpTransport {
    /// Create a new HTTP transport.
    ///
    /// - Set `config.endpoint` to enable sending (POST to URL).
    /// - Set `config.listen` to enable receiving (embedded HTTP server, requires `http-server` feature).
    ///
    /// # Errors
    ///
    /// Returns error if the listen address is invalid or the server fails to bind.
    pub async fn new(config: &HttpTransportConfig) -> TransportResult<Self> {
        // Default path: no pressure governor -> byte-identical to before.
        Self::new_inner(
            config,
            #[cfg(feature = "governor")]
            None,
        )
        .await
    }

    /// Create an HTTP transport bound to a pressure governor (G3, `governor`
    /// feature).
    ///
    /// Identical to [`new`](Self::new) except the embedded receive server
    /// consults `pressure` BEFORE enqueuing each request: while
    /// [`UnifiedPressure::should_hold`](crate::governor::UnifiedPressure::should_hold)
    /// holds, the handler returns 503 (SERVICE_UNAVAILABLE) -- the same status
    /// the existing channel-full backpressure path uses. Passing `None` is
    /// exactly equivalent to [`new`](Self::new).
    ///
    /// # Errors
    ///
    /// Same as [`new`](Self::new).
    #[cfg(feature = "governor")]
    pub async fn with_pressure(
        config: &HttpTransportConfig,
        pressure: Option<Arc<crate::governor::UnifiedPressure>>,
    ) -> TransportResult<Self> {
        Self::new_inner(config, pressure).await
    }

    async fn new_inner(
        config: &HttpTransportConfig,
        #[cfg(feature = "governor")] pressure: Option<Arc<crate::governor::UnifiedPressure>>,
    ) -> TransportResult<Self> {
        // `mut` is only used on the TLS path; harmless without the feature.
        #[cfg_attr(not(feature = "tls"), allow(unused_mut))]
        let mut client_builder = reqwest::Client::builder()
            .connect_timeout(std::time::Duration::from_millis(config.connect_timeout_ms))
            .timeout(std::time::Duration::from_millis(config.send_timeout_ms));

        // Private-CA trust via the unified TLS module (augments native roots).
        #[cfg(feature = "tls")]
        if let Some(ref ca) = config.tls_ca_path {
            let trust = crate::tls::TlsTrust {
                native_roots: true,
                webpki_roots: false,
                extra_roots: vec![ca.into()],
                extra_intermediates: Vec::new(),
                exclusive: false,
            };
            let tls_cfg =
                crate::tls::build_client_config(crate::tls::TlsConfigSource::Trust(trust))
                    .map_err(|e| TransportError::Config(format!("HTTP client TLS: {e}")))?;
            client_builder = client_builder.use_preconfigured_tls((*tls_cfg).clone());
        }
        #[cfg(not(feature = "tls"))]
        if config.tls_ca_path.is_some() {
            tracing::warn!(
                "http transport tls_ca_path is set but the `tls` feature is disabled -- \
                 ignoring (using reqwest default roots)"
            );
        }

        let client = client_builder
            .build()
            .map_err(|e| TransportError::Config(format!("failed to create HTTP client: {e}")))?;

        #[cfg(feature = "http-server")]
        let (receiver, shutdown_tx, server_handle) = if let Some(listen) = &config.listen {
            let addr: std::net::SocketAddr = listen
                .parse()
                .map_err(|e| TransportError::Config(format!("invalid listen address: {e}")))?;

            let (tx, rx) = tokio::sync::mpsc::channel(config.recv_buffer_size);
            let (sd_tx, sd_rx) = tokio::sync::oneshot::channel::<()>();

            let sequence = Arc::new(AtomicU64::new(0));
            let recv_path = config.recv_path.clone();

            let app = build_receiver_router(
                tx,
                sequence,
                &recv_path,
                config.max_body_bytes,
                #[cfg(feature = "governor")]
                pressure.clone(),
            );

            let listener = tokio::net::TcpListener::bind(addr).await.map_err(|e| {
                TransportError::Connection(format!("failed to bind to {addr}: {e}"))
            })?;

            let handle = tokio::spawn(async move {
                axum::serve(
                    listener,
                    app.into_make_service_with_connect_info::<std::net::SocketAddr>(),
                )
                .with_graceful_shutdown(async {
                    sd_rx.await.ok();
                })
                .await
                .ok();
            });

            (Some(tokio::sync::Mutex::new(rx)), Some(sd_tx), Some(handle))
        } else {
            (None, None, None)
        };

        // When the governor feature is on but http-server is off there is no
        // receive server to attach the pressure governor to; consume it so the
        // signature stays uniform without an unused-variable warning.
        #[cfg(all(feature = "governor", not(feature = "http-server")))]
        let _ = pressure;

        #[cfg(feature = "logger")]
        tracing::info!(
            endpoint = ?config.endpoint,
            listen = ?config.listen,
            "HTTP transport opened"
        );

        // Fail loud on bad filter config -- silently disabling filters
        // turns a misconfigured `drop` / `dlq` rule into a permanent pass.
        let filter_engine = super::filter::TransportFilterEngine::new(
            &config.filters_in,
            &config.filters_out,
            &crate::transport::filter::TransportFilterTierConfig::from_cascade(),
        )?;

        let closed = Arc::new(AtomicBool::new(false));

        #[cfg(feature = "health")]
        {
            let h = Arc::clone(&closed);
            crate::health::HealthRegistry::register("transport:http", move || {
                if h.load(Ordering::Relaxed) {
                    crate::health::HealthStatus::Unhealthy
                } else {
                    crate::health::HealthStatus::Healthy
                }
            });
        }

        Ok(Self {
            client,
            endpoint: config.endpoint.clone(),
            #[cfg(feature = "http-server")]
            receiver,
            #[cfg(feature = "http-server")]
            shutdown_tx: parking_lot::Mutex::new(shutdown_tx),
            #[cfg(feature = "http-server")]
            _server_handle: server_handle,
            closed,
            #[cfg(feature = "http-server")]
            recv_timeout_ms: config.recv_timeout_ms,
            filter_engine,
        })
    }
}

/// Build the axum router for the receive side.
#[cfg(feature = "http-server")]
fn build_receiver_router(
    sender: tokio::sync::mpsc::Sender<Message<HttpToken>>,
    sequence: Arc<AtomicU64>,
    recv_path: &str,
    max_body_bytes: usize,
    #[cfg(feature = "governor")] pressure: Option<Arc<crate::governor::UnifiedPressure>>,
) -> axum::Router {
    use axum::routing::post;

    let state = ReceiverState {
        sender,
        sequence,
        #[cfg(feature = "governor")]
        pressure,
    };

    axum::Router::new()
        .route(recv_path, post(ingest_handler))
        // Reject oversized bodies with 413 before the handler buffers them.
        .layer(axum::extract::DefaultBodyLimit::max(max_body_bytes))
        .with_state(state)
}

/// Shared state for the receive handler.
#[cfg(feature = "http-server")]
#[derive(Clone)]
struct ReceiverState {
    sender: tokio::sync::mpsc::Sender<Message<HttpToken>>,
    sequence: Arc<AtomicU64>,
    /// Optional pressure governor (G3, `governor` feature). `None` by default
    /// -> the handler never consults it and behaviour is byte-identical. When
    /// `Some`, the handler rejects with 503 while [`UnifiedPressure::should_hold`]
    /// holds -- pressure-driven shedding ON TOP of the existing channel-full
    /// 503, never replacing it.
    #[cfg(feature = "governor")]
    pressure: Option<Arc<crate::governor::UnifiedPressure>>,
}

/// POST handler that accepts raw bytes and queues them into the mpsc channel.
#[cfg(feature = "http-server")]
async fn ingest_handler(
    axum::extract::State(state): axum::extract::State<ReceiverState>,
    axum::extract::ConnectInfo(addr): axum::extract::ConnectInfo<std::net::SocketAddr>,
    headers: axum::http::HeaderMap,
    body: axum::body::Bytes,
) -> axum::response::Response {
    use axum::response::IntoResponse as _;

    if body.is_empty() {
        return axum::http::StatusCode::BAD_REQUEST.into_response();
    }

    // G3 pressure-driven shedding (governor feature, opt-in). BEFORE enqueuing,
    // if a governor is wired and it says hold, shed the request with 503 --
    // consistent with the existing channel-full 503 below (NOT 429). Default
    // `None` -> this is skipped and behaviour is byte-identical.
    #[cfg(feature = "governor")]
    if let Some(pressure) = &state.pressure
        && pressure.should_hold()
    {
        #[cfg(feature = "metrics")]
        metrics::counter!("dfe_transport_backpressured_total", "transport" => "http", "reason" => "pressure")
            .increment(1);
        return shed_503();
    }

    // Extract W3C traceparent from incoming HTTP headers for distributed tracing
    #[cfg(feature = "transport-trace")]
    if let Some(tp) = headers
        .get(super::propagation::TRACEPARENT_HEADER)
        .and_then(|v| v.to_str().ok())
        && super::propagation::is_valid_traceparent(tp)
    {
        tracing::Span::current().record("traceparent", tp);
    }

    // Suppress unused variable warning when otel feature is disabled
    #[cfg(not(feature = "otel"))]
    let _ = &headers;

    let seq = state.sequence.fetch_add(1, Ordering::Relaxed);
    let format = PayloadFormat::detect(&body);
    let timestamp_ms = chrono::Utc::now().timestamp_millis();

    // `body` is `axum::body::Bytes` (= `bytes::Bytes`) -- move it directly,
    // no copy needed.
    let msg = Message {
        key: None,
        payload: body,
        token: HttpToken::with_source(seq, addr.to_string()),
        timestamp_ms: Some(timestamp_ms),
        format,
    };

    match state.sender.try_send(msg) {
        Ok(()) => {
            #[cfg(feature = "metrics")]
            metrics::counter!("dfe_transport_sent_total", "transport" => "http").increment(1);
            axum::http::StatusCode::OK.into_response()
        }
        Err(tokio::sync::mpsc::error::TrySendError::Full(_)) => {
            #[cfg(feature = "metrics")]
            metrics::counter!("dfe_transport_backpressured_total", "transport" => "http")
                .increment(1);
            shed_503()
        }
        Err(tokio::sync::mpsc::error::TrySendError::Closed(_)) => {
            #[cfg(feature = "metrics")]
            metrics::counter!("dfe_transport_refused_total", "transport" => "http").increment(1);
            axum::http::StatusCode::GONE.into_response()
        }
    }
}

/// 503 shed response with a `Retry-After: 1` hint, so a well-behaved sender
/// backs off briefly instead of hot-retrying into a pod that is already holding.
#[cfg(feature = "http-server")]
fn shed_503() -> axum::response::Response {
    use axum::response::IntoResponse as _;
    (
        axum::http::StatusCode::SERVICE_UNAVAILABLE,
        [(axum::http::header::RETRY_AFTER, "1")],
    )
        .into_response()
}

impl TransportBase for HttpTransport {
    async fn close(&self) -> TransportResult<()> {
        self.closed.store(true, Ordering::Relaxed);
        // Stop the embedded server now (graceful shutdown) rather than on Drop.
        // take() => idempotent: a later close()/drop() is a no-op.
        #[cfg(feature = "http-server")]
        if let Some(tx) = self.shutdown_tx.lock().take() {
            let _ = tx.send(());
        }
        Ok(())
    }

    fn is_healthy(&self) -> bool {
        !self.closed.load(Ordering::Relaxed)
    }

    fn name(&self) -> &'static str {
        "http"
    }
}

impl TransportSender for HttpTransport {
    async fn send(&self, key: &str, payload: bytes::Bytes) -> SendResult {
        if self.closed.load(Ordering::Relaxed) {
            return SendResult::Fatal(TransportError::Closed);
        }

        // Outbound filter check
        if self.filter_engine.has_outbound_filters() {
            match self.filter_engine.apply_outbound(&payload) {
                super::filter::FilterDisposition::Pass => {}
                super::filter::FilterDisposition::Drop => return SendResult::Ok,
                super::filter::FilterDisposition::Dlq => return SendResult::FilteredDlq,
            }
        }

        let Some(base_url) = &self.endpoint else {
            return SendResult::Fatal(TransportError::Config(
                "no endpoint configured for sending".into(),
            ));
        };

        // Build URL: {base_url}/{key} if key is non-empty, otherwise just {base_url}
        let url = if key.is_empty() {
            base_url.clone()
        } else {
            let base = base_url.trim_end_matches('/');
            let suffix = key.trim_start_matches('/');
            format!("{base}/{suffix}")
        };

        #[cfg(feature = "metrics")]
        let start = std::time::Instant::now();

        // Build request with optional W3C traceparent header for distributed tracing
        let request_builder = self
            .client
            .post(&url)
            .header("content-type", "application/octet-stream");

        #[cfg(feature = "transport-trace")]
        let request_builder = if let Some(tp) = super::propagation::current_traceparent() {
            request_builder.header(super::propagation::TRACEPARENT_HEADER, tp)
        } else {
            request_builder
        };

        #[cfg(feature = "logger")]
        let payload_len = payload.len();
        let result = match request_builder.body(payload).send().await {
            Ok(resp) if resp.status().is_success() => {
                #[cfg(feature = "logger")]
                tracing::debug!(url = %url, bytes = payload_len, "HTTP transport: POST sent");

                #[cfg(feature = "metrics")]
                metrics::counter!("dfe_transport_sent_total", "transport" => "http").increment(1);
                SendResult::Ok
            }
            Ok(resp)
                if resp.status() == reqwest::StatusCode::TOO_MANY_REQUESTS
                    || resp.status() == reqwest::StatusCode::SERVICE_UNAVAILABLE =>
            {
                #[cfg(feature = "logger")]
                tracing::warn!(status = %resp.status(), url = %url, "HTTP transport: backpressure");

                #[cfg(feature = "metrics")]
                metrics::counter!("dfe_transport_backpressured_total", "transport" => "http")
                    .increment(1);
                SendResult::Backpressured
            }
            Ok(resp) => {
                #[cfg(feature = "logger")]
                tracing::warn!(status = %resp.status(), url = %url, "HTTP transport: send error");

                #[cfg(feature = "metrics")]
                metrics::counter!("dfe_transport_send_errors_total", "transport" => "http")
                    .increment(1);
                SendResult::Fatal(TransportError::Send(format!(
                    "HTTP {} from {}",
                    resp.status(),
                    url
                )))
            }
            Err(e) => {
                #[cfg(feature = "logger")]
                tracing::warn!(error = %e, url = %url, "HTTP transport: request failed");

                #[cfg(feature = "metrics")]
                metrics::counter!("dfe_transport_send_errors_total", "transport" => "http")
                    .increment(1);
                SendResult::Fatal(TransportError::Send(format!("HTTP request failed: {e}")))
            }
        };

        #[cfg(feature = "metrics")]
        metrics::histogram!("dfe_transport_send_duration_seconds", "transport" => "http")
            .record(start.elapsed().as_secs_f64());

        result
    }
}

impl TransportReceiver for HttpTransport {
    type Token = HttpToken;

    async fn recv(&self, max: usize) -> TransportResult<WorkBatch<Self::Token>> {
        if self.closed.load(Ordering::Relaxed) {
            return Err(TransportError::Closed);
        }

        #[cfg(feature = "http-server")]
        {
            let Some(receiver) = &self.receiver else {
                return Err(TransportError::Config(
                    "no listen address configured for receiving".into(),
                ));
            };

            let mut rx = receiver.lock().await;
            let mut messages = Vec::with_capacity(max.min(100));

            for _ in 0..max {
                let result = if self.recv_timeout_ms == 0 {
                    match rx.try_recv() {
                        Ok(msg) => Some(msg),
                        Err(tokio::sync::mpsc::error::TryRecvError::Empty) => break,
                        Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => {
                            return Err(TransportError::Closed);
                        }
                    }
                } else if messages.is_empty() {
                    // First message: wait with timeout
                    match tokio::time::timeout(
                        std::time::Duration::from_millis(self.recv_timeout_ms),
                        rx.recv(),
                    )
                    .await
                    {
                        Ok(Some(msg)) => Some(msg),
                        Ok(None) => return Err(TransportError::Closed),
                        Err(_) => break, // Timeout
                    }
                } else {
                    // Subsequent: non-blocking drain
                    match rx.try_recv() {
                        Ok(msg) => Some(msg),
                        Err(_) => break,
                    }
                };

                if let Some(msg) = result {
                    messages.push(msg);
                }
            }

            // Apply inbound filters via the shared partition helper; DLQ
            // entries are returned in the RecvBatch for the caller to route.
            let batch = self.filter_engine.partition_batch(
                messages,
                |m| m.payload.as_ref(),
                |m| m.key.clone(),
                |m| m.token.clone(),
            );
            let messages = batch.messages;
            let dlq_entries = batch.dlq_entries;
            let filtered_tokens = batch.filtered_tokens;

            #[cfg(feature = "logger")]
            if !messages.is_empty() {
                tracing::debug!(messages = messages.len(), "HTTP transport: batch received");
            }

            // New default (metrics audit): inbound throughput (was only emitted
            // by file/pipe/redis). Counts post-filter delivered messages.
            #[cfg(feature = "metrics")]
            if !messages.is_empty() {
                metrics::counter!("dfe_transport_received_total", "transport" => "http")
                    .increment(messages.len() as u64);
            }

            Ok(RecvBatch {
                messages,
                dlq_entries,
                filtered_tokens,
            }
            .into())
        }

        #[cfg(not(feature = "http-server"))]
        {
            let _ = max;
            Err(TransportError::Config(
                "HTTP receive requires the 'http-server' feature".into(),
            ))
        }
    }

    async fn commit(&self, _tokens: &[Self::Token]) -> TransportResult<()> {
        // HTTP is fire-and-forget -- commit is a no-op.
        Ok(())
    }
}

impl Drop for HttpTransport {
    fn drop(&mut self) {
        #[cfg(feature = "http-server")]
        if let Some(tx) = self.shutdown_tx.lock().take() {
            let _ = tx.send(());
        }
    }
}

impl super::traits::FromCascade for HttpTransportConfig {}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn http_token_display() {
        let token = HttpToken::new(42);
        assert_eq!(format!("{token}"), "http:42");
    }

    #[test]
    fn http_token_display_with_source() {
        let token = HttpToken::with_source(7, "192.168.1.1:54321".to_string());
        assert_eq!(format!("{token}"), "http:192.168.1.1:54321:7");
    }

    #[test]
    fn config_defaults() {
        let config = HttpTransportConfig::default();
        assert!(config.endpoint.is_none());
        assert!(config.listen.is_none());
        assert_eq!(config.recv_path, "/ingest");
        assert_eq!(config.recv_buffer_size, 10_000);
        assert_eq!(config.recv_timeout_ms, 100);
    }

    #[test]
    fn config_sender_helper() {
        let config = HttpTransportConfig::sender("http://localhost:8080/ingest");
        assert_eq!(
            config.endpoint.as_deref(),
            Some("http://localhost:8080/ingest")
        );
        assert!(config.listen.is_none());
    }

    #[test]
    fn config_receiver_helper() {
        let config = HttpTransportConfig::receiver("0.0.0.0:8080");
        assert!(config.endpoint.is_none());
        assert_eq!(config.listen.as_deref(), Some("0.0.0.0:8080"));
    }

    #[tokio::test]
    async fn send_only_transport() {
        // Send-only config (no endpoint = send disabled, but transport creates fine)
        let config = HttpTransportConfig::default();
        let transport = HttpTransport::new(&config).await.unwrap();

        assert!(transport.is_healthy());
        assert_eq!(transport.name(), "http");

        // Send without endpoint should fail
        let result = transport
            .send("test", bytes::Bytes::from_static(b"payload"))
            .await;
        assert!(result.is_fatal());

        // Commit is always ok
        transport.commit(&[]).await.unwrap();
    }

    #[tokio::test]
    async fn close_prevents_send() {
        let config = HttpTransportConfig::sender("http://localhost:19999/test");
        let transport = HttpTransport::new(&config).await.unwrap();

        transport.close().await.unwrap();
        assert!(!transport.is_healthy());

        let result = transport
            .send("test", bytes::Bytes::from_static(b"data"))
            .await;
        assert!(result.is_fatal());
    }

    #[tokio::test]
    async fn close_prevents_recv() {
        let config = HttpTransportConfig::default();
        let transport = HttpTransport::new(&config).await.unwrap();

        transport.close().await.unwrap();
        let result = transport.recv(1).await;
        assert!(result.is_err());
    }

    /// Full send + receive round-trip test.
    /// Requires both `transport-http` and `http-server` features.
    #[cfg(feature = "http-server")]
    #[tokio::test]
    async fn send_and_receive_roundtrip() {
        // Start receiver on a random available port
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        drop(listener); // Free the port for the transport to bind

        let recv_config = HttpTransportConfig {
            listen: Some(addr.to_string()),
            recv_path: "/ingest".to_string(),
            recv_buffer_size: 100,
            recv_timeout_ms: 1000,
            ..Default::default()
        };
        let receiver = HttpTransport::new(&recv_config).await.unwrap();

        // Give the server a moment to start
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        // Send a message via a separate sender transport
        let send_config =
            HttpTransportConfig::sender(&format!("http://127.0.0.1:{}/ingest", addr.port()));
        let sender = HttpTransport::new(&send_config).await.unwrap();

        let result = sender
            .send("", bytes::Bytes::from_static(b"{\"msg\":\"hello\"}"))
            .await;
        assert!(result.is_ok(), "send failed: {result:?}");

        // Receive it
        let batch = receiver.recv(10).await.unwrap();
        assert_eq!(batch.records.len(), 1);
        assert_eq!(batch.records[0].payload.as_ref(), b"{\"msg\":\"hello\"}");
        // The source address rides on the batch commit token, not the record.
        assert!(batch.commit_tokens[0].source_addr.is_some());

        // Cleanup
        sender.close().await.unwrap();
        receiver.close().await.unwrap();
    }

    /// Test that the receiver rejects empty bodies.
    #[cfg(feature = "http-server")]
    #[tokio::test]
    async fn receive_rejects_empty_body() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        drop(listener);

        let recv_config = HttpTransportConfig {
            listen: Some(addr.to_string()),
            recv_timeout_ms: 200,
            ..Default::default()
        };
        let receiver = HttpTransport::new(&recv_config).await.unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        // Send empty body
        let client = reqwest::Client::new();
        let resp = client
            .post(format!("http://127.0.0.1:{}/ingest", addr.port()))
            .body(Vec::<u8>::new())
            .send()
            .await
            .unwrap();

        assert_eq!(resp.status(), reqwest::StatusCode::BAD_REQUEST);

        // recv should timeout with no messages
        let records = receiver.recv(10).await.unwrap().records;
        assert!(records.is_empty());

        receiver.close().await.unwrap();
    }

    /// Oversized bodies are rejected with 413 before the handler buffers them.
    #[cfg(feature = "http-server")]
    #[tokio::test]
    async fn receive_rejects_oversized_body() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        drop(listener);

        let recv_config = HttpTransportConfig {
            listen: Some(addr.to_string()),
            recv_timeout_ms: 200,
            max_body_bytes: 1024, // 1 KiB cap
            ..Default::default()
        };
        let receiver = HttpTransport::new(&recv_config).await.unwrap();
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;

        // POST 8 KiB -- over the 1 KiB cap.
        let client = reqwest::Client::new();
        let resp = client
            .post(format!("http://127.0.0.1:{}/ingest", addr.port()))
            .body(vec![b'x'; 8 * 1024])
            .send()
            .await
            .unwrap();

        assert_eq!(resp.status(), reqwest::StatusCode::PAYLOAD_TOO_LARGE);

        // The oversized POST never reached the queue.
        let records = receiver.recv(10).await.unwrap().records;
        assert!(records.is_empty(), "oversized body must not be queued");

        receiver.close().await.unwrap();
    }

    /// Test recv without listen returns config error.
    #[cfg(feature = "http-server")]
    #[tokio::test]
    async fn recv_without_listen_returns_error() {
        let config = HttpTransportConfig::sender("http://localhost:9999");
        let transport = HttpTransport::new(&config).await.unwrap();

        let result = transport.recv(10).await;
        assert!(result.is_err());
    }

    /// G3: with a pressure governor pinned HIGH, the ingest handler sheds with
    /// 503 (SERVICE_UNAVAILABLE) -- the same status as the channel-full path.
    /// With `None` (the default `new`), POST is accepted (200) as before.
    #[cfg(all(feature = "http-server", feature = "governor"))]
    #[tokio::test]
    async fn pressure_high_sheds_with_503_and_none_is_normal() {
        use crate::governor::{Hysteresis, MemoryPressureSource, PressureSource, UnifiedPressure};
        use crate::memory::{MemoryGuard, MemoryGuardConfig};

        // --- None default: POST accepted (200) ---
        {
            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            let addr = listener.local_addr().unwrap();
            drop(listener);
            let cfg = HttpTransportConfig {
                listen: Some(addr.to_string()),
                recv_timeout_ms: 200,
                ..Default::default()
            };
            let receiver = HttpTransport::new(&cfg).await.unwrap();
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;

            let client = reqwest::Client::new();
            let resp = client
                .post(format!("http://127.0.0.1:{}/ingest", addr.port()))
                .body(b"{\"msg\":\"ok\"}".to_vec())
                .send()
                .await
                .unwrap();
            assert_eq!(
                resp.status(),
                reqwest::StatusCode::OK,
                "no governor -> accepted"
            );
            receiver.close().await.unwrap();
        }

        // --- Governor pinned HIGH (HARD memory source at 95%): 503 ---
        {
            let guard = Arc::new(MemoryGuard::new(MemoryGuardConfig {
                limit_bytes: 1000,
                pressure_threshold: 0.80,
                ..Default::default()
            }));
            guard.add_bytes(950); // 95% -> well above pause_above
            let src = MemoryPressureSource::new(Arc::clone(&guard));
            let pressure = Arc::new(UnifiedPressure::new(
                vec![Arc::new(src) as Arc<dyn PressureSource>],
                Hysteresis::new(0.80, 0.65).expect("valid band"),
            ));
            // Sanity: the latch is armed.
            assert!(pressure.should_hold(), "pinned-high governor must hold");

            let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            let addr = listener.local_addr().unwrap();
            drop(listener);
            let cfg = HttpTransportConfig {
                listen: Some(addr.to_string()),
                recv_timeout_ms: 200,
                ..Default::default()
            };
            let receiver = HttpTransport::with_pressure(&cfg, Some(Arc::clone(&pressure)))
                .await
                .unwrap();
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;

            let client = reqwest::Client::new();
            let resp = client
                .post(format!("http://127.0.0.1:{}/ingest", addr.port()))
                .body(b"{\"msg\":\"shed\"}".to_vec())
                .send()
                .await
                .unwrap();
            assert_eq!(
                resp.status(),
                reqwest::StatusCode::SERVICE_UNAVAILABLE,
                "pinned-high governor must shed with 503"
            );
            // The 503 carries a Retry-After hint so a well-behaved sender backs
            // off instead of hot-retrying into a holding pod.
            assert_eq!(
                resp.headers()
                    .get(reqwest::header::RETRY_AFTER)
                    .and_then(|v| v.to_str().ok()),
                Some("1"),
                "503 shed must carry Retry-After"
            );

            // The shed POST never reached the queue.
            let records = receiver.recv(10).await.unwrap().records;
            assert!(records.is_empty(), "shed request must not be queued");
            receiver.close().await.unwrap();
        }
    }

    #[test]
    fn config_serde_roundtrip() {
        let config = HttpTransportConfig {
            endpoint: Some("http://example.com/ingest".into()),
            listen: Some("0.0.0.0:8080".into()),
            recv_path: "/custom".into(),
            recv_buffer_size: 5000,
            recv_timeout_ms: 250,
            ..Default::default()
        };

        let json = serde_json::to_string(&config).unwrap();
        let parsed: HttpTransportConfig = serde_json::from_str(&json).unwrap();

        assert_eq!(parsed.endpoint, config.endpoint);
        assert_eq!(parsed.listen, config.listen);
        assert_eq!(parsed.recv_path, config.recv_path);
        assert_eq!(parsed.recv_buffer_size, config.recv_buffer_size);
        assert_eq!(parsed.recv_timeout_ms, config.recv_timeout_ms);
    }
}