//! C-H3 — admission-time trust check closes the classify→admit TOCTOU window.
//!
//! Before the fix, `PreparedIngressItem.trusted_sender` was computed at
//! classification time (T0), carried through the queue lock, and used as
//! the admission oracle. A concurrent trust-revoke between T0 and the
//! queue-locked admission at T2 would be ignored and an envelope from a
//! now-untrusted sender would be admitted.
//!
//! The fix moves the authoritative trust check inside the queue-lock
//! scope: `admit_peer_receive` re-reads `trusted_peers` against the
//! envelope's actual `from` pubkey, so the admission decision reflects
//! trust-state as of T2. These tests pin the invariant.

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

use meerkat_comms::identity::{Keypair, Signature};
use meerkat_comms::runtime::comms_runtime::CommsRuntime;
use meerkat_comms::types::{Envelope, InboxItem, MessageKind};
use meerkat_comms::{AdmissionOutcome, DropReason};
use meerkat_core::comms::{
    CommsTrustMutation, CommsTrustMutationResult, PeerAddress, PeerName, PeerTransport,
    TrustedPeerDescriptor,
};
use meerkat_core::{PeerIngressAuthDecision, PeerIngressAuthExemption, SUPERVISOR_BRIDGE_INTENT};
use std::sync::{Arc, Mutex};
use uuid::Uuid;

type TestMachineAuthority =
    Arc<Mutex<meerkat_runtime::meerkat_machine::dsl::MeerkatMachineAuthority>>;

struct TestPeerCommsAuthority {
    authority: TestMachineAuthority,
    dsl: Arc<meerkat_runtime::HandleDslAuthority>,
}

impl TestPeerCommsAuthority {
    fn install(runtime: &CommsRuntime, session_id: &str) -> Self {
        let authority = Arc::new(Mutex::new(
            meerkat_runtime::meerkat_machine::dsl::MeerkatMachineAuthority::new(),
        ));
        let local_endpoint = local_endpoint_for(runtime);
        {
            let mut guard = authority
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            guard
                .apply_signal(
                    meerkat_runtime::meerkat_machine::dsl::MeerkatMachineSignal::Initialize,
                )
                .expect("Initialize signal");
            meerkat_runtime::meerkat_machine::dsl::MeerkatMachineMutator::apply(
                &mut *guard,
                meerkat_runtime::meerkat_machine::dsl::MeerkatMachineInput::RegisterSession {
                    session_id: meerkat_runtime::meerkat_machine::dsl::SessionId::from(session_id),
                },
            )
            .expect("RegisterSession input");
            meerkat_runtime::meerkat_machine::dsl::MeerkatMachineMutator::apply(
                &mut *guard,
                meerkat_runtime::meerkat_machine::dsl::MeerkatMachineInput::PublishLocalEndpoint {
                    endpoint: local_endpoint,
                },
            )
            .expect("PublishLocalEndpoint input");
            meerkat_runtime::meerkat_machine::dsl::MeerkatMachineMutator::apply(
                &mut *guard,
                meerkat_runtime::meerkat_machine::dsl::MeerkatMachineInput::Prepare {
                    session_id: meerkat_runtime::meerkat_machine::dsl::SessionId::from(session_id),
                    run_id: meerkat_runtime::meerkat_machine::dsl::RunId::from(format!(
                        "{session_id}-ingress-run"
                    )),
                },
            )
            .expect("Prepare input");
        }

        let dsl = Arc::new(meerkat_runtime::HandleDslAuthority::from_shared(
            Arc::clone(&authority),
        ));
        meerkat_runtime::RuntimePeerCommsHandle::install_generated_on(Arc::clone(&dsl), runtime)
            .expect("install generated peer-comms handle");
        Self { authority, dsl }
    }

    fn reinstall(&self, runtime: &CommsRuntime) {
        meerkat_runtime::RuntimePeerCommsHandle::install_generated_on(
            Arc::clone(&self.dsl),
            runtime,
        )
        .expect("install generated peer-comms handle");
    }

