//! `A2aDispatcherBuilder` shape tests — mirrors the
//! `McpServerBuilder` happy/cancel/build_arc surface.
//!
//! The builder closed an architecture-gate asymmetry where A2A used
//! post-construction `with_cancel_subscription(&Arc<Self>)` while MCP
//! used a builder flag + `build_arc` finalizer. These tests pin the
//! contract so the asymmetry cannot regress.

#![cfg(feature = "http")]

use klieo_auth_common::{AllowAnonymous, Authenticator};
use klieo_a2a::handler::{A2aHandler, EchoHandler};
use klieo_a2a::server::{A2aDispatcher, A2aDispatcherBuilder};
use klieo_bus_memory::MemoryBus;
use klieo_core::Pubsub;
use std::sync::Arc;
use std::time::Duration;
use tokio_util::sync::CancellationToken;

const REPLICA_ATTACH_DELAY: Duration = Duration::from_millis(50);
const CANCEL_DEADLINE: Duration = Duration::from_millis(500);
const CUSTOM_PUBLISH_CONCURRENCY: usize = 7;

fn handler() -> Arc<dyn A2aHandler> {
    Arc::new(EchoHandler::default())
}

fn authenticator() -> Arc<dyn Authenticator> {
    Arc::new(AllowAnonymous)
}

#[tokio::test(flavor = "current_thread")]
async fn builder_happy_path_constructs_dispatcher() {
    let bus = MemoryBus::new();
    let pubsub: Arc<dyn Pubsub> = bus.pubsub.clone();
    let dispatcher = A2aDispatcher::builder()
        .handler(handler())
        .authenticator(authenticator())
        .pubsub(pubsub.clone())
        .build()
        .expect("test builder");
    assert!(
        Arc::ptr_eq(dispatcher.pubsub(), &pubsub),
        "builder must thread the configured pubsub through to the dispatcher",
    );
}

#[tokio::test(flavor = "current_thread")]
async fn builder_build_returns_cancel_requires_arc_error() {
    let bus = MemoryBus::new();
    let pubsub: Arc<dyn Pubsub> = bus.pubsub.clone();
    let result = A2aDispatcherBuilder::default()
        .handler(handler())
        .authenticator(authenticator())
        .pubsub(pubsub)
        .with_cancel_subscription()
        .build();
    assert!(
        matches!(result, Err(klieo_a2a::A2aBuilderError::CancelRequiresArc)),
        "build() with cancel subscription must return CancelRequiresArc error",
    );
}

#[tokio::test(flavor = "current_thread")]
async fn builder_without_handler_returns_missing_handler_error() {
    let bus = MemoryBus::new();
    let pubsub: Arc<dyn Pubsub> = bus.pubsub.clone();
    let result = A2aDispatcherBuilder::default()
        .authenticator(authenticator())
        .pubsub(pubsub)
        .build();
    assert!(matches!(
        result,
        Err(klieo_a2a::A2aBuilderError::MissingHandler)
    ));
}

#[tokio::test(flavor = "current_thread")]
async fn builder_without_authenticator_returns_missing_authenticator_error() {
    let bus = MemoryBus::new();
    let pubsub: Arc<dyn Pubsub> = bus.pubsub.clone();
    let result = A2aDispatcherBuilder::default()
        .handler(handler())
        .pubsub(pubsub)
        .build();
    assert!(
        matches!(
            result,
            Err(klieo_a2a::A2aBuilderError::MissingAuthenticator)
        ),
        "build() without authenticator must return MissingAuthenticator",
    );
}

#[tokio::test(flavor = "current_thread")]
async fn builder_without_pubsub_returns_missing_pubsub_error() {
    let result = A2aDispatcherBuilder::default()
        .handler(handler())
        .authenticator(authenticator())
        .build();
    assert!(
        matches!(result, Err(klieo_a2a::A2aBuilderError::MissingPubsub)),
        "build() without pubsub must return MissingPubsub",
    );
}

