//! CR-7 — portable per-group MLS state snapshot.
//!
//! What this proves: a `GroupStateSnapshot` exported by one device's `Conversation`
//! can be imported by *another device* (same user identity) on a fresh
//! `MessagingClient` and used to decrypt new traffic in the same group, without that
//! second device having to be re-Welcomed by peers.
//!
//! This is the linking + DG-recovery invariant. After CR-4 made the OpenMLS state
//! cold-restartable on the SAME device, CR-7 generalises that to cross-device
//! state transfer where both ends authenticate to the same user identity.

#![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, GroupStateSnapshot, Identity, MessageEnvelope, MessagingClient,
    Storage, StorageBackend, SyncCursor, Transport, GROUP_SNAPSHOT_HARD_CAP,
    GROUP_SNAPSHOT_VERSION,
};

#[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 open_client(
    label: &str,
    transport: Arc<CapturingTransport>,
    identity: Identity,
) -> Arc<MessagingClient> {
    let cfg = ClientConfig {
        identity,
        device_label: label.into(),
        storage: Arc::new(MemStorage::default()) as Arc<dyn Storage>,
        transport,
        now_ms: 1_000,
        storage_backend: StorageBackend::Memory,
        device_signing_secret_key: None,
    };
    block_on(MessagingClient::init(cfg)).expect("init")
}

#[test]
fn snapshot_round_trips_through_encode_decode() {
    // A creates a group, snapshots it, decodes the bytes, and asserts the fields are
    // intact. No cross-device transfer yet — just locks down the wire format.
    let transport = Arc::new(CapturingTransport::default());
    let alice = open_client("A", transport, Identity::generate());

    let conv = block_on(alice.create_conversation(Some("snap".into()), 1_000)).expect("create");
    let convs = alice.list_conversations();
    let convo_meta = convs.iter().find(|m| m.id == conv).expect("conv meta");

    // No public `Conversation` accessor on MessagingClient — but `MessagingClient` itself
    // doesn't expose conversation handles. So we exercise via build_linking_ticket which
    // calls export_state_snapshot internally per CR-13's CR-7 wiring.
    let other = open_client(
        "B",
        Arc::new(CapturingTransport::default()),
        Identity::generate(),
    );
    let kp = other.fresh_key_package().expect("kp");
    let ticket = block_on(alice.build_linking_ticket(other.device_id(), kp, Vec::new(), 2_000))
        .expect("ticket");

    // The catchup snapshot's `conversation_metas[i].group_state_bytes` is what
    // CR-7 produces. Decode each and verify.
    let catchup =
        ping_core::CatchupSnapshot::decode(&ticket.catchup_snapshot).expect("decode catchup");
    let group_entry = catchup
        .conversation_metas
        .iter()
        .find(|e| e.conversation_id == conv)
        .expect("group meta in catchup");
    assert!(
        !group_entry.group_state_bytes.is_empty(),
        "CR-7 must populate group_state_bytes"
    );

    let snap = GroupStateSnapshot::decode(&group_entry.group_state_bytes).expect("decode snapshot");
    assert_eq!(snap.v, GROUP_SNAPSHOT_VERSION);
    assert_eq!(snap.group_id, conv);
    assert_eq!(
        snap.openmls_storage_version,
        openmls_traits::storage::CURRENT_VERSION
    );
    assert!(!snap.entries.is_empty(), "non-empty group state");
    assert_eq!(convo_meta.epoch, 0); // single-member group
}