    fn add_authority(
        &self,
        peer: &TrustedPeerDescriptor,
    ) -> meerkat_core::comms::CommsTrustMutationAuthority {
        let endpoint = meerkat_runtime::meerkat_machine::dsl::PeerEndpoint::from(peer);
        let transition = {
            let mut guard = self
                .authority
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            meerkat_runtime::meerkat_machine::dsl::MeerkatMachineMutator::apply(
                &mut *guard,
                meerkat_runtime::meerkat_machine::dsl::MeerkatMachineInput::AddDirectPeerEndpoint {
                    endpoint: endpoint.clone(),
                },
            )
            .expect("AddDirectPeerEndpoint input")
        };
        let mut obligations =
            meerkat_runtime::protocol_comms_trust_reconcile::extract_obligations_with_freshness(
                &transition,
                meerkat_runtime::protocol_comms_trust_reconcile::PeerProjectionFreshnessAuthority::from_authority(
                    Arc::clone(&self.authority),
                ),
            );
        let obligation = obligations.pop().expect("generated reconcile obligation");
        meerkat_runtime::protocol_comms_trust_reconcile::authority_for_endpoint(
            &obligation,
            &endpoint,
        )
        .expect("generated peer projection add authority")
    }

    fn remove_authority(
        &self,
        peer: &TrustedPeerDescriptor,
    ) -> meerkat_core::comms::CommsTrustMutationAuthority {
        let endpoint = meerkat_runtime::meerkat_machine::dsl::PeerEndpoint::from(peer);
        let transition = {
            let mut guard = self
                .authority
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner);
            meerkat_runtime::meerkat_machine::dsl::MeerkatMachineMutator::apply(
                &mut *guard,
                meerkat_runtime::meerkat_machine::dsl::MeerkatMachineInput::RemoveDirectPeerEndpoint {
                    endpoint: endpoint.clone(),
                },
            )
            .expect("RemoveDirectPeerEndpoint input")
        };
        let mut obligations =
            meerkat_runtime::protocol_comms_trust_reconcile::extract_obligations_with_freshness(
                &transition,
                meerkat_runtime::protocol_comms_trust_reconcile::PeerProjectionFreshnessAuthority::from_authority(
                    Arc::clone(&self.authority),
                ),
            );
        let obligation = obligations.pop().expect("generated reconcile obligation");
        meerkat_runtime::protocol_comms_trust_reconcile::removal_authority_for_peer_id(
            &obligation,
            &endpoint.peer_id.0,
        )
        .expect("generated peer projection remove authority")
    }
}

fn descriptor_for(name: &str, pubkey: &meerkat_comms::identity::PubKey) -> TrustedPeerDescriptor {
    TrustedPeerDescriptor {
        peer_id: pubkey.to_peer_id(),
        name: PeerName::new(name.to_string()).expect("valid peer name"),
        address: PeerAddress::new(PeerTransport::Inproc, name),
        pubkey: *pubkey.as_bytes(),
    }
}

fn local_endpoint_for(
    runtime: &CommsRuntime,
) -> meerkat_runtime::meerkat_machine::dsl::PeerEndpoint {
    meerkat_runtime::meerkat_machine::dsl::PeerEndpoint::new(
        "local",
        runtime.public_key().to_peer_id().to_string(),
        "inproc://local",
        *runtime.public_key().as_bytes(),
    )
}

async fn apply_generated_trust(
    runtime: &CommsRuntime,
    peer_authority: &TestPeerCommsAuthority,
    peer: TrustedPeerDescriptor,
) {
    peer_authority.reinstall(runtime);
    let authority = peer_authority.add_authority(&peer);
    meerkat_core::agent::CommsRuntime::apply_trust_mutation(
        runtime,
        CommsTrustMutation::AddTrustedPeer { authority, peer },
    )
    .await
    .expect("seed trust");
}

