camel_component_grpc/

consumer.rs

use std::collections::HashMap as StdHashMap;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::OnceLock;
use std::sync::atomic::{AtomicU64, Ordering};

use async_trait::async_trait;
use base64::Engine;
use bytes::BytesMut;
use camel_api::security_policy::AuthPrincipal;
use camel_api::store_principal_properties;
use camel_api::{Body, CamelError, Exchange, Message, Value};
use camel_auth::{AuthenticatedPrincipal, CredentialSource, enforce_dispatch, install_carrier};
use camel_component_api::{
    ConcurrencyModel, Consumer, ConsumerContext, ConsumerStartupMode, ExchangeEnvelope,
    SecurityContext,
};
use camel_proto_compiler::ProtoCache;
use prost::Message as _;
use prost_reflect::{DynamicMessage, MessageDescriptor};
use tokio::sync::mpsc;
use tonic::Status;
use tracing::{debug, info};

use crate::config::GrpcServerConfig;
use crate::mode::GrpcMode;
use crate::server::GrpcDispatchTable;
use crate::server::GrpcKernelAuth;
use crate::server::GrpcServerRegistry;

static PROTO_CACHE: OnceLock<ProtoCache> = OnceLock::new();

/// Per-request kernel authentication bundle: the sealed principal minted at
/// the transport boundary plus the plan it must stay bound to
/// (`unify-transport-auth`, Task 2.1).
pub(crate) struct KernelRequestAuth {
    plan: camel_api::security_policy::RouteSecurityPlan,
    principal: AuthenticatedPrincipal,
}

impl KernelRequestAuth {
    /// Install the typed carrier on a fresh request exchange, then enforce
    /// the route binding. The carrier is installed BEFORE the pipeline runs
    /// — a fresh exchange is created per request, so every dispatched
    /// exchange must carry its own principal (the Task 2.9 dispatch check
    /// relies on this). The principal is also mirrored to exchange
    /// properties so route processors can observe the subject.
    /// `enforce_dispatch` fails closed (`Public`
    /// short-circuits to Ok) and denials map to the transport idiom.
    fn apply_to(&self, exchange: &mut Exchange) -> Result<(), Status> {
        store_principal_properties(exchange, self.principal.principal());
        install_carrier(exchange, &self.principal);
        enforce_dispatch(&self.plan, exchange).map_err(|e| match e {
            CamelError::Unauthenticated(msg) => Status::unauthenticated(msg),
            other => Status::internal(other.to_string()),
        })
    }
}

/// Bind a minted kernel principal to its route plan for one request.
///
/// `None` when the route has no kernel state (plan-less: Public
/// pass-through) or when the request carries no minted principal
/// (Public plans pass through without extraction).
fn kernel_request_auth(
    kernel: Option<&GrpcKernelAuth>,
    principal: Option<AuthenticatedPrincipal>,
) -> Option<KernelRequestAuth> {
    let kernel = kernel?;
    let principal = principal?;
    Some(KernelRequestAuth {
        plan: kernel.plan.clone(),
        principal,
    })
}

fn proto_cache() -> &'static ProtoCache {
    PROTO_CACHE.get_or_init(ProtoCache::new)
}

/// Map a pipeline error onto the transport denial idiom.
///
/// A pipeline policy denial (`CamelError::Unauthorized`, what
/// `SecurityPolicyService` returns) is `PERMISSION_DENIED` — the status
/// the deleted transport-side scratch evaluation used to emit, so denial
/// semantics survive with enforcement living wholly in the pipeline
/// layer. Every other pipeline error is system-broken, not a denial:
/// INTERNAL, unchanged.
fn pipeline_error_to_status(e: CamelError) -> Status {
    match e {
        CamelError::Unauthorized(msg) => Status::permission_denied(msg),
        other => Status::internal(format!("pipeline error: {other}")),
    }
}

/// Resolve the gRPC mode (unary/streaming) for a given method without creating a consumer.
pub fn resolve_grpc_mode(
    proto_path: &PathBuf,
    service_name: &str,
    method_name: &str,
) -> Result<GrpcMode, CamelError> {
    let cache = proto_cache();
    let pool = cache
        .get_or_compile(proto_path, std::iter::empty::<&std::path::Path>())
        .map_err(|e| CamelError::EndpointCreationFailed(format!("failed to compile proto: {e}")))?;

    let svc = pool.get_service_by_name(service_name).ok_or_else(|| {
        CamelError::EndpointCreationFailed(format!(
            "service descriptor not found: {}",
            service_name
        ))
    })?;

    let method = svc
        .methods()
        .find(|m| m.name() == method_name)
        .ok_or_else(|| {
            CamelError::EndpointCreationFailed(format!(
                "method descriptor not found: {}/{}",
                service_name, method_name
            ))
        })?;

    Ok(GrpcMode::from_method(&method))
}

