bdk_turso 0.2.2

//! Async Turso storage for BDK wallet and chain changesets.
//!
//! [`Store`] owns a Turso database connection pool and exposes methods to migrate the schema and
//! read or write persisted changesets.

use std::{fmt, sync::Arc};

use bdk_chain::{
    BlockId, ConfirmationBlockTime, DescriptorId, bitcoin, keychain_txout, local_chain, tx_graph,
};
use bitcoin::{Amount, BlockHash, OutPoint, ScriptBuf, Transaction, TxOut, Txid, consensus};
pub use turso::EncryptionOpts;
use turso::{Builder, Connection, Database, transaction::Transaction as DatabaseTransaction};

use crate::Error;

/// Opt-in Turso features applied when opening a [`Store`].
///
/// All experimental features are disabled by default. Configure an `Options` value before opening
/// the database; these settings cannot be changed on an already-open database.
#[derive(Clone, Default)]
pub struct Options {
    /// Enables Turso's experimental index-method support.
    index_method: bool,
    /// Enables opening the database from multiple processes.
    multiprocess_wal: bool,
    /// Enables the experimental `VACUUM` command.
    vacuum: bool,
    /// Enables `ATTACH DATABASE` and `DETACH DATABASE`.
    attach: bool,
    /// Enables generated columns.
    generated_columns: bool,
    /// Enables `WITHOUT ROWID` tables.
    without_rowid: bool,
    /// Enables the MVCC passive checkpoint option.
    mvcc_passive_checkpoint: bool,
    /// Enables encryption and supplies its cipher and key.
    ///
    /// Encrypted database files are not readable by SQLite-compatible backends.
    encryption: Option<EncryptionOpts>,
}

impl Options {
    /// Create options with every experimental feature disabled.
    ///
    /// Enable only the capabilities needed by the application by chaining the relevant `with_*`
    /// methods before passing the options to [`Store::new_with_options`] or
    /// [`Store::new_memory_with_options`].
    pub fn new() -> Self {
        Self::default()
    }

    /// Enable Turso's experimental index-method support for the opened database.
    ///
    /// This setting must be enabled before opening the database and only affects databases opened
    /// with the returned options.
    pub fn with_index_method(mut self) -> Self {
        self.index_method = true;
        self
    }

    /// Enable multi-process access to the database through Turso's WAL coordinator.
    ///
    /// Use this only when the same database file must be shared between operating-system
    /// processes; it is disabled by default.
    pub fn with_multiprocess_wal(mut self) -> Self {
        self.multiprocess_wal = true;
        self
    }

    /// Enable Turso's experimental `VACUUM` support for the opened database.
    ///
    /// The option makes the command available; it does not run vacuuming automatically.
    pub fn with_vacuum(mut self) -> Self {
        self.vacuum = true;
        self
    }

    /// Enable SQL `ATTACH DATABASE` and `DETACH DATABASE` statements.
    ///
    /// This is disabled by default and is only applied when the database is opened.
    pub fn with_attach(mut self) -> Self {
        self.attach = true;
        self
    }

    /// Enable Turso's experimental generated-column support for SQL schemas.
    ///
    /// This setting does not alter the crate's wallet schema; it allows applications to use
    /// generated columns in their own database operations.
    pub fn with_generated_columns(mut self) -> Self {
        self.generated_columns = true;
        self
    }

    /// Enable `WITHOUT ROWID` table definitions in the opened database.
    ///
    /// Enable this only if application SQL needs to create such tables.
    pub fn with_without_rowid(mut self) -> Self {
        self.without_rowid = true;
        self
    }

    /// Enable Turso's experimental passive-checkpoint behavior for MVCC databases.
    ///
    /// This setting is opt-in and does not initiate a checkpoint by itself.
    pub fn with_mvcc_passive_checkpoint(mut self) -> Self {
        self.mvcc_passive_checkpoint = true;
        self
    }

    /// Configure at-rest encryption using Turso's cipher name and hexadecimal key.
    ///
    /// Encryption is enabled when this option is present. Keep the key secret and use the same
    /// encryption settings whenever reopening the database. Encrypted files cannot be opened by
    /// SQLite-compatible backends such as `bdk_sqlite`.
    pub fn with_encryption(mut self, encryption: EncryptionOpts) -> Self {
        self.encryption = Some(encryption);
        self
    }
}

