camel-processor 0.54.0

//! ## Stop semantics (ADR-0025)
//!
//! This segment implements `OutcomePipeline` and propagates `PipelineOutcome::Stopped(ex)`
//! with the exchange state intact. See ADR-0025 §3.

use crate::do_try::CatchMatcher;
use crate::error_handler::record_span_error;
use camel_api::error_handler::ExceptionDisposition;
use camel_api::outcome_pipeline::OutcomePipeline;
use camel_api::pipeline_outcome::PipelineOutcome;
use camel_api::{CamelError, Exchange, FilterPredicate};
use std::future::Future;
use std::pin::Pin;

/// - `Handled`: the exchange continues through finally and downstream.
/// - `Propagate`: the catch body runs for side effects, finally runs with the
///   catch body's exchange, and the original try-error re-throws as `Failed`.
#[derive(Clone)]
pub struct CatchClauseSegment {
    pub matcher: CatchMatcher,
    pub on_when: Option<FilterPredicate>,
    pub body: camel_api::OutcomeSegment,
    pub disposition: ExceptionDisposition,
}

/// Compilable segment for a `doFinally` clause within a `DoTrySegment`.
#[derive(Clone)]
pub struct FinallyClauseSegment {
    pub on_when: Option<FilterPredicate>,
    pub body: camel_api::OutcomeSegment,
}

/// Outcome-aware structural EIP segment for the doTry/doCatch/doFinally pattern.
///
/// Operates at the `PipelineOutcome` layer so that `Stopped(ex)` from a
/// sub-step (e.g. Stop EIP) is preserved with the exchange including all
/// mutations. See ADR-0025 §5.1 for full semantics.
pub struct DoTrySegment {
    pub try_body: camel_api::OutcomeSegment,
    pub catches: Vec<CatchClauseSegment>,
    pub finally: Option<FinallyClauseSegment>,
}

impl Clone for DoTrySegment {
    fn clone(&self) -> Self {
        Self {
            try_body: self.try_body.clone(),
            catches: self.catches.clone(),
            finally: self.finally.clone(),
        }
    }
}

/// Narrow outcome type for `run_finally_body` so the compiler enforces
/// exhaustiveness at the two call sites instead of `unreachable!()`.
/// This is a private, transient type; the `Stopped` exchange is boxed to keep
/// the `Err` variant of `run_finally_body` small (clippy::result_large_err).
enum FinallyOutcome {
    Stopped(Box<Exchange>),
    Failed(CamelError),
}

/// Run the finally body if present and on_when permits.
/// Free function (not a method) to avoid borrow conflict with
/// `self.catches.iter_mut()` inside the catch loop.
/// The `FinallyOutcome` variants move the `Exchange` (the catch loop needs
/// its state); boxing would rewrite the segment call chain for a lint
/// threshold.
#[allow(clippy::result_large_err)]
async fn run_finally_body(
    finally: &mut Option<FinallyClauseSegment>,
    ex: Exchange,
) -> Result<Exchange, FinallyOutcome> {
    let Some(f) = finally.as_mut() else {
        return Ok(ex);
    };
    if !f.on_when.as_ref().map(|p| p(&ex)).unwrap_or(true) {
        return Ok(ex);
    }
    match f.body.run(ex).await {
        PipelineOutcome::Completed(e) => Ok(e),
        PipelineOutcome::Stopped(e) => Err(FinallyOutcome::Stopped(Box::new(e))),
        PipelineOutcome::Failed(e) => Err(FinallyOutcome::Failed(e)),
    }
}

impl OutcomePipeline for DoTrySegment {
    fn clone_box(&self) -> Box<dyn OutcomePipeline> {
        Box::new(self.clone())
    }

