encryption.rs

use crate::common::{
    do_flush, run_query, run_query_on_row, ExecRows, TempDatabase, TempDatabaseBuilder,
};
use rand::{rng, RngCore};
use std::sync::Arc;
use turso_core::{
    CipherMode, Database, DatabaseOpts, EncryptionKey, EncryptionOpts, OpenFlags, PlatformIO, Row,
    IO,
};

const ENABLE_ENCRYPTION: bool = true;

fn run_non_4k_page_size_encryption_test(
    tmp_db: &TempDatabase,
    enable_mvcc: bool,
) -> anyhow::Result<()> {
    let _ = env_logger::try_init();
    let db_path = tmp_db.path.clone();

    {
        let conn = tmp_db.connect_limbo();
        // Set page size to 8k (8192 bytes) and test encryption. Default page size is 4k.
        run_query(tmp_db, &conn, "PRAGMA page_size = 8192;")?;
        run_query(
            tmp_db,
            &conn,
            "PRAGMA hexkey = 'b1bbfda4f589dc9daaf004fe21111e00dc00c98237102f5c7002a5669fc76327';",
        )?;
        run_query(tmp_db, &conn, "PRAGMA cipher = 'aegis256';")?;
        if enable_mvcc {
            run_query(tmp_db, &conn, "PRAGMA journal_mode = 'mvcc';")?;
        }
        run_query(
            tmp_db,
            &conn,
            "CREATE TABLE test (id INTEGER PRIMARY KEY, value TEXT);",
        )?;
        run_query(
            tmp_db,
            &conn,
            "INSERT INTO test (value) VALUES ('Hello, World!')",
        )?;
        let mut row_count = 0;
        run_query_on_row(tmp_db, &conn, "SELECT * FROM test", |row: &Row| {
            assert_eq!(row.get::<i64>(0).unwrap(), 1);
            assert_eq!(row.get::<String>(1).unwrap(), "Hello, World!");
            row_count += 1;
        })?;

        assert_eq!(row_count, 1);
        do_flush(&conn, tmp_db)?;
    }

    {
        // Reopen the existing db with 8k page size and test encryption
        let uri = format!(
            "file:{}?cipher=aegis256&hexkey=b1bbfda4f589dc9daaf004fe21111e00dc00c98237102f5c7002a5669fc76327",
            db_path.to_str().unwrap()
        );
        let (_io, conn) = turso_core::Connection::from_uri(
            &uri,
            DatabaseOpts::new().with_encryption(ENABLE_ENCRYPTION),
        )?;
        run_query_on_row(tmp_db, &conn, "SELECT * FROM test", |row: &Row| {
            assert_eq!(row.get::<i64>(0).unwrap(), 1);
            assert_eq!(row.get::<String>(1).unwrap(), "Hello, World!");
        })?;
    }

    Ok(())
}

fn run_corruption_associated_data_bytes_test(
    tmp_db: &TempDatabase,
    enable_mvcc: bool,
) -> anyhow::Result<()> {
    let _ = env_logger::try_init();
    let db_path = tmp_db.path.clone();

    {
        let conn = tmp_db.connect_limbo();
        run_query(
            tmp_db,
            &conn,
            "PRAGMA hexkey = 'b1bbfda4f589dc9daaf004fe21111e00dc00c98237102f5c7002a5669fc76327';",
        )?;
        run_query(tmp_db, &conn, "PRAGMA cipher = 'aegis256';")?;
        if enable_mvcc {
            run_query(tmp_db, &conn, "PRAGMA journal_mode = 'mvcc';")?;
        }
        run_query(
            tmp_db,
            &conn,
            "CREATE TABLE test (id INTEGER PRIMARY KEY, value TEXT);",
        )?;
        run_query(
            tmp_db,
            &conn,
            "INSERT INTO test (value) VALUES ('Test AD corruption')",
        )?;
        run_query(tmp_db, &conn, "PRAGMA wal_checkpoint(TRUNCATE);")?;
        do_flush(&conn, tmp_db)?;
    }

    // test corruption at different positions in the header (the first 100 bytes)
    let corruption_positions = [3, 7, 16, 30, 50, 70, 99];

    for &corrupt_pos in &corruption_positions {
        let test_tmp_db = TempDatabaseBuilder::new().build();
        let test_db_path = test_tmp_db.path.clone();
        std::fs::copy(&db_path, &test_db_path)?;

        {
            // corrupt one byte
            use std::fs::OpenOptions;
            use std::io::{Read, Seek, SeekFrom, Write};

            let mut file = OpenOptions::new()
                .read(true)
                .write(true)
                .open(&test_db_path)?;

            file.seek(SeekFrom::Start(corrupt_pos as u64))?;
            let mut original_byte = [0u8; 1];
            file.read_exact(&mut original_byte)?;

            // corrupt it by flipping all bits
            let corrupted_byte = [!original_byte[0]];

            file.seek(SeekFrom::Start(corrupt_pos as u64))?;
            file.write_all(&corrupted_byte)?;
        }

        let uri = format!(
            "file:{}?cipher=aegis256&hexkey=b1bbfda4f589dc9daaf004fe21111e00dc00c98237102f5c7002a5669fc76327",
            test_db_path.to_str().unwrap()
        );

        let (_io, conn) = turso_core::Connection::from_uri(
            &uri,
            DatabaseOpts::new().with_encryption(ENABLE_ENCRYPTION),
        )
        .expect("opening the corrupted DB should not fail at the URI level");

        let result = run_query_on_row(tmp_db, &conn, "SELECT * FROM test", |_row: &Row| {});

        assert!(
            result.is_err(),
            "should return error when accessing encrypted DB with corrupted associated data at position {corrupt_pos}",
        );
    }

    Ok(())
}

