iris_chat_protocol/protocol_engine/
engine_fact_ingest.rs

1use super::*;
2
3impl ProtocolEngine {
4    /// Install metadata copied from an authenticated sibling device. Signed
5    /// roster facts can advance this copy, but restoring removed membership
6    /// requires a newer revision.
7    pub fn install_device_sync_group(&mut self, snapshot: GroupSnapshot) -> anyhow::Result<bool> {
8        if self
9            .group_manager
10            .group(&snapshot.group_id)
11            .is_some_and(|current| current.protocol != snapshot.protocol)
12            && self
13                .group_roster_fact_histories
14                .get(&snapshot.group_id)
15                .is_some_and(|history| {
16                    project_group_roster_fact_events(history.events.iter())
17                        .iter()
18                        .any(|signed| signed.protocol != snapshot.protocol)
19                })
20        {
21            // A sibling copy cannot change the protocol established by signed
22            // roster facts. Consult history only for conflicting protocols;
23            // ordinary metadata sync needs no additional event verification.
24            return Ok(false);
25        }
26        let checkpoint = self.state_checkpoint();
27        let applied = self.install_group_roster_snapshot(snapshot)?;
28        if applied {
29            if let Err(error) = self.persist() {
30                self.restore_checkpoint(checkpoint);
31                return Err(error);
32            }
33        }
34        Ok(applied)
35    }
36
37    pub fn ingest_group_roster_fact_event(
38        &mut self,
39        event: &Event,
40    ) -> anyhow::Result<Option<ProtocolGroupRosterIngestionResult>> {
41        if !is_group_roster_fact_event(event) {
42            return Ok(None);
43        }
44        let fact = parse_group_roster_fact_event(event)?;
45        if !self.group_roster_fact_is_authorized(&fact) {
46            return Ok(None);
47        }
48
49        let checkpoint = self.state_checkpoint();
50        let Some(snapshot) = self.remember_group_roster_fact_event(&fact.group_id, event) else {
51            return Ok(None);
52        };
53        let applied = match self.install_group_roster_snapshot(snapshot.clone()) {
54            Ok(applied) => applied,
55            Err(error) => {
56                self.restore_checkpoint(checkpoint);
57                return Err(error);
58            }
59        };
60        let mut retry_batch = if applied {
61            match self.retry_pending_protocol(NdrUnixSeconds(event.created_at.as_secs())) {
62                Ok(retry_batch) => retry_batch,
63                Err(error) => {
64                    self.restore_checkpoint(checkpoint);
65                    return Err(error);
66                }
67            }
68        } else {
69            ProtocolRetryBatch::default()
70        };
71        if applied {
72            match self.sync_group_to_local_siblings(&snapshot) {
73                Ok(effects) => {
74                    retry_batch.group_result.effects.extend(effects);
75                }
76                Err(error) => {
77                    self.restore_checkpoint(checkpoint);
78                    return Err(error);
79                }
80            }
81        }
82        if let Err(error) = self.persist() {
83            self.restore_checkpoint(checkpoint);
84            return Err(error);
85        }
86
87        Ok(Some(ProtocolGroupRosterIngestionResult {
88            snapshot: applied.then_some(snapshot),
89            retry_batch,
90        }))
91    }
92
93    fn remember_group_roster_fact_event(
94        &mut self,
95        group_id: &str,
96        event: &Event,
97    ) -> Option<GroupSnapshot> {
98        let history = self
99            .group_roster_fact_histories
100            .entry(group_id.to_string())
101            .or_default();
102        if history
103            .events
104            .iter()
105            .any(|existing| existing.id == event.id)
106        {
107            return None;
108        }
109        history.events.push(event.clone());
110        history.events.sort_by(|left, right| {
111            left.created_at
112                .as_secs()
113                .cmp(&right.created_at.as_secs())
114                .then_with(|| left.id.to_hex().cmp(&right.id.to_hex()))
115        });
116        let excess = history
117            .events
118            .len()
119            .saturating_sub(GROUP_ROSTER_FACT_EVENT_HISTORY_LIMIT);
120        if excess > 0 {
121            history.events.drain(0..excess);
122        }
123        project_group_roster_fact_events(history.events.iter())
124            .into_iter()
125            .find(|snapshot| snapshot.group_id == group_id)
126    }
127
128    fn group_roster_fact_is_authorized(&self, fact: &GroupRosterFact) -> bool {
129        let signer_owner = ndr_owner(fact.signer_pubkey);
130        if let Some(current) = self.group_manager.group(&fact.group_id) {
131            return current.admins.contains(&signer_owner);
132        }
133
134        let local_owner = self.group_manager.snapshot().local_owner_pubkey;
135        fact.snapshot.admins.contains(&signer_owner) && fact.snapshot.members.contains(&local_owner)
136    }
137
138    pub(super) fn install_group_roster_snapshot(
139        &mut self,
140        snapshot: GroupSnapshot,
141    ) -> anyhow::Result<bool> {
142        if self.group_roster_snapshot_is_stale(&snapshot) {
143            return Ok(false);
144        }
145
146        let mut manager_snapshot = self.group_manager.snapshot();
147        match manager_snapshot
148            .groups
149            .iter_mut()
150            .find(|group| group.group_id == snapshot.group_id)
151        {
152            Some(group) if *group == snapshot => return Ok(false),
153            Some(group) => *group = snapshot,
154            None => manager_snapshot.groups.push(snapshot),
155        }
156        manager_snapshot
157            .groups
158            .sort_by(|left, right| left.group_id.cmp(&right.group_id));
159        self.group_manager = GroupEventManager::from_snapshot(manager_snapshot)?;
160        for pending in &mut self.pending_group_pairwise_payloads {
161            pending.next_retry_at_secs = 0;
162        }
163        Ok(true)
164    }
165
166    fn group_roster_snapshot_is_stale(&self, incoming: &GroupSnapshot) -> bool {
167        self.group_manager
168            .group(&incoming.group_id)
169            .is_some_and(|current| {
170                current.revision > incoming.revision
171                    || (current.revision == incoming.revision
172                        && current.updated_at > incoming.updated_at)
173                    // Sibling snapshots do not carry a signed event tie-breaker.
174                    // A competing fact at the same revision cannot undo removal.
175                    || (incoming.revision <= current.revision
176                        && !current.members.contains(&self.local_owner)
177                        && incoming.members.contains(&self.local_owner))
178            })
179    }
180}
iris_chat_protocol/protocol_engine/engine_fact_ingest.rs

iris_chat_protocol/protocol_engine/
engine_fact_ingest.rs
