xmtp_db/
lib.rs

1#![warn(clippy::unwrap_used)]
2
3pub mod encrypted_store;
4mod errors;
5pub mod serialization;
6pub use serialization::*;
7pub mod sql_key_store;
8mod traits;
9pub use traits::*;
10pub mod xmtp_openmls_provider;
11pub use xmtp_openmls_provider::{
12    TransactionOutcome, XmtpMlsStorageProvider, XmtpOpenMlsProvider, XmtpOpenMlsProviderRef,
13    XmtpOpenMlsProviderRefMut,
14};
15#[cfg(any(feature = "test-utils", test))]
16pub mod mock;
17
18/// Benchmark-only latency-injecting SQLite VFS. Native + `bench` feature only.
19#[cfg(all(not(target_arch = "wasm32"), feature = "bench"))]
20pub mod latency_vfs;
21
22#[cfg(any(test, feature = "test-utils"))]
23pub mod test_utils;
24#[cfg(any(test, feature = "test-utils"))]
25pub use test_utils::*;
26
27pub use diesel;
28pub use encrypted_store::*;
29pub use errors::*;
30pub use xmtp_proto as proto;
31
32use diesel::connection::SimpleConnection;
33
34use crate::sql_key_store::SqlKeyStore;
35
36/// The default platform-specific store
37pub type DefaultStore = EncryptedMessageStore<database::DefaultDatabase>;
38pub type DefaultDbConnection = <DefaultStore as XmtpDb>::DbQuery;
39pub type DefaultMlsStore = SqlKeyStore<<DefaultStore as XmtpDb>::DbQuery>;
40
41pub mod prelude {
42    pub use super::ReadOnly;
43    pub use super::association_state::QueryAssociationStateCache;
44    pub use super::consent_record::QueryConsentRecord;
45    pub use super::conversation_list::QueryConversationList;
46    pub use super::delivery::QueryDelivery;
47    pub use super::group::QueryDms;
48    pub use super::group::QueryGroup;
49    pub use super::group::QueryGroupVersion;
50    pub use super::group_intent::QueryGroupIntent;
51    pub use super::group_intent::QueryPreparedEnvelope;
52    pub use super::group_message::QueryGroupMessage;
53    pub use super::identity::QueryIdentity;
54    pub use super::identity_cache::QueryIdentityCache;
55    pub use super::identity_update::QueryIdentityUpdates;
56    pub use super::incoming_envelope::QueryIncomingEnvelope;
57    pub use super::key_package_history::QueryKeyPackageHistory;
58    pub use super::key_store_entry::QueryKeyStoreEntry;
59    pub use super::local_commit_log::QueryLocalCommitLog;
60    pub use super::migrations::QueryMigrations;
61    pub use super::notifications::QueryNotifications;
62    pub use super::pragmas::Pragmas;
63    pub use super::processed_device_sync_messages::QueryDeviceSyncMessages;
64    pub use super::readd_status::QueryReaddStatus;
65    pub use super::refresh_state::QueryRefreshState;
66    pub use super::remote_commit_log::QueryRemoteCommitLog;
67    pub use super::server_configuration::QueryServerConfiguration;
68    pub use super::tasks::QueryTasks;
69    pub use super::traits::*;
70}
71
72pub trait ReadOnly {
73    #[allow(unused)]
74    fn enable_readonly(&self) -> Result<(), StorageError>;
75
76    #[allow(unused)]
77    fn disable_readonly(&self) -> Result<(), StorageError>;
78}
79
80impl<C: ConnectionExt> ReadOnly for DbConnection<C> {
81    #[allow(unused)]
82    fn enable_readonly(&self) -> Result<(), StorageError> {
83        self.raw_query(|conn| conn.batch_execute("PRAGMA query_only = ON;"))?;
84        Ok(())
85    }
86
87    #[allow(unused)]
88    fn disable_readonly(&self) -> Result<(), StorageError> {
89        self.raw_query(|conn| conn.batch_execute("PRAGMA query_only = OFF;"))?;
90        Ok(())
91    }
92}
93
94impl<T> ReadOnly for &T
95where
96    T: ReadOnly,
97{
98    #[allow(unused)]
99    fn enable_readonly(&self) -> Result<(), StorageError> {
100        (**self).enable_readonly()
101    }
102
103    #[allow(unused)]
104    fn disable_readonly(&self) -> Result<(), StorageError> {
105        (**self).disable_readonly()
106    }
107}
108
109#[cfg(target_arch = "wasm32")]
110pub async fn init_sqlite() {
111    // This is a no-op for wasm32
112}
113#[cfg_attr(not(target_arch = "wasm32"), ctor::ctor(unsafe))]
114#[cfg(all(test, not(target_arch = "wasm32")))]
115fn test_setup() {
116    xmtp_common::logger();
117}
118
119#[cfg(not(target_arch = "wasm32"))]
120pub async fn init_sqlite() {}
121
122#[cfg(any(test, feature = "test-utils"))]
123pub mod test_util {
124    #![allow(clippy::unwrap_used)]
125
126    use crate::group_message::{ContentType, GroupMessageKind, StoredGroupMessage};
127
128    use super::*;
129    use ascii_table::AsciiTable;
130    use diesel::{
131        ExpressionMethods, QueryDsl, RunQueryDsl, SelectableHelper, connection::LoadConnection,
132        deserialize::FromSqlRow, sql_query,
133    };
134
135    impl<C: ConnectionExt> DbConnection<C> {
136        /// Create a new table and register triggers for tracking column updates
137        pub fn register_triggers(&self) {
138            tracing::info!("Registering triggers");
139            let queries = vec![
140                r#"
141                 CREATE TABLE test_metadata (
142                     intents_created INT DEFAULT 0,
143                     intents_published INT DEFAULT 0,
144                     intents_deleted INT DEFAULT 0,
145                     intents_processed INT DEFAULT 0,
146                     rowid integer PRIMARY KEY CHECK (rowid = 1) -- There can only be one meta
147                 );
148                 "#,
149                r#"
150                 -- Create a table to store history of deleted intent payload hashes
151                 CREATE TABLE deleted_intents_history (
152                     id INTEGER PRIMARY KEY AUTOINCREMENT,
153                     intent_id INTEGER NOT NULL,
154                     payload_hash BLOB,
155                     deleted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
156                 );
157                 "#,
158                r#"
159                 -- Create a table to store history of key package rotation timestamps
160                 CREATE TABLE key_package_rotation_history (
161                     id INTEGER PRIMARY KEY AUTOINCREMENT,
162                     next_key_package_rotation_ns BIGINT,
163                     updated_at BIGINT NOT NULL
164                 );
165                 "#,
166                r#"
167                 -- Modify the deletion trigger to record payload hash history
168                 CREATE TRIGGER intents_deleted_tracking AFTER DELETE ON group_intents
169                 FOR EACH ROW
170                 BEGIN
171                     -- Update the counter in test_metadata
172                     UPDATE test_metadata SET intents_deleted = intents_deleted + 1;
173                     -- Insert the deleted intent's information into history table
174                     INSERT INTO deleted_intents_history (intent_id, payload_hash)
175                     VALUES (OLD.id, OLD.payload_hash);
176                 END;
177                 "#,
178                r#"CREATE TRIGGER intents_created_tracking AFTER INSERT on group_intents
179                 BEGIN
180                     UPDATE test_metadata SET intents_created = intents_created + 1;
181                 END;"#,
182                r#"CREATE TRIGGER intents_published_tracking AFTER UPDATE OF state ON group_intents
183                 FOR EACH ROW
184                 WHEN NEW.state = 2 AND OLD.state !=2
185                 BEGIN
186                     UPDATE test_metadata SET intents_published = intents_published + 1;
187                 END;"#,
188                r#"CREATE TRIGGER intents_processed_tracking AFTER UPDATE OF state ON group_intents
189                 FOR EACH ROW
190                 WHEN NEW.state = 5
191                 BEGIN
192                     UPDATE test_metadata SET intents_processed = intents_processed + 1;
193                 END;"#,
194                r#"
195                 CREATE TRIGGER track_key_package_rotation AFTER UPDATE OF next_key_package_rotation_ns ON identity
196                 FOR EACH ROW
197                 WHEN OLD.next_key_package_rotation_ns IS NOT NEW.next_key_package_rotation_ns
198                 BEGIN
199                     INSERT INTO key_package_rotation_history (next_key_package_rotation_ns, updated_at)
200                     VALUES (NEW.next_key_package_rotation_ns, (strftime('%s', 'now') || substr(strftime('%f', 'now'), 4)) * 1);
201                 END;
202                 "#,
203                r#"INSERT INTO test_metadata (
204                     intents_created,
205                     intents_deleted,
206                     intents_published,
207                     intents_processed
208                 ) VALUES (0, 0, 0, 0);"#,
209            ];
210            for query in queries {
211                let query = diesel::sql_query(query);
212                let _ = self.raw_query(|conn| query.execute(conn)).unwrap();
213            }
214        }
215
216        /// Disable sqlcipher memory security
217        pub fn disable_memory_security(&self) {
218            let query = r#"PRAGMA cipher_memory_security = OFF"#;
219            let query = diesel::sql_query(query);
220            let _ = self.raw_query(|c| query.clone().execute(c)).unwrap();
221            let _ = self.raw_query(|c| query.execute(c)).unwrap();
222        }
223
224        pub fn intents_published(&self) -> i32 {
225            self.raw_query(|conn| {
226                let mut row = conn
227                    .load(sql_query(
228                        "SELECT intents_published FROM test_metadata WHERE rowid = 1",
229                    ))
230                    .unwrap();
231                let row = row.next().unwrap().unwrap();
232                Ok(
233                    <i32 as FromSqlRow<diesel::sql_types::Integer, _>>::build_from_row(&row)
234                        .unwrap(),
235                )
236            })
237            .unwrap()
238        }
239
240        pub fn intents_processed(&self) -> i32 {
241            self.raw_query(|conn| {
242                let mut row = conn
243                    .load(sql_query(
244                        "SELECT intents_processed FROM test_metadata WHERE rowid = 1",
245                    ))
246                    .unwrap();
247                let row = row.next().unwrap().unwrap();
248                Ok(
249                    <i32 as FromSqlRow<diesel::sql_types::Integer, _>>::build_from_row(&row)
250                        .unwrap(),
251                )
252            })
253            .unwrap()
254        }
255
256        pub fn intents_deleted(&self) -> i32 {
257            self.raw_query(|conn| {
258                let mut row = conn
259                    .load(sql_query("SELECT intents_deleted FROM test_metadata"))
260                    .unwrap();
261                let row = row.next().unwrap().unwrap();
262                Ok(
263                    <i32 as FromSqlRow<diesel::sql_types::Integer, _>>::build_from_row(&row)
264                        .unwrap(),
265                )
266            })
267            .unwrap()
268        }
269
270        pub fn intent_payloads_deleted(&self) -> Vec<Vec<u8>> {
271            let mut hashes = vec![];
272            self.raw_query(|conn| {
273                let row = conn
274                    .load(sql_query(
275                        "SELECT payload_hash FROM deleted_intents_history",
276                    ))
277                    .unwrap();
278                for r in row {
279                    hashes.push(
280                        <Vec<u8> as FromSqlRow<diesel::sql_types::Binary, _>>::build_from_row(
281                            &r.unwrap(),
282                        )
283                        .unwrap(),
284                    );
285                }
286                Ok(())
287            })
288            .unwrap();
289            hashes
290        }
291
292        pub fn intents_created(&self) -> i32 {
293            self.raw_query(|conn| {
294                let mut row = conn
295                    .load(sql_query("SELECT intents_created FROM test_metadata"))
296                    .unwrap();
297                let row = row.next().unwrap().unwrap();
298                Ok(
299                    <i32 as FromSqlRow<diesel::sql_types::Integer, _>>::build_from_row(&row)
300                        .unwrap(),
301                )
302            })
303            .unwrap()
304        }
305
306        pub fn missing_messages(&self, sequence_ids: &[u64]) -> Vec<StoredGroupMessage> {
307            use crate::schema::group_messages::{self, dsl};
308            use diesel::QueryDsl;
309            let sequence_ids: Vec<i64> = sequence_ids.iter().copied().map(|id| id as i64).collect();
310            let query = dsl::group_messages
311                .filter(dsl::sequence_id.is_not_null())
312                .filter(group_messages::sequence_id.ne_all(sequence_ids))
313                .filter(group_messages::kind.eq(GroupMessageKind::Application))
314                .order(group_messages::sequence_id.asc());
315
316            self.raw_query(|conn| query.select(StoredGroupMessage::as_select()).load(conn))
317                .unwrap()
318        }
319
320        pub fn key_package_rotation_history(&self) -> Vec<(i64, i64)> {
321            let mut history = vec![];
322            self.raw_query(|conn| {
323                 let rows = conn
324                     .load(sql_query(
325                         "SELECT next_key_package_rotation_ns, updated_at FROM key_package_rotation_history ORDER BY id ASC",
326                     ))
327                     .unwrap();
328                 for row in rows {
329                     let row = row.unwrap();
330                     let rotation_ns = <i64 as FromSqlRow<diesel::sql_types::BigInt, _>>::build_from_row(&row)
331                         .unwrap();
332                     let updated_at = <i64 as FromSqlRow<diesel::sql_types::BigInt, _>>::build_from_row(&row)
333                         .unwrap();
334                     history.push((rotation_ns, updated_at));
335                 }
336                 Ok(())
337             })
338             .unwrap();
339            history
340        }
341
342        /// print refresh state and group message tables of the database to stdout in
343        /// column format.
344        pub fn print_db(&self) {
345            // matches
346            // <CR>$xmtp.org application/x-protobuf
347            // can see actual format with hex -> ascii converter
348            // this allows us to ignore noise from protobuf encoded bytes in the message field
349            let proto_content_type_header = hex::decode(
350                "0a240a08786d74702e6f726712166170706c69636174696f6e2f782d70726f746f627566",
351            )
352            .unwrap();
353            let format_msg = |m: &StoredGroupMessage| -> String {
354                if m.kind == GroupMessageKind::MembershipChange {
355                    return "transcript".to_string();
356                }
357                if m.content_type != ContentType::Unknown {
358                    return "encoded message".to_string();
359                }
360
361                if m.decrypted_message_bytes
362                    .starts_with(&proto_content_type_header)
363                {
364                    return "unknown encoded type".to_string();
365                }
366
367                match String::from_utf8(m.decrypted_message_bytes.clone()) {
368                    Ok(s) => s,
369                    Err(_) => "unknown encoded type".to_string(),
370                }
371            };
372            let mut t = AsciiTable::default();
373
374            println!("\n=== group_messages ===");
375            let msgs: Vec<crate::group_message::StoredGroupMessage> = self
376                .raw_query(|c| {
377                    crate::schema::group_messages::table
378                        .select(StoredGroupMessage::as_select())
379                        .load(c)
380                })
381                .unwrap_or_default();
382            t.column(0).set_header("id");
383            t.column(1).set_header("group_id");
384            t.column(2).set_header("sent_at");
385            t.column(3).set_header("kind");
386            t.column(4).set_header("sender_inbox_id");
387            t.column(5).set_header("delivery_status");
388            t.column(6).set_header("content_type");
389            t.column(7).set_header("sequence_id");
390            t.column(8).set_header("message");
391            let rows: Vec<Vec<String>> = msgs
392                .iter()
393                .map(|m| {
394                    vec![
395                        hex::encode(&m.id)[..16].to_string(),
396                        hex::encode(m.group_id)[..16].to_string(),
397                        m.sent_at_ns.to_string(),
398                        format!("{:?}", m.kind),
399                        m.sender_inbox_id.clone(),
400                        format!("{:?}", m.delivery_status),
401                        m.content_type.to_string(),
402                        m.sequence_id.to_string(),
403                        format_msg(m),
404                    ]
405                })
406                .collect();
407            if rows.is_empty() {
408                println!("(empty)");
409            } else {
410                t.println(rows);
411            }
412
413            let mut t = AsciiTable::default();
414            println!("\n=== refresh_state ===");
415            let states: Vec<crate::refresh_state::RefreshState> = self
416                .raw_query(|c| {
417                    crate::schema::refresh_state::table
418                        .select(crate::refresh_state::RefreshState::as_select())
419                        .load(c)
420                })
421                .unwrap_or_default();
422            t.column(0).set_header("entity_id");
423            t.column(1).set_header("entity_kind");
424            t.column(2).set_header("sequence_id");
425            let rows: Vec<Vec<String>> = states
426                .iter()
427                .map(|s| {
428                    vec![
429                        hex::encode(&s.entity_id)[..16.min(s.entity_id.len() * 2)].to_string(),
430                        format!("{:?}", s.entity_kind),
431                        s.sequence_id.to_string(),
432                    ]
433                })
434                .collect();
435            if rows.is_empty() {
436                println!("(empty)");
437            } else {
438                t.println(rows);
439            }
440        }
441    }
442}
xmtp_db/lib.rs

xmtp_db/
lib.rs
