camel-component-wasm 0.24.0

//! Source host state, linker functions, and host imports.
//!
//! This module provides:
//! - [`SourceHostState`]: host state for the source world (separate from `WasmHostState`)
//! - [`SourceChannels`]: bounded tokio channels bridging async guest imports to host tasks
//! - `HostWithStore` impl for async `accept-http`, `submit-exchange`; sync `Host` impl for
//!   `is-cancelled`
//! - [`run_http_listener`]: async axum task feeding HTTP requests into the channel
//! - [`run_pipeline_bridge`]: async task forwarding exchanges to the pipeline

use std::sync::Arc;

use bytes::Bytes;
use tokio::sync::{Notify, mpsc, oneshot};
use tokio_util::sync::CancellationToken;
use wasmtime::component::{Accessor, AccessorTask, HasSelf, Linker, Resource, ResourceTable};

use camel_api::{Body, CamelError, Exchange, ExchangePattern, Message, StreamBody, Value};
use camel_component_api::consumer::ConsumerContext;

use crate::return_stream::{
    DEFAULT_DRAIN_CHANNEL_BOUND, DrainCoord, DrainEvent, StreamReturnable, TerminalSlot,
    drain_guest_stream, receiver_to_body_stream,
};
use crate::source_bindings::camel::plugin::source_host::{HttpRequest, SubmitOutcome};
use crate::source_bindings::camel::plugin::types::{
    WasmBody, WasmExchange, WasmMessage, WasmPattern,
};

/// Concrete type for the http-listener resource in the ResourceTable.
/// Stateless — the handle is a marker that the guest holds while running.
/// Mapped via `with:` in source_bindings.rs to replace the empty enum
/// generated by bindgen for the `resource http-listener` declaration.
pub struct HttpListenerHandle;

/// Materialized HTTP request metadata (everything except the body).
///
/// Sent through the request channel alongside a per-request body receiver.
/// The body itself is streamed incrementally through the receiver — never
/// materialized as a single `Vec<u8>` — so large payloads are not capped.
pub struct HttpMeta {
    pub method: String,
    pub path: String,
    pub headers: Vec<(String, String)>,
}

/// Per-request body chunk receiver. The host's axum handler streams body
/// frames into the matching `mpsc::Sender`; the guest reads via the
/// `stream-body-handle` minted by `accept_http`.
type BodyChunkRx = mpsc::Receiver<Result<Bytes, CamelError>>;

/// Request channel payload: metadata + body streaming receiver.
type RequestChannelItem = (HttpMeta, BodyChunkRx);

// ─── Channel capacity constants ───────────────────────────────────────────

/// Capacity for the HTTP request channel (host listener → guest).
///
/// 1 propagates backpressure all the way to the HTTP client: the axum handler
/// `.send().await`s onto this channel, so once the guest is busy draining a
/// prior exchange into the (also capacity-1) pipeline, further inbound
/// requests park on the send until the guest calls `accept-http` again. A
/// larger buffer would let a burst of clients receive 202 responses for work
/// the pipeline has not yet accepted, silently decoupling client-visible
/// success from actual delivery — undesirable for a source that must not
/// acknowledge messages it may drop on shutdown.
pub const REQUEST_CHANNEL_CAPACITY: usize = 1;

/// Capacity for the exchange channel (guest → pipeline bridge).
/// 1 enforces strict backpressure — guest blocks until pipeline accepts.
pub const EXCHANGE_CHANNEL_CAPACITY: usize = 1;

// ─── SourceChannels ───────────────────────────────────────────────────────

/// Paired channel endpoints for bridging the guest to async host tasks.
///
/// Created once per source consumer instance. The `*_tx` halves go to the
/// HTTP listener and pipeline bridge tasks; `request_rx` lives in
/// [`SourceHostState`] inside the wasmtime `Store`, wrapped in an
/// `Arc<tokio::sync::Mutex<…>>` so the async host import (which receives only
/// `&Accessor`, not `&mut self`) can `lock().await` it to call `recv().await`.
///
/// The request channel carries `(HttpMeta, body_rx)` — metadata plus a
/// per-request bounded channel streaming body chunks. The body is never
/// materialized as a single `Vec<u8>`, removing the old 10 MiB cap.
pub struct SourceChannels {
    pub request_tx: mpsc::Sender<RequestChannelItem>,
    pub request_rx: Arc<tokio::sync::Mutex<mpsc::Receiver<RequestChannelItem>>>,
    pub exchange_tx: mpsc::Sender<(Exchange, oneshot::Sender<SubmitOutcome>)>,
    pub exchange_rx: mpsc::Receiver<(Exchange, oneshot::Sender<SubmitOutcome>)>,
}

