//! Wave-c C-T §6 #3 — `CommsTrustReconciler` add-failure contract.
//!
//! Ports the blocker stub that was originally scaffolded in
//! `meerkat-comms/tests/trust_reconcile_add_failure.rs` at c.0. See
//! the sibling `trust_reconcile_concurrency.rs` for the rationale on
//! why the tests live in `meerkat-runtime/tests/` instead — the
//! reconciler lives in `meerkat-runtime`, and `meerkat-comms` cannot
//! dev-dep on it without introducing a circular crate dependency.
//!
//! Invariant pinned (§6 #3): when the generated trust mutation seam
//! returns an error, the reconciler must
//!
//!   (a) surface the failure as
//!       `CommsTrustReconcileError::AddTrustFailed` with the typed
//!       peer id echoed back,
//!   (b) NOT mutate the canonical trust store — later reconciles
//!       re-read the same "empty" state, so a retry will try to add
//!       the same peer again rather than the reconciler leaking a
//!       phantom "already applied" entry.

#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]

use std::collections::BTreeSet;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};

use async_trait::async_trait;
use meerkat_core::agent::{CommsCapabilityError, CommsRuntime};
use meerkat_core::comms::{
    CommsTrustMutation, CommsTrustMutationResult, GeneratedCommsTrustAuthoritySourceKind,
    SendError, TrustedPeerDescriptor,
};
use meerkat_core::{
    PeerIngressAuthorityPhase, PeerIngressQueueSnapshot, PeerIngressRuntimeSnapshot,
};
use meerkat_runtime::comms_trust_reconcile::{CommsTrustReconcileError, CommsTrustReconciler};
use meerkat_runtime::meerkat_machine::dsl::{
    MeerkatMachineAuthority, MeerkatMachineInput, MeerkatMachineMutator, MeerkatMachineSignal,
    PeerAddress, PeerEndpoint, PeerId, PeerName, PeerSigningKey, SessionId,
};
use meerkat_runtime::protocol_comms_trust_reconcile::CommsTrustReconcileObligation;

const UUID_A: &str = "f805a14c-4089-5328-b4cb-39ede8b4464d";
const LOCAL_UUID: &str = "00000000-0000-4000-8000-000000000000";

fn endpoint(name: &str, peer_id_uuid: &str) -> PeerEndpoint {
    PeerEndpoint {
        name: PeerName(format!("ep-{name}")),
        peer_id: PeerId(peer_id_uuid.to_string()),
        address: PeerAddress(format!("inproc://{name}")),
        signing_key: PeerSigningKey([name.as_bytes()[0]; 32]),
    }
}

fn local_peer_id() -> meerkat_core::comms::PeerId {
    meerkat_core::comms::PeerId::parse(LOCAL_UUID).expect("valid local test peer id")
}

fn obligation(
    epoch: u64,
    direct_peer_endpoints: BTreeSet<PeerEndpoint>,
) -> CommsTrustReconcileObligation {
    let mut authority = MeerkatMachineAuthority::new();
    authority
        .apply_signal(MeerkatMachineSignal::Initialize)
        .expect("Initialize signal");
    MeerkatMachineMutator::apply(
        &mut authority,
        MeerkatMachineInput::RegisterSession {
            session_id: SessionId::from("trust-reconcile-add-failure-test"),
        },
    )
    .expect("RegisterSession input");
    MeerkatMachineMutator::apply(
        &mut authority,
        MeerkatMachineInput::PublishLocalEndpoint {
            endpoint: endpoint("local", LOCAL_UUID),
        },
    )
    .expect("PublishLocalEndpoint input");
    let projection_epoch = epoch.max(1);
    let mut transition = None;
    for overlay_epoch in 1..=projection_epoch {
        transition = Some(
            MeerkatMachineMutator::apply(
                &mut authority,
                MeerkatMachineInput::ApplyMobPeerOverlay {
                    epoch: overlay_epoch,
                    endpoints: direct_peer_endpoints.clone(),
                },
            )
            .expect("ApplyMobPeerOverlay input"),
        );
    }
    let transition = transition.expect("projection epoch loop produces transition");
    meerkat_runtime::protocol_comms_trust_reconcile::extract_obligations_with_freshness(
        &transition,
        meerkat_runtime::protocol_comms_trust_reconcile::PeerProjectionFreshnessAuthority::from_authority(
            Arc::new(std::sync::Mutex::new(authority)),
        ),
    )
    .into_iter()
    .next()
    .expect("generated reconcile obligation")
}

/// `CommsRuntime` mock whose next generated trust add mutation returns
/// `SendError::Unsupported`. Once fired, the flag resets so a
/// subsequent retry succeeds — this lets us assert the retry path
/// sees the peer as still absent from the applied view.
#[derive(Default)]
struct AddFailingCommsRuntime {
    fail_next_add: AtomicBool,
    successful_adds: std::sync::Mutex<Vec<TrustedPeerDescriptor>>,
}