// TODO: mvcc does not error here
#[turso_macros::test]
fn test_per_page_encryption(tmp_db: TempDatabase) -> anyhow::Result<()> {
    let _ = env_logger::try_init();
    let db_path = tmp_db.path.clone();
    let opts = tmp_db.db_opts;

    {
        let conn = tmp_db.connect_limbo();
        run_query(
            &tmp_db,
            &conn,
            "PRAGMA hexkey = 'b1bbfda4f589dc9daaf004fe21111e00dc00c98237102f5c7002a5669fc76327';",
        )?;
        run_query(&tmp_db, &conn, "PRAGMA cipher = 'aegis256';")?;
        run_query(
            &tmp_db,
            &conn,
            "CREATE TABLE test (id INTEGER PRIMARY KEY, value TEXT);",
        )?;
        run_query(
            &tmp_db,
            &conn,
            "INSERT INTO test (value) VALUES ('Hello, World!')",
        )?;
        let mut row_count = 0;
        run_query_on_row(&tmp_db, &conn, "SELECT * FROM test", |row: &Row| {
            assert_eq!(row.get::<i64>(0).unwrap(), 1);
            assert_eq!(row.get::<String>(1).unwrap(), "Hello, World!");
            row_count += 1;
        })?;
        assert_eq!(row_count, 1);
        do_flush(&conn, &tmp_db)?;
    }

    {
        //test connecting to the encrypted db using correct URI
        let uri = format!(
            "file:{}?cipher=aegis256&hexkey=b1bbfda4f589dc9daaf004fe21111e00dc00c98237102f5c7002a5669fc76327",
            db_path.to_str().unwrap()
        );
        let (_io, conn) = turso_core::Connection::from_uri(&uri, opts)?;
        let mut row_count = 0;
        run_query_on_row(&tmp_db, &conn, "SELECT * FROM test", |row: &Row| {
            assert_eq!(row.get::<i64>(0).unwrap(), 1);
            assert_eq!(row.get::<String>(1).unwrap(), "Hello, World!");
            row_count += 1;
        })?;
        assert_eq!(row_count, 1);
    }
    {
        //Try to create a table after reopening the encrypted db.
        let uri = format!(
            "file:{}?cipher=aegis256&hexkey=b1bbfda4f589dc9daaf004fe21111e00dc00c98237102f5c7002a5669fc76327",
            db_path.to_str().unwrap()
        );
        let (_io, conn) = turso_core::Connection::from_uri(&uri, opts)?;
        run_query(
            &tmp_db,
            &conn,
            "CREATE TABLE test1 (id INTEGER PRIMARY KEY, value TEXT);",
        )?;
        do_flush(&conn, &tmp_db)?;
    }
    {
        //Try to create a table after reopening the encrypted db.
        let uri = format!(
            "file:{}?cipher=aegis256&hexkey=b1bbfda4f589dc9daaf004fe21111e00dc00c98237102f5c7002a5669fc76327",
            db_path.to_str().unwrap()
        );
        let (_io, conn) = turso_core::Connection::from_uri(&uri, opts)?;
        run_query(
            &tmp_db,
            &conn,
            "INSERT INTO test1 (value) VALUES ('Hello, World!')",
        )?;
        let mut row_count = 0;
        run_query_on_row(&tmp_db, &conn, "SELECT * FROM test", |row: &Row| {
            assert_eq!(row.get::<i64>(0).unwrap(), 1);
            assert_eq!(row.get::<String>(1).unwrap(), "Hello, World!");
            row_count += 1;
        })?;

        assert_eq!(row_count, 1);
        do_flush(&conn, &tmp_db)?;
    }
    {
        // test connecting to encrypted db using wrong key (key ends with 77, correct key ends with 27).
        let uri = format!(
            "file:{}?cipher=aegis256&hexkey=b1bbfda4f589dc9daaf004fe21111e00dc00c98237102f5c7002a5669fc76377",
            db_path.to_str().unwrap()
        );
        let (_io, conn) = turso_core::Connection::from_uri(&uri, opts)?;
        let result = run_query_on_row(&tmp_db, &conn, "SELECT * FROM test", |_row: &Row| {});
        assert!(
            result.is_err(),
            "should return error when accessing encrypted DB with wrong key"
        );
    }
    {
        // test connecting to encrypted db using insufficient encryption parameters in URI.
        let uri = format!("file:{}?cipher=aegis256", db_path.to_str().unwrap());
        let result = turso_core::Connection::from_uri(&uri, opts);
        assert!(
            result.is_err(),
            "should return error when accessing encrypted DB without passing hexkey in URI"
        );
    }
    {
        let uri = format!(
            "file:{}?hexkey=b1bbfda4f589dc9daaf004fe21111e00dc00c98237102f5c7002a5669fc76327",
            db_path.to_str().unwrap()
        );
        let result = turso_core::Connection::from_uri(&uri, opts);
        assert!(
            result.is_err(),
            "should return error when accessing encrypted DB without passing cipher in URI"
        );
    }
    {
        // test connecting to encrypted db without using URI.
        let conn = tmp_db.connect_limbo();
        let result = run_query_on_row(&tmp_db, &conn, "SELECT * FROM test", |_row: &Row| {});
        assert!(
            result.is_err(),
            "should return error when accessing encrypted DB without using URI"
        );
    }

    Ok(())
}

#[turso_macros::test]
fn test_non_4k_page_size_encryption(tmp_db: TempDatabase) -> anyhow::Result<()> {
    run_non_4k_page_size_encryption_test(&tmp_db, false)
}

#[turso_macros::test]
fn test_non_4k_page_size_encryption_mvcc(tmp_db: TempDatabase) -> anyhow::Result<()> {
    run_non_4k_page_size_encryption_test(&tmp_db, true)
}

#[turso_macros::test]
fn test_mvcc_rejects_late_encryption_pragmas(tmp_db: TempDatabase) -> anyhow::Result<()> {
    let conn = tmp_db.connect_limbo();
    run_query(&tmp_db, &conn, "PRAGMA journal_mode = 'mvcc';")?;

    // Insert data before the (rejected) late encryption pragmas.
    run_query(
        &tmp_db,
        &conn,
        "CREATE TABLE pre (id INTEGER PRIMARY KEY, v TEXT);",
    )?;
    run_query(
        &tmp_db,
        &conn,
        "INSERT INTO pre (v) VALUES ('before_late_pragma')",
    )?;

    let key_err = run_query(
        &tmp_db,
        &conn,
        "PRAGMA hexkey = 'b1bbfda4f589dc9daaf004fe21111e00dc00c98237102f5c7002a5669fc76327';",
    )
    .unwrap_err();
    assert!(
        key_err
            .to_string()
            .contains("configure encryption before PRAGMA journal_mode='mvcc'"),
        "unexpected error: {key_err:?}"
    );

    let cipher_err = run_query(&tmp_db, &conn, "PRAGMA cipher = 'aegis256';").unwrap_err();
    assert!(
        cipher_err
            .to_string()
            .contains("configure encryption before PRAGMA journal_mode='mvcc'"),
        "unexpected error: {cipher_err:?}"
    );

    // Data inserted before the rejected pragmas must still be readable.
    let mut pre_count = 0;
    run_query_on_row(&tmp_db, &conn, "SELECT v FROM pre", |row: &Row| {
        assert_eq!(row.get::<String>(0).unwrap(), "before_late_pragma");
        pre_count += 1;
    })?;
    assert_eq!(pre_count, 1);

    run_query(
        &tmp_db,
        &conn,
        "CREATE TABLE test (id INTEGER PRIMARY KEY, value TEXT);",
    )?;
    run_query(
        &tmp_db,
        &conn,
        "INSERT INTO test (value) VALUES ('still plaintext')",
    )?;
    do_flush(&conn, &tmp_db)?;

    let reopened = tmp_db.connect_limbo();
    let mut row_count = 0;
    run_query_on_row(&tmp_db, &reopened, "SELECT value FROM test", |row: &Row| {
        assert_eq!(row.get::<String>(0).unwrap(), "still plaintext");
        row_count += 1;
    })?;
    assert_eq!(row_count, 1);

    Ok(())
}

