camel_api/

splitter.rs

use std::pin::Pin;
use std::sync::Arc;

use futures::Stream;

use crate::body::{Body, body_type_name};
use crate::error::CamelError;
use crate::exchange::Exchange;
use crate::message::Message;
use crate::value::Value;

/// A function that splits a single exchange into multiple fragment exchanges.
///
/// Built-in expressions return [`CamelError::TypeConversionFailed`] when the
/// body type does not match their input contract; empty-content bodies yield
/// `Ok(Vec::new())` (pass-through).
pub type SplitExpression =
    Arc<dyn Fn(&Exchange) -> Result<Vec<Exchange>, CamelError> + Send + Sync>;

/// Fallible split source: a synchronous [`SplitExpression`] or a
/// language-backed asynchronous expression.
///
/// The async arm evaluates to a [`Value`]; fragments are derived with the
/// canonical split rules (string → non-empty line fragments, array → one
/// fragment per element). Evaluation errors propagate before any fragment
/// is materialized.
#[derive(Clone)]
#[non_exhaustive]
pub enum SplitSource {
    /// Programmatic synchronous splitter.
    Sync(SplitExpression),
    /// Language-backed asynchronous expression.
    Async(Arc<dyn Fn(&Exchange) -> crate::filter::BoxValueFuture + Send + Sync>),
}

impl From<SplitExpression> for SplitSource {
    fn from(f: SplitExpression) -> Self {
        Self::Sync(f)
    }
}

impl SplitSource {
    /// Split `exchange` into fragments, propagating evaluation failures.
    pub async fn split(&self, exchange: &Exchange) -> Result<Vec<Exchange>, CamelError> {
        match self {
            Self::Sync(f) => f(exchange),
            Self::Async(f) => derive_fragments(exchange, f(exchange).await?),
        }
    }
}

/// Derive fragment exchanges from an evaluated language value.
///
/// Mirrors the rules previously inlined in camel-core `splitting.rs`
/// (DeclarativeSplit): a string yields one fragment per non-empty line; an
/// array yields one fragment per element (string elements become text
/// bodies, everything else JSON); anything else is a type error.
fn derive_fragments(exchange: &Exchange, value: Value) -> Result<Vec<Exchange>, CamelError> {
    match value {
        Value::String(s) => Ok(s
            .lines()
            .filter(|line| !line.is_empty())
            .map(|line| {
                let mut fragment = exchange.clone();
                fragment.input.body = Body::from(line.to_string());
                fragment
            })
            .collect()),
        Value::Array(arr) => Ok(arr
            .into_iter()
            .map(|v| {
                let mut fragment = exchange.clone();
                fragment.input.body = match v {
                    Value::String(s) => Body::Text(s),
                    other => Body::Json(other),
                };
                fragment
            })
            .collect()),
        other => Err(CamelError::TypeConversionFailed(format!(
            "declarative split requires a text or array value, got {received}; add an unmarshal step before split",
            received = crate::value_type_name(&other)
        ))),
    }
}

/// A function that lazily produces a stream of exchange fragments.
///
/// Used by the streaming split segment (camel-processor) for v1 sequential streaming split
/// (e.g., ZIP entry extraction, CSV/JSON streaming in future work).
///
/// Each call returns a `Stream` that yields fragments one at a time.
pub type StreamingSplitExpression = Arc<
    dyn Fn(Exchange) -> Pin<Box<dyn Stream<Item = Result<Exchange, CamelError>> + Send>>
        + Send
        + Sync,
>;

/// Typed error returned when a streaming split receives a body that is not
/// `Body::Stream`. Shared by the compiled production expression, the test
/// mirrors, and examples so the message cannot drift between copies.
pub fn streaming_split_type_error(body: &Body) -> CamelError {
    CamelError::TypeConversionFailed(format!(
        "streaming split requires body type stream, got {}; add an unmarshal step before split",
        body_type_name(body)
    ))
}

/// Strategy for aggregating fragment results back into a single exchange.
#[derive(Clone, Default)]
#[non_exhaustive]
pub enum AggregationStrategy {
    /// Result is the last fragment's exchange (default).
    #[default]
    LastWins,
    /// Collects all fragment bodies into a JSON array.
    CollectAll,
    /// Returns the original exchange unchanged.
    Original,
    /// Custom aggregation function: `(accumulated, next) -> merged`.
    Custom(Arc<dyn Fn(Exchange, Exchange) -> Exchange + Send + Sync>),
}

