xmtp_mls/groups/intents/
queue.rs

1use crate::groups::intents::GROUP_KEY_ROTATION_INTERVAL_NS;
2use crate::groups::{GroupError, MlsGroup, XmtpSharedContext};
3use derive_builder::Builder;
4use xmtp_db::TransactionOutcome::Continue;
5use xmtp_db::{
6    DbQuery, TransactionOutcome,
7    group_intent::{IntentKind, IntentState, NewGroupIntent, StoredGroupIntent},
8    prelude::*,
9};
10
11#[derive(Builder, Clone, Debug)]
12#[builder(setter(strip_option), build_fn(error = "GroupError", private))]
13pub struct QueueIntent {
14    #[builder(setter(into))]
15    kind: IntentKind,
16    /// not specifying data will be empty vec
17    #[builder(setter(into), default)]
18    data: Vec<u8>,
19    #[builder(setter(into), default)]
20    should_push: bool,
21}
22
23impl QueueIntentBuilder {
24    pub fn queue<C>(&mut self, group: &MlsGroup<C>) -> Result<StoredGroupIntent, GroupError>
25    where
26        C: XmtpSharedContext,
27    {
28        crate::state_tx::state_write(group.context.mls_storage(), |tx| {
29            let storage = tx.storage();
30            let db = storage.db();
31            self.queue_in(&db, group).map(Continue)
32        })
33        .map(TransactionOutcome::into_continued)
34    }
35
36    /// Queue an intent as part of the caller's state transaction.
37    pub(crate) fn queue_in<C>(
38        &mut self,
39        db: &impl DbQuery,
40        group: &MlsGroup<C>,
41    ) -> Result<StoredGroupIntent, GroupError>
42    where
43        C: XmtpSharedContext,
44    {
45        self.build()?.queue_with_conn(db, group)
46    }
47}
48
49impl QueueIntent {
50    /// Create an intent to send a message
51    pub fn send_message() -> QueueIntentBuilder {
52        let mut this = QueueIntent::builder();
53        this.kind = Some(IntentKind::SendMessage);
54        this
55    }
56
57    /// Create an intent to update the keys for a group
58    pub fn key_update() -> QueueIntentBuilder {
59        let mut this = QueueIntent::builder();
60        this.kind = Some(IntentKind::KeyUpdate);
61        this
62    }
63
64    /// Create an intent to update the metadata of a group
65    pub fn metadata_update() -> QueueIntentBuilder {
66        let mut this = QueueIntent::builder();
67        this.kind = Some(IntentKind::MetadataUpdate);
68        this
69    }
70
71    /// Create an intent to update the membership of a group
72    pub fn update_group_membership() -> QueueIntentBuilder {
73        let mut this = QueueIntent::builder();
74        this.kind = Some(IntentKind::UpdateGroupMembership);
75        this
76    }
77
78    /// Create an intent to update the admin list of a group
79    pub fn update_admin_list() -> QueueIntentBuilder {
80        let mut this = QueueIntent::builder();
81        this.kind = Some(IntentKind::UpdateAdminList);
82        this
83    }
84
85    /// create an intent to update the permissions of a group
86    pub fn update_permission() -> QueueIntentBuilder {
87        let mut this = QueueIntent::builder();
88        this.kind = Some(IntentKind::UpdatePermission);
89        this
90    }
91
92    pub fn readd_installations() -> QueueIntentBuilder {
93        let mut this = QueueIntent::builder();
94        this.kind = Some(IntentKind::ReaddInstallations);
95        this
96    }
97
98    /// Create an intent to propose member updates (adds and/or removes)
99    pub fn propose_member_update() -> QueueIntentBuilder {
100        let mut this = QueueIntent::builder();
101        this.kind = Some(IntentKind::ProposeMemberUpdate);
102        this
103    }
104
105    /// Create an intent to propose group context extensions update
106    pub fn propose_group_context_extensions() -> QueueIntentBuilder {
107        let mut this = QueueIntent::builder();
108        this.kind = Some(IntentKind::ProposeGroupContextExtensions);
109        this
110    }
111
112    /// Create an intent to commit pending proposals
113    pub fn commit_pending_proposals() -> QueueIntentBuilder {
114        let mut this = QueueIntent::builder();
115        this.kind = Some(IntentKind::CommitPendingProposals);
116        this
117    }
118
119    /// Create an intent to fire the one-shot AppData-migration
120    /// bootstrap commit. The intent payload is a
121    /// `ProposeGroupContextExtensionsIntentData` carrying the
122    /// target extensions blob (four legacy XMTP extensions removed,
123    /// `RequiredCapabilities` updated to require
124    /// `ExtensionType::AppDataDictionary` and drop the legacy
125    /// extension types). The handler synthesizes the per-component
126    /// dict seeds and bundles everything into a single commit.
127    pub fn bootstrap_migration() -> QueueIntentBuilder {
128        let mut this = QueueIntent::builder();
129        this.kind = Some(IntentKind::BootstrapMigration);
130        this
131    }
132
133    /// Create an intent to write to an AppData well-known or app-range
134    /// component. The intent's `data` field carries a serialized
135    /// [`crate::groups::intents::AppDataUpdateIntentData`] — see that
136    /// type's documentation for the merge semantics of each op flavor.
137    pub fn app_data_update() -> QueueIntentBuilder {
138        let mut this = QueueIntent::builder();
139        this.kind = Some(IntentKind::AppDataUpdate);
140        this
141    }
142
143    fn builder() -> QueueIntentBuilder {
144        QueueIntentBuilder::default()
145    }
146
147    fn queue_with_conn<Ctx>(
148        self,
149        conn: &impl DbQuery,
150        group: &MlsGroup<Ctx>,
151    ) -> Result<StoredGroupIntent, GroupError>
152    where
153        Ctx: XmtpSharedContext,
154    {
155        if self.kind == IntentKind::SendMessage {
156            if let Some(existing) = conn
157                .find_group_intents(
158                    group.group_id,
159                    Some(vec![
160                        IntentState::ToPublish,
161                        IntentState::Published,
162                        IntentState::Committed,
163                    ]),
164                    Some(vec![IntentKind::SendMessage]),
165                )?
166                .into_iter()
167                .find(|intent| intent.data == self.data)
168            {
169                return Ok(existing);
170            }
171            self.maybe_insert_key_update_intent(conn, group)?;
172        }
173
174        let Self {
175            kind: intent_kind,
176            data: intent_data,
177            should_push,
178        } = self;
179
180        let intent = conn.insert_group_intent(NewGroupIntent::new(
181            intent_kind,
182            group.group_id,
183            intent_data,
184            should_push,
185        ))?;
186
187        if intent_kind != IntentKind::SendMessage {
188            conn.update_rotated_at_ns(&group.group_id)?;
189        }
190        tracing::debug!(inbox_id = group.context.inbox_id(), intent_kind = %intent_kind, "queued intent");
191
192        Ok(intent)
193    }
194
195    #[tracing::instrument(level = "trace", skip_all)]
196    fn maybe_insert_key_update_intent<Ctx>(
197        &self,
198        conn: &impl DbQuery,
199        group: &MlsGroup<Ctx>,
200    ) -> Result<(), GroupError>
201    where
202        Ctx: XmtpSharedContext,
203    {
204        let last_rotated_at_ns = conn.get_rotated_at_ns(&group.group_id)?;
205        let now_ns = xmtp_common::time::now_ns();
206        let elapsed_ns = now_ns - last_rotated_at_ns;
207        if elapsed_ns > GROUP_KEY_ROTATION_INTERVAL_NS {
208            QueueIntent::key_update()
209                .build()?
210                .queue_with_conn(conn, group)?;
211        }
212        Ok(())
213    }
214}
xmtp_mls/groups/intents/queue.rs

xmtp_mls/groups/intents/
queue.rs