const RESERVED_METADATA_KEYS: &[&str] = &[
    "content-type",
    "te",
    "grpc-encoding",
    "grpc-accept-encoding",
    "grpc-status",
    "grpc-message",
    "grpc-status-details-bin",
    "user-agent",
];

fn extract_metadata(metadata: &tonic::metadata::MetadataMap) -> Vec<(String, serde_json::Value)> {
    let mut headers = Vec::new();
    for key_and_value in metadata.iter() {
        use tonic::metadata::KeyAndValueRef;
        match key_and_value {
            KeyAndValueRef::Ascii(key, value) => {
                let key_str = key.as_str();
                if RESERVED_METADATA_KEYS.contains(&key_str) {
                    continue;
                }
                if let Ok(v) = value.to_str() {
                    headers.push((
                        key_str.to_string(),
                        serde_json::Value::String(v.to_string()),
                    ));
                }
            }
            KeyAndValueRef::Binary(key, value) => {
                let key_str = key.as_str();
                if RESERVED_METADATA_KEYS.contains(&key_str) {
                    continue;
                }
                let encoded = base64::engine::general_purpose::STANDARD.encode(value);
                headers.push((format!("bin:{key_str}"), serde_json::Value::String(encoded)));
            }
        }
    }
    headers
}

pub(crate) enum GrpcStreamItem {
    Message(Vec<u8>),
    Error(tonic::Status),
    Done,
}

pub(crate) enum GrpcReply {
    Ok(Vec<u8>),
    Err(tonic::Status),
}

/// Request envelope crossing the server→consumer boundary.
///
/// `kernel_principal` is the sealed principal minted by
/// `kernel_authenticate` at the transport boundary
/// (`unify-transport-auth`, Task 2.1) and is installed as the exchange's
/// typed carrier before the pipeline runs. Policy evaluation is NOT done
/// at the transport (the legacy scratch arm was deleted in
/// `finish-auth-flip`): enforcement lives in the pipeline layer plus the
/// strict dispatch check.
pub(crate) enum GrpcRequestEnvelope {
    Unary {
        metadata: tonic::metadata::MetadataMap,
        body: Vec<u8>,
        reply_tx: tokio::sync::oneshot::Sender<GrpcReply>,
        kernel_principal: Option<AuthenticatedPrincipal>,
    },
    ServerStreaming {
        metadata: tonic::metadata::MetadataMap,
        body: Vec<u8>,
        reply_tx: mpsc::Sender<GrpcStreamItem>,
        kernel_principal: Option<AuthenticatedPrincipal>,
    },
    ClientStreaming {
        metadata: tonic::metadata::MetadataMap,
        body_rx: mpsc::Receiver<Vec<u8>>,
        reply_tx: tokio::sync::oneshot::Sender<GrpcReply>,
        kernel_principal: Option<AuthenticatedPrincipal>,
    },
    Bidi {
        metadata: tonic::metadata::MetadataMap,
        body_rx: mpsc::Receiver<Vec<u8>>,
        reply_tx: mpsc::Sender<GrpcStreamItem>,
        kernel_principal: Option<AuthenticatedPrincipal>,
    },
}

// ── Observer registry ──────────────────────────────────────────────────────

static OBSERVER_REGISTRY: OnceLock<std::sync::Mutex<StdHashMap<String, GrpcStreamObserver>>> =
    OnceLock::new();

static OBSERVER_COUNTER: AtomicU64 = AtomicU64::new(0);

fn next_observer_id() -> String {
    let n = OBSERVER_COUNTER.fetch_add(1, Ordering::Relaxed);
    format!("obs-{n}")
}

fn observer_registry() -> &'static std::sync::Mutex<StdHashMap<String, GrpcStreamObserver>> {
    OBSERVER_REGISTRY.get_or_init(|| std::sync::Mutex::new(StdHashMap::new()))
}

fn register_observer(id: String, observer: GrpcStreamObserver) {
    let registry = observer_registry();
    let mut registry = match registry.lock() {
        Ok(g) => g,
        Err(poisoned) => poisoned.into_inner(),
    };
    registry.insert(id, observer);
}

fn remove_observer(id: &str) -> Option<GrpcStreamObserver> {
    let registry = observer_registry();
    let mut registry = match registry.lock() {
        Ok(g) => g,
        Err(poisoned) => poisoned.into_inner(),
    };
    registry.remove(id)
}

pub fn take_stream_observer(exchange: &Exchange) -> Option<GrpcStreamObserver> {
    let id = exchange
        .properties
        .get("CamelGrpcStreamObserverId")?
        .as_str()?;
    remove_observer(id)
}

