camel-component-wasm 0.22.0

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

use std::sync::Arc;

use tokio::sync::{mpsc, oneshot};
use tokio_util::sync::CancellationToken;
use wasmtime::component::{HasSelf, Linker, ResourceTable};

use camel_api::{CamelError, Exchange, Message};
use camel_component_api::consumer::ConsumerContext;

use crate::source_bindings::camel::plugin::source_host::{HttpRequest, SubmitOutcome};
use crate::source_bindings::camel::plugin::types::WasmExchange;

/// 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;

// ─── 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 sync guest to async host tasks.
///
/// Created once per source consumer instance. The `*_tx` halves go to the
/// HTTP listener and pipeline bridge tasks; the `*_rx` halves live in
/// [`SourceHostState`] inside the wasmtime `Store`.
pub struct SourceChannels {
    pub request_tx: mpsc::Sender<HttpRequest>,
    pub request_rx: mpsc::Receiver<HttpRequest>,
    pub exchange_tx: mpsc::Sender<(WasmExchange, oneshot::Sender<SubmitOutcome>)>,
    pub exchange_rx: mpsc::Receiver<(WasmExchange, 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,
            exchange_tx,
            exchange_rx,
        }
    }
}

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

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

/// 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.
pub struct SourceHostState {
    pub table: ResourceTable,
    pub wasi: wasmtime_wasi::WasiCtx,
    pub request_rx: mpsc::Receiver<HttpRequest>,
    pub exchange_tx: mpsc::Sender<(WasmExchange, oneshot::Sender<SubmitOutcome>)>,
    pub cancel_token: CancellationToken,
}

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 accept_http(
        &mut self,
        _listener: wasmtime::component::Resource<HttpListenerHandle>,
    ) -> Result<Option<HttpRequest>, SourceWasmError> {
        // blocking_recv: guest runs in spawn_blocking (Task 5), so this is safe.
        // Returns None when the channel is closed (listener task exited / cancelled).
        match self.request_rx.blocking_recv() {
            Some(req) => Ok(Some(req)),
            None => Ok(None),
        }
    }

    fn submit_exchange(
        &mut self,
        exchange: WasmExchange,
    ) -> Result<SubmitOutcome, SourceWasmError> {
        let (reply_tx, reply_rx) = oneshot::channel();
        // blocking_send: fails only if the receiver is dropped (pipeline bridge exited = clean shutdown).
        if self
            .exchange_tx
            .blocking_send((exchange, reply_tx))
            .is_err()
        {
            return Ok(SubmitOutcome::Stopped);
        }

        // Wait for the pipeline bridge to confirm delivery.
        match reply_rx.blocking_recv() {
            Ok(outcome) => Ok(outcome),
            Err(_) => Ok(SubmitOutcome::Stopped),
        }
    }

    fn is_cancelled(&mut self) -> bool {
        self.cancel_token.is_cancelled()
    }
}

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

