1use std::boxed::Box;
8use std::collections::{BTreeMap, BTreeSet};
9use std::path::{Path, PathBuf};
10use std::rc::Rc;
11use std::string::{String, ToString};
12use std::time::Duration;
13use std::vec::Vec;
14
15use db_management::migration::SqliteMigrator;
16use db_management::pool_manager::{Pool, SqlitePoolManager};
17use deadpool::Runtime;
18use miden_client::Word;
19use miden_client::account::{
20 Account,
21 AccountCode,
22 AccountHeader,
23 AccountId,
24 AccountStorage,
25 Address,
26 StorageMapKey,
27 StorageSlotName,
28};
29use miden_client::asset::{Asset, AssetVault, AssetWitness};
30use miden_client::block::BlockHeader;
31use miden_client::crypto::{InOrderIndex, MmrPeaks};
32use miden_client::note::{BlockNumber, NoteScript, NoteTag, Nullifier};
33use miden_client::store::{
34 AccountRecord,
35 AccountStatus,
36 AccountStorageFilter,
37 BlockRelevance,
38 ClientAccountType,
39 InputNoteCursor,
40 InputNoteRecord,
41 NoteFilter,
42 OutputNoteRecord,
43 PartialBlockchainFilter,
44 SettingMutation,
45 SettingScope,
46 Store,
47 StoreError,
48 TransactionFilter,
49};
50use miden_client::sync::{NoteTagRecord, StateSyncUpdate};
51use miden_client::transaction::{TransactionRecord, TransactionStoreUpdate};
52use miden_client::utils::Serializable;
53use miden_protocol::Felt;
54use miden_protocol::account::StorageMapWitness;
55use miden_protocol::asset::AssetId;
56use rusqlite::Connection;
57use rusqlite::types::Value;
58use sql_error::SqlResultExt;
59
60use crate::account::rows::query_vault_assets;
61
62mod account;
63mod builder;
64mod chain_data;
65mod db_management;
66mod forest;
67mod note;
68mod settings;
69mod sql_error;
70mod sync;
71mod transaction;
72
73pub use builder::ClientBuilderSqliteExt;
74
75pub struct SqliteStore {
83 pub(crate) pool: Pool,
84 database_filepath: PathBuf,
85}
86
87impl SqliteStore {
88 pub async fn new(database_filepath: PathBuf) -> Result<Self, StoreError> {
93 if database_filepath.to_str().is_none() {
94 return Err(database_error(format!(
95 "database path is not valid UTF-8: {}",
96 database_filepath.display()
97 )));
98 }
99
100 let sqlite_pool_manager = SqlitePoolManager::new(database_filepath.clone());
101 let pool = Pool::builder(sqlite_pool_manager)
102 .wait_timeout(Some(Duration::from_secs(30)))
103 .runtime(Runtime::Tokio1)
104 .build()
105 .map_err(database_error)?;
106
107 Self::migrate(&pool, SqliteMigrator::client()).await?;
108
109 Ok(SqliteStore { pool, database_filepath })
112 }
113
114 pub fn database_filepath(&self) -> &Path {
116 &self.database_filepath
117 }
118
119 async fn migrate(pool: &Pool, migration: &'static SqliteMigrator) -> Result<(), StoreError> {
124 let conn = pool.get().await.map_err(database_error)?;
125
126 conn.interact(move |conn| migration.apply(conn))
127 .await
128 .map_err(database_error)?
129 .map_err(database_error)
130 }
131
132 async fn interact_with_connection<F, R>(&self, f: F) -> Result<R, StoreError>
139 where
140 F: FnOnce(&mut Connection) -> Result<R, StoreError> + Send + 'static,
141 R: Send + 'static,
142 {
143 self.pool
144 .get()
145 .await
146 .map_err(database_error)?
147 .interact(f)
148 .await
149 .map_err(database_error)?
150 }
151}
152
153#[async_trait::async_trait]
158impl Store for SqliteStore {
159 fn identifier(&self) -> &str {
160 self.database_filepath
161 .to_str()
162 .expect("rejected by SqliteStore::new when not UTF-8")
163 }
164
165 fn get_current_timestamp(&self) -> Option<u64> {
166 Some(current_timestamp_u64())
167 }
168
169 async fn get_note_tags(&self) -> Result<Vec<NoteTagRecord>, StoreError> {
170 self.interact_with_connection(SqliteStore::get_note_tags).await
171 }
172
173 async fn get_unique_note_tags(&self) -> Result<BTreeSet<NoteTag>, StoreError> {
174 self.interact_with_connection(SqliteStore::get_unique_note_tags).await
175 }
176
177 async fn add_note_tag(&self, tag: NoteTagRecord) -> Result<bool, StoreError> {
178 self.interact_with_connection(move |conn| SqliteStore::add_note_tag(conn, tag))
179 .await
180 }
181
182 async fn remove_note_tag(&self, tag: NoteTagRecord) -> Result<usize, StoreError> {
183 self.interact_with_connection(move |conn| SqliteStore::remove_note_tag(conn, tag))
184 .await
185 }
186
187 async fn get_sync_height(&self) -> Result<BlockNumber, StoreError> {
188 self.interact_with_connection(SqliteStore::get_sync_height).await
189 }
190
191 async fn apply_state_sync(&self, state_sync_update: StateSyncUpdate) -> Result<(), StoreError> {
192 self.interact_with_connection(move |conn| {
193 SqliteStore::apply_state_sync(conn, state_sync_update)
194 })
195 .await
196 }
197
198 async fn get_transactions(
199 &self,
200 transaction_filter: TransactionFilter,
201 ) -> Result<Vec<TransactionRecord>, StoreError> {
202 self.interact_with_connection(move |conn| {
203 SqliteStore::get_transactions(conn, &transaction_filter)
204 })
205 .await
206 }
207
208 async fn apply_transaction(&self, tx_update: TransactionStoreUpdate) -> Result<(), StoreError> {
209 self.interact_with_connection(move |conn| SqliteStore::apply_transaction(conn, &tx_update))
210 .await
211 }
212
213 async fn apply_transaction_batch(
214 &self,
215 tx_updates: Vec<TransactionStoreUpdate>,
216 ) -> Result<(), StoreError> {
217 self.interact_with_connection(move |conn| {
218 SqliteStore::apply_transaction_batch(conn, &tx_updates)
219 })
220 .await
221 }
222
223 async fn get_input_notes(
224 &self,
225 filter: NoteFilter,
226 ) -> Result<Vec<InputNoteRecord>, StoreError> {
227 self.interact_with_connection(move |conn| SqliteStore::get_input_notes(conn, &filter))
228 .await
229 }
230
231 async fn get_output_notes(
232 &self,
233 note_filter: NoteFilter,
234 ) -> Result<Vec<OutputNoteRecord>, StoreError> {
235 self.interact_with_connection(move |conn| SqliteStore::get_output_notes(conn, ¬e_filter))
236 .await
237 }
238
239 async fn get_input_note_after(
240 &self,
241 filter: NoteFilter,
242 consumer: AccountId,
243 block_start: Option<BlockNumber>,
244 block_end: Option<BlockNumber>,
245 cursor: Option<InputNoteCursor>,
246 ) -> Result<Option<InputNoteRecord>, StoreError> {
247 self.interact_with_connection(move |conn| {
248 SqliteStore::get_input_note_after(
249 conn,
250 &filter,
251 consumer,
252 block_start,
253 block_end,
254 cursor,
255 )
256 })
257 .await
258 }
259
260 async fn upsert_input_notes(&self, notes: &[InputNoteRecord]) -> Result<(), StoreError> {
261 let notes = notes.to_vec();
262 self.interact_with_connection(move |conn| SqliteStore::upsert_input_notes(conn, ¬es))
263 .await
264 }
265
266 async fn get_note_script(&self, script_root: Word) -> Result<NoteScript, StoreError> {
267 self.interact_with_connection(move |conn| SqliteStore::get_note_script(conn, script_root))
268 .await
269 }
270
271 async fn upsert_note_scripts(&self, note_scripts: &[NoteScript]) -> Result<(), StoreError> {
272 let note_scripts = note_scripts.to_vec();
273 self.interact_with_connection(move |conn| {
274 SqliteStore::upsert_note_scripts(conn, ¬e_scripts)
275 })
276 .await
277 }
278
279 async fn insert_block_header(
280 &self,
281 block_header: &BlockHeader,
282 nodes: &[(InOrderIndex, Word)],
283 has_client_notes: bool,
284 ) -> Result<(), StoreError> {
285 let block_header = block_header.clone();
286 let nodes = nodes.to_vec();
287 self.interact_with_connection(move |conn| {
288 SqliteStore::insert_block_header(conn, &block_header, &nodes, has_client_notes)
289 })
290 .await
291 }
292
293 async fn untrack_and_prune_irrelevant_blocks(
294 &self,
295 blocks_to_untrack: &[BlockNumber],
296 node_indices_to_remove: &[InOrderIndex],
297 ) -> Result<(), StoreError> {
298 let blocks_to_untrack = blocks_to_untrack.to_vec();
299 let node_indices_to_remove = node_indices_to_remove.to_vec();
300 self.interact_with_connection(move |conn| {
301 SqliteStore::untrack_and_prune_irrelevant_blocks(
302 conn,
303 &blocks_to_untrack,
304 &node_indices_to_remove,
305 )
306 })
307 .await
308 }
309
310 async fn prune_account_history(
311 &self,
312 account_id: AccountId,
313 up_to_nonce: Felt,
314 ) -> Result<usize, StoreError> {
315 self.interact_with_connection(move |conn| {
316 SqliteStore::prune_account_history(conn, account_id, up_to_nonce)
317 })
318 .await
319 }
320
321 async fn get_block_headers(
322 &self,
323 block_numbers: &BTreeSet<BlockNumber>,
324 ) -> Result<Vec<(BlockHeader, BlockRelevance)>, StoreError> {
325 let block_numbers = block_numbers.clone();
326 self.interact_with_connection(move |conn| {
327 SqliteStore::get_block_headers(conn, &block_numbers)
328 })
329 .await
330 }
331
332 async fn get_tracked_block_headers(&self) -> Result<Vec<BlockHeader>, StoreError> {
333 self.interact_with_connection(SqliteStore::get_tracked_block_headers).await
334 }
335
336 async fn get_tracked_block_header_numbers(&self) -> Result<BTreeSet<usize>, StoreError> {
337 self.interact_with_connection(SqliteStore::get_tracked_block_header_numbers)
338 .await
339 }
340
341 async fn get_partial_blockchain_nodes(
342 &self,
343 filter: PartialBlockchainFilter,
344 ) -> Result<BTreeMap<InOrderIndex, Word>, StoreError> {
345 self.interact_with_connection(move |conn| {
346 SqliteStore::get_partial_blockchain_nodes(conn, &filter)
347 })
348 .await
349 }
350
351 async fn get_current_blockchain_peaks(&self) -> Result<MmrPeaks, StoreError> {
352 self.interact_with_connection(SqliteStore::get_current_blockchain_peaks).await
353 }
354
355 async fn insert_account(
356 &self,
357 account: &Account,
358 initial_address: Address,
359 client_account_type: ClientAccountType,
360 ) -> Result<(), StoreError> {
361 let cloned_account = account.clone();
362
363 self.interact_with_connection(move |conn| {
364 SqliteStore::insert_account(
365 conn,
366 &cloned_account,
367 &initial_address,
368 client_account_type,
369 )
370 })
371 .await
372 }
373
374 async fn update_account(&self, account: &Account) -> Result<(), StoreError> {
375 let cloned_account = account.clone();
376
377 self.interact_with_connection(move |conn| {
378 SqliteStore::update_account(conn, &cloned_account)
379 })
380 .await
381 }
382
383 async fn get_account_ids(&self) -> Result<Vec<AccountId>, StoreError> {
384 self.interact_with_connection(SqliteStore::get_account_ids).await
385 }
386
387 async fn get_account_headers(&self) -> Result<Vec<(AccountHeader, AccountStatus)>, StoreError> {
388 self.interact_with_connection(SqliteStore::get_account_headers).await
389 }
390
391 async fn get_account_header(
392 &self,
393 account_id: AccountId,
394 ) -> Result<Option<(AccountHeader, AccountStatus)>, StoreError> {
395 self.interact_with_connection(move |conn| SqliteStore::get_account_header(conn, account_id))
396 .await
397 }
398
399 async fn get_account_header_by_commitment(
400 &self,
401 account_commitment: Word,
402 ) -> Result<Option<AccountHeader>, StoreError> {
403 self.interact_with_connection(move |conn| {
404 SqliteStore::get_account_header_by_commitment(conn, account_commitment)
405 })
406 .await
407 }
408
409 async fn get_account(
410 &self,
411 account_id: AccountId,
412 ) -> Result<Option<AccountRecord>, StoreError> {
413 self.interact_with_connection(move |conn| SqliteStore::get_account(conn, account_id))
414 .await
415 }
416
417 async fn get_account_code(
418 &self,
419 account_id: AccountId,
420 ) -> Result<Option<AccountCode>, StoreError> {
421 self.interact_with_connection(move |conn| {
422 SqliteStore::get_account_code_by_id(conn, account_id)
423 })
424 .await
425 }
426
427 async fn upsert_foreign_account_code(
428 &self,
429 account_id: AccountId,
430 code: AccountCode,
431 ) -> Result<(), StoreError> {
432 self.interact_with_connection(move |conn| {
433 SqliteStore::upsert_foreign_account_code(conn, account_id, &code)
434 })
435 .await
436 }
437
438 async fn get_foreign_account_code(
439 &self,
440 account_ids: Vec<AccountId>,
441 ) -> Result<BTreeMap<AccountId, AccountCode>, StoreError> {
442 self.interact_with_connection(move |conn| {
443 SqliteStore::get_foreign_account_code(conn, account_ids)
444 })
445 .await
446 }
447
448 async fn set_setting(
449 &self,
450 scope: SettingScope,
451 key: String,
452 value: Vec<u8>,
453 ) -> Result<(), StoreError> {
454 self.interact_with_connection(move |conn| {
455 SqliteStore::set_setting(conn, scope, &key, &value)
456 })
457 .await
458 }
459
460 async fn get_setting(
461 &self,
462 scope: SettingScope,
463 key: String,
464 ) -> Result<Option<Vec<u8>>, StoreError> {
465 self.interact_with_connection(move |conn| SqliteStore::get_setting(conn, scope, &key))
466 .await
467 }
468
469 async fn remove_setting(&self, scope: SettingScope, key: String) -> Result<bool, StoreError> {
470 self.interact_with_connection(move |conn| SqliteStore::remove_setting(conn, scope, &key))
471 .await
472 }
473
474 async fn list_setting_keys(&self, scope: SettingScope) -> Result<Vec<String>, StoreError> {
475 self.interact_with_connection(move |conn| SqliteStore::list_setting_keys(conn, scope))
476 .await
477 }
478
479 async fn apply_settings_mutations(
480 &self,
481 scope: SettingScope,
482 mutations: Vec<SettingMutation>,
483 ) -> Result<(), StoreError> {
484 self.interact_with_connection(move |conn| {
485 with_write_tx(conn, |tx| {
486 for mutation in &mutations {
487 match mutation {
488 SettingMutation::Set { key, value } => {
489 SqliteStore::set_setting(tx, scope, key, value)?;
490 },
491 SettingMutation::Remove { key } => {
492 SqliteStore::remove_setting(tx, scope, key)?;
493 },
494 }
495 }
496 Ok(())
497 })
498 })
499 .await
500 }
501
502 async fn get_unspent_input_note_nullifiers(&self) -> Result<Vec<Nullifier>, StoreError> {
503 self.interact_with_connection(SqliteStore::get_unspent_input_note_nullifiers)
504 .await
505 }
506
507 async fn get_account_vault(&self, account_id: AccountId) -> Result<AssetVault, StoreError> {
508 self.interact_with_connection(move |conn| SqliteStore::get_account_vault(conn, account_id))
509 .await
510 }
511
512 async fn get_account_assets(&self, account_id: AccountId) -> Result<Vec<Asset>, StoreError> {
513 self.interact_with_connection(move |conn| query_vault_assets(conn, account_id))
514 .await
515 }
516
517 async fn get_vault_asset_witnesses(
518 &self,
519 account_id: AccountId,
520 vault_root: Word,
521 asset_ids: BTreeSet<AssetId>,
522 ) -> Result<Vec<AssetWitness>, StoreError> {
523 self.interact_with_connection(move |conn| {
524 SqliteStore::get_vault_asset_witnesses(conn, account_id, vault_root, asset_ids)
525 })
526 .await
527 }
528
529 async fn get_account_asset(
530 &self,
531 account_id: AccountId,
532 asset_id: AssetId,
533 ) -> Result<Option<(Asset, AssetWitness)>, StoreError> {
534 self.interact_with_connection(move |conn| {
535 SqliteStore::get_account_asset(conn, account_id, asset_id)
536 })
537 .await
538 }
539
540 async fn get_account_map_item(
541 &self,
542 account_id: AccountId,
543 slot_name: StorageSlotName,
544 key: StorageMapKey,
545 ) -> Result<(Word, StorageMapWitness), StoreError> {
546 self.interact_with_connection(move |conn| {
547 SqliteStore::get_account_map_item(conn, account_id, slot_name, key)
548 })
549 .await
550 }
551
552 async fn get_account_storage(
553 &self,
554 account_id: AccountId,
555 filter: AccountStorageFilter,
556 ) -> Result<AccountStorage, StoreError> {
557 self.interact_with_connection(move |conn| {
558 SqliteStore::get_account_storage(conn, account_id, &filter)
559 })
560 .await
561 }
562
563 async fn get_addresses_by_account_id(
564 &self,
565 account_id: AccountId,
566 ) -> Result<Vec<Address>, StoreError> {
567 self.interact_with_connection(move |conn| {
568 SqliteStore::get_account_addresses(conn, account_id)
569 })
570 .await
571 }
572
573 async fn insert_address(
574 &self,
575 address: Address,
576 account_id: AccountId,
577 ) -> Result<(), StoreError> {
578 self.interact_with_connection(move |conn| {
579 SqliteStore::insert_address(conn, &address, account_id)
580 })
581 .await
582 }
583
584 async fn remove_address(&self, address: Address) -> Result<bool, StoreError> {
585 self.interact_with_connection(move |conn| SqliteStore::remove_address(conn, &address))
586 .await
587 }
588
589 async fn get_minimal_partial_account(
590 &self,
591 account_id: AccountId,
592 ) -> Result<Option<AccountRecord>, StoreError> {
593 self.interact_with_connection(move |conn| {
594 SqliteStore::get_minimal_partial_account(conn, account_id)
595 })
596 .await
597 }
598}
599
600fn database_error(err: impl core::fmt::Display) -> StoreError {
604 StoreError::DatabaseError(err.to_string())
605}
606
607pub(crate) fn current_timestamp_u64() -> u64 {
609 let now = chrono::Utc::now();
610 u64::try_from(now.timestamp()).expect("timestamp is always after epoch")
611}
612
613pub(crate) fn column_value_as_u64<I: rusqlite::RowIndex>(
618 row: &rusqlite::Row<'_>,
619 index: I,
620) -> rusqlite::Result<u64> {
621 let value: i64 = row.get(index)?;
622 #[allow(
623 clippy::cast_sign_loss,
624 reason = "We store u64 as i64 as sqlite only allows the latter."
625 )]
626 Ok(value as u64)
627}
628
629pub(crate) fn u64_to_value(v: u64) -> Value {
634 #[allow(
635 clippy::cast_possible_wrap,
636 reason = "We store u64 as i64 as sqlite only allows the latter."
637 )]
638 Value::Integer(v as i64)
639}
640
641pub(crate) fn blob_array<T: Serializable>(items: impl IntoIterator<Item = T>) -> Rc<Vec<Value>> {
648 Rc::new(items.into_iter().map(|item| Value::Blob(item.to_bytes())).collect())
649}
650
651pub(crate) fn int_array(items: impl IntoIterator<Item = u64>) -> Rc<Vec<Value>> {
654 Rc::new(items.into_iter().map(u64_to_value).collect())
655}
656
657pub(crate) fn text_array(items: impl IntoIterator<Item = String>) -> Rc<Vec<Value>> {
659 Rc::new(items.into_iter().map(Value::Text).collect())
660}
661
662pub(crate) fn with_write_tx<R>(
668 conn: &mut Connection,
669 f: impl FnOnce(&rusqlite::Transaction<'_>) -> Result<R, StoreError>,
670) -> Result<R, StoreError> {
671 let tx = conn
672 .transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)
673 .into_store_error()?;
674 let result = f(&tx)?;
675 tx.commit().into_store_error()?;
676 Ok(result)
677}
678
679#[cfg(test)]
683pub mod tests {
684 use std::boxed::Box;
685 use std::sync::LazyLock;
686
687 use miden_client::store::Store;
688 use miden_client::testing::common::create_test_store_path;
689
690 use super::db_management::migration::SqliteMigrator;
691 use super::db_management::migration::tests::damaging_migration;
692 use super::db_management::pool_manager::SqlitePoolManager;
693 use super::{Pool, SqliteStore, StoreError, column_value_as_u64, u64_to_value, with_write_tx};
694
695 static DAMAGING_MIGRATION: LazyLock<SqliteMigrator> = LazyLock::new(damaging_migration);
698
699 #[tokio::test]
700 async fn failed_migration_leaves_the_store_as_it_was() {
701 let database_filepath = create_test_store_path();
702 drop(SqliteStore::new(database_filepath.clone()).await.unwrap());
703
704 let pool = Pool::builder(SqlitePoolManager::new(database_filepath.clone()))
705 .build()
706 .unwrap();
707 let err = SqliteStore::migrate(&pool, &DAMAGING_MIGRATION).await.unwrap_err();
708
709 assert!(
710 err.to_string().contains("produced a schema this client does not expect"),
711 "the migration should have been rejected, got {err}"
712 );
713 SqliteStore::new(database_filepath).await.unwrap();
715 }
716
717 fn assert_send_sync<T: Send + Sync>() {}
718
719 #[test]
722 fn u64_column_round_trip() {
723 let conn = rusqlite::Connection::open_in_memory().unwrap();
724 for value in [0u64, 1, 1 << 63, u64::MAX] {
725 let read: u64 = conn
726 .query_row("SELECT ?1", [u64_to_value(value)], |row| column_value_as_u64(row, 0))
727 .unwrap();
728 assert_eq!(read, value);
729 }
730 }
731
732 #[test]
733 fn with_write_tx_rolls_back_on_error() {
734 let mut conn = rusqlite::Connection::open_in_memory().unwrap();
735 conn.execute_batch("CREATE TABLE t (id INTEGER PRIMARY KEY);").unwrap();
736
737 let result = with_write_tx(&mut conn, |tx| {
738 tx.execute("INSERT INTO t (id) VALUES (1)", []).unwrap();
739 Err::<(), _>(StoreError::DatabaseError("forced failure".into()))
740 });
741 assert!(result.is_err());
742
743 let count: i64 = conn.query_row("SELECT COUNT(*) FROM t", [], |row| row.get(0)).unwrap();
744 assert_eq!(count, 0, "the insert must roll back when the closure errors");
745 }
746
747 #[test]
748 fn with_write_tx_commits_on_success() {
749 let mut conn = rusqlite::Connection::open_in_memory().unwrap();
750 conn.execute_batch("CREATE TABLE t (id INTEGER PRIMARY KEY);").unwrap();
751
752 with_write_tx(&mut conn, |tx| {
753 tx.execute("INSERT INTO t (id) VALUES (1)", []).unwrap();
754 Ok(())
755 })
756 .unwrap();
757
758 let count: i64 = conn.query_row("SELECT COUNT(*) FROM t", [], |row| row.get(0)).unwrap();
759 assert_eq!(count, 1);
760 }
761
762 #[test]
763 fn is_send_sync() {
764 assert_send_sync::<SqliteStore>();
765 assert_send_sync::<Box<dyn Store>>();
766 }
767
768 async fn dyn_trait_send_fut(store: Box<dyn Store>) {
770 let res = store.get_tracked_block_headers().await;
772 assert!(res.is_ok());
773 }
774
775 #[tokio::test]
776 async fn future_is_send() {
777 let client = SqliteStore::new(create_test_store_path()).await.unwrap();
778 let client: Box<SqliteStore> = client.into();
779 tokio::task::spawn(async move { dyn_trait_send_fut(client).await });
780 }
781
782 pub(crate) async fn create_test_store() -> SqliteStore {
783 SqliteStore::new(create_test_store_path()).await.unwrap()
784 }
785}