// ── Observer guard (auto-cleanup on Drop) ──────────────────────────────────

struct ObserverGuard {
    id: String,
}

impl ObserverGuard {
    fn new(id: String) -> Self {
        Self { id }
    }
}

impl Drop for ObserverGuard {
    fn drop(&mut self) {
        remove_observer(&self.id);
    }
}

// ── GrpcStreamObserver ─────────────────────────────────────────────────────

#[derive(Clone)]
pub struct GrpcStreamObserver {
    tx: mpsc::Sender<GrpcStreamItem>,
    resp_desc: MessageDescriptor,
}

impl GrpcStreamObserver {
    pub(crate) fn new(tx: mpsc::Sender<GrpcStreamItem>, resp_desc: MessageDescriptor) -> Self {
        Self { tx, resp_desc }
    }

    pub async fn on_next(&self, json: serde_json::Value) -> Result<(), CamelError> {
        let encoded = json_to_protobuf_bytes(json, self.resp_desc.clone())
            .map_err(|e| CamelError::ProcessorError(format!("failed to encode protobuf: {e}")))?;
        self.tx
            .send(GrpcStreamItem::Message(encoded))
            .await
            .map_err(|_| CamelError::ProcessorError("stream observer channel closed".into()))
    }

    pub async fn on_error(&self, status: Status) {
        if self.tx.send(GrpcStreamItem::Error(status)).await.is_err() {
            tracing::debug!("grpc stream observer: failed to send error, channel closed");
        }
    }

    pub async fn on_completed(&self) {
        if self.tx.send(GrpcStreamItem::Done).await.is_err() {
            tracing::debug!("grpc stream observer: failed to send done, channel closed");
        }
    }
}

// ── Helper ─────────────────────────────────────────────────────────────────

fn json_to_protobuf_bytes(
    json: serde_json::Value,
    desc: MessageDescriptor,
) -> Result<Vec<u8>, Status> {
    let json_str = serde_json::to_string(&json)
        .map_err(|e| Status::internal(format!("failed to serialize JSON: {e}")))?;
    let mut de = serde_json::Deserializer::from_str(&json_str);
    let resp_dyn = DynamicMessage::deserialize(desc, &mut de)
        .map_err(|e| Status::internal(format!("failed to parse JSON into protobuf: {e}")))?;
    let mut buf = BytesMut::new();
    prost::Message::encode(&resp_dyn, &mut buf)
        .map_err(|e| Status::internal(format!("failed to encode protobuf: {e}")))?;
    Ok(buf.to_vec())
}

// ── GrpcConsumer ───────────────────────────────────────────────────────────

/// Reject credential sources the gRPC transport cannot carry.
///
/// gRPC metadata maps to HTTP headers only: `authorization_header` maps to the
/// `authorization` metadata key and `{header: {name}}` to the same-named
/// metadata key. Query parameters and cookies have no gRPC metadata
/// representation, so a route declaring them must fail at load (ADR-0033
/// fail-closed), not silently authenticate nothing at request time.
pub(crate) fn validate_credential_sources(sources: &[CredentialSource]) -> Result<(), CamelError> {
    for source in sources {
        let source_kind = match source {
            CredentialSource::QueryParam { .. } => "query_param",
            CredentialSource::Cookie { .. } => "cookie",
            _ => continue,
        };
        return Err(CamelError::Config(format!(
            "grpc routes cannot carry {source_kind} credential sources; supported: authorization_header, header" // allow-secret: field names in error text, not values
        )));
    }
    Ok(())
}

pub struct GrpcConsumer {
    host: String,
    port: u16,
    path: String,
    proto_path: PathBuf,
    service_name: String,
    method_name: String,
    mode: GrpcMode,
    security_ctx: Option<SecurityContext>,
    runtime: Arc<dyn camel_component_api::RuntimeObservability>,
    server_config: GrpcServerConfig,
}

impl GrpcConsumer {
    #[allow(clippy::too_many_arguments)]
    pub fn new(
        host: String,
        port: u16,
        path: String,
        proto_path: PathBuf,
        service_name: String,
        method_name: String,
        mode: GrpcMode,
        runtime: Arc<dyn camel_component_api::RuntimeObservability>,
        server_config: GrpcServerConfig,
    ) -> Self {
        Self {
            host,
            port,
            path,
            proto_path,
            service_name,
            method_name,
            mode,
            security_ctx: None,
            runtime,
            server_config,
        }
    }

