camel-processor 0.21.0

use super::*;
use camel_api::{BoxProcessorExt, Message};
use std::sync::Arc;
use std::sync::atomic::Ordering;
use tower::ServiceExt;

// ── Test helpers ───────────────────────────────────────────────────

fn make_exchange(body: &str) -> Exchange {
    Exchange::new(Message::new(body))
}

fn uppercase_processor() -> BoxProcessor {
    BoxProcessor::from_fn(|mut ex: Exchange| {
        Box::pin(async move {
            if let Body::Text(s) = &ex.input.body {
                ex.input.body = Body::Text(s.to_uppercase());
            }
            Ok(ex)
        })
    })
}

fn failing_processor() -> BoxProcessor {
    BoxProcessor::from_fn(|_ex| Box::pin(async { Err(CamelError::ProcessorError("boom".into())) }))
}

#[test]
fn test_multicast_zero_parallel_limit_rejected() {
    let config = MulticastConfig::new().parallel(true).parallel_limit(0);
    let result = MulticastService::new(vec![passthrough_processor()], config);
    assert!(result.is_err(), "zero parallel_limit should return Err");
}

fn passthrough_processor() -> BoxProcessor {
    BoxProcessor::from_fn(|ex| Box::pin(async move { Ok(ex) }))
}

// ── 1. Sequential + LastWins ───────────────────────────────────────

#[tokio::test]
async fn test_multicast_sequential_last_wins() {
    let endpoints = vec![
        uppercase_processor(),
        uppercase_processor(),
        uppercase_processor(),
    ];

    let config = MulticastConfig::new(); // LastWins by default
    let mut svc = MulticastService::new(endpoints, config).unwrap();

    let result = svc
        .ready()
        .await
        .unwrap()
        .call(make_exchange("hello"))
        .await
        .unwrap();

    assert_eq!(result.input.body.as_text(), Some("HELLO"));
}

// ── 2. Sequential + CollectAll ─────────────────────────────────────

#[tokio::test]
async fn test_multicast_sequential_collect_all() {
    let endpoints = vec![
        uppercase_processor(),
        uppercase_processor(),
        uppercase_processor(),
    ];

    let config = MulticastConfig::new().aggregation(MulticastStrategy::CollectAll);
    let mut svc = MulticastService::new(endpoints, config).unwrap();

    let result = svc
        .ready()
        .await
        .unwrap()
        .call(make_exchange("hello"))
        .await
        .unwrap();

    let expected = serde_json::json!(["HELLO", "HELLO", "HELLO"]);
    match &result.input.body {
        Body::Json(v) => assert_eq!(*v, expected),
        other => panic!("expected JSON body, got {other:?}"),
    }
}

// ── 3. Sequential + Original ───────────────────────────────────────

#[tokio::test]
async fn test_multicast_sequential_original() {
    let endpoints = vec![
        uppercase_processor(),
        uppercase_processor(),
        uppercase_processor(),
    ];

    let config = MulticastConfig::new().aggregation(MulticastStrategy::Original);
    let mut svc = MulticastService::new(endpoints, config).unwrap();

    let result = svc
        .ready()
        .await
        .unwrap()
        .call(make_exchange("hello"))
        .await
        .unwrap();

    // Original body should be unchanged
    assert_eq!(result.input.body.as_text(), Some("hello"));
}

// ── 4. Sequential + Custom aggregation ─────────────────────────────

#[tokio::test]
async fn test_multicast_sequential_custom_aggregation() {
    let joiner: Arc<dyn Fn(Exchange, Exchange) -> Exchange + Send + Sync> =
        Arc::new(|mut acc: Exchange, next: Exchange| {
            let acc_text = acc.input.body.as_text().unwrap_or("").to_string();
            let next_text = next.input.body.as_text().unwrap_or("").to_string();
            acc.input.body = Body::Text(format!("{acc_text}+{next_text}"));
            acc
        });

    let endpoints = vec![
        uppercase_processor(),
        uppercase_processor(),
        uppercase_processor(),
    ];

    let config = MulticastConfig::new().aggregation(MulticastStrategy::Custom(joiner));
    let mut svc = MulticastService::new(endpoints, config).unwrap();

    let result = svc
        .ready()
        .await
        .unwrap()
        .call(make_exchange("a"))
        .await
        .unwrap();

    assert_eq!(result.input.body.as_text(), Some("A+A+A"));
}

