//! CR-2 — `revoke_device` end-to-end.
//!
//! The scenario the CR is built for: a user wants to revoke one of their devices.
//! `revoke_device(device_id)` walks the locally-known device→leaf maps and emits one
//! Commit envelope per conversation it could resolve. After processing those commits,
//! the revoked device's leaf is gone from each affected group.

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

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 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")
}

fn last_welcome(t: &CapturingTransport) -> MessageEnvelope {
    t.sent
        .lock()
        .iter()
        .rev()
        .find(|e| matches!(e.kind, ping_core::message::MessageKind::Welcome))
        .cloned()
        .expect("welcome captured")
}

#[test]
fn revoke_removes_locally_admitted_device_from_one_conversation() {
    // A creates a conversation, admits B. A calls revoke_device(B). Result: one
    // Commit envelope, returned and broadcast; B's leaf is gone from the conversation.
    let transport_a = Arc::new(CapturingTransport::default());
    let transport_b = Arc::new(CapturingTransport::default());
    let client_a = make_client("A", transport_a.clone());
    let client_b = make_client("B", transport_b.clone());

    let b_kp = client_b.fresh_key_package().expect("kp");
    let conv = block_on(client_a.create_conversation(None, 1_000)).expect("create");
    block_on(client_a.add_members(conv, vec![(client_b.device_id(), b_kp)], 2_000)).expect("add B");
    let before_epoch = client_a
        .list_conversations()
        .into_iter()
        .find(|m| m.id == conv)
        .map(|m| m.epoch)
        .expect("conv meta");
    let before_members = client_a
        .list_conversations()
        .into_iter()
        .find(|m| m.id == conv)
        .map(|m| m.member_count)
        .expect("conv meta");
    assert_eq!(before_members, 2);

    let envelopes = block_on(client_a.revoke_device(client_b.device_id(), 3_000)).expect("revoke");
    assert_eq!(envelopes.len(), 1, "one Commit per affected conversation");
    assert_eq!(envelopes[0].conversation_id, conv);
    assert!(matches!(
        envelopes[0].kind,
        ping_core::message::MessageKind::Commit
    ));

    let after = client_a
        .list_conversations()
        .into_iter()
        .find(|m| m.id == conv)
        .expect("conv still present on A");
    assert!(after.epoch > before_epoch, "epoch must advance");
    assert_eq!(after.member_count, 1, "B leaf gone");
}

#[test]
fn revoke_handles_multiple_conversations_in_one_call() {
    // A admits B into two conversations. revoke_device(B) returns two Commits.
    let transport_a = Arc::new(CapturingTransport::default());
    let transport_b = Arc::new(CapturingTransport::default());
    let client_a = make_client("A", transport_a.clone());
    let client_b = make_client("B", transport_b.clone());

    let conv1 = block_on(client_a.create_conversation(Some("one".into()), 1_000)).expect("c1");
    let conv2 = block_on(client_a.create_conversation(Some("two".into()), 1_100)).expect("c2");

    let kp1 = client_b.fresh_key_package().expect("kp1");
    block_on(client_a.add_members(conv1, vec![(client_b.device_id(), kp1)], 2_000)).unwrap();
    let kp2 = client_b.fresh_key_package().expect("kp2");
    block_on(client_a.add_members(conv2, vec![(client_b.device_id(), kp2)], 2_100)).unwrap();

    let envelopes = block_on(client_a.revoke_device(client_b.device_id(), 3_000)).expect("revoke");
    assert_eq!(envelopes.len(), 2, "one Commit per conversation");
    let conv_ids: std::collections::HashSet<_> =
        envelopes.iter().map(|e| e.conversation_id).collect();
    assert!(conv_ids.contains(&conv1) && conv_ids.contains(&conv2));

    for meta in client_a.list_conversations() {
        assert_eq!(meta.member_count, 1, "B revoked from {}", meta.id.as_hex());
    }
}