    fn resolve_descriptors(&self) -> Result<(MessageDescriptor, MessageDescriptor), CamelError> {
        let cache = proto_cache();
        let pool = cache
            .get_or_compile(&self.proto_path, std::iter::empty::<&std::path::Path>())
            .map_err(|e| {
                CamelError::EndpointCreationFailed(format!("failed to compile proto: {e}"))
            })?;

        let svc = pool
            .get_service_by_name(&self.service_name)
            .ok_or_else(|| {
                CamelError::EndpointCreationFailed(format!(
                    "service descriptor not found: {}",
                    self.service_name
                ))
            })?;

        let method = svc
            .methods()
            .find(|m| m.name() == self.method_name)
            .ok_or_else(|| {
                CamelError::EndpointCreationFailed(format!(
                    "method descriptor not found: {}/{}",
                    self.service_name, self.method_name
                ))
            })?;

        Ok((method.input(), method.output()))
    }

    /// Validate every credential-source list this route can extract from:
    /// the configured sources and, when a compiled plan is present, the
    /// plan's own sources (fail-closed at load, ADR-0033).
    fn validate_route_credential_sources(&self) -> Result<(), CamelError> {
        let Some(sec_ctx) = &self.security_ctx else {
            return Ok(());
        };
        validate_credential_sources(&sec_ctx.credential_sources)?;
        if let Some(plan) = &sec_ctx.plan {
            validate_credential_sources(&plan.credential_sources)?;
        }
        Ok(())
    }

    pub async fn start_with_listener(
        &mut self,
        ctx: ConsumerContext,
        listener: tokio::net::TcpListener,
    ) -> Result<(), CamelError> {
        self.validate_route_credential_sources()?;
        let dispatch = GrpcServerRegistry::global()
            .get_or_spawn_with_listener(
                listener,
                &self.host,
                self.port,
                self.server_config.clone(),
                Arc::clone(&self.runtime),
            )
            .await?;
        self.start_inner(ctx, dispatch).await
    }

    async fn start_inner(
        &mut self,
        ctx: ConsumerContext,
        dispatch: GrpcDispatchTable,
    ) -> Result<(), CamelError> {
        let (req_desc, resp_desc) = self.resolve_descriptors()?;
        let mode = self.mode;

        let (env_tx, mut env_rx) = mpsc::channel::<GrpcRequestEnvelope>(64);
        // Kernel interceptor state is captured HERE, at dispatch-entry
        // construction, from the security context wired before start
        // (Task 2.1 construction-order lifecycle fix). The per-request
        // handlers are built from this entry, so the plan is present
        // before any request arrives — never patched on afterwards.
        let kernel = self
            .security_ctx
            .as_ref()
            .and_then(GrpcKernelAuth::from_security_context)
            .map(Arc::new);
        {
            let mut table = dispatch.write().await;
            if table.contains_key(&self.path) {
                return Err(CamelError::EndpointCreationFailed(format!(
                    "duplicate gRPC consumer path: {}",
                    self.path
                )));
            }
            table.insert(self.path.clone(), (env_tx, mode, kernel.clone()));
        }

        let path = self.path.clone();
        let host = self.host.clone();
        let port = self.port;
        let sender = ctx.sender();

        info!(
            path = %path,
            host = %host,
            port = port,
            mode = ?mode,
            "grpc consumer started, waiting for requests"
        );

        // NOTE: Long-running bidi streams hold a semaphore permit for their duration.
        // If this becomes an issue, consider separate concurrency limits for streaming vs unary.
        let semaphore = std::sync::Arc::new(tokio::sync::Semaphore::new(64));
        let mut join_set = tokio::task::JoinSet::new();

        loop {
            tokio::select! {
                biased;
                _ = ctx.cancelled() => {
                    info!(
                        path = %path,
                        "grpc consumer cancelled, shutting down"
                    );
                    break;
                }
                envelope = env_rx.recv() => {
                    let Some(envelope) = envelope else { break };

                    let sem = semaphore.clone();
                    let permit = sem.acquire_owned().await.map_err(|_| CamelError::ChannelClosed)?;
                    let req_desc = req_desc.clone();
                    let resp_desc = resp_desc.clone();
                    let sender = sender.clone();
                    let correlation_id = next_observer_id();
                    let path_for_log = path.clone();
                    // Kernel state captured at dispatch-entry construction,
                    // cloned per request; the principal minted by the
                    // interceptor binds to this plan for the carrier install.
                    // Per-request policy evaluation is NOT done here —
                    // enforcement lives in the pipeline layer plus the
                    // strict dispatch check.
                    let kernel = kernel.clone();

                    debug!(
                        path = %path_for_log,
                        correlation_id = %correlation_id,
                        "grpc consumer received request"
                    );

                    join_set.spawn(async move {
                        let _permit = permit;
                        match envelope {
                            GrpcRequestEnvelope::Unary { metadata, body, reply_tx, kernel_principal } => {
                                debug!(
                                    path = %path_for_log,
                                    correlation_id = %correlation_id,
                                    size = body.len(),
                                    "grpc consumer processing unary request"
                                );

                                let kernel_auth = kernel_request_auth(kernel.as_deref(), kernel_principal);
                                let result = process_unary_request(
                                    body, metadata, req_desc, resp_desc, sender, kernel_auth,
                                ).await;
                                let reply = match result {
                                    Ok(bytes) => GrpcReply::Ok(bytes),
                                    Err(status) => GrpcReply::Err(status),
                                };
                                let _ = reply_tx.send(reply);
                            }
                            GrpcRequestEnvelope::ServerStreaming { metadata, body, reply_tx, kernel_principal } => {
                                debug!(
                                    path = %path_for_log,
                                    correlation_id = %correlation_id,
                                    size = body.len(),
                                    "grpc consumer processing server streaming request"
                                );

                                let kernel_auth = kernel_request_auth(kernel.as_deref(), kernel_principal);
                                process_server_streaming_request(
                                    body, metadata, req_desc, resp_desc, sender, reply_tx, kernel_auth,
                                ).await;
                            }
                            GrpcRequestEnvelope::ClientStreaming { metadata, body_rx, reply_tx, kernel_principal } => {
                                debug!(
                                    path = %path_for_log,
                                    correlation_id = %correlation_id,
                                    "grpc consumer processing client streaming request"
                                );

                                let kernel_auth = kernel_request_auth(kernel.as_deref(), kernel_principal);
                                process_client_streaming_request(
                                    body_rx, metadata, req_desc, resp_desc, sender, reply_tx, kernel_auth,
                                ).await;
                            }
                            GrpcRequestEnvelope::Bidi { metadata, body_rx, reply_tx, kernel_principal } => {
                                debug!(
                                    path = %path_for_log,
                                    correlation_id = %correlation_id,
                                    "grpc consumer processing bidi streaming request"
                                );

                                let kernel_auth = kernel_request_auth(kernel.as_deref(), kernel_principal);
                                process_bidi_request(
                                    body_rx, metadata, req_desc, resp_desc, sender, reply_tx, kernel_auth,
                                ).await;
                            }
                        }
                    });
                }
            }
        }

        join_set.shutdown().await;

        GrpcServerRegistry::global()
            .unregister(&host, port, &path)
            .await;

        info!(
            path = %path,
            "grpc consumer stopped"
        );

        Ok(())
    }
}

