//! CR-2 conformance runner — Rust binding.
//!
//! The CR-2 conformance is an invariant fixture (see scenario.json) — exact wire bytes
//! differ across bindings because each OpenMLS run produces fresh group state, but the
//! invariants (envelope count, conversation routing, post-revoke member count, scope
//! limit on peer-admitted devices) must hold on every binding's facade.

#![cfg(feature = "test-utils")]

use std::path::PathBuf;
use std::sync::Arc;

use futures::executor::block_on;
use parking_lot::Mutex;

use ping_core::{
    storage::memory::MemStorage, transport::DiscoveredDevice, transport::TransportSubscription,
    ClientConfig, ConversationId, Identity, MessageEnvelope, MessagingClient, Storage, SyncCursor,
    Transport,
};

#[derive(Debug, Default)]
struct CapturingTransport {
    sent: Mutex<Vec<MessageEnvelope>>,
}

impl Transport for CapturingTransport {
    fn send<'a>(
        &'a self,
        envelope: MessageEnvelope,
    ) -> ping_core::transport::TransportFuture<'a, ()> {
        Box::pin(async move {
            self.sent.lock().push(envelope);
            Ok(())
        })
    }
    fn fetch_since<'a>(
        &'a self,
        _conv: ConversationId,
        _cursor: SyncCursor,
        _limit: u32,
    ) -> ping_core::transport::TransportFuture<'a, Vec<MessageEnvelope>> {
        Box::pin(async move { Ok(Vec::new()) })
    }
    fn subscribe<'a>(
        &'a self,
        _cb: Arc<dyn Fn(MessageEnvelope) + Send + Sync>,
    ) -> ping_core::transport::TransportFuture<'a, TransportSubscription> {
        Box::pin(async move { Ok(TransportSubscription::new(|| ())) })
    }
    fn discover_devices<'a>(
        &'a self,
        _user_id: ping_core::UserId,
    ) -> ping_core::transport::TransportFuture<'a, Vec<DiscoveredDevice>> {
        Box::pin(async move { Ok(Vec::new()) })
    }
}

fn fixture_path() -> PathBuf {
    let mut p = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
    p.push("../../protocol/conformance/cr2_revoke_device/scenario.json");
    p
}

fn make_client(label: &str, transport: Arc<CapturingTransport>) -> Arc<MessagingClient> {
    let cfg = ClientConfig {
        identity: Identity::generate(),
        device_label: label.into(),
        storage: Arc::new(MemStorage::default()) as Arc<dyn Storage>,
        transport,
        now_ms: 1_000,
        storage_backend: ping_core::StorageBackend::Memory,
        device_signing_secret_key: None,
    };
    block_on(MessagingClient::init(cfg)).expect("init")
}

