xmtp_db/encrypted_store/database/native/
sqlcipher_connection.rs

1//! SQLCipher-specific Connection
2use bon::Builder;
3use diesel::{
4    connection::{LoadConnection, SimpleConnection},
5    deserialize::FromSqlRow,
6    prelude::*,
7    result::{DatabaseErrorKind, Error as DieselError},
8    sql_query,
9};
10use std::{
11    fmt::Display,
12    fs::File,
13    io::{BufReader, Read, Write},
14    path::{Path, PathBuf},
15};
16use xmtp_configuration::BUSY_TIMEOUT;
17
18use super::PlatformStorageError;
19use crate::{
20    NotFound,
21    database::instrumentation::TestInstrumentation,
22    native::{ConnectionOptions, connection_pragmas},
23};
24
25use crate::{EncryptionKey, StorageOption};
26
27pub type Salt = [u8; 16];
28const PLAINTEXT_HEADER_SIZE: usize = 32;
29const SALT_FILE_NAME: &str = "sqlcipher_salt";
30const VALIDATION_QUERY: &str = "SELECT count(*) FROM sqlite_master;";
31
32// For PRAGMA query log statements
33#[derive(QueryableByName, Debug)]
34struct CipherVersion {
35    #[diesel(sql_type = diesel::sql_types::Text)]
36    cipher_version: String,
37}
38
39// For PRAGMA query log statements
40#[derive(QueryableByName, Debug)]
41struct CipherProviderVersion {
42    #[diesel(sql_type = diesel::sql_types::Text)]
43    cipher_provider_version: String,
44}
45
46/// Specialized Connection for r2d2 connection pool.
47#[derive(Clone, Debug, Builder, zeroize::ZeroizeOnDrop)]
48pub struct EncryptedConnection {
49    key: EncryptionKey,
50    /// We don't store the salt for Ephemeral Dbs
51    salt: Option<Salt>,
52    options: StorageOption,
53}
54
55impl EncryptedConnection {
56    /// Creates a file for the salt and stores it
57    pub fn new(key: EncryptionKey, opts: &StorageOption) -> Result<Self, PlatformStorageError> {
58        use crate::StorageOption::*;
59
60        let salt = match opts {
61            Ephemeral => None,
62            Persistent(db_path) => {
63                {
64                    let mut conn = SqliteConnection::establish(db_path)?;
65                    Self::check_for_sqlcipher(opts, &mut conn)?;
66                }
67                let mut salt = [0u8; 16];
68                let db_pathbuf = PathBuf::from(db_path);
69                let salt_path = Self::salt_file(db_path)?;
70
71                match (salt_path.try_exists()?, db_pathbuf.try_exists()?) {
72                    // db and salt exist
73                    (true, true) => {
74                        tracing::debug!(
75                            salt = %salt_path.display(),
76                            db = %db_pathbuf.display(),
77                            "salt and database exist, db=[{}], salt=[{}]",
78                            db_pathbuf.display(),
79                            salt_path.display(),
80                        );
81                        let file = BufReader::new(File::open(salt_path)?);
82                        salt = <Salt as hex::FromHex>::from_hex(
83                            file.bytes().take(32).collect::<Result<Vec<u8>, _>>()?,
84                        )?;
85                    }
86                    // the db exists and needs to be migrated
87                    (false, true) => {
88                        tracing::debug!(
89                            "migrating sqlcipher db=[{}] to plaintext header with salt=[{}]",
90                            db_pathbuf.display(),
91                            salt_path.display()
92                        );
93                        Self::migrate(db_path, &key, &mut salt)?;
94                    }
95                    // the db doesn't exist yet and needs to be created
96                    (false, false) => {
97                        tracing::debug!(
98                            "creating new sqlcipher db=[{}] with salt=[{}]",
99                            db_pathbuf.display(),
100                            salt_path.display()
101                        );
102                        Self::create(db_path, &key, &mut salt)?;
103                    }
104                    // the db doesn't exist but the salt does
105                    // This generally doesn't make sense & shouldn't happen.
106                    // Create a new database and delete the salt file.
107                    (true, false) => {
108                        tracing::debug!(
109                            "database [{}] does not exist, but the salt [{}] does, re-creating",
110                            db_pathbuf.display(),
111                            salt_path.display(),
112                        );
113                        std::fs::remove_file(salt_path)?;
114                        Self::create(db_path, &key, &mut salt)?;
115                    }
116                }
117                tracing::info!("db_path=[{}]", db_path);
118                Some(salt)
119            }
120        };
121
122        Ok(Self {
123            key,
124            salt,
125            options: opts.clone(),
126        })
127    }
128
129    /// create a new database + salt file.
130    /// writes the 16-bytes hex-encoded salt to `salt`
131    fn create(
132        path: &String,
133        key: &EncryptionKey,
134        salt: &mut [u8],
135    ) -> Result<(), PlatformStorageError> {
136        let conn = &mut SqliteConnection::establish(path)?;
137        conn.batch_execute(&format!(
138            r#"
139            {}
140            {}
141            "#,
142            pragma_key(hex::encode(key)),
143            pragma_plaintext_header()
144        ))?;
145
146        Self::write_salt(path, conn, salt)?;
147        Ok(())
148    }
149
150    /// Executes the steps outlined in the [SQLCipher Docs](https://www.zetetic.net/sqlcipher/sqlcipher-api/#cipher_plaintext_header_size)
151    /// Migrates the database to `cipher_plaintext_header_size` and returns the salt after
152    /// persisting it to SALT_FILE_NAME.
153    ///
154    /// if the salt file already exists, deletes it.
155    fn migrate(
156        path: &String,
157        key: &EncryptionKey,
158        salt: &mut [u8],
159    ) -> Result<(), PlatformStorageError> {
160        let conn = &mut SqliteConnection::establish(path)?;
161
162        conn.batch_execute(&format!(
163            r#"
164            {}
165            select count(*) from sqlite_master; -- trigger header read, currently it is encrypted
166        "#,
167            pragma_key(hex::encode(key))
168        ))?;
169
170        // get the salt and save it for later use
171        Self::write_salt(path, conn, salt)?;
172
173        conn.batch_execute(&format!(
174            r#"
175            {}
176            PRAGMA user_version = 1; -- force header write
177        "#,
178            pragma_plaintext_header()
179        ))?;
180
181        Ok(())
182    }
183
184    /// Get the salt from the opened database, write it to `Self::salt_file(db_path)` as hex-encoded
185    /// bytes, and then copy it to `buf` after decoding hex bytes.
186    fn write_salt(
187        path: &String,
188        conn: &mut SqliteConnection,
189        buf: &mut [u8],
190    ) -> Result<(), PlatformStorageError> {
191        let mut row_iter = conn.load(sql_query("PRAGMA cipher_salt"))?;
192        // cipher salt should always exist. if it doesn't SQLCipher is misconfigured.
193        let row = row_iter
194            .next()
195            .ok_or(NotFound::CipherSalt(path.to_string()))??;
196        let salt = <String as FromSqlRow<diesel::sql_types::Text, _>>::build_from_row(&row)?;
197        tracing::debug!(
198            salt,
199            file = %Self::salt_file(PathBuf::from(path))?.display(),
200            "writing salt to file"
201        );
202        let mut f = File::create(Self::salt_file(PathBuf::from(path))?)?;
203
204        f.write_all(salt.as_bytes())?;
205        let mut perms = f.metadata()?.permissions();
206        perms.set_readonly(true);
207        f.set_permissions(perms)?;
208
209        let salt = hex::decode(salt)?;
210        buf.copy_from_slice(&salt);
211        Ok(())
212    }
213
214    /// Salt file is stored next to the sqlite3 db3 file as `{db_file_name}.SALT_FILE_NAME`.
215    /// If the db file is named `sqlite3_xmtp_db.db3`, the salt file would
216    /// be stored next to this file as `sqlite3_xmtp_db.db3.sqlcipher_salt`
217    pub(crate) fn salt_file<P: AsRef<Path>>(db_path: P) -> std::io::Result<PathBuf> {
218        let db_path: &Path = db_path.as_ref();
219        let name = db_path.file_name().ok_or(std::io::Error::new(
220            std::io::ErrorKind::NotFound,
221            "database file has no name",
222        ))?;
223        let db_path = db_path.parent().ok_or(std::io::Error::new(
224            std::io::ErrorKind::NotFound,
225            "Parent directory could not be found",
226        ))?;
227        Ok(db_path.join(format!("{}.{}", name.to_string_lossy(), SALT_FILE_NAME)))
228    }
229
230    /// Output the correct order of PRAGMAS to instantiate a connection
231    fn pragmas(&self) -> impl Display {
232        let Self { key, salt, .. } = self;
233
234        if let Some(s) = salt {
235            format!(
236                "{}\n{}\n{}",
237                pragma_key(hex::encode(key)),
238                pragma_plaintext_header(),
239                pragma_salt(hex::encode(s))
240            )
241        } else {
242            format!(
243                "{}\n{}",
244                pragma_key(hex::encode(key)),
245                pragma_plaintext_header()
246            )
247        }
248    }
249
250    fn check_for_sqlcipher(
251        opts: &StorageOption,
252        conn: &mut SqliteConnection,
253    ) -> Result<CipherVersion, PlatformStorageError> {
254        if cfg!(any(test, feature = "test-utils")) {
255            conn.batch_execute("pragma cipher_log = stdout; pragma cipher_log_level = NONE;")?;
256        }
257
258        if let Some(path) = opts.path() {
259            let exists = std::path::Path::new(path).exists();
260            tracing::debug!("db @ [{}] exists? [{}]", path, exists);
261        }
262        let mut cipher_version = sql_query("PRAGMA cipher_version").load::<CipherVersion>(conn)?;
263        if cipher_version.is_empty() {
264            return Err(PlatformStorageError::SqlCipherNotLoaded);
265        }
266        Ok(cipher_version.pop().expect("checked for empty"))
267    }
268}
269
270impl ConnectionOptions for EncryptedConnection {
271    fn options(&self) -> &StorageOption {
272        &self.options
273    }
274}
275
276impl super::ValidatedConnection for EncryptedConnection {
277    fn validate(&self, conn: &mut SqliteConnection) -> Result<(), PlatformStorageError> {
278        let sqlcipher_version = EncryptedConnection::check_for_sqlcipher(&self.options, conn)?;
279
280        conn.batch_execute(&self.pragmas().to_string())?;
281        // Validation uses a separate connection before pool setup. A closing WAL
282        // connection can still hold a lock, so set the timeout before this read.
283        conn.batch_execute(&format!("PRAGMA busy_timeout = {BUSY_TIMEOUT};"))?;
284
285        // Test the key by reading the schema, after applying the encryption settings.
286        // https://www.zetetic.net/sqlcipher/sqlcipher-api/#testing-the-key
287        conn.batch_execute(VALIDATION_QUERY)
288            .map_err(validation_error)?;
289
290        let CipherProviderVersion {
291            cipher_provider_version,
292        } = sql_query("PRAGMA cipher_provider_version")
293            .get_result::<CipherProviderVersion>(conn)?;
294        tracing::info!(
295            "Sqlite cipher_version={:?}, cipher_provider_version={:?}",
296            sqlcipher_version.cipher_version,
297            cipher_provider_version
298        );
299        let log = std::env::var("SQLCIPHER_LOG");
300        let is_sqlcipher_log_enabled = matches!(log, Ok(s) if s == "true" || s == "1");
301        //TODO: Can log to logcat/oslog for ios/android
302        if is_sqlcipher_log_enabled {
303            conn.batch_execute("PRAGMA cipher_log = stderr; PRAGMA cipher_log_level = INFO;")
304                .ok();
305        }
306        tracing::debug!("SQLCipher Database validated.");
307        Ok(())
308    }
309}
310
311/// Keep unreadable encrypted databases non-retryable without hiding other errors.
312fn validation_error(error: DieselError) -> PlatformStorageError {
313    tracing::error!("SQLCipher schema validation failed: {error:?}");
314    // The pinned Diesel SQLite backend drops the numeric error code. SQLCipher
315    // also uses SQLITE_ERROR ("SQL logic error") for failed page authentication.
316    // Match these unreadable-database signals only for the fixed schema query,
317    // not arbitrary SQL or pragma setup. Locks and I/O errors keep their cause.
318    match error {
319        DieselError::DatabaseError(DatabaseErrorKind::Unknown, ref info)
320            if matches!(
321                info.message(),
322                "file is not a database" | "database disk image is malformed" | "SQL logic error"
323            ) || info.message().starts_with("malformed database schema (") =>
324        {
325            PlatformStorageError::SqlCipherKeyIncorrect
326        }
327        error => PlatformStorageError::DieselResult(error),
328    }
329}
330
331impl diesel::r2d2::CustomizeConnection<SqliteConnection, diesel::r2d2::Error>
332    for EncryptedConnection
333{
334    fn on_acquire(&self, conn: &mut SqliteConnection) -> Result<(), diesel::r2d2::Error> {
335        if cfg!(any(test, feature = "test-utils")) {
336            conn.set_instrumentation(TestInstrumentation);
337        }
338        conn.batch_execute(&format!("{}", self.pragmas(),))
339            .map_err(diesel::r2d2::Error::QueryError)?;
340        connection_pragmas(conn)?;
341        Ok(())
342    }
343}
344
345fn pragma_key(key: impl Display) -> impl Display {
346    format!(r#"PRAGMA key = "x'{key}'";"#)
347}
348
349fn pragma_salt(salt: impl Display) -> impl Display {
350    format!(r#"PRAGMA cipher_salt="x'{salt}'";"#)
351}
352
353fn pragma_plaintext_header() -> impl Display {
354    format!(r#"PRAGMA cipher_plaintext_header_size={PLAINTEXT_HEADER_SIZE};"#)
355}
356
357#[cfg(test)]
358mod tests {
359    use crate::{EncryptedMessageStore, NativeDb, ValidatedConnection, XmtpTestDb};
360    use diesel::connection::InstrumentationEvent;
361    use diesel_migrations::MigrationHarness;
362    use std::{fs::File, sync::mpsc};
363    use xmtp_common::{ErrorCode, RetryableError, time::Duration, tmp_path};
364
365    use super::*;
366    const SQLITE3_PLAINTEXT_HEADER: &str = "SQLite format 3\0";
367    const VALIDATION_KEY: [u8; 32] = [7; 32];
368    const LOCK_PROBE_TIMEOUT: Duration = Duration::from_millis(100);
369    const VALIDATION_START_TIMEOUT: Duration = Duration::from_secs(10);
370    use StorageOption::*;
371
372    /// Keep a real encrypted WAL database locked until the caller drops the connection.
373    fn locked_encrypted_database() -> (String, EncryptedConnection, SqliteConnection) {
374        let path = tmp_path();
375        let customizer =
376            EncryptedConnection::new(VALIDATION_KEY.into(), &Persistent(path.clone())).unwrap();
377        let mut conn = SqliteConnection::establish(&path).unwrap();
378        customizer.validate(&mut conn).unwrap();
379        conn.batch_execute(
380            "CREATE TABLE validation_fixture (value INTEGER NOT NULL);
381             INSERT INTO validation_fixture VALUES (1);
382             PRAGMA journal_mode = WAL;
383             PRAGMA locking_mode = EXCLUSIVE;
384             BEGIN EXCLUSIVE;",
385        )
386        .unwrap();
387        (path, customizer, conn)
388    }
389
390    #[xmtp_common::test(unwrap_try = true)]
391    async fn same_key_reopen_waits_for_transient_lock() {
392        let (path, customizer, locked) = locked_encrypted_database();
393        let mut reopened = SqliteConnection::establish(&path)?;
394        let (started_tx, started_rx) = mpsc::channel();
395        let (finished_tx, finished_rx) = mpsc::channel();
396        reopened.set_instrumentation(move |event: InstrumentationEvent<'_>| {
397            if let InstrumentationEvent::StartQuery { query, .. } = event
398                && query.to_string() == VALIDATION_QUERY
399            {
400                started_tx.send(()).unwrap();
401            }
402        });
403        let validation = std::thread::spawn(move || {
404            let result = customizer.validate(&mut reopened);
405            finished_tx.send(()).unwrap();
406            result.map(|()| reopened)
407        });
408
409        // Wait until the schema read starts before checking that it waits for the lock.
410        let started = started_rx.recv_timeout(VALIDATION_START_TIMEOUT);
411        let waiting = finished_rx.recv_timeout(LOCK_PROBE_TIMEOUT);
412        drop(locked);
413        let result = validation.join().unwrap();
414        started?;
415        assert!(matches!(waiting, Err(mpsc::RecvTimeoutError::Timeout)));
416        let mut reopened = result?;
417        let timeout = {
418            let mut rows = reopened.load(sql_query("PRAGMA busy_timeout"))?;
419            let row = rows.next().unwrap()?;
420            <i32 as FromSqlRow<diesel::sql_types::Integer, _>>::build_from_row(&row)?
421        };
422        assert_eq!(timeout, BUSY_TIMEOUT);
423        let value = diesel::select(diesel::dsl::sql::<diesel::sql_types::Integer>(
424            "(SELECT value FROM validation_fixture)",
425        ))
426        .get_result::<i32>(&mut reopened)?;
427        assert_eq!(value, 1);
428        drop(reopened);
429        EncryptedMessageStore::<()>::remove_db_files(path);
430    }
431
432    #[xmtp_common::test(unwrap_try = true)]
433    async fn expired_reopen_lock_keeps_database_error() {
434        let (path, _customizer, locked) = locked_encrypted_database();
435        let error = NativeDb::builder()
436            .persistent(path.clone())
437            .key(VALIDATION_KEY)
438            .build()
439            .unwrap_err();
440        drop(locked);
441        assert!(error.is_retryable());
442        assert!(matches!(
443            error,
444            crate::StorageError::Platform(PlatformStorageError::DieselResult(
445                DieselError::DatabaseError(DatabaseErrorKind::Unknown, ref info)
446            )) if info.message() == "database is locked"
447        ));
448        EncryptedMessageStore::<()>::remove_db_files(path);
449    }
450
451    #[rstest::rstest]
452    #[case("file is not a database")]
453    #[case("database disk image is malformed")]
454    #[case("malformed database schema (identity) - invalid rootpage")]
455    #[xmtp_common::test(unwrap_try = true)]
456    async fn schema_corruption_stays_non_retryable(#[case] message: &str) {
457        let error = validation_error(DieselError::DatabaseError(
458            DatabaseErrorKind::Unknown,
459            Box::new(message.to_string()),
460        ));
461        assert!(matches!(error, PlatformStorageError::SqlCipherKeyIncorrect));
462        assert_eq!(
463            error.error_code(),
464            "PlatformStorageError::SqlCipherKeyIncorrect"
465        );
466        assert!(!error.is_retryable());
467    }
468
469    #[tokio::test]
470    async fn test_sqlcipher_version() {
471        let db_path = tmp_path();
472        {
473            let opts = Persistent(db_path.clone());
474            let mut conn = SqliteConnection::establish(&db_path).unwrap();
475            let v = EncryptedConnection::check_for_sqlcipher(&opts, &mut conn).unwrap();
476            println!("SQLCipher Version {}", v.cipher_version);
477        }
478    }
479
480    #[tokio::test]
481    async fn test_db_creates_with_plaintext_header() {
482        let db_path = tmp_path();
483        {
484            let _ = crate::TestDb::create_persistent_store(Some(db_path.clone())).await;
485
486            assert!(EncryptedConnection::salt_file(&db_path).unwrap().exists());
487            let bytes = std::fs::read(EncryptedConnection::salt_file(&db_path).unwrap()).unwrap();
488            let salt = hex::decode(bytes).unwrap();
489            assert_eq!(salt.len(), 16);
490
491            let mut plaintext_header = [0; 16];
492            let mut file = File::open(&db_path).unwrap();
493            file.read_exact(&mut plaintext_header).unwrap();
494
495            assert_eq!(
496                SQLITE3_PLAINTEXT_HEADER,
497                String::from_utf8(plaintext_header.into()).unwrap()
498            );
499        }
500        EncryptedMessageStore::<()>::remove_db_files(db_path)
501    }
502
503    #[xmtp_common::test(unwrap_try = true)]
504    async fn test_db_migrates() {
505        use crate::{ConnectionExt, XmtpDb};
506        use diesel::sql_types::{Integer, Text};
507        use std::collections::BTreeMap;
508
509        #[derive(QueryableByName)]
510        struct TableName {
511            #[diesel(sql_type = Text)]
512            name: String,
513        }
514        #[derive(QueryableByName)]
515        struct Column {
516            #[diesel(sql_type = Text)]
517            name: String,
518            #[diesel(sql_type = Text)]
519            r#type: String,
520            #[diesel(sql_type = Integer)]
521            notnull: i32,
522            #[diesel(sql_type = Integer)]
523            pk: i32,
524        }
525
526        let db = NativeDb::builder().ephemeral().build_unencrypted()?;
527        db.init()?;
528        let actual = db.conn().raw_query(|conn| {
529            let tables = diesel::sql_query(
530                "SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%' AND name != '__diesel_schema_migrations' ORDER BY name",
531            ).load::<TableName>(conn)?;
532            let mut schema = BTreeMap::new();
533            for table in tables {
534                let columns = diesel::sql_query(format!("PRAGMA table_info('{}')", table.name))
535                    .load::<Column>(conn)?;
536                let has_primary_key = columns.iter().any(|column| column.pk != 0);
537                let mut columns: Vec<_> = columns.into_iter().map(|column| {
538                    let sql_type = match column.r#type.to_uppercase().as_str() {
539                        "INT" | "INTEGER" => "Integer",
540                        "BIGINT" => "BigInt",
541                        "BLOB" => "Binary",
542                        "TEXT" => "Text",
543                        "BOOL" | "BOOLEAN" => "Bool",
544                        "REAL" => "Float",
545                        other => panic!("unexpected SQL type {other}"),
546                    };
547                    let sql_type = if column.notnull == 0 {
548                        format!("Nullable<{sql_type}>")
549                    } else { sql_type.to_string() };
550                    (column.name, sql_type)
551                }).collect();
552                if !has_primary_key { columns.insert(0, ("rowid".into(), "Integer".into())); }
553                schema.insert(table.name, columns);
554            }
555            diesel::sql_query("SELECT * FROM conversation_list").execute(conn)?;
556            assert_eq!(conn.applied_migrations().unwrap().len(), 1);
557            Ok(schema)
558        })?;
559
560        let mut expected = BTreeMap::new();
561        let mut table = None;
562        for line in include_str!("../../schema_gen.rs").lines().map(str::trim) {
563            if line.ends_with("{") && line.contains(" (") {
564                let name = line.split_once(" (").unwrap().0.to_string();
565                expected.insert(name.clone(), Vec::new());
566                table = Some(name);
567            } else if line == "}" {
568                table = None;
569            } else if table.is_some()
570                && let Some((name, sql_type)) = line.split_once(" -> ")
571            {
572                expected
573                    .get_mut(table.as_ref().unwrap())
574                    .unwrap()
575                    .push((name.to_string(), sql_type.trim_end_matches(',').to_string()));
576            }
577        }
578        assert_eq!(actual, expected);
579    }
580
581    #[xmtp_common::test(unwrap_try = true)]
582    async fn rejects_pre_transition_database_before_migration() {
583        use crate::{ConnectionExt, StorageError, XmtpDb};
584        use xmtp_common::{ErrorCode, RetryableError};
585
586        let path = tmp_path();
587        let db = NativeDb::builder()
588            .persistent(path.clone())
589            .build_unencrypted()?;
590        db.conn().raw_query(|conn| conn.batch_execute(
591            "CREATE TABLE __diesel_schema_migrations (version VARCHAR(50) PRIMARY KEY NOT NULL, run_on TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP); INSERT INTO __diesel_schema_migrations (version) VALUES ('20250820174800');",
592        ))?;
593        let result = EncryptedMessageStore::new(db);
594        let error = match result {
595            Err(error) => error,
596            Ok(_) => panic!("legacy database was accepted"),
597        };
598        assert!(matches!(error, StorageError::PreTransitionDatabase));
599        assert_eq!(error.error_code(), "StorageError::PreTransitionDatabase");
600        assert!(!error.is_retryable());
601        let mut conn = SqliteConnection::establish(&path)?;
602        assert!(
603            diesel::sql_query("SELECT * FROM group_messages")
604                .execute(&mut conn)
605                .is_err()
606        );
607        assert_eq!(conn.applied_migrations()?.len(), 1);
608        drop(conn);
609        std::fs::remove_file(path)?;
610    }
611
612    #[xmtp_common::test(unwrap_try = true)]
613    async fn rejects_old_self_hosted_format_without_changing_data() {
614        use crate::encrypted_store::EmbeddedMigrationsExt;
615        use crate::{ConnectionExt, StorageError, TestDb, XmtpDb, XmtpTestDb};
616        use diesel::sql_types::Text;
617
618        let database = TestDb::create_database(None).await;
619        let connection = database.conn();
620        connection.raw_query(|conn| {
621            conn.batch_execute(
622                "CREATE TABLE __diesel_schema_migrations (version VARCHAR(50) PRIMARY KEY NOT NULL, run_on TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP); CREATE TABLE refresh_state (entity_id BLOB NOT NULL, entity_kind INTEGER NOT NULL, sequence_id BIGINT NOT NULL, PRIMARY KEY(entity_id, entity_kind)); INSERT INTO refresh_state VALUES (x'01', 2, 42);",
623            )?;
624            diesel::sql_query("INSERT INTO __diesel_schema_migrations(version) VALUES (?)")
625                .bind::<Text, _>(crate::MIGRATIONS.final_migration()).execute(conn)?;
626            Ok(())
627        })?;
628        let result = EncryptedMessageStore::new(database);
629        assert!(matches!(result, Err(StorageError::OldStreamDatabase)));
630        let value = connection.raw_query(|conn| {
631            crate::schema::refresh_state::table
632                .select(crate::schema::refresh_state::sequence_id)
633                .first::<i64>(conn)
634        })?;
635        assert_eq!(value, 42);
636        assert!(
637            connection
638                .raw_query(
639                    |conn| diesel::sql_query("SELECT * FROM incoming_envelopes").execute(conn)
640                )
641                .is_err()
642        );
643    }
644
645    /// Spec 006 CFG-044: the baseline gained `server_configuration`. Diesel
646    /// records one version for the whole baseline, so a database from an
647    /// earlier self-hosted build is never re-migrated. Initialization must
648    /// reject it here rather than let the configuration queries meet
649    /// "no such table: server_configuration" at client build.
650    #[xmtp_common::test(unwrap_try = true)]
651    async fn rejects_a_database_missing_the_server_configuration_table() {
652        use crate::encrypted_store::EmbeddedMigrationsExt;
653        use crate::{ConnectionExt, StorageError, TestDb, XmtpDb, XmtpTestDb};
654        use diesel::sql_types::Text;
655
656        let database = TestDb::create_database(None).await;
657        let connection = database.conn();
658        connection.raw_query(|conn| {
659            conn.batch_execute(
660                "CREATE TABLE __diesel_schema_migrations (version VARCHAR(50) PRIMARY KEY NOT NULL, run_on TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP); CREATE TABLE refresh_state (entity_id BLOB NOT NULL, entity_kind INTEGER NOT NULL, sequence_id BIGINT NOT NULL, received_sequence_id BIGINT NOT NULL DEFAULT 0, PRIMARY KEY(entity_id, entity_kind));",
661            )?;
662            diesel::sql_query("INSERT INTO __diesel_schema_migrations(version) VALUES (?)")
663                .bind::<Text, _>(crate::MIGRATIONS.final_migration())
664                .execute(conn)?;
665            Ok(())
666        })?;
667        let result = EncryptedMessageStore::new(database);
668        assert!(matches!(result, Err(StorageError::OldStreamDatabase)));
669    }
670}
xmtp_db/encrypted_store/database/native/sqlcipher_connection.rs

xmtp_db/encrypted_store/database/native/
sqlcipher_connection.rs