impl SourceChannels {
    /// Create a new set of channels with the documented capacities.
    pub fn new() -> Self {
        let (request_tx, request_rx) = mpsc::channel(REQUEST_CHANNEL_CAPACITY);
        let (exchange_tx, exchange_rx) = mpsc::channel(EXCHANGE_CHANNEL_CAPACITY);
        Self {
            request_tx,
            request_rx: Arc::new(tokio::sync::Mutex::new(request_rx)),
            exchange_tx,
            exchange_rx,
        }
    }
}

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

// ─── SourceHostState ──────────────────────────────────────────────────────

/// Default maximum request body size in bytes (10 MiB).
///
/// Matches [`camel_api::body::DEFAULT_MATERIALIZE_LIMIT`] — the same cap the
/// old `to_bytes` path enforced. Restores the DoS backstop that was
/// inadvertently removed when the body switched from materialized to streamed.
pub const DEFAULT_MAX_REQUEST_BODY_BYTES: u64 = 10 * 1024 * 1024;

/// Host state for a source consumer. Lives inside the `wasmtime::Store`.
///
/// Separate from [`crate::runtime::WasmHostState`] because the source world
/// has a fundamentally different lifecycle: the guest IS the source, owns a
/// run loop, and communicates via channels rather than direct call-process.
///
/// `request_rx` is wrapped in `Arc<tokio::sync::Mutex<…>>` because the async
/// host imports receive `&Accessor` (shared access), not `&mut self`; the
/// Mutex provides the interior mutability needed to `recv().await` the
/// receiver across import calls.
pub struct SourceHostState {
    pub table: ResourceTable,
    pub wasi: wasmtime_wasi::WasiCtx,
    pub request_rx: Arc<tokio::sync::Mutex<mpsc::Receiver<RequestChannelItem>>>,
    pub exchange_tx: mpsc::Sender<(Exchange, oneshot::Sender<SubmitOutcome>)>,
    pub cancel_token: CancellationToken,
    /// Maximum bytes the guest may read from a single request body before
    /// the stream producer terminates with an overflow error. Restores the
    /// DoS backstop that the old `to_bytes(MAX_BODY_BYTES)` path provided.
    pub max_request_body_bytes: u64,
}

impl wasmtime_wasi::WasiView for SourceHostState {
    fn ctx(&mut self) -> wasmtime_wasi::WasiCtxView<'_> {
        wasmtime_wasi::WasiCtxView {
            ctx: &mut self.wasi,
            table: &mut self.table,
        }
    }
}

// ─── Host trait implementation ────────────────────────────────────────────

type SourceWasmError = crate::source_bindings::camel::plugin::types::WasmError;

// The http-listener resource has no methods in WIT, but wasmtime bindgen
// generates a HostHttpListener trait that must be implemented.
impl crate::source_bindings::camel::plugin::source_host::HostHttpListener for SourceHostState {
    fn drop(
        &mut self,
        _resource: wasmtime::component::Resource<HttpListenerHandle>,
    ) -> wasmtime::Result<()> {
        // HttpListenerHandle is stateless; nothing to clean up.
        Ok(())
    }
}

impl crate::source_bindings::camel::plugin::source_host::Host for SourceHostState {
    fn is_cancelled(&mut self) -> bool {
        // Stays sync (a `with`-style peek that must not yield). Returns
        // immediately so tight guest loops can poll cancellation cheaply.
        self.cancel_token.is_cancelled()
    }
}

