xmtp_api_backend/
envelope.rs

1//! Decode backend envelopes once for queries and streams.
2use prost::Message;
3use std::collections::HashMap;
4use xmtp_proto::{
5    ConversionError, backend_v1 as wire,
6    types::{self, IncomingBatchLimits, OrderedEnvelopeBatch, Topic, TopicCursor, TopicKind},
7};
8
9/// A backend envelope cannot be decoded. These errors are not retryable.
10#[derive(Debug, thiserror::Error)]
11pub enum EnvelopeError {
12    /// A required field is absent or invalid. Not retryable.
13    #[error(transparent)]
14    Conversion(#[from] ConversionError),
15    /// A payload cannot be parsed. Not retryable.
16    #[error(transparent)]
17    Validation(#[from] xmtp_mls_validation::ValidationError),
18    /// The complete delivery cannot fit the receive buffer. No cursor advances.
19    #[error("incoming delivery exceeds its row or byte limit")]
20    Capacity,
21}
22impl xmtp_common::RetryableError for EnvelopeError {
23    fn is_retryable(&self) -> bool {
24        false
25    }
26}
27impl From<EnvelopeError> for xmtp_proto::api::ApiClientError {
28    fn from(error: EnvelopeError) -> Self {
29        Self::other(error)
30    }
31}
32
33fn invalid(item: &'static str) -> ConversionError {
34    ConversionError::InvalidValue {
35        item,
36        expected: "valid backend envelope",
37        got: "missing or invalid field".into(),
38    }
39}
40
41/// Validate a complete ordered read without parsing or changing MLS bytes.
42/// Preserve each topic's input cursor as `after`; sequence gaps are valid.
43/// Advance only these in-memory cursors on success, never durable receipt `F`.
44// implements: PROC-001
45pub fn ordered_batches(
46    cursors: &mut TopicCursor,
47    envelopes: Vec<wire::ServerEnvelope>,
48    limits: IncomingBatchLimits,
49) -> Result<Vec<OrderedEnvelopeBatch>, EnvelopeError> {
50    let bytes = envelopes.iter().try_fold(0usize, |bytes, envelope| {
51        bytes.checked_add(envelope.encoded_len())
52    });
53    if envelopes.len() > limits.max_rows || bytes.is_none_or(|bytes| bytes > limits.max_bytes) {
54        return Err(EnvelopeError::Capacity);
55    }
56    let mut next = cursors.clone();
57    let mut batches: Vec<OrderedEnvelopeBatch> = Vec::new();
58    let mut indices = HashMap::new();
59    for envelope in envelopes {
60        let meta = envelope.meta.as_ref().ok_or_else(|| invalid("metadata"))?;
61        let topic = Topic::parse(&meta.topic.as_ref().ok_or_else(|| invalid("topic"))?.topic)?;
62        let (_, sequence, _) = metadata(meta, topic.kind())?;
63        let cursor = next
64            .get_mut(&topic)
65            .ok_or_else(|| invalid("unrequested topic"))?;
66        if sequence <= *cursor {
67            return Err(invalid("ordered cursor").into());
68        }
69        let index = *indices.entry(topic.clone()).or_insert_with(|| {
70            let index = batches.len();
71            batches.push(OrderedEnvelopeBatch {
72                topic,
73                after: *cursor,
74                envelopes: Vec::new(),
75            });
76            index
77        });
78        batches[index].envelopes.push(envelope);
79        *cursor = sequence;
80    }
81    *cursors = next;
82    Ok(batches)
83}
84
85/// Check every fixed target against the exact registered topic set.
86/// Require one target per topic. A target may be below the requested start.
87pub fn registration_targets(
88    starts: &TopicCursor,
89    targets: Vec<wire::CatchupTarget>,
90) -> Result<TopicCursor, EnvelopeError> {
91    if targets.len() != starts.len() {
92        return Err(invalid("registration target count").into());
93    }
94    let mut output = TopicCursor::new();
95    for target in targets {
96        let topic = Topic::parse(&target.topic.ok_or_else(|| invalid("target topic"))?.topic)?;
97        if !starts.contains_key(&topic)
98            || target.through_sequence_id > i64::MAX as u64
99            || output
100                .insert(topic, types::Cursor(target.through_sequence_id))
101                .is_some()
102        {
103            return Err(invalid("registration target").into());
104        }
105    }
106    Ok(output)
107}
108
109/// Check the topic, cursor, hash, and server timestamp.
110pub fn metadata(
111    meta: &wire::EnvelopeMeta,
112    kind: TopicKind,
113) -> Result<(Topic, types::Cursor, chrono::DateTime<chrono::Utc>), ConversionError> {
114    let topic = Topic::parse(&meta.topic.as_ref().ok_or_else(|| invalid("topic"))?.topic)?;
115    if topic.kind() != kind {
116        return Err(invalid("topic kind"));
117    }
118    let seq = meta
119        .cursor
120        .as_ref()
121        .ok_or_else(|| invalid("cursor"))?
122        .sequence_id;
123    if seq == 0 || seq > i64::MAX as u64 {
124        return Err(invalid("sequence id"));
125    }
126    let timestamp = i64::try_from(meta.server_ns).map_err(|_| invalid("server timestamp"))?;
127    message_hash(meta)?;
128    Ok((
129        topic,
130        types::Cursor(seq),
131        chrono::DateTime::from_timestamp_nanos(timestamp),
132    ))
133}
134
135/// Read the SHA-256 envelope hash from server metadata.
136pub fn message_hash(meta: &wire::EnvelopeMeta) -> Result<Vec<u8>, ConversionError> {
137    match meta.message_hash.as_ref().and_then(|h| h.hash.as_ref()) {
138        Some(wire::message_hash::Hash::Sha256(bytes)) if bytes.len() == 32 => Ok(bytes.clone()),
139        _ => Err(invalid("message hash")),
140    }
141}
142
143fn parts(
144    envelope: wire::ServerEnvelope,
145    kind: TopicKind,
146) -> Result<(wire::EnvelopeMeta, wire::client_envelope::Payload), EnvelopeError> {
147    let meta = envelope.meta.ok_or_else(|| invalid("metadata"))?;
148    metadata(&meta, kind)?;
149    let payload = envelope
150        .envelope
151        .and_then(|e| e.payload)
152        .ok_or_else(|| invalid("payload"))?;
153    Ok((meta, payload))
154}
155
156/// Decode an MLS message and keep its payload hash and server metadata.
157pub fn decode_group_message(
158    envelope: wire::ServerEnvelope,
159) -> Result<types::GroupMessage, EnvelopeError> {
160    let (meta, payload) = parts(envelope, TopicKind::GroupMessagesV1)?;
161    let (topic, cursor, created_ns) = metadata(&meta, TopicKind::GroupMessagesV1)?;
162    let wire::client_envelope::Payload::GroupMessage(group) = payload else {
163        return Err(invalid("group payload").into());
164    };
165    let message = xmtp_mls_validation::parse_group_message(&group.data)?;
166    if message.group_id().as_slice() != topic.identifier() {
167        return Err(invalid("group id").into());
168    }
169    Ok(types::GroupMessage {
170        cursor,
171        created_ns,
172        group_id: topic.identifier().try_into()?,
173        message,
174        payload_hash: xmtp_common::sha256_array(&group.data).to_vec(),
175        sender_hmac: group.sender_hmac,
176        should_push: group.should_push,
177        envelope_hash: Some(message_hash(&meta)?),
178        expiry_ns: Some(meta.expiry_ns),
179    })
180}
181
182/// Decode the newest group metadata without loading its payload.
183pub fn decode_group_message_metadata(
184    meta: wire::EnvelopeMeta,
185) -> Result<types::GroupMessageMetadata, EnvelopeError> {
186    let (topic, cursor, created_ns) = metadata(&meta, TopicKind::GroupMessagesV1)?;
187    Ok(types::GroupMessageMetadata {
188        cursor,
189        created_ns,
190        group_id: topic.identifier().try_into()?,
191        envelope_hash: Some(message_hash(&meta)?),
192        expiry_ns: Some(meta.expiry_ns),
193    })
194}
195
196/// Decode an inline welcome or encrypted welcome pointer.
197pub fn decode_welcome_message(
198    envelope: wire::ServerEnvelope,
199) -> Result<types::WelcomeMessage, EnvelopeError> {
200    let (meta, payload) = parts(envelope, TopicKind::WelcomeMessagesV1)?;
201    let (topic, cursor, created_ns) = metadata(&meta, TopicKind::WelcomeMessagesV1)?;
202    let wire::client_envelope::Payload::WelcomeMessage(welcome) = payload else {
203        return Err(invalid("welcome payload").into());
204    };
205    let variant = match welcome.version.ok_or_else(|| invalid("welcome version"))? {
206        wire::welcome_message::Version::V1(v) => {
207            if v.installation_key != topic.identifier() {
208                return Err(invalid("installation id").into());
209            }
210            types::WelcomeMessageV1 {
211                installation_key: v.installation_key.try_into()?,
212                hpke_public_key: v.hpke_public_key,
213                wrapper_algorithm: v
214                    .wrapper_algorithm
215                    .try_into()
216                    .map_err(ConversionError::from)?,
217                data: v.data,
218                welcome_metadata: v.welcome_metadata,
219            }
220            .into()
221        }
222        wire::welcome_message::Version::WelcomePointer(v) => {
223            if v.installation_key != topic.identifier() {
224                return Err(invalid("installation id").into());
225            }
226            types::WelcomePointer {
227                installation_key: v.installation_key.try_into()?,
228                hpke_public_key: v.hpke_public_key,
229                wrapper_algorithm: v
230                    .wrapper_algorithm
231                    .try_into()
232                    .map_err(ConversionError::from)?,
233                welcome_pointer: v.welcome_pointer,
234            }
235            .into()
236        }
237    };
238    Ok(types::WelcomeMessage {
239        cursor,
240        created_ns,
241        variant,
242    })
243}
244
245/// Decode a key package returned by a full newest query.
246pub fn decode_key_package(
247    envelope: wire::ServerEnvelope,
248) -> Result<wire::KeyPackage, EnvelopeError> {
249    let (_, payload) = parts(envelope, TopicKind::KeyPackagesV1)?;
250    match payload {
251        wire::client_envelope::Payload::KeyPackage(key) => Ok(key),
252        _ => Err(invalid("key package payload").into()),
253    }
254}
255
256/// Decode a signed commit-log record without verifying its signature.
257pub fn decode_commit_log_entry(
258    envelope: wire::ServerEnvelope,
259) -> Result<types::CommitLogEntry, EnvelopeError> {
260    let (meta, payload) = parts(envelope, TopicKind::CommitLogEntriesV1)?;
261    let wire::client_envelope::Payload::CommitLogEntry(payload) = payload else {
262        return Err(invalid("commit log payload").into());
263    };
264    let entry =
265        xmtp_mls_common::commit_log::decode_commit_log(&payload.serialized_commit_log_entry)
266            .map_err(ConversionError::from)?;
267    let (topic, _, _) = metadata(&meta, TopicKind::CommitLogEntriesV1)?;
268    if entry.group_id != topic.identifier() {
269        return Err(invalid("commit log group id").into());
270    }
271    Ok(types::CommitLogEntry {
272        meta,
273        entry,
274        payload,
275    })
276}
277
278/// Decode an identity update and check its inbox topic.
279pub fn decode_identity_update(
280    envelope: wire::ServerEnvelope,
281) -> Result<types::IdentityUpdateLog, EnvelopeError> {
282    let (meta, payload) = parts(envelope, TopicKind::IdentityUpdatesV1)?;
283    let wire::client_envelope::Payload::IdentityUpdate(update) = payload else {
284        return Err(invalid("identity payload").into());
285    };
286    let (topic, _, _) = metadata(&meta, TopicKind::IdentityUpdatesV1)?;
287    if hex::decode(&update.inbox_id).map_err(|_| invalid("inbox id"))? != topic.identifier() {
288        return Err(invalid("inbox id").into());
289    }
290    Ok(types::IdentityUpdateLog { meta, update })
291}
292
293#[cfg(test)]
294mod tests {
295    use super::*;
296
297    fn envelope(topic: &Topic, sequence: u64) -> wire::ServerEnvelope {
298        wire::ServerEnvelope {
299            meta: Some(wire::EnvelopeMeta {
300                topic: Some(wire::Topic {
301                    topic: topic.cloned_vec(),
302                }),
303                cursor: Some(wire::Cursor {
304                    sequence_id: sequence,
305                }),
306                message_hash: Some(wire::MessageHash {
307                    hash: Some(wire::message_hash::Hash::Sha256(vec![7; 32])),
308                }),
309                ..Default::default()
310            }),
311            // Receipt deliberately does not parse a malformed MLS payload.
312            envelope: Some(wire::ClientEnvelope {
313                payload: Some(wire::client_envelope::Payload::GroupMessage(
314                    wire::GroupMessage {
315                        data: vec![255],
316                        ..Default::default()
317                    },
318                )),
319            }),
320        }
321    }
322
323    // verifies: PROC-001
324    #[xmtp_common::test(unwrap_try = true)]
325    fn ordered_batches_keep_sparse_positions_and_authoritative_bytes() {
326        let a = Topic::new_group_message([1; 16]);
327        let b = Topic::new_group_message([2; 16]);
328        let mut cursors = [(a.clone(), types::Cursor(2)), (b.clone(), types::Cursor(0))].into();
329        let original = envelope(&a, 8);
330        let batches = ordered_batches(
331            &mut cursors,
332            vec![original.clone(), envelope(&b, 11), envelope(&a, 20)],
333            IncomingBatchLimits {
334                max_rows: 3,
335                max_bytes: 4096,
336            },
337        )?;
338        assert_eq!(batches.len(), 2);
339        assert_eq!(batches[0].after, types::Cursor(2));
340        assert_eq!(batches[0].envelopes[0], original);
341        assert_eq!(batches[1].after, types::Cursor(0));
342        assert_eq!(cursors[&a], types::Cursor(20));
343        assert_eq!(cursors[&b], types::Cursor(11));
344    }
345
346    // verifies: PROC-001
347    #[xmtp_common::test(unwrap_try = true)]
348    fn failed_frame_does_not_advance_any_topic() {
349        let a = Topic::new_group_message([1; 16]);
350        let b = Topic::new_group_message([2; 16]);
351        let starts: TopicCursor =
352            [(a.clone(), types::Cursor(2)), (b.clone(), types::Cursor(0))].into();
353        for envelopes in [
354            vec![envelope(&a, 8), envelope(&b, 0)],
355            vec![envelope(&a, 8), envelope(&a, 7)],
356        ] {
357            let mut cursors = starts.clone();
358            assert!(
359                ordered_batches(
360                    &mut cursors,
361                    envelopes,
362                    IncomingBatchLimits {
363                        max_rows: 2,
364                        max_bytes: 4096
365                    }
366                )
367                .is_err()
368            );
369            assert_eq!(cursors, starts);
370        }
371        let mut cursors = starts.clone();
372        assert!(matches!(
373            ordered_batches(
374                &mut cursors,
375                vec![envelope(&a, 8)],
376                IncomingBatchLimits {
377                    max_rows: 1,
378                    max_bytes: 1
379                }
380            ),
381            Err(EnvelopeError::Capacity)
382        ));
383        assert_eq!(cursors, starts);
384    }
385}
xmtp_api_backend/envelope.rs

xmtp_api_backend/
envelope.rs