// TODO: mvcc for some reason does not error on corruption here
#[turso_macros::test]
fn test_corruption_turso_magic_bytes(tmp_db: TempDatabase) -> anyhow::Result<()> {
    let _ = env_logger::try_init();
    let db_path = tmp_db.path.clone();

    let opts = tmp_db.db_opts;

    {
        let conn = tmp_db.connect_limbo();
        run_query(
            &tmp_db,
            &conn,
            "PRAGMA hexkey = 'b1bbfda4f589dc9daaf004fe21111e00dc00c98237102f5c7002a5669fc76327';",
        )?;
        run_query(&tmp_db, &conn, "PRAGMA cipher = 'aegis256';")?;
        run_query(
            &tmp_db,
            &conn,
            "CREATE TABLE test (id INTEGER PRIMARY KEY, value TEXT);",
        )?;
        run_query(
            &tmp_db,
            &conn,
            "INSERT INTO test (value) VALUES ('Test corruption')",
        )?;
        run_query(&tmp_db, &conn, "PRAGMA wal_checkpoint(TRUNCATE);")?;
        do_flush(&conn, &tmp_db)?;
    }

    // corrupt the Turso magic bytes by changing "Turso" to "Vurso" (the db name as it was intended)
    {
        use std::fs::OpenOptions;
        use std::io::{Seek, SeekFrom, Write};

        let mut file = OpenOptions::new().write(true).open(&db_path)?;

        file.seek(SeekFrom::Start(0))?;
        file.write_all(b"V")?;
    }

    // try to connect to the corrupted database - this should return a decryption error
    {
        let uri = format!(
            "file:{}?cipher=aegis256&hexkey=b1bbfda4f589dc9daaf004fe21111e00dc00c98237102f5c7002a5669fc76327",
            db_path.to_str().unwrap()
        );

        let (_io, conn) = turso_core::Connection::from_uri(&uri, opts)?;
        let result = run_query_on_row(&tmp_db, &conn, "SELECT * FROM test", |_row: &Row| {});

        assert!(
            result.is_err(),
            "should return error when accessing encrypted DB with corrupted Turso magic bytes"
        );
        let err_msg = result.unwrap_err().to_string();
        assert!(
            err_msg.contains("Decryption failed"),
            "error should indicate decryption failure, got: {err_msg}"
        );
    }

    Ok(())
}

#[turso_macros::test]
fn test_corruption_associated_data_bytes(tmp_db: TempDatabase) -> anyhow::Result<()> {
    run_corruption_associated_data_bytes_test(&tmp_db, false)
}

#[turso_macros::test]
fn test_corruption_associated_data_bytes_mvcc(tmp_db: TempDatabase) -> anyhow::Result<()> {
    run_corruption_associated_data_bytes_test(&tmp_db, true)
}

#[turso_macros::test(mvcc)]
fn test_turso_header_structure(db: TempDatabase) -> anyhow::Result<()> {
    let _ = env_logger::try_init();

    let verify_header =
        |db_path: &str, expected_cipher_id: u8, description: &str| -> anyhow::Result<()> {
            use std::fs::File;
            use std::io::{Read, Seek, SeekFrom};

            let mut file = File::open(db_path)?;
            let mut header = [0u8; 16];
            file.seek(SeekFrom::Start(0))?;
            file.read_exact(&mut header)?;

            assert_eq!(
                &header[0..5],
                b"Turso",
                "Magic bytes should be 'Turso' for {description}"
            );
            assert_eq!(header[5], 0x00, "Version should be 0x00 for {description}");
            assert_eq!(
                header[6], expected_cipher_id,
                "Cipher ID should be {expected_cipher_id} for {description}"
            );

            // the unused bytes should be zeroed
            for (i, &byte) in header[7..16].iter().enumerate() {
                assert_eq!(
                    byte,
                    0,
                    "Unused byte at position {} should be 0 for {}",
                    i + 7,
                    description
                );
            }

            println!("Verified {} header: cipher ID = {}", description, header[6]);
            Ok(())
        };

    let test_cases = [
        (
            "aes128gcm",
            1,
            "AES-128-GCM",
            "b1bbfda4f589dc9daaf004fe21111e00",
        ),
        (
            "aes256gcm",
            2,
            "AES-256-GCM",
            "b1bbfda4f589dc9daaf004fe21111e00dc00c98237102f5c7002a5669fc76327",
        ),
        (
            "aegis256",
            3,
            "AEGIS-256",
            "b1bbfda4f589dc9daaf004fe21111e00dc00c98237102f5c7002a5669fc76327",
        ),
        (
            "aegis256x2",
            4,
            "AEGIS-256X2",
            "b1bbfda4f589dc9daaf004fe21111e00dc00c98237102f5c7002a5669fc76327",
        ),
        (
            "aegis128l",
            6,
            "AEGIS-128L",
            "b1bbfda4f589dc9daaf004fe21111e00",
        ),
        (
            "aegis128x2",
            7,
            "AEGIS-128X2",
            "b1bbfda4f589dc9daaf004fe21111e00",
        ),
        (
            "aegis128x4",
            8,
            "AEGIS-128X4",
            "b1bbfda4f589dc9daaf004fe21111e00",
        ),
    ];
    let opts = db.db_opts;
    let flags = db.db_flags;

    for (cipher_name, expected_id, description, hexkey) in test_cases {
        let tmp_db = TempDatabase::builder()
            .with_opts(opts)
            .with_flags(flags)
            .build();
        let db_path = tmp_db.path.clone();

        {
            let conn = tmp_db.connect_limbo();
            run_query(&tmp_db, &conn, &format!("PRAGMA hexkey = '{hexkey}';"))?;
            run_query(&tmp_db, &conn, &format!("PRAGMA cipher = '{cipher_name}';"))?;
            run_query(
                &tmp_db,
                &conn,
                "CREATE TABLE test (id INTEGER PRIMARY KEY, value TEXT);",
            )?;
            do_flush(&conn, &tmp_db)?;
        }

        verify_header(db_path.to_str().unwrap(), expected_id, description)?;
    }
    Ok(())
}