/// Async host imports. The guest's `run` (async) awaits these; each receives
/// `&Accessor<SourceHostState, HasSelf<SourceHostState>>` (shared access into
/// the store, NOT `&mut self`). We snapshot owned values (clones) out of a
/// `with` call, then `.await` OUTSIDE it — `Accessor::with` is sync and must
/// not be held across an await. Cancellation races every await against the
/// shared `CancellationToken` so `stop()` unblocks a parked import promptly.
impl crate::source_bindings::camel::plugin::source_host::HostWithStore<SourceHostState>
    for wasmtime::component::HasSelf<SourceHostState>
{
    async fn accept_http(
        accessor: &Accessor<SourceHostState, wasmtime::component::HasSelf<SourceHostState>>,
        _listener: Resource<HttpListenerHandle>,
    ) -> Result<Option<HttpRequest>, SourceWasmError> {
        // Snapshot clones via `with`, then await outside. request_rx is an
        // Arc<Mutex<Receiver>>; cloning the Arc is cheap and lets us lock it
        // across the await without borrowing the accessor.
        // Single `with` snapshot — one TLS lookup for all three fields.
        let (request_rx, cancel_token, max_body) = accessor.with(|mut view| {
            let state = view.get();
            (
                state.request_rx.clone(),
                state.cancel_token.clone(),
                state.max_request_body_bytes,
            )
        });

        // Uncontended lock: the guest runs a single sequential run loop, so
        // only one accept_http is in flight at a time.
        let mut guard = request_rx.lock().await;
        let meta_and_body = tokio::select! {
            r = guard.recv() => r,
            _ = cancel_token.cancelled() => {
                return Ok(None);
            }
        };
        // Drop the lock before minting the StreamReader (needs accessor.with).
        drop(guard);

        // None => channel closed (listener exited); treat as cancellation.
        let Some((meta, mut body_rx)) = meta_and_body else {
            return Ok(None);
        };

        // Build a BoxStream from the per-request body channel via poll_fn.
        // NOT receiver_to_body_stream — that is DrainEvent-specific (the
        // submit-exchange direction). Here the channel carries
        // Result<Bytes, CamelError> directly.
        let box_stream: futures::stream::BoxStream<'static, Result<Bytes, CamelError>> =
            Box::pin(futures::stream::poll_fn(move |cx| body_rx.poll_recv(cx)));

        // Mint the StreamReader + terminal FutureReader inside accessor.with.
        // assemble_stream_body_source takes &Accessor<S, U> and internally
        // calls accessor.with for each reader creation.
        let cancel = cancel_token.clone();
        let handle = crate::stream_bridge::assemble_stream_body_source(
            accessor, box_stream, cancel, max_body,
        );
        let request = match handle {
            Ok(body) => Some(HttpRequest {
                method: meta.method,
                path: meta.path,
                headers: meta.headers,
                body,
            }),
            Err(e) => {
                // log-policy: system-broken
                tracing::error!("failed to assemble stream body for accept-http: {e}");
                None
            }
        };
        Ok(request)
    }

    async fn submit_exchange(
        accessor: &Accessor<SourceHostState, wasmtime::component::HasSelf<SourceHostState>>,
        mut exchange: WasmExchange,
    ) -> Result<SubmitOutcome, SourceWasmError> {
        // Snapshot owned values (clones) + extract the stream (if any) inside a
        // single `with`. The closure must return before `spawn` is called:
        // `Accessor::spawn` internally calls `with`, which panics if nested
        // (concurrent.rs:454). `take_stream` replaces a `WasmBody::Stream` with
        // `Empty`, moving the live `StreamReader`/`FutureReader` out so the
        // metadata conversion below does not trip over the non-cloneable handle.
        let (exchange_tx, cancel_token, stream_parts) = accessor.with(|mut view| {
            let state = view.get();
            let exchange_tx = state.exchange_tx.clone();
            let cancel_token = state.cancel_token.clone();
            let stream_parts = exchange.take_stream();
            (exchange_tx, cancel_token, stream_parts)
        });
        // `with` guard dropped here — safe to call `accessor.spawn` below.

        // Build the native Exchange from the (now stream-stripped) metadata.
        // The body is `Empty` for the stream case; it is overwritten with
        // `Body::Stream` after the drain is spawned.
        let mut native = source_exchange_to_native(exchange);
        let (reply_tx, reply_rx) = oneshot::channel();

        if let Some((stream_reader, terminal, stream_metadata)) = stream_parts {
            // Streaming submit-exchange: fire-and-return.
            //
            // The guest emits a `stream<u8>` body. Reading a wasmtime stream
            // requires the concurrent execution context (an `&Accessor`), which
            // is only live while the event loop driving `run` polls. We CANNOT
            // `tokio::spawn` a moved store + `run_concurrent` (the source
            // world's `run` is already inside `run_concurrent`;
            // `check_recursive_run` panics). Instead, `Accessor::spawn` queues
            // the drain onto the SAME event loop, where it progresses
            // concurrently with the guest fiber.
            let (chunk_tx, chunk_rx) = mpsc::channel::<DrainEvent>(DEFAULT_DRAIN_CHANNEL_BOUND);
            // Non-blocking terminal slot: written BEFORE sender drop so the
            // consumer observes the terminal result after buffered chunks drain.
            // Created ONCE and shared between the drain task and the body stream.
            let source_terminal: TerminalSlot = Arc::new(std::sync::Mutex::new(None));
            // progress/receiver_gone unused under source (no watchdog);
            // required by shared drain_guest_stream signature.
            let coord = DrainCoord {
                cancel: cancel_token.clone(),
                progress: Arc::new(Notify::new()),
                receiver_gone: Arc::new(Notify::new()),
                terminal_slot: source_terminal.clone(),
            };

            // Register the drain BEFORE sending the exchange. Ordering is
            // load-bearing: the chunk-channel receiver only gets a consumer
            // (the downstream `Body::Stream` reader) once the exchange is
            // delivered. If we awaited drain completion before sending, the
            // bounded chunk channel would fill, backpressure-stall the drain,
            // and self-deadlock.
            accessor.spawn(SubmitExchangeDrain {
                stream_reader,
                terminal,
                chunk_tx,
                coord,
            });

            // Attach the lazy body. Downstream reads from this receiver; the
            // spawned drain feeds it from the guest's stream.
            native.input.body = Body::Stream(StreamBody {
                stream: Arc::new(tokio::sync::Mutex::new(Some(Box::pin(
                    receiver_to_body_stream(crate::return_stream::DrainReceiver {
                        rx: chunk_rx,
                        terminal: source_terminal,
                    }),
                )))),
                metadata: stream_metadata,
            });
        }

        // Send the (native) exchange and await the pipeline's acceptance reply.
        // submit-exchange returns once the pipeline ACCEPTS the envelope —
        // before full body drain (fire-and-return). Race against cancellation
        // so `stop()` unblocks a parked submit promptly.
        let outcome = tokio::select! {
            res = async {
                if exchange_tx.send((native, reply_tx)).await.is_err() {
                    // Bridge gone (clean shutdown) — report Stopped.
                    return SubmitOutcome::Stopped;
                }
                // Pipeline bridge replies with the delivery outcome; a dropped
                // reply (bridge exited after accept) is also a clean stop.
                reply_rx.await.unwrap_or(SubmitOutcome::Stopped)
            } => res,
            _ = cancel_token.cancelled() => SubmitOutcome::Stopped,
        };
        Ok(outcome)
    }
}