#[async_trait]
impl Consumer for GrpcConsumer {
    async fn start(&mut self, ctx: ConsumerContext) -> Result<(), CamelError> {
        self.validate_route_credential_sources()?;
        info!(
            host = %self.host,
            port = self.port,
            service = %self.service_name,
            method = %self.method_name,
            mode = ?self.mode,
            "grpc consumer starting"
        );
        let dispatch = GrpcServerRegistry::global()
            .get_or_spawn(
                &self.host,
                self.port,
                self.server_config.clone(),
                Arc::clone(&self.runtime),
            )
            .await?;
        // gRPC listener is bound inside get_or_spawn (TcpListener::bind
        // before tokio::spawn). Signal readiness now that the bind succeeded.
        ctx.mark_ready();
        self.start_inner(ctx, dispatch).await
    }

    async fn stop(&mut self) -> Result<(), CamelError> {
        info!(
            host = %self.host,
            port = self.port,
            service = %self.service_name,
            method = %self.method_name,
            "grpc consumer stopping"
        );
        GrpcServerRegistry::global()
            .unregister(&self.host, self.port, &self.path)
            .await;
        Ok(())
    }

    fn concurrency_model(&self) -> ConcurrencyModel {
        ConcurrencyModel::Concurrent { max: None }
    }

    fn startup_mode(&self) -> ConsumerStartupMode {
        ConsumerStartupMode::Explicit
    }

    fn set_security_context(&mut self, ctx: SecurityContext) {
        self.security_ctx = Some(ctx);
    }
}

// ── Unary processor (unchanged) ────────────────────────────────────────────