/// this is a smoll test for database registry caching encryption keys
///
/// Previously, the DATABASE_MANAGER cached Database instances with keys. Which led to:
/// 1. Open database with correct key -> Database cached with correct encryption_key
/// 2. Open database with WRONG key or no key -> Cached Database returned
/// 3. Decryption succeeds because cached Database has correct key
///
/// This test ensures that opening with wrong encryption key (or no key) fails even after
/// the database has been opened with the correct key (which populates the cache).
#[turso_macros::test(mvcc)]
fn test_encryption_key_validation_with_cached_database(_db: TempDatabase) -> anyhow::Result<()> {
    let _ = env_logger::try_init();

    let temp_dir = tempfile::tempdir()?;
    let db_path = temp_dir
        .path()
        .join(format!("test-enc-cache-{}.db", rng().next_u32()));
    let db_path_str = db_path.to_str().unwrap();

    let correct_key = "b1bbfda4f589dc9daaf004fe21111e00dc00c98237102f5c7002a5669fc76327";
    let wrong_key = "aaaaaaa4f589dc9daaf004fe21111e00dc00c98237102f5c7002a5669fc76327";

    let io = Arc::new(PlatformIO::new()?);
    let opts = DatabaseOpts::new().with_encryption(ENABLE_ENCRYPTION);

    let correct_encryption_opts = Some(EncryptionOpts {
        cipher: "aegis256".to_string(),
        hexkey: correct_key.to_string(),
    });

    let main_db = Database::open_file_with_flags(
        io.clone(),
        db_path_str,
        OpenFlags::Create,
        opts,
        correct_encryption_opts.clone(),
    )?;

    // step 1: Create encrypted database with correct key
    {
        let correct_encryption_key =
            turso_core::EncryptionKey::from_hex_string(correct_key).unwrap();

        let conn = main_db.connect()?;
        conn.set_encryption_cipher(turso_core::CipherMode::Aegis256)?;
        conn.set_encryption_key(correct_encryption_key)?;

        conn.execute("CREATE TABLE secret_data (id INTEGER PRIMARY KEY, value TEXT)")?;
        conn.execute("INSERT INTO secret_data (value) VALUES ('top secret')")?;
        conn.query("PRAGMA wal_checkpoint(TRUNCATE)")?;
        for completion in conn.cacheflush()? {
            io.wait_for_completion(completion)?;
        }
    }

    // Step 2: re-open with correct key (this uses the DATABASE_MANAGER cache)
    {
        let correct_encryption_key =
            turso_core::EncryptionKey::from_hex_string(correct_key).unwrap();

        let db = Database::open_file_with_flags(
            io.clone(),
            db_path_str,
            OpenFlags::default(),
            opts,
            correct_encryption_opts.clone(),
        )?;

        let conn = db.connect()?;
        conn.set_encryption_cipher(turso_core::CipherMode::Aegis256)?;
        conn.set_encryption_key(correct_encryption_key)?;

        let rows = conn.query("SELECT * FROM secret_data")?;
        let mut row_count = 0;
        if let Some(mut rows) = rows {
            loop {
                match rows.step()? {
                    turso_core::StepResult::Row => {
                        let row = rows.row().unwrap();
                        assert_eq!(row.get::<String>(1).unwrap(), "top secret");
                        row_count += 1;
                    }
                    turso_core::StepResult::Done => break,
                    turso_core::StepResult::Interrupt => break,
                    turso_core::StepResult::Busy | turso_core::StepResult::IO => continue,
                }
            }
        }
        assert_eq!(row_count, 1, "Should read data with correct key");
    }

    // Step 3: Opening with wrong key succeeds, but reading data fails with decryption error
    {
        let wrong_encryption_key = turso_core::EncryptionKey::from_hex_string(wrong_key).unwrap();

        let db = Database::open_file_with_flags(
            io.clone(),
            db_path_str,
            OpenFlags::default(),
            opts,
            Some(EncryptionOpts {
                cipher: "aegis256".to_string(),
                hexkey: wrong_key.to_string(),
            }),
        )?;

        // opening succeeds - the key is not validated at open time
        let conn = db.connect()?;
        conn.set_encryption_cipher(turso_core::CipherMode::Aegis256)?;
        conn.set_encryption_key(wrong_encryption_key)?;

        // Reading data should fail with a decryption error
        let read_failed = match conn.query("SELECT * FROM secret_data") {
            Err(_) => true,
            Ok(Some(mut rows)) => loop {
                match rows.step() {
                    Err(_) => break true,                            // Error - read failed
                    Ok(turso_core::StepResult::Done) => break false, // Completed without error
                    Ok(turso_core::StepResult::Interrupt) => break false,
                    Ok(turso_core::StepResult::Row) => break false, // Got data - unexpected!!
                    Ok(turso_core::StepResult::Busy) | Ok(turso_core::StepResult::IO) => continue,
                }
            },
            Ok(None) => false,
        };
        assert!(
            read_failed,
            "Reading data with wrong key should fail with decryption error"
        );
    }

    // Step 4: Opening without encryption options should fail immediately
    {
        let result = Database::open_file_with_flags(
            io.clone(),
            db_path_str,
            OpenFlags::default(),
            opts,
            None,
        );

        assert!(
            result.is_err(),
            "Opening encrypted database without encryption options should fail"
        );
        let err = result.unwrap_err();
        assert!(
            err.to_string()
                .contains("Database is encrypted but no encryption options provided"),
            "Error message should indicate missing encryption options"
        );
    }

    // Step 5: verify correct key still works after wrong key attempt
    {
        let correct_encryption_key =
            turso_core::EncryptionKey::from_hex_string(correct_key).unwrap();

        let db = Database::open_file_with_flags(
            io.clone(),
            db_path_str,
            OpenFlags::default(),
            opts,
            correct_encryption_opts.clone(),
        )?;

        let conn = db.connect()?;
        conn.set_encryption_cipher(turso_core::CipherMode::Aegis256)?;
        conn.set_encryption_key(correct_encryption_key)?;

        let rows = conn.query("SELECT * FROM secret_data")?;
        let mut row_count = 0;
        if let Some(mut rows) = rows {
            loop {
                match rows.step()? {
                    turso_core::StepResult::Row => {
                        let row = rows.row().unwrap();
                        assert_eq!(row.get::<String>(1).unwrap(), "top secret");
                        row_count += 1;
                    }
                    turso_core::StepResult::Done => break,
                    turso_core::StepResult::Interrupt => break,
                    turso_core::StepResult::Busy | turso_core::StepResult::IO => continue,
                }
            }
        }
        assert_eq!(
            row_count, 1,
            "Should still read data with correct key after wrong key attempt"
        );
    }
    Ok(())
}

// Two different keys/ciphers for the two attached databases
const KEY_A: &str = "b1bbfda4f589dc9daaf004fe21111e00dc00c98237102f5c7002a5669fc76327";
const CIPHER_A: &str = "aegis256";
const KEY_B: &str = "a2ccfeb5f690ed0ebf01a843f22222f11ed11d9348213f6d8113b677ae84ad38";
const CIPHER_B: &str = "aes256gcm";

/// Helper: create an encrypted database file with the given cipher, hexkey, table name, and value.
/// Returns the file path.
fn create_encrypted_db(
    cipher: &str,
    hexkey: &str,
    table_name: &str,
    value: &str,
) -> anyhow::Result<std::path::PathBuf> {
    let temp_dir = tempfile::tempdir()?;
    let db_path = temp_dir
        .path()
        .join(format!("enc-{}-{}.db", table_name, rng().next_u32()));
    let db_path_str = db_path.to_str().unwrap();

    let io: Arc<dyn IO + Send> = Arc::new(PlatformIO::new()?);
    let opts = DatabaseOpts::new().with_encryption(true);
    let encryption_opts = Some(EncryptionOpts {
        cipher: cipher.to_string(),
        hexkey: hexkey.to_string(),
    });

    let db = Database::open_file_with_flags(
        io.clone(),
        db_path_str,
        OpenFlags::Create,
        opts,
        encryption_opts,
    )?;

    let conn = db.connect()?;
    let cipher_mode = CipherMode::try_from(cipher)?;
    let key = EncryptionKey::from_hex_string(hexkey)?;
    conn.set_encryption_cipher(cipher_mode)?;
    conn.set_encryption_key(key)?;

    conn.execute(format!(
        "CREATE TABLE {table_name} (id INTEGER PRIMARY KEY, value TEXT)"
    ))?;
    conn.execute(format!(
        "INSERT INTO {table_name} (value) VALUES ('{value}')"
    ))?;
    conn.query("PRAGMA wal_checkpoint(TRUNCATE)")?;
    for c in conn.cacheflush()? {
        io.wait_for_completion(c)?;
    }

    // Keep the temp dir alive by leaking it (the test process will clean up)
    std::mem::forget(temp_dir);
    Ok(db_path)
}

