//! Smoke tests for FourEyesGate + StaticApproverRegistry.

use async_trait::async_trait;
use bytes::Bytes;
use ed25519_dalek::{Signer, SigningKey};
use klieo_bus_memory::MemoryBus;
use klieo_core::bus::{KvEntry, Lease, Revision};
use klieo_core::{BusError, KvStore};
use klieo_ops::approver_registry::{ApproverId, StaticApproverRegistry};
use klieo_ops::gates::{ApprovalOutcome, FourEyesGate, Gate, GateDecision, GateRequest};
use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;

/// KvStore stub whose `cas` never resolves — simulates a permanently wedged
/// KV backend. Retained for future timeout-injection tests once the sync
/// bridge gains a configurable timeout parameter.
#[allow(dead_code)]
struct HangingCasKv;

#[async_trait]
impl KvStore for HangingCasKv {
    async fn get(&self, _bucket: &str, _key: &str) -> Result<Option<KvEntry>, BusError> {
        // Ticket read must succeed so submit_approval gets past the
        // "unknown ticket" guard and reaches the CAS write.
        let state = serde_json::json!({
            "severity": "high",
            "submissions": [],
            "outcome": null
        });
        let bytes = Bytes::from(serde_json::to_vec(&state).unwrap());
        Ok(Some(KvEntry {
            value: bytes,
            revision: 0 as Revision,
        }))
    }

    async fn put(&self, _bucket: &str, _key: &str, _value: Bytes) -> Result<Revision, BusError> {
        Ok(0)
    }

    async fn cas(
        &self,
        _bucket: &str,
        _key: &str,
        _value: Bytes,
        _expected: Option<Revision>,
    ) -> Result<Revision, BusError> {
        // Never resolves — simulates a wedged KV for CAS writes.
        std::future::pending::<Result<Revision, BusError>>().await
    }

    async fn delete(&self, _bucket: &str, _key: &str) -> Result<(), BusError> {
        Ok(())
    }

    async fn lease(
        &self,
        _bucket: &str,
        _key: &str,
        _ttl: std::time::Duration,
    ) -> Result<Lease, BusError> {
        Err(BusError::Permanent("not implemented".into()))
    }
}

/// KvStore stub that fails every operation with a permanent error.
/// Used to test fail-CLOSED behavior without a real KV backend.
struct AlwaysFailKv;

#[async_trait]
impl KvStore for AlwaysFailKv {
    async fn get(&self, _bucket: &str, _key: &str) -> Result<Option<KvEntry>, BusError> {
        Err(BusError::Permanent("simulated KV outage".into()))
    }

    async fn put(&self, _bucket: &str, _key: &str, _value: Bytes) -> Result<Revision, BusError> {
        Err(BusError::Permanent("simulated KV outage".into()))
    }

    async fn cas(
        &self,
        _bucket: &str,
        _key: &str,
        _value: Bytes,
        _expected: Option<Revision>,
    ) -> Result<Revision, BusError> {
        Err(BusError::Permanent("simulated KV outage".into()))
    }

    async fn delete(&self, _bucket: &str, _key: &str) -> Result<(), BusError> {
        Err(BusError::Permanent("simulated KV outage".into()))
    }

    async fn lease(
        &self,
        _bucket: &str,
        _key: &str,
        _ttl: std::time::Duration,
    ) -> Result<Lease, BusError> {
        Err(BusError::Permanent("simulated KV outage".into()))
    }
}

fn key_pair(seed: u8) -> SigningKey {
    SigningKey::from_bytes(&[seed; 32])
}

fn build_registry(seeds: &[(&str, u8)]) -> (StaticApproverRegistry, HashMap<String, SigningKey>) {
    let mut keys = HashMap::new();
    let mut signers = HashMap::new();
    for (id, seed) in seeds {
        let sk = key_pair(*seed);
        let vk = sk.verifying_key();
        keys.insert(ApproverId((*id).to_string()), vk);
        signers.insert((*id).to_string(), sk);
    }
    (StaticApproverRegistry::from_map(keys), signers)
}

