iris-chat-protocol 0.3.7

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
121
122
123
124
125
126
127
128
129
130
131
132

// Retry clocks are mutable; the large event/envelope body is shared across
// checkpoint and rollback clones. Any body mutation starts a fresh cache
// generation, including owner resolution and imported-envelope changes.
#[derive(Clone)]
struct CheckpointInbound {
    value: Arc<ProtocolPendingInboundBody>,
    cached: Arc<std::sync::OnceLock<CheckpointInboundBodies>>,
}

#[derive(Clone)]
struct CheckpointInboundBodies {
    event: Arc<crate::StorageCheckpointBody>,
    envelope: Arc<crate::StorageCheckpointBody>,
}

impl From<ProtocolPendingInboundBody> for CheckpointInbound {
    fn from(value: ProtocolPendingInboundBody) -> Self {
        Self {
            value: Arc::new(value),
            cached: Arc::default(),
        }
    }
}

impl std::fmt::Debug for CheckpointInbound {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        self.value.fmt(f)
    }
}

impl PartialEq for CheckpointInbound {
    fn eq(&self, other: &Self) -> bool {
        self.value == other.value
    }
}
impl Eq for CheckpointInbound {}

impl std::ops::Deref for ProtocolPendingInbound {
    type Target = ProtocolPendingInboundBody;
    fn deref(&self) -> &Self::Target {
        &self.body.value
    }
}

impl std::ops::DerefMut for ProtocolPendingInbound {
    fn deref_mut(&mut self) -> &mut Self::Target {
        self.body.cached = Arc::default();
        Arc::make_mut(&mut self.body.value)
    }
}

impl Serialize for CheckpointInbound {
    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        self.value.serialize(serializer)
    }
}

impl<'de> Deserialize<'de> for CheckpointInbound {
    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        ProtocolPendingInboundBody::deserialize(deserializer).map(Self::from)
    }
}

impl CheckpointInbound {
    fn checkpoint_bodies(&self) -> Result<CheckpointInboundBodies, serde_json::Error> {
        if let Some(cached) = self.cached.get() {
            return Ok(cached.clone());
        }
        let bodies = CheckpointInboundBodies {
            event: Arc::new(crate::StorageCheckpointBody::new(
                serde_json::value::to_raw_value(&self.value.event)?.into(),
            )),
            envelope: Arc::new(crate::StorageCheckpointBody::new(
                serde_json::value::to_raw_value(&self.value.envelope)?.into(),
            )),
        };
        let _ = self.cached.set(bodies);
        Ok(self
            .cached
            .get()
            .expect("initialized inbound checkpoint bodies")
            .clone())
    }
}

fn append_checkpoint_inbound(
    checkpoint: &mut crate::StorageCheckpoint,
    inbound: &[ProtocolPendingInbound],
) -> Result<(), serde_json::Error> {
    #[derive(Serialize)]
    struct Clocks<'a> {
        created_at_secs: u64,
        next_retry_at_secs: u64,
        event_id: &'a str,
    }
    #[derive(Serialize)]
    struct Owners<'a> {
        sender_message_pubkey_hex: &'a Option<String>,
        resolved_owner_pubkey_hex: &'a Option<String>,
        claimed_owner_pubkey_hex: &'a Option<String>,
        metadata_verified: bool,
    }
    for (index, pending) in inbound.iter().enumerate() {
        let bodies = pending.body.checkpoint_bodies()?;
        let clocks = serde_json::to_string(&Clocks {
            created_at_secs: pending.created_at_secs,
            next_retry_at_secs: pending.next_retry_at_secs,
            event_id: &pending.event_id,
        })?;
        let owners = serde_json::to_string(&Owners {
            sender_message_pubkey_hex: &pending.sender_message_pubkey_hex,
            resolved_owner_pubkey_hex: &pending.resolved_owner_pubkey_hex,
            claimed_owner_pubkey_hex: &pending.claimed_owner_pubkey_hex,
            metadata_verified: pending.metadata_verified,
        })?;
        checkpoint.push_record(
            2,
            bodies.event,
            format!("{}{{\"event\":", if index == 0 { "" } else { "," }),
            format!(",{},", &clocks[1..clocks.len() - 1]),
            0,
        );
        checkpoint.push_record(
            3,
            bodies.envelope,
            "\"envelope\":".into(),
            format!(",{}", &owners[1..]),
            0,
        );
    }
    Ok(())
}