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

xmtp_mls/groups/intents/
queue.rs
