xmtp_mls_identity/identity_updates/
state_store.rs

1//! Resolving and storing association states. A client stores each state it
2//! computes through one write that also reports a changed display handle.
3
4use super::IdentityUpdateError;
5use crate::identity_updates::IdentityContext;
6use xmtp_db::{
7    StorageError, TransactionOutcome, identity_update::StoredIdentityUpdate, prelude::*,
8};
9use xmtp_id::{
10    associations::{
11        AssociationError, AssociationState, get_state, unverified::UnverifiedIdentityUpdate,
12        verify_updates,
13    },
14    scw_verifier::SmartContractSignatureVerifier,
15};
16
17use crate::identity_ext::IdentityExt;
18
19/// An association state read from the cache, or one computed from the stored
20/// identity updates that the caller must store.
21pub(super) enum ResolvedState {
22    Cached(AssociationState),
23    Computed {
24        sequence_id: i64,
25        state: AssociationState,
26    },
27}
28
29/// Read the association state of `inbox_id` at `to_sequence_id`, or at the
30/// newest stored update, from the cache. On a miss, compute it from the
31/// stored updates and leave the store to the caller.
32pub(super) async fn resolve_association_state(
33    conn: &impl DbQuery,
34    inbox_id: &str,
35    to_sequence_id: Option<i64>,
36    scw_verifier: &impl SmartContractSignatureVerifier,
37) -> Result<ResolvedState, IdentityUpdateError> {
38    let updates = conn.get_identity_updates(inbox_id, None, to_sequence_id)?;
39    let last_sequence_id = updates
40        .last()
41        .ok_or::<IdentityUpdateError>(AssociationError::MissingIdentityUpdate.into())?
42        .sequence_id;
43    if let Some(to_sequence_id) = to_sequence_id
44        && to_sequence_id != last_sequence_id
45    {
46        return Err(AssociationError::MissingIdentityUpdate.into());
47    }
48
49    if let Some(association_state) = conn.read_from_cache(inbox_id, last_sequence_id)? {
50        return Ok(ResolvedState::Cached(
51            association_state.try_into().map_err(StorageError::from)?,
52        ));
53    }
54
55    let unverified_updates = updates
56        .into_iter()
57        // deserialize identity update payload
58        .map(StoredIdentityUpdate::to_unverified)
59        .collect::<Result<Vec<UnverifiedIdentityUpdate>, AssociationError>>()?;
60    let updates = verify_updates(unverified_updates, scw_verifier).await?;
61
62    Ok(ResolvedState::Computed {
63        sequence_id: last_sequence_id,
64        state: get_state(updates)?,
65    })
66}
67
68/// [`get_association_state_with_verifier`](super::get_association_state_with_verifier)
69/// for a client: a computed state is
70/// stored by [`store_association_state`], which emits `handle.changed`.
71pub async fn get_association_state_for_client(
72    context: &impl IdentityContext,
73    conn: &impl DbQuery,
74    inbox_id: &str,
75    to_sequence_id: Option<i64>,
76) -> Result<AssociationState, IdentityUpdateError> {
77    let verifier = context.scw_verifier();
78    match resolve_association_state(conn, inbox_id, to_sequence_id, &verifier).await? {
79        ResolvedState::Cached(state) => Ok(state),
80        ResolvedState::Computed { sequence_id, state } => {
81            store_association_state(context, sequence_id, &state)?;
82            Ok(state)
83        }
84    }
85}
86
87/// Store the association state of an inbox at `sequence_id` in the client's
88/// cache. In the same write, read the newest state the client already holds
89/// for the inbox, and buffer `handle.changed` when the stored state changes
90/// its display handle. The event is emitted after the write commits.
91// implements: EVENT-058
92pub fn store_association_state(
93    context: &impl IdentityContext,
94    sequence_id: i64,
95    state: &AssociationState,
96) -> Result<(), IdentityUpdateError> {
97    use xmtp_events::{ClientEvent, EventWriter};
98    let inbox_id = state.inbox_id();
99    xmtp_mls_core::state_tx::state_write_with_events(
100        context.mls_storage(),
101        context.events(),
102        |tx, events| {
103            let storage = tx.storage();
104            let db = storage.db();
105            let held = db
106                .newest_cached_states(&[inbox_id])?
107                .remove(inbox_id)
108                .map(|(held_sequence, held)| {
109                    AssociationState::try_from(held).map(|held| (held_sequence, held))
110                })
111                .transpose()
112                .map_err(StorageError::from)?;
113            db.write_to_cache(inbox_id.to_owned(), sequence_id, state.clone().into())?;
114            if let Some(change) = handle_change(held.as_ref(), sequence_id, state) {
115                events.emit(Some(ClientEvent::HandleChanged(change)), None);
116            }
117            Ok::<_, StorageError>(TransactionOutcome::Continue(()))
118        },
119    )?
120    .into_continued();
121    Ok(())
122}
123
124/// The `handle.changed` event for storing `state` at `sequence_id`, given
125/// the newest state the client held for the inbox before the store. Storing
126/// the first state of an inbox, or a state no newer than the held one, is not
127/// a change.
128fn handle_change(
129    held: Option<&(i64, AssociationState)>,
130    sequence_id: i64,
131    state: &AssociationState,
132) -> Option<xmtp_events::HandleChanged> {
133    let (held_sequence, held) = held?;
134    if sequence_id <= *held_sequence {
135        return None;
136    }
137    let handle = state.display_handle()?;
138    (held.display_handle() != Some(handle)).then(|| xmtp_events::HandleChanged {
139        inbox_id: state.inbox_id().to_owned(),
140        handle: handle.to_string(),
141    })
142}
xmtp_mls_identity/identity_updates/state_store.rs

xmtp_mls_identity/identity_updates/
state_store.rs