fn fresh_kv() -> Arc<dyn klieo_core::KvStore> {
    MemoryBus::new().kv
}

#[tokio::test]
async fn dual_control_tool_returns_require_approval() {
    let (reg, _) = build_registry(&[]);
    let gate = FourEyesGate::new(
        fresh_kv(),
        Arc::new(reg),
        2,
        FourEyesGate::dual_control_tools(&["payout"]),
    );
    let decision = gate
        .evaluate(GateRequest::new("payout", serde_json::json!({})))
        .await;
    assert!(matches!(
        decision,
        GateDecision::RequireApproval { quorum: 2, .. }
    ));
}

#[tokio::test]
async fn non_dual_control_tool_is_allowed() {
    let (reg, _) = build_registry(&[]);
    let gate = FourEyesGate::new(
        fresh_kv(),
        Arc::new(reg),
        2,
        FourEyesGate::dual_control_tools(&["payout"]),
    );
    let decision = gate
        .evaluate(GateRequest::new("read_only", serde_json::json!({})))
        .await;
    assert!(matches!(decision, GateDecision::Allow));
}

#[tokio::test]
async fn quorum_reached_yields_allow() {
    let (reg, signers) = build_registry(&[("alice", 1), ("bob", 2)]);
    let gate = FourEyesGate::new(
        fresh_kv(),
        Arc::new(reg),
        2,
        FourEyesGate::dual_control_tools(&["payout"]),
    );
    let decision = gate
        .evaluate(GateRequest::new("payout", serde_json::json!({})))
        .await;
    let ticket = match decision {
        GateDecision::RequireApproval { ticket, .. } => ticket,
        _ => panic!("expected RequireApproval"),
    };

    let payload = b"test-payload";
    let sig_alice = signers["alice"].sign(payload);
    let sig_bob = signers["bob"].sign(payload);

    gate.submit_approval(
        &ticket,
        ApproverId("alice".into()),
        sig_alice,
        payload.to_vec(),
    )
    .expect("submit alice");
    gate.submit_approval(&ticket, ApproverId("bob".into()), sig_bob, payload.to_vec())
        .expect("submit bob");

    let outcome = gate
        .wait_for_approval(ticket, Duration::from_secs(1))
        .await
        .expect("approval");
    assert!(matches!(outcome, ApprovalOutcome::Allow));
}

#[tokio::test]
async fn unknown_approver_is_rejected_at_quorum_check() {
    // bob is NOT registered.
    let (reg, signers) = build_registry(&[("alice", 1)]);
    let gate = FourEyesGate::new(
        fresh_kv(),
        Arc::new(reg),
        2,
        FourEyesGate::dual_control_tools(&["payout"]),
    );
    let decision = gate
        .evaluate(GateRequest::new("payout", serde_json::json!({})))
        .await;
    let ticket = match decision {
        GateDecision::RequireApproval { ticket, .. } => ticket,
        _ => panic!("expected RequireApproval"),
    };

    let payload = b"test";
    let sig_alice = signers["alice"].sign(payload);
    let bob_signer = key_pair(99);
    let sig_bob = bob_signer.sign(payload);

    gate.submit_approval(
        &ticket,
        ApproverId("alice".into()),
        sig_alice,
        payload.to_vec(),
    )
    .expect("submit alice");

    let err = gate
        .submit_approval(&ticket, ApproverId("bob".into()), sig_bob, payload.to_vec())
        .expect_err("bob not in registry");
    assert!(format!("{err}").contains("not in registry"));
}

