iris-chat-protocol 0.3.19

use super::*;

pub(super) fn queued_group_history(count: usize) -> (ProtocolEngine, Keys, Keys) {
    let owner = Keys::generate();
    let device = Keys::generate();
    let sender = Keys::generate();
    let sender_device = Keys::generate();
    let author = ndr_device(Keys::generate().public_key());
    let mut engine = test_engine(&owner, &device);
    engine
        .ingest_app_keys_event(&signed_app_keys(&sender, &[sender_device.public_key()], 1))
        .unwrap();
    let group = group_snapshot_for_test(
        "queued-group",
        "History",
        1,
        &sender,
        &[owner.public_key(), sender.public_key()],
    );
    let initial = nostr_double_ratchet::SenderKeyState::new(1, [9; 32], 0);
    engine.group_manager = GroupEventManager::from_snapshot(GroupManagerSnapshot {
        local_owner_pubkey: engine.local_owner,
        groups: vec![group.clone()],
        sender_keys: vec![incoming_sender_key_record(
            &group.group_id,
            sender.public_key(),
            sender_device.public_key(),
            author,
            &initial,
        )],
    })
    .unwrap();
    let mut encryptor = initial;
    for index in 0..count {
        let plaintext = JsonGroupPayloadCodecV1
            .encode_sender_key_plaintext(
                nostr_double_ratchet::GroupPayloadEncodeContext {
                    local_device_pubkey: ndr_device(sender_device.public_key()),
                    created_at: NdrUnixSeconds(20),
                },
                &nostr_double_ratchet::GroupSenderKeyPlaintext {
                    group_id: group.group_id.clone(),
                    revision: 1,
                    body: format!("history-{index}").into_bytes(),
                },
            )
            .unwrap();
        let (message_number, ciphertext) = encryptor.encrypt_to_bytes(&plaintext).unwrap();
        engine.pending_group_sender_key_messages.push(
            nostr_double_ratchet::wire::ParsedGroupSenderKeyMessageEvent {
                sender_event_pubkey: author,
                key_id: 1,
                message_number,
                encrypted_header: Some("hidden position".into()),
                created_at: NdrUnixSeconds(20),
                ciphertext,
            },
        );
    }
    (engine, sender, sender_device)
}

#[test]
fn group_retry_budget_preserves_backlog_and_yields_between_batches() {
    let (mut engine, _, _) = queued_group_history(20);
    let first = engine.retry_pending_protocol(NdrUnixSeconds(30)).unwrap();
    assert!(
        first.group_result.events.len() <= 8,
        "one turn must not decrypt the whole history queue"
    );
    assert_eq!(
        first.group_result.events.len() + engine.pending_group_sender_key_messages.len(),
        20
    );
    assert!(
        engine.has_due_pending_retry_work(NdrUnixSeconds(30)),
        "the retained tail must schedule continuation"
    );
    let mut delivered = first.group_result.events.len();
    for _ in 0..20 {
        let next = engine.retry_pending_protocol(NdrUnixSeconds(30)).unwrap();
        assert!(next.group_result.events.len() <= 8);
        delivered += next.group_result.events.len();
        if engine.pending_group_sender_key_messages.is_empty() {
            break;
        }
    }
    assert_eq!(delivered, 20);
    assert!(engine.pending_group_sender_key_messages.is_empty());
}

#[test]
fn group_retry_does_not_run_for_an_ordinary_direct_message() {
    let (mut engine, sender, sender_device) = queued_group_history(20);
    let result = engine
        .process_group_pairwise_payload(
            b"ordinary direct message",
            sender.public_key(),
            Some(sender_device.public_key()),
        )
        .unwrap();
    assert!(!result.consumed);
    assert!(
        result.events.is_empty(),
        "a direct message must not synchronously decrypt unrelated group history"
    );
    assert_eq!(engine.pending_group_sender_key_messages.len(), 20);
}

pub(super) struct GroupRetryCountingStorage {
    pub(super) inner: Arc<dyn StorageAdapter>,
    pub(super) puts: std::sync::atomic::AtomicUsize,
    pub(super) fail_next: std::sync::atomic::AtomicBool,
}

impl StorageAdapter for GroupRetryCountingStorage {
    fn get(&self, key: &str) -> StorageResult<Option<String>> {
        self.inner.get(key)
    }
    fn put(&self, key: &str, value: String) -> StorageResult<()> {
        self.puts.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        if self
            .fail_next
            .swap(false, std::sync::atomic::Ordering::Relaxed)
        {
            return Err(StorageError::new("injected group retry save failure"));
        }
        self.inner.put(key, value)
    }
    fn put_checkpoint(&self, key: &str, value: &crate::StorageCheckpoint) -> StorageResult<()> {
        self.puts.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        if self
            .fail_next
            .swap(false, std::sync::atomic::Ordering::Relaxed)
        {
            return Err(StorageError::new("injected group retry save failure"));
        }
        self.inner.put_checkpoint(key, value)
    }
    fn del(&self, key: &str) -> StorageResult<()> {
        self.inner.del(key)
    }
    fn list(&self, prefix: &str) -> StorageResult<Vec<String>> {
        self.inner.list(prefix)
    }
}