/// Helper: open a plain (unencrypted) main database with attach + encryption enabled.
fn open_main_db() -> anyhow::Result<(TempDatabase, Arc<turso_core::Connection>)> {
    let tmp_db = TempDatabase::builder()
        .with_opts(DatabaseOpts::new().with_encryption(true).with_attach(true))
        .build();
    let conn = tmp_db.connect_limbo();
    Ok((tmp_db, conn))
}

#[turso_macros::test(mvcc)]
fn test_attach_encrypted_database(_tmp_db: TempDatabase) -> anyhow::Result<()> {
    let _ = env_logger::try_init();

    // Create two encrypted databases with different keys and ciphers
    let path_a = create_encrypted_db(CIPHER_A, KEY_A, "secret_a", "data from A")?;
    let path_b = create_encrypted_db(CIPHER_B, KEY_B, "secret_b", "data from B")?;

    // --- Test 1: Happy path — attach both with correct keys ---
    {
        let (main_db, conn) = open_main_db()?;
        let attach_a = format!(
            "ATTACH 'file:{}?cipher={}&hexkey={}' AS aux_a",
            path_a.to_str().unwrap(),
            CIPHER_A,
            KEY_A
        );
        let attach_b = format!(
            "ATTACH 'file:{}?cipher={}&hexkey={}' AS aux_b",
            path_b.to_str().unwrap(),
            CIPHER_B,
            KEY_B
        );
        run_query(&main_db, &conn, &attach_a)?;
        run_query(&main_db, &conn, &attach_b)?;

        let mut row_count = 0;
        run_query_on_row(
            &main_db,
            &conn,
            "SELECT value FROM aux_a.secret_a",
            |row: &Row| {
                assert_eq!(row.get::<String>(0).unwrap(), "data from A");
                row_count += 1;
            },
        )?;
        assert_eq!(row_count, 1, "Should read one row from aux_a");

        let mut row_count = 0;
        run_query_on_row(
            &main_db,
            &conn,
            "SELECT value FROM aux_b.secret_b",
            |row: &Row| {
                assert_eq!(row.get::<String>(0).unwrap(), "data from B");
                row_count += 1;
            },
        )?;
        assert_eq!(row_count, 1, "Should read one row from aux_b");
    }

    // --- Test 2: Wrong key for db_a — use key_b instead ---
    {
        let (main_db, conn) = open_main_db()?;
        let attach_wrong = format!(
            "ATTACH 'file:{}?cipher={}&hexkey={}' AS aux_a",
            path_a.to_str().unwrap(),
            CIPHER_A,
            KEY_B // wrong key!
        );
        // Attaching may succeed (key is not validated at open time), but reading must fail
        let attach_result = run_query(&main_db, &conn, &attach_wrong);
        if attach_result.is_ok() {
            let read_result = run_query_on_row(
                &main_db,
                &conn,
                "SELECT value FROM aux_a.secret_a",
                |_: &Row| {},
            );
            assert!(
                read_result.is_err(),
                "Reading with wrong key should fail with decryption error"
            );
        }
        // If attach itself failed, that's also acceptable
    }

    // --- Test 3: Swapped keys — db_a gets key_b, db_b gets key_a ---
    {
        let (main_db, conn) = open_main_db()?;
        let attach_a_wrong = format!(
            "ATTACH 'file:{}?cipher={}&hexkey={}' AS aux_a",
            path_a.to_str().unwrap(),
            CIPHER_A,
            KEY_B
        );
        let attach_b_wrong = format!(
            "ATTACH 'file:{}?cipher={}&hexkey={}' AS aux_b",
            path_b.to_str().unwrap(),
            CIPHER_B,
            KEY_A
        );

        // Attach may succeed at open time; reading should fail for both
        let _ = run_query(&main_db, &conn, &attach_a_wrong);
        let _ = run_query(&main_db, &conn, &attach_b_wrong);

        let read_a = run_query_on_row(
            &main_db,
            &conn,
            "SELECT value FROM aux_a.secret_a",
            |_: &Row| {},
        );
        let read_b = run_query_on_row(
            &main_db,
            &conn,
            "SELECT value FROM aux_b.secret_b",
            |_: &Row| {},
        );
        assert!(
            read_a.is_err() || read_b.is_err(),
            "At least one read with swapped keys must fail"
        );
    }

    // --- Test 4: Missing hexkey in URI ---
    {
        let (main_db, conn) = open_main_db()?;
        let attach_no_hexkey = format!(
            "ATTACH 'file:{}?cipher={}' AS aux_a",
            path_a.to_str().unwrap(),
            CIPHER_A
        );
        let result = run_query(&main_db, &conn, &attach_no_hexkey);
        assert!(
            result.is_err(),
            "ATTACH with cipher but no hexkey should fail"
        );
    }

    // --- Test 5: Missing cipher in URI ---
    {
        let (main_db, conn) = open_main_db()?;
        let attach_no_cipher = format!(
            "ATTACH 'file:{}?hexkey={}' AS aux_a",
            path_a.to_str().unwrap(),
            KEY_A
        );
        let result = run_query(&main_db, &conn, &attach_no_cipher);
        assert!(
            result.is_err(),
            "ATTACH with hexkey but no cipher should fail"
        );
    }

    // --- Test 6: No encryption params at all ---
    {
        let (main_db, conn) = open_main_db()?;
        let attach_no_enc = format!("ATTACH '{}' AS aux_a", path_a.to_str().unwrap());
        let result = run_query(&main_db, &conn, &attach_no_enc);
        // Opening an encrypted DB without key should fail
        assert!(
            result.is_err(),
            "ATTACH encrypted DB without key should fail"
        );
    }

    // --- Test 7: Correct key after wrong key attempt ---
    {
        let (main_db, conn) = open_main_db()?;

        // First, try wrong key
        let attach_wrong = format!(
            "ATTACH 'file:{}?cipher={}&hexkey={}' AS aux_a",
            path_a.to_str().unwrap(),
            CIPHER_A,
            KEY_B
        );
        let _ = run_query(&main_db, &conn, &attach_wrong);
        // Detach (ignore error if attach failed)
        let _ = run_query(&main_db, &conn, "DETACH aux_a");

        // Now attach with correct key
        let attach_correct = format!(
            "ATTACH 'file:{}?cipher={}&hexkey={}' AS aux_a",
            path_a.to_str().unwrap(),
            CIPHER_A,
            KEY_A
        );
        run_query(&main_db, &conn, &attach_correct)?;

        let mut row_count = 0;
        run_query_on_row(
            &main_db,
            &conn,
            "SELECT value FROM aux_a.secret_a",
            |row: &Row| {
                assert_eq!(row.get::<String>(0).unwrap(), "data from A");
                row_count += 1;
            },
        )?;
        assert_eq!(
            row_count, 1,
            "Should read data after re-attaching with correct key"
        );
    }

    // --- Test 8: Write to encrypted attached DB ---
    {
        let (main_db, conn) = open_main_db()?;
        let attach_a = format!(
            "ATTACH 'file:{}?cipher={}&hexkey={}' AS aux_a",
            path_a.to_str().unwrap(),
            CIPHER_A,
            KEY_A
        );
        run_query(&main_db, &conn, &attach_a)?;

        run_query(
            &main_db,
            &conn,
            "INSERT INTO aux_a.secret_a (value) VALUES ('new data')",
        )?;
        do_flush(&conn, &main_db)?;

        let mut values = Vec::new();
        run_query_on_row(
            &main_db,
            &conn,
            "SELECT value FROM aux_a.secret_a ORDER BY id",
            |row: &Row| {
                values.push(row.get::<String>(0).unwrap());
            },
        )?;
        assert_eq!(values.len(), 2, "Should have original + new row");
        assert_eq!(values[0], "data from A");
        assert_eq!(values[1], "new data");
    }

    // --- Test 9: Different ciphers on same connection ---
    {
        let (main_db, conn) = open_main_db()?;
        let attach_a = format!(
            "ATTACH 'file:{}?cipher={}&hexkey={}' AS aux_a",
            path_a.to_str().unwrap(),
            CIPHER_A,
            KEY_A
        );
        let attach_b = format!(
            "ATTACH 'file:{}?cipher={}&hexkey={}' AS aux_b",
            path_b.to_str().unwrap(),
            CIPHER_B,
            KEY_B
        );
        run_query(&main_db, &conn, &attach_a)?;
        run_query(&main_db, &conn, &attach_b)?;

        // Verify both are readable with their respective keys/ciphers
        let mut val_a = String::new();
        let mut val_b = String::new();
        run_query_on_row(
            &main_db,
            &conn,
            "SELECT value FROM aux_a.secret_a WHERE id = 1",
            |row: &Row| {
                val_a = row.get::<String>(0).unwrap();
            },
        )?;
        run_query_on_row(
            &main_db,
            &conn,
            "SELECT value FROM aux_b.secret_b WHERE id = 1",
            |row: &Row| {
                val_b = row.get::<String>(0).unwrap();
            },
        )?;
        assert_eq!(val_a, "data from A");
        assert_eq!(val_b, "data from B");
    }

    Ok(())
}

