iris-chat-protocol 0.3.17

Reusable Iris Chat double-ratchet protocol engine
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120

use super::*;

impl ProtocolEngine {
    /// Runtime hold while private block state is being applied. The owning app
    /// restores it from signed account state before allowing retry work.
    pub fn hold_pending_direct_sends(&mut self, peer: PublicKey, hold: bool) {
        if hold {
            self.held_direct_chats.insert(peer.to_hex());
        } else {
            self.held_direct_chats.remove(&peer.to_hex());
        }
    }

    /// Durably cancel unsent direct-chat intents after a local block. Group
    /// membership/key controls addressed to this peer keep their group chat ID
    /// and are retained, as are every other conversation's pending deliveries.
    pub fn retire_pending_direct_sends(
        &mut self,
        peer: PublicKey,
        through: Option<u64>,
    ) -> anyhow::Result<usize> {
        let chat_id = peer.to_hex();
        self.with_state_checkpoint(|engine| {
            let before = engine.pending_remote_sends.len()
                + engine.pending_local_sibling_sends.len()
                + engine.ready_group_publications.len()
                + engine.pending_group_fanouts.len();
            let fanouts_before = engine.pending_group_fanouts.len();
            engine.pending_group_fanouts.retain(|pending| {
                let GroupPendingFanout::Remote {
                    recipient_owner,
                    payload,
                } = pending.fanout.as_ref()
                else {
                    return true;
                };
                *recipient_owner != ndr_owner(peer)
                    || !serde_json::from_slice::<UnsignedEvent>(payload)
                        .is_ok_and(|event| event.kind.as_u16() == 21_113)
                    || through.is_some_and(|cutoff| {
                        pending_direct_authored_at(payload, pending.created_at_secs) > cutoff
                    })
            });
            if engine.pending_group_fanouts.len() != fanouts_before {
                engine.invalidate_pending_group_fanout_retry_deadline();
            }
            engine.ready_group_publications.retain(|_, pending| {
                pending.wake_recipient != Some(ndr_owner(peer))
                    || through.is_some_and(|cutoff| {
                        pending
                            .publish
                            .authored_at_secs
                            .unwrap_or(pending.publish.event.created_at.as_secs())
                            > cutoff
                    })
            });
            engine.pending_remote_sends.retain(|pending| {
                pending.recipient_owner != ndr_owner(peer)
                    || pending.chat_id != chat_id
                    || through.is_some_and(|cutoff| {
                        pending_direct_authored_at(&pending.payload, pending.created_at_secs)
                            > cutoff
                    })
            });
            engine.pending_local_sibling_sends.retain(|pending| {
                pending.chat_id != chat_id
                    || through.is_some_and(|cutoff| {
                        decode_local_sibling_payload(&pending.payload).map_or(
                            pending.created_at_secs,
                            |(_, _, payload)| {
                                pending_direct_authored_at(&payload, pending.created_at_secs)
                            },
                        ) > cutoff
                    })
            });
            let retired = before
                - engine.pending_remote_sends.len()
                - engine.pending_local_sibling_sends.len()
                - engine.ready_group_publications.len()
                - engine.pending_group_fanouts.len();
            if retired > 0 {
                // A surrounding batch must not defer this privacy barrier.
                engine.persist_now()?;
            }
            Ok(retired)
        })
    }

    /// Retire matching plaintext sibling intents, including intents waiting for
    /// roster discovery. Already sealed events and all other queues are untouched.
    /// Callers must durably migrate the source data first and must not expose this
    /// engine for sending if retirement fails.
    pub fn retire_pending_local_sibling_events(
        &mut self,
        matches: impl Fn(PublicKey, &UnsignedEvent) -> bool,
    ) -> anyhow::Result<usize> {
        self.with_state_checkpoint(|engine| {
            let before = engine.pending_local_sibling_sends.len();
            engine.pending_local_sibling_sends.retain(|pending| {
                let Some((owner, _, payload)) = decode_local_sibling_payload(&pending.payload)
                else {
                    return true;
                };
                !serde_json::from_slice::<UnsignedEvent>(&payload)
                    .is_ok_and(|event| matches(owner, &event))
            });
            let retired = before - engine.pending_local_sibling_sends.len();
            if retired > 0 {
                // This upgrade barrier cannot be deferred by a surrounding batch.
                engine.persist_now()?;
            }
            Ok(retired)
        })
    }
}

pub(crate) fn pending_direct_authored_at(payload: &[u8], fallback: u64) -> u64 {
    serde_json::from_slice::<UnsignedEvent>(payload)
        .map_or(fallback, |event| event.created_at.as_secs())
}