fn deferred_group_retry_query_fixture() -> (ProtocolEngine, Keys, Arc<GroupRetryCountingStorage>) {
    let (mut engine, sender, device) = queued_group_history(1);
    let candidates = std::mem::take(&mut engine.pending_group_sender_key_messages);
    engine.pending_inbound = synthetic_pending_inbound(&mut engine, 128, 0, 30);
    engine.pending_group_sender_key_messages = candidates;
    engine.pending_group_fanouts = (0..256)
        .map(|index| ProtocolPendingGroupFanout {
            group_id: "queued-group".into(),
            fanout: GroupPendingFanout::Remote {
                recipient_owner: ndr_owner(sender.public_key()),
                payload: format!("deferred-{index}").into_bytes(),
            }
            .into(),
            remaining_devices: Some(BTreeSet::from([ndr_device(device.public_key())])),
            inner_event_id: Some(format!("deferred-{index}")),
            created_at_secs: 20,
            next_retry_at_secs: 90,
        })
        .collect();
    let storage = Arc::new(GroupRetryCountingStorage {
        inner: engine.storage.clone(),
        puts: 0.into(),
        fail_next: false.into(),
    });
    engine.storage = storage.clone();
    (engine, device, storage)
}

#[test]
fn unchanged_invite_replay_preserves_deferred_work_without_checkpointing() {
    let (mut engine, _, storage) = deferred_group_retry_query_fixture();
    let peer = Keys::generate();
    let peer_device = Keys::generate();
    let peer_engine = test_engine(&peer, &peer_device);
    let invite = invite_unsigned_event(&peer_engine.local_invite().unwrap())
        .unwrap()
        .sign_with_keys(&peer_device)
        .unwrap();
    engine.observe_invite_event(&invite).unwrap();
    // Pending work can arrive after the invite. Replaying unchanged discovery
    // must not wake it or rewrite its checkpoint on every relay catch-up.
    engine.pending_inbound = synthetic_pending_inbound(&mut engine, 128, 0, 30);
    let before = serde_json::to_value((
        &engine.pending_inbound,
        &engine.pending_group_fanouts,
        &engine.pending_group_pairwise_payloads,
    ))
    .unwrap();
    let writes = storage.puts.load(std::sync::atomic::Ordering::Relaxed);
    for _ in 0..16 {
        assert!(engine.observe_invite_event(&invite).unwrap().is_empty());
    }
    assert_eq!(
        storage.puts.load(std::sync::atomic::Ordering::Relaxed),
        writes,
        "unchanged invitations must not checkpoint the entire backlog"
    );
    assert_eq!(
        serde_json::to_value((
            &engine.pending_inbound,
            &engine.pending_group_fanouts,
            &engine.pending_group_pairwise_payloads,
        ))
        .unwrap(),
        before,
        "ordinary retry deadlines belong to the retry scheduler"
    );
}

#[test]
fn metadata_outcome_reports_authorization_changes_even_without_retry_effects() {
    let mut engine = test_engine(&Keys::generate(), &Keys::generate());
    let peer = Keys::generate();
    let device = Keys::generate();
    let sibling = Keys::generate();
    let roster = signed_app_keys(&peer, &[device.public_key()], 10);
    let first = engine.ingest_app_keys_event_with_outcome(&roster).unwrap();
    assert!(first.changed);
    assert!(first.retry_batch.is_empty());
    assert!(
        !engine
            .ingest_app_keys_event_with_outcome(&roster)
            .unwrap()
            .changed
    );
    let stale = signed_app_keys(&peer, &[device.public_key()], 9);
    assert!(
        !engine
            .ingest_app_keys_event_with_outcome(&stale)
            .unwrap()
            .changed
    );

    // A projection is insufficient when exact signed proof was lost during
    // migration. Replaying it must restore authorization and report progress.
    engine.invite_owner_app_keys_evidence.clear();
    let repaired = engine.ingest_app_keys_event_with_outcome(&roster).unwrap();
    assert!(repaired.changed);
    assert!(repaired.retry_batch.is_empty());
    assert_eq!(
        engine.invite_owner_app_keys_membership(ndr_owner(peer.public_key()), device.public_key()),
        Some(true)
    );

    let conflict = signed_app_keys(&peer, &[sibling.public_key()], 10);
    assert!(
        engine
            .ingest_app_keys_event_with_outcome(&conflict)
            .unwrap()
            .changed
    );
    assert_eq!(
        engine.invite_owner_app_keys_membership(ndr_owner(peer.public_key()), device.public_key()),
        None
    );
    let removal = signed_app_keys(&peer, &[sibling.public_key()], 11);
    assert!(
        engine
            .ingest_app_keys_event_with_outcome(&removal)
            .unwrap()
            .changed
    );
    assert_eq!(
        engine.invite_owner_app_keys_membership(ndr_owner(peer.public_key()), device.public_key()),
        Some(false)
    );
}