    fn run<'a>(
        &'a mut self,
        exchange: Exchange,
    ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
        Box::pin(async move {
            // Pre-clone for the "no catch matched" path; `exchange` itself is
            // consumed by `try_body.run(...)` below.
            let exchange_for_unmatched = exchange.clone();

            // 1. Run try_body. If Stopped, return immediately (skip catch AND finally).
            let try_outcome = self.try_body.run(exchange).await;
            let returned_ex = match try_outcome {
                PipelineOutcome::Stopped(ex) => return PipelineOutcome::Stopped(ex),
                PipelineOutcome::Completed(ex) => ex,
                PipelineOutcome::Failed(err) => {
                    // 2. try failed — try each catch in order. Walk the catch
                    // chain starting from the failed try-exchange. On the first
                    // matching catch whose `on_when` is true (or unset), run it.
                    let mut current_ex = exchange_for_unmatched;
                    current_ex.set_error(err.clone());
                    for catch in self.catches.iter_mut() {
                        if catch.matcher.matches(&err, &current_ex)
                            && catch
                                .on_when
                                .as_ref()
                                .map(|p| p(&current_ex))
                                .unwrap_or(true)
                        {
                            match catch.body.run(current_ex).await {
                                // Stop in catch: skip finally AND outer route halts.
                                PipelineOutcome::Stopped(stopped_ex) => {
                                    return PipelineOutcome::Stopped(stopped_ex);
                                }
                                // Catch handled — exit catch chain with this exchange.
                                PipelineOutcome::Completed(next) => {
                                    match catch.disposition {
                                        ExceptionDisposition::Handled => {
                                            current_ex = next;
                                            break;
                                        }
                                        ExceptionDisposition::Continued => {
                                            current_ex = next;
                                            break;
                                        }
                                        // Propagate and any future variant run finally then
                                        // surface the original error (fail-closed).
                                        _ => {
                                            match run_finally_body(&mut self.finally, next).await {
                                                Ok(_) => {}
                                                Err(FinallyOutcome::Stopped(e)) => {
                                                    return PipelineOutcome::Stopped(*e);
                                                }
                                                Err(FinallyOutcome::Failed(_finally_err)) => {
                                                    tracing::warn!(
                                                        error = %err,
                                                        "doFinally threw during Propagate; \
                                                         restoring original"
                                                    );
                                                    return PipelineOutcome::Failed(err);
                                                }
                                            }
                                            return PipelineOutcome::Failed(err);
                                        }
                                    }
                                }
                                // Catch-body Failed: surface THAT error to outer
                                // route. Per ADR-0025 invariant #4 ("doTry is a
                                // local error-handler island"), a failing catch
                                // body propagates as Failed — it does NOT re-enter
                                // the catch chain (no recursive catch-of-catch in
                                // Camel). Skip remaining catches and finally.
                                PipelineOutcome::Failed(catch_err) => {
                                    // Sealed failure envelope: the catch error
                                    // stays the main error (returned below);
                                    // the original try error is surfaced
                                    // through the unconditional warn record
                                    // and, when a span is active, the span
                                    // error record. No new span is created.
                                    tracing::warn!(
                                        original_error = %err,
                                        catch_error = %catch_err,
                                        "do_try catch block failed; catch error supersedes original"
                                    );
                                    record_span_error(&catch_err);
                                    return PipelineOutcome::Failed(catch_err);
                                }
                            }
                        }
                    }
                    current_ex
                }
            };
            // 3. Run finally if present (skip if try/catch returned Stopped —
            // already returned above; skip if catch-body Failed — also returned).
            match run_finally_body(&mut self.finally, returned_ex).await {
                Ok(ex) => PipelineOutcome::Completed(ex),
                Err(FinallyOutcome::Stopped(e)) => PipelineOutcome::Stopped(*e),
                Err(FinallyOutcome::Failed(finally_err)) => {
                    tracing::warn!(
                        error = %finally_err,
                        "doFinally threw during/after catch; surfacing finally error"
                    );
                    PipelineOutcome::Failed(finally_err)
                }
            }
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::do_try::CatchMatcher;
    use crate::test_log_capture::{capture_debugs_with_span_records, captured_field, record_field};
    use camel_api::pipeline_outcome::PipelineOutcome;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicU32, Ordering};

    // ── Helpers for DoTrySegment tests ──

    struct CompleteSegment;

    impl OutcomePipeline for CompleteSegment {
        fn clone_box(&self) -> Box<dyn OutcomePipeline> {
            Box::new(CompleteSegment)
        }
        fn run<'a>(
            &'a mut self,
            exchange: Exchange,
        ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
            Box::pin(async move { PipelineOutcome::Completed(exchange) })
        }
    }