// ── 5. Stop on exception ───────────────────────────────────────────

#[tokio::test]
async fn test_multicast_stop_on_exception() {
    let endpoints = vec![
        uppercase_processor(),
        failing_processor(),
        uppercase_processor(),
    ];

    let config = MulticastConfig::new().stop_on_exception(true);
    let mut svc = MulticastService::new(endpoints, config).unwrap();

    let result = svc
        .ready()
        .await
        .unwrap()
        .call(make_exchange("hello"))
        .await;

    assert!(result.is_err(), "expected error due to stop_on_exception");
}

// ── 6. Continue on exception ───────────────────────────────────────

#[tokio::test]
async fn test_multicast_continue_on_exception() {
    let endpoints = vec![
        uppercase_processor(),
        failing_processor(),
        uppercase_processor(),
    ];

    let config = MulticastConfig::new()
        .stop_on_exception(false)
        .aggregation(MulticastStrategy::LastWins);
    let mut svc = MulticastService::new(endpoints, config).unwrap();

    let result = svc
        .ready()
        .await
        .unwrap()
        .call(make_exchange("hello"))
        .await;

    // LastWins: last endpoint succeeded, so result should be OK
    assert!(result.is_ok(), "last endpoint should succeed");
    assert_eq!(result.unwrap().input.body.as_text(), Some("HELLO"));
}

// ── 7. Stop on exception with fail_on_nth ─────────────────────────────

#[tokio::test]
async fn test_multicast_stop_on_exception_halts_early() {
    use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};

    // Track which endpoints actually execute
    let executed = Arc::new(AtomicUsize::new(0));

    let exec_clone1 = Arc::clone(&executed);
    let endpoint0 = BoxProcessor::from_fn(move |ex: Exchange| {
        let e = Arc::clone(&exec_clone1);
        Box::pin(async move {
            e.fetch_add(1, AtomicOrdering::SeqCst);
            Ok(ex)
        })
    });

    let exec_clone2 = Arc::clone(&executed);
    let endpoint1 = BoxProcessor::from_fn(move |_ex: Exchange| {
        let e = Arc::clone(&exec_clone2);
        Box::pin(async move {
            e.fetch_add(1, AtomicOrdering::SeqCst);
            Err(CamelError::ProcessorError("fail on 1".into()))
        })
    });

    let exec_clone3 = Arc::clone(&executed);
    let endpoint2 = BoxProcessor::from_fn(move |ex: Exchange| {
        let e = Arc::clone(&exec_clone3);
        Box::pin(async move {
            e.fetch_add(1, AtomicOrdering::SeqCst);
            Ok(ex)
        })
    });

    let endpoints = vec![endpoint0, endpoint1, endpoint2];
    let config = MulticastConfig::new().stop_on_exception(true);
    let mut svc = MulticastService::new(endpoints, config).unwrap();

    let result = svc.ready().await.unwrap().call(make_exchange("x")).await;
    assert!(result.is_err(), "should fail at endpoint 1");

    // Only endpoints 0 and 1 should have executed (2 should be skipped)
    let count = executed.load(AtomicOrdering::SeqCst);
    assert_eq!(
        count, 2,
        "endpoint 2 should not have executed due to stop_on_exception"
    );
}

// ── 8. Continue on exception with fail_on_nth ─────────────────────────