// ─── submit-exchange background drain (AccessorTask) ──────────────────────

/// `AccessorTask` driving the guest→host body drain for a streaming
/// `submit-exchange`.
///
/// Unlike the plugin/bean return path (`spawn_return_drain`), the source world
/// CANNOT `tokio::spawn` a moved store and open a fresh `run_concurrent` — the
/// source `run` is already inside `run_concurrent` and `check_recursive_run`
/// panics on a nested call. `Accessor::spawn` instead registers this task on
/// the SAME event loop already driving `run`, so the drain progresses
/// concurrently with the guest fiber (and the guest's `spawn_local` writer).
struct SubmitExchangeDrain {
    stream_reader: wasmtime::component::StreamReader<u8>,
    terminal: wasmtime::component::FutureReader<Result<(), SourceWasmError>>,
    chunk_tx: mpsc::Sender<DrainEvent>,
    coord: DrainCoord,
}

impl AccessorTask<SourceHostState, HasSelf<SourceHostState>> for SubmitExchangeDrain {
    async fn run(
        self,
        accessor: &Accessor<SourceHostState, HasSelf<SourceHostState>>,
    ) -> wasmtime::Result<()> {
        drain_guest_stream::<SourceWasmError, SourceHostState>(
            accessor,
            self.stream_reader,
            self.terminal,
            self.chunk_tx,
            self.coord,
        )
        .await;
        // Terminal errors are surfaced through the chunk channel
        // (`DrainEvent::Error`), not the task return — the downstream
        // `Body::Stream` reader observes them. Returning `Ok` keeps the
        // event loop's task accounting clean.
        Ok(())
    }
}

