iris-chat-protocol 0.3.19

use super::*;

#[derive(Clone)]
pub(super) struct ActiveGroupSenderKeyDecrypt {
    pub(super) fingerprint: String,
    cursor: nostr_double_ratchet::GroupSenderKeyDecryptCursor,
    message: GroupSenderKeyMessage,
    // Validated against the durable queue before each continuation. Only a queue
    // mutation can force a scan; ordinary slices touch just this one ciphertext.
    pub(super) pending_index: usize,
}

impl ActiveGroupSenderKeyDecrypt {
    pub(super) fn matches(
        &self,
        parsed: &nostr_double_ratchet::wire::ParsedGroupSenderKeyMessageEvent,
    ) -> bool {
        self.message.sender_event_pubkey == parsed.sender_event_pubkey
            && self.message.created_at == parsed.created_at
            && self.message.key_id == parsed.key_id
            && self.message.message_number == parsed.message_number
            && self.message.encrypted_header == parsed.encrypted_header
            && self.message.ciphertext == parsed.ciphertext
    }
}

impl ProtocolEngine {
    pub(super) fn prepare_group_sender_key_message(
        &mut self,
        message: &GroupSenderKeyMessage,
        fingerprint: &str,
    ) -> nostr_double_ratchet::Result<Option<nostr_double_ratchet::GroupSenderKeyReceivePlan>> {
        {
            let mut retry = self.group_sender_key_retry.borrow_mut();
            retry.yielded = false;
            if retry
                .prepared
                .as_ref()
                .is_some_and(|(ready, _)| ready == fingerprint)
            {
                return Ok(retry.prepared.take().map(|(_, plan)| plan));
            }
        }
        if message.encrypted_header.is_none() {
            let mut remaining = usize::MAX;
            return self.group_manager.plan_sender_key_message_with_budget(
                message.clone(),
                &mut None,
                &mut remaining,
            );
        }
        {
            let mut retry = self.group_sender_key_retry.borrow_mut();
            if let Some(index) = retry
                .fresh
                .iter()
                .position(|fresh| fresh.fingerprint == fingerprint)
            {
                if index != 0 || !self.group_retry_pacing.borrow().fresh_delay().is_zero() {
                    retry.yielded = true;
                    return Ok(None);
                }
                retry.fresh.pop_front();
                drop(retry);
                let result = self.prepare_fresh_group_sender_key_message(message);
                self.group_sender_key_retry.borrow_mut().yielded = matches!(result, Ok(None));
                return result;
            }
        }
        if !self.group_retry_pacing.borrow().delay().is_zero() {
            self.group_sender_key_retry.borrow_mut().yielded = true;
            return Ok(None);
        }
        let (active, mut remaining) = {
            let mut retry = self.group_sender_key_retry.borrow_mut();
            if retry
                .active
                .as_ref()
                .is_some_and(|active| active.fingerprint != fingerprint)
                || retry.key_trials_remaining == 0
            {
                retry.yielded = true;
                return Ok(None);
            }
            (retry.active.take(), retry.key_trials_remaining)
        };
        let (mut cursor, cached_message, pending_index) = active.map_or_else(
            || (None, message.clone(), usize::MAX),
            |active| (Some(active.cursor), active.message, active.pending_index),
        );
        let decrypt_started = std::time::Instant::now();
        #[cfg(test)]
        let trials_before = remaining;
        let result = self.group_manager.plan_sender_key_message_with_budget(
            message.clone(),
            &mut cursor,
            &mut remaining,
        );
        let decrypt_elapsed = decrypt_started.elapsed();
        #[cfg(test)]
        {
            let mut pacing = self.group_retry_pacing.borrow_mut();
            pacing.key_trials += trials_before.saturating_sub(remaining);
            pacing.work_calls += 1;
        }
        self.group_retry_pacing
            .borrow_mut()
            .record_work(decrypt_elapsed);
        #[cfg(all(test, target_os = "linux"))]
        tests::audit_group_retry_decrypt_elapsed(decrypt_elapsed);
        let mut retry = self.group_sender_key_retry.borrow_mut();
        retry.key_trials_remaining = remaining;
        retry.yielded = matches!(result, Ok(None));
        retry.active = cursor.map(|cursor| ActiveGroupSenderKeyDecrypt {
            fingerprint: fingerprint.to_owned(),
            cursor,
            message: cached_message,
            pending_index,
        });
        result
    }

    /// Pure search does not clone rollback state or write a checkpoint per slice.
    /// Final ratchet application still uses the caller's atomic persistence batch.
    pub(super) fn advance_group_sender_key_continuation(&mut self) -> anyhow::Result<bool> {
        if self.group_sender_key_retry.borrow().active.is_some()
            && !self.group_retry_pacing.borrow().delay().is_zero()
        {
            // Repeated unrelated core events must not clone the ciphertext or
            // advance the same expensive cursor before its next work allowance.
            return Ok(true);
        }
        let Some(mut active) = self.group_sender_key_retry.borrow_mut().active.take() else {
            return Ok(false);
        };
        let matches_cached = self
            .pending_group_sender_key_messages
            .get(active.pending_index)
            .is_some_and(|parsed| active.matches(parsed));
        if !matches_cached {
            let Some(index) = self
                .pending_group_sender_key_messages
                .iter()
                .position(|parsed| active.matches(parsed))
            else {
                return Ok(false);
            };
            active.pending_index = index;
        }
        if self.local_owner_is_inactive_for_group(&active.message.group_id)
            || self.unmapped_group_sender_key_candidate_is_known_message_author(
                &self.pending_group_sender_key_messages[active.pending_index],
            )
        {
            return Ok(false);
        }
        let fingerprint = active.fingerprint.clone();
        let message = active.message.clone();
        self.group_sender_key_retry.borrow_mut().active = Some(active);
        let result = self.prepare_group_sender_key_message(&message, &fingerprint)?;
        let mut retry = self.group_sender_key_retry.borrow_mut();
        retry.remove_ready_for_continuation(&fingerprint);
        retry.queued.insert(fingerprint.clone());
        let pending = result.is_none();
        if let Some(plan) = result {
            retry.prepared = Some((fingerprint.clone(), plan));
        }
        if retry.active.is_some() || retry.prepared.is_some() {
            retry
                .ready
                .push_front((message.sender_event_pubkey, fingerprint));
        } else {
            // A changed active ratchet cancels this generation. Requeue behind
            // other senders so fresh traffic cannot repeatedly win the front.
            retry
                .ready
                .push_back((message.sender_event_pubkey, fingerprint));
        }
        Ok(pending)
    }
}