#[tokio::test(flavor = "current_thread")]
async fn builder_build_arc_with_cancel_subscription_dispatches_cancel() {
    let bus = MemoryBus::new();
    let pubsub: Arc<dyn Pubsub> = bus.pubsub.clone();

    let dispatcher_a = A2aDispatcher::builder()
        .handler(handler())
        .authenticator(authenticator())
        .pubsub(pubsub.clone())
        .with_cancel_subscription()
        .build_arc()
        .expect("test builder");
    let dispatcher_b = A2aDispatcher::builder()
        .handler(handler())
        .authenticator(authenticator())
        .pubsub(pubsub)
        .with_cancel_subscription()
        .build_arc()
        .expect("test builder");
    tokio::time::sleep(REPLICA_ATTACH_DELAY).await;

    let token = CancellationToken::new();
    dispatcher_a
        .cancel_registry()
        .register("t-1".into(), token.clone());

    dispatcher_b.publish_cancel("t-1").await.unwrap();

    tokio::time::timeout(CANCEL_DEADLINE, token.cancelled())
        .await
        .expect("replica A's token must fire after replica B's cancel publish");
}

#[tokio::test(flavor = "current_thread")]
async fn builder_with_publish_concurrency_applies() {
    let bus = MemoryBus::new();
    let pubsub: Arc<dyn Pubsub> = bus.pubsub.clone();
    let dispatcher = A2aDispatcher::builder()
        .handler(handler())
        .authenticator(authenticator())
        .pubsub(pubsub)
        .with_publish_concurrency(CUSTOM_PUBLISH_CONCURRENCY)
        .build()
        .expect("test builder");
    assert_eq!(
        dispatcher.publish_permits().available_permits(),
        CUSTOM_PUBLISH_CONCURRENCY,
        "with_publish_concurrency must size the dispatcher's publish semaphore",
    );
}

#[tokio::test(flavor = "current_thread")]
async fn builder_with_leader_election_threads_kv() {
    use klieo_core::KvStore;

    let bus = klieo_bus_memory::MemoryBus::new();
    let kv: Arc<dyn KvStore> = bus.kv.clone();
    let handler: Arc<dyn klieo_a2a::handler::A2aHandler> =
        Arc::new(klieo_a2a::handler::EchoHandler::default());
    let auth: Arc<dyn klieo_auth_common::Authenticator> = Arc::new(klieo_auth_common::AllowAnonymous);
    let pubsub: Arc<dyn klieo_core::Pubsub> = bus.pubsub.clone();

    let dispatcher = A2aDispatcher::builder()
        .handler(handler)
        .authenticator(auth)
        .pubsub(pubsub)
        .with_leader_election(kv)
        .build()
        .expect("test builder");

    assert!(
        dispatcher.leader_registry().is_some(),
        "with_leader_election must wire a LeaderRegistry into the dispatcher",
    );
}

#[tokio::test(flavor = "current_thread")]
async fn builder_with_tenant_binding_threads_kv() {
    let bus = klieo_bus_memory::MemoryBus::new();
    let kv: Arc<dyn klieo_core::KvStore> = bus.kv.clone();
    let handler: Arc<dyn klieo_a2a::handler::A2aHandler> =
        Arc::new(klieo_a2a::handler::EchoHandler::default());
    let auth: Arc<dyn klieo_auth_common::Authenticator> = Arc::new(klieo_auth_common::AllowAnonymous);
    let pubsub: Arc<dyn klieo_core::Pubsub> = bus.pubsub.clone();

    let dispatcher = A2aDispatcher::builder()
        .handler(handler)
        .authenticator(auth)
        .pubsub(pubsub)
        .with_tenant_binding(kv)
        .build()
        .expect("test builder");

    assert!(
        dispatcher.ownership_registry().is_some(),
        "with_tenant_binding must wire an OwnershipRegistry",
    );
}