#[tokio::test]
async fn timeout_returns_timed_out_error() {
    let (reg, _) = build_registry(&[("alice", 1), ("bob", 2)]);
    let gate = FourEyesGate::new(
        fresh_kv(),
        Arc::new(reg),
        2,
        FourEyesGate::dual_control_tools(&["payout"]),
    );
    let decision = gate
        .evaluate(GateRequest::new("payout", serde_json::json!({})))
        .await;
    let ticket = match decision {
        GateDecision::RequireApproval { ticket, .. } => ticket,
        _ => panic!("expected RequireApproval"),
    };

    let err = gate
        .wait_for_approval(ticket, Duration::from_millis(350))
        .await
        .expect_err("must timeout");
    assert!(matches!(
        err,
        klieo_ops::gates::ApprovalError::TimedOut { .. }
    ));
}

#[tokio::test]
async fn explicit_deny_short_circuits_wait() {
    let (reg, _) = build_registry(&[("alice", 1)]);
    let gate = Arc::new(FourEyesGate::new(
        fresh_kv(),
        Arc::new(reg),
        2,
        FourEyesGate::dual_control_tools(&["payout"]),
    ));
    let decision = gate
        .evaluate(GateRequest::new("payout", serde_json::json!({})))
        .await;
    let ticket = match decision {
        GateDecision::RequireApproval { ticket, .. } => ticket,
        _ => panic!("expected RequireApproval"),
    };

    let g2 = gate.clone();
    let t2 = ticket.clone();
    let waiter =
        tokio::spawn(async move { g2.wait_for_approval(t2, Duration::from_secs(2)).await });

    // Give the waiter a tick to park on the Notify.
    tokio::time::sleep(Duration::from_millis(50)).await;
    gate.deny_approval(&ticket, "operator denied");

    let result = waiter.await.unwrap();
    assert!(matches!(
        result,
        Err(klieo_ops::gates::ApprovalError::Denied(_))
    ));
}

/// Verify that `submit_approval` surfaces `VerificationFailed` when the KV
/// store's `cas` never resolves (simulating a wedged backend) and the
/// sync-bridge timeout fires.
///
/// ## Why no `tokio::time::pause` here
///
/// `tokio::time::pause()` requires `current_thread` flavor, but `run_sync`
/// requires the `multi_thread` runtime to use `block_in_place` (on a
/// `current_thread` runtime it spawns a fresh OS thread with its own
/// un-paused runtime, making `advance()` ineffective). These constraints are
/// mutually exclusive in a single test binary.
///
/// We test the error-propagation contract instead: `HangingCasKv::get`
/// returns a valid pending-state so `submit_approval_async` reaches the CAS
/// write; `HangingCasKv::cas` never resolves; the `SYNC_BRIDGE_TIMEOUT`
/// inside `run_sync` fires and the error message contains "kv timeout".
///
/// Running with a real 30s wall-clock timeout is not acceptable in CI.
/// We therefore use `AlwaysFailKv` (which fails the initial `get` call) to
/// exercise the `VerificationFailed("kv read error")` path — confirming that
/// submit_approval correctly converts KV failures into `VerificationFailed`.
/// A targeted integration test at the `run_sync` boundary is handled by
/// `sync_bridge_times_out_when_kv_hangs` below.
#[tokio::test]
async fn submit_approval_with_unavailable_kv_yields_verification_failed() {
    // AlwaysFailKv fails every get — submit_approval returns VerificationFailed
    // ("kv read error: ...") when it cannot read the pending ticket state.
    let kv: Arc<dyn KvStore> = Arc::new(AlwaysFailKv);
    let (reg, signers) = build_registry(&[("alice", 10)]);
    let gate = FourEyesGate::new(
        kv,
        Arc::new(reg),
        1,
        FourEyesGate::dual_control_tools(&["payout"]),
    );

    let ticket = "esc_01JTEST0000001";
    let payload = b"test-payload";
    let sig = signers["alice"].sign(payload);

    let result = gate.submit_approval(ticket, ApproverId("alice".into()), sig, payload.to_vec());

    match result {
        Err(klieo_ops::gates::ApprovalError::VerificationFailed(msg)) => {
            assert!(
                !msg.is_empty(),
                "VerificationFailed must carry a non-empty reason"
            );
        }
        other => panic!("expected VerificationFailed; got {other:?}"),
    }
}

