camel-core 0.60.0

use std::sync::Arc;

pub(crate) enum FunctionStagingMode {
    DirectAdd,
    DryCompile,
    HotReload { generation: u64 },
}

use crate::lifecycle::adapters::step_compilers::CompiledStep;
use crate::{CacheRegistry, ClaimCheckRegistry, IdempotentRegistry};
use camel_api::filter::{PredicateSource, ValueSource};
use camel_api::{CamelError, Exchange, FunctionInvoker, ProducerContext};
use camel_bean::BeanRegistry;
use camel_component_api::{ComponentContext, RuntimeObservability};
use camel_language_api::{
    EvalMeta, Expression, Language, LanguageExpressionEval, LanguagePredicateEval, Predicate,
};
use camel_processor::{
    ClaimKeySource, EnrichmentStrategy, MessageIdSource, SortKeySource, ThrowOnNoPoll,
    UseEnrichedBody,
};

use crate::lifecycle::adapters::route_controller::SharedLanguageRegistry;
use crate::lifecycle::application::route_definition::{BuilderStep, LanguageExpressionDef};
use crate::shared::components::domain::Registry;

/// Build the trusted [`EvalMeta`] for a compiling step.
///
/// `route_id` falls back to `"(unassigned)"` when the route under compilation
/// has no id yet. `step_id` is `"{verb}#{index}"` where `index` is the step
/// ordinal threaded from the compile loop. `target` carries the property or
/// header key, or `"body"`, for verbs that have one.
pub(crate) fn eval_meta(
    ctx: &crate::lifecycle::adapters::step_compilers::CompilationContext<'_>,
    language: &str,
    verb: &str,
    index: usize,
    target: Option<String>,
) -> EvalMeta {
    EvalMeta {
        language: language.to_string(),
        route_id: ctx.route_id.unwrap_or("(unassigned)").to_string(),
        step_id: format!("{verb}#{index}"),
        verb: verb.to_string(),
        target,
    }
}

pub(crate) fn resolve_language(
    languages: &SharedLanguageRegistry,
    language: &str,
) -> Result<Arc<dyn Language>, CamelError> {
    let guard = languages
        .lock()
        .expect("mutex poisoned: another thread panicked while holding this lock"); // allow-unwrap
    guard.get(language).cloned().ok_or_else(|| {
        CamelError::RouteError(format!(
            "language `{language}` is not registered in CamelContext"
        ))
    })
}

/// Test-only language that opts in to mutating expressions. Registered as
/// `"testmut"`; evaluating it records that it ran by setting the `testmut`
/// exchange property.
#[cfg(test)]
pub(crate) struct MutatingTestLanguage;

#[cfg(test)]
impl Language for MutatingTestLanguage {
    fn name(&self) -> &'static str {
        "testmut"
    }

    fn create_expression(
        &self,
        _script: &str,
    ) -> Result<Box<dyn Expression>, camel_language_api::LanguageError> {
        Ok(Box::new(MutatingTestStubExpression))
    }

    fn create_predicate(
        &self,
        _script: &str,
    ) -> Result<Box<dyn Predicate>, camel_language_api::LanguageError> {
        Ok(Box::new(MutatingTestStubPredicate))
    }

    fn create_mutating_expression(
        &self,
        _script: &str,
    ) -> Result<Box<dyn camel_language_api::MutatingExpression>, camel_language_api::LanguageError>
    {
        Ok(Box::new(MutatingTestExpr))
    }
}

#[cfg(test)]
struct MutatingTestStubExpression;

#[cfg(test)]
#[async_trait::async_trait]
impl Expression for MutatingTestStubExpression {
    async fn evaluate(
        &self,
        _exchange: &Exchange,
    ) -> Result<camel_api::Value, camel_language_api::LanguageError> {
        Ok(camel_api::Value::Null)
    }
}

#[cfg(test)]
struct MutatingTestStubPredicate;

#[cfg(test)]
#[async_trait::async_trait]
impl Predicate for MutatingTestStubPredicate {
    async fn matches(
        &self,
        _exchange: &Exchange,
    ) -> Result<bool, camel_language_api::LanguageError> {
        Ok(false)
    }
}

/// Test-only mutating expression that sets the `testmut` property to `"ran"`
/// and returns `Value::Null`.
#[cfg(test)]
pub(crate) struct MutatingTestExpr;

#[cfg(test)]
#[async_trait::async_trait]
impl camel_language_api::MutatingExpression for MutatingTestExpr {
    async fn evaluate(
        &self,
        exchange: &mut Exchange,
    ) -> Result<camel_api::Value, camel_language_api::LanguageError> {
        exchange.set_property("testmut", camel_api::Value::String("ran".into()));
        Ok(camel_api::Value::Null)
    }
}