impl std::fmt::Debug for AggregationStrategy {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            AggregationStrategy::LastWins => f.write_str("LastWins"),
            AggregationStrategy::CollectAll => f.write_str("CollectAll"),
            AggregationStrategy::Original => f.write_str("Original"),
            AggregationStrategy::Custom(_) => f.write_str("Custom(..)"),
        }
    }
}

/// The streaming format to use when splitting a stream body.
#[derive(
    Clone,
    Debug,
    Default,
    PartialEq,
    Eq,
    serde::Serialize,
    serde::Deserialize,
    schemars::JsonSchema,
    ts_rs::TS,
)]
#[serde(rename_all = "snake_case")]
#[ts(rename_all = "snake_case")]
#[non_exhaustive]
pub enum StreamSplitFormat {
    /// Auto-detect the format from the body content.
    #[default]
    Auto,
    /// Newline-delimited JSON — each line is a complete JSON value.
    Ndjson,
    /// Split by newlines, each line becomes a text fragment.
    Lines,
    /// Split into fixed-size byte chunks.
    Chunks,
    /// ZIP archive — materialized format, each entry becomes a fragment exchange.
    Zip,
    /// TAR archive — materialized format, each regular-file entry becomes a fragment exchange.
    Tar,
    /// GZIP-compressed TAR archive — materialized format, each regular-file entry becomes a fragment exchange.
    #[serde(rename = "tar.gz")]
    #[ts(rename = "tar.gz")]
    TarGz,
    /// Materialized MIME message — one fragment per attachment part.
    Mime,
}

/// Configuration for splitting a streaming body into fragments.
///
/// Controls how the stream splitter processes the body, including format
/// detection, sizing limits, and metadata propagation.
#[derive(
    Clone,
    Debug,
    PartialEq,
    Eq,
    serde::Serialize,
    serde::Deserialize,
    schemars::JsonSchema,
    ts_rs::TS,
)]
#[serde(rename_all = "snake_case")]
#[ts(rename_all = "snake_case")]
pub struct StreamSplitConfig {
    /// The streaming format to use.
    pub format: StreamSplitFormat,
    /// Maximum size (in bytes) of a single record or chunk.
    pub max_record_bytes: usize,
    /// Number of records/chunks to collect into a single exchange batch.
    pub batch_size: usize,
    /// Explicit chunk size in bytes (required when format is [`Chunks`](StreamSplitFormat::Chunks)).
    pub chunk_size: Option<usize>,
    /// Maximum number of attachment parts to materialize (MIME only).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_parts: Option<usize>,
    /// Maximum size (in bytes) of a single attachment part (MIME only).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub max_part_size: Option<usize>,
    /// Content types to include when splitting a MIME message (MIME only).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub content_types: Option<Vec<String>>,
    /// Whether to include origin metadata in each fragment.
    pub include_origin: bool,
}

impl Default for StreamSplitConfig {
    fn default() -> Self {
        Self {
            format: StreamSplitFormat::Auto,
            max_record_bytes: 1024 * 1024,
            batch_size: 1,
            chunk_size: None,
            max_parts: None,
            max_part_size: None,
            content_types: None,
            include_origin: true,
        }
    }
}