/// Register the source-host interface and WASI p2 into a linker for [`SourceHostState`].
///
/// Uses the synchronous WASI linker: the guest runs on a dedicated OS thread
/// (see [`crate::source_consumer`]) and its host imports block on tokio
/// channels, which is only valid off the async runtime.
pub fn add_to_linker(linker: &mut Linker<SourceHostState>) -> Result<(), wasmtime::Error> {
    // WASI p2 preview — required because the guest targets wasm32-wasip2.
    wasmtime_wasi::p2::add_to_linker_sync(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) ───────────────────────────────────────────

/// Maximum accepted inbound request body size (10 MiB). Payloads exceeding
/// this are rejected with 413 rather than silently truncated to empty.
const MAX_BODY_BYTES: usize = 10 * 1024 * 1024;

/// 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.
///
/// 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<HttpRequest>,
    cancel: CancellationToken,
) -> Result<(), CamelError> {
    use axum::Router;
    use axum::extract::State;
    use axum::http::Request;
    use axum::response::Response;
    use axum::routing::any;

    struct ListenerState {
        tx: mpsc::Sender<HttpRequest>,
    }

    async fn handler(
        State(state): State<Arc<ListenerState>>,
        req: Request<axum::body::Body>,
    ) -> Response {
        let (parts, body) = req.into_parts();
        // Surface body-read failures instead of `.unwrap_or_default()`-ing them
        // into an empty Vec. Previously a too-large or reset body was silently
        // accepted as an empty exchange. Oversized → 413, anything else → 400.
        let body_bytes = match axum::body::to_bytes(body, MAX_BODY_BYTES).await {
            Ok(bytes) => bytes,
            Err(e) => {
                // LengthLimitError lives in the error source chain; axum::Error
                // itself exposes no is::<T>(). Oversized → 413, else → 400.
                use std::error::Error as _;
                let oversized = e
                    .source()
                    .is_some_and(|src| src.is::<http_body_util::LengthLimitError>());
                let status = if oversized {
                    axum::http::StatusCode::PAYLOAD_TOO_LARGE
                } else {
                    axum::http::StatusCode::BAD_REQUEST
                };
                tracing::debug!(status = %status, error = %e, "source HTTP body read failed");
                return Response::builder()
                    .status(status)
                    .body(axum::body::Body::empty())
                    .unwrap(); // allow-unwrap
            }
        };

        let http_request = HttpRequest {
            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(),
            body: body_bytes.to_vec(),
        };

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

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

    let state = Arc::new(ListenerState { tx: request_tx });

    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) ─────────────────────────────────────────

/// Field-by-field converter between two bindgen-generated, structurally identical types.
/// Exists because dual `bindgen!` invocations produce distinct Rust types (TODO WIT-001).
/// Deletes when WIT-001 unifies the type definitions.
fn to_plugin_wasm_exchange(
    src: crate::source_bindings::camel::plugin::types::WasmExchange,
) -> crate::bindings::camel::plugin::types::WasmExchange {
    use crate::bindings::camel::plugin::types as plugin;
    use crate::source_bindings::camel::plugin::types as source;

    let convert_body = |body: source::WasmBody| -> plugin::WasmBody {
        match body {
            source::WasmBody::Empty => plugin::WasmBody::Empty,
            source::WasmBody::Text(s) => plugin::WasmBody::Text(s),
            source::WasmBody::Bytes(b) => plugin::WasmBody::Bytes(b),
            source::WasmBody::Json(s) => plugin::WasmBody::Json(s),
            source::WasmBody::Xml(s) => plugin::WasmBody::Xml(s),
        }
    };

    let convert_message = |msg: source::WasmMessage| -> plugin::WasmMessage {
        plugin::WasmMessage {
            headers: msg.headers,
            body: convert_body(msg.body),
        }
    };

    let convert_pattern = |p: source::WasmPattern| -> plugin::WasmPattern {
        match p {
            source::WasmPattern::InOnly => plugin::WasmPattern::InOnly,
            source::WasmPattern::InOut => plugin::WasmPattern::InOut,
        }
    };

    plugin::WasmExchange {
        input: convert_message(src.input),
        output: src.output.map(convert_message),
        properties: src.properties,
        pattern: convert_pattern(src.pattern),
        correlation_id: src.correlation_id,
        route_id: src.route_id,
        message_id: src.message_id,
    }
}

/// Receive exchanges from the guest, convert to native [`Exchange`], and
/// forward to the pipeline via [`ConsumerContext`].
///
/// Sends [`SubmitOutcome::Accepted`] or [`SubmitOutcome::Stopped`] back to
/// the guest via the oneshot reply channel.
pub async fn run_pipeline_bridge(
    mut exchange_rx: mpsc::Receiver<(WasmExchange, oneshot::Sender<SubmitOutcome>)>,
    ctx: ConsumerContext,
) -> Result<(), CamelError> {
    while let Some((wasm_exchange, reply_tx)) = exchange_rx.recv().await {
        // Convert source_bindings WasmExchange → bindings WasmExchange → native Exchange.
        let plugin_exchange = to_plugin_wasm_exchange(wasm_exchange);
        let mut exchange = Exchange::new(Message::default());
        crate::serde_bridge::wasm_to_exchange(plugin_exchange, &mut exchange);

        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::<HttpRequest>(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::<(WasmExchange, 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 test_to_plugin_wasm_exchange_converts_fields() {
        use crate::source_bindings::camel::plugin::types as src;
        let src_exchange = 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 plugin = to_plugin_wasm_exchange(src_exchange);
        assert_eq!(plugin.input.headers.len(), 1);
        assert_eq!(plugin.correlation_id, "corr-1");
        assert_eq!(plugin.route_id, Some("route-1".to_string()));
        assert!(matches!(
            plugin.pattern,
            crate::bindings::camel::plugin::types::WasmPattern::InOnly
        ));
    }
}