use std::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};
use turso::{Builder, Connection, Database, transaction::Transaction as DatabaseTransaction};
use crate::Error;
#[derive(Debug, Clone)]
pub struct Store {
pub(crate) database: Database,
}
impl Store {
pub async fn new_memory() -> Result<Self, Error> {
Self::new(":memory:").await
}
pub async fn new(path: &str) -> Result<Self, Error> {
let database = Builder::new_local(path).build().await?;
Ok(Self { database })
}
pub fn from_database(database: Database) -> Self {
Self { database }
}
pub(crate) fn connect(&self) -> Result<Connection, Error> {
Ok(self.database.connect()?)
}
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(())
}
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(())
}
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(())
}
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(())
}
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)
}
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)
}
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;
#[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(())
}
}