/// Test that VACUUM INTO on an encrypted database results in a clean, unencrypted
/// database that can be read without any encryption keys.
#[turso_macros::test]
fn test_vacuum_into_unencrypts(tmp_db: TempDatabase) -> anyhow::Result<()> {
    use tempfile::TempDir;

    let _ = env_logger::try_init();
    let hexkey = "b1bbfda4f589dc9daaf004fe21111e00dc00c98237102f5c7002a5669fc76327";
    let cipher = "aegis256";

    // 1. Create an encrypted source database and insert data
    {
        let conn = tmp_db.connect_limbo();
        conn.execute(format!("PRAGMA hexkey = '{hexkey}'"))?;
        conn.execute(format!("PRAGMA cipher = '{cipher}'"))?;

        conn.execute("CREATE TABLE secret_data (id INTEGER PRIMARY KEY, content TEXT)")?;
        conn.execute("INSERT INTO secret_data (content) VALUES ('this was encrypted')")?;
        do_flush(&conn, &tmp_db)?;
    }

    let dest_dir = TempDir::new()?;
    let dest_path = dest_dir.path().join("exported.db");
    let dest_path_str = dest_path.to_str().unwrap();

    // 2. Demonstrate that the encrypted source CANNOT be read or vacuumed without keys
    {
        let unauthorized_db = TempDatabase::new_with_existent(&tmp_db.path);
        let unauthorized_conn = unauthorized_db.connect_limbo();

        // Reading should fail
        let result = unauthorized_conn.execute("SELECT * FROM secret_data");
        assert!(
            result.is_err(),
            "Encrypted source should not be readable as plaintext"
        );
        let err_msg = result.err().unwrap().to_string();
        assert!(
            err_msg.contains("Corrupt database"),
            "Error message should indicate that the encrypted database cannot be read: '{err_msg}'"
        );

        // VACUUM INTO should also fail because it cannot read the source schema/data
        let fail_path = dest_dir.path().join("should_fail.db");
        let fail_path_str = fail_path.to_str().unwrap();
        let result = unauthorized_conn.execute(format!("VACUUM INTO '{fail_path_str}'"));
        assert!(
            result.is_err(),
            "VACUUM INTO should fail on encrypted database when no keys are provided"
        );
        let err_msg = result.err().unwrap().to_string();
        assert!(
            err_msg.contains("Corrupt database"),
            "Error message should indicate that the encrypted database cannot be read: '{err_msg}'"
        );
    }

    // 3. Execute VACUUM INTO using an authorized connection
    {
        let conn = tmp_db.connect_limbo();
        conn.execute(format!("PRAGMA hexkey = '{hexkey}'"))?;
        conn.execute(format!("PRAGMA cipher = '{cipher}'"))?;
        conn.execute(format!("VACUUM INTO '{dest_path_str}'"))?;
    }

    // 4. Prove the destination IS readable without keys (it was unencrypted)
    {
        let dest_db = TempDatabase::new_with_existent(&dest_path);
        let dest_conn = dest_db.connect_limbo();

        let rows: Vec<(i64, String)> = dest_conn.exec_rows("SELECT id, content FROM secret_data");
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].1, "this was encrypted");
    }

    Ok(())
}

#[test]
fn test_encrypted_db_then_enable_mvcc_large_payload_chunked() -> anyhow::Result<()> {
    let _ = env_logger::try_init();

    let temp_dir = tempfile::tempdir()?;
    let db_path = temp_dir.path().join("test.db");
    let db_path_str = db_path.to_str().unwrap();

    let hex_key = "b1bbfda4f589dc9daaf004fe21111e00dc00c98237102f5c7002a5669fc76327";
    let large_value = "x".repeat(100_000);
    let io = Arc::new(PlatformIO::new()?);
    let opts = DatabaseOpts::new().with_encryption(true);
    let enc_opts = Some(EncryptionOpts {
        cipher: "aes256gcm".to_string(),
        hexkey: hex_key.to_string(),
    });

    {
        let db = Database::open_file_with_flags(
            io.clone(),
            db_path_str,
            OpenFlags::Create,
            opts,
            enc_opts.clone(),
        )?;
        let key = EncryptionKey::from_hex_string(hex_key)?;
        let conn = db.connect_with_encryption(Some(key))?;
        conn.execute("PRAGMA journal_mode = 'mvcc'")?;
        conn.execute("CREATE TABLE test (id INTEGER PRIMARY KEY, value TEXT)")?;
        conn.execute(format!("INSERT INTO test (value) VALUES ('{large_value}')"))?;
        for c in conn.cacheflush()? {
            io.wait_for_completion(c)?;
        }
    }

    let log_path = db_path.with_extension("db-log");
    let log_bytes = std::fs::read(&log_path)?;
    assert!(
        log_bytes.len() > 64 * 1024,
        "multi-chunk encrypted MVCC log should exceed 64 KiB"
    );

    {
        let db = Database::open_file_with_flags(
            io.clone(),
            db_path_str,
            OpenFlags::default(),
            opts,
            enc_opts,
        )?;
        let key = EncryptionKey::from_hex_string(hex_key)?;
        let conn = db.connect_with_encryption(Some(key))?;

        let rows: Vec<(i64, i64, String, String)> = conn.exec_rows(
            "SELECT id, length(value), substr(value, 1, 16), substr(value, length(value) - 15, 16) FROM test",
        );
        assert_eq!(
            rows.len(),
            1,
            "Should recover multi-chunk MVCC payload after restart"
        );
        assert_eq!(rows[0].0, 1);
        assert_eq!(rows[0].1, large_value.len() as i64);
        assert_eq!(rows[0].2, "xxxxxxxxxxxxxxxx");
        assert_eq!(rows[0].3, "xxxxxxxxxxxxxxxx");
    }

    Ok(())
}