#[test]
fn cr2_conformance_scenarios() {
    // Sanity: the fixture file is reachable.
    assert!(
        std::fs::metadata(fixture_path()).is_ok(),
        "CR-2 scenario.json missing at {}",
        fixture_path().display()
    );

    // --- single_conv_locally_admitted ---
    {
        let t_a = Arc::new(CapturingTransport::default());
        let t_b = Arc::new(CapturingTransport::default());
        let a = make_client("A", t_a.clone());
        let b = make_client("B", t_b.clone());
        let kp = block_on(b.fresh_key_package()).unwrap();
        let conv = block_on(a.create_conversation(None, 1_000)).unwrap();
        block_on(a.add_members(conv, vec![(b.device_id(), kp)], 2_000)).unwrap();
        let before = a
            .list_conversations()
            .into_iter()
            .find(|m| m.id == conv)
            .unwrap();
        let envs = block_on(a.revoke_device(b.device_id(), 3_000)).unwrap();
        assert_eq!(envs.len(), 1, "single_conv_locally_admitted: envs.length");
        assert_eq!(envs[0].conversation_id, conv);
        assert!(matches!(
            envs[0].kind,
            ping_core::message::MessageKind::Commit
        ));
        let after = a
            .list_conversations()
            .into_iter()
            .find(|m| m.id == conv)
            .unwrap();
        assert_eq!(after.member_count, 1);
        assert!(after.epoch > before.epoch);
    }

    // --- multi_conv_locally_admitted ---
    {
        let t_a = Arc::new(CapturingTransport::default());
        let t_b = Arc::new(CapturingTransport::default());
        let a = make_client("A", t_a.clone());
        let b = make_client("B", t_b.clone());
        let conv1 = block_on(a.create_conversation(Some("c1".into()), 1_000)).unwrap();
        let conv2 = block_on(a.create_conversation(Some("c2".into()), 1_100)).unwrap();
        let kp1 = block_on(b.fresh_key_package()).unwrap();
        block_on(a.add_members(conv1, vec![(b.device_id(), kp1)], 2_000)).unwrap();
        let kp2 = block_on(b.fresh_key_package()).unwrap();
        block_on(a.add_members(conv2, vec![(b.device_id(), kp2)], 2_100)).unwrap();
        let envs = block_on(a.revoke_device(b.device_id(), 3_000)).unwrap();
        assert_eq!(envs.len(), 2, "multi_conv: two envelopes");
        let conv_ids: std::collections::HashSet<_> =
            envs.iter().map(|e| e.conversation_id).collect();
        assert!(conv_ids.contains(&conv1));
        assert!(conv_ids.contains(&conv2));
        for m in a.list_conversations() {
            assert_eq!(m.member_count, 1, "multi_conv: post-revoke member count");
        }
    }

    // --- peer_admitted_scope_limit ---
    {
        let t_a = Arc::new(CapturingTransport::default());
        let t_b = Arc::new(CapturingTransport::default());
        let t_c = Arc::new(CapturingTransport::default());
        let a = make_client("A", t_a.clone());
        let b = make_client("B", t_b.clone());
        let c = make_client("C", t_c.clone());
        let conv = block_on(a.create_conversation(None, 1_000)).unwrap();
        let b_kp = block_on(b.fresh_key_package()).unwrap();
        block_on(a.add_members(conv, vec![(b.device_id(), b_kp)], 2_000)).unwrap();
        let welcome = t_a
            .sent
            .lock()
            .iter()
            .rev()
            .find(|e| matches!(e.kind, ping_core::message::MessageKind::Welcome))
            .cloned()
            .expect("welcome");
        block_on(b.join_conversation(&welcome, 2_500)).unwrap();
        let c_kp = block_on(c.fresh_key_package()).unwrap();
        block_on(b.add_members(conv, vec![(c.device_id(), c_kp)], 3_000)).unwrap();

        let envs_a = block_on(a.revoke_device(c.device_id(), 4_000)).unwrap();
        assert_eq!(
            envs_a.len(),
            0,
            "peer_admitted_scope_limit: A has no local C→leaf, silent skip"
        );
        let envs_b = block_on(b.revoke_device(c.device_id(), 5_000)).unwrap();
        assert_eq!(envs_b.len(), 1, "B admitted C, so B *can* revoke C");
    }

    // --- unknown_device_is_noop ---
    {
        let t = Arc::new(CapturingTransport::default());
        let a = make_client("A", t);
        let _conv = block_on(a.create_conversation(None, 1_000)).unwrap();
        let envs = block_on(a.revoke_device(ping_core::DeviceId(vec![0xab; 32]), 2_000)).unwrap();
        assert!(envs.is_empty(), "unknown_device_is_noop");
    }

    // --- idempotent_after_manual_remove ---
    {
        let t_a = Arc::new(CapturingTransport::default());
        let t_b = Arc::new(CapturingTransport::default());
        let a = make_client("A", t_a.clone());
        let b = make_client("B", t_b.clone());
        let conv = block_on(a.create_conversation(None, 1_000)).unwrap();
        let kp = block_on(b.fresh_key_package()).unwrap();
        block_on(a.add_members(conv, vec![(b.device_id(), kp)], 2_000)).unwrap();
        block_on(a.remove_members(conv, vec![1], 3_000)).unwrap();
        let envs = block_on(a.revoke_device(b.device_id(), 4_000)).unwrap();
        assert!(
            envs.is_empty(),
            "idempotent_after_manual_remove: remove_members pruned the map"
        );
    }
}
