xmtp_db/encrypted_store/
remote_commit_log.rs

1use diesel::RunQueryDsl;
2
3use crate::{
4    ConnectionExt, DbConnection, impl_store, schema::remote_commit_log,
5    schema::remote_commit_log::dsl,
6};
7use diesel::{
8    Insertable, Queryable,
9    backend::Backend,
10    deserialize::{self, FromSql, FromSqlRow},
11    expression::AsExpression,
12    prelude::*,
13    serialize::{self, IsNull, Output, ToSql},
14    sql_types::Integer,
15    sqlite::Sqlite,
16};
17
18use serde::{Deserialize, Serialize};
19use xmtp_common::snippet::Snippet;
20use xmtp_proto::xmtp::mls::message_contents::CommitResult as ProtoCommitResult;
21
22use xmtp_proto::types::GroupId;
23#[derive(Insertable, Debug, Clone)]
24#[diesel(table_name = remote_commit_log)]
25pub struct NewRemoteCommitLog {
26    pub log_sequence_id: i64,
27    pub group_id: GroupId,
28    pub commit_sequence_id: i64,
29    pub commit_result: CommitResult,
30    pub applied_epoch_number: i64,
31    pub applied_epoch_authenticator: Vec<u8>,
32}
33
34impl_store!(NewRemoteCommitLog, remote_commit_log);
35
36#[derive(Insertable, Queryable, Clone)]
37#[diesel(table_name = remote_commit_log)]
38#[diesel(primary_key(rowid))]
39pub struct RemoteCommitLog {
40    pub rowid: i32,
41    // The sequence ID of the log entry on the server
42    pub log_sequence_id: i64,
43    // The group ID of the conversation
44    pub group_id: GroupId,
45    // The sequence ID of the commit being referenced
46    pub commit_sequence_id: i64,
47    // Whether the commit was successfully applied or not
48    // 1 = Applied, all other values are failures matching the protobuf enum
49    pub commit_result: CommitResult,
50    // The epoch number after the commit was applied, or the existing number otherwise
51    pub applied_epoch_number: i64,
52    // The state after the commit was applied, or the existing state otherwise
53    pub applied_epoch_authenticator: Vec<u8>,
54}
55
56impl_store!(RemoteCommitLog, remote_commit_log);
57
58#[repr(i32)]
59#[derive(Copy, Clone, Serialize, Deserialize, Eq, PartialEq, AsExpression, FromSqlRow)]
60#[diesel(sql_type = Integer)]
61pub enum CommitResult {
62    Unknown = 0,
63    Success = 1,
64    WrongEpoch = 2,
65    Undecryptable = 3,
66    Invalid = 4,
67}
68
69impl std::fmt::Debug for CommitResult {
70    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
71        let s = match self {
72            CommitResult::Unknown => "Unknown",
73            CommitResult::Success => "Success",
74            CommitResult::WrongEpoch => "WrongEpoch",
75            CommitResult::Undecryptable => "Undecryptable",
76            CommitResult::Invalid => "Invalid",
77        };
78        write!(f, "{}", s)
79    }
80}
81
82impl std::fmt::Debug for RemoteCommitLog {
83    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
84        write!(
85            f,
86            "RemoteCommitLog {{ rowid: {:?}, log_sequence_id: {:?}, group_id {:?}, commit_sequence_id: {:?}, commit_result: {:?}, applied_epoch_number: {:?}, applied_epoch_authenticator: {:?} }}",
87            self.rowid,
88            self.log_sequence_id,
89            self.group_id.as_slice().snippet(),
90            self.commit_sequence_id,
91            self.commit_result,
92            self.applied_epoch_number,
93            self.applied_epoch_authenticator.snippet()
94        )
95    }
96}
97
98impl ToSql<Integer, Sqlite> for CommitResult
99where
100    i32: ToSql<Integer, Sqlite>,
101{
102    fn to_sql<'b>(&'b self, out: &mut Output<'b, '_, Sqlite>) -> serialize::Result {
103        out.set_value(*self as i32);
104        Ok(IsNull::No)
105    }
106}
107
108impl FromSql<Integer, Sqlite> for CommitResult
109where
110    i32: FromSql<Integer, Sqlite>,
111{
112    fn from_sql(bytes: <Sqlite as Backend>::RawValue<'_>) -> deserialize::Result<Self> {
113        match i32::from_sql(bytes)? {
114            0 => Ok(Self::Unknown),
115            1 => Ok(Self::Success),
116            2 => Ok(Self::WrongEpoch),
117            3 => Ok(Self::Undecryptable),
118            4 => Ok(Self::Invalid),
119            x => Err(format!("Unrecognized variant {}", x).into()),
120        }
121    }
122}
123
124impl From<ProtoCommitResult> for CommitResult {
125    fn from(value: ProtoCommitResult) -> Self {
126        match value {
127            ProtoCommitResult::Applied => Self::Success,
128            ProtoCommitResult::WrongEpoch => Self::WrongEpoch,
129            ProtoCommitResult::Undecryptable => Self::Undecryptable,
130            ProtoCommitResult::Invalid => Self::Invalid,
131            ProtoCommitResult::Unspecified => Self::Unknown,
132        }
133    }
134}
135
136pub enum RemoteCommitLogOrder {
137    AscendingByRowid,
138    DescendingByRowid,
139}
140
141pub trait QueryRemoteCommitLog {
142    fn get_latest_remote_log_for_group(
143        &self,
144        group_id: &GroupId,
145    ) -> Result<Option<RemoteCommitLog>, crate::ConnectionError>;
146
147    fn get_remote_commit_log_after_cursor(
148        &self,
149        group_id: &GroupId,
150        after_cursor: i64,
151        order_by: RemoteCommitLogOrder,
152    ) -> Result<Vec<RemoteCommitLog>, crate::ConnectionError>;
153}
154
155impl<T> QueryRemoteCommitLog for &T
156where
157    T: QueryRemoteCommitLog,
158{
159    fn get_latest_remote_log_for_group(
160        &self,
161        group_id: &GroupId,
162    ) -> Result<Option<RemoteCommitLog>, crate::ConnectionError> {
163        (**self).get_latest_remote_log_for_group(group_id)
164    }
165
166    fn get_remote_commit_log_after_cursor(
167        &self,
168        group_id: &GroupId,
169        after_cursor: i64,
170        order_by: RemoteCommitLogOrder,
171    ) -> Result<Vec<RemoteCommitLog>, crate::ConnectionError> {
172        (**self).get_remote_commit_log_after_cursor(group_id, after_cursor, order_by)
173    }
174}
175
176impl<C: ConnectionExt> QueryRemoteCommitLog for DbConnection<C> {
177    fn get_latest_remote_log_for_group(
178        &self,
179        group_id: &GroupId,
180    ) -> Result<Option<RemoteCommitLog>, crate::ConnectionError> {
181        self.raw_query(|db| {
182            dsl::remote_commit_log
183                .filter(remote_commit_log::group_id.eq(group_id))
184                .order(remote_commit_log::log_sequence_id.desc())
185                .limit(1)
186                .first(db)
187                .optional()
188        })
189    }
190
191    fn get_remote_commit_log_after_cursor(
192        &self,
193        group_id: &GroupId,
194        after_cursor: i64,
195        order: RemoteCommitLogOrder,
196    ) -> Result<Vec<RemoteCommitLog>, crate::ConnectionError> {
197        // If a group hits more than 2^31 entries on the remote commit log rowid, we will hit this error
198        // If we want to address this we can make a new sqlite cursor table/row that stores u64 values
199        if after_cursor > i32::MAX as i64 {
200            return Err(crate::ConnectionError::Database(
201                diesel::result::Error::QueryBuilderError("Cursor value exceeds i32::MAX".into()),
202            ));
203        }
204        let after_cursor: i32 = after_cursor as i32;
205
206        let query = dsl::remote_commit_log
207            .filter(dsl::group_id.eq(group_id))
208            .filter(dsl::rowid.gt(after_cursor))
209            .filter(dsl::commit_sequence_id.ne(0));
210
211        self.raw_query(|db| match order {
212            RemoteCommitLogOrder::AscendingByRowid => query.order_by(dsl::rowid.asc()).load(db),
213            RemoteCommitLogOrder::DescendingByRowid => query.order_by(dsl::rowid.desc()).load(db),
214        })
215    }
216}
xmtp_db/encrypted_store/remote_commit_log.rs

xmtp_db/encrypted_store/
remote_commit_log.rs