#[test]
fn revoke_unknown_device_is_noop() {
    // Revoking a device this client never admitted returns an empty Vec, not an error.
    let transport = Arc::new(CapturingTransport::default());
    let client = make_client("A", transport);
    let _conv = block_on(client.create_conversation(None, 1_000)).expect("create");

    let bogus_device = ping_core::DeviceId(vec![0xff; 32]);
    let envelopes = block_on(client.revoke_device(bogus_device, 2_000)).expect("revoke");
    assert!(envelopes.is_empty(), "no locally-known leaves to revoke");
}

#[test]
fn peer_admitted_device_is_silently_skipped() {
    // Three clients. A admits B. Then B admits C. C now controls a leaf in the
    // conversation, but A never recorded C's device_id → leaf mapping (because A
    // didn't do the admit). A's `revoke_device(C)` returns an empty list.
    // This documents the v0.1 scope limitation called out in the doc-comment.
    let transport_a = Arc::new(CapturingTransport::default());
    let transport_b = Arc::new(CapturingTransport::default());
    let transport_c = Arc::new(CapturingTransport::default());
    let client_a = make_client("A", transport_a.clone());
    let client_b = make_client("B", transport_b.clone());
    let client_c = make_client("C", transport_c.clone());

    // A creates the group and admits B.
    let conv = block_on(client_a.create_conversation(None, 1_000)).expect("create");
    let b_kp = client_b.fresh_key_package().expect("kp");
    block_on(client_a.add_members(conv, vec![(client_b.device_id(), b_kp)], 2_000)).unwrap();

    // B joins via the Welcome A captured.
    let welcome = last_welcome(&transport_a);
    block_on(client_b.join_conversation(&welcome, 2_500)).expect("B join");

    // B admits C — A doesn't know C's leaf.
    let c_kp = client_c.fresh_key_package().expect("c kp");
    block_on(client_b.add_members(conv, vec![(client_c.device_id(), c_kp)], 3_000))
        .expect("B admits C");

    // A's view of C: unknown leaf. Revoke returns nothing.
    let envelopes =
        block_on(client_a.revoke_device(client_c.device_id(), 4_000)).expect("A revoke C");
    assert!(
        envelopes.is_empty(),
        "A locally has no mapping for C — silent skip per CR-2 v0.1 scope"
    );

    // Sanity: B *does* know C's leaf and CAN revoke it.
    let envelopes_b =
        block_on(client_b.revoke_device(client_c.device_id(), 5_000)).expect("B revoke C");
    assert_eq!(envelopes_b.len(), 1, "B admitted C, so B can revoke C");
}

#[test]
fn revoke_after_remove_is_idempotent_noop() {
    // A admits B, then manually removes B's leaf (by index). After that, revoke_device
    // should find no locally-known mapping and return empty — the prior remove_members
    // pruned the map.
    let transport_a = Arc::new(CapturingTransport::default());
    let transport_b = Arc::new(CapturingTransport::default());
    let client_a = make_client("A", transport_a.clone());
    let client_b = make_client("B", transport_b.clone());

    let conv = block_on(client_a.create_conversation(None, 1_000)).expect("create");
    let b_kp = client_b.fresh_key_package().expect("kp");
    block_on(client_a.add_members(conv, vec![(client_b.device_id(), b_kp)], 2_000)).unwrap();

    // Resolve B's leaf via the (otherwise internal) capture path: take the most-
    // recent member count delta. Simpler: just call remove_members(1) — B is leaf 1
    // (A created the group so A is leaf 0).
    block_on(client_a.remove_members(conv, vec![1], 3_000)).expect("manual remove");

    let envelopes = block_on(client_a.revoke_device(client_b.device_id(), 4_000)).expect("revoke");
    assert!(
        envelopes.is_empty(),
        "remove_members pruned the device→leaf map; revoke is now a no-op"
    );
}