impl fmt::Debug for Options {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Options")
            .field("index_method", &self.index_method)
            .field("multiprocess_wal", &self.multiprocess_wal)
            .field("vacuum", &self.vacuum)
            .field("attach", &self.attach)
            .field("generated_columns", &self.generated_columns)
            .field("without_rowid", &self.without_rowid)
            .field("mvcc_passive_checkpoint", &self.mvcc_passive_checkpoint)
            .field("encryption_enabled", &self.encryption.is_some())
            .finish()
    }
}

/// An asynchronous store for BDK wallet changesets backed by a local Turso database.
///
/// Use [`Store::new`] for a database with default settings, or [`Store::new_with_options`] to opt
/// into Turso experimental features. Call [`Store::migrate`] before using the low-level read/write
/// methods; initialization through `AsyncWalletPersister` performs migration automatically.
#[derive(Debug, Clone)]
pub struct Store {
    pub(crate) database: Database,
    options: Options,
}

impl Store {
    /// Create an in-memory database with all experimental options disabled.
    ///
    /// The database exists for the lifetime of the returned store and is not persisted to a file.
    /// Use [`Store::new_memory_with_options`] when an in-memory database needs experimental
    /// features.
    pub async fn new_memory() -> Result<Self, Error> {
        Self::new_memory_with_options(Options::default()).await
    }

    /// Create an in-memory database using the supplied experimental options.
    ///
    /// The options are applied while Turso opens the database. The database remains in memory for
    /// the lifetime of the store and is not persisted to disk.
    pub async fn new_memory_with_options(options: Options) -> Result<Self, Error> {
        Self::new_with_options(":memory:", options).await
    }

    /// Create or open a local database using default options.
    ///
    /// This constructor leaves all experimental options disabled. It does not create the BDK
    /// schema; call [`Store::migrate`] before using the persistence methods, unless the store is
    /// initialized through `AsyncWalletPersister`.
    pub async fn new(path: &str) -> Result<Self, Error> {
        Self::new_with_options(path, Options::default()).await
    }

    /// Create or open a local database with the supplied Turso options.
    ///
    /// The options are applied only while opening the database. All experimental features remain
    /// disabled unless explicitly enabled on `options`. This constructor does not run the BDK
    /// schema migration; call [`Store::migrate`] before using the persistence methods.
    ///
    /// # Examples
    ///
    /// ```no_run
    /// use bdk_turso::{Options, Store};
    ///
    /// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
    /// let options = Options::new().with_attach().with_generated_columns();
    /// let store = Store::new_with_options("wallet.sqlite", options).await?;
    /// store.migrate().await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn new_with_options(path: &str, options: Options) -> Result<Self, Error> {
        let mut builder = Builder::new_local(path)
            .experimental_index_method(options.index_method)
            .experimental_multiprocess_wal(options.multiprocess_wal)
            .experimental_vacuum(options.vacuum)
            .experimental_attach(options.attach)
            .experimental_generated_columns(options.generated_columns)
            .experimental_without_rowid(options.without_rowid)
            .experimental_mvcc_passive_checkpoint(options.mvcc_passive_checkpoint);

        if let Some(encryption) = &options.encryption {
            builder = builder
                .experimental_encryption(true)
                .with_encryption(encryption.clone());
        }

