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

xmtp_api_backend/
envelope.rs