// ─── source WasmExchange → native Exchange conversion ──────────────────────

/// Convert a (stream-stripped) source-world `WasmExchange` into a native
/// [`Exchange`]. Inlines the `serde_bridge` field-mapping because the source
/// bindings generate distinct types from the plugin bindings (TODO WIT-001).
///
/// A `WasmBody::Stream` never reaches here: `submit_exchange` calls
/// `take_stream` (replacing it with `Empty`) before conversion and attaches a
/// `Body::Stream` separately.
fn source_exchange_to_native(wasm: WasmExchange) -> Exchange {
    let input = source_message_to_native(wasm.input);
    let mut exchange = Exchange::new(input);
    // Preserve the guest's correlation_id (Exchange::new generates a random one).
    if !wasm.correlation_id.is_empty() {
        exchange.correlation_id = wasm.correlation_id;
    }
    if let Some(out) = wasm.output {
        exchange.output = Some(source_message_to_native(out));
    }
    for (key, raw) in wasm.properties {
        let value = serde_json::from_str::<Value>(&raw).unwrap_or(Value::String(raw));
        exchange.properties.insert(key, value);
    }
    exchange.pattern = match wasm.pattern {
        WasmPattern::InOnly => ExchangePattern::InOnly,
        WasmPattern::InOut => ExchangePattern::InOut,
    };
    exchange
}

fn source_message_to_native(msg: WasmMessage) -> Message {
    let headers = msg
        .headers
        .into_iter()
        .map(|(k, raw)| {
            let value = serde_json::from_str::<Value>(&raw).unwrap_or(Value::String(raw));
            (k, value)
        })
        .collect();
    Message {
        headers,
        body: source_body_to_native(msg.body),
    }
}

fn source_body_to_native(body: WasmBody) -> Body {
    match body {
        WasmBody::Empty => Body::Empty,
        WasmBody::Text(s) => Body::Text(s),
        WasmBody::Bytes(v) => Body::Bytes(Bytes::from(v)),
        WasmBody::Json(s) => serde_json::from_str::<Value>(&s)
            .map(Body::Json)
            .unwrap_or(Body::Text(s)),
        WasmBody::Xml(s) => Body::Xml(s),
        // Unreachable: `submit_exchange` extracts the stream via `take_stream`
        // before conversion. Defensive guard (drop + warn) mirrors
        // `serde_bridge::wasm_to_body`.
        WasmBody::Stream(_) => {
            tracing::warn!("source: undrained stream body reached conversion — dropped");
            Body::Empty
        }
    }
}

// ─── Linker setup ─────────────────────────────────────────────────────────

/// Register the source-host interface and WASI p2 into a linker for [`SourceHostState`].
///
/// Uses the ASYNCHRONOUS WASI linker: the guest's `run` export is async and is
/// driven by `Store::run_concurrent` on a tokio task (see
/// [`crate::source_consumer`]). Its async host imports (`accept-http`,
/// `submit-exchange`) receive an `&Accessor` and use async channel ops.
pub fn add_to_linker(linker: &mut Linker<SourceHostState>) -> Result<(), wasmtime::Error> {
    // WASI p2 preview — async variant required for the async source world.
    wasmtime_wasi::p2::add_to_linker_async(linker)?;

    // Source-host interface (accept-http, submit-exchange, is-cancelled).
    crate::source_bindings::camel::plugin::source_host::add_to_linker::<_, HasSelf<_>>(
        linker,
        |state| state,
    )?;

    Ok(())
}

// ─── HTTP listener task (async) ───────────────────────────────────────────