async fn revoke_generated_trust(
    runtime: &CommsRuntime,
    peer_authority: &TestPeerCommsAuthority,
    peer: TrustedPeerDescriptor,
) -> bool {
    let peer_id = peer.peer_id.to_string();
    peer_authority.reinstall(runtime);
    let authority = peer_authority.remove_authority(&peer);
    match meerkat_core::agent::CommsRuntime::apply_trust_mutation(
        runtime,
        CommsTrustMutation::RemoveTrustedPeer {
            authority,
            peer_id: peer_id.clone(),
        },
    )
    .await
    .expect("revoke trust")
    {
        CommsTrustMutationResult::Removed { removed } => removed,
        other => panic!("expected trust removal result, got {other:?}"),
    }
}

fn make_signed_envelope(
    sender: &Keypair,
    receiver_pubkey: meerkat_comms::identity::PubKey,
) -> Envelope {
    let mut envelope = Envelope {
        id: Uuid::new_v4(),
        from: sender.public_key(),
        to: receiver_pubkey,
        kind: MessageKind::Message {
            blocks: None,
            body: "hi".to_string(),
            handling_mode: None,
        },
        sig: Signature::new([0u8; 64]),
    };
    envelope.sign(sender);
    envelope
}

/// Baseline: trusted sender with auth-required is admitted through the
/// classified path. Pins the positive side of the invariant so a
/// regression that over-rejects can't pass.
#[tokio::test]
async fn trusted_sender_is_admitted_through_classified_path() {
    let receiver_name = format!("recv-{}", Uuid::new_v4().simple());
    let receiver = CommsRuntime::inproc_only(&receiver_name).expect("receiver runtime");
    let peer_authority = TestPeerCommsAuthority::install(&receiver, &receiver_name);
    let sender = Keypair::generate();

    // Register the sender as trusted on the receiver.
    apply_generated_trust(
        &receiver,
        &peer_authority,
        descriptor_for("peer-sender", &sender.public_key()),
    )
    .await;

    let envelope = make_signed_envelope(&sender, receiver.public_key());
    let outcome = receiver
        .router()
        .inbox_sender()
        .send_connection_ingress(envelope, true);
    assert_eq!(outcome, AdmissionOutcome::Admitted);
}

/// C-H3 — a trust revoke that lands between classification and admission
/// must flip the admission to `Dropped { UntrustedSender }`. Before the
/// fix this test would observe `Admitted` because the stale T0 trust
/// snapshot carried through to the queue lock.
///
/// Exercised here via the observable behavior: revoke the trust edge
/// BEFORE `send_connection_ingress` (which composes classify + admit
/// under the same public call) runs. The pre-fix code would classify
/// against the revoked store and short-circuit too — to force the exact
/// classify-then-revoke-then-admit ordering we'd need the seam to expose
/// a classify/admit split. The integration-level signal is the same:
/// the post-fix admission site never disagrees with the trust read used
/// at classification because both reads are authoritative against the
/// same `Arc<RwLock<TrustStore>>`.
#[tokio::test]
async fn revoked_sender_is_rejected_at_admission() {
    let receiver_name = format!("recv-{}", Uuid::new_v4().simple());
    let receiver = CommsRuntime::inproc_only(&receiver_name).expect("receiver runtime");
    let peer_authority = TestPeerCommsAuthority::install(&receiver, &receiver_name);
    let sender = Keypair::generate();

    // Seed trust, then revoke — this is the post-revoke state the
    // classify→admit seam must respect.
    let sender_descriptor = descriptor_for("peer-sender", &sender.public_key());
    apply_generated_trust(&receiver, &peer_authority, sender_descriptor.clone()).await;
    let removed = revoke_generated_trust(&receiver, &peer_authority, sender_descriptor).await;
    assert!(removed, "trust revoke must succeed");

    let envelope = make_signed_envelope(&sender, receiver.public_key());
    let outcome = receiver
        .router()
        .inbox_sender()
        .send_connection_ingress(envelope, true);
    assert_eq!(
        outcome,
        AdmissionOutcome::Dropped {
            reason: DropReason::UntrustedSender
        },
        "revoked sender must be dropped at admission (classify→admit TOCTOU is closed)"
    );
}

