//! Send-then-merge rollback for group membership commits.
//!
//! Guarantees the local MLS epoch never runs ahead of the server: a Commit the
//! server REJECTS is rolled back (no epoch advance, group still usable), while an
//! AMBIGUOUS transport failure (no response — possible masked success) merges
//! locally so we don't strand ourselves a step behind. This is what prevents the
//! `epoch_advanced` (409) desync that otherwise permanently bricks a 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, Error, Identity, MessageEnvelope, MessageKind, MessagingClient,
    Storage, SyncCursor, Transport,
};

// 0 = accept, 1 = reject Commits with a DEFINITE 4xx, 2 = fail Commits with an
// AMBIGUOUS network error (no server response).
#[derive(Debug)]
struct ToggleTransport {
    mode: Mutex<u8>,
    sent: Mutex<Vec<MessageEnvelope>>,
}

impl ToggleTransport {
    fn new() -> Self {
        Self {
            mode: Mutex::new(0),
            sent: Mutex::new(Vec::new()),
        }
    }
    fn set_mode(&self, m: u8) {
        *self.mode.lock() = m;
    }
}

impl Transport for ToggleTransport {
    fn send<'a>(
        &'a self,
        envelope: MessageEnvelope,
    ) -> ping_core::transport::TransportFuture<'a, ()> {
        Box::pin(async move {
            let mode = *self.mode.lock();
            if matches!(envelope.kind, MessageKind::Commit) {
                match mode {
                    1 => return Err(Error::Transport("network: http 409 epoch_advanced".into())),
                    2 => return Err(Error::Transport("network error: fetch failed".into())),
                    _ => {}
                }
            }
            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<ToggleTransport>) -> 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 epoch_of(c: &MessagingClient, conv: ConversationId) -> u64 {
    c.list_conversations()
        .into_iter()
        .find(|m| m.id == conv)
        .map(|m| m.epoch)
        .expect("conv meta")
}
fn members_of(c: &MessagingClient, conv: ConversationId) -> u32 {
    c.list_conversations()
        .into_iter()
        .find(|m| m.id == conv)
        .map(|m| m.member_count)
        .expect("conv meta")
}

#[test]
fn rejected_commit_rolls_back_then_retry_succeeds() {
    let ta = Arc::new(ToggleTransport::new());
    let tb = Arc::new(ToggleTransport::new());
    let a = make_client("A", ta.clone());
    let b = make_client("B", tb.clone());

    let conv = block_on(a.create_conversation(None, 1_000)).expect("create");
    let e0 = epoch_of(&a, conv);
    assert_eq!(members_of(&a, conv), 1);

    // Server REJECTS the Commit (definite 409). The add must fail AND leave the
    // group exactly where it was — no epoch advance, no phantom member.
    ta.set_mode(1);
    let kp = block_on(b.fresh_key_package()).expect("kp");
    let res = block_on(a.add_members(conv, vec![(b.device_id(), kp)], 2_000));
    assert!(res.is_err(), "a rejected Commit must surface as an error");
    assert_eq!(
        epoch_of(&a, conv),
        e0,
        "epoch must NOT advance on a rejected Commit"
    );
    assert_eq!(members_of(&a, conv), 1, "no phantom member after rollback");

    // The group is still operational: with the server now accepting, a retry
    // succeeds and advances the epoch in lockstep.
    ta.set_mode(0);
    let kp2 = block_on(b.fresh_key_package()).expect("kp2");
    block_on(a.add_members(conv, vec![(b.device_id(), kp2)], 3_000)).expect("retry add");
    assert!(
        epoch_of(&a, conv) > e0,
        "epoch advances after the accepted retry"
    );
    assert_eq!(members_of(&a, conv), 2, "B added on the retry");
}

#[test]
fn ambiguous_failure_merges_to_match_possible_masked_success() {
    let ta = Arc::new(ToggleTransport::new());
    let tb = Arc::new(ToggleTransport::new());
    let a = make_client("A", ta.clone());
    let b = make_client("B", tb.clone());

    let conv = block_on(a.create_conversation(None, 1_000)).expect("create");
    let e0 = epoch_of(&a, conv);

    // Ambiguous failure (no server response): the server MAY have applied the
    // Commit, so we merge locally to match a masked success rather than fall a
    // step behind. The call still surfaces the error so the host can react.
    ta.set_mode(2);
    let kp = block_on(b.fresh_key_package()).expect("kp");
    let res = block_on(a.add_members(conv, vec![(b.device_id(), kp)], 2_000));
    assert!(res.is_err(), "ambiguous failure still surfaces an error");
    assert!(
        epoch_of(&a, conv) > e0,
        "epoch advances (merge-on-ambiguous)"
    );
    assert_eq!(members_of(&a, conv), 2, "member merged locally");
}