#[test]
fn cross_device_import_decrypts_subsequent_traffic() {
    // The core CR-7 invariant. Setup:
    //   A is on device A1.
    //   A1 creates a conv with peer P, exchanges messages.
    //   A "links a new device A2": A2 has the same identity, fresh device key.
    //   A1 builds a LinkingTicket carrying the conv's GroupStateSnapshot.
    //   A2 imports the snapshot → A2 holds the conv's MLS state.
    //
    // After import, A2 can decrypt new traffic in the conv. This is what makes
    // multi-device-after-the-fact work without peer re-add.
    //
    // **Caveat**: post-import, A2 has *A1's* leaf key material. Real recovery would
    // immediately issue an MLS Update Proposal to rotate A2's leaf keys — that's the
    // follow-up step documented in the memo. Here we only validate the snapshot
    // round-trip + first-decrypt invariant.

    let transport_a = Arc::new(CapturingTransport::default());
    let transport_p = Arc::new(CapturingTransport::default());

    let alice_identity = Identity::generate();
    let alice_identity_bytes = alice_identity.export();
    let alice1 = open_client("A1", transport_a.clone(), alice_identity);
    let peer = open_client("P", transport_p.clone(), Identity::generate());

    // A1 creates the conv and admits P.
    let conv = block_on(alice1.create_conversation(Some("with-P".into()), 1_000)).expect("create");
    let p_kp = peer.fresh_key_package().expect("p kp");
    block_on(alice1.add_members(conv, vec![(peer.device_id(), p_kp)], 2_000)).expect("add P");

    // P joins via the welcome from A1's transport.
    let welcome = transport_a
        .sent
        .lock()
        .iter()
        .rev()
        .find(|e| matches!(e.kind, ping_core::message::MessageKind::Welcome))
        .cloned()
        .expect("welcome");
    block_on(peer.join_conversation(&welcome, 3_000)).expect("P join");

    // A1 sends a message → P decrypts. Just to advance epoch / generate live state.
    let env_first = block_on(alice1.send(conv, b"hi from A1".to_vec(), 4_000)).expect("send 1");
    let msg = block_on(peer.process_envelope(&env_first, 4_500))
        .expect("P process 1")
        .expect("app msg");
    assert_eq!(msg.plaintext, b"hi from A1");

    // Now A1 builds a linking ticket for A2. The ticket's catchup_snapshot carries
    // the conv's GroupStateSnapshot.
    let alice2 = open_client(
        "A2",
        Arc::new(CapturingTransport::default()),
        Identity::import(&alice_identity_bytes).expect("identity"),
    );
    let a2_kp = alice2.fresh_key_package().expect("a2 kp");
    let ticket =
        block_on(alice1.build_linking_ticket(alice2.device_id(), a2_kp, Vec::new(), 5_000))
            .expect("build_linking_ticket");

    // A2 decodes the catchup; finds the conv's group_state_bytes; imports.
    let catchup =
        ping_core::CatchupSnapshot::decode(&ticket.catchup_snapshot).expect("catchup decode");
    let conv_entry = catchup
        .conversation_metas
        .iter()
        .find(|e| e.conversation_id == conv)
        .expect("conv in catchup");
    let imported_conv =
        block_on(alice2.import_state_snapshot(&conv_entry.group_state_bytes, 6_000))
            .expect("import_state_snapshot");
    assert_eq!(imported_conv, conv);
    assert!(
        alice2.list_conversations().iter().any(|m| m.id == conv),
        "A2's list_conversations must include the imported conv"
    );

    // Peer P sends a message. A2 receives it and decrypts via the imported state.
    //
    // **Why not have A1 send?** A2 imported A1's leaf state — both A1 and A2 think
    // they are leaf 0. If A1 sends, MLS rejects A2's decrypt attempt with "Cannot
    // create decryption secrets from own sender ratchet" because A2's `own_leaf_index`
    // collides with the sender. Real linking flows resolve this by having A2 issue
    // an MLS Update Proposal post-import (rotating onto a fresh leaf and bumping the
    // epoch) — that's the v1.1 step explicitly documented in
    // `docs/design/CR4_CR7_PERSISTENCE.md` open question #3.
    //
    // What v0.1 of CR-7 DOES deliver: A2 can decrypt traffic from OTHER members of
    // the group using the imported leaf-0 secrets. That's the testable invariant
    // here.
    let env_from_p =
        block_on(peer.send(conv, b"hello from P".to_vec(), 7_000)).expect("P send post-import");
    let msg2 = block_on(alice2.process_envelope(&env_from_p, 8_000))
        .expect("A2 process — THIS is the CR-7 v0.1 invariant")
        .expect("app msg");
    assert_eq!(
        msg2.plaintext, b"hello from P",
        "A2 must decrypt peer-originated traffic after importing A1's snapshot"
    );
}