#[tokio::test]
async fn test_multicast_continue_on_exception_executes_all() {
    use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};

    // Track which endpoints actually execute
    let executed = Arc::new(AtomicUsize::new(0));

    let exec_clone1 = Arc::clone(&executed);
    let endpoint0 = BoxProcessor::from_fn(move |ex: Exchange| {
        let e = Arc::clone(&exec_clone1);
        Box::pin(async move {
            e.fetch_add(1, AtomicOrdering::SeqCst);
            Ok(ex)
        })
    });

    let exec_clone2 = Arc::clone(&executed);
    let endpoint1 = BoxProcessor::from_fn(move |_ex: Exchange| {
        let e = Arc::clone(&exec_clone2);
        Box::pin(async move {
            e.fetch_add(1, AtomicOrdering::SeqCst);
            Err(CamelError::ProcessorError("fail on 1".into()))
        })
    });

    let exec_clone3 = Arc::clone(&executed);
    let endpoint2 = BoxProcessor::from_fn(move |ex: Exchange| {
        let e = Arc::clone(&exec_clone3);
        Box::pin(async move {
            e.fetch_add(1, AtomicOrdering::SeqCst);
            Ok(ex)
        })
    });

    let endpoints = vec![endpoint0, endpoint1, endpoint2];
    let config = MulticastConfig::new()
        .stop_on_exception(false)
        .aggregation(MulticastStrategy::LastWins);
    let mut svc = MulticastService::new(endpoints, config).unwrap();

    let result = svc.ready().await.unwrap().call(make_exchange("x")).await;
    assert!(result.is_ok(), "last endpoint should succeed");

    // All 3 endpoints should have executed
    let count = executed.load(AtomicOrdering::SeqCst);
    assert_eq!(
        count, 3,
        "all endpoints should have executed despite error in endpoint 1"
    );
}

// ── 9. Empty endpoints ─────────────────────────────────────────────

#[tokio::test]
async fn test_multicast_empty_endpoints() {
    let endpoints: Vec<BoxProcessor> = vec![];

    let config = MulticastConfig::new();
    let mut svc = MulticastService::new(endpoints, config).unwrap();

    let mut ex = make_exchange("hello");
    ex.set_property("marker", Value::Bool(true));

    let result = svc.ready().await.unwrap().call(ex).await.unwrap();
    assert_eq!(result.input.body.as_text(), Some("hello"));
    assert_eq!(result.property("marker"), Some(&Value::Bool(true)));
}

// ── 10. Metadata properties ─────────────────────────────────────────

#[tokio::test]
async fn test_multicast_metadata_properties() {
    // Use a pipeline that records the metadata into the body as JSON
    let recorder = BoxProcessor::from_fn(|ex: Exchange| {
        Box::pin(async move {
            let idx = ex.property(CAMEL_MULTICAST_INDEX).cloned();
            let complete = ex.property(CAMEL_MULTICAST_COMPLETE).cloned();
            let body = serde_json::json!({
                "index": idx,
                "complete": complete,
            });
            let mut out = ex;
            out.input.body = Body::Json(body);
            Ok(out)
        })
    });

    let endpoints = vec![recorder.clone(), recorder.clone(), recorder];

    let config = MulticastConfig::new().aggregation(MulticastStrategy::CollectAll);
    let mut svc = MulticastService::new(endpoints, config).unwrap();

    let result = svc
        .ready()
        .await
        .unwrap()
        .call(make_exchange("x"))
        .await
        .unwrap();

    let expected = serde_json::json!([
        {"index": 0, "complete": false},
        {"index": 1, "complete": false},
        {"index": 2, "complete": true},
    ]);
    match &result.input.body {
        Body::Json(v) => assert_eq!(*v, expected),
        other => panic!("expected JSON body, got {other:?}"),
    }
}

// ── 11. poll_ready returns Ready immediately ───────────────────────