    fn seg_complete() -> camel_api::OutcomeSegment {
        camel_api::OutcomeSegment::new(Box::new(CompleteSegment))
    }

    struct FailSegment(CamelError);

    impl OutcomePipeline for FailSegment {
        fn clone_box(&self) -> Box<dyn OutcomePipeline> {
            Box::new(FailSegment(self.0.clone()))
        }
        fn run<'a>(
            &'a mut self,
            _exchange: Exchange,
        ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
            let e = self.0.clone();
            Box::pin(async move { PipelineOutcome::Failed(e) })
        }
    }

    fn seg_fail(err: CamelError) -> camel_api::OutcomeSegment {
        camel_api::OutcomeSegment::new(Box::new(FailSegment(err)))
    }

    struct MutateThenStop {
        mutator: Arc<dyn Fn(&mut Exchange) + Send + Sync>,
    }

    impl OutcomePipeline for MutateThenStop {
        fn clone_box(&self) -> Box<dyn OutcomePipeline> {
            Box::new(MutateThenStop {
                mutator: Arc::clone(&self.mutator),
            })
        }
        fn run<'a>(
            &'a mut self,
            mut exchange: Exchange,
        ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
            let m = Arc::clone(&self.mutator);
            Box::pin(async move {
                m(&mut exchange);
                PipelineOutcome::Stopped(exchange)
            })
        }
    }

    fn seg_stop_with(
        mutator: impl Fn(&mut Exchange) + Send + Sync + 'static,
    ) -> camel_api::OutcomeSegment {
        camel_api::OutcomeSegment::new(Box::new(MutateThenStop {
            mutator: Arc::new(mutator),
        }))
    }

    struct RecordCall {
        counter: Arc<AtomicU32>,
    }

    impl OutcomePipeline for RecordCall {
        fn clone_box(&self) -> Box<dyn OutcomePipeline> {
            Box::new(RecordCall {
                counter: Arc::clone(&self.counter),
            })
        }
        fn run<'a>(
            &'a mut self,
            exchange: Exchange,
        ) -> Pin<Box<dyn Future<Output = PipelineOutcome> + Send + 'a>> {
            let c = Arc::clone(&self.counter);
            Box::pin(async move {
                c.fetch_add(1, Ordering::SeqCst);
                PipelineOutcome::Completed(exchange)
            })
        }
    }

    fn seg_record(counter: Arc<AtomicU32>) -> camel_api::OutcomeSegment {
        camel_api::OutcomeSegment::new(Box::new(RecordCall { counter }))
    }

    // ── 5 New tests for DoTrySegment (ADR-0025) ──

    #[tokio::test]
    async fn stop_inside_try_skips_catch_and_finally() {
        let catch_call = Arc::new(AtomicU32::new(0));
        let finally_call = Arc::new(AtomicU32::new(0));

        let mut seg = DoTrySegment {
            try_body: seg_stop_with(|ex| {
                ex.set_property("mutated", camel_api::Value::Bool(true));
            }),
            catches: vec![CatchClauseSegment {
                matcher: CatchMatcher::ByVariant(vec!["*".into()]),
                on_when: None,
                body: seg_record(catch_call.clone()),
                disposition: ExceptionDisposition::Handled,
            }],
            finally: Some(FinallyClauseSegment {
                on_when: None,
                body: seg_record(finally_call.clone()),
            }),
        };

        let result = seg.run(Exchange::default()).await;
        match result {
            PipelineOutcome::Stopped(ex) => {
                assert_eq!(
                    ex.properties.get("mutated"),
                    Some(&camel_api::Value::Bool(true)),
                    "try body mutation must be preserved in Stopped exchange"
                );
            }
            other => panic!("expected Stopped, got {:?}", other),
        }
        assert_eq!(
            catch_call.load(Ordering::SeqCst),
            0,
            "catch must NOT run when try stops"
        );
        assert_eq!(
            finally_call.load(Ordering::SeqCst),
            0,
            "finally must NOT run when try stops"
        );
    }

    #[tokio::test]
    async fn stop_inside_catch_skips_finally() {
        let finally_call = Arc::new(AtomicU32::new(0));

        let mut seg = DoTrySegment {
            try_body: seg_fail(CamelError::ProcessorError("boom".into())),
            catches: vec![CatchClauseSegment {
                matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
                on_when: None,
                body: seg_stop_with(|ex| {
                    ex.set_property("catch_mutated", camel_api::Value::Bool(true));
                }),
                disposition: ExceptionDisposition::Handled,
            }],
            finally: Some(FinallyClauseSegment {
                on_when: None,
                body: seg_record(finally_call.clone()),
            }),
        };

        let result = seg.run(Exchange::default()).await;
        match result {
            PipelineOutcome::Stopped(ex) => {
                assert_eq!(
                    ex.properties.get("catch_mutated"),
                    Some(&camel_api::Value::Bool(true)),
                    "catch body mutation must be preserved in Stopped exchange"
                );
            }
            other => panic!("expected Stopped, got {:?}", other),
        }
        assert_eq!(
            finally_call.load(Ordering::SeqCst),
            0,
            "finally must NOT run when catch stops"
        );
    }

    #[tokio::test]
    async fn stop_inside_finally_stops_outer_route() {
        let mut seg = DoTrySegment {
            try_body: seg_complete(),
            catches: vec![],
            finally: Some(FinallyClauseSegment {
                on_when: None,
                body: seg_stop_with(|ex| {
                    ex.set_property("finally_mutated", camel_api::Value::Bool(true));
                }),
            }),
        };

        let result = seg.run(Exchange::default()).await;
        match result {
            PipelineOutcome::Stopped(ex) => {
                assert_eq!(
                    ex.properties.get("finally_mutated"),
                    Some(&camel_api::Value::Bool(true)),
                    "finally body mutation must be preserved in Stopped exchange"
                );
            }
            other => panic!("expected Stopped, got {:?}", other),
        }
    }

    #[tokio::test]
    async fn catch_on_when_false_falls_through_to_next_catch() {
        let first_call = Arc::new(AtomicU32::new(0));
        let second_call = Arc::new(AtomicU32::new(0));

        let mut seg = DoTrySegment {
            try_body: seg_fail(CamelError::Io("disk err".into())),
            catches: vec![
                CatchClauseSegment {
                    matcher: CatchMatcher::ByVariant(vec!["Io".into()]),
                    on_when: Some(FilterPredicate::new(|_ex| false)),
                    body: seg_record(first_call.clone()),
                    disposition: ExceptionDisposition::Handled,
                },
                CatchClauseSegment {
                    matcher: CatchMatcher::ByVariant(vec!["*".into()]),
                    on_when: None,
                    body: seg_record(second_call.clone()),
                    disposition: ExceptionDisposition::Handled,
                },
            ],
            finally: None,
        };

        let result = seg.run(Exchange::default()).await;
        assert!(
            matches!(result, PipelineOutcome::Completed(_)),
            "expected Completed after second catch"
        );
        assert_eq!(
            first_call.load(Ordering::SeqCst),
            0,
            "first catch must NOT fire (on_when=false)"
        );
        assert_eq!(
            second_call.load(Ordering::SeqCst),
            1,
            "second catch must fire"
        );
    }

    #[tokio::test]
    async fn finally_on_when_false_skips_finally_entirely() {
        let finally_call = Arc::new(AtomicU32::new(0));
        let mut ex = Exchange::default();
        ex.set_property("try_set", camel_api::Value::Bool(true));

        let mut seg = DoTrySegment {
            try_body: seg_complete(),
            catches: vec![],
            finally: Some(FinallyClauseSegment {
                on_when: Some(FilterPredicate::new(|_ex| false)),
                body: seg_record(finally_call.clone()),
            }),
        };

        let result = seg.run(ex).await;
        match result {
            PipelineOutcome::Completed(ex) => {
                assert_eq!(
                    ex.properties.get("try_set"),
                    Some(&camel_api::Value::Bool(true)),
                    "exchange state from try must be preserved"
                );
            }
            other => panic!("expected Completed, got {:?}", other),
        }
        assert_eq!(
            finally_call.load(Ordering::SeqCst),
            0,
            "finally must NOT run when on_when=false"
        );
    }

    // ── Catch-body failure envelope (compiled-segment path) ──

    fn catch_fails_segment() -> DoTrySegment {
        DoTrySegment {
            try_body: seg_fail(CamelError::ProcessorError("orig".into())),
            catches: vec![CatchClauseSegment {
                matcher: CatchMatcher::ByVariant(vec!["ProcessorError".into()]),
                on_when: None,
                body: seg_fail(CamelError::Io("catch-fail".into())),
                disposition: ExceptionDisposition::Handled,
            }],
            finally: None,
        }
    }

    #[test]
    fn catch_body_failure_returns_catch_err_and_marks_span() {
        let mut seg = catch_fails_segment();
        let (result, _captured, span_records) = capture_debugs_with_span_records(|| {
            // Declared `error` field so record_span_error's record lands.
            let span = tracing::info_span!("dotry_seg_test", error = tracing::field::Empty);
            let _guard = span.enter();
            tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .expect("current-thread runtime")
                .block_on(async { seg.run(Exchange::default()).await })
        });

        assert!(
            matches!(result, PipelineOutcome::Failed(CamelError::Io(_))),
            "catch failure must surface the catch error as Failed(Io), got: {result:?}"
        );
        // The span recorded the `error` field with the CATCH error.
        // Structured field lookup: substring `error=` would also match
        // `original_error=` suffixes.
        assert!(
            span_records.iter().any(|line| {
                captured_field(line, "error").is_some_and(|v| v.contains("catch-fail"))
            }),
            "expected span error record carrying the catch error, span records: {span_records:?}"
        );
    }

    #[test]
    fn catch_body_failure_emits_envelope_log() {
        let mut seg = catch_fails_segment();
        let (result, captured, _span_records) = capture_debugs_with_span_records(|| {
            tokio::runtime::Builder::new_current_thread()
                .enable_all()
                .build()
                .expect("current-thread runtime")
                .block_on(async { seg.run(Exchange::default()).await })
        });

        assert!(
            matches!(result, PipelineOutcome::Failed(CamelError::Io(_))),
            "catch failure must surface the catch error as Failed(Io), got: {result:?}"
        );
        // Structured field lookup on the envelope record, not message
        // formatting.
        let original = record_field(&captured, "do_try catch block failed", "original_error")
            .unwrap_or_else(|| {
                panic!("envelope record missing original_error field, captured: {captured:?}")
            });
        let catch = record_field(&captured, "do_try catch block failed", "catch_error")
            .unwrap_or_else(|| {
                panic!("envelope record missing catch_error field, captured: {captured:?}")
            });
        assert!(
            original.contains("orig"),
            "original_error must carry the original error, got: {original}"
        );
        assert!(
            catch.contains("catch-fail"),
            "catch_error must carry the catch error, got: {catch}"
        );
    }
}