#[test]
fn import_rejects_already_open_conversation() {
    // Snapshotting then re-importing into the same client without a close in between
    // is a programmer error — guard against it.
    let transport = Arc::new(CapturingTransport::default());
    let alice = open_client("A", transport, Identity::generate());
    let other = open_client(
        "B",
        Arc::new(CapturingTransport::default()),
        Identity::generate(),
    );

    let conv = block_on(alice.create_conversation(None, 1_000)).expect("create");
    let kp = other.fresh_key_package().expect("kp");
    let ticket = block_on(alice.build_linking_ticket(other.device_id(), kp, Vec::new(), 2_000))
        .expect("ticket");

    let catchup =
        ping_core::CatchupSnapshot::decode(&ticket.catchup_snapshot).expect("catchup decode");
    let conv_entry = catchup
        .conversation_metas
        .iter()
        .find(|e| e.conversation_id == conv)
        .expect("conv in catchup");

    let err =
        block_on(alice.import_state_snapshot(&conv_entry.group_state_bytes, 3_000)).unwrap_err();
    assert!(
        format!("{err}").contains("already open"),
        "expected 'already open' error, got: {err}"
    );
}

#[test]
fn snapshot_caps_enforced() {
    // Encode-side hard cap is what `encode()` returns errors for; we synthesise a
    // bloated snapshot directly because building a 256 KB+ real MLS state is
    // expensive. This pins the same behaviour the decode side enforces.
    use ping_core::{GroupSnapshotEntry, GroupStateSnapshot};

    let bloat = vec![0xaa; GROUP_SNAPSHOT_HARD_CAP + 1];
    let snap = GroupStateSnapshot {
        v: GROUP_SNAPSHOT_VERSION,
        group_id: ConversationId([0u8; 16]),
        openmls_storage_version: openmls_traits::storage::CURRENT_VERSION,
        snapshot_created_at_ms: 0,
        entries: vec![GroupSnapshotEntry {
            key: b"k".to_vec(),
            value: bloat,
        }],
    };
    let err = snap.encode().unwrap_err();
    assert!(format!("{err}").contains("hard cap"));

    // Decode side rejects oversize inputs without trying to parse.
    let too_big = vec![0u8; GROUP_SNAPSHOT_HARD_CAP + 100];
    let err = GroupStateSnapshot::decode(&too_big).unwrap_err();
    assert!(format!("{err}").contains("hard cap"));

    // Unknown version on decode.
    let weird = GroupStateSnapshot {
        v: 99,
        group_id: ConversationId([0u8; 16]),
        openmls_storage_version: openmls_traits::storage::CURRENT_VERSION,
        snapshot_created_at_ms: 0,
        entries: Vec::new(),
    };
    let bytes = weird.encode().expect("encode v=99 succeeds");
    let err = GroupStateSnapshot::decode(&bytes).unwrap_err();
    assert!(format!("{err}").contains("v=99 not supported"));
}

#[test]
fn import_rejects_mismatched_openmls_storage_version() {
    use ping_core::{GroupSnapshotEntry, GroupStateSnapshot};
    let snap = GroupStateSnapshot {
        v: GROUP_SNAPSHOT_VERSION,
        group_id: ConversationId([7u8; 16]),
        // Force a version we don't support.
        openmls_storage_version: 999,
        snapshot_created_at_ms: 0,
        entries: vec![GroupSnapshotEntry {
            key: b"placeholder".to_vec(),
            value: b"placeholder".to_vec(),
        }],
    };
    let bytes = snap.encode().expect("encode");
    let alice = open_client(
        "A",
        Arc::new(CapturingTransport::default()),
        Identity::generate(),
    );
    let err = block_on(alice.import_state_snapshot(&bytes, 1_000)).unwrap_err();
    assert!(
        format!("{err}").contains("openmls_storage_version"),
        "expected storage-version error, got: {err}"
    );
}