#[tokio::test]
async fn test_poll_ready_returns_ready_immediately() {
    use std::sync::atomic::AtomicBool;

    // A service that is never ready on its own.
    #[derive(Clone)]
    struct NeverReady {
        _ready: Arc<AtomicBool>,
    }

    impl Service<Exchange> for NeverReady {
        type Response = Exchange;
        type Error = CamelError;
        type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;

        fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
            cx.waker().wake_by_ref();
            Poll::Pending
        }

        fn call(&mut self, exchange: Exchange) -> Self::Future {
            Box::pin(async move { Ok(exchange) })
        }
    }

    let inner = NeverReady {
        _ready: Arc::new(AtomicBool::new(false)),
    };
    let boxed: BoxProcessor = BoxProcessor::new(inner);

    let config = MulticastConfig::new();
    let mut svc = MulticastService::new(vec![boxed], config).unwrap();

    // MulticastService::poll_ready should return Ready(Ok(())) immediately,
    // even when the underlying endpoint is not ready. Readiness is deferred
    // to inside call() where stop_on_exception is respected.
    let waker = futures::task::noop_waker();
    let mut cx = Context::from_waker(&waker);
    let poll = Pin::new(&mut svc).poll_ready(&mut cx);
    assert!(
        matches!(poll, Poll::Ready(Ok(()))),
        "expected Ready(Ok(())) even when endpoint is not ready"
    );
}

// ── 12. CollectAll with error propagates ────────────────────────────

#[tokio::test]
async fn test_multicast_collect_all_error_propagates() {
    let endpoints = vec![
        uppercase_processor(),
        failing_processor(),
        uppercase_processor(),
    ];

    let config = MulticastConfig::new()
        .stop_on_exception(false)
        .aggregation(MulticastStrategy::CollectAll);
    let mut svc = MulticastService::new(endpoints, config).unwrap();

    let result = svc
        .ready()
        .await
        .unwrap()
        .call(make_exchange("hello"))
        .await;

    assert!(result.is_err(), "CollectAll should propagate first error");
}

// ── 13. LastWins with error last returns error ──────────────────────

#[tokio::test]
async fn test_multicast_last_wins_error_last() {
    let endpoints = vec![
        uppercase_processor(),
        uppercase_processor(),
        failing_processor(),
    ];

    let config = MulticastConfig::new()
        .stop_on_exception(false)
        .aggregation(MulticastStrategy::LastWins);
    let mut svc = MulticastService::new(endpoints, config).unwrap();

    let result = svc
        .ready()
        .await
        .unwrap()
        .call(make_exchange("hello"))
        .await;

    assert!(result.is_err(), "LastWins should return last error");
}

// ── 14. Custom aggregation with error propagates ────────────────────

#[tokio::test]
async fn test_multicast_custom_error_propagates() {
    let joiner: Arc<dyn Fn(Exchange, Exchange) -> Exchange + Send + Sync> =
        Arc::new(|acc: Exchange, _next: Exchange| acc);

    let endpoints = vec![
        uppercase_processor(),
        failing_processor(),
        uppercase_processor(),
    ];

    let config = MulticastConfig::new()
        .stop_on_exception(false)
        .aggregation(MulticastStrategy::Custom(joiner));
    let mut svc = MulticastService::new(endpoints, config).unwrap();

    let result = svc
        .ready()
        .await
        .unwrap()
        .call(make_exchange("hello"))
        .await;

    assert!(
        result.is_err(),
        "Custom aggregation should propagate errors"
    );
}

// ── 15. Parallel + CollectAll basic ─────────────────────────────────

#[tokio::test]
async fn test_multicast_parallel_basic() {
    let endpoints = vec![uppercase_processor(), uppercase_processor()];

    let config = MulticastConfig::new()
        .parallel(true)
        .aggregation(MulticastStrategy::CollectAll);
    let mut svc = MulticastService::new(endpoints, config).unwrap();

    let result = svc
        .ready()
        .await
        .unwrap()
        .call(make_exchange("test"))
        .await
        .unwrap();

    // Both endpoints uppercase "test" → ["TEST", "TEST"]
    // Note: parallel order is not guaranteed for CollectAll, but with identical processors it doesn't matter
    match &result.input.body {
        Body::Json(v) => {
            let arr = v.as_array().expect("expected array");
            assert_eq!(arr.len(), 2);
            assert!(arr.iter().all(|v| v.as_str() == Some("TEST")));
        }
        other => panic!("expected JSON body, got {:?}", other),
    }
}