async fn process_unary_request(
    body: Vec<u8>,
    metadata: tonic::metadata::MetadataMap,
    req_desc: MessageDescriptor,
    resp_desc: MessageDescriptor,
    sender: mpsc::Sender<ExchangeEnvelope>,
    kernel_auth: Option<KernelRequestAuth>,
) -> Result<Vec<u8>, Status> {
    let req_dyn = DynamicMessage::decode(req_desc, body.as_slice())
        .map_err(|e| Status::invalid_argument(format!("failed to decode protobuf: {e}")))?;

    let json = serde_json::to_value(&req_dyn).map_err(|e| {
        Status::invalid_argument(format!("failed to convert protobuf to JSON: {e}"))
    })?;

    let mut msg = Message::new(Body::Json(json));
    for (k, v) in extract_metadata(&metadata) {
        msg.set_header(k, v);
    }

    let mut exchange = Exchange::new(msg);
    if let Some(auth) = kernel_auth.as_ref() {
        // Carrier install + route-binding enforcement before the pipeline.
        auth.apply_to(&mut exchange)?;
    }

    let (reply_tx, reply_rx) = tokio::sync::oneshot::channel();
    let envelope = ExchangeEnvelope {
        exchange,
        reply_tx: Some(reply_tx),
    };

    sender
        .send(envelope)
        .await
        .map_err(|_| Status::internal("pipeline channel closed"))?;

    let result = reply_rx
        .await
        .map_err(|_| Status::internal("pipeline reply dropped"))?
        .map_err(pipeline_error_to_status)?;

    let resp_json = match result.input.body {
        Body::Json(v) => v,
        other => {
            return Err(Status::internal(format!(
                "expected JSON response body from pipeline, got {other:?}"
            )));
        }
    };

    let json_str = serde_json::to_string(&resp_json)
        .map_err(|e| Status::internal(format!("failed to serialize response JSON: {e}")))?;
    let mut de = serde_json::Deserializer::from_str(&json_str);
    let resp_dyn = DynamicMessage::deserialize(resp_desc, &mut de)
        .map_err(|e| Status::internal(format!("failed to parse JSON into protobuf: {e}")))?;

    let mut buf = BytesMut::new();
    resp_dyn
        .encode(&mut buf)
        .map_err(|e| Status::internal(format!("failed to encode protobuf response: {e}")))?;

    Ok(buf.to_vec())
}

// ── Server-streaming processor ─────────────────────────────────────────────

async fn process_server_streaming_request(
    body: Vec<u8>,
    metadata: tonic::metadata::MetadataMap,
    req_desc: MessageDescriptor,
    resp_desc: MessageDescriptor,
    sender: mpsc::Sender<ExchangeEnvelope>,
    reply_tx: mpsc::Sender<GrpcStreamItem>,
    kernel_auth: Option<KernelRequestAuth>,
) {
    let req_dyn = match DynamicMessage::decode(req_desc, body.as_slice()) {
        Ok(m) => m,
        Err(e) => {
            let _ = reply_tx
                .send(GrpcStreamItem::Error(Status::invalid_argument(format!(
                    "failed to decode protobuf: {e}"
                ))))
                .await;
            return;
        }
    };

    let json = match serde_json::to_value(&req_dyn) {
        Ok(v) => v,
        Err(e) => {
            let _ = reply_tx
                .send(GrpcStreamItem::Error(Status::invalid_argument(format!(
                    "failed to convert protobuf to JSON: {e}"
                ))))
                .await;
            return;
        }
    };

    let mut msg = Message::new(Body::Json(json));
    for (k, v) in extract_metadata(&metadata) {
        msg.set_header(k, v);
    }

    let observer = GrpcStreamObserver::new(reply_tx.clone(), resp_desc);
    let observer_id = next_observer_id();
    register_observer(observer_id.clone(), observer.clone());
    let _guard = ObserverGuard::new(observer_id.clone());

    let mut exchange = Exchange::new(msg);
    if let Some(auth) = kernel_auth.as_ref()
        && let Err(status) = auth.apply_to(&mut exchange)
    {
        let _ = reply_tx.send(GrpcStreamItem::Error(status)).await;
        return;
    }
    exchange.set_property("CamelGrpcStreamObserverId", Value::String(observer_id));

    // The envelope carries a pipeline reply channel so a pipeline error
    // (policy denial included) reaches this processor instead of dying
    // with `reply_tx: None` — the regression where a denial ended the
    // stream as a silent, empty success.
    let (pipeline_reply_tx, pipeline_reply_rx) = tokio::sync::oneshot::channel();
    let envelope = ExchangeEnvelope {
        exchange,
        reply_tx: Some(pipeline_reply_tx),
    };

    if sender.send(envelope).await.is_err() {
        let _ = reply_tx
            .send(GrpcStreamItem::Error(Status::internal(
                "pipeline channel closed",
            )))
            .await;
        return;
    }

    // The pipeline verdict decides the stream's terminal frame: a
    // pipeline error is surfaced client-visibly via the observer (the
    // same denial idiom the deleted transport-side scratch evaluation
    // emitted). A successful result streamed through the observer adds
    // nothing; so does a dropped reply sender (route stand-ins that
    // never reply) — the observer stream stays the truth either way.
    if let Ok(Err(e)) = pipeline_reply_rx.await {
        observer.on_error(pipeline_error_to_status(e)).await;
    }

    // Wait for the stream receiver to be dropped (stream complete).
    // This keeps the guard alive so the observer stays registered until
    // the route is done. If take_stream_observer was called, the guard's
    // Drop is a no-op. If not, the guard cleans up the leaked observer.
    reply_tx.closed().await;
}

