Skip to main content

ping_core/
conversation.rs

1//! Conversation state — wraps an OpenMLS `MlsGroup`.
2//!
3//! Each external conversation maps 1:1 to an MLS group whose leaves are devices. The DeviceGroup
4//! (one per user, devices only) is just a special-cased conversation with the same wrapper.
5//!
6//! Persistence: we snapshot the `MlsGroup` after every state-changing operation under
7//! `groups/{conversation_id}` and cache the result in-memory.
8
9use openmls::{
10    framing::{MlsMessageOut, ProcessedMessageContent},
11    group::{MlsGroup, MlsGroupCreateConfig, MlsGroupJoinConfig},
12    prelude::{
13        tls_codec::{Deserialize as TlsDeserialize, Serialize as TlsSerialize},
14        BasicCredential, Capabilities, Ciphersuite, CredentialWithKey, Extension, ExtensionType,
15        Extensions, MlsMessageBodyIn, MlsMessageIn, ProcessedMessage, ProtocolMessage,
16        ProtocolVersion, RequiredCapabilitiesExtension, UnknownExtension,
17    },
18};
19use openmls_basic_credential::SignatureKeyPair;
20use openmls_traits::OpenMlsProvider;
21use ping_mls_store::PersistentMlsProvider;
22use serde::{Deserialize, Serialize};
23use std::collections::BTreeMap;
24use std::sync::Arc;
25use ulid::Ulid;
26use zeroize::Zeroizing;
27
28use crate::{
29    clock::Hlc,
30    codec,
31    device::{DeviceId, GroupSnapshotEntry, GroupStateSnapshot, GROUP_SNAPSHOT_VERSION},
32    error::{Error, Result},
33    identity::UserId,
34    message::{IncomingMessage, MessageEnvelope, MessageKind},
35    storage::Storage,
36    sync::SyncCursor,
37};
38
39const DEFAULT_CIPHERSUITE: Ciphersuite = Ciphersuite::MLS_128_DHKEMX25519_AES128GCM_SHA256_Ed25519;
40
41/// Map an OpenMLS `WelcomeError` to a typed [`Error`]. The two variants that
42/// mean "the KeyPackage this Welcome was bound to (or its private init key) is
43/// no longer in our keystore" — i.e. the KeyPackage was already consumed
44/// (single-use reuse) or never persisted — become [`Error::KeyPackageNotFound`]
45/// so the host can top up KeyPackages and retry rather than treating it as an
46/// opaque MLS failure. Everything else stays a generic MLS error.
47fn map_welcome_error<S: std::fmt::Display>(e: openmls::group::WelcomeError<S>) -> Error {
48    use openmls::group::WelcomeError;
49    match e {
50        WelcomeError::NoMatchingKeyPackage | WelcomeError::PrivateInitKeyNotFound => {
51            Error::KeyPackageNotFound
52        }
53        other => Error::mls(other),
54    }
55}
56
57/// MLS GroupContext extension type carrying the human conversation `name`.
58///
59/// Stored in the group context so the name is part of shared MLS group state and
60/// therefore present on EVERY device that holds the group — including a device
61/// that joins via a Welcome. Without this the name was creator-local and a
62/// joiner/linked device saw `None`. A private-use `Unknown` extension type
63/// (not a GREASE `0x?A?A` value, not a registered type 0x0001–0x0005).
64///
65/// It is added WITHOUT a `RequiredCapabilities` extension, so openmls imposes no
66/// per-member capability check — existing KeyPackages keep working and no
67/// re-link is required. The value is UTF-8 bytes of the name.
68const GROUP_NAME_EXTENSION_TYPE: u16 = 0xFF00;
69
70/// Read the conversation `name` from a group's GroupContext extensions, if set.
71fn group_name_from_extensions(extensions: &Extensions) -> Option<String> {
72    extensions.iter().find_map(|ext| match ext {
73        Extension::Unknown(ext_type, data) if *ext_type == GROUP_NAME_EXTENSION_TYPE => {
74            String::from_utf8(data.0.clone())
75                .ok()
76                .filter(|s| !s.is_empty())
77        }
78        _ => None,
79    })
80}
81
82/// Build the GroupContext extensions carrying `name` (empty when `name` is
83/// `None`/blank), for `MlsGroupCreateConfig::with_group_context_extensions`.
84fn group_context_extensions_for_name(name: Option<&str>) -> Extensions {
85    match name {
86        Some(n) if !n.is_empty() => Extensions::single(Extension::Unknown(
87            GROUP_NAME_EXTENSION_TYPE,
88            UnknownExtension(n.as_bytes().to_vec()),
89        )),
90        _ => Extensions::empty(),
91    }
92}
93
94/// Leaf [`Capabilities`] advertising support for the group-name GroupContext
95/// extension ([`GROUP_NAME_EXTENSION_TYPE`]) ON TOP OF the MLS defaults.
96///
97/// Required so a later [`Conversation::set_name`] (rename / avatar-id change via
98/// `update_group_context_extensions`) passes openmls' GCE-proposal validation:
99/// that path demands every group-context extension be listed in a
100/// `RequiredCapabilities`, and that every member's leaf advertise those
101/// extension types. The CREATE path does not need this (genesis extensions skip
102/// the proposal validator), but a post-create UPDATE does. New KeyPackages and
103/// the creator's own leaf carry this; existing devices must RE-LINK once to pick
104/// it up (a pre-production-acceptable cost, aligned with the shared-identity
105/// re-link).
106pub(crate) fn ping_leaf_capabilities() -> Capabilities {
107    ping_leaf_capabilities_for(false)
108}
109
110/// Leaf capabilities, optionally also advertising the `LastResort` extension.
111///
112/// A KeyPackage marked last-resort (`mark_as_last_resort`) carries the
113/// `LastResort` extension on its leaf; openmls' KeyPackage validation rejects
114/// any leaf whose capabilities do not advertise every extension the leaf
115/// actually carries ("A key package extension is not supported in the leaf's
116/// capabilities"). So last-resort KeyPackages MUST list `LastResort` here, while
117/// ordinary KeyPackages must NOT (they don't carry it).
118pub(crate) fn ping_leaf_capabilities_for(last_resort: bool) -> Capabilities {
119    let extensions = if last_resort {
120        vec![
121            ExtensionType::Unknown(GROUP_NAME_EXTENSION_TYPE),
122            ExtensionType::LastResort,
123        ]
124    } else {
125        vec![ExtensionType::Unknown(GROUP_NAME_EXTENSION_TYPE)]
126    };
127    Capabilities::new(None, None, Some(&extensions), None, None)
128}
129
130/// GroupContext extensions for a NAME UPDATE commit (`set_name`), as opposed to
131/// genesis ([`group_context_extensions_for_name`]). The GCE-proposal validator
132/// requires every group-context extension to appear in a `RequiredCapabilities`,
133/// so we attach one listing [`GROUP_NAME_EXTENSION_TYPE`] alongside the name
134/// extension. Clearing the name (`None`/blank) keeps the `RequiredCapabilities`
135/// (harmless: required caps may list a type that isn't currently present).
136fn group_context_extensions_for_name_update(name: Option<&str>) -> Extensions {
137    let required = Extension::RequiredCapabilities(RequiredCapabilitiesExtension::new(
138        &[ExtensionType::Unknown(GROUP_NAME_EXTENSION_TYPE)],
139        &[],
140        &[],
141    ));
142    match name {
143        Some(n) if !n.is_empty() => Extensions::from_vec(vec![
144            Extension::Unknown(
145                GROUP_NAME_EXTENSION_TYPE,
146                UnknownExtension(n.as_bytes().to_vec()),
147            ),
148            required,
149        ])
150        // Two distinct extension types — `from_vec` only rejects duplicates.
151        .expect("name + required-capabilities are distinct extension types"),
152        _ => Extensions::single(required),
153    }
154}
155
156/// 16-byte conversation identifier (ULID encoded). Stable across epochs.
157#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize, Deserialize)]
158pub struct ConversationId(#[serde(with = "serde_bytes_array16")] pub [u8; 16]);
159
160impl ConversationId {
161    pub fn new() -> Self {
162        Self(Ulid::new().to_bytes())
163    }
164    pub fn as_hex(&self) -> String {
165        hex::encode(self.0)
166    }
167}
168
169impl Default for ConversationId {
170    fn default() -> Self {
171        Self::new()
172    }
173}
174
175mod serde_bytes_array16 {
176    use serde::{Deserializer, Serializer};
177    pub fn serialize<S: Serializer>(b: &[u8; 16], s: S) -> Result<S::Ok, S::Error> {
178        serde_bytes::serialize(b.as_slice(), s)
179    }
180    pub fn deserialize<'de, D: Deserializer<'de>>(d: D) -> Result<[u8; 16], D::Error> {
181        let v: Vec<u8> = serde_bytes::deserialize(d)?;
182        v.try_into()
183            .map_err(|_| serde::de::Error::custom("expected 16 bytes"))
184    }
185}
186
187#[derive(Debug, Clone, Serialize, Deserialize)]
188pub struct ConversationMeta {
189    pub id: ConversationId,
190    pub name: Option<String>,
191    pub epoch: u64,
192    pub member_count: u32,
193    pub is_device_group: bool,
194    pub created_at_ms: u64,
195}
196
197/// One member leaf of a conversation's MLS group: the member's [`UserId`]
198/// (recovered from the leaf's `BasicCredential`) and its ratchet-tree leaf
199/// index. A user with multiple devices appears once **per device leaf** —
200/// callers that want a per-user roster should dedup by `user_id`.
201#[derive(Debug, Clone, Serialize, Deserialize)]
202pub struct MemberInfo {
203    pub user_id: UserId,
204    pub leaf_index: u32,
205}
206
207/// In-memory conversation handle. Holds the OpenMLS group plus our wire-level cursor.
208pub struct Conversation {
209    pub(crate) id: ConversationId,
210    pub(crate) meta: ConversationMeta,
211    pub(crate) group: MlsGroup,
212    pub(crate) crypto: Arc<PersistentMlsProvider>,
213    pub(crate) signing: Arc<SignatureKeyPair>,
214    pub(crate) own_device: DeviceId,
215    pub(crate) seq: u64,
216    pub(crate) hlc: Hlc,
217    pub(crate) cursor: SyncCursor,
218    pub(crate) storage: Arc<dyn Storage>,
219    /// Local device→leaf-index map for [CR-2] revocation.
220    ///
221    /// Populated when this device either (a) admits a peer via [`Self::add_members`] —
222    /// every entry in the `Vec<(DeviceId, KeyPackage)>` is recorded after the commit
223    /// merges — or (b) joins as the receiving device via [`Self::join`], at which point
224    /// we record our own leaf. Pruned when [`Self::remove_members`] is called.
225    ///
226    /// Not authoritative for *peers' devices we didn't admit*: those are visible in
227    /// `group.members()` but their device_ids are opaque to this client. `revoke_device`
228    /// is therefore best-effort across conversations we ourselves invited the device
229    /// into; see [`MessagingClient::revoke_device`] for the documented scope.
230    pub(crate) device_leaves: BTreeMap<DeviceId, u32>,
231}
232
233impl std::fmt::Debug for Conversation {
234    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
235        f.debug_struct("Conversation")
236            .field("id", &self.id.as_hex())
237            .field("meta", &self.meta)
238            .finish()
239    }
240}
241
242impl Conversation {
243    pub fn id(&self) -> ConversationId {
244        self.id
245    }
246    pub fn meta(&self) -> &ConversationMeta {
247        &self.meta
248    }
249
250    /// Current member roster, recovered locally from the MLS group's leaf
251    /// credentials — no network and no out-of-band `ping.profile` message.
252    /// Each `BasicCredential` was built from the member's `UserId`
253    /// (`BasicCredential::new(own_user.0.clone())` in [`Self::create`] /
254    /// [`Self::join`]), so we round-trip it back here. Each entry is one
255    /// leaf; a multi-device user appears once per device leaf.
256    pub fn members(&self) -> Vec<MemberInfo> {
257        self.group
258            .members()
259            .filter_map(|m| {
260                let basic = BasicCredential::try_from(m.credential).ok()?;
261                Some(MemberInfo {
262                    user_id: UserId(basic.identity().to_vec()),
263                    leaf_index: m.index.u32(),
264                })
265            })
266            .collect()
267    }
268
269    pub fn epoch(&self) -> u64 {
270        self.group.epoch().as_u64()
271    }
272    pub fn cursor(&self) -> &SyncCursor {
273        &self.cursor
274    }
275
276    /// Create a new conversation, with `self` as the only initial member.
277    // 8 args is a lot, but they're all needed for an internal constructor and a builder
278    // would be over-engineered for v0.1.
279    #[allow(clippy::too_many_arguments)]
280    pub(crate) fn create(
281        id: ConversationId,
282        name: Option<String>,
283        own_device: DeviceId,
284        own_user: &UserId,
285        crypto: Arc<PersistentMlsProvider>,
286        signing: Arc<SignatureKeyPair>,
287        storage: Arc<dyn Storage>,
288        now_ms: u64,
289    ) -> Result<Self> {
290        let credential = BasicCredential::new(own_user.0.clone());
291        let credential_with_key = CredentialWithKey {
292            credential: credential.into(),
293            signature_key: signing.public().into(),
294        };
295        // Carry the conversation name in the GroupContext so it travels in the
296        // Welcome to every joiner (see `GROUP_NAME_EXTENSION_TYPE`). No
297        // RequiredCapabilities is added, so this imposes no member capability
298        // check and stays compatible with existing KeyPackages.
299        let cfg = MlsGroupCreateConfig::builder()
300            .ciphersuite(DEFAULT_CIPHERSUITE)
301            .use_ratchet_tree_extension(true)
302            // Advertise the group-name extension capability on the creator's leaf
303            // so a later `set_name` (rename / avatar change) passes the GCE
304            // capability check. See `ping_leaf_capabilities`.
305            .capabilities(ping_leaf_capabilities())
306            .with_group_context_extensions(group_context_extensions_for_name(name.as_deref()))
307            .map_err(Error::mls)?
308            .build();
309        let group = MlsGroup::new_with_group_id(
310            crypto.as_ref(),
311            signing.as_ref(),
312            &cfg,
313            openmls::group::GroupId::from_slice(&id.0),
314            credential_with_key,
315        )
316        .map_err(Error::mls)?;
317
318        let meta = ConversationMeta {
319            id,
320            name,
321            epoch: 0,
322            member_count: 1,
323            is_device_group: false,
324            created_at_ms: now_ms,
325        };
326        // [CR-2] Group creator is always leaf 0; record so revoke_device can target it.
327        let mut device_leaves = BTreeMap::new();
328        device_leaves.insert(own_device.clone(), group.own_leaf_index().u32());
329        Ok(Self {
330            id,
331            meta,
332            group,
333            crypto,
334            signing,
335            own_device,
336            seq: 0,
337            hlc: Hlc::ZERO.tick(now_ms),
338            cursor: SyncCursor::default(),
339            storage,
340            device_leaves,
341        })
342    }
343
344    /// Join an existing conversation from a Welcome message.
345    ///
346    /// Maps the OpenMLS `WelcomeError` for a missing KeyPackage / init key to
347    /// the typed [`Error::KeyPackageNotFound`] (see `map_welcome_error`) so the
348    /// host can react (top up KeyPackages) instead of treating it as an opaque
349    /// MLS failure.
350    pub(crate) fn join(
351        welcome_bytes: &[u8],
352        own_device: DeviceId,
353        crypto: Arc<PersistentMlsProvider>,
354        signing: Arc<SignatureKeyPair>,
355        storage: Arc<dyn Storage>,
356        now_ms: u64,
357    ) -> Result<Self> {
358        let mls_in = MlsMessageIn::tls_deserialize_exact(welcome_bytes).map_err(Error::mls)?;
359        let welcome = match mls_in.extract() {
360            MlsMessageBodyIn::Welcome(w) => w,
361            _ => return Err(Error::Invalid("expected Welcome".into())),
362        };
363        let cfg = MlsGroupJoinConfig::builder()
364            .use_ratchet_tree_extension(true)
365            .build();
366        let staged =
367            openmls::group::StagedWelcome::new_from_welcome(crypto.as_ref(), &cfg, welcome, None)
368                .map_err(map_welcome_error)?;
369        let group = staged.into_group(crypto.as_ref()).map_err(Error::mls)?;
370
371        let id_bytes: [u8; 16] = group
372            .group_id()
373            .as_slice()
374            .try_into()
375            .map_err(|_| Error::Invalid("group id must be 16 bytes".into()))?;
376        let id = ConversationId(id_bytes);
377        // The creator stamped the name into the GroupContext, which is part of
378        // the shared group state the Welcome reconstructs — so a joining device
379        // recovers the real name here instead of starting with `None`.
380        let name = group_name_from_extensions(group.extensions());
381        let meta = ConversationMeta {
382            id,
383            name,
384            epoch: group.epoch().as_u64(),
385            member_count: group.members().count() as u32,
386            is_device_group: false,
387            created_at_ms: now_ms,
388        };
389
390        // Seed the cursor at the join epoch so subsequent fetches skip pre-join Commits
391        // (notably the Add commit that produced this Welcome — it lives in the conversation
392        // log at `epoch - 1`, which the joiner must not try to apply on top of its
393        // already-advanced group state).
394        let join_epoch = group.epoch().as_u64();
395        // [CR-2] Record our own (device_id → leaf_index) so the host can later revoke us
396        // via the standard `revoke_device` flow. `own_leaf_index()` is stable for the
397        // lifetime of this group membership.
398        let own_leaf = group.own_leaf_index().u32();
399        let mut device_leaves = BTreeMap::new();
400        device_leaves.insert(own_device.clone(), own_leaf);
401        Ok(Self {
402            id,
403            meta,
404            group,
405            crypto,
406            signing,
407            own_device,
408            seq: 0,
409            hlc: Hlc::ZERO.tick(now_ms),
410            cursor: SyncCursor {
411                epoch: join_epoch,
412                ..Default::default()
413            },
414            storage,
415            device_leaves,
416        })
417    }
418
419    /// The conversation name recovered from MLS GroupContext state (set at
420    /// creation, carried in the Welcome to every joiner). `None` if unnamed.
421    pub(crate) fn name_from_group_state(&self) -> Option<String> {
422        group_name_from_extensions(self.group.extensions())
423    }
424
425    /// [CR-4] Rehydrate a previously-persisted conversation on cold restart.
426    ///
427    /// Loads the OpenMLS group state via `MlsGroup::load` (which reads from the
428    /// provider's storage — populated by the SQLite-backed checkpoint on the
429    /// previous run). Pairs the loaded MLS state with the meta + cursor + device→leaf
430    /// map the host-side `Storage` trait kept for us. Returns `Ok(None)` if OpenMLS
431    /// finds no state for `id` — the host's `groups` namespace had a stale entry.
432    #[allow(clippy::too_many_arguments)]
433    pub(crate) fn load(
434        id: ConversationId,
435        meta: ConversationMeta,
436        cursor: SyncCursor,
437        device_leaves: BTreeMap<DeviceId, u32>,
438        own_device: DeviceId,
439        crypto: Arc<PersistentMlsProvider>,
440        signing: Arc<SignatureKeyPair>,
441        storage: Arc<dyn Storage>,
442        now_ms: u64,
443    ) -> Result<Option<Self>> {
444        use openmls::group::GroupId;
445        let group_id = GroupId::from_slice(&id.0);
446        let group = match MlsGroup::load(crypto.storage(), &group_id).map_err(Error::mls)? {
447            Some(g) => g,
448            None => return Ok(None),
449        };
450        // Restore the local outgoing-send counter from the persisted cursor. The cursor
451        // tracks the highest applied (epoch, sender, seq) for every device — including
452        // our own — so we can recover `self.seq` from `cursor.last_seq_per_device[own]`.
453        // Without this, the next `send_application()` re-uses an already-consumed seq
454        // and receivers silently dedupe (cursor.is_new returns false on their side).
455        let seq = cursor
456            .last_seq_per_device
457            .get(&own_device)
458            .copied()
459            .unwrap_or(0);
460        Ok(Some(Self {
461            id,
462            meta,
463            group,
464            crypto,
465            signing,
466            own_device,
467            seq,
468            hlc: Hlc::ZERO.tick(now_ms),
469            cursor,
470            storage,
471            device_leaves,
472        }))
473    }
474
475    /// Encrypt an application message and produce a wire envelope ready for transport.
476    ///
477    /// Uses the [CR-6] plaintext content_hash path: the envelope's `content_hash` is
478    /// `SHA-256(plaintext)`, not the MLS ciphertext. This is what makes rebase clean
479    /// and gives cross-binding hash parity.
480    pub fn send_application(&mut self, plaintext: &[u8], now_ms: u64) -> Result<MessageEnvelope> {
481        let out = self
482            .group
483            .create_message(self.crypto.as_ref(), self.signing.as_ref(), plaintext)
484            .map_err(Error::mls)?;
485
486        self.seq += 1;
487        self.hlc = self.hlc.tick(now_ms);
488        let bytes = out.tls_serialize_detached().map_err(Error::mls)?;
489        let env = MessageEnvelope::new_application(
490            self.id,
491            self.epoch(),
492            self.own_device.clone(),
493            self.seq,
494            self.hlc,
495            bytes,
496            plaintext,
497        );
498        // Advance the local cursor past our own send so a subsequent catch-up sync doesn't
499        // pull this envelope back to us (we've already applied it locally — re-processing
500        // would either fail or duplicate-deliver).
501        self.cursor.advance(
502            env.epoch,
503            self.own_device.clone(),
504            self.seq,
505            self.hlc,
506            now_ms,
507        );
508        Ok(env)
509    }
510
511    /// Add members by KeyPackage. Produces the Commit envelope to broadcast plus the Welcome
512    /// envelope(s) to deliver out-of-band to the newly-added devices.
513    ///
514    /// [CR-2] takes a `Vec<(DeviceId, KeyPackage)>` instead of a bare `Vec<KeyPackage>`. The
515    /// `DeviceId` for each entry is the *caller's* assertion of which device owns that
516    /// KeyPackage — hosts typically get it from the directory service alongside the
517    /// KeyPackage itself. The mapping is persisted per-conversation so [`MessagingClient::revoke_device`]
518    /// can later locate the leaf to remove without a fresh directory lookup. The SDK does
519    /// not cryptographically verify the device claim; that's a host policy concern
520    /// (typically: the directory authenticates the key_package_id → device_id mapping).
521    pub fn add_members(
522        &mut self,
523        entries: Vec<(DeviceId, Vec<u8>)>,
524        now_ms: u64,
525    ) -> Result<AddOutcome> {
526        // All-in-one (stage + immediate merge) for callers that commit and
527        // persist synchronously with NO networked rollback window — e.g. the
528        // device-group / device-linking paths. The networked group path in
529        // `client.rs` instead uses `stage_add_members` + `confirm_staged` /
530        // `abort_staged`, so a Commit the server REJECTS can be rolled back
531        // rather than leaving the local epoch ahead of the server (the desync
532        // that permanently bricks a group: every later Commit then 409s and
533        // peers can't decrypt our epoch).
534        let staged = self.stage_add_members(entries, now_ms)?;
535        self.confirm_staged(&staged, now_ms)?;
536        let StagedCommit {
537            commit, welcome, ..
538        } = staged;
539        let welcome =
540            welcome.ok_or_else(|| Error::Invalid("add_members produced no Welcome".into()))?;
541        Ok(AddOutcome { commit, welcome })
542    }
543
544    /// Stage an add-members Commit WITHOUT merging it — the group keeps a
545    /// *pending* commit and the local epoch is UNCHANGED. Returns the Commit +
546    /// Welcome envelopes to send. The caller MUST follow with exactly one of
547    /// [`Self::confirm_staged`] (server accepted the Commit → merge locally) or
548    /// [`Self::abort_staged`] (server rejected it → discard, epoch never moves).
549    ///
550    /// This is what makes the local epoch advance *only after* the server
551    /// accepts the Commit (send-then-merge), so a rejected/conflicting Commit
552    /// can't desync the group. Safe because the JS worker serializes top-level
553    /// requests — nothing else touches this group during the send round-trip
554    /// between stage and confirm/abort (worker.ts "SERIALIZE … dispatch").
555    pub(crate) fn stage_add_members(
556        &mut self,
557        entries: Vec<(DeviceId, Vec<u8>)>,
558        now_ms: u64,
559    ) -> Result<StagedCommit> {
560        let mut kps = Vec::with_capacity(entries.len());
561        // Track signature_key → device_id so we can resolve leaf indices post-commit.
562        let mut sig_to_device: Vec<(Vec<u8>, DeviceId)> = Vec::with_capacity(entries.len());
563        for (device_id, raw) in &entries {
564            let mls_in = MlsMessageIn::tls_deserialize_exact(raw).map_err(Error::mls)?;
565            let kp_in = match mls_in.extract() {
566                MlsMessageBodyIn::KeyPackage(kp) => kp,
567                _ => return Err(Error::Invalid("expected KeyPackage".into())),
568            };
569            // KeyPackages on the wire are unvalidated (`KeyPackageIn`); validate against the
570            // crypto provider before handing them to OpenMLS.
571            let kp = kp_in
572                .validate(self.crypto.crypto(), ProtocolVersion::default())
573                .map_err(Error::mls)?;
574            let sig_key = kp.leaf_node().signature_key().as_slice().to_vec();
575            sig_to_device.push((sig_key, device_id.clone()));
576            kps.push(kp);
577        }
578
579        // The Commit's wire `epoch` is the *source* epoch (where it was crafted). The
580        // Welcome's `epoch` is the *post-commit* epoch. An MLS Commit advances the epoch by
581        // exactly 1, so we can name the post-commit epoch as `pre + 1` WITHOUT merging.
582        let pre_commit_epoch = self.epoch();
583        let post_commit_epoch = pre_commit_epoch + 1;
584
585        let (commit_out, welcome_out, _gi) = self
586            .group
587            .add_members(self.crypto.as_ref(), self.signing.as_ref(), &kps)
588            .map_err(Error::mls)?;
589        // NB: NO merge here — the pending commit is merged by `confirm_staged`
590        // only once the server has accepted the Commit send.
591
592        let next_seq = self.seq + 1;
593        let next_hlc = self.hlc.tick(now_ms);
594
595        let commit_bytes = mls_message_out_bytes(commit_out)?;
596        let commit_env = MessageEnvelope::new(
597            self.id,
598            pre_commit_epoch,
599            MessageKind::Commit,
600            self.own_device.clone(),
601            next_seq,
602            next_hlc,
603            commit_bytes,
604        );
605
606        let welcome_bytes = mls_message_out_bytes(welcome_out)?;
607        let welcome_env = MessageEnvelope::new(
608            self.id,
609            post_commit_epoch,
610            MessageKind::Welcome,
611            self.own_device.clone(),
612            next_seq,
613            next_hlc,
614            welcome_bytes,
615        );
616
617        Ok(StagedCommit {
618            commit: commit_env,
619            welcome: Some(welcome_env),
620            next_seq,
621            next_hlc,
622            leaf_update: StagedLeafUpdate::Add(sig_to_device),
623        })
624    }
625
626    pub fn remove_members(
627        &mut self,
628        leaf_indexes: Vec<u32>,
629        now_ms: u64,
630    ) -> Result<MessageEnvelope> {
631        // All-in-one (stage + immediate merge). See `add_members` for why the
632        // networked path uses stage/confirm/abort instead.
633        let staged = self.stage_remove_members(leaf_indexes, now_ms)?;
634        self.confirm_staged(&staged, now_ms)?;
635        let StagedCommit { commit, .. } = staged;
636        Ok(commit)
637    }
638
639    /// Stage a remove-members Commit WITHOUT merging it — see
640    /// [`Self::stage_add_members`]. No Welcome (removals don't admit anyone).
641    pub(crate) fn stage_remove_members(
642        &mut self,
643        leaf_indexes: Vec<u32>,
644        now_ms: u64,
645    ) -> Result<StagedCommit> {
646        use openmls::prelude::LeafNodeIndex;
647        let leaves: Vec<LeafNodeIndex> = leaf_indexes
648            .iter()
649            .copied()
650            .map(LeafNodeIndex::new)
651            .collect();
652
653        let pre_commit_epoch = self.epoch();
654
655        let (commit_out, _welcome_opt, _gi) = self
656            .group
657            .remove_members(self.crypto.as_ref(), self.signing.as_ref(), &leaves)
658            .map_err(Error::mls)?;
659        // NB: NO merge here — see `stage_add_members`.
660
661        let next_seq = self.seq + 1;
662        let next_hlc = self.hlc.tick(now_ms);
663        let bytes = mls_message_out_bytes(commit_out)?;
664        let commit_env = MessageEnvelope::new(
665            self.id,
666            pre_commit_epoch,
667            MessageKind::Commit,
668            self.own_device.clone(),
669            next_seq,
670            next_hlc,
671            bytes,
672        );
673
674        let removed: std::collections::HashSet<u32> = leaf_indexes.iter().copied().collect();
675        Ok(StagedCommit {
676            commit: commit_env,
677            welcome: None,
678            next_seq,
679            next_hlc,
680            leaf_update: StagedLeafUpdate::Remove(removed),
681        })
682    }
683
684    /// Create a self-Remove PROPOSAL — "leave the group".
685    ///
686    /// MLS does not let a member commit their OWN removal, so leaving is a
687    /// two-step dance: this member broadcasts a Remove proposal for its own
688    /// leaf, and a DIFFERENT member commits it (see
689    /// [`Self::stage_commit_pending_proposals`]). A proposal does NOT advance
690    /// the epoch; we bump only our own seq/hlc and the caller then deletes the
691    /// conversation locally. Returns the proposal envelope
692    /// (`MessageKind::Proposal`) to broadcast on the conversation channel.
693    pub(crate) fn leave_group(&mut self, now_ms: u64) -> Result<MessageEnvelope> {
694        let proposal_out = self
695            .group
696            .leave_group(self.crypto.as_ref(), self.signing.as_ref())
697            .map_err(Error::mls)?;
698        let next_seq = self.seq + 1;
699        let next_hlc = self.hlc.tick(now_ms);
700        let bytes = mls_message_out_bytes(proposal_out)?;
701        let env = MessageEnvelope::new(
702            self.id,
703            self.epoch(),
704            MessageKind::Proposal,
705            self.own_device.clone(),
706            next_seq,
707            next_hlc,
708            bytes,
709        );
710        self.seq = next_seq;
711        self.hlc = next_hlc;
712        Ok(env)
713    }
714
715    /// True when OpenMLS has buffered pending proposals for this group (e.g. a
716    /// peer's leave proposal awaiting a Commit).
717    pub(crate) fn has_pending_proposals(&self) -> bool {
718        self.group.pending_proposals().next().is_some()
719    }
720
721    /// Stage a Commit covering all buffered pending proposals (the canonical
722    /// case: a peer's "leave" Remove proposal). Returns `None` when there is
723    /// nothing pending. Send-then-merge like add/remove so a server-rejected
724    /// Commit rolls back without desyncing the epoch. The removed leaves are
725    /// read off the pending Remove proposals so the device→leaf map is pruned
726    /// on confirm.
727    pub(crate) fn stage_commit_pending_proposals(
728        &mut self,
729        now_ms: u64,
730    ) -> Result<Option<StagedCommit>> {
731        use openmls::messages::proposals::Proposal;
732        let removed: std::collections::HashSet<u32> = self
733            .group
734            .pending_proposals()
735            .filter_map(|qp| match qp.proposal() {
736                Proposal::Remove(r) => Some(r.removed().u32()),
737                _ => None,
738            })
739            .collect();
740        if self.group.pending_proposals().next().is_none() {
741            return Ok(None);
742        }
743
744        let pre_commit_epoch = self.epoch();
745        let (commit_out, _welcome, _gi) = self
746            .group
747            .commit_to_pending_proposals(self.crypto.as_ref(), self.signing.as_ref())
748            .map_err(Error::mls)?;
749        // NB: NO merge here — confirm_staged merges once the server accepts.
750
751        let next_seq = self.seq + 1;
752        let next_hlc = self.hlc.tick(now_ms);
753        let bytes = mls_message_out_bytes(commit_out)?;
754        let commit_env = MessageEnvelope::new(
755            self.id,
756            pre_commit_epoch,
757            MessageKind::Commit,
758            self.own_device.clone(),
759            next_seq,
760            next_hlc,
761            bytes,
762        );
763
764        Ok(Some(StagedCommit {
765            commit: commit_env,
766            welcome: None,
767            next_seq,
768            next_hlc,
769            leaf_update: StagedLeafUpdate::Remove(removed),
770        }))
771    }
772
773    /// Change the conversation `name` carried in the GroupContext (RFC 9420
774    /// GroupContextExtensions commit). Unlike a hydration broadcast, this rides
775    /// MLS group STATE, so every member — and every future joiner via the
776    /// ratchet-tree/GroupInfo — converges on the new name. Hosts use this to make
777    /// a rename or an embedded avatar-media-id change BULLETPROOF (the name field
778    /// carries the `ping:meta:v1:` blob).
779    ///
780    /// Produces a Commit to broadcast; NO Welcome (membership is unchanged). All
781    /// members must advertise the group-name extension capability
782    /// ([`ping_leaf_capabilities`]) — i.e. have re-linked since that shipped —
783    /// else openmls rejects the commit (`RequiredExtensionNotSupportedByAllMembers`).
784    ///
785    /// All-in-one (stage + immediate merge). The networked group path uses
786    /// [`Self::stage_set_name`] + [`Self::confirm_staged`]/[`Self::abort_staged`]
787    /// so a server-rejected commit can roll back without desyncing the epoch.
788    pub fn set_name(&mut self, name: Option<String>, now_ms: u64) -> Result<MessageEnvelope> {
789        let staged = self.stage_set_name(name, now_ms)?;
790        self.confirm_staged(&staged, now_ms)?;
791        let StagedCommit { commit, .. } = staged;
792        Ok(commit)
793    }
794
795    /// Stage a name-update Commit WITHOUT merging it — see
796    /// [`Self::stage_add_members`]. No Welcome (no membership change). The local
797    /// epoch advances only on [`Self::confirm_staged`].
798    pub(crate) fn stage_set_name(
799        &mut self,
800        name: Option<String>,
801        now_ms: u64,
802    ) -> Result<StagedCommit> {
803        let pre_commit_epoch = self.epoch();
804        let extensions = group_context_extensions_for_name_update(name.as_deref());
805
806        let (commit_out, _welcome_opt, _gi) = self
807            .group
808            .update_group_context_extensions(
809                self.crypto.as_ref(),
810                extensions,
811                self.signing.as_ref(),
812            )
813            .map_err(Error::mls)?;
814        // NB: NO merge here — confirm_staged merges once the server accepts.
815
816        let next_seq = self.seq + 1;
817        let next_hlc = self.hlc.tick(now_ms);
818        let bytes = mls_message_out_bytes(commit_out)?;
819        let commit_env = MessageEnvelope::new(
820            self.id,
821            pre_commit_epoch,
822            MessageKind::Commit,
823            self.own_device.clone(),
824            next_seq,
825            next_hlc,
826            bytes,
827        );
828
829        Ok(StagedCommit {
830            commit: commit_env,
831            welcome: None,
832            next_seq,
833            next_hlc,
834            leaf_update: StagedLeafUpdate::None,
835        })
836    }
837
838    /// Merge a previously [staged](Self::stage_add_members) Commit into the local
839    /// group — call ONLY after the server has accepted the Commit send. Advances
840    /// the epoch, updates the roster + device→leaf map, bumps seq/hlc, and moves
841    /// the sync cursor past our own Commit so catch-up doesn't re-apply it.
842    pub(crate) fn confirm_staged(&mut self, staged: &StagedCommit, now_ms: u64) -> Result<()> {
843        self.group
844            .merge_pending_commit(self.crypto.as_ref())
845            .map_err(Error::mls)?;
846        self.meta.epoch = self.epoch();
847        self.meta.member_count = self.group.members().count() as u32;
848        // A GroupContextExtensions commit (e.g. `set_name`) changes the name
849        // carried in group state — refresh the cached meta name. Harmless for
850        // add/remove commits (the name is unchanged).
851        self.meta.name = self.name_from_group_state();
852
853        match &staged.leaf_update {
854            StagedLeafUpdate::Add(sig_to_device) => {
855                // [CR-2] Resolve leaf indexes for the devices we just added (match by the
856                // per-device MLS signature_key, unique per device).
857                for member in self.group.members() {
858                    if let Some((_, device_id)) = sig_to_device
859                        .iter()
860                        .find(|(sig, _)| sig.as_slice() == member.signature_key.as_slice())
861                    {
862                        self.device_leaves
863                            .insert(device_id.clone(), member.index.u32());
864                    }
865                }
866            }
867            StagedLeafUpdate::Remove(removed) => {
868                // [CR-2] Prune the device→leaf map for removed leaves. Other entries' leaf
869                // indexes are stable (OpenMLS reuses blank slots, doesn't reshuffle).
870                self.device_leaves.retain(|_, idx| !removed.contains(idx));
871            }
872            StagedLeafUpdate::None => {
873                // No membership change (name-update commit) — leaf map unchanged.
874            }
875        }
876
877        self.seq = staged.next_seq;
878        self.hlc = staged.next_hlc;
879        self.cursor.advance(
880            self.meta.epoch,
881            self.own_device.clone(),
882            self.seq,
883            self.hlc,
884            now_ms,
885        );
886        Ok(())
887    }
888
889    /// Discard a previously [staged](Self::stage_add_members) Commit — call when
890    /// the server REJECTED the Commit send. Clears the pending commit so the
891    /// local epoch stays exactly where it was (no desync) and the conversation
892    /// is operational again. Idempotent / safe if there is no pending commit.
893    pub(crate) fn abort_staged(&mut self) -> Result<()> {
894        self.group
895            .clear_pending_commit(self.crypto.storage())
896            .map_err(Error::mls)?;
897        Ok(())
898    }
899
900    /// Process an inbound envelope. Returns Some(IncomingMessage) for application traffic.
901    pub fn process(
902        &mut self,
903        env: &MessageEnvelope,
904        now_ms: u64,
905    ) -> Result<Option<IncomingMessage>> {
906        if !self.cursor.is_new(env.epoch, &env.sender_device, env.seq) {
907            return Ok(None); // dedupe: already applied
908        }
909        let mls_in = MlsMessageIn::tls_deserialize_exact(&env.payload).map_err(Error::mls)?;
910
911        // OpenMLS' `process_message` expects an `impl Into<ProtocolMessage>`. `MlsMessageIn`
912        // itself doesn't implement that; we have to extract the body and convert the inner
913        // private/public message. Welcomes are handled at the client level, not here.
914        let protocol_msg: ProtocolMessage = match mls_in.extract() {
915            MlsMessageBodyIn::PrivateMessage(m) => m.into(),
916            MlsMessageBodyIn::PublicMessage(m) => m.into(),
917            MlsMessageBodyIn::Welcome(_) => {
918                return Err(Error::Invalid(
919                    "Welcome must be handled at client level, not in-group".into(),
920                ));
921            }
922            _ => return Err(Error::Invalid("unsupported MLS message body".into())),
923        };
924
925        let processed: ProcessedMessage = self
926            .group
927            .process_message(self.crypto.as_ref(), protocol_msg)
928            .map_err(Error::mls)?;
929
930        // Recover the sender's account-level `UserId` from their authenticated
931        // leaf credential BEFORE `into_content()` consumes `processed`. Same
932        // round-trip as `members()`: the leaf was built as
933        // `BasicCredential::new(user.0)`, so `identity()` is the `UserId` bytes.
934        // This lets the host attribute messages to the right account across
935        // every linked device without any device→account side channel.
936        let sender_user_id = BasicCredential::try_from(processed.credential().clone())
937            .map(|c| UserId(c.identity().to_vec()))
938            .unwrap_or_else(|_| UserId(Vec::new()));
939
940        let out = match processed.into_content() {
941            ProcessedMessageContent::ApplicationMessage(app) => {
942                let pt = app.into_bytes();
943                // CR-6: for v=2 application envelopes the wire-contract validator can't
944                // check `content_hash` (the hash is over plaintext, which it didn't have).
945                // We can now: verify SHA-256(pt) == env.content_hash and reject mismatches.
946                // For v=1 envelopes the wire-contract validator already checked the
947                // ciphertext-based hash, so no extra work here.
948                if env.v >= 2 {
949                    let computed = crate::message::hash_application_plaintext(&pt);
950                    if computed != env.content_hash {
951                        return Err(Error::Invalid(
952                            "v=2 application content_hash mismatch".into(),
953                        ));
954                    }
955                }
956                Some(IncomingMessage {
957                    conversation_id: self.id,
958                    sender_device: env.sender_device.clone(),
959                    sender_user_id,
960                    epoch: env.epoch,
961                    hlc: env.hlc,
962                    plaintext: pt,
963                    content_hash: env.content_hash,
964                })
965            }
966            ProcessedMessageContent::StagedCommitMessage(staged) => {
967                self.group
968                    .merge_staged_commit(self.crypto.as_ref(), *staged)
969                    .map_err(Error::mls)?;
970                self.meta.epoch = self.epoch();
971                self.meta.member_count = self.group.members().count() as u32;
972                // A remote `set_name` (GroupContextExtensions) commit changes the
973                // name in group state — refresh the cached meta so this device
974                // picks up the rename / avatar-id change WITHOUT a side broadcast.
975                self.meta.name = self.name_from_group_state();
976                None
977            }
978            ProcessedMessageContent::ProposalMessage(qp) => {
979                // OpenMLS does NOT auto-buffer a processed proposal — the caller
980                // must explicitly store it, or `pending_proposals()` stays empty
981                // and `commit_to_pending_proposals` covers nothing. This is the
982                // mechanism behind "leave": a peer's self-Remove proposal is
983                // stored here so a remaining member can commit it (evicting the
984                // leaver) via `commit_pending_proposals`.
985                self.group
986                    .store_pending_proposal(self.crypto.storage(), *qp)
987                    .map_err(Error::mls)?;
988                None
989            }
990            ProcessedMessageContent::ExternalJoinProposalMessage(_) => {
991                // External-join proposals are not part of any current flow; drop.
992                None
993            }
994        };
995
996        self.cursor.advance(
997            env.epoch,
998            env.sender_device.clone(),
999            env.seq,
1000            env.hlc,
1001            now_ms,
1002        );
1003        Ok(out)
1004    }
1005
1006    /// Export a derived secret keyed to this group's current epoch ([CR-8]).
1007    ///
1008    /// Wraps `MlsGroup::export_secret` (the MLS exporter, RFC 9420 §8.5) and surfaces the
1009    /// bytes in a `Zeroizing<Vec<u8>>` so the local copy is wiped on drop. Used by the host
1010    /// to seed:
1011    ///   * the ephemeral channel (`ping/ephemeral`, §5.4 of the architecture)
1012    ///   * call media keys (`ping/calls/media/{call_id}`, §7.2)
1013    ///   * call-ephemeral framer keys (`ping/calls/ephemeral/{call_id}`, §7.5)
1014    ///
1015    /// `label` should use the documented `ping/*` namespacing convention. There is no
1016    /// runtime enforcement — cross-binding parity is enforced by conformance fixtures
1017    /// pinning specific label strings.
1018    ///
1019    /// Output is the secret. Callers MUST treat the buffer as a secret: never log, never
1020    /// persist unencrypted. The wrapper zeroes our local copy on drop; the caller is
1021    /// responsible for zeroing any copy they make.
1022    pub fn export_secret(
1023        &self,
1024        label: &str,
1025        context: &[u8],
1026        length: usize,
1027    ) -> Result<Zeroizing<Vec<u8>>> {
1028        if length == 0 {
1029            return Err(Error::Invalid("export_secret length must be > 0".into()));
1030        }
1031        // Soft cap to prevent runaway allocations from a malformed caller. Real labels never
1032        // need more than ~64 bytes (AES-256 key + 96-bit nonce + slack); 1 KiB is generous.
1033        if length > 1024 {
1034            return Err(Error::Invalid(
1035                "export_secret length exceeds 1024-byte cap".into(),
1036            ));
1037        }
1038        let bytes = self
1039            .group
1040            .export_secret(self.crypto.as_ref(), label, context, length)
1041            .map_err(Error::mls)?;
1042        Ok(Zeroizing::new(bytes))
1043    }
1044
1045    /// [CR-7] Export a portable snapshot of this group's MLS state.
1046    ///
1047    /// Walks the provider's working set, picks every entry whose key references this
1048    /// group's id, and bundles them with format metadata. Returns CBOR-encoded bytes
1049    /// suitable for inclusion in:
1050    ///   * `LinkingTicket.catchup_snapshot.conversation_metas[i].group_state_bytes`
1051    ///     (via [CR-13] — host calls this and passes the bytes through);
1052    ///   * `IdentityBackup.device_group_snapshot` (the Permissive-recovery path per
1053    ///     `docs/architecture/recovery.md`).
1054    ///
1055    /// Returns `Err` if the encoded snapshot exceeds [`GROUP_SNAPSHOT_HARD_CAP`].
1056    /// Output is wrapped in `Zeroizing` because the bytes contain past epoch secrets;
1057    /// the caller's copy on the FFI side is the host's responsibility to wipe.
1058    pub fn export_state_snapshot(&self, now_ms: u64) -> Result<Zeroizing<Vec<u8>>> {
1059        let entries = self.crypto.group_scoped_entries(&self.id.0);
1060        let snap = GroupStateSnapshot {
1061            v: GROUP_SNAPSHOT_VERSION,
1062            group_id: self.id,
1063            openmls_storage_version: openmls_traits::storage::CURRENT_VERSION,
1064            snapshot_created_at_ms: now_ms,
1065            entries: entries
1066                .into_iter()
1067                .map(|(key, value)| GroupSnapshotEntry { key, value })
1068                .collect(),
1069        };
1070        Ok(Zeroizing::new(snap.encode()?))
1071    }
1072
1073    /// Look up the leaf index this device controls, if known ([CR-2]).
1074    ///
1075    /// Returns the locally-tracked leaf for `device_id`. Only populated for devices we
1076    /// added via [`Self::add_members`] or for our own leaf via [`Self::create`] /
1077    /// [`Self::join`]. Devices a peer admitted on our behalf are not in this map.
1078    pub fn leaf_index_of(&self, device_id: &DeviceId) -> Option<u32> {
1079        self.device_leaves.get(device_id).copied()
1080    }
1081
1082    /// Synchronously capture everything [`ConversationSnapshot::flush`]
1083    /// needs to persist this conversation, so a caller can DROP the
1084    /// `conversations` lock BEFORE awaiting the async writes.
1085    ///
1086    /// Holding a `parking_lot` guard across `.await` is a latent bug: on
1087    /// the single-threaded wasm worker, a second client call that lands
1088    /// while the first is suspended (a waiting writer + a new reader)
1089    /// makes `parking_lot` try to PARK, and its wasm stub `panic!`s with
1090    /// "Parking not supported on this platform" — poisoning the module.
1091    /// Splitting the synchronous capture (under the lock) from the async
1092    /// flush (lock released) removes that hazard everywhere the snapshot
1093    /// runs for a conversation that lives inside the shared map. The
1094    /// capture is a consistent point-in-time view (cursor + meta + leaves
1095    /// + the Arc'd provider/storage handles).
1096    pub(crate) fn snapshot_inputs(&self) -> Result<ConversationSnapshot> {
1097        // [CR-2] Stable BTreeMap-of-pairs encoding → canonical CBOR so
1098        // every platform decodes identical bytes.
1099        let leaves_vec: Vec<(DeviceId, u32)> = self
1100            .device_leaves
1101            .iter()
1102            .map(|(d, i)| (d.clone(), *i))
1103            .collect();
1104        Ok(ConversationSnapshot {
1105            id: self.id,
1106            crypto: self.crypto.clone(),
1107            storage: self.storage.clone(),
1108            cursor: self.cursor.encode()?,
1109            meta: codec::encode(&self.meta)?,
1110            device_leaves: codec::encode(&leaves_vec)?,
1111        })
1112    }
1113
1114    /// Persist this conversation's state. Convenience wrapper used by
1115    /// call sites that hold an OWNED `Conversation` (not borrowed from the
1116    /// shared map) — e.g. just-created/just-joined conversations before
1117    /// they're inserted, where no lock is held across the await. Map-
1118    /// resident callers MUST instead use `snapshot_inputs()` + drop the
1119    /// guard + `flush().await` (see client.rs) to avoid the wasm parking
1120    /// panic described on `snapshot_inputs`.
1121    pub(crate) async fn snapshot_to_storage(&self) -> Result<()> {
1122        self.snapshot_inputs()?.flush().await
1123    }
1124}
1125
1126/// Point-in-time, lock-free snapshot of a [`Conversation`]'s persistable
1127/// state. Produced synchronously by [`Conversation::snapshot_inputs`] (so
1128/// the `conversations` lock can be dropped) and flushed asynchronously by
1129/// [`Self::flush`].
1130pub(crate) struct ConversationSnapshot {
1131    id: ConversationId,
1132    crypto: Arc<PersistentMlsProvider>,
1133    storage: Arc<dyn Storage>,
1134    cursor: Vec<u8>,
1135    meta: Vec<u8>,
1136    device_leaves: Vec<u8>,
1137}
1138
1139impl ConversationSnapshot {
1140    /// Flush the captured state to storage. Safe to `.await` with NO
1141    /// `conversations` lock held — it only touches the Arc'd provider +
1142    /// storage handles, never the shared map.
1143    pub(crate) async fn flush(self) -> Result<()> {
1144        // CRASH-CONSISTENCY ORDERING. The host `Storage` is a non-transactional
1145        // key-value store (iOS SQLCipher row, web IndexedDB slot, memory), so we
1146        // cannot make these four writes atomic without changing the host API on
1147        // every platform. Instead we order them so a crash between any two
1148        // leaves a SELF-HEALING state, never a permanent gap:
1149        //
1150        //   1. MLS working set (`checkpoint_async`) — the authoritative crypto
1151        //      state (epoch, ratchet keys). Written FIRST so the persisted MLS
1152        //      state is always >= what the cursor claims we've processed.
1153        //   2. meta (name / member_count) — cosmetic, re-derivable from group state.
1154        //   3. device→leaf map ([CR-2]) — needed by revoke_device after restart.
1155        //   4. cursor — the "processed up to here" gate, written LAST.
1156        //
1157        // Why cursor LAST is the key invariant: the cursor decides which events
1158        // we re-fetch on restart. If it committed BEFORE the MLS checkpoint, a
1159        // crash in between would leave the cursor ahead of the persisted MLS
1160        // state — we'd skip events the group never actually applied, a permanent
1161        // gap (the "stranded" condition). Writing MLS first and the cursor last
1162        // guarantees the cursor is never ahead of durable state: a crash just
1163        // means we re-fetch a few already-applied events, which `process` dedups
1164        // via `SyncCursor::is_new`. meta/device_leaves lagging the cursor is the
1165        // only residual skew and both are re-derivable from MLS group state.
1166        //
1167        // [CR-4] checkpoint MUST happen on every state-changing op so a cold
1168        // restart (iOS NSE, web SW) finds the latest epoch. `checkpoint_async`
1169        // is required for the WASM `IndexedDb` backend (IDB is async-only);
1170        // native Memory / Sqlite await trivially (their I/O is sync internally).
1171        self.crypto
1172            .checkpoint_async()
1173            .await
1174            .map_err(|e| Error::Storage(format!("checkpoint: {e}")))?;
1175
1176        let hex = self.id.as_hex();
1177        self.storage
1178            .put("groups", &format!("{hex}/meta"), self.meta)
1179            .await?;
1180        self.storage
1181            .put("device_leaves", &hex, self.device_leaves)
1182            .await?;
1183        // Cursor written LAST — see the ordering rationale above.
1184        self.storage.put("cursors", &hex, self.cursor).await?;
1185        Ok(())
1186    }
1187}
1188
1189/// Both halves of an Add commit. The Commit goes on the conversation channel; the Welcome is
1190/// delivered to the new members via whatever out-of-band path the host uses (often the same
1191/// transport, addressed to the new device's mailbox).
1192#[derive(Debug, Clone)]
1193pub struct AddOutcome {
1194    pub commit: MessageEnvelope,
1195    pub welcome: MessageEnvelope,
1196}
1197
1198/// The local-state mutation a staged Commit will apply on
1199/// [`Conversation::confirm_staged`]. Captured at stage time so confirm can run
1200/// after the (async) Commit send without re-deriving anything.
1201pub(crate) enum StagedLeafUpdate {
1202    /// Add: signature_key → device_id for each added device, resolved to a leaf
1203    /// index against the merged tree in `confirm_staged`.
1204    Add(Vec<(Vec<u8>, DeviceId)>),
1205    /// Remove: the leaf indexes being dropped from the device→leaf map.
1206    Remove(std::collections::HashSet<u32>),
1207    /// No membership change (e.g. a GroupContextExtensions / name-update commit).
1208    /// The device→leaf map is untouched.
1209    None,
1210}
1211
1212/// A Commit produced but NOT yet merged (see [`Conversation::stage_add_members`]).
1213/// Held by the client across the Commit send; merged via
1214/// [`Conversation::confirm_staged`] on success or discarded via
1215/// [`Conversation::abort_staged`] on a server rejection. This is the unit of the
1216/// send-then-merge protocol that keeps the local epoch from ever running ahead of
1217/// the server.
1218pub(crate) struct StagedCommit {
1219    pub commit: MessageEnvelope,
1220    pub welcome: Option<MessageEnvelope>,
1221    next_seq: u64,
1222    next_hlc: Hlc,
1223    leaf_update: StagedLeafUpdate,
1224}
1225
1226fn mls_message_out_bytes(m: MlsMessageOut) -> Result<Vec<u8>> {
1227    m.tls_serialize_detached().map_err(Error::mls)
1228}