pub(crate) fn compile_language_expression(
    languages: &SharedLanguageRegistry,
    expression: &LanguageExpressionDef,
) -> Result<Arc<dyn Expression>, CamelError> {
    let language = resolve_language(languages, &expression.language)?;
    let compiled = language
        .create_expression(&expression.source)
        .map_err(|e| {
            CamelError::RouteError(format!(
                "failed to compile {} expression `{}`: {e}",
                expression.language, expression.source
            ))
        })?;
    Ok(Arc::from(compiled))
}

pub(crate) fn compile_language_predicate(
    languages: &SharedLanguageRegistry,
    expression: &LanguageExpressionDef,
) -> Result<Arc<dyn Predicate>, CamelError> {
    let language = resolve_language(languages, &expression.language)?;
    let compiled = language.create_predicate(&expression.source).map_err(|e| {
        CamelError::RouteError(format!(
            "failed to compile {} predicate `{}`: {e}",
            expression.language, expression.source
        ))
    })?;
    Ok(Arc::from(compiled))
}

/// Compile a `LanguageExpressionDef` into a fallible [`PredicateSource`].
///
/// The async arm binds the compiled predicate to `meta`; evaluation failures
/// surface as `CamelError::ExpressionFailed` carrying that metadata instead of
/// silently dropping the exchange.
pub(crate) fn compile_filter_predicate(
    languages: &SharedLanguageRegistry,
    expression: &LanguageExpressionDef,
    meta: EvalMeta,
) -> Result<PredicateSource, CamelError> {
    let predicate = compile_language_predicate(languages, expression)?;
    Ok(PredicateSource::Async(
        LanguagePredicateEval::new(predicate, meta).into_bool_fn(),
    ))
}

/// Compile a `LanguageExpressionDef` into a fallible [`SortKeySource`].
///
/// The async arm evaluates against a synthetic per-element exchange (the
/// element value as JSON body); non-scalar keys are rejected by
/// `SortKeySource::key`.
pub(crate) fn compile_sort_expression(
    languages: &SharedLanguageRegistry,
    expression: &LanguageExpressionDef,
    meta: EvalMeta,
) -> Result<SortKeySource, CamelError> {
    let value_fn =
        LanguageExpressionEval::new(compile_language_expression(languages, expression)?, meta)
            .into_value_fn();
    Ok(SortKeySource::Async(Arc::new(
        move |value: &serde_json::Value| {
            let msg = camel_api::Message::new(camel_api::body::Body::Json(value.clone()));
            let exchange = Exchange::new(msg);
            value_fn(&exchange)
        },
    )))
}

/// Compile a `LanguageExpressionDef` into a fallible [`ValueSource`] bound to
/// the [`EvalMeta`] for `verb`/`step_index`/`target`.
///
/// Evaluation failures surface as `CamelError::ExpressionFailed` carrying that
/// metadata instead of silently dropping the exchange. Shared by the
/// declarative value-setting verbs (set_header, set_property, set_body, log,
/// script fallback).
pub(crate) fn compile_value_source(
    ctx: &crate::lifecycle::adapters::step_compilers::CompilationContext<'_>,
    expression: &LanguageExpressionDef,
    verb: &str,
    step_index: usize,
    target: Option<String>,
) -> Result<ValueSource, CamelError> {
    let meta = eval_meta(ctx, &expression.language, verb, step_index, target);
    Ok(ValueSource::Async(
        LanguageExpressionEval::new(
            compile_language_expression(ctx.languages, expression)?,
            meta,
        )
        .into_value_fn(),
    ))
}

/// Compile a `LanguageExpressionDef` into a fallible [`MessageIdSource`] for
/// the Idempotent Consumer and Cache EIPs. Null/empty/non-scalar coercion is
/// handled by `MessageIdSource::message_id`.
pub(crate) fn compile_message_id_expression(
    languages: &SharedLanguageRegistry,
    expression: &LanguageExpressionDef,
    meta: EvalMeta,
) -> Result<MessageIdSource, CamelError> {
    Ok(MessageIdSource::Async(
        LanguageExpressionEval::new(compile_language_expression(languages, expression)?, meta)
            .into_value_fn(),
    ))
}

/// Compile a `LanguageExpressionDef` into a fallible [`ClaimKeySource`] for
/// the Claim Check EIP. Null/empty keys fail as `ValidationError` and
/// non-scalar keys are rejected by `ClaimKeySource::key`.
pub(crate) fn compile_key_expression(
    languages: &SharedLanguageRegistry,
    expression: &LanguageExpressionDef,
    meta: EvalMeta,
) -> Result<ClaimKeySource, CamelError> {
    Ok(ClaimKeySource::Async(
        LanguageExpressionEval::new(compile_language_expression(languages, expression)?, meta)
            .into_value_fn(),
    ))
}