// ── Client-streaming processor ─────────────────────────────────────────────

async fn process_client_streaming_request(
    mut body_rx: mpsc::Receiver<Vec<u8>>,
    metadata: tonic::metadata::MetadataMap,
    req_desc: MessageDescriptor,
    resp_desc: MessageDescriptor,
    sender: mpsc::Sender<ExchangeEnvelope>,
    reply_tx: tokio::sync::oneshot::Sender<GrpcReply>,
    kernel_auth: Option<KernelRequestAuth>,
) {
    while let Some(body) = body_rx.recv().await {
        let req_dyn = match DynamicMessage::decode(req_desc.clone(), body.as_slice()) {
            Ok(d) => d,
            Err(e) => {
                let _ = reply_tx.send(GrpcReply::Err(Status::invalid_argument(format!(
                    "failed to decode protobuf: {e}"
                ))));
                return;
            }
        };

        let json = match serde_json::to_value(&req_dyn) {
            Ok(j) => j,
            Err(e) => {
                let _ = reply_tx.send(GrpcReply::Err(Status::internal(format!(
                    "failed to convert protobuf to JSON: {e}"
                ))));
                return;
            }
        };

        let mut msg = Message::new(Body::Json(json));
        for (k, v) in extract_metadata(&metadata) {
            msg.set_header(k, v);
        }
        msg.set_header(
            "CamelGrpcClientStreaming".to_string(),
            serde_json::Value::Bool(true),
        );

        let mut exchange = Exchange::new(msg);
        if let Some(auth) = kernel_auth.as_ref()
            && let Err(status) = auth.apply_to(&mut exchange)
        {
            let _ = reply_tx.send(GrpcReply::Err(status));
            return;
        }
        let (reply_tx_pipe, reply_rx_pipe) = tokio::sync::oneshot::channel();
        let envelope = ExchangeEnvelope {
            exchange,
            reply_tx: Some(reply_tx_pipe),
        };

        if sender.send(envelope).await.is_err() {
            let _ = reply_tx.send(GrpcReply::Err(Status::internal("pipeline channel closed")));
            return;
        }

        // Intentionally discard intermediate replies — only the completion exchange's reply matters.
        let _ = reply_rx_pipe.await;
    }

    // Stream complete — send final Exchange with completion marker
    let mut completion_msg = Message::new(Body::Json(serde_json::Value::Null));
    for (k, v) in extract_metadata(&metadata) {
        completion_msg.set_header(k, v);
    }
    completion_msg.set_header(
        "CamelGrpcClientStreaming".to_string(),
        serde_json::Value::Bool(true),
    );
    completion_msg.set_header(
        "CamelGrpcClientStreamComplete".to_string(),
        serde_json::Value::Bool(true),
    );

    let mut completion_exchange = Exchange::new(completion_msg);
    if let Some(auth) = kernel_auth.as_ref()
        && let Err(status) = auth.apply_to(&mut completion_exchange)
    {
        let _ = reply_tx.send(GrpcReply::Err(status));
        return;
    }
    let (reply_tx_pipe, reply_rx_pipe) = tokio::sync::oneshot::channel();
    let envelope = ExchangeEnvelope {
        exchange: completion_exchange,
        reply_tx: Some(reply_tx_pipe),
    };

    if sender.send(envelope).await.is_err() {
        let _ = reply_tx.send(GrpcReply::Err(Status::internal("pipeline channel closed")));
        return;
    }

    // The route's response to the completion Exchange becomes the gRPC response
    let result = match reply_rx_pipe.await {
        Ok(Ok(exchange)) => exchange,
        Ok(Err(e)) => {
            let _ = reply_tx.send(GrpcReply::Err(pipeline_error_to_status(e)));
            return;
        }
        Err(_) => {
            let _ = reply_tx.send(GrpcReply::Err(Status::internal("pipeline reply dropped")));
            return;
        }
    };

    let resp_json = match result.input.body {
        Body::Json(v) => v,
        other => {
            let _ = reply_tx.send(GrpcReply::Err(Status::internal(format!(
                "expected JSON response body from pipeline, got {other:?}"
            ))));
            return;
        }
    };

    let encoded = match json_to_protobuf_bytes(resp_json, resp_desc) {
        Ok(b) => b,
        Err(e) => {
            let _ = reply_tx.send(GrpcReply::Err(Status::internal(format!(
                "failed to encode response: {e}",
            ))));
            return;
        }
    };

    let _ = reply_tx.send(GrpcReply::Ok(encoded));
}