#[test]
fn metadata_outcome_distinguishes_invite_replacement_from_exact_replay() {
    let local = Keys::generate();
    let mut engine = test_engine(&local, &local);
    let peer = Keys::generate();
    let peer_engine = test_engine(&peer, &peer);
    let mut invite = peer_engine.local_invite().unwrap();
    invite.created_at = NdrUnixSeconds(10);
    let first_event = invite_unsigned_event(&invite)
        .unwrap()
        .sign_with_keys(&peer)
        .unwrap();
    let first = engine
        .observe_invite_event_with_outcome(&first_event)
        .unwrap();
    assert!(first.changed);
    assert!(first.retry_batch.is_empty());
    assert!(
        !engine
            .observe_invite_event_with_outcome(&first_event)
            .unwrap()
            .changed
    );

    let mut replacement = test_engine(&peer, &peer).local_invite().unwrap();
    replacement.created_at = NdrUnixSeconds(11);
    let replacement = invite_unsigned_event(&replacement)
        .unwrap()
        .sign_with_keys(&peer)
        .unwrap();
    let replaced = engine
        .observe_invite_event_with_outcome(&replacement)
        .unwrap();
    assert!(replaced.changed);
    assert!(replaced.retry_batch.is_empty());
    assert!(
        !engine
            .observe_invite_event_with_outcome(&replacement)
            .unwrap()
            .changed
    );
    let own = invite_unsigned_event(&engine.local_invite().unwrap())
        .unwrap()
        .sign_with_keys(&local)
        .unwrap();
    let own = engine.observe_invite_event_with_outcome(&own).unwrap();
    assert!(!own.changed && own.retry_batch.is_empty());
}

fn assert_ready_retry_queries_skip_deferred_work(
    engine: &ProtocolEngine,
    storage: &GroupRetryCountingStorage,
) {
    let queues = || {
        serde_json::to_value((
            &engine.pending_remote_sends,
            &engine.pending_local_sibling_sends,
            &engine.pending_inbound,
            &engine.pending_group_fanouts,
            &engine.pending_group_pairwise_payloads,
            &engine.pending_group_sender_key_repairs,
            &engine.pending_group_sender_key_messages,
            &engine.pending_decrypted_deliveries,
        ))
        .unwrap()
    };
    let queues_before = queues();
    let journal = engine.pending_decrypted_deliveries.clone();
    let group = engine.group_manager.snapshot();
    let sessions = serde_json::to_value(engine.session_manager.snapshot()).unwrap();
    let before = engine.group_sender_key_retry.borrow().clone();
    let writes = storage.puts.load(std::sync::atomic::Ordering::Relaxed);
    engine.invalidate_pending_inbound_retry_deadline();
    take_inbound_retry_deadline_scans();
    for _ in 0..32 {
        assert!(engine.has_due_pending_retry_work(NdrUnixSeconds(30)));
    }
    // The unknown inbound deadline precedes fanouts in the fallback, so zero
    // visits also proves that these queries never reached the deferred fanouts.
    assert_eq!(take_inbound_retry_deadline_scans(), 0);
    assert!(
        queues() == queues_before,
        "queries must preserve every retry queue"
    );
    assert_eq!(engine.pending_decrypted_deliveries, journal);
    assert_eq!(engine.group_manager.snapshot(), group);
    assert_eq!(
        serde_json::to_value(engine.session_manager.snapshot()).unwrap(),
        sessions
    );
    assert_eq!(
        storage.puts.load(std::sync::atomic::Ordering::Relaxed),
        writes
    );
    let retry = engine.group_sender_key_retry.borrow();
    assert!(retry.inputs == before.inputs);
    assert_eq!(retry.ready, before.ready);
    assert_eq!(retry.queued, before.queued);
    assert_eq!(retry.started, before.started);
    assert_eq!(retry.attempts, before.attempts);
    assert_eq!(retry.key_trials_remaining, before.key_trials_remaining);
    assert_eq!(retry.yielded, before.yielded);
    assert_eq!(retry.total_attempts, before.total_attempts);
    assert_eq!(retry.receive_checkpoints, before.receive_checkpoints);
    assert_eq!(
        retry.continuation_ready_comparisons,
        before.continuation_ready_comparisons
    );
    assert_eq!(
        retry
            .active
            .as_ref()
            .map(|active| (&active.fingerprint, active.pending_index)),
        before
            .active
            .as_ref()
            .map(|active| (&active.fingerprint, active.pending_index))
    );
    assert_eq!(
        retry.prepared.as_ref().map(|(fingerprint, _)| fingerprint),
        before.prepared.as_ref().map(|(fingerprint, _)| fingerprint)
    );
}