        let database = builder.build().await?;
        Ok(Self { database, options })
    }

    /// Wrap an already-open Turso database in a `Store`.
    ///
    /// This does not open a path or run the BDK schema migration. Call [`Store::migrate`] before
    /// using persistence methods. Because the database was created externally, its opening options
    /// are not inferred by the store.
    pub fn from_database(database: Database) -> Self {
        Self {
            database,
            options: Options::default(),
        }
    }

    /// Return the options retained by this store.
    ///
    /// For stores created with [`Store::new_with_options`] or
    /// [`Store::new_memory_with_options`], this is the configuration passed to Turso. For stores
    /// wrapped with [`Store::from_database`], the original database settings are unknown and the
    /// store retains default options.
    pub fn options(&self) -> &Options {
        &self.options
    }

    pub(crate) fn connect(&self) -> Result<Connection, Error> {
        Ok(self.database.connect()?)
    }

    /// Apply the bundled BDK persistence schema in a transaction.
    ///
    /// This migration is safe to run repeatedly. It must be run before calling the low-level
    /// changeset read/write methods directly; `AsyncWalletPersister::initialize` runs it for you.
    pub async fn migrate(&self) -> Result<(), Error> {
        let mut conn = self.connect()?;
        let txn = conn.transaction().await?;
        txn.execute_batch(include_str!("../migrations/0001_schema.up.sql"))
            .await?;
        txn.commit().await?;
        Ok(())
    }

    /// Write transaction-graph records using an existing database transaction.
    ///
    /// Persists transaction bytes, seen/evicted timestamps, transaction outputs, and confirmation
    /// anchors. Existing transaction and output records are updated; duplicate anchors are
    /// ignored. This helper does not commit `conn`, so the caller controls transaction completion.
    pub async fn write_tx_graph(
        conn: &DatabaseTransaction<'_>,
        tx_graph: &tx_graph::ChangeSet<ConfirmationBlockTime>,
    ) -> Result<(), Error> {
        for tx in &tx_graph.txs {
            let txid = tx.compute_txid();
            conn.execute(
                "INSERT INTO tx(txid, tx) VALUES(?1, ?2) ON CONFLICT DO UPDATE SET tx = ?2",
                (txid.to_string(), consensus::encode::serialize(tx)),
            )
            .await?;
        }
        for (txid, time) in &tx_graph.first_seen {
            conn.execute(
                "INSERT INTO tx(txid, first_seen) VALUES(?1, ?2) ON CONFLICT DO UPDATE SET first_seen = ?2",
                (txid.to_string(), i64::try_from(*time)?),
            )
            .await?;
        }
        for (txid, time) in &tx_graph.last_seen {
            conn.execute(
                "INSERT INTO tx(txid, last_seen) VALUES(?1, ?2) ON CONFLICT DO UPDATE SET last_seen = ?2",
                (txid.to_string(), i64::try_from(*time)?),
            )
            .await?;
        }
        for (txid, time) in &tx_graph.last_evicted {
            conn.execute(
                "INSERT INTO tx(txid, last_evicted) VALUES(?1, ?2) ON CONFLICT DO UPDATE SET last_evicted = ?2",
                (txid.to_string(), i64::try_from(*time)?),
            )
            .await?;
        }
        for (outpoint, txout) in &tx_graph.txouts {
            let OutPoint { txid, vout } = outpoint;
            let TxOut {
                value,
                script_pubkey,
            } = txout;
            conn.execute(
                "INSERT INTO txout(txid, vout, value, script) VALUES(?1, ?2, ?3, ?4) ON CONFLICT DO UPDATE SET value = ?3, script = ?4",
                (
                    txid.to_string(),
                    i64::from(*vout),
                    i64::try_from(value.to_sat())?,
                    script_pubkey.to_bytes(),
                ),
            )
            .await?;
        }
        for (anchor, txid) in &tx_graph.anchors {
            let BlockId { height, hash } = anchor.block_id;
            conn.execute(
                "INSERT OR IGNORE INTO anchor(block_height, block_hash, txid, confirmation_time) VALUES(?1, ?2, ?3, ?4)",
                (
                    i64::from(height),
                    hash.to_string(),
                    txid.to_string(),
                    i64::try_from(anchor.confirmation_time)?,
                ),
            )
            .await?;
        }

        Ok(())
    }

    /// Apply local-chain block changes using an existing database transaction.
    ///
    /// A block with a hash is inserted at its height if that height is not already present; a
    /// block with no hash removes the row at that height. This helper does not commit `conn`.
    pub async fn write_local_chain(
        conn: &DatabaseTransaction<'_>,
        local_chain: &local_chain::ChangeSet,
    ) -> Result<(), Error> {
        for (&height, hash) in &local_chain.blocks {
            match hash {
                Some(hash) => {
                    conn.execute(
                        "INSERT OR IGNORE INTO block(height, hash) VALUES(?1, ?2)",
                        (i64::from(height), hash.to_string()),
                    )
                    .await?;
                }
                None => {
                    conn.execute("DELETE FROM block WHERE height = ?1", [i64::from(height)])
                        .await?;
                }
            }
        }

        Ok(())
    }

    /// Persist keychain reveal indices and cached scripts using an existing transaction.
    ///
    /// Reveal indices are updated when present in the changeset, while cached scripts are inserted
    /// only if they do not already exist. This helper does not commit `conn`.
    pub async fn write_keychain_txout(
        conn: &DatabaseTransaction<'_>,
        keychain_txout: &keychain_txout::ChangeSet,
    ) -> Result<(), Error> {
        for (descriptor_id, last_revealed) in &keychain_txout.last_revealed {
            conn.execute(
                "INSERT INTO keychain_last_revealed(descriptor_id, last_revealed) VALUES(?1, ?2) ON CONFLICT DO UPDATE SET last_revealed = ?2",
                (descriptor_id.to_string(), i64::from(*last_revealed)),
            )
            .await?;
        }
        for (descriptor_id, spk_cache) in &keychain_txout.spk_cache {
            for (derivation_index, script) in spk_cache {
                conn.execute(
                    "INSERT OR IGNORE INTO keychain_script_pubkey(descriptor_id, derivation_index, script) VALUES(?1, ?2, ?3)",
                    (
                        descriptor_id.to_string(),
                        i64::from(*derivation_index),
                        script.to_bytes(),
                    ),
                )
                .await?;
            }
        }

        Ok(())
    }

    /// Reconstruct transaction-graph changes from the database.
    ///
    /// Reads transactions, timestamps, outputs, and confirmation anchors visible through `conn`.
    /// The caller owns the transaction and controls when it is completed.
    pub async fn read_tx_graph(
        conn: &DatabaseTransaction<'_>,
    ) -> Result<tx_graph::ChangeSet<ConfirmationBlockTime>, Error> {
        let mut changeset = tx_graph::ChangeSet::default();

        let mut rows = conn
            .query(
                "SELECT txid, tx, first_seen, last_seen, last_evicted FROM tx",
                (),
            )
            .await?;
        while let Some(row) = rows.next().await? {
            let txid: Txid = row.get::<String>(0)?.parse()?;
            let tx: Option<Vec<u8>> = row.get(1)?;
            let first_seen: Option<i64> = row.get(2)?;
            let last_seen: Option<i64> = row.get(3)?;
            let last_evicted: Option<i64> = row.get(4)?;
            if let Some(data) = tx {
                let tx: Transaction = consensus::encode::deserialize(&data)?;
                changeset.txs.insert(Arc::new(tx));
            }
            if let Some(first_seen) = first_seen {
                changeset.first_seen.insert(txid, first_seen.try_into()?);
            }
            if let Some(last_seen) = last_seen {
                changeset.last_seen.insert(txid, last_seen.try_into()?);
            }
            if let Some(last_evicted) = last_evicted {
                changeset
                    .last_evicted
                    .insert(txid, last_evicted.try_into()?);
            }
        }

        let mut rows = conn
            .query("SELECT txid, vout, value, script FROM txout", ())
            .await?;
        while let Some(row) = rows.next().await? {
            let txid: Txid = row.get::<String>(0)?.parse()?;
            let vout = u32::try_from(row.get::<i64>(1)?)?;
            let value = Amount::from_sat(u64::try_from(row.get::<i64>(2)?)?);
            let script_pubkey = ScriptBuf::from_bytes(row.get::<Vec<u8>>(3)?);
            changeset.txouts.insert(
                OutPoint { txid, vout },
                TxOut {
                    value,
                    script_pubkey,
                },
            );
        }

        let mut rows = conn
            .query(
                "SELECT block_height, block_hash, txid, confirmation_time FROM anchor",
                (),
            )
            .await?;
        while let Some(row) = rows.next().await? {
            let height = u32::try_from(row.get::<i64>(0)?)?;
            let hash: BlockHash = row.get::<String>(1)?.parse()?;
            let txid: Txid = row.get::<String>(2)?.parse()?;
            let confirmation_time = u64::try_from(row.get::<i64>(3)?)?;
            changeset.anchors.insert((
                ConfirmationBlockTime {
                    block_id: BlockId { height, hash },
                    confirmation_time,
                },
                txid,
            ));
        }

        Ok(changeset)
    }

    /// Read all persisted local-chain blocks into a chain changeset.
    ///
    /// Each stored height and hash is returned as a block entry with a hash. The caller owns `conn`
    /// and controls the surrounding transaction.
    pub async fn read_local_chain(
        conn: &DatabaseTransaction<'_>,
    ) -> Result<local_chain::ChangeSet, Error> {
        let mut changeset = local_chain::ChangeSet::default();
        let mut rows = conn.query("SELECT height, hash FROM block", ()).await?;
        while let Some(row) = rows.next().await? {
            let height = u32::try_from(row.get::<i64>(0)?)?;
            let hash: BlockHash = row.get::<String>(1)?.parse()?;
            changeset.blocks.insert(height, Some(hash));
        }

        Ok(changeset)
    }

    /// Read persisted reveal indices and cached scripts into a keychain changeset.
    ///
    /// The returned changeset includes entries present in the database; an empty database yields
    /// empty maps. The caller owns `conn` and controls the surrounding transaction.
    pub async fn read_keychain_txout(
        conn: &DatabaseTransaction<'_>,
    ) -> Result<keychain_txout::ChangeSet, Error> {
        let mut changeset = keychain_txout::ChangeSet::default();

        let mut rows = conn
            .query(
                "SELECT descriptor_id, last_revealed FROM keychain_last_revealed",
                (),
            )
            .await?;
        while let Some(row) = rows.next().await? {
            let descriptor_id: DescriptorId = row.get::<String>(0)?.parse()?;
            let last_revealed = u32::try_from(row.get::<i64>(1)?)?;
            changeset.last_revealed.insert(descriptor_id, last_revealed);
        }

        let mut rows = conn
            .query(
                "SELECT descriptor_id, derivation_index, script FROM keychain_script_pubkey",
                (),
            )
            .await?;
        while let Some(row) = rows.next().await? {
            let descriptor_id: DescriptorId = row.get::<String>(0)?.parse()?;
            let derivation_index = u32::try_from(row.get::<i64>(1)?)?;
            let script = ScriptBuf::from_bytes(row.get::<Vec<u8>>(2)?);
            changeset
                .spk_cache
                .entry(descriptor_id)
                .or_default()
                .insert(derivation_index, script);
        }

        Ok(changeset)
    }
}