// ── Bidi-streaming processor ───────────────────────────────────────────────

async fn process_bidi_request(
    mut body_rx: mpsc::Receiver<Vec<u8>>,
    metadata: tonic::metadata::MetadataMap,
    req_desc: MessageDescriptor,
    resp_desc: MessageDescriptor,
    sender: mpsc::Sender<ExchangeEnvelope>,
    reply_tx: mpsc::Sender<GrpcStreamItem>,
    kernel_auth: Option<KernelRequestAuth>,
) {
    let observer = GrpcStreamObserver::new(reply_tx.clone(), resp_desc);
    let observer_id = next_observer_id();
    register_observer(observer_id.clone(), observer.clone());
    let _guard = ObserverGuard::new(observer_id.clone());

    // Spawn a task to forward messages from the client stream to the pipeline
    let sender_clone = sender.clone();
    let metadata_clone = metadata.clone();
    let req_desc_clone = req_desc.clone();

    let forward_task = tokio::spawn(async move {
        let mut sequence: u64 = 0;
        while let Some(body) = body_rx.recv().await {
            let req_dyn = match DynamicMessage::decode(req_desc_clone.clone(), body.as_slice()) {
                Ok(m) => m,
                Err(e) => {
                    let _ = observer
                        .on_error(Status::invalid_argument(format!(
                            "failed to decode protobuf: {e}"
                        )))
                        .await;
                    continue;
                }
            };

            let json = match serde_json::to_value(&req_dyn) {
                Ok(v) => v,
                Err(e) => {
                    let _ = observer
                        .on_error(Status::invalid_argument(format!(
                            "failed to convert protobuf to JSON: {e}"
                        )))
                        .await;
                    continue;
                }
            };

            let mut msg = Message::new(Body::Json(json));
            for (k, v) in extract_metadata(&metadata_clone) {
                msg.set_header(k, v);
            }

            msg.set_header(
                "CamelGrpcBidiSequence",
                serde_json::Value::Number(sequence.into()),
            );
            sequence += 1;

            let mut exchange = Exchange::new(msg);
            if let Some(auth) = kernel_auth.as_ref()
                && let Err(status) = auth.apply_to(&mut exchange)
            {
                let _ = observer.on_error(status).await;
                break;
            }
            exchange.set_property(
                "CamelGrpcStreamObserverId",
                Value::String(observer_id.clone()),
            );

            // Each message envelope carries a pipeline reply channel so
            // a pipeline error (policy denial included) becomes a
            // client-visible stream error instead of dying with
            // `reply_tx: None`. The forwarding loop stays non-blocking:
            // a per-message watcher renders the verdict via the
            // observer.
            let (pipeline_reply_tx, pipeline_reply_rx) = tokio::sync::oneshot::channel();
            let envelope = ExchangeEnvelope {
                exchange,
                reply_tx: Some(pipeline_reply_tx),
            };

            if sender_clone.send(envelope).await.is_err() {
                let _ = observer
                    .on_error(Status::internal("pipeline channel closed"))
                    .await;
                break;
            }

            let verdict_observer = observer.clone();
            tokio::spawn(async move {
                if let Ok(Err(e)) = pipeline_reply_rx.await {
                    verdict_observer.on_error(pipeline_error_to_status(e)).await;
                }
            });
        }

        // Signal completion when client stream ends
        observer.on_completed().await;
    });

    // Wait for the forward task to complete
    let _ = forward_task.await;
}

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

    #[test]
    fn grpc_credential_sources_uncarryable_rejected_at_load() {
        let query = CredentialSource::QueryParam {
            param: "ticket".to_string(),
        };
        let err = validate_credential_sources(&[query]).unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("query_param"), "message was: {msg}");
        assert!(msg.contains("grpc"), "message was: {msg}");

        let cookie = CredentialSource::Cookie {
            name: "session".to_string(),
        };
        let err = validate_credential_sources(&[cookie]).unwrap_err();
        let msg = err.to_string();
        assert!(msg.contains("cookie"), "message was: {msg}");
        assert!(msg.contains("grpc"), "message was: {msg}");

        // Carryable sources pass validation.
        let carryable = vec![
            CredentialSource::AuthorizationHeader,
            CredentialSource::Header {
                name: "x-api-key".to_string(),
            },
        ];
        assert!(validate_credential_sources(&carryable).is_ok());
        assert!(validate_credential_sources(&[]).is_ok());
    }
}