/// Create an encrypted database with existing data, then switch to MVCC mode.
/// Verifies that pre-existing rows survive the journal mode switch and that
/// new MVCC writes produce an encrypted log.
#[test]
fn test_encrypted_db_with_data_then_enable_mvcc() -> anyhow::Result<()> {
    let _ = env_logger::try_init();

    let temp_dir = tempfile::tempdir()?;
    let db_path = temp_dir.path().join("test.db");
    let db_path_str = db_path.to_str().unwrap();

    let hex_key = "b1bbfda4f589dc9daaf004fe21111e00dc00c98237102f5c7002a5669fc76327";
    let io = Arc::new(PlatformIO::new()?);
    let opts = DatabaseOpts::new().with_encryption(true);
    let enc_opts = Some(EncryptionOpts {
        cipher: "aes256gcm".to_string(),
        hexkey: hex_key.to_string(),
    });

    // Phase 1: Create encrypted DB in WAL mode, insert data
    {
        let db = Database::open_file_with_flags(
            io.clone(),
            db_path_str,
            OpenFlags::Create,
            opts,
            enc_opts.clone(),
        )?;
        let key = EncryptionKey::from_hex_string(hex_key)?;
        let conn = db.connect_with_encryption(Some(key))?;
        conn.execute("CREATE TABLE test (id INTEGER PRIMARY KEY, value TEXT)")?;
        conn.execute("INSERT INTO test (value) VALUES ('before mvcc')")?;
        for c in conn.cacheflush()? {
            io.wait_for_completion(c)?;
        }
    }

    // Phase 2: Reopen, switch to MVCC, insert more data
    {
        let db = Database::open_file_with_flags(
            io.clone(),
            db_path_str,
            OpenFlags::default(),
            opts,
            enc_opts.clone(),
        )?;
        let key = EncryptionKey::from_hex_string(hex_key)?;
        let conn = db.connect_with_encryption(Some(key))?;

        // Pre-existing row should be readable
        let rows: Vec<(String,)> = conn.exec_rows("SELECT value FROM test");
        assert_eq!(rows.len(), 1, "pre-existing row must be readable");
        assert_eq!(rows[0].0, "before mvcc");

        // Switch to MVCC
        conn.execute("PRAGMA journal_mode = 'mvcc'")?;
        conn.execute("INSERT INTO test (value) VALUES ('after mvcc')")?;
        for c in conn.cacheflush()? {
            io.wait_for_completion(c)?;
        }
    }

    // Verify the MVCC log is encrypted
    {
        let log_path = db_path.with_extension("db-log");
        let log_bytes = std::fs::read(&log_path)?;
        assert!(
            log_bytes.len() > 56,
            "MVCC log should contain data beyond the header"
        );
        for plaintext in [b"before mvcc" as &[u8], b"after mvcc"] {
            assert!(
                !log_bytes.windows(plaintext.len()).any(|w| w == plaintext),
                "MVCC log must not contain plaintext '{}' when encryption is enabled",
                std::str::from_utf8(plaintext).unwrap()
            );
        }
    }

    // Phase 3: Reopen — MVCC recovery should replay, both rows visible
    {
        let db = Database::open_file_with_flags(
            io.clone(),
            db_path_str,
            OpenFlags::default(),
            opts,
            enc_opts.clone(),
        )?;
        let key = EncryptionKey::from_hex_string(hex_key)?;
        let conn = db.connect_with_encryption(Some(key))?;

        let rows: Vec<(String,)> = conn.exec_rows("SELECT value FROM test ORDER BY id");
        assert_eq!(
            rows.iter().map(|r| r.0.as_str()).collect::<Vec<_>>(),
            vec!["before mvcc", "after mvcc"]
        );
    }

    Ok(())
}

#[turso_macros::test]
fn test_non_4k_page_size_encryption_enable_mvcc_after_encryption(
    tmp_db: TempDatabase,
) -> anyhow::Result<()> {
    let _ = env_logger::try_init();
    let conn = tmp_db.connect_limbo();

    [
        "PRAGMA page_size = 8192;",
        "PRAGMA hexkey = 'b1bbfda4f589dc9daaf004fe21111e00dc00c98237102f5c7002a5669fc76327';",
        "PRAGMA cipher = 'aegis256';",
        "PRAGMA journal_mode = 'mvcc';",
        "CREATE TABLE test (id INTEGER PRIMARY KEY, value TEXT);",
        "INSERT INTO test (value) VALUES ('Hello, World!')",
    ]
    .iter()
    .try_for_each(|query| run_query(&tmp_db, &conn, query))?;
    do_flush(&conn, &tmp_db)
}

// Regression coverage for https://github.com/tursodatabase/turso/issues/7375.
// PRAGMA cipher/hexkey before PRAGMA page_size used to leave EncryptionContext
// pinned to the default 4096-byte page size; the first write then panicked.

const ISSUE_7375_KEY_256: &str = "000102030405060708090a0b0c0d0e0f101112131415161718191a1b1c1d1e1f";
const ISSUE_7375_KEY_128: &str = "000102030405060708090a0b0c0d0e0f";