/// Run an axum HTTP server that feeds incoming requests into `request_tx`.
///
/// The caller must have already bound the TCP listener (see
/// [`crate::source_consumer::WasmSourceConsumer::start`]) so that a bind
/// failure surfaces synchronously as a `start()` error rather than a
/// background warning that leaves the route appearing healthy with no
/// listener accepting requests.
///
/// Each request's body is streamed incrementally into a per-request bounded
/// channel — the old 10 MiB `to_bytes` materialization cap is removed. The
/// guest reads the body via the `stream-body-handle` returned by
/// `accept-http`.
///
/// # Response timing (fire-and-forget semantic)
///
/// Returns **202 Accepted** once the request metadata is handed to the guest
/// via the request channel. The body is streamed asynchronously *after* the
/// response is sent — the handler does NOT wait for the full body before
/// responding. This is inherent to the streaming shape: you cannot stream
/// and wait for full receipt simultaneously. Mid-body connection drops
/// surface as stream errors to the guest (the body channel receives an
/// `Err` frame), not as HTTP-level failures to the client. The source
/// accepts the request for processing; it does not guarantee full body
/// receipt.
///
/// Shuts down gracefully when `cancel` is triggered.
pub async fn run_http_listener(
    listener: tokio::net::TcpListener,
    path_filter: Option<String>,
    request_tx: mpsc::Sender<RequestChannelItem>,
    cancel: CancellationToken,
) -> Result<(), CamelError> {
    use axum::Router;
    use axum::extract::State;
    use axum::http::Request;
    use axum::response::Response;
    use axum::routing::any;
    use http_body_util::BodyExt;

    struct ListenerState {
        tx: mpsc::Sender<RequestChannelItem>,
        cancel: CancellationToken,
    }

    async fn handler(
        State(state): State<Arc<ListenerState>>,
        req: Request<axum::body::Body>,
    ) -> Response {
        let (parts, body) = req.into_parts();

        // Per-request bounded channel for streaming the body to the guest.
        // Capacity matches the drain-channel bound used elsewhere for
        // guest→host streaming (return_stream.rs).
        let (body_tx, body_rx) = mpsc::channel::<Result<Bytes, CamelError>>(
            crate::return_stream::DEFAULT_DRAIN_CHANNEL_BOUND,
        );

        let http_meta = HttpMeta {
            method: parts.method.to_string(),
            path: parts.uri.path().to_string(),
            headers: parts
                .headers
                .iter()
                .map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_string()))
                .collect(),
        };

        if state.tx.send((http_meta, body_rx)).await.is_err() {
            return Response::builder()
                .status(503)
                .body(axum::body::Body::from("service unavailable"))
                .unwrap(); // allow-unwrap
        }

        // Stream the axum body into body_tx incrementally. The spawned task
        // owns the body and the sender; when the body is fully consumed (or
        // errors), body_tx drops = EOF on the guest's StreamReader.
        // Racing against cancel ensures the drain exits promptly on source
        // shutdown rather than relying solely on connection-close propagation.
        let drain_cancel = state.cancel.clone();
        tokio::spawn(async move {
            let mut body = body;
            loop {
                tokio::select! {
                    frame = body.frame() => {
                        match frame {
                            Some(Ok(frame)) => {
                                if let Ok(data) = frame.into_data()
                                    && body_tx.send(Ok(data)).await.is_err()
                                {
                                    // Guest stopped reading — abandon the stream.
                                    break;
                                }
                                // Trailers frame: silently ignored (HTTP metadata, not body bytes).
                            }
                            Some(Err(e)) => {
                                let _ = body_tx
                                    .send(Err(CamelError::Io(format!("body read: {e}"))))
                                    .await;
                                break;
                            }
                            // None = clean EOF from hyper — body fully consumed.
                            None => break,
                        }
                    }
                    _ = drain_cancel.cancelled() => break,
                }
            }
            // body_tx drops here → guest observes clean EOF.
        });

        Response::builder()
            .status(202)
            .body(axum::body::Body::from("accepted"))
            .unwrap() // allow-unwrap
    }

    let state = Arc::new(ListenerState {
        tx: request_tx,
        cancel: cancel.clone(),
    });

    let route_path = path_filter
        .filter(|path| !path.is_empty())
        .map(|path| {
            if path.starts_with('/') {
                path
            } else {
                format!("/{path}")
            }
        })
        .unwrap_or_else(|| "/{*path}".to_string());

    let app = Router::new()
        .route(&route_path, any(handler))
        .with_state(state);

    // Listener is already bound by the caller; capture the address for logging.
    let local = listener.local_addr().ok();

    if let Some(addr) = &local {
        tracing::info!(%addr, "source HTTP listener started");
    }

    axum::serve(listener, app)
        .with_graceful_shutdown(async move { cancel.cancelled().await })
        .await
        .map_err(|e| CamelError::Io(format!("HTTP listener error: {e}")))?;

    if let Some(addr) = &local {
        tracing::info!(%addr, "source HTTP listener stopped");
    }
    Ok(())
}