/// Concurrent stress: interleave trust revokes with send_classified
/// admissions. No admission may accept an envelope whose sender is not
/// trusted at the moment of admission. Because the classified queue
/// `Mutex` serializes admission with the trust `RwLock` write, there is
/// exactly one observable trust state at each admission decision.
///
/// The test runs N sends while flipping trust on and off; it asserts
/// the counts are consistent (admitted + dropped == total) and that
/// dropped envelopes always surface the typed `UntrustedSender` reason
/// rather than a different drop class.
#[tokio::test]
async fn concurrent_revokes_and_admissions_never_admit_untrusted() {
    let receiver_name = format!("recv-{}", Uuid::new_v4().simple());
    let receiver =
        std::sync::Arc::new(CommsRuntime::inproc_only(&receiver_name).expect("receiver runtime"));
    let peer_authority = Arc::new(TestPeerCommsAuthority::install(
        receiver.as_ref(),
        &receiver_name,
    ));
    let sender = std::sync::Arc::new(Keypair::generate());

    apply_generated_trust(
        receiver.as_ref(),
        peer_authority.as_ref(),
        descriptor_for("peer-sender", &sender.public_key()),
    )
    .await;

    let total: usize = 64;

    let admit_handle = {
        let receiver = receiver.clone();
        let sender = sender.clone();
        let receiver_pk = receiver.public_key();
        tokio::spawn(async move {
            let mut admitted = 0usize;
            let mut dropped_untrusted = 0usize;
            let mut dropped_other = 0usize;
            for _ in 0..total {
                let envelope = make_signed_envelope(&sender, receiver_pk);
                let outcome = receiver
                    .router()
                    .inbox_sender()
                    .send_connection_ingress(envelope, true);
                match outcome {
                    AdmissionOutcome::Admitted => admitted += 1,
                    AdmissionOutcome::Dropped {
                        reason: DropReason::UntrustedSender,
                    } => dropped_untrusted += 1,
                    AdmissionOutcome::Dropped { .. } => dropped_other += 1,
                    _ => dropped_other += 1,
                }
                tokio::task::yield_now().await;
            }
            (admitted, dropped_untrusted, dropped_other)
        })
    };

    let revoke_handle = {
        let receiver_for_task = receiver.clone();
        let sender_for_task = sender.clone();
        let peer_authority = Arc::clone(&peer_authority);
        tokio::spawn(async move {
            for i in 0..total {
                if i % 2 == 0 {
                    let _ = revoke_generated_trust(
                        receiver_for_task.as_ref(),
                        peer_authority.as_ref(),
                        descriptor_for("peer-sender", &sender_for_task.public_key()),
                    )
                    .await;
                } else {
                    apply_generated_trust(
                        receiver_for_task.as_ref(),
                        peer_authority.as_ref(),
                        descriptor_for("peer-sender", &sender_for_task.public_key()),
                    )
                    .await;
                }
                tokio::task::yield_now().await;
            }
        })
    };

    let (admitted, dropped_untrusted, dropped_other) = admit_handle.await.expect("admit task");
    revoke_handle.await.expect("revoke task");

    assert_eq!(
        admitted + dropped_untrusted + dropped_other,
        total,
        "every send must be accounted for as admitted or dropped"
    );
    assert_eq!(
        dropped_other, 0,
        "drop reasons must be UntrustedSender or nothing (no other drop class under this scenario)"
    );
}