// ── 16. Parallel with concurrency limit ─────────────────────────────

#[tokio::test]
async fn test_multicast_parallel_with_limit() {
    use std::sync::atomic::{AtomicUsize, Ordering as AtomicOrdering};

    let concurrent = Arc::new(AtomicUsize::new(0));
    let max_concurrent = Arc::new(AtomicUsize::new(0));

    let endpoints: Vec<BoxProcessor> = (0..4)
        .map(|_| {
            let c = Arc::clone(&concurrent);
            let mc = Arc::clone(&max_concurrent);
            BoxProcessor::from_fn(move |ex: Exchange| {
                let c = Arc::clone(&c);
                let mc = Arc::clone(&mc);
                Box::pin(async move {
                    let current = c.fetch_add(1, AtomicOrdering::SeqCst) + 1;
                    mc.fetch_max(current, AtomicOrdering::SeqCst);
                    tokio::task::yield_now().await;
                    c.fetch_sub(1, AtomicOrdering::SeqCst);
                    Ok(ex)
                })
            })
        })
        .collect();

    let config = MulticastConfig::new().parallel(true).parallel_limit(2);
    let mut svc = MulticastService::new(endpoints, config).unwrap();

    let _ = svc.ready().await.unwrap().call(make_exchange("x")).await;

    let observed_max = max_concurrent.load(std::sync::atomic::Ordering::SeqCst);
    assert!(
        observed_max <= 2,
        "max concurrency was {}, expected <= 2",
        observed_max
    );
}

// ── 17. Stream body creates valid JSON placeholder ────────────────────

async fn setup_multicast_stream_test(origin: Option<String>) -> Exchange {
    use bytes::Bytes;
    use camel_api::{Body, StreamBody, StreamMetadata};
    use futures::stream;
    use std::sync::Arc;
    use tokio::sync::Mutex;

    let chunks = vec![Ok(Bytes::from("test"))];
    let stream_body = StreamBody {
        stream: Arc::new(Mutex::new(Some(Box::pin(stream::iter(chunks))))),
        metadata: StreamMetadata {
            origin,
            ..Default::default()
        },
    };

    let stream_body_clone = stream_body.clone();
    let endpoints = vec![BoxProcessor::from_fn(move |ex: Exchange| {
        let body_clone = stream_body_clone.clone();
        Box::pin(async move {
            let mut out = ex;
            out.input.body = Body::Stream(body_clone);
            Ok(out)
        })
    })];

    let config = MulticastConfig::new().aggregation(MulticastStrategy::CollectAll);
    let mut svc = MulticastService::new(endpoints, config).unwrap();

    svc.ready()
        .await
        .unwrap()
        .call(Exchange::new(Message::new("")))
        .await
        .unwrap()
}

#[tokio::test]
async fn test_multicast_stream_bodies_creates_valid_json() {
    use camel_api::Body;

    let result = setup_multicast_stream_test(Some("http://example.com/data".to_string())).await;

    let Body::Json(value) = &result.input.body else {
        panic!("Expected Json body, got {:?}", result.input.body);
    };

    let json_str = serde_json::to_string(&value).unwrap();
    let parsed: serde_json::Value = serde_json::from_str(&json_str).unwrap();

    assert!(parsed.is_array());
    let arr = parsed.as_array().unwrap();
    assert_eq!(arr.len(), 1);
    assert!(arr[0]["_stream"].is_object());
    assert_eq!(arr[0]["_stream"]["origin"], "http://example.com/data");
    assert_eq!(arr[0]["_stream"]["placeholder"], true);
}

// ── 18. Stream body with None origin creates valid JSON ──────────────