// ─── Pipeline bridge task (async) ─────────────────────────────────────────

/// Receive native exchanges from the guest's `submit-exchange` import and
/// forward them to the pipeline via [`ConsumerContext`].
///
/// The WasmExchange→native conversion (and streaming-body drain setup) happens
/// inside the import now (see [`source_exchange_to_native`]), so the bridge is
/// reduced to a pure forwarder. Sends [`SubmitOutcome::Accepted`] /
/// [`SubmitOutcome::Stopped`] back to the guest via the oneshot reply channel.
pub async fn run_pipeline_bridge(
    mut exchange_rx: mpsc::Receiver<(Exchange, oneshot::Sender<SubmitOutcome>)>,
    ctx: ConsumerContext,
) -> Result<(), CamelError> {
    while let Some((exchange, reply_tx)) = exchange_rx.recv().await {
        let outcome = match ctx.send(exchange).await {
            Ok(()) => SubmitOutcome::Accepted,
            Err(_) => SubmitOutcome::Stopped,
        };
        // If the guest has already abandoned the exchange (unlikely), this
        // send fails silently.
        let _ = reply_tx.send(outcome);
    }
    Ok(())
}

// ─── Tests ────────────────────────────────────────────────────────────────

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

    #[test]
    fn test_source_channels_new() {
        let channels = SourceChannels::new();
        assert!(!channels.request_tx.is_closed());
        assert!(!channels.exchange_tx.is_closed());
    }

    #[test]
    fn test_source_channels_default() {
        let channels = SourceChannels::default();
        assert!(!channels.request_tx.is_closed());
    }

    #[test]
    fn test_request_channel_close_returns_none() {
        let (tx, mut rx) = mpsc::channel::<RequestChannelItem>(1);
        drop(tx);
        let result = std::thread::spawn(move || rx.blocking_recv())
            .join()
            .unwrap();
        assert!(result.is_none());
    }

    #[test]
    fn test_exchange_channel_close_detected() {
        let (tx, rx) = mpsc::channel::<(Exchange, oneshot::Sender<SubmitOutcome>)>(1);
        drop(rx);
        assert!(tx.is_closed());
    }

    #[test]
    fn test_cancel_token_is_cancelled() {
        let token = CancellationToken::new();
        assert!(!token.is_cancelled());
        token.cancel();
        assert!(token.is_cancelled());
    }

    #[test]
    fn test_submit_outcome_variants() {
        let accepted = SubmitOutcome::Accepted;
        let stopped = SubmitOutcome::Stopped;
        assert!(matches!(accepted, SubmitOutcome::Accepted));
        assert!(matches!(stopped, SubmitOutcome::Stopped));
    }

    #[test]
    fn source_exchange_to_native_maps_fields() {
        use crate::source_bindings::camel::plugin::types as src;
        let wasm = src::WasmExchange {
            input: src::WasmMessage {
                headers: vec![("key".to_string(), "val".to_string())],
                body: src::WasmBody::Text("hello".to_string()),
            },
            output: None,
            properties: vec![("p".to_string(), "v".to_string())],
            pattern: src::WasmPattern::InOnly,
            correlation_id: "corr-1".to_string(),
            route_id: Some("route-1".to_string()),
            message_id: Some("msg-1".to_string()),
        };

        let native = source_exchange_to_native(wasm);
        assert_eq!(native.input.headers.len(), 1);
        assert_eq!(
            native.input.headers.get("key"),
            Some(&camel_api::Value::String("val".to_string()))
        );
        assert!(matches!(native.input.body, Body::Text(_)));
        assert_eq!(
            native.properties.get("p"),
            Some(&camel_api::Value::String("v".to_string()))
        );
        assert!(matches!(native.pattern, ExchangePattern::InOnly));
        assert_eq!(native.correlation_id, "corr-1");
    }
}