xmtp_db/encrypted_store/
local_commit_log.rs

1use super::{DbConnection, remote_commit_log::CommitResult, schema::local_commit_log::dsl};
2use crate::{ConnectionExt, impl_store, schema::local_commit_log};
3use diesel::{Insertable, Queryable, prelude::*};
4use xmtp_common::snippet::Snippet;
5use xmtp_proto::xmtp::mls::message_contents::PlaintextCommitLogEntry;
6
7use xmtp_proto::types::GroupId;
8pub enum CommitType {
9    GroupCreation,
10    BackupRestore,
11    Welcome,
12    KeyUpdate,
13    MetadataUpdate,
14    UpdateGroupMembership,
15    UpdateAdminList,
16    UpdatePermission,
17    /// A commit (authored by anyone) that removed this installation's leaf
18    /// from the group. The member merges only the public part of such a
19    /// commit and cannot derive the new epoch's secrets, so the logged entry
20    /// records the pre-commit epoch and authenticator.
21    RemovedFromGroup,
22}
23
24impl std::fmt::Display for CommitType {
25    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
26        let description = match self {
27            CommitType::GroupCreation => "GroupCreation",
28            CommitType::BackupRestore => "BackupRestore",
29            CommitType::Welcome => "Welcome",
30            CommitType::KeyUpdate => "KeyUpdate",
31            CommitType::MetadataUpdate => "MetadataUpdate",
32            CommitType::UpdateGroupMembership => "UpdateGroupMembership",
33            CommitType::UpdateAdminList => "UpdateAdminList",
34            CommitType::UpdatePermission => "UpdatePermission",
35            CommitType::RemovedFromGroup => "RemovedFromGroup",
36        };
37        write!(f, "{}", description)
38    }
39}
40
41#[derive(Insertable, Debug, Clone)]
42#[diesel(table_name = local_commit_log)]
43pub struct NewLocalCommitLog {
44    pub group_id: GroupId,
45    pub commit_sequence_id: i64,
46    pub last_epoch_authenticator: Vec<u8>,
47    pub commit_result: CommitResult,
48    pub applied_epoch_number: i64,
49    pub applied_epoch_authenticator: Vec<u8>,
50    pub error_message: Option<String>,
51    pub sender_inbox_id: Option<String>,
52    pub sender_installation_id: Option<Vec<u8>>,
53    pub commit_type: Option<String>,
54}
55
56#[derive(Queryable, Clone)]
57#[diesel(table_name = local_commit_log)]
58#[diesel(primary_key(id))]
59pub struct LocalCommitLog {
60    pub rowid: i32,
61    pub group_id: GroupId,
62    pub commit_sequence_id: i64,
63    pub last_epoch_authenticator: Vec<u8>,
64    pub commit_result: CommitResult,
65    pub applied_epoch_number: i64,
66    pub applied_epoch_authenticator: Vec<u8>,
67    pub error_message: Option<String>,
68    pub sender_inbox_id: Option<String>,
69    pub sender_installation_id: Option<Vec<u8>>,
70    pub commit_type: Option<String>,
71}
72
73impl From<&LocalCommitLog> for PlaintextCommitLogEntry {
74    fn from(local_commit_log: &LocalCommitLog) -> Self {
75        PlaintextCommitLogEntry {
76            group_id: local_commit_log.group_id.to_vec(),
77            commit_sequence_id: local_commit_log.commit_sequence_id as u64,
78            last_epoch_authenticator: local_commit_log.last_epoch_authenticator.clone(),
79            commit_result: local_commit_log.commit_result.into(),
80            applied_epoch_number: local_commit_log.applied_epoch_number as u64,
81            applied_epoch_authenticator: local_commit_log.applied_epoch_authenticator.clone(),
82        }
83    }
84}
85
86impl From<CommitResult> for i32 {
87    fn from(commit_result: CommitResult) -> Self {
88        match commit_result {
89            CommitResult::Success => {
90                xmtp_proto::xmtp::mls::message_contents::CommitResult::Applied as i32
91            }
92            CommitResult::WrongEpoch => {
93                xmtp_proto::xmtp::mls::message_contents::CommitResult::WrongEpoch as i32
94            }
95            CommitResult::Undecryptable => {
96                xmtp_proto::xmtp::mls::message_contents::CommitResult::Undecryptable as i32
97            }
98            CommitResult::Invalid => {
99                xmtp_proto::xmtp::mls::message_contents::CommitResult::Invalid as i32
100            }
101            CommitResult::Unknown => {
102                xmtp_proto::xmtp::mls::message_contents::CommitResult::Unspecified as i32
103            }
104        }
105    }
106}
107
108impl_store!(NewLocalCommitLog, local_commit_log);
109
110impl std::fmt::Debug for LocalCommitLog {
111    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
112        write!(
113            f,
114            "LocalCommitLog {{ rowid: {:?}, group_id {:?}, commit_sequence_id: {:?}, last_epoch_authenticator: {:?}, commit_result: {:?}, error_message: {:?}, applied_epoch_number: {:?}, applied_epoch_authenticator: {:?}, sender_inbox_id: {:?}, sender_installation_id: {:?}, commit_type: {:?} }}",
115            self.rowid,
116            self.group_id.as_slice().snippet(),
117            self.commit_sequence_id,
118            self.last_epoch_authenticator.snippet(),
119            self.commit_result,
120            self.error_message,
121            self.applied_epoch_number,
122            self.applied_epoch_authenticator.snippet(),
123            self.sender_inbox_id.snippet(),
124            self.sender_installation_id.snippet(),
125            self.commit_type
126        )
127    }
128}
129
130pub enum LocalCommitLogOrder {
131    AscendingByRowid,
132    DescendingByRowid,
133}
134
135pub trait QueryLocalCommitLog {
136    fn get_group_logs(
137        &self,
138        group_id: &GroupId,
139    ) -> Result<Vec<LocalCommitLog>, crate::ConnectionError>;
140
141    // Local commit log entries are returned sorted in ascending order of `rowid`
142    // Entries with `commit_sequence_id` = 0 should not be published to the remote commit log
143    fn get_local_commit_log_after_cursor(
144        &self,
145        group_id: &GroupId,
146        after_cursor: i64,
147        order_by: LocalCommitLogOrder,
148    ) -> Result<Vec<LocalCommitLog>, crate::ConnectionError>;
149
150    fn get_latest_log_for_group(
151        &self,
152        group_id: &GroupId,
153    ) -> Result<Option<LocalCommitLog>, crate::ConnectionError>;
154
155    fn get_local_commit_log_cursor(
156        &self,
157        group_id: &GroupId,
158    ) -> Result<Option<i32>, crate::ConnectionError>;
159
160    /// Rowid of the most recent chain-start entry for this group, if any.
161    /// Chain-start entries have `commit_sequence_id == 0` (Welcome /
162    /// GroupCreation / BackupRestore) and mark the beginning of the member's
163    /// current membership session.
164    fn get_latest_chain_start_rowid(
165        &self,
166        group_id: &GroupId,
167    ) -> Result<Option<i32>, crate::ConnectionError>;
168}
169
170impl<T> QueryLocalCommitLog for &T
171where
172    T: QueryLocalCommitLog,
173{
174    fn get_group_logs(
175        &self,
176        group_id: &GroupId,
177    ) -> Result<Vec<LocalCommitLog>, crate::ConnectionError> {
178        (**self).get_group_logs(group_id)
179    }
180
181    fn get_local_commit_log_after_cursor(
182        &self,
183        group_id: &GroupId,
184        after_cursor: i64,
185        order_by: LocalCommitLogOrder,
186    ) -> Result<Vec<LocalCommitLog>, crate::ConnectionError> {
187        (**self).get_local_commit_log_after_cursor(group_id, after_cursor, order_by)
188    }
189
190    fn get_latest_log_for_group(
191        &self,
192        group_id: &GroupId,
193    ) -> Result<Option<LocalCommitLog>, crate::ConnectionError> {
194        (**self).get_latest_log_for_group(group_id)
195    }
196
197    fn get_local_commit_log_cursor(
198        &self,
199        group_id: &GroupId,
200    ) -> Result<Option<i32>, crate::ConnectionError> {
201        (**self).get_local_commit_log_cursor(group_id)
202    }
203
204    fn get_latest_chain_start_rowid(
205        &self,
206        group_id: &GroupId,
207    ) -> Result<Option<i32>, crate::ConnectionError> {
208        (**self).get_latest_chain_start_rowid(group_id)
209    }
210}
211
212impl<C: ConnectionExt> QueryLocalCommitLog for DbConnection<C> {
213    fn get_group_logs(
214        &self,
215        group_id: &GroupId,
216    ) -> Result<Vec<LocalCommitLog>, crate::ConnectionError> {
217        self.raw_query(|db| {
218            dsl::local_commit_log
219                .filter(dsl::group_id.eq(group_id))
220                .order_by(dsl::rowid.asc())
221                .load(db)
222        })
223    }
224
225    // Local commit log entries are sorted by `rowid`
226    // Entries with `commit_sequence_id` = 0 should not be published to the remote commit log
227    fn get_local_commit_log_after_cursor(
228        &self,
229        group_id: &GroupId,
230        after_cursor: i64,
231        order: LocalCommitLogOrder,
232    ) -> Result<Vec<LocalCommitLog>, crate::ConnectionError> {
233        // i64 cursor is populated by i32 local_commit_log rowid value, so we should never hit this error
234        if after_cursor > i32::MAX as i64 {
235            return Err(crate::ConnectionError::Database(
236                diesel::result::Error::QueryBuilderError("Cursor value exceeds i32::MAX".into()),
237            ));
238        }
239        let after_cursor = after_cursor as i32;
240
241        let query = dsl::local_commit_log
242            .filter(dsl::group_id.eq(group_id))
243            .filter(dsl::rowid.gt(after_cursor))
244            .filter(dsl::commit_sequence_id.ne(0));
245
246        self.raw_query(|db| match order {
247            LocalCommitLogOrder::AscendingByRowid => query.order_by(dsl::rowid.asc()).load(db),
248            LocalCommitLogOrder::DescendingByRowid => query.order_by(dsl::rowid.desc()).load(db),
249        })
250    }
251
252    fn get_latest_log_for_group(
253        &self,
254        group_id: &GroupId,
255    ) -> Result<Option<LocalCommitLog>, crate::ConnectionError> {
256        self.raw_query(|db| {
257            dsl::local_commit_log
258                .filter(dsl::group_id.eq(group_id))
259                .order_by(dsl::rowid.desc())
260                .limit(1)
261                .first(db)
262                .optional()
263        })
264    }
265
266    fn get_local_commit_log_cursor(
267        &self,
268        group_id: &GroupId,
269    ) -> Result<Option<i32>, crate::ConnectionError> {
270        let query = dsl::local_commit_log
271            .filter(dsl::group_id.eq(group_id))
272            .select(dsl::rowid)
273            .order(dsl::rowid.desc())
274            .limit(1);
275
276        self.raw_query(|conn| query.first::<i32>(conn).optional())
277    }
278
279    fn get_latest_chain_start_rowid(
280        &self,
281        group_id: &GroupId,
282    ) -> Result<Option<i32>, crate::ConnectionError> {
283        let query = dsl::local_commit_log
284            .filter(dsl::group_id.eq(group_id))
285            .filter(dsl::commit_sequence_id.eq(0))
286            .select(dsl::rowid)
287            .order(dsl::rowid.desc())
288            .limit(1);
289
290        self.raw_query(|conn| query.first::<i32>(conn).optional())
291    }
292}
xmtp_db/encrypted_store/local_commit_log.rs

xmtp_db/encrypted_store/
local_commit_log.rs