#[cfg(test)]
mod test {
    use super::*;
    use bitcoin::hashes::Hash;

    #[test]
    fn options_default_to_disabled() {
        let options = Options::new();

        assert!(!options.index_method);
        assert!(!options.multiprocess_wal);
        assert!(!options.vacuum);
        assert!(!options.attach);
        assert!(!options.generated_columns);
        assert!(!options.without_rowid);
        assert!(!options.mvcc_passive_checkpoint);
        assert!(options.encryption.is_none());
    }

    #[tokio::test]
    async fn block_table_height_is_unique() -> anyhow::Result<()> {
        let mut changeset = local_chain::ChangeSet::default();
        changeset.blocks.insert(0, Some(Hash::hash(b"0")));
        changeset.blocks.insert(1, Some(Hash::hash(b"1")));

        let store = Store::new_memory().await?;
        store.migrate().await?;

        let mut conn = store.connect()?;
        let txn = conn.transaction().await?;
        Store::write_local_chain(&txn, &changeset).await?;
        txn.commit().await?;

        changeset.blocks.insert(1, Some(Hash::hash(b"1a")));
        let mut conn = store.connect()?;
        let txn = conn.transaction().await?;
        Store::write_local_chain(&txn, &changeset).await?;
        txn.commit().await?;

        let conn = store.connect()?;
        let mut rows = conn
            .query("SELECT height, hash FROM block WHERE height = 1", ())
            .await?;
        let row = rows.next().await?.expect("expected block row");
        let row_hash: String = row.get(1)?;
        let expected_hash = BlockHash::hash(b"1").to_string();
        assert_eq!(row_hash, expected_hash);

        let mut changeset = local_chain::ChangeSet::default();
        changeset.blocks.insert(1, None);
        let mut conn = store.connect()?;
        let txn = conn.transaction().await?;
        Store::write_local_chain(&txn, &changeset).await?;
        txn.commit().await?;

        changeset.blocks.insert(1, Some(Hash::hash(b"1a")));
        let mut conn = store.connect()?;
        let txn = conn.transaction().await?;
        Store::write_local_chain(&txn, &changeset).await?;
        txn.commit().await?;

        let conn = store.connect()?;
        let mut rows = conn
            .query("SELECT height, hash FROM block WHERE height = 1", ())
            .await?;
        let row = rows.next().await?.expect("expected block row");
        let row_hash: String = row.get(1)?;
        let expected_hash = BlockHash::hash(b"1a").to_string();
        assert_eq!(row_hash, expected_hash);

        Ok(())
    }
}