#[test]
fn retry_query_established_continuations_skip_deferred_queues_without_mutation() {
    for prepared in [false, true] {
        let (mut engine, device, storage) = deferred_group_retry_query_fixture();
        long_group_candidate(&mut engine, &device, 700);
        assert!(engine
            .retry_pending_protocol(NdrUnixSeconds(30))
            .unwrap()
            .is_empty());
        assert!(engine.group_sender_key_retry.borrow().active.is_some());
        if prepared {
            for _ in 0..32 {
                if engine.group_sender_key_retry.borrow().prepared.is_some() {
                    break;
                }
                engine.group_sender_key_retry.borrow_mut().reset_budget();
                engine.advance_group_sender_key_continuation().unwrap();
            }
            assert!(engine.group_sender_key_retry.borrow().prepared.is_some());
        }
        assert!(engine.pending_decrypted_deliveries.is_empty());
        assert_ready_retry_queries_skip_deferred_work(&engine, &storage);

        if prepared {
            assert_eq!(drain_group_retry(&mut engine, NdrUnixSeconds(30)).len(), 1);
            engine.ack_pending_decrypted_deliveries().unwrap();
        } else {
            // An orphan continuation with no ready entry must use the fallback.
            let ready = std::mem::take(&mut engine.group_sender_key_retry.borrow_mut().ready);
            assert!(!engine.has_due_pending_retry_work(NdrUnixSeconds(30)));
            assert_eq!(
                take_inbound_retry_deadline_scans(),
                engine.pending_inbound.len()
            );
            engine.group_sender_key_retry.borrow_mut().ready = ready;
            // Empty durable work must clear stale continuation state normally.
            engine.pending_group_sender_key_messages.retain(|_| false);
        }
        engine.invalidate_pending_inbound_retry_deadline();
        take_inbound_retry_deadline_scans();
        assert!(!engine.has_due_pending_retry_work(NdrUnixSeconds(30)));
        assert_eq!(
            take_inbound_retry_deadline_scans(),
            engine.pending_inbound.len()
        );
        let retry = engine.group_sender_key_retry.borrow();
        assert!(retry.active.is_none() && retry.prepared.is_none());
        assert!(retry.inputs.is_empty() && retry.ready.is_empty() && retry.queued.is_empty());
        drop(retry);
        assert!(!engine.has_due_pending_retry_work(NdrUnixSeconds(81)));
        assert!(engine.has_due_pending_retry_work(NdrUnixSeconds(82)));
        assert_eq!(take_inbound_retry_deadline_scans(), 0);
    }
}

#[test]
fn retry_query_delivery_journal_skips_deferred_queues_until_acknowledged() {
    let (mut engine, _, storage) = deferred_group_retry_query_fixture();
    assert_eq!(drain_group_retry(&mut engine, NdrUnixSeconds(30)).len(), 1);
    assert!(engine.pending_group_sender_key_messages.is_empty());
    assert_eq!(engine.pending_decrypted_deliveries.len(), 1);
    assert_ready_retry_queries_skip_deferred_work(&engine, &storage);
    engine.ack_pending_decrypted_deliveries().unwrap();
    assert!(!engine.has_due_pending_retry_work(NdrUnixSeconds(30)));
    assert_eq!(
        take_inbound_retry_deadline_scans(),
        engine.pending_inbound.len()
    );
    assert!(engine.has_due_pending_retry_work(NdrUnixSeconds(82)));
}

#[test]
fn retry_query_without_continuation_preserves_refreshing_fallback() {
    let (mut engine, _, storage) = deferred_group_retry_query_fixture();
    engine.refresh_group_sender_key_retry_inputs();
    assert_eq!(engine.group_sender_key_retry.borrow().ready.len(), 1);
    let mut group = engine.group_manager.snapshot();
    group.groups[0].revision += 1;
    engine.group_manager = GroupEventManager::from_snapshot(group.clone()).unwrap();
    let writes = storage.puts.load(std::sync::atomic::Ordering::Relaxed);
    engine.invalidate_pending_inbound_retry_deadline();
    take_inbound_retry_deadline_scans();
    assert!(engine.has_due_pending_retry_work(NdrUnixSeconds(30)));
    assert_eq!(
        take_inbound_retry_deadline_scans(),
        engine.pending_inbound.len()
    );
    assert_eq!(engine.group_manager.snapshot(), group);
    assert_eq!(
        storage.puts.load(std::sync::atomic::Ordering::Relaxed),
        writes
    );
    let retry = engine.group_sender_key_retry.borrow();
    assert!(retry.active.is_none() && retry.prepared.is_none());
    assert_eq!(retry.ready.len(), 1);
    assert_eq!(retry.total_attempts, 0);
    assert_eq!(
        retry.inputs[&engine.pending_group_sender_key_messages[0].sender_event_pubkey].revision,
        group.groups[0].revision
    );
}

