//! Cluster-0.25 failover-attempt-cap tunable.
//!
//! Drives `A2aDispatcher::handle_dead_leader_orphan` with a pre-seeded
//! `LeaderEntry` carrying a non-default `attempt` to verify
//! `A2aDispatcherBuilder::with_max_failover_attempts` actually changes
//! the gate's terminate threshold. The cluster-0.24
//! `failover_reinvoke.rs` suite already covers the default-cap case;
//! this file only proves the dial is wired.

#![cfg(all(feature = "http", feature = "test-fixtures"))]

use async_trait::async_trait;
use bytes::Bytes;
use klieo_a2a::auth::RequestContext;
use klieo_a2a::envelope::A2aHeaders;
use klieo_a2a::error::A2aError;
use klieo_a2a::handler::A2aHandler;
use klieo_a2a::server::A2aDispatcher;
use klieo_a2a::task_store::{A2aTaskStore, DEFAULT_BUCKET};
use klieo_a2a::types::{
    Message, Part, Role, SendMessageParams, SendMessageResult, Task, TaskStatus,
};
use klieo_auth_common::{AllowAnonymous, Authenticator, Identity};
use klieo_bus_memory::MemoryBus;
use klieo_core::resume::{KvResumeBuffer, ResumeBuffer};
use klieo_core::{KvStore, LeaderEntry, Pubsub};
use std::sync::Arc;
use uuid::Uuid;

const TASK_ID: &str = "t-failover-tunable";
const LEADER_BUCKET: &str = "klieo-leaders";
const PRINCIPAL: &str = "alice@example.com";
const CUSTOM_CAP: u32 = 5;

#[derive(Default)]
struct IdempotentEchoHandler;

#[async_trait]
impl A2aHandler for IdempotentEchoHandler {
    async fn send_message(
        &self,
        _ctx: &RequestContext,
        params: SendMessageParams,
    ) -> Result<SendMessageResult, A2aError> {
        let task = Task {
            id: TASK_ID.into(),
            contextId: "ctx-failover-tunable".into(),
            status: TaskStatus::Completed,
            artifacts: vec![],
            history: vec![params.message],
            metadata: None,
        };
        Ok(SendMessageResult::Task(task))
    }

    fn is_idempotent(&self) -> bool {
        true
    }
}

fn build_dispatcher_with_cap(
    pubsub: Arc<dyn Pubsub>,
    kv: Arc<dyn KvStore>,
    cap: u32,
) -> Arc<A2aDispatcher> {
    let handler: Arc<dyn A2aHandler> = Arc::new(IdempotentEchoHandler);
    let auth: Arc<dyn Authenticator> = Arc::new(AllowAnonymous);
    A2aDispatcher::builder()
        .handler(handler)
        .authenticator(auth)
        .pubsub(pubsub)
        .with_leader_election(kv)
        .with_max_failover_attempts(cap)
        .build_arc()
        .expect("test builder")
}

fn anon_ctx() -> RequestContext {
    RequestContext::new(
        A2aHeaders::decode_from(&klieo_core::Headers::new()),
        Some(Identity::anonymous()),
    )
}

fn build_payload(text: &str) -> Bytes {
    let params = SendMessageParams {
        message: Message {
            messageId: format!("m-{}", Uuid::new_v4()),
            contextId: Some("ctx-failover-tunable".into()),
            taskId: Some(TASK_ID.into()),
            role: Role::User,
            parts: vec![Part::Text {
                content: text.into(),
                metadata: None,
                mediaType: None,
                filename: None,
            }],
            metadata: None,
            extensions: vec![],
            referenceTaskIds: vec![],
        },
        configuration: None,
    };
    Bytes::from(serde_json::to_vec(&params).unwrap())
}

async fn seed_leader_entry(kv: &Arc<dyn KvStore>, entry: LeaderEntry) {
    let key = format!("a2a.{TASK_ID}");
    kv.put(LEADER_BUCKET, &key, entry.to_bytes().unwrap())
        .await
        .unwrap();
}

async fn seed_resume_event(buf: &Arc<KvResumeBuffer>, id: u64) {
    buf.record(TASK_ID, id, Bytes::from_static(b"prior-leader-event"))
        .await
        .unwrap();
}

#[tokio::test(flavor = "current_thread")]
async fn custom_cap_allows_higher_attempt() {
    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_cap(pubsub.clone(), kv.clone(), CUSTOM_CAP);

    let task_store = Arc::new(
        A2aTaskStore::new(kv.clone(), DEFAULT_BUCKET.into())
            .with_event_sink(dispatcher.event_sink()),
    );
    let buffer = Arc::new(KvResumeBuffer::new(kv.clone()));
    seed_resume_event(&buffer, 1).await;

    // attempt = 4 < CUSTOM_CAP = 5 — re-invoke must run.
    let entry = LeaderEntry {
        replica_id: "replica-prior".into(),
        payload: Some(build_payload("at-attempt-four")),
        attempt: 4,
        principal: Some(PRINCIPAL.into()),
    };
    seed_leader_entry(&kv, entry).await;

    let result = dispatcher
        .handle_dead_leader_orphan(&anon_ctx(), TASK_ID, &task_store, buffer.as_ref())
        .await
        .expect("re-invoke gate must not error");
    assert!(
        result.is_some(),
        "with custom cap=5 and attempt=4 the gate must re-invoke (Some stream)",
    );

    let updated = dispatcher
        .leader_registry()
        .unwrap()
        .lookup_entry(&format!("a2a.{TASK_ID}"))
        .await
        .unwrap()
        .expect("re-invoke must leave a fresh leader entry");
    assert_eq!(
        updated.attempt, 5,
        "re-invoke at attempt=4 must bump attempt to 5",
    );
}

#[tokio::test(flavor = "current_thread")]
async fn custom_cap_terminates_at_configured_threshold() {
    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_cap(pubsub.clone(), kv.clone(), CUSTOM_CAP);

    let task_store = Arc::new(
        A2aTaskStore::new(kv.clone(), DEFAULT_BUCKET.into())
            .with_event_sink(dispatcher.event_sink()),
    );
    let buffer = Arc::new(KvResumeBuffer::new(kv.clone()));
    seed_resume_event(&buffer, 1).await;

    // attempt = CUSTOM_CAP = 5 — terminate.
    let entry = LeaderEntry {
        replica_id: "replica-at-cap".into(),
        payload: Some(build_payload("at-custom-cap")),
        attempt: CUSTOM_CAP,
        principal: Some(PRINCIPAL.into()),
    };
    seed_leader_entry(&kv, entry).await;

    let err = match dispatcher
        .handle_dead_leader_orphan(&anon_ctx(), TASK_ID, &task_store, buffer.as_ref())
        .await
    {
        Err(e) => e,
        Ok(_) => panic!("attempt at custom cap must terminate (LeaderDied)"),
    };
    assert!(matches!(err, A2aError::LeaderDied { .. }));
    assert!(
        buffer.is_terminal(TASK_ID).await.unwrap(),
        "terminate path must mark the resume buffer terminal",
    );
}
