iris_chat_protocol/protocol_engine/
engine_group_wakes.rs

1use super::*;
2
3pub(super) const MAX_READY_GROUP_PUBLICATIONS: usize = 4_096;
4
5impl ProtocolEngine {
6    pub fn owns_group_data_stream(&self, chat_id: &str, author: PublicKey) -> bool {
7        let Some(group) = chat_id.strip_prefix("group:") else {
8            return false;
9        };
10        self.group_manager.group_has_member(group, self.local_owner) == Some(true)
11            && self.group_manager.is_local_sender_key_stream(
12                group,
13                ndr_device(author),
14                self.local_device,
15            )
16    }
17
18    /// Authors learned from private distributions for groups this device belongs to.
19    pub fn subscribable_group_sender_event_pubkeys(&self) -> Vec<PublicKey> {
20        self.group_manager
21            .subscribable_sender_event_pubkeys()
22            .into_iter()
23            .filter_map(|a| public_device(a).ok())
24            .collect()
25    }
26
27    fn ready_group_publication_is_allowed(&self, pending: &ProtocolReadyGroupPublish) -> bool {
28        // Envelopes sealed before carriers were recorded still name their
29        // recipient device. They are retired here, never offered again.
30        pending.publish.is_publishable()
31            && (pending.publish.carrier != ProtocolCarrier::SharedGroup
32                || self
33                    .owns_group_data_stream(&pending.publish.chat_id, pending.publish.event.pubkey))
34            && !pending
35                .publish
36                .chat_id
37                .strip_prefix("group:")
38                .is_some_and(|group| self.local_owner_is_inactive_for_group(group))
39            && !pending
40                .wake_recipient
41                .is_some_and(|owner| self.held_direct_chats.contains(&owner.to_hex()))
42    }
43
44    pub(super) fn retire_ineligible_group_publications(&mut self) -> anyhow::Result<()> {
45        let rejected = self
46            .ready_group_publications
47            .iter()
48            .filter(|(_, pending)| !self.ready_group_publication_is_allowed(pending))
49            .map(|(id, _)| id.clone())
50            .collect::<Vec<_>>();
51        if rejected.is_empty() {
52            return Ok(());
53        }
54        self.with_state_checkpoint(|engine| {
55            for id in rejected {
56                engine.ready_group_publications.remove(&id);
57            }
58            engine.persist_now()
59        })
60    }
61
62    pub(super) fn retain_group_publications(
63        &mut self,
64        effects: &[ProtocolEffect],
65        wake_recipient: Option<NdrOwnerPubkey>,
66    ) -> anyhow::Result<()> {
67        let added = effects
68            .iter()
69            .filter(|effect| {
70                let ProtocolEffect::Publish(publish) = effect;
71                !self
72                    .ready_group_publications
73                    .contains_key(&publish.event.id.to_hex())
74            })
75            .count();
76        anyhow::ensure!(
77            self.ready_group_publications.len().saturating_add(added)
78                <= MAX_READY_GROUP_PUBLICATIONS,
79            "Outgoing group queue is full"
80        );
81        for effect in effects {
82            let ProtocolEffect::Publish(publish) = effect;
83            self.ready_group_publications
84                .entry(publish.event.id.to_hex())
85                .or_insert_with(|| ProtocolReadyGroupPublish {
86                    publish: publish.clone(),
87                    wake_recipient,
88                });
89        }
90        Ok(())
91    }
92
93    /// Ready group ciphertext remains here until the application's durable
94    /// outbox owns it. Reoffering the same signed events never advances ratchets.
95    pub fn pending_group_publish_effects(&self) -> Vec<ProtocolEffect> {
96        self.ready_group_publications
97            .values()
98            .filter(|pending| self.ready_group_publication_is_allowed(pending))
99            .map(|pending| ProtocolEffect::Publish(pending.publish.clone()))
100            .collect()
101    }
102
103    pub fn has_pending_group_publish(&self, event_id: &str) -> bool {
104        self.ready_group_publications.contains_key(event_id)
105    }
106
107    /// Call only after the complete publication batch is committed to the
108    /// application outbox (or durably rejected by its block/removal policy).
109    pub fn ack_group_publish_effects(&mut self, effects: &[ProtocolEffect]) -> anyhow::Result<()> {
110        if !effects.iter().any(|effect| {
111            let ProtocolEffect::Publish(publish) = effect;
112            self.has_pending_group_publish(&publish.event.id.to_hex())
113        }) {
114            return Ok(());
115        }
116        self.with_state_checkpoint(|engine| {
117            for effect in effects {
118                let ProtocolEffect::Publish(publish) = effect;
119                engine
120                    .ready_group_publications
121                    .remove(&publish.event.id.to_hex());
122            }
123            engine.persist_now()
124        })
125    }
126}
iris_chat_protocol/protocol_engine/engine_group_wakes.rs

iris_chat_protocol/protocol_engine/
engine_group_wakes.rs