#[test]
fn group_retry_unchanged_metadata_does_not_repeat_blind_search_or_checkpoint() {
    let (mut engine, sender, sender_device) = queued_group_history(1);
    engine.pending_group_sender_key_messages[0].ciphertext =
        nostr_double_ratchet::SenderKeyState::new(999, [99; 32], 0)
            .encrypt_to_bytes(b"synthetic unknown ciphertext")
            .unwrap()
            .1;
    let mut group_state = engine.group_manager.snapshot();
    let distribution = nostr_double_ratchet::SenderKeyDistribution {
        group_id: "queued-group".into(),
        key_id: 1,
        sender_event_pubkey: group_state.sender_keys[0].sender_event_pubkey,
        chain_key: [9; 32],
        iteration: 0,
        created_at: NdrUnixSeconds(20),
    };
    group_state.sender_keys[0]
        .distribution_history
        .push(distribution.clone());
    engine.group_manager = GroupEventManager::from_snapshot(group_state).unwrap();
    let storage = Arc::new(GroupRetryCountingStorage {
        inner: engine.storage.clone(),
        puts: 0.into(),
        fail_next: false.into(),
    });
    engine.storage = storage.clone();
    let now = NdrUnixSeconds(unix_now().get());
    let started = std::time::Instant::now();
    drain_group_retry(&mut engine, now);
    println!("one failed blind candidate: {:?}", started.elapsed());
    assert_eq!(engine.group_sender_key_retry.borrow().total_attempts, 1);
    assert_eq!(engine.pending_group_sender_key_messages.len(), 1);
    assert!(!engine.has_ready_group_sender_key_retry_work());
    let mut presentation_only = engine.group_manager.snapshot();
    presentation_only.groups[0].name = "Renamed locally".into();
    presentation_only.groups[0].picture = Some("https://example.invalid/picture".into());
    presentation_only.groups[0].updated_at = now;
    engine.group_manager = GroupEventManager::from_snapshot(presentation_only).unwrap();
    assert!(
        !engine.has_ready_group_sender_key_retry_work(),
        "display metadata alone cannot unlock ciphertext"
    );
    let payload = JsonGroupPayloadCodecV1
        .encode_pairwise_command(
            nostr_double_ratchet::GroupPayloadEncodeContext {
                local_device_pubkey: ndr_device(sender_device.public_key()),
                created_at: now,
            },
            &GroupPairwiseCommand::MetadataSnapshot {
                snapshot: engine.group_manager.group("queued-group").unwrap(),
            },
        )
        .unwrap();
    let distribution_payload = JsonGroupPayloadCodecV1
        .encode_pairwise_command(
            nostr_double_ratchet::GroupPayloadEncodeContext {
                local_device_pubkey: ndr_device(sender_device.public_key()),
                created_at: now,
            },
            &GroupPairwiseCommand::SenderKeyDistribution { distribution },
        )
        .unwrap();
    let writes = storage.puts.load(std::sync::atomic::Ordering::Relaxed);
    for payload in [&payload, &distribution_payload]
        .into_iter()
        .cycle()
        .take(6)
    {
        engine
            .process_group_pairwise_payload(
                payload,
                sender.public_key(),
                Some(sender_device.public_key()),
            )
            .unwrap();
        engine.retry_pending_protocol(now).unwrap();
    }
    assert_eq!(
        engine.group_sender_key_retry.borrow().total_attempts,
        1,
        "identical metadata cannot unlock different key material"
    );
    assert_eq!(
        storage.puts.load(std::sync::atomic::Ordering::Relaxed),
        writes,
        "idle retry bookkeeping must not rewrite the durable backlog"
    );
}

#[test]
fn group_retry_budget_is_shared_by_a_whole_outer_batch_and_survives_restart() {
    let (mut engine, _, _) = queued_group_history(20);
    let owner = engine.owner_pubkey;
    let keys = Keys::new(nostr::SecretKey::from_slice(&engine.local_device_secret).unwrap());
    let storage = engine.storage.clone();
    engine.enter_batch();
    let mut delivered = 0;
    for _ in 0..16 {
        delivered += engine
            .retry_pending_protocol(NdrUnixSeconds(30))
            .unwrap()
            .group_result
            .events
            .len();
    }
    assert!(
        delivered <= 8,
        "sixteen background events still share one retry budget"
    );
    engine.exit_batch().unwrap();
    drop(engine);
    let mut restored =
        ProtocolEngine::load_or_create_for_local_device(storage, owner, &keys).unwrap();
    assert_eq!(
        restored.pending_group_sender_key_messages.len(),
        20 - delivered
    );
    assert!(restored.has_due_pending_retry_work(NdrUnixSeconds(30)));
    for _ in 0..20 {
        delivered += restored
            .retry_pending_protocol(NdrUnixSeconds(30))
            .unwrap()
            .group_result
            .events
            .len();
        if restored.pending_group_sender_key_messages.is_empty() {
            break;
        }
    }
    assert_eq!(delivered, 20);
    assert!(!restored.has_ready_group_sender_key_retry_work());
}