impl std::fmt::Debug for AddFailingCommsRuntime {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("AddFailingCommsRuntime").finish()
    }
}

#[async_trait]
impl CommsRuntime for AddFailingCommsRuntime {
    async fn drain_messages(&self) -> Vec<String> {
        Vec::new()
    }

    fn inbox_notify(&self) -> Arc<tokio::sync::Notify> {
        Arc::new(tokio::sync::Notify::new())
    }

    async fn apply_trust_mutation(
        &self,
        mutation: CommsTrustMutation,
    ) -> Result<CommsTrustMutationResult, SendError> {
        match mutation {
            CommsTrustMutation::AddTrustedPeer { peer, authority } => {
                authority
                    .validate_public_add(Some(local_peer_id()), &peer)
                    .map_err(SendError::Validation)?;
                if self.fail_next_add.swap(false, Ordering::SeqCst) {
                    return Err(SendError::Unsupported("synthetic add failure".into()));
                }
                let mut successful_adds = self
                    .successful_adds
                    .lock()
                    .unwrap_or_else(std::sync::PoisonError::into_inner);
                let created = !successful_adds
                    .iter()
                    .any(|existing| existing.peer_id == peer.peer_id);
                successful_adds.push(peer);
                Ok(CommsTrustMutationResult::Added { created })
            }
            _ => Err(SendError::Unsupported(
                "test runtime only supports generated public trust adds".into(),
            )),
        }
    }

    async fn peer_ingress_runtime_snapshot(
        &self,
    ) -> Result<PeerIngressRuntimeSnapshot, CommsCapabilityError> {
        Ok(PeerIngressRuntimeSnapshot {
            self_peer_id: local_peer_id(),
            auth_required: true,
            authority_phase: PeerIngressAuthorityPhase::Received,
            trusted_peers: self.successful_add_calls(),
            submission_queue_len: 0,
            queue: PeerIngressQueueSnapshot::default(),
        })
    }

    async fn public_trusted_peer_projection_snapshot(
        &self,
    ) -> Result<Vec<TrustedPeerDescriptor>, CommsCapabilityError> {
        Ok(self.successful_add_calls())
    }

    async fn trusted_peer_projection_snapshot_for_source(
        &self,
        source_kind: GeneratedCommsTrustAuthoritySourceKind,
    ) -> Result<Vec<TrustedPeerDescriptor>, CommsCapabilityError> {
        if source_kind == GeneratedCommsTrustAuthoritySourceKind::MeerkatMachinePeerProjection {
            Ok(self.successful_add_calls())
        } else {
            Ok(Vec::new())
        }
    }
}

impl AddFailingCommsRuntime {
    fn successful_add_calls(&self) -> Vec<TrustedPeerDescriptor> {
        self.successful_adds
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .clone()
    }
}

/// Add failure surfaces a typed `AddTrustFailed` error. The
/// canonical trust store is NOT mutated — a subsequent retry with the
/// same peer set re-reads the peer as absent and tries the add again.
#[tokio::test]
async fn add_failure_surfaces_typed_error_and_preserves_canonical_store() {
    let comms = Arc::new(AddFailingCommsRuntime::default());
    comms.fail_next_add.store(true, Ordering::SeqCst);
    let reconciler = CommsTrustReconciler::new(comms.clone());

    // First reconcile: add fails.
    let err = reconciler
        .reconcile(&obligation(1, BTreeSet::from([endpoint("A", UUID_A)])))
        .await
        .expect_err("add_trust failure must surface");
    match err {
        CommsTrustReconcileError::AddTrustFailed { peer_id, .. } => {
            assert_eq!(
                peer_id, UUID_A,
                "typed error must echo the peer_id that failed",
            );
        }
        other => panic!("expected AddTrustFailed, got {other:?}"),
    }

    // Canonical store must still be empty; the only attempted add
    // errored before it could commit.
    assert!(
        comms.successful_add_calls().is_empty(),
        "no successful trust-store adds yet — the only attempt errored",
    );

    // Retry the same reconcile at the same epoch. Now the failure
    // flag is cleared; the retry succeeds because the reconciler
    // re-reads the canonical store and still sees the peer absent.
    let retry = reconciler
        .reconcile(&obligation(1, BTreeSet::from([endpoint("A", UUID_A)])))
        .await
        .expect("retry succeeds with flag cleared");
    assert_eq!(retry.applied_epoch, 1);
    assert_eq!(retry.added.len(), 1, "retry treats peer A as newly added");
    assert!(retry.removed.is_empty());
    assert_eq!(
        comms.successful_add_calls().len(),
        1,
        "trust store now carries peer A after the retry",
    );
    let trusted_after_retry = comms.successful_add_calls();
    assert_eq!(trusted_after_retry.len(), 1);
    assert_eq!(trusted_after_retry[0].peer_id.to_string(), UUID_A);
}