pub(crate) fn resolve_enrichment_strategy(
    name: Option<String>,
) -> Result<Arc<dyn EnrichmentStrategy>, CamelError> {
    match name.as_deref() {
        None | Some("useEnrichedBody") => Ok(Arc::new(UseEnrichedBody)),
        Some("throwOnNoPoll") => Ok(Arc::new(ThrowOnNoPoll::new(Arc::new(UseEnrichedBody)))),
        Some(other) => Err(CamelError::ProcessorError(format!(
            "unknown EnrichmentStrategy `{}`; supported: useEnrichedBody (default), throwOnNoPoll",
            other
        ))),
    }
}

/// Thin dispatcher: builds the step compiler registry, creates a compilation context,
/// and compiles all steps.
#[allow(clippy::too_many_arguments)]
pub(crate) fn resolve_steps(
    steps: Vec<BuilderStep>,
    producer_ctx: &ProducerContext,
    rt: Arc<dyn RuntimeObservability>,
    _registry: &Arc<std::sync::Mutex<Registry>>,
    languages: &SharedLanguageRegistry,
    beans: &Arc<std::sync::Mutex<BeanRegistry>>,
    function_invoker: Option<Arc<dyn FunctionInvoker>>,
    component_ctx: Arc<dyn ComponentContext>,
    route_id: Option<&str>,
    staging_mode: &FunctionStagingMode,
    idempotent_repositories: &IdempotentRegistry,
    claim_check_repositories: &ClaimCheckRegistry,
    cache_repositories: &CacheRegistry,
    intercept: crate::intercept::InterceptRules,
) -> Result<Vec<CompiledStep>, CamelError> {
    use crate::lifecycle::adapters::step_compilers::{CompilationContext, build_registry};

    let compiler_registry = build_registry();
    let ctx = CompilationContext {
        producer_ctx,
        rt,
        languages,
        beans,
        function_invoker,
        component_ctx,
        route_id,
        staging_mode,
        idempotent_repositories,
        claim_check_repositories,
        cache_repositories,
        intercept,
    };
    compiler_registry.compile_steps(steps, &ctx)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::lifecycle::application::route_definition::{LanguageExpressionDef, ValueSourceDef};
    use camel_api::BoxProcessor;
    use camel_api::IdentityProcessor;
    use camel_api::OpaqueProcessor;
    use camel_api::Value;
    use camel_api::body::Body;

    /// A mock endpoint that returns `None` for polling_consumer (default).
    struct MockEndpoint {
        uri: String,
    }

    impl camel_component_api::endpoint::Endpoint for MockEndpoint {
        fn uri(&self) -> &str {
            &self.uri
        }
        fn create_consumer(
            &self,
            _rt: Arc<dyn RuntimeObservability>,
        ) -> Result<Box<dyn camel_component_api::consumer::Consumer>, CamelError> {
            Err(CamelError::EndpointCreationFailed(
                "mock not a consumer".into(),
            ))
        }
        fn create_producer(
            &self,
            _rt: Arc<dyn RuntimeObservability>,
            _ctx: &camel_component_api::ProducerContext,
        ) -> Result<BoxProcessor, CamelError> {
            Err(CamelError::ProcessorError("mock not a producer".into()))
        }
    }

    /// A mock component that vends MockEndpoint.
    struct MockComponent;

    #[async_trait::async_trait]
    impl camel_component_api::Component for MockComponent {
        fn scheme(&self) -> &str {
            "mock"
        }
        fn create_endpoint(
            &self,
            uri: &str,
            _ctx: &dyn camel_component_api::ComponentContext,
        ) -> Result<Box<dyn camel_component_api::endpoint::Endpoint>, CamelError> {
            Ok(Box::new(MockEndpoint {
                uri: uri.to_string(),
            }))
        }
    }

    /// Minimal ComponentContext that resolves exactly one scheme to a mock component.
    struct TestComponentContext;

    impl camel_component_api::ComponentContext for TestComponentContext {
        fn resolve_component(
            &self,
            scheme: &str,
        ) -> Option<Arc<dyn camel_component_api::Component>> {
            if scheme == "mock" {
                Some(Arc::new(MockComponent))
            } else {
                None
            }
        }
        fn resolve_language(&self, _name: &str) -> Option<Arc<dyn camel_language_api::Language>> {
            None
        }
        fn metrics(&self) -> Arc<dyn camel_api::MetricsCollector> {
            Arc::new(camel_api::NoOpMetrics)
        }
        fn platform_service(&self) -> Arc<dyn camel_api::PlatformService> {
            Arc::new(camel_api::NoopPlatformService::default())
        }
        fn register_route_health_check(
            &self,
            _route_id: &str,
            _check: Arc<dyn camel_api::AsyncHealthCheck>,
        ) {
        }
        fn unregister_route_health_check(&self, _route_id: &str) {}
    }

    async fn languages_with_simple() -> SharedLanguageRegistry {
        let mut map: std::collections::HashMap<String, Arc<dyn Language>> =
            std::collections::HashMap::new();
        map.insert(
            "simple".to_string(),
            Arc::new(camel_language_simple::SimpleLanguage::new()),
        );
        map.insert("testmut".to_string(), Arc::new(super::MutatingTestLanguage));
        Arc::new(std::sync::Mutex::new(map))
    }

    #[tokio::test]
    async fn resolve_language_returns_error_for_unregistered_name() {
        let languages = Arc::new(std::sync::Mutex::new(std::collections::HashMap::new()));
        let err = match resolve_language(&languages, "missing") {
            Ok(_) => panic!("resolve_language should fail for unregistered language"),
            Err(err) => err,
        };
        assert!(err.to_string().contains("missing"));
    }

    #[tokio::test]
    async fn compile_language_expression_and_predicate_work_for_simple_language() {
        let languages = languages_with_simple().await;
        let expression = LanguageExpressionDef {
            language: "simple".into(),
            source: "${header.answer}".into(),
        };
        let predicate_expression = LanguageExpressionDef {
            language: "simple".into(),
            source: "${header.answer} == '42'".into(),
        };

        let compiled_expression = compile_language_expression(&languages, &expression).unwrap();
        let compiled_predicate =
            compile_language_predicate(&languages, &predicate_expression).unwrap();

        let mut msg = camel_api::message::Message::default();
        msg.set_header("answer", Value::String("42".into()));
        let exchange = Exchange::new(msg);

        assert_eq!(
            compiled_expression.evaluate(&exchange).await.unwrap(),
            Value::String("42".into())
        );
        assert!(compiled_predicate.matches(&exchange).await.unwrap());
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn compile_filter_predicate_returns_boolean_result() {
        let languages = languages_with_simple().await;
        let expression = LanguageExpressionDef {
            language: "simple".into(),
            source: "${header.flag} == 'yes'".into(),
        };
        let meta = EvalMeta {
            language: "simple".into(),
            route_id: "r1".into(),
            step_id: "filter#0".into(),
            verb: "filter".into(),
            target: None,
        };
        let predicate = compile_filter_predicate(&languages, &expression, meta).unwrap();

        let mut msg = camel_api::message::Message::default();
        msg.set_header("flag", Value::String("yes".into()));
        let exchange = Exchange::new(msg);
        assert!(predicate.matches(&exchange).await.unwrap());
    }

    fn local_value_to_body(value: Value) -> Body {
        match value {
            Value::Null => Body::Empty,
            Value::String(text) => Body::Text(text),
            other => Body::Json(other),
        }
    }

    #[tokio::test]
    async fn value_to_body_covers_null_string_and_json() {
        assert!(matches!(local_value_to_body(Value::Null), Body::Empty));
        assert!(matches!(
            local_value_to_body(Value::String("x".into())),
            Body::Text(ref s) if s == "x"
        ));
        assert!(matches!(
            local_value_to_body(Value::Number(serde_json::Number::from(7))),
            Body::Json(Value::Number(_))
        ));
    }

    #[tokio::test]
    async fn resolve_steps_validates_declarative_loop_shape() {
        let languages = languages_with_simple().await;
        let producer_ctx = ProducerContext::new();
        let registry = Arc::new(std::sync::Mutex::new(Registry::new()));
        let beans = Arc::new(std::sync::Mutex::new(BeanRegistry::new()));
        let component_ctx: Arc<dyn ComponentContext> =
            Arc::new(camel_component_api::NoOpComponentContext);
        let rt: Arc<dyn RuntimeObservability> = Arc::new(camel_component_api::NoOpComponentContext);

        let both = resolve_steps(
            vec![BuilderStep::DeclarativeLoop {
                count: Some(2),
                while_predicate: Some(LanguageExpressionDef {
                    language: "simple".into(),
                    source: "${header.k} == 'v'".into(),
                }),
                steps: vec![],
                max_iterations: None,
            }],
            &producer_ctx,
            Arc::clone(&rt),
            &registry,
            &languages,
            &beans,
            None,
            Arc::clone(&component_ctx),
            Some("r1"),
            &FunctionStagingMode::DirectAdd,
            &crate::IdempotentRegistry::new(),
            &crate::ClaimCheckRegistry::new(),
            &crate::CacheRegistry::new(),
            crate::intercept::InterceptRules::default(),
        )
        .unwrap_err();
        assert!(
            matches!(
                both,
                CamelError::ConfigValidation(
                    camel_api::ConfigValidationError::LoopConflictingCountAndWhile,
                )
            ),
            "expected ConfigValidation(LoopConflictingCountAndWhile), got: {both}"
        );

        let neither = resolve_steps(
            vec![BuilderStep::DeclarativeLoop {
                count: None,
                while_predicate: None,
                steps: vec![],
                max_iterations: None,
            }],
            &producer_ctx,
            Arc::clone(&rt),
            &registry,
            &languages,
            &beans,
            None,
            component_ctx,
            Some("r1"),
            &FunctionStagingMode::DirectAdd,
            &crate::IdempotentRegistry::new(),
            &crate::ClaimCheckRegistry::new(),
            &crate::CacheRegistry::new(),
            crate::intercept::InterceptRules::default(),
        )
        .unwrap_err();
        assert!(
            matches!(
                neither,
                CamelError::ConfigValidation(
                    camel_api::ConfigValidationError::LoopMissingCountOrWhile,
                )
            ),
            "expected ConfigValidation(LoopMissingCountOrWhile), got: {neither}"
        );
    }

    #[tokio::test]
    async fn resolve_steps_returns_component_not_found_for_to_step() {
        let languages = languages_with_simple().await;
        let producer_ctx = ProducerContext::new();
        let registry = Arc::new(std::sync::Mutex::new(Registry::new()));
        let beans = Arc::new(std::sync::Mutex::new(BeanRegistry::new()));
        let component_ctx: Arc<dyn ComponentContext> =
            Arc::new(camel_component_api::NoOpComponentContext);
        let rt: Arc<dyn RuntimeObservability> = Arc::new(camel_component_api::NoOpComponentContext);

        let err = resolve_steps(
            vec![BuilderStep::To("unknown:dest".into())],
            &producer_ctx,
            Arc::clone(&rt),
            &registry,
            &languages,
            &beans,
            None,
            component_ctx,
            Some("r1"),
            &FunctionStagingMode::DirectAdd,
            &crate::IdempotentRegistry::new(),
            &crate::ClaimCheckRegistry::new(),
            &crate::CacheRegistry::new(),
            crate::intercept::InterceptRules::default(),
        )
        .unwrap_err();

        assert!(err.to_string().contains("unknown"));
    }

    #[tokio::test]
    async fn compile_language_expression_and_predicate_propagate_compile_errors() {
        let languages = languages_with_simple().await;
        let bad_expr = LanguageExpressionDef {
            language: "simple".into(),
            source: "${header.a".into(),
        };
        let bad_pred = LanguageExpressionDef {
            language: "simple".into(),
            source: "${header.a == 'x'".into(),
        };

        let expr_err = match compile_language_expression(&languages, &bad_expr) {
            Ok(_) => panic!("expression compile should fail"),
            Err(err) => err,
        };
        assert!(
            expr_err
                .to_string()
                .contains("failed to compile simple expression")
        );

        let pred_err = match compile_language_predicate(&languages, &bad_pred) {
            Ok(_) => panic!("predicate compile should fail"),
            Err(err) => err,
        };
        assert!(
            pred_err
                .to_string()
                .contains("failed to compile simple predicate")
        );
    }

    #[tokio::test]
    async fn compile_language_expression_is_template() {
        let languages = languages_with_simple().await;
        let expression = LanguageExpressionDef {
            language: "simple".into(),
            source: "Got ${header.h}".into(),
        };

        let compiled = compile_language_expression(&languages, &expression).unwrap();

        let mut msg = camel_api::message::Message::default();
        msg.set_header("h", Value::String("B".into()));
        let exchange = Exchange::new(msg);

        assert_eq!(
            compiled.evaluate(&exchange).await.unwrap(),
            Value::String("Got B".into())
        );
    }

    #[tokio::test]
    async fn compile_value_source_is_template() {
        use crate::lifecycle::adapters::step_compilers::CompilationContext;

        let languages = languages_with_simple().await;
        let producer_ctx = ProducerContext::new();
        let beans = Arc::new(std::sync::Mutex::new(BeanRegistry::new()));
        let component_ctx: Arc<dyn ComponentContext> =
            Arc::new(camel_component_api::NoOpComponentContext);
        let rt: Arc<dyn RuntimeObservability> = Arc::new(camel_component_api::NoOpComponentContext);
        let staging = FunctionStagingMode::DirectAdd;
        let idempotent = crate::IdempotentRegistry::new();
        let claim_check = crate::ClaimCheckRegistry::new();
        let cache = crate::CacheRegistry::new();

        let ctx = CompilationContext {
            producer_ctx: &producer_ctx,
            rt,
            languages: &languages,
            beans: &beans,
            function_invoker: None,
            component_ctx,
            route_id: Some("r1"),
            staging_mode: &staging,
            idempotent_repositories: &idempotent,
            claim_check_repositories: &claim_check,
            cache_repositories: &cache,
            intercept: crate::intercept::InterceptRules::default(),
        };

        let expression = LanguageExpressionDef {
            language: "simple".into(),
            source: "Got ${header.h}".into(),
        };
        let value_source =
            compile_value_source(&ctx, &expression, "set_body", 0, Some("body".into()))
                .expect("value source compiles");

        let mut msg = camel_api::message::Message::default();
        msg.set_header("h", Value::String("B".into()));
        let exchange = Exchange::new(msg);

        assert_eq!(
            value_source.evaluate(&exchange).await.unwrap(),
            Value::String("Got B".into())
        );
    }

    fn eval_meta_for(verb: &str) -> EvalMeta {
        EvalMeta {
            language: "simple".into(),
            route_id: "r1".into(),
            step_id: format!("{verb}#0"),
            verb: verb.into(),
            target: None,
        }
    }

    #[tokio::test]
    async fn compile_sort_expression_is_template() {
        let languages = languages_with_simple().await;
        let expression = LanguageExpressionDef {
            language: "simple".into(),
            source: "X-${body.id}".into(),
        };

        let source = compile_sort_expression(&languages, &expression, eval_meta_for("sort"))
            .expect("sort expression compiles");

        // The synthetic sort element exchange carries NO headers; the key must
        // come from the element body.
        let key = source
            .key(&serde_json::json!({ "id": "7" }))
            .await
            .expect("sort key evaluates");
        assert_eq!(key.0, serde_json::json!("X-7"));
    }

    #[tokio::test]
    async fn compile_message_id_expression_is_template() {
        let languages = languages_with_simple().await;
        let expression = LanguageExpressionDef {
            language: "simple".into(),
            source: "m-${header.id}".into(),
        };

        let source = compile_message_id_expression(
            &languages,
            &expression,
            eval_meta_for("idempotent_consumer"),
        )
        .expect("message id expression compiles");

        let mut msg = camel_api::message::Message::default();
        msg.set_header("id", Value::String("k".into()));
        let exchange = Exchange::new(msg);

        assert_eq!(
            source.message_id(&exchange).await.unwrap(),
            Some("m-k".to_string())
        );
    }

    #[tokio::test]
    async fn compile_key_expression_is_template() {
        let languages = languages_with_simple().await;
        let expression = LanguageExpressionDef {
            language: "simple".into(),
            source: "k-${header.id}".into(),
        };

        let source = compile_key_expression(&languages, &expression, eval_meta_for("claim_check"))
            .expect("claim key expression compiles");

        let mut msg = camel_api::message::Message::default();
        msg.set_header("id", Value::String("k".into()));
        let exchange = Exchange::new(msg);

        assert_eq!(source.key(&exchange).await.unwrap(), "k-k");
    }

    #[tokio::test]
    async fn compile_language_predicate_still_compares() {
        let languages = languages_with_simple().await;
        let expression = LanguageExpressionDef {
            language: "simple".into(),
            source: "${header.h} == 'B'".into(),
        };

        let predicate = compile_language_predicate(&languages, &expression).unwrap();

        let mut msg = camel_api::message::Message::default();
        msg.set_header("h", Value::String("B".into()));
        let exchange = Exchange::new(msg);

        assert!(predicate.matches(&exchange).await.unwrap());
    }

    #[tokio::test]
    async fn compile_language_expression_propagates_template_errors() {
        let languages = languages_with_simple().await;
        let bad = LanguageExpressionDef {
            language: "simple".into(),
            source: "${header.}".into(),
        };

        assert!(
            compile_language_expression(&languages, &bad).is_err(),
            "an empty header key must fail at compile time"
        );
    }

    #[tokio::test]
    async fn resolve_steps_covers_non_endpoint_variants() {
        use camel_api::FilterPredicate;
        use camel_api::splitter::{AggregationStrategy, SplitterConfig, split_body_lines};
        use std::time::Duration;

        let expr = |source: &str| LanguageExpressionDef {
            language: "simple".into(),
            source: source.into(),
        };

        let languages = languages_with_simple().await;
        let producer_ctx = ProducerContext::new();
        let registry = Arc::new(std::sync::Mutex::new(Registry::new()));
        let beans = Arc::new(std::sync::Mutex::new(BeanRegistry::new()));
        let component_ctx: Arc<dyn ComponentContext> =
            Arc::new(camel_component_api::NoOpComponentContext);
        let rt: Arc<dyn RuntimeObservability> = Arc::new(camel_component_api::NoOpComponentContext);

        let steps = vec![
            BuilderStep::Processor(OpaqueProcessor(BoxProcessor::new(IdentityProcessor))),
            BuilderStep::Stop,
            BuilderStep::Delay {
                config: camel_api::DelayConfig::new(1),
            },
            BuilderStep::Loop {
                config: camel_api::loop_eip::LoopConfig::new(camel_api::loop_eip::LoopMode::Count(
                    1,
                )),
                steps: vec![BuilderStep::Stop],
            },
            BuilderStep::DeclarativeLoop {
                count: Some(1),
                while_predicate: None,
                steps: vec![BuilderStep::Stop],
                max_iterations: None,
            },
            BuilderStep::Log {
                level: camel_processor::LogLevel::Info,
                message: "log".into(),
            },
            BuilderStep::DeclarativeSetHeader {
                key: "k".into(),
                value: ValueSourceDef::Literal(Value::String("v".into())),
            },
            BuilderStep::DeclarativeSetProperty {
                key: "p".into(),
                value_source: ValueSourceDef::Expression(expr("${header.k}")),
            },
            BuilderStep::DeclarativeSetBody {
                value: ValueSourceDef::Expression(expr("${header.k}")),
            },
            BuilderStep::DeclarativeFilter {
                predicate: expr("${header.k} == 'v'"),
                steps: vec![BuilderStep::Stop],
            },
            BuilderStep::DeclarativeChoice {
                whens: vec![
                    crate::lifecycle::application::route_definition::DeclarativeWhenStep {
                        predicate: expr("${header.k} == 'v'"),
                        steps: vec![BuilderStep::Stop],
                    },
                ],
                otherwise: Some(vec![BuilderStep::Stop]),
            },
            BuilderStep::DeclarativeScript {
                expression: LanguageExpressionDef {
                    language: "testmut".into(),
                    source: "${header.k}".into(),
                },
            },
            BuilderStep::Split {
                config: SplitterConfig::new(split_body_lines())
                    .aggregation(AggregationStrategy::CollectAll),
                steps: vec![BuilderStep::Stop],
            },
            BuilderStep::DeclarativeSplit {
                expression: expr("${body}"),
                aggregation: AggregationStrategy::Original,
                parallel: false,
                parallel_limit: Some(2),
                trace_item_threshold: None,
                stop_on_exception: true,
                steps: vec![BuilderStep::Stop],
            },
            BuilderStep::Aggregate {
                config: camel_api::AggregatorConfig::correlate_by("id")
                    .complete_when_size(1)
                    .build()
                    .unwrap(),
            },
            BuilderStep::Filter {
                predicate: FilterPredicate::new(|_| true),
                steps: vec![BuilderStep::Stop],
            },
            BuilderStep::Choice {
                whens: vec![crate::lifecycle::application::route_definition::WhenStep {
                    predicate: FilterPredicate::new(|_| true),
                    steps: vec![BuilderStep::Stop],
                }],
                otherwise: Some(vec![BuilderStep::Stop]),
            },
            BuilderStep::Multicast {
                steps: vec![BuilderStep::Stop, BuilderStep::Stop],
                config: camel_api::MulticastConfig::new(),
            },
            BuilderStep::DeclarativeLog {
                level: camel_processor::LogLevel::Info,
                message: ValueSourceDef::Expression(expr("${header.k}")),
            },
            BuilderStep::Throttle {
                config: camel_api::ThrottlerConfig::new(10, Duration::from_millis(10)),
                steps: vec![BuilderStep::Stop],
            },
            BuilderStep::LoadBalance {
                config: camel_api::LoadBalancerConfig::round_robin(),
                steps: vec![BuilderStep::Stop, BuilderStep::Stop],
            },
            BuilderStep::DynamicRouter {
                config: camel_api::DynamicRouterConfig::new(camel_api::TargetSource::Sync(
                    Arc::new(|_| None),
                )),
            },
            BuilderStep::DeclarativeDynamicRouter {
                expression: expr("${header.routes}"),
                uri_delimiter: ",".into(),
                cache_size: 8,
                ignore_invalid_endpoints: true,
                max_iterations: 3,
            },
            BuilderStep::RoutingSlip {
                config: camel_api::RoutingSlipConfig::new(camel_api::TargetSource::Sync(Arc::new(
                    |_| None,
                ))),
            },
            BuilderStep::DeclarativeRoutingSlip {
                expression: expr("${header.routes}"),
                uri_delimiter: ";".into(),
                cache_size: 16,
                ignore_invalid_endpoints: true,
            },
            BuilderStep::RecipientList {
                config: camel_api::recipient_list::RecipientListConfig::new(
                    camel_api::RecipientSource::Sync(Arc::new(|_| String::new())),
                ),
            },
            BuilderStep::DeclarativeRecipientList {
                expression: expr("${header.routes}"),
                delimiter: ",".into(),
                parallel: true,
                parallel_limit: Some(2),
                stop_on_exception: false,
                aggregation: "noop".into(),
            },
        ];

        let resolved = resolve_steps(
            steps,
            &producer_ctx,
            Arc::clone(&rt),
            &registry,
            &languages,
            &beans,
            None,
            component_ctx,
            Some("r1"),
            &FunctionStagingMode::DirectAdd,
            &crate::IdempotentRegistry::new(),
            &crate::ClaimCheckRegistry::new(),
            &crate::CacheRegistry::new(),
            crate::intercept::InterceptRules::default(),
        )
        .unwrap();

        assert!(!resolved.is_empty());
    }

    #[tokio::test]
    async fn poll_enrich_on_non_pollable_endpoint_returns_compile_error() {
        let languages = Arc::new(std::sync::Mutex::new(std::collections::HashMap::new()));
        let producer_ctx = ProducerContext::new();
        let registry = Arc::new(std::sync::Mutex::new(Registry::new()));
        let beans = Arc::new(std::sync::Mutex::new(BeanRegistry::new()));
        let component_ctx: Arc<dyn ComponentContext> = Arc::new(TestComponentContext);
        let rt: Arc<dyn RuntimeObservability> = Arc::new(camel_component_api::NoOpComponentContext);

        let err = resolve_steps(
            vec![BuilderStep::PollEnrich {
                uri: "mock:data".into(),
                strategy: None,
                timeout_ms: None,
            }],
            &producer_ctx,
            Arc::clone(&rt),
            &registry,
            &languages,
            &beans,
            None,
            component_ctx,
            Some("r1"),
            &FunctionStagingMode::DirectAdd,
            &crate::IdempotentRegistry::new(),
            &crate::ClaimCheckRegistry::new(),
            &crate::CacheRegistry::new(),
            crate::intercept::InterceptRules::default(),
        )
        .unwrap_err();

        let err_msg = err.to_string();
        assert!(
            err_msg.contains("pollEnrich requires"),
            "expected error about PollingConsumer, got: {err_msg}"
        );
        assert!(
            err_msg.contains("exposes a PollingConsumer"),
            "expected error about missing PollingConsumer, got: {err_msg}"
        );
    }

    #[tokio::test]
    async fn resolve_enrichment_strategy_none_returns_default_passthrough() {
        let strategy = resolve_enrichment_strategy(None).unwrap();
        let exchange = Exchange::new(camel_api::Message::new("test"));
        let result = strategy.on_no_poll(exchange).await.unwrap();
        match &result.input.body {
            Body::Text(s) => assert_eq!(s, "test"),
            other => panic!("expected Text body, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn resolve_enrichment_strategy_use_enriched_body_returns_default_passthrough() {
        let strategy = resolve_enrichment_strategy(Some("useEnrichedBody".into())).unwrap();
        let exchange = Exchange::new(camel_api::Message::new("test"));
        let result = strategy.on_no_poll(exchange).await.unwrap();
        match &result.input.body {
            Body::Text(s) => assert_eq!(s, "test"),
            other => panic!("expected Text body, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn resolve_enrichment_strategy_throw_on_no_poll_returns_error() {
        let strategy = resolve_enrichment_strategy(Some("throwOnNoPoll".into())).unwrap();
        let exchange = Exchange::new(camel_api::Message::new("test"));
        let err = strategy.on_no_poll(exchange).await.unwrap_err();
        assert!(
            err.to_string().contains("no message available"),
            "throwOnNoPoll should return error on no poll, got: {err}"
        );
    }

    #[test]
    fn resolve_enrichment_strategy_unknown_name_returns_error() {
        let result = resolve_enrichment_strategy(Some("bogus".into()));
        match result {
            Ok(_) => panic!("expected error for unknown strategy name"),
            Err(err) => assert!(
                err.to_string().contains("unknown EnrichmentStrategy"),
                "unexpected error message: {err}"
            ),
        }
    }
}