#[tokio::test]
async fn test_multicast_stream_with_none_origin_creates_valid_json() {
    use camel_api::Body;

    let result = setup_multicast_stream_test(None).await;

    let Body::Json(value) = &result.input.body else {
        panic!("Expected Json body, got {:?}", result.input.body);
    };

    let json_str = serde_json::to_string(&value).unwrap();
    let parsed: serde_json::Value = serde_json::from_str(&json_str).unwrap();

    assert!(parsed.is_array());
    let arr = parsed.as_array().unwrap();
    assert_eq!(arr.len(), 1);
    assert!(arr[0]["_stream"].is_object());
    assert_eq!(arr[0]["_stream"]["origin"], serde_json::Value::Null);
    assert_eq!(arr[0]["_stream"]["placeholder"], true);
}

// ── 19. poll_ready error does not poison multicast ──────────────────

#[tokio::test]
async fn test_poll_ready_error_does_not_poison_multicast() {
    use std::sync::atomic::AtomicUsize;

    // A service whose poll_ready always returns Ready(Err(...)).
    #[derive(Clone)]
    struct FailingReadyService {
        call_count: Arc<AtomicUsize>,
    }

    impl Service<Exchange> for FailingReadyService {
        type Response = Exchange;
        type Error = CamelError;
        type Future = Pin<Box<dyn Future<Output = Result<Exchange, CamelError>> + Send>>;

        fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
            Poll::Ready(Err(CamelError::ProcessorError("ready-fail".into())))
        }

        fn call(&mut self, exchange: Exchange) -> Self::Future {
            self.call_count.fetch_add(1, Ordering::SeqCst);
            Box::pin(async move { Ok(exchange) })
        }
    }

    let call_count = Arc::new(AtomicUsize::new(0));
    let failing_svc = FailingReadyService {
        call_count: Arc::clone(&call_count),
    };
    let failing_boxed: BoxProcessor = BoxProcessor::new(failing_svc);

    let endpoints = vec![uppercase_processor(), failing_boxed, uppercase_processor()];

    let config = MulticastConfig::new()
        .stop_on_exception(false)
        .aggregation(MulticastStrategy::LastWins);
    let mut svc = MulticastService::new(endpoints, config).unwrap();

    // With the fix, poll_ready returns Ready(Ok(())) so ready() succeeds.
    // The failing endpoint's readiness error is handled inside call(),
    // where stop_on_exception=false means processing continues.
    let result = svc.ready().await;
    assert!(
        result.is_ok(),
        "poll_ready should not fail-fast; got error: {:?}",
        result.err()
    );

    let result = result.unwrap().call(make_exchange("hello")).await;
    assert!(
        result.is_ok(),
        "LastWins with last endpoint succeeding should return Ok; got: {:?}",
        result.err()
    );

    // The successful endpoints (0 and 2) should have been processed.
    // The failing endpoint (1) should have had its call bypassed due to
    // readiness failure inside process_sequential, but the others continue.
    assert_eq!(
        result.unwrap().input.body.as_text(),
        Some("HELLO"),
        "last successful endpoint should have uppercased"
    );

    // Verify the failing endpoint's call() was never invoked
    assert_eq!(call_count.load(Ordering::SeqCst), 0);
}

#[tokio::test]
async fn test_multicast_collect_all_converts_bytes_xml_and_empty() {
    let endpoints = vec![
        BoxProcessor::from_fn(|mut ex: Exchange| {
            Box::pin(async move {
                ex.input.body = Body::Bytes(vec![65, 66].into());
                Ok(ex)
            })
        }),
        BoxProcessor::from_fn(|mut ex: Exchange| {
            Box::pin(async move {
                ex.input.body = Body::Xml("<a/>".to_string());
                Ok(ex)
            })
        }),
        BoxProcessor::from_fn(|mut ex: Exchange| {
            Box::pin(async move {
                ex.input.body = Body::Empty;
                Ok(ex)
            })
        }),
    ];

    let config = MulticastConfig::new().aggregation(MulticastStrategy::CollectAll);
    let mut svc = MulticastService::new(endpoints, config).unwrap();
    let result = svc
        .ready()
        .await
        .unwrap()
        .call(make_exchange("x"))
        .await
        .unwrap();

    match result.input.body {
        Body::Json(v) => assert_eq!(v, serde_json::json!(["AB", "<a/>", null])),
        _ => panic!("expected json"),
    }
}