#[test]
fn group_retry_error_retains_the_candidate_and_unprocessed_tail() {
    let (mut engine, _, _) = queued_group_history(3);
    // The known-position path surfaces malformed ciphertext as an error.
    engine.pending_group_sender_key_messages[0].encrypted_header = None;
    engine.pending_group_sender_key_messages[0]
        .ciphertext
        .clear();
    let malformed = engine.pending_group_sender_key_messages[0].clone();
    let recovered = drain_group_retry(&mut engine, NdrUnixSeconds(30));
    assert_eq!(recovered.len(), 2);
    assert_eq!(&*engine.pending_group_sender_key_messages, &[malformed]);
    // Another success changes the ratchet once; after that input is attempted,
    // malformed ciphertext cannot spin indefinitely.
    drain_group_retry(&mut engine, NdrUnixSeconds(30));
    assert!(!engine.has_ready_group_sender_key_retry_work());
}

#[test]
fn group_retry_hot_stream_does_not_starve_another_stream_at_the_tail() {
    let (mut engine, _, _) = queued_group_history(20);
    let (other, _, other_device) = queued_group_history(1);
    let mut state = engine.group_manager.snapshot();
    let other_state = other.group_manager.snapshot();
    state.groups[0]
        .members
        .push(other_state.sender_keys[0].sender_owner);
    state.sender_keys.extend(other_state.sender_keys);
    engine.group_manager = GroupEventManager::from_snapshot(state).unwrap();
    for pending in other.pending_group_sender_key_messages.iter().cloned() {
        engine.pending_group_sender_key_messages.push(pending);
    }
    let mut found_tail = false;
    for _ in 0..21 {
        let batch = engine.retry_pending_protocol(NdrUnixSeconds(30)).unwrap();
        found_tail |= batch.group_result.events.iter().any(|event| matches!(event,
            GroupIncomingEvent::Message(message) if message.sender_device == Some(ndr_device(other_device.public_key()))));
        // Advancing the hot stream re-admits its old failures behind the tail.
        engine.has_ready_group_sender_key_retry_work();
        if found_tail {
            break;
        }
    }
    assert!(
        found_tail,
        "new work on the first stream cannot reset FIFO progress"
    );
}

#[test]
fn group_retry_new_revision_resumes_a_pending_ciphertext_after_sibling_sync() {
    let (mut engine, _, _) = queued_group_history(1);
    let mut state = engine.group_manager.snapshot();
    state.groups[0].revision = 0;
    engine.group_manager = GroupEventManager::from_snapshot(state).unwrap();
    engine.retry_pending_protocol(NdrUnixSeconds(30)).unwrap();
    assert_eq!(engine.pending_group_sender_key_messages.len(), 1);
    assert!(!engine.has_ready_group_sender_key_retry_work());
    let mut group = engine.group_manager.group("queued-group").unwrap();
    group.revision = 1;
    assert!(engine.install_device_sync_group(group).unwrap());
    assert!(engine.has_due_pending_retry_work(NdrUnixSeconds(30)));
    let recovered = engine.retry_pending_protocol(NdrUnixSeconds(30)).unwrap();
    assert_eq!(recovered.group_result.events.len(), 1);
    assert!(engine.pending_group_sender_key_messages.is_empty());
}

#[test]
fn group_retry_ratchet_progress_unlocks_an_earlier_out_of_window_candidate() {
    let (mut engine, _, _) = queued_group_history(1);
    let template = engine.pending_group_sender_key_messages[0].clone();
    let mut bridge = None;
    let mut future = None;
    // Keep the production 10,000-key search window. A later queued message
    // advances the chain enough to unlock the previously failed ciphertext.
    let decoded = JsonGroupPayloadCodecV1
        .encode_sender_key_plaintext(
            nostr_double_ratchet::GroupPayloadEncodeContext {
                local_device_pubkey: engine.group_manager.snapshot().sender_keys[0].sender_device,
                created_at: NdrUnixSeconds(20),
            },
            &nostr_double_ratchet::GroupSenderKeyPlaintext {
                group_id: "queued-group".into(),
                revision: 1,
                body: b"recovered".to_vec(),
            },
        )
        .unwrap();
    let mut encryptor = nostr_double_ratchet::SenderKeyState::new(1, [9; 32], 0);
    for number in 0..=10_001 {
        let (_, ciphertext) = encryptor.encrypt_to_bytes(&decoded).unwrap();
        if number == 5_000 {
            bridge = Some(ciphertext.clone());
        }
        if number == 10_001 {
            future = Some(ciphertext);
        }
    }
    engine.pending_group_sender_key_messages = vec![
        nostr_double_ratchet::wire::ParsedGroupSenderKeyMessageEvent {
            ciphertext: future.unwrap(),
            ..template.clone()
        },
        nostr_double_ratchet::wire::ParsedGroupSenderKeyMessageEvent {
            ciphertext: bridge.unwrap(),
            ..template
        },
    ]
    .into();
    let mut delivered = 0;
    for _ in 0..256 {
        delivered += engine
            .retry_pending_protocol(NdrUnixSeconds(30))
            .unwrap()
            .group_result
            .events
            .len();
        if engine.pending_group_sender_key_messages.is_empty() {
            break;
        }
    }
    assert_eq!(delivered, 2);
    assert!(engine.pending_group_sender_key_messages.is_empty());
}

