iris_chat_protocol/protocol_engine/
engine_pending_retirement.rs

1use super::*;
2
3impl ProtocolEngine {
4    /// Runtime hold while private block state is being applied. The owning app
5    /// restores it from signed account state before allowing retry work.
6    pub fn hold_pending_direct_sends(&mut self, peer: PublicKey, hold: bool) {
7        if hold {
8            self.held_direct_chats.insert(peer.to_hex());
9        } else {
10            self.held_direct_chats.remove(&peer.to_hex());
11        }
12    }
13
14    /// Durably cancel unsent direct-chat intents after a local block. Group
15    /// membership/key controls addressed to this peer keep their group chat ID
16    /// and are retained, as are every other conversation's pending deliveries.
17    pub fn retire_pending_direct_sends(
18        &mut self,
19        peer: PublicKey,
20        through: Option<u64>,
21    ) -> anyhow::Result<usize> {
22        let chat_id = peer.to_hex();
23        self.with_state_checkpoint(|engine| {
24            let before = engine.pending_remote_sends.len()
25                + engine.pending_local_sibling_sends.len()
26                + engine.ready_group_publications.len()
27                + engine.pending_group_fanouts.len();
28            let fanouts_before = engine.pending_group_fanouts.len();
29            engine.pending_group_fanouts.retain(|pending| {
30                let GroupPendingFanout::Remote {
31                    recipient_owner,
32                    payload,
33                } = pending.fanout.as_ref()
34                else {
35                    return true;
36                };
37                *recipient_owner != ndr_owner(peer)
38                    || !serde_json::from_slice::<UnsignedEvent>(payload)
39                        .is_ok_and(|event| event.kind.as_u16() == 21_113)
40                    || through.is_some_and(|cutoff| {
41                        pending_direct_authored_at(payload, pending.created_at_secs) > cutoff
42                    })
43            });
44            if engine.pending_group_fanouts.len() != fanouts_before {
45                engine.invalidate_pending_group_fanout_retry_deadline();
46            }
47            engine.ready_group_publications.retain(|_, pending| {
48                pending.wake_recipient != Some(ndr_owner(peer))
49                    || through.is_some_and(|cutoff| {
50                        pending
51                            .publish
52                            .authored_at_secs
53                            .unwrap_or(pending.publish.event.created_at.as_secs())
54                            > cutoff
55                    })
56            });
57            engine.pending_remote_sends.retain(|pending| {
58                pending.recipient_owner != ndr_owner(peer)
59                    || pending.chat_id != chat_id
60                    || through.is_some_and(|cutoff| {
61                        pending_direct_authored_at(&pending.payload, pending.created_at_secs)
62                            > cutoff
63                    })
64            });
65            engine.pending_local_sibling_sends.retain(|pending| {
66                pending.chat_id != chat_id
67                    || through.is_some_and(|cutoff| {
68                        decode_local_sibling_payload(&pending.payload).map_or(
69                            pending.created_at_secs,
70                            |(_, _, payload)| {
71                                pending_direct_authored_at(&payload, pending.created_at_secs)
72                            },
73                        ) > cutoff
74                    })
75            });
76            let retired = before
77                - engine.pending_remote_sends.len()
78                - engine.pending_local_sibling_sends.len()
79                - engine.ready_group_publications.len()
80                - engine.pending_group_fanouts.len();
81            if retired > 0 {
82                // A surrounding batch must not defer this privacy barrier.
83                engine.persist_now()?;
84            }
85            Ok(retired)
86        })
87    }
88
89    /// Retire matching plaintext sibling intents, including intents waiting for
90    /// roster discovery. Already sealed events and all other queues are untouched.
91    /// Callers must durably migrate the source data first and must not expose this
92    /// engine for sending if retirement fails.
93    pub fn retire_pending_local_sibling_events(
94        &mut self,
95        matches: impl Fn(PublicKey, &UnsignedEvent) -> bool,
96    ) -> anyhow::Result<usize> {
97        self.with_state_checkpoint(|engine| {
98            let before = engine.pending_local_sibling_sends.len();
99            engine.pending_local_sibling_sends.retain(|pending| {
100                let Some((owner, _, payload)) = decode_local_sibling_payload(&pending.payload)
101                else {
102                    return true;
103                };
104                !serde_json::from_slice::<UnsignedEvent>(&payload)
105                    .is_ok_and(|event| matches(owner, &event))
106            });
107            let retired = before - engine.pending_local_sibling_sends.len();
108            if retired > 0 {
109                // This upgrade barrier cannot be deferred by a surrounding batch.
110                engine.persist_now()?;
111            }
112            Ok(retired)
113        })
114    }
115}
116
117pub(crate) fn pending_direct_authored_at(payload: &[u8], fallback: u64) -> u64 {
118    serde_json::from_slice::<UnsignedEvent>(payload)
119        .map_or(fallback, |event| event.created_at.as_secs())
120}
iris_chat_protocol/protocol_engine/engine_pending_retirement.rs

iris_chat_protocol/protocol_engine/
engine_pending_retirement.rs