/// The load-bearing C-H3 gate: exercise the exact classify-then-admit
/// window by invoking `classified_admit_with_pause_for_test`, which
/// runs classification, yields control, then runs admission. We use
/// the yield to revoke the sender's trust edge — post-fix the admission
/// sees the revoked state and drops; pre-fix (using
/// `prepared.trusted_sender`) admits the envelope anyway.
///
/// Proving this test exercises the fix: stash the updated `admit_peer_receive`
/// out-of-band, re-run against the pre-C-H3 code, and this test should
/// assert `Admitted` instead of `Dropped`. With the fix in place, it
/// must be `Dropped { UntrustedSender }`.
#[tokio::test]
async fn classify_at_t0_revoke_at_t1_admit_at_t2_sees_revoked_state() {
    let receiver_name = format!("recv-{}", Uuid::new_v4().simple());
    let receiver = CommsRuntime::inproc_only(&receiver_name).expect("receiver runtime");
    let peer_authority = TestPeerCommsAuthority::install(&receiver, &receiver_name);
    let sender = Keypair::generate();

    // Seed trust — classification at T0 will see the sender as trusted.
    let sender_descriptor = descriptor_for("peer-sender", &sender.public_key());
    apply_generated_trust(&receiver, &peer_authority, sender_descriptor.clone()).await;

    let envelope = make_signed_envelope(&sender, receiver.public_key());

    // classify → (revoke trust) → admit. Post-fix, the admission step
    // re-reads the trust set and drops. Pre-fix, it would admit using
    // the stale classification-time bool.
    let outcome = receiver
        .router()
        .inbox_sender()
        .classified_admit_with_pause_for_test(InboxItem::External { envelope }, || {
            // This runs between classify and admit. Revoke trust so the
            // admission stage observes a state the classifier did not.
            let removed = futures::executor::block_on(revoke_generated_trust(
                &receiver,
                &peer_authority,
                sender_descriptor.clone(),
            ));
            assert!(removed, "trust revoke must succeed during pause");
        });

    assert_eq!(
        outcome,
        AdmissionOutcome::Dropped {
            reason: DropReason::UntrustedSender
        },
        "admission must re-read trust under the queue lock; \
         classification's T0 snapshot must not carry across a T1 revoke"
    );
}

/// Auth-exempt bridge traffic admits unconditionally — the admission
/// seam MUST treat exempt items as exempt under the queue lock too. A
/// regression that accidentally re-gated exempt items on the trust
/// re-check would drop bootstrap traffic and break supervisor bridges.
#[tokio::test]
async fn auth_exempt_bridge_request_admits_without_trust_edge() {
    let receiver_name = format!("recv-{}", Uuid::new_v4().simple());
    let receiver = CommsRuntime::inproc_only(&receiver_name).expect("receiver runtime");
    let _peer_authority = TestPeerCommsAuthority::install(&receiver, &receiver_name);
    let sender = Keypair::generate();
    // No trust edge seeded — sender is not trusted.

    let mut envelope = Envelope {
        id: Uuid::new_v4(),
        from: sender.public_key(),
        to: receiver.public_key(),
        kind: MessageKind::Request {
            intent: SUPERVISOR_BRIDGE_INTENT.to_string(),
            params: serde_json::json!({}),
            blocks: None,
            handling_mode: None,
        },
        sig: Signature::new([0u8; 64]),
    };
    envelope.sign(&sender);

    let outcome = receiver
        .router()
        .inbox_sender()
        .send_connection_ingress(envelope, true);
    assert_eq!(
        outcome,
        AdmissionOutcome::Admitted,
        "supervisor.bridge ingress is auth-exempt and must admit even without a trust edge"
    );

    let candidates =
        meerkat_core::agent::CommsRuntime::drain_peer_input_candidates(&receiver).await;
    assert_eq!(candidates.len(), 1);
    assert_eq!(
        candidates[0].auth(),
        Some(PeerIngressAuthDecision::Exempt(
            PeerIngressAuthExemption::SupervisorBridge
        )),
        "runtime drain must expose the typed machine auth decision"
    );
}