impl StreamSplitConfig {
    /// Validates the configuration.
    ///
    /// # Errors
    ///
    /// Returns [`CamelError::Config`] if:
    /// - `batch_size` is `0`
    /// - `max_record_bytes` is `0`
    /// - `format` is [`Chunks`](StreamSplitFormat::Chunks) but `chunk_size` is `None`
    /// - `format` is a materialized archive format ([`Zip`](StreamSplitFormat::Zip),
    ///   [`Tar`](StreamSplitFormat::Tar), [`TarGz`](StreamSplitFormat::TarGz)) but
    ///   `chunk_size` is `Some(...)`
    /// - `chunk_size` is `Some(0)`
    pub fn validate(&self) -> Result<(), CamelError> {
        if self.batch_size == 0 {
            return Err(CamelError::Config(
                "stream split batch_size must be > 0".into(),
            ));
        }
        if self.max_record_bytes == 0 {
            return Err(CamelError::Config(
                "stream split max_record_bytes must be > 0".into(),
            ));
        }
        if self.format == StreamSplitFormat::Chunks && self.chunk_size.is_none() {
            return Err(CamelError::Config(
                "stream split format=Chunks requires chunk_size".into(),
            ));
        }
        // Materialized archive formats split whole entries, not byte chunks,
        // so chunk_size is meaningless. The Zip+chunk_size check must come
        // before the generic chunk_size zero/exceeds checks so that
        // `Zip + Some(0)` yields the more specific error.
        if self.format == StreamSplitFormat::Zip && self.chunk_size.is_some() {
            return Err(CamelError::Config(
                "stream split format=Zip does not support chunk_size".into(),
            ));
        }
        if matches!(
            self.format,
            StreamSplitFormat::Tar | StreamSplitFormat::TarGz
        ) && self.chunk_size.is_some()
        {
            return Err(CamelError::Config(format!(
                "stream split format={:?} is a materialized archive format and does not support chunk_size",
                self.format
            )));
        }
        if let Some(cs) = self.chunk_size
            && cs == 0
        {
            return Err(CamelError::Config(
                "stream split chunk_size must be > 0".into(),
            ));
        }
        if self.format == StreamSplitFormat::Chunks
            && let Some(cs) = self.chunk_size
            && cs > self.max_record_bytes
        {
            return Err(CamelError::Config(
                "stream split chunk_size must be <= max_record_bytes".into(),
            ));
        }
        // MIME materializes attachment parts, not byte chunks or record
        // batches. The mime-only options are rejected on every other format
        // so a stray option cannot silently do nothing.
        if self.format == StreamSplitFormat::Mime {
            if self.chunk_size.is_some() {
                return Err(CamelError::Config(
                    "stream split format=Mime does not support chunk_size".into(),
                ));
            }
            if self.batch_size != 1 {
                return Err(CamelError::Config(
                    "stream split format=Mime requires batch_size=1".into(),
                ));
            }
            if self.max_parts == Some(0) {
                return Err(CamelError::Config(
                    "stream split max_parts must be > 0".into(),
                ));
            }
            if self.max_part_size == Some(0) {
                return Err(CamelError::Config(
                    "stream split max_part_size must be > 0".into(),
                ));
            }
        } else {
            if self.max_parts.is_some() {
                return Err(CamelError::Config(
                    "stream split max_parts is only supported for format=Mime".into(),
                ));
            }
            if self.max_part_size.is_some() {
                return Err(CamelError::Config(
                    "stream split max_part_size is only supported for format=Mime".into(),
                ));
            }
            if self.content_types.is_some() {
                return Err(CamelError::Config(
                    "stream split content_types is only supported for format=Mime".into(),
                ));
            }
        }
        Ok(())
    }
}

/// Default threshold above which split fragments start new traces (0 = off).
pub const DEFAULT_TRACE_ITEM_THRESHOLD: usize = 100;

/// Default cap on the number of fragments an eager split may materialize.
pub const DEFAULT_MAX_FRAGMENTS: usize = 100_000;

/// Configuration for the Splitter EIP.
#[derive(Clone)]
pub struct SplitterConfig {
    /// Expression that splits an exchange into fragments.
    pub expression: SplitSource,
    /// How to aggregate fragment results.
    pub aggregation: AggregationStrategy,
    /// Whether to process fragments in parallel.
    pub parallel: bool,
    /// Maximum number of parallel fragments (None = unlimited).
    pub parallel_limit: Option<usize>,
    /// Whether to stop processing on the first exception.
    ///
    /// In parallel mode this only affects aggregation (the first error is
    /// propagated), **not** in-flight futures — `join_all` cannot cancel
    /// already-spawned work.
    pub stop_on_exception: bool,
    /// Maximum number of fragments materialized by `expression` (DoS cap, R3-M4).
    ///
    /// The eager splitter materializes the whole `Vec<Exchange>` before
    /// processing; this cap is enforced by `SplitSegment` after eager
    /// materialization, rejecting a split that would explode memory. Default
    /// [`DEFAULT_MAX_FRAGMENTS`]. A segment constructed directly with `0`
    /// fails closed for any non-empty split. For unbounded/lazy input use
    /// `StreamingSplitter`.
    pub max_fragments: usize,
    /// Threshold above which split fragments start new traces (0 = off).
    pub trace_item_threshold: usize,
}

impl std::fmt::Debug for SplitterConfig {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SplitterConfig")
            .field("expression", &"<split-expression>")
            .field("aggregation", &self.aggregation)
            .field("parallel", &self.parallel)
            .field("parallel_limit", &self.parallel_limit)
            .field("stop_on_exception", &self.stop_on_exception)
            .field("max_fragments", &self.max_fragments)
            .field("trace_item_threshold", &self.trace_item_threshold)
            .finish()
    }
}

impl SplitterConfig {
    /// Create a new splitter config with the given split expression.
    pub fn new(expression: impl Into<SplitSource>) -> Self {
        Self {
            expression: expression.into(),
            aggregation: AggregationStrategy::default(),
            parallel: false,
            parallel_limit: None,
            stop_on_exception: true,
            max_fragments: DEFAULT_MAX_FRAGMENTS,
            trace_item_threshold: DEFAULT_TRACE_ITEM_THRESHOLD,
        }
    }

