//! Backpressure semantics for [`TaskEventSink::send`].
//!
//! When the dispatcher's publish-concurrency semaphore is saturated,
//! `send` MUST drop the event with `Ok(())` (best-effort fanout) and
//! a `tracing::warn` rather than enqueueing more work onto the bus
//! task. Building the dispatcher with `with_publish_concurrency(0)`
//! exercises the saturation branch deterministically — every send
//! finds zero permits available, so subscribers receive nothing.

use klieo_a2a::handler::EchoHandler;
use klieo_a2a::server::{A2aDispatcher, TaskEvent};
use klieo_a2a::types::TaskStatus;
use klieo_auth_common::{AllowAnonymous, Authenticator};
use klieo_bus_memory::MemoryBus;
use klieo_core::{DurableName, Pubsub};
use std::sync::Arc;
use std::time::Duration;
use tokio_stream::StreamExt as _;

const SATURATED_PERMITS: usize = 0;
const EVENT_COUNT: usize = 5;
const SUBSCRIBE_TIMEOUT: Duration = Duration::from_millis(100);
const TASK_ID: &str = "t-backpressure";

#[tokio::test]
async fn publish_backpressure_drops_on_saturation() {
    let bus = MemoryBus::new();
    let pubsub: Arc<dyn Pubsub> = bus.pubsub.clone();
    let handler: Arc<dyn klieo_a2a::handler::A2aHandler> = Arc::new(EchoHandler::default());
    let auth: Arc<dyn Authenticator> = Arc::new(AllowAnonymous);

    // Zero-permit semaphore: every send hits the saturation branch.
    let dispatcher = A2aDispatcher::new(handler, auth, pubsub.clone())
        .with_publish_concurrency(SATURATED_PERMITS);

    // Subscribe to the per-task subject before any sends so we'd
    // observe any publish that actually made it through.
    let durable = DurableName::new("backpressure-test-subscriber");
    let subject = format!("klieo.a2a.task.{TASK_ID}");
    let mut stream = pubsub
        .subscribe(&subject, durable)
        .await
        .expect("subscribe");

    let sink = dispatcher.event_sink();
    for n in 0..EVENT_COUNT {
        let event = TaskEvent::new(TASK_ID.to_string(), TaskStatus::Working, None, false)
            .with_event_id(n as u64);
        // Best-effort fanout: every send must return Ok despite saturation.
        sink.send(event)
            .await
            .expect("saturation must surface as Ok(()), not error");
    }

    // Subscriber sees nothing within the timeout — every event dropped.
    let received = tokio::time::timeout(SUBSCRIBE_TIMEOUT, stream.next()).await;
    assert!(
        received.is_err(),
        "subscriber received an event but every publish should have been dropped on saturation",
    );
}