fn assert_encrypted_page_size_after_key_and_cipher(
    page_size: i64,
    cipher: &str,
    hexkey: &str,
) -> anyhow::Result<()> {
    let tmp_db = TempDatabaseBuilder::new()
        .with_opts(DatabaseOpts::new().with_encryption(true))
        .build();

    let conn = tmp_db.connect_limbo();
    conn.execute(format!("PRAGMA cipher = '{cipher}'"))?;
    conn.execute(format!("PRAGMA hexkey = '{hexkey}'"))?;
    conn.execute(format!("PRAGMA page_size = {page_size}"))?;
    conn.execute("CREATE TABLE t(a INTEGER PRIMARY KEY, b BLOB)")?;
    conn.execute("INSERT INTO t VALUES(1, randomblob(300))")?;

    let rows: Vec<(i64,)> = conn.exec_rows("SELECT count(*) FROM t");
    assert_eq!(rows[0].0, 1, "row count mismatch for page_size={page_size}");
    let ps: Vec<(i64,)> = conn.exec_rows("PRAGMA page_size");
    assert_eq!(ps[0].0, page_size, "page_size readback mismatch");

    do_flush(&conn, &tmp_db)?;

    let uri = format!(
        "file:{}?cipher={cipher}&hexkey={hexkey}",
        tmp_db.path.to_str().unwrap()
    );
    let (_io, reopened) =
        turso_core::Connection::from_uri(&uri, DatabaseOpts::new().with_encryption(true))?;
    let rows: Vec<(i64,)> = reopened.exec_rows("SELECT count(*) FROM t");
    assert_eq!(
        rows[0].0, 1,
        "row count after reopen mismatch for page_size={page_size}"
    );

    Ok(())
}

#[turso_macros::test]
fn test_encrypted_page_size_after_key_and_cipher(_tmp_db: TempDatabase) -> anyhow::Result<()> {
    let _ = env_logger::try_init();
    for page_size in [512, 4096, 65536] {
        assert_encrypted_page_size_after_key_and_cipher(
            page_size,
            "aegis256x4",
            ISSUE_7375_KEY_256,
        )?;
    }
    // Different key size / metadata path.
    assert_encrypted_page_size_after_key_and_cipher(512, "aes128gcm", ISSUE_7375_KEY_128)?;
    Ok(())
}

#[turso_macros::test]
fn test_uri_encryption_then_page_size(_tmp_db: TempDatabase) -> anyhow::Result<()> {
    let _ = env_logger::try_init();
    let tmp_db = TempDatabaseBuilder::new()
        .with_opts(DatabaseOpts::new().with_encryption(true))
        .build();

    let uri = format!(
        "file:{}?cipher=aegis256x4&hexkey={ISSUE_7375_KEY_256}",
        tmp_db.path.to_str().unwrap()
    );

    {
        let (io, conn) =
            turso_core::Connection::from_uri(&uri, DatabaseOpts::new().with_encryption(true))?;
        conn.execute("PRAGMA page_size = 512")?;
        conn.execute("CREATE TABLE t(a INTEGER PRIMARY KEY, b BLOB)")?;
        conn.execute("INSERT INTO t VALUES(1, randomblob(300))")?;

        let rows: Vec<(i64,)> = conn.exec_rows("SELECT count(*) FROM t");
        assert_eq!(rows[0].0, 1);
        let ps: Vec<(i64,)> = conn.exec_rows("PRAGMA page_size");
        assert_eq!(ps[0].0, 512);

        for c in conn.cacheflush()? {
            io.wait_for_completion(c)?;
        }
    }

    let (_io, reopened) =
        turso_core::Connection::from_uri(&uri, DatabaseOpts::new().with_encryption(true))?;
    let rows: Vec<(i64,)> = reopened.exec_rows("SELECT count(*) FROM t");
    assert_eq!(rows[0].0, 1);
    let ps: Vec<(i64,)> = reopened.exec_rows("PRAGMA page_size");
    assert_eq!(ps[0].0, 512);

    Ok(())
}

#[turso_macros::test]
fn test_fresh_attach_encrypted_non_4k_page_size(_tmp_db: TempDatabase) -> anyhow::Result<()> {
    let _ = env_logger::try_init();
    let aux_dir = tempfile::tempdir()?;
    let aux_path = aux_dir.path().join("aux.db");
    let aux_uri = format!(
        "file:{}?cipher=aegis256x4&hexkey={ISSUE_7375_KEY_256}",
        aux_path.to_str().unwrap()
    );

    {
        let main_db = TempDatabaseBuilder::new()
            .with_opts(DatabaseOpts::new().with_encryption(true).with_attach(true))
            .build();
        let conn = main_db.connect_limbo();
        conn.execute("PRAGMA page_size = 512")?;
        conn.execute(format!("ATTACH '{aux_uri}' AS aux"))?;
        conn.execute("CREATE TABLE aux.t(a INTEGER PRIMARY KEY, b BLOB)")?;
        conn.execute("INSERT INTO aux.t VALUES(1, randomblob(300))")?;

        let rows: Vec<(i64,)> = conn.exec_rows("SELECT count(*) FROM aux.t");
        assert_eq!(rows[0].0, 1);
        // Layout inheritance: the attached pager must adopt the main
        // connection's page size, not the fresh-DB 4096 default.
        let aux_ps: Vec<(i64,)> = conn.exec_rows("PRAGMA aux.page_size");
        assert_eq!(aux_ps[0].0, 512);

        // Force aux WAL onto its main file so the standalone reopen sees the row.
        conn.execute("PRAGMA aux.wal_checkpoint(TRUNCATE)")?;
        do_flush(&conn, &main_db)?;
    }

    let (_io, aux_conn) =
        turso_core::Connection::from_uri(&aux_uri, DatabaseOpts::new().with_encryption(true))?;
    let rows: Vec<(i64,)> = aux_conn.exec_rows("SELECT count(*) FROM t");
    assert_eq!(rows[0].0, 1);
    let ps: Vec<(i64,)> = aux_conn.exec_rows("PRAGMA page_size");
    assert_eq!(ps[0].0, 512);

    Ok(())
}

#[turso_macros::test]
fn test_inplace_vacuum_non_4k_encryption(_tmp_db: TempDatabase) -> anyhow::Result<()> {
    let _ = env_logger::try_init();
    let tmp_db = TempDatabaseBuilder::new()
        .with_opts(DatabaseOpts::new().with_encryption(true))
        .build();

    let conn = tmp_db.connect_limbo();
    // Safe creation order so setup itself does not trigger the issue path.
    conn.execute("PRAGMA page_size = 512")?;
    conn.execute("PRAGMA cipher = 'aegis256x4'")?;
    conn.execute(format!("PRAGMA hexkey = '{ISSUE_7375_KEY_256}'"))?;

    conn.execute("CREATE TABLE t(id INTEGER PRIMARY KEY, payload BLOB)")?;
    for i in 0..100i64 {
        conn.execute(format!("INSERT INTO t VALUES({i}, randomblob(300))"))?;
    }
    conn.execute("DELETE FROM t WHERE id % 3 = 0")?;

    conn.execute("VACUUM")?;

    let rows: Vec<(i64,)> = conn.exec_rows("SELECT count(*) FROM t");
    assert_eq!(rows[0].0, 66);
    let ps: Vec<(i64,)> = conn.exec_rows("PRAGMA page_size");
    assert_eq!(ps[0].0, 512);

    do_flush(&conn, &tmp_db)?;

    let uri = format!(
        "file:{}?cipher=aegis256x4&hexkey={ISSUE_7375_KEY_256}",
        tmp_db.path.to_str().unwrap()
    );
    let (_io, reopened) =
        turso_core::Connection::from_uri(&uri, DatabaseOpts::new().with_encryption(true))?;
    let rows: Vec<(i64,)> = reopened.exec_rows("SELECT count(*) FROM t");
    assert_eq!(rows[0].0, 66);

    Ok(())
}