#[test]
fn group_retry_storage_failure_restores_ratchet_and_delivers_once_after_retry() {
    let (mut engine, _, _) = queued_group_history(3);
    engine.persist().unwrap();
    let before = engine.pending_group_sender_key_messages.clone();
    let group_before = engine.group_manager.snapshot();
    let storage = Arc::new(GroupRetryCountingStorage {
        inner: engine.storage.clone(),
        puts: 0.into(),
        fail_next: true.into(),
    });
    engine.storage = storage.clone();
    let error = engine
        .retry_pending_protocol(NdrUnixSeconds(30))
        .unwrap_err();
    assert!(error.downcast_ref::<StorageError>().is_some());
    assert_eq!(engine.pending_group_sender_key_messages, before);
    assert_eq!(engine.group_manager.snapshot(), group_before);
    assert!(engine.pending_decrypted_deliveries.is_empty());
    assert!(engine.has_ready_group_sender_key_retry_work());
    let mut delivered = 0;
    for _ in 0..3 {
        delivered += engine
            .retry_pending_protocol(NdrUnixSeconds(30))
            .unwrap()
            .group_result
            .events
            .len();
        if engine.pending_group_sender_key_messages.is_empty() {
            break;
        }
    }
    assert_eq!(delivered, 3);
    assert!(engine.pending_group_sender_key_messages.is_empty());
    assert_eq!(engine.pending_decrypted_deliveries.len(), 3);
    engine.ack_pending_decrypted_deliveries().unwrap();
    assert!(engine
        .retry_pending_protocol(NdrUnixSeconds(30))
        .unwrap()
        .is_empty());
    let keys = Keys::new(nostr::SecretKey::from_slice(&engine.local_device_secret).unwrap());
    let mut restored =
        ProtocolEngine::load_or_create_for_local_device(storage, engine.owner_pubkey, &keys)
            .unwrap();
    assert!(restored.pending_group_sender_key_messages.is_empty());
    assert!(restored
        .retry_pending_protocol(NdrUnixSeconds(30))
        .unwrap()
        .is_empty());
}

#[test]
fn group_retry_new_candidate_on_unchanged_stream_is_admitted_once() {
    let (mut engine, _, _) = queued_group_history(2);
    let next = engine.pending_group_sender_key_messages.pop().unwrap();
    engine.pending_group_sender_key_messages[0].ciphertext =
        nostr_double_ratchet::SenderKeyState::new(999, [99; 32], 0)
            .encrypt_to_bytes(b"synthetic unknown ciphertext")
            .unwrap()
            .1;
    drain_group_retry(&mut engine, NdrUnixSeconds(30));
    assert!(!engine.has_ready_group_sender_key_retry_work());
    engine
        .queue_pending_group_sender_key_message(next.clone())
        .unwrap();
    assert!(engine.has_ready_group_sender_key_retry_work());
    let recovered = drain_group_retry(&mut engine, NdrUnixSeconds(30));
    assert_eq!(recovered.len(), 1);
    engine.queue_pending_group_sender_key_message(next).unwrap();
    drain_group_retry(&mut engine, NdrUnixSeconds(30));
    assert!(!engine.has_ready_group_sender_key_retry_work());
}

#[test]
fn group_retry_large_valid_history_advances_in_bounded_passes() {
    let (mut engine, _, _) = queued_group_history(512);
    let started = std::time::Instant::now();
    let mut delivered = 0;
    let mut passes = 0;
    let mut longest = std::time::Duration::ZERO;
    while engine.has_ready_group_sender_key_retry_work() {
        let pass_started = std::time::Instant::now();
        engine.enter_batch();
        let batch = engine.retry_pending_protocol(NdrUnixSeconds(30)).unwrap();
        assert!(batch.group_result.events.len() <= GROUP_SENDER_KEY_RETRY_LIMIT);
        delivered += batch.group_result.events.len();
        engine.exit_batch().unwrap();
        longest = longest.max(pass_started.elapsed());
        passes += 1;
        assert!(passes <= 512, "every ready pass must progress");
    }
    assert_eq!(delivered, 512);
    println!(
        "valid history: messages={delivered} passes={passes} total_ms={:.3} max_pass_ms={:.3}",
        started.elapsed().as_secs_f64() * 1000.0,
        longest.as_secs_f64() * 1000.0
    );
}