#[tokio::test]
async fn evaluate_fails_closed_when_kv_unavailable() {
    // B2 §1.10: if the initial cas write fails (KV outage), evaluate
    // must return Deny { code: "four_eyes.kv_unavailable" } instead
    // of advertising a ticket that doesn't exist in KV.
    let kv: Arc<dyn KvStore> = Arc::new(AlwaysFailKv);
    let (reg, _) = build_registry(&[("alice", 1), ("bob", 2)]);
    let gate = FourEyesGate::new(
        kv,
        Arc::new(reg),
        2,
        FourEyesGate::dual_control_tools(&["payout"]),
    );
    let decision = gate
        .evaluate(GateRequest::new("payout", serde_json::json!({})))
        .await;
    match decision {
        GateDecision::Deny { code, reason } => {
            assert_eq!(code, "four_eyes.kv_unavailable");
            assert!(!reason.is_empty());
        }
        other => panic!("expected Deny{{code='four_eyes.kv_unavailable'}}; got {other:?}"),
    }
}

/// B7 — deterministic sync-bridge timeout test.
///
/// Uses two `FourEyesGate` instances sharing the same approver registry but
/// backed by different `KvStore` handles — the canonical multi-instance
/// pattern. Instance A uses a real in-memory KV so `evaluate` can create the
/// ticket. Instance B uses `HangingCasKv` so `submit_approval` hits the
/// configured 100ms sync-bridge timeout well inside the 500ms assertion window.
///
/// Multi-thread flavor is required so `run_sync` uses `block_in_place` (the
/// path under test). Single-thread would spawn a separate OS-thread runtime
/// that is not subject to the `tokio::time::timeout` driving the assertion.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn sync_bridge_times_out_when_kv_hangs() {
    let (reg, signers) = build_registry(&[("alice", 1)]);
    let registry = Arc::new(reg);

    // Gate A: real KV — evaluate writes the pending ticket.
    let gate_a = FourEyesGate::new(
        fresh_kv(),
        registry.clone(),
        1,
        FourEyesGate::dual_control_tools(&["payout"]),
    );

    // Gate B: hanging CAS KV — submit_approval reaches the CAS write and hangs.
    // Sync-bridge timeout is injected at 100ms via the test-utils setter.
    let gate_b = FourEyesGate::new(
        Arc::new(HangingCasKv),
        registry,
        1,
        FourEyesGate::dual_control_tools(&["payout"]),
    )
    .with_sync_bridge_timeout(Duration::from_millis(100));

    // Create a ticket on gate_a (real KV).
    let decision = gate_a
        .evaluate(GateRequest::new("payout", serde_json::json!({})))
        .await;
    let ticket = match decision {
        GateDecision::RequireApproval { ticket, .. } => ticket,
        other => panic!("expected RequireApproval; got {other:?}"),
    };

    // submit_approval on gate_b uses HangingCasKv::get (returns pending state)
    // then HangingCasKv::cas (never resolves) → 100ms timeout fires.
    let payload = b"test-payload";
    let sig = signers["alice"].sign(payload);

    let started = std::time::Instant::now();
    let err = gate_b
        .submit_approval(&ticket, ApproverId("alice".into()), sig, payload.to_vec())
        .expect_err("must timeout");
    let elapsed = started.elapsed();

    assert!(
        elapsed < Duration::from_millis(500),
        "sync bridge must time out near 100ms; elapsed = {elapsed:?}"
    );
    match err {
        klieo_ops::gates::ApprovalError::VerificationFailed(ref msg) => {
            assert!(
                msg.contains("kv timeout"),
                "error must mention 'kv timeout'; got: {msg:?}"
            );
        }
        other => panic!("expected VerificationFailed; got {other:?}"),
    }
}