    /// Set the aggregation strategy for combining fragment results.
    pub fn aggregation(mut self, strategy: AggregationStrategy) -> Self {
        self.aggregation = strategy;
        self
    }

    /// Enable or disable parallel fragment processing.
    pub fn parallel(mut self, parallel: bool) -> Self {
        self.parallel = parallel;
        self
    }

    /// Set the maximum number of concurrent fragments in parallel mode.
    pub fn parallel_limit(mut self, limit: usize) -> Self {
        self.parallel_limit = Some(limit);
        self
    }

    /// Control whether processing stops on the first fragment error.
    ///
    /// In parallel mode this only affects aggregation — see the field-level
    /// doc comment for details.
    pub fn stop_on_exception(mut self, stop: bool) -> Self {
        self.stop_on_exception = stop;
        self
    }

    /// Set the maximum number of fragments the eager splitter will materialize.
    pub fn max_fragments(mut self, max: usize) -> Self {
        self.max_fragments = max;
        self
    }

    /// Set the threshold above which split fragments start new traces (0 = off).
    pub fn trace_item_threshold(mut self, threshold: usize) -> Self {
        self.trace_item_threshold = threshold;
        self
    }

    /// Validates the configuration.
    ///
    /// Returns `Err(CamelError::Config)` if `parallel_limit` is set to 0,
    /// which would cause a `Semaphore::new(0)` panic at runtime.
    pub fn validate(&self) -> Result<(), CamelError> {
        if self.parallel && self.parallel_limit == Some(0) {
            return Err(CamelError::Config(
                "splitter parallel_limit must be > 0".to_string(),
            ));
        }
        if self.max_fragments == 0 {
            return Err(CamelError::Config(
                "splitter max_fragments must be > 0".to_string(),
            ));
        }
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/// Create a fragment exchange that inherits headers, properties, and OTel context
/// from the parent, but with a new body.
///
/// # OpenTelemetry Trace Propagation
///
/// Each fragment inherits the live segment step span context: the splitter step's
/// span, not the route root and not a previous fragment's span. A fragment-driven
/// sub-route root therefore opens as a child of that segment span in the same trace,
/// creating a natural fan-out relationship in the distributed trace:
///
/// ```text
/// Segment step (span A)
///   ├─ Fragment 1 sub-route root (span B, child of A)
///   ├─ Fragment 2 sub-route root (span C, child of A)
///   └─ Fragment N sub-route root (span N, child of A)
/// ```
///
/// Restoring the entry context after the segment completes is the segment wrapper's
/// job (`route_compiler::TracedSegmentStep` in `camel-core`), not
/// `fragment_exchange`'s job.
pub fn fragment_exchange(parent: &Exchange, body: Body) -> Exchange {
    let mut msg = Message::new(body);
    msg.headers = parent.input.headers.clone();
    let mut ex = Exchange::new(msg);
    ex.properties = parent.properties.clone();
    ex.pattern = parent.pattern;
    // Inherit OTel context so fragment spans are children of the parent span
    ex.otel_context = parent.otel_context.clone();
    // claimfamily: each fragment is a fanout branch exchange of the
    // parent — split a sibling claim onto it so drain stays covered
    // while the fragment is in flight.
    ex.in_flight_claim = parent.in_flight_claim.as_ref().map(|claim| claim.split());
    ex
}

/// Split the exchange body by newlines. Returns one fragment per line.
///
/// Empty bodies pass through with zero fragments. Wrong-type bodies return a
/// [`CamelError::TypeConversionFailed`] naming the received body type and the
/// expected `text` input.
pub fn split_body_lines() -> SplitExpression {
    Arc::new(|exchange: &Exchange| {
        let text = match &exchange.input.body {
            Body::Text(s) => s.as_str(),
            Body::Empty => return Ok(Vec::new()),
            _ => {
                return Err(CamelError::TypeConversionFailed(format!(
                    "split expression 'body_lines' requires body type text, got {received}; add an unmarshal step before split",
                    received = body_type_name(&exchange.input.body)
                )));
            }
        };
        Ok(text
            .lines()
            .map(|line| fragment_exchange(exchange, Body::Text(line.to_string())))
            .collect())
    })
}

/// Split a JSON array body into one fragment per element.
///
/// Fragment typing is element-driven: string elements produce [`Body::Text`]
/// fragments carrying the raw string (so `${body}` renders it unquoted);
/// number, boolean, object, nested-array, and null elements produce
/// [`Body::Json`] fragments.
///
/// Empty bodies and empty arrays pass through with zero fragments. Non-array
/// JSON and wrong-type bodies return a [`CamelError::TypeConversionFailed`]
/// naming the received body type and the expected `json (array)` input.
pub fn split_body_json_array() -> SplitExpression {
    Arc::new(|exchange: &Exchange| {
        let arr = match &exchange.input.body {
            Body::Json(serde_json::Value::Array(arr)) => arr,
            Body::Empty => return Ok(Vec::new()),
            Body::Json(_) => {
                return Err(CamelError::TypeConversionFailed(
                    "split expression 'body_json_array' requires body type json (array), got json (non-array); add an unmarshal step before split"
                        .to_string(),
                ))
            }
            _ => {
                return Err(CamelError::TypeConversionFailed(format!(
                    "split expression 'body_json_array' requires body type json (array), got {received}; add an unmarshal step before split",
                    received = body_type_name(&exchange.input.body)
                )))
            }
        };
        Ok(arr
            .iter()
            .map(|val| match val {
                serde_json::Value::String(s) => fragment_exchange(exchange, Body::Text(s.clone())),
                other => fragment_exchange(exchange, Body::Json(other.clone())),
            })
            .collect())
    })
}

/// Split the exchange body using a custom function that operates on the body.
///
/// Custom closures stay infallible: they own their body-type policy and an
/// empty `Vec` keeps pass-through semantics.
pub fn split_body<F>(f: F) -> SplitExpression
where
    F: Fn(&Body) -> Vec<Body> + Send + Sync + 'static,
{
    Arc::new(move |exchange: &Exchange| {
        Ok(f(&exchange.input.body)
            .into_iter()
            .map(|body| fragment_exchange(exchange, body))
            .collect())
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::in_flight::{InFlightClaim, InFlightGauge};
    use crate::value::Value;
    use std::sync::Arc;

    #[test]
    fn test_split_body_lines() {
        let mut ex = Exchange::new(Message::new("a\nb\nc"));
        ex.input.set_header("source", Value::String("test".into()));
        ex.set_property("trace", Value::Bool(true));

        let fragments = split_body_lines()(&ex).unwrap();
        assert_eq!(fragments.len(), 3);
        assert_eq!(fragments[0].input.body.as_text(), Some("a"));
        assert_eq!(fragments[1].input.body.as_text(), Some("b"));
        assert_eq!(fragments[2].input.body.as_text(), Some("c"));

        // Verify headers and properties inherited
        for frag in &fragments {
            assert_eq!(
                frag.input.header("source"),
                Some(&Value::String("test".into()))
            );
            assert_eq!(frag.property("trace"), Some(&Value::Bool(true)));
        }
    }

    #[test]
    fn test_split_body_lines_empty() {
        let ex = Exchange::new(Message::default()); // Body::Empty
        let fragments = split_body_lines()(&ex).unwrap();
        assert!(fragments.is_empty());
    }

    #[test]
    fn fragment_exchange_splits_parent_claim() {
        let gauge = Arc::new(InFlightGauge::new());
        let mut parent = Exchange::new(Message::new(Body::Text("a,b".into())));
        parent.in_flight_claim = Some(InFlightClaim::attach(&gauge));

        let frag = fragment_exchange(&parent, Body::Text("a".into()));
        assert_eq!(gauge.total(), 2);
        assert!(frag.in_flight_claim.is_some());
        assert!(parent.in_flight_claim.is_some());

        drop(frag);
        assert_eq!(gauge.total(), 1);
    }

    #[test]
    fn test_split_body_json_array() {
        let arr = serde_json::json!([1, 2, 3]);
        let ex = Exchange::new(Message::new(arr));

        let fragments = split_body_json_array()(&ex).unwrap();
        assert_eq!(fragments.len(), 3);
        assert!(matches!(&fragments[0].input.body, Body::Json(v) if *v == serde_json::json!(1)));
        assert!(matches!(&fragments[1].input.body, Body::Json(v) if *v == serde_json::json!(2)));
        assert!(matches!(&fragments[2].input.body, Body::Json(v) if *v == serde_json::json!(3)));
    }

    #[test]
    fn split_body_json_array_string_elements_become_text() {
        // The leading empty-string element pins the accepted delta from
        // mission 251 (bd rc-etf0q): an empty element yields an empty
        // `Body::Text`, not a quoted `""` JSON string.
        let ex = Exchange::new(Message::new(serde_json::json!(["", "a", "b"])));

        let fragments = split_body_json_array()(&ex).unwrap();
        assert_eq!(fragments.len(), 3);
        assert!(
            matches!(&fragments[0].input.body, Body::Text(s) if s.is_empty()),
            "fragment 0 must be Body::Text(\"\"), got {:?}",
            fragments[0].input.body
        );
        assert!(matches!(&fragments[1].input.body, Body::Text(s) if s == "a"));
        assert!(matches!(&fragments[2].input.body, Body::Text(s) if s == "b"));
        // No fragment body carries a literal quote character: string elements
        // are raw text, so `${body}` renders them unquoted.
        for frag in &fragments {
            let text = match &frag.input.body {
                Body::Text(s) => s.as_str(),
                other => panic!("expected Body::Text fragment, got {other:?}"),
            };
            assert!(
                !text.contains('"'),
                "fragment body must not carry a quote character, got {text:?}"
            );
        }
    }

    #[test]
    fn test_split_body_json_array_non_string_elements_stay_json() {
        let ex = Exchange::new(Message::new(serde_json::json!([1, {"k": "v"}, null])));

        let fragments = split_body_json_array()(&ex).unwrap();
        assert_eq!(fragments.len(), 3);
        assert!(matches!(&fragments[0].input.body, Body::Json(v) if *v == serde_json::json!(1)));
        assert!(matches!(&fragments[1].input.body, Body::Json(v)
                if *v == serde_json::json!({"k": "v"})));
        assert!(matches!(&fragments[2].input.body, Body::Json(v) if v.is_null()));
    }

    #[test]
    fn test_split_body_json_array_not_array() {
        let obj = serde_json::json!({"not": "array"});
        let ex = Exchange::new(Message::new(obj));

        let err = split_body_json_array()(&ex).unwrap_err();
        assert!(matches!(err, CamelError::TypeConversionFailed(_)));
        assert!(err.to_string().contains("json (non-array)"));
    }

    #[test]
    fn test_split_body_lines_wrong_type_json_errors() {
        let ex = Exchange::new(Message::new(serde_json::json!({"a": 1})));

        let err = split_body_lines()(&ex).unwrap_err();
        let msg = err.to_string();
        assert!(matches!(err, CamelError::TypeConversionFailed(_)));
        for needle in [
            "body_lines",
            "json",
            "text",
            "add an unmarshal step before split",
        ] {
            assert!(msg.contains(needle), "message '{msg}' missing '{needle}'");
        }
    }

    #[test]
    fn test_split_body_json_array_wrong_type_text_errors() {
        let ex = Exchange::new(Message::new("x"));

        let err = split_body_json_array()(&ex).unwrap_err();
        let msg = err.to_string();
        assert!(matches!(err, CamelError::TypeConversionFailed(_)));
        for needle in [
            "body_json_array",
            "text",
            "json (array)",
            "add an unmarshal step before split",
        ] {
            assert!(msg.contains(needle), "message '{msg}' missing '{needle}'");
        }
    }

    #[test]
    fn test_split_body_json_array_non_array_json_errors() {
        let ex = Exchange::new(Message::new(serde_json::json!({"o": 1})));

        let err = split_body_json_array()(&ex).unwrap_err();
        let msg = err.to_string();
        assert!(matches!(err, CamelError::TypeConversionFailed(_)));
        assert!(msg.contains("json (non-array)"));
    }

    #[test]
    fn test_split_body_lines_empty_body_ok() {
        let ex = Exchange::new(Message::default()); // Body::Empty
        let fragments = split_body_lines()(&ex).unwrap();
        assert!(fragments.is_empty());
    }

    #[test]
    fn test_split_body_json_array_empty_body_ok() {
        let ex = Exchange::new(Message::default()); // Body::Empty
        let fragments = split_body_json_array()(&ex).unwrap();
        assert!(fragments.is_empty());
    }

    #[test]
    fn test_split_body_json_array_empty_array_ok() {
        let ex = Exchange::new(Message::new(serde_json::json!([])));
        let fragments = split_body_json_array()(&ex).unwrap();
        assert!(fragments.is_empty());
    }

    #[test]
    fn test_split_body_lines_empty_text_ok() {
        let ex = Exchange::new(Message::new(""));
        let fragments = split_body_lines()(&ex).unwrap();
        assert!(fragments.is_empty());
    }

    #[test]
    fn test_split_error_omits_payload() {
        let ex = Exchange::new(Message::new(serde_json::json!({
            "secret": "SECRET-8f31a"
        })));

        let err = split_body_lines()(&ex).unwrap_err();
        let msg = err.to_string();
        assert!(matches!(err, CamelError::TypeConversionFailed(_)));
        for needle in [
            "body_lines",
            "json",
            "text",
            "add an unmarshal step before split",
        ] {
            assert!(msg.contains(needle), "message '{msg}' missing '{needle}'");
        }
        assert!(
            !msg.contains("SECRET-8f31a"),
            "message '{msg}' leaks payload"
        );
    }

    #[test]
    fn test_split_body_custom() {
        let splitter = split_body(|body: &Body| match body {
            Body::Text(s) => s
                .split(',')
                .map(|part| Body::Text(part.trim().to_string()))
                .collect(),
            _ => Vec::new(),
        });

        let mut ex = Exchange::new(Message::new("x, y, z"));
        ex.set_property("id", Value::from(42));

        let fragments = splitter(&ex).unwrap();
        assert_eq!(fragments.len(), 3);
        assert_eq!(fragments[0].input.body.as_text(), Some("x"));
        assert_eq!(fragments[1].input.body.as_text(), Some("y"));
        assert_eq!(fragments[2].input.body.as_text(), Some("z"));

        // Properties inherited
        for frag in &fragments {
            assert_eq!(frag.property("id"), Some(&Value::from(42)));
        }
    }

    #[test]
    fn test_splitter_config_defaults() {
        let config = SplitterConfig::new(split_body_lines());
        assert!(matches!(config.aggregation, AggregationStrategy::LastWins));
        assert!(!config.parallel);
        assert!(config.parallel_limit.is_none());
        assert!(config.stop_on_exception);
    }

    #[test]
    fn test_splitter_config_builder() {
        let config = SplitterConfig::new(split_body_lines())
            .aggregation(AggregationStrategy::CollectAll)
            .parallel(true)
            .parallel_limit(4)
            .stop_on_exception(false);

        assert!(matches!(
            config.aggregation,
            AggregationStrategy::CollectAll
        ));
        assert!(config.parallel);
        assert_eq!(config.parallel_limit, Some(4));
        assert!(!config.stop_on_exception);
    }

    #[test]
    fn test_splitter_config_default_max_fragments() {
        let cfg = SplitterConfig::new(Arc::new(|_: &Exchange| Ok(Vec::new())) as SplitExpression);
        assert_eq!(cfg.max_fragments, 100_000);
    }

    #[test]
    fn test_splitter_config_rejects_zero_max_fragments() {
        let cfg = SplitterConfig::new(Arc::new(|_: &Exchange| Ok(Vec::new())) as SplitExpression)
            .max_fragments(0);
        assert!(cfg.validate().is_err());
    }

    #[test]
    fn test_fragment_exchange_inherits_otel_context() {
        use opentelemetry::Context;
        use opentelemetry::trace::{SpanContext, SpanId, TraceContextExt, TraceFlags, TraceId};

        // Create parent exchange with a valid span context
        let mut parent = Exchange::new(Message::new("test"));
        let trace_id = TraceId::from_bytes([0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 123]);
        let span_id = SpanId::from_bytes([0, 0, 0, 0, 0, 0, 1, 200]);
        let span_context = SpanContext::new(
            trace_id,
            span_id,
            TraceFlags::SAMPLED,
            true,
            Default::default(),
        );
        let expected_trace_id = span_context.trace_id();
        parent.otel_context = Context::current().with_remote_span_context(span_context);

        // Create fragment via split_body_lines
        let fragments = split_body_lines()(&parent).unwrap();
        assert!(!fragments.is_empty(), "Should have at least one fragment");

        // Verify each fragment has the same span context as parent
        for fragment in &fragments {
            let span = fragment.otel_context.span();
            let frag_span_ctx = span.span_context();
            assert!(
                frag_span_ctx.is_valid(),
                "Fragment should have valid span context"
            );
            assert_eq!(
                frag_span_ctx.trace_id(),
                expected_trace_id,
                "Fragment should have same trace ID as parent"
            );
        }
    }

    #[test]
    fn test_stream_split_config_defaults_valid() {
        let config = StreamSplitConfig::default();
        assert!(config.validate().is_ok());
    }

    #[test]
    fn test_stream_split_config_batch_size_zero_rejected() {
        let config = StreamSplitConfig {
            batch_size: 0,
            ..Default::default()
        };
        let err = config.validate().unwrap_err();
        assert!(err.to_string().contains("batch_size"));
    }

    #[test]
    fn test_stream_split_config_max_record_bytes_zero_rejected() {
        let config = StreamSplitConfig {
            max_record_bytes: 0,
            ..Default::default()
        };
        let err = config.validate().unwrap_err();
        assert!(err.to_string().contains("max_record_bytes"));
    }

    #[test]
    fn test_stream_split_config_chunks_requires_chunk_size() {
        let config = StreamSplitConfig {
            format: StreamSplitFormat::Chunks,
            chunk_size: None,
            ..Default::default()
        };
        let err = config.validate().unwrap_err();
        assert!(err.to_string().contains("Chunks requires chunk_size"));
    }

    #[test]
    fn test_stream_split_config_chunk_size_zero_rejected() {
        let config = StreamSplitConfig {
            format: StreamSplitFormat::Chunks,
            chunk_size: Some(0),
            ..Default::default()
        };
        let err = config.validate().unwrap_err();
        assert!(err.to_string().contains("chunk_size must be > 0"));
    }

    #[test]
    fn test_stream_split_config_chunk_size_exceeds_max_record_bytes() {
        let config = StreamSplitConfig {
            format: StreamSplitFormat::Chunks,
            chunk_size: Some(2000),
            max_record_bytes: 1000,
            ..Default::default()
        };
        let err = config.validate().unwrap_err();
        assert!(
            err.to_string()
                .contains("chunk_size must be <= max_record_bytes")
        );
    }

    #[test]
    fn test_stream_split_config_zip_rejects_chunk_size() {
        let config = StreamSplitConfig {
            format: StreamSplitFormat::Zip,
            chunk_size: Some(1024),
            ..Default::default()
        };
        let err = config.validate().unwrap_err();
        assert!(err.to_string().contains("Zip does not support chunk_size"));
    }

    #[test]
    fn test_all_fragments_share_same_trace_context() {
        use opentelemetry::Context;
        use opentelemetry::trace::{SpanContext, SpanId, TraceContextExt, TraceFlags, TraceId};

        // Create parent with a specific trace ID
        let mut parent = Exchange::new(Message::new("line1\nline2\nline3"));
        let trace_id =
            TraceId::from_bytes([0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0x3B, 0x9A, 0xCA, 0x09]);
        let span_id = SpanId::from_bytes([0, 0, 0, 0, 0, 0, 0, 111]);
        let span_context = SpanContext::new(
            trace_id,
            span_id,
            TraceFlags::SAMPLED,
            true,
            Default::default(),
        );
        parent.otel_context = Context::current().with_remote_span_context(span_context);

        let fragments = split_body_lines()(&parent).unwrap();
        assert_eq!(fragments.len(), 3);

        // All fragments should share the same trace ID
        let trace_ids: Vec<_> = fragments
            .iter()
            .map(|f| {
                let span = f.otel_context.span();
                span.span_context().trace_id()
            })
            .collect();

        assert!(
            trace_ids.iter().all(|&id| id == trace_id),
            "All fragments should have the same trace ID"
        );
    }

    #[test]
    fn trace_item_threshold_defaults_to_100() {
        let config = SplitterConfig::new(split_body_lines());
        assert_eq!(config.trace_item_threshold, 100);
    }

    #[test]
    fn trace_item_threshold_builder_sets_value() {
        let config = SplitterConfig::new(split_body_lines()).trace_item_threshold(0);
        assert_eq!(config.trace_item_threshold, 0);

        let config = SplitterConfig::new(split_body_lines()).trace_item_threshold(7);
        assert_eq!(config.trace_item_threshold, 7);
    }

    #[test]
    fn canonical_split_spec_carries_trace_item_threshold() {
        use crate::runtime::{CanonicalSplitAggregationSpec, CanonicalSplitExpressionSpec};

        let with_threshold = crate::runtime::CanonicalStepSpec::Split {
            expression: CanonicalSplitExpressionSpec::BodyLines,
            aggregation: CanonicalSplitAggregationSpec::CollectAll,
            parallel: false,
            parallel_limit: None,
            stop_on_exception: true,
            trace_item_threshold: Some(5),
            steps: Vec::new(),
        };
        let without_threshold = crate::runtime::CanonicalStepSpec::Split {
            expression: CanonicalSplitExpressionSpec::BodyLines,
            aggregation: CanonicalSplitAggregationSpec::CollectAll,
            parallel: false,
            parallel_limit: None,
            stop_on_exception: true,
            trace_item_threshold: None,
            steps: Vec::new(),
        };

        // Mirror the derive(PartialEq)/destructure path `parallel_limit` uses.
        let crate::runtime::CanonicalStepSpec::Split {
            trace_item_threshold,
            ..
        } = &with_threshold
        else {
            unreachable!()
        };
        assert_eq!(trace_item_threshold, &Some(5));

        let crate::runtime::CanonicalStepSpec::Split {
            trace_item_threshold,
            ..
        } = &without_threshold
        else {
            unreachable!()
        };
        assert_eq!(trace_item_threshold, &None);

        assert_ne!(with_threshold, without_threshold);
    }
}