#[test]
fn group_retry_checkpoint_restores_eligibility_without_restarting_forever() {
    let (mut engine, _, _) = queued_group_history(1);
    assert!(engine.has_ready_group_sender_key_retry_work());
    let before = engine.state_checkpoint();
    assert_eq!(
        engine
            .retry_pending_protocol(NdrUnixSeconds(30))
            .unwrap()
            .group_result
            .events
            .len(),
        1
    );
    assert!(!engine.has_ready_group_sender_key_retry_work());
    engine.restore_checkpoint(before);
    assert!(engine.has_ready_group_sender_key_retry_work());
    assert_eq!(
        engine
            .retry_pending_protocol(NdrUnixSeconds(30))
            .unwrap()
            .group_result
            .events
            .len(),
        1
    );
    assert!(!engine.has_ready_group_sender_key_retry_work());
}

#[test]
fn group_sender_key_delivery_journal_preserves_author_scope_and_survives_restart() {
    let (mut engine, sender, sender_device) = queued_group_history(1);
    let owner = engine.owner_pubkey;
    let keys = Keys::new(nostr::SecretKey::from_slice(&engine.local_device_secret).unwrap());
    let parsed = engine.pending_group_sender_key_messages.pop().unwrap();
    let message = engine
        .group_sender_key_message_from_parsed(&parsed)
        .unwrap();
    let result = engine
        .handle_group_sender_key_message(message.clone())
        .unwrap();
    assert_eq!(result.events.len(), 1);
    assert_eq!(engine.pending_decrypted_deliveries.len(), 1);
    assert!(engine
        .handle_group_sender_key_message(message)
        .unwrap()
        .events
        .is_empty());
    let storage = engine.storage.clone();
    drop(engine);
    let mut restored =
        ProtocolEngine::load_or_create_for_local_device(storage, owner, &keys).unwrap();
    let batch = restored.retry_pending_protocol(NdrUnixSeconds(30)).unwrap();
    assert_eq!(batch.direct_messages.len(), 1);
    let delivery = &batch.direct_messages[0];
    assert_eq!(delivery.sender, sender.public_key());
    assert_eq!(delivery.sender_device, Some(sender_device.public_key()));
    assert_eq!(delivery.created_at_secs, 20);
    let Some(GroupPairwiseCommand::GroupMessage {
        group_id,
        revision,
        body,
    }) = JsonGroupPayloadCodecV1
        .decode_pairwise_command(delivery.content.as_bytes())
        .unwrap()
    else {
        panic!("scoped group wrapper");
    };
    assert_eq!(group_id, "queued-group");
    assert_eq!(revision, 1);
    assert_eq!(body, b"history-0");
    let Some(GroupIncomingEvent::Message(received)) = restored
        .pending_group_decrypted_delivery(delivery.event_id.as_deref().unwrap())
        .unwrap()
    else {
        panic!("journaled group message");
    };
    assert_eq!(received.group_id, group_id);
    assert_eq!(received.sender_owner, ndr_owner(sender.public_key()));
    assert_eq!(
        received.sender_device,
        Some(ndr_device(sender_device.public_key()))
    );
    assert_eq!(received.body, body);
    assert!(restored
        .pending_group_decrypted_delivery("group-delivery:forged")
        .is_err());
    assert!(
        restored
            .process_group_pairwise_payload(
                delivery.content.as_bytes(),
                delivery.sender,
                delivery.sender_device
            )
            .unwrap()
            .events
            .is_empty(),
        "external legacy group-message payloads remain unsupported"
    );
    restored
        .ack_decrypted_delivery_ids(&HashSet::from(["unrelated".into()]))
        .unwrap();
    assert_eq!(restored.pending_decrypted_deliveries.len(), 1);
    restored
        .ack_decrypted_delivery_ids(&HashSet::from([delivery.event_id.clone().unwrap()]))
        .unwrap();
    assert!(restored
        .retry_pending_protocol(NdrUnixSeconds(31))
        .unwrap()
        .direct_messages
        .is_empty());
}

#[test]
fn group_sender_key_receive_storage_failure_retries_without_losing_plaintext() {
    let (mut engine, _, _) = queued_group_history(1);
    let parsed = engine.pending_group_sender_key_messages.pop().unwrap();
    let message = engine
        .group_sender_key_message_from_parsed(&parsed)
        .unwrap();
    let before = engine.group_manager.snapshot();
    engine.persist().unwrap();
    engine.storage = Arc::new(GroupRetryCountingStorage {
        inner: engine.storage.clone(),
        puts: 0.into(),
        fail_next: true.into(),
    });
    assert!(engine
        .handle_group_sender_key_message(message.clone())
        .unwrap_err()
        .is::<StorageError>());
    assert_eq!(engine.group_manager.snapshot(), before);
    assert!(engine.pending_decrypted_deliveries.is_empty());
    assert!(!engine
        .processed_group_sender_key_messages
        .contains(&group_sender_key_fingerprint(&message)));
    let result = engine.handle_group_sender_key_message(message).unwrap();
    assert_eq!(result.events.len(), 1);
    assert_eq!(engine.pending_decrypted_deliveries.len(), 1);
}