//! Cluster-0.25 KV reaper end-to-end wiring through `A2aHttpServer`.
//!
//! `A2aDispatcherBuilder::with_kv_reaper(interval)` stores the
//! interval on the dispatcher; the actual scan task is spawned by
//! [`A2aHttpServer::with_resume_buffer`] which has access to both the
//! dispatcher's leader/ownership KV references AND the resume buffer
//! the reaper queries for terminal liveness.
//!
//! This test seeds a `klieo-leaders` entry, marks the corresponding
//! resume-buffer stream terminal, then waits long enough for one
//! reaper scan to fire and asserts the entry has been evicted.

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

use bytes::Bytes;
use klieo_a2a::handler::{A2aHandler, EchoHandler};
use klieo_a2a::http::A2aHttpServer;
use klieo_a2a::server::A2aDispatcher;
use klieo_a2a::task_store::{A2aTaskStore, DEFAULT_BUCKET};
use klieo_auth_common::{AllowAnonymous, Authenticator};
use klieo_bus_memory::MemoryBus;
use klieo_core::resume::{KvResumeBuffer, ResumeBuffer};
use klieo_core::{KvStore, Pubsub};
use std::sync::Arc;
use std::time::Duration;
use tokio_util::sync::CancellationToken;

const LEADER_BUCKET: &str = "klieo-leaders";
const TASK_ID: &str = "t-orphan-reaper";
const SCAN_INTERVAL: Duration = Duration::from_millis(50);
// Two full scans + buffer should comfortably evict in CI.
const POST_SCAN_WAIT: Duration = Duration::from_millis(250);

fn build_dispatcher_with_reaper(
    pubsub: Arc<dyn Pubsub>,
    kv: Arc<dyn KvStore>,
) -> Arc<A2aDispatcher> {
    let handler: Arc<dyn A2aHandler> = Arc::new(EchoHandler::default());
    let auth: Arc<dyn Authenticator> = Arc::new(AllowAnonymous);
    A2aDispatcher::builder()
        .handler(handler)
        .authenticator(auth)
        .pubsub(pubsub)
        .with_leader_election(kv)
        .with_kv_reaper(SCAN_INTERVAL)
        .build_arc()
        .expect("test builder")
}

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

    let dispatcher = build_dispatcher_with_reaper(pubsub.clone(), kv.clone());
    assert_eq!(dispatcher.kv_reaper_interval(), Some(SCAN_INTERVAL));

    // Pre-seed leader entry the reaper should evict.
    let key = format!("a2a.{TASK_ID}");
    kv.put(LEADER_BUCKET, &key, Bytes::from_static(b"replica-dead"))
        .await
        .unwrap();
    assert!(
        kv.get(LEADER_BUCKET, &key).await.unwrap().is_some(),
        "seed must land before the reaper starts",
    );

    // Resume buffer must report terminal for `TASK_ID` so the reaper
    // sees the entry as an orphan candidate.
    let buffer = Arc::new(KvResumeBuffer::new(kv.clone()));
    buffer
        .record(TASK_ID, 1, Bytes::from_static(b"final"))
        .await
        .unwrap();
    buffer.close(TASK_ID).await.unwrap();
    assert!(
        buffer.is_terminal(TASK_ID).await.unwrap(),
        "buffer.close must flip the stream to terminal",
    );

    // Build server — `with_resume_buffer` is the wiring point that
    // spawns the reaper task. Hold the server in scope so the
    // `_kv_reaper` field stays alive during the scan window.
    let task_store = Arc::new(
        A2aTaskStore::new(kv.clone(), DEFAULT_BUCKET.into())
            .with_event_sink(dispatcher.event_sink()),
    );
    let server = Arc::new(
        A2aHttpServer::new(dispatcher.clone(), task_store, CancellationToken::new())
            .with_resume_buffer(buffer.clone() as Arc<dyn ResumeBuffer>),
    );

    // One scan interval + slack for the eviction round-trip.
    tokio::time::sleep(POST_SCAN_WAIT).await;

    assert!(
        kv.get(LEADER_BUCKET, &key).await.unwrap().is_none(),
        "reaper must evict the leader entry whose buffer reports terminal",
    );

    drop(server);
}

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

    let handler: Arc<dyn A2aHandler> = Arc::new(EchoHandler::default());
    let auth: Arc<dyn Authenticator> = Arc::new(AllowAnonymous);
    let dispatcher = A2aDispatcher::builder()
        .handler(handler)
        .authenticator(auth)
        .pubsub(pubsub)
        .with_leader_election(kv.clone())
        .build_arc()
        .expect("test builder");
    assert!(dispatcher.kv_reaper_interval().is_none());

    let key = format!("a2a.{TASK_ID}");
    kv.put(LEADER_BUCKET, &key, Bytes::from_static(b"replica-dead"))
        .await
        .unwrap();

    let buffer = Arc::new(KvResumeBuffer::new(kv.clone()));
    buffer
        .record(TASK_ID, 1, Bytes::from_static(b"final"))
        .await
        .unwrap();
    buffer.close(TASK_ID).await.unwrap();

    let task_store = Arc::new(
        A2aTaskStore::new(kv.clone(), DEFAULT_BUCKET.into())
            .with_event_sink(dispatcher.event_sink()),
    );
    let _server = Arc::new(
        A2aHttpServer::new(dispatcher.clone(), task_store, CancellationToken::new())
            .with_resume_buffer(buffer as Arc<dyn ResumeBuffer>),
    );

    tokio::time::sleep(POST_SCAN_WAIT).await;

    assert!(
        kv.get(LEADER_BUCKET, &key).await.unwrap().is_some(),
        "without with_kv_reaper the entry must NOT be evicted",
    );
}
