diff --git a/Cargo.lock b/Cargo.lock index d82d83b56c..efc9c71e0f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4333,6 +4333,7 @@ dependencies = [ "async-trait", "common", "derive_more", + "enum_from", "futures 0.3.15", "hex 0.4.2", "itertools", diff --git a/mm2src/coins/utxo/utxo_block_header_storage/mod.rs b/mm2src/coins/utxo/utxo_block_header_storage/mod.rs index 1932d9afda..888d067f02 100644 --- a/mm2src/coins/utxo/utxo_block_header_storage/mod.rs +++ b/mm2src/coins/utxo/utxo_block_header_storage/mod.rs @@ -58,7 +58,6 @@ impl BlockHeaderStorage { }) } - #[cfg(test)] pub(crate) fn into_inner(self) -> Box { self.inner } } @@ -108,7 +107,7 @@ impl BlockHeaderStorageOps for BlockHeaderStorage { async fn is_table_empty(&self) -> Result<(), BlockHeaderStorageError> { self.inner.is_table_empty().await } } -#[cfg(test)] +#[cfg(any(test, target_arch = "wasm32"))] mod block_headers_storage_tests { use super::*; use chain::BlockHeaderBits; @@ -313,7 +312,7 @@ mod native_tests { fn test_remove_headers_up_to_height() { block_on(test_remove_headers_up_to_height_impl(FOR_COIN_GET)) } } -#[cfg(all(test, target_arch = "wasm32"))] +#[cfg(target_arch = "wasm32")] mod wasm_test { use super::*; use crate::utxo::utxo_block_header_storage::block_headers_storage_tests::*; @@ -323,7 +322,7 @@ mod wasm_test { wasm_bindgen_test_configure!(run_in_browser); - const FOR_COIN: &str = "RICK"; + const FOR_COIN: &str = "tBTC"; #[wasm_bindgen_test] async fn test_storage_init() { diff --git a/mm2src/coins/utxo/utxo_block_header_storage/wasm/block_header_table.rs b/mm2src/coins/utxo/utxo_block_header_storage/wasm/block_header_table.rs index 79121043d6..951d0aa938 100644 --- a/mm2src/coins/utxo/utxo_block_header_storage/wasm/block_header_table.rs +++ b/mm2src/coins/utxo/utxo_block_header_storage/wasm/block_header_table.rs @@ -1,8 +1,8 @@ -use mm2_db::indexed_db::{DbUpgrader, OnUpgradeResult, TableSignature}; +use mm2_db::indexed_db::{BeBigUint, DbUpgrader, OnUpgradeResult, TableSignature}; #[derive(Clone, Debug, Deserialize, Serialize)] pub struct BlockHeaderStorageTable { - pub height: u64, + pub height: BeBigUint, pub bits: u32, pub hash: String, pub raw_header: String, @@ -10,7 +10,7 @@ pub struct BlockHeaderStorageTable { } impl BlockHeaderStorageTable { - pub const HEIGHT_TICKER_INDEX: &str = "block_height_ticker_index"; + pub const TICKER_HEIGHT_INDEX: &str = "block_height_ticker_index"; pub const HASH_TICKER_INDEX: &str = "block_hash_ticker_index"; } @@ -21,7 +21,7 @@ impl TableSignature for BlockHeaderStorageTable { match (old_version, new_version) { (0, 1) => { let table = upgrader.create_table(Self::table_name())?; - table.create_multi_index(Self::HEIGHT_TICKER_INDEX, &["height", "ticker"], true)?; + table.create_multi_index(Self::TICKER_HEIGHT_INDEX, &["ticker", "height"], true)?; table.create_multi_index(Self::HASH_TICKER_INDEX, &["hash", "ticker"], true)?; table.create_index("ticker", false)?; }, diff --git a/mm2src/coins/utxo/utxo_block_header_storage/wasm/indexeddb_block_header_storage.rs b/mm2src/coins/utxo/utxo_block_header_storage/wasm/indexeddb_block_header_storage.rs index 63d2fa9950..3e19792548 100644 --- a/mm2src/coins/utxo/utxo_block_header_storage/wasm/indexeddb_block_header_storage.rs +++ b/mm2src/coins/utxo/utxo_block_header_storage/wasm/indexeddb_block_header_storage.rs @@ -3,10 +3,10 @@ use super::BlockHeaderStorageTable; use async_trait::async_trait; use chain::BlockHeader; use mm2_core::mm_ctx::MmArc; -use mm2_db::indexed_db::cursor_prelude::{CollectCursor, WithOnly}; -use mm2_db::indexed_db::{ConstructibleDb, DbIdentifier, DbInstance, DbLocked, IndexedDb, IndexedDbBuilder, +use mm2_db::indexed_db::{BeBigUint, ConstructibleDb, DbIdentifier, DbInstance, DbLocked, IndexedDb, IndexedDbBuilder, InitDbResult, MultiIndex, SharedDb}; use mm2_err_handle::prelude::*; +use num_traits::ToPrimitive; use primitives::hash::H256; use serialization::Reader; use spv_validation::storage::{BlockHeaderStorageError, BlockHeaderStorageOps}; @@ -93,15 +93,15 @@ impl BlockHeaderStorageOps for IDBBlockHeadersStorage { let bits: u32 = header.bits.into(); let headers_to_store = BlockHeaderStorageTable { ticker: ticker.clone(), - height, + height: BeBigUint::from(height), bits, hash, raw_header, }; - let index_keys = MultiIndex::new(BlockHeaderStorageTable::HEIGHT_TICKER_INDEX) - .with_value(&height) - .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))? + let index_keys = MultiIndex::new(BlockHeaderStorageTable::TICKER_HEIGHT_INDEX) .with_value(&ticker) + .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))? + .with_value(&BeBigUint::from(height)) .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))?; block_headers_db @@ -149,10 +149,10 @@ impl BlockHeaderStorageOps for IDBBlockHeadersStorage { .table::() .await .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))?; - let index_keys = MultiIndex::new(BlockHeaderStorageTable::HEIGHT_TICKER_INDEX) - .with_value(&height) - .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))? + let index_keys = MultiIndex::new(BlockHeaderStorageTable::TICKER_HEIGHT_INDEX) .with_value(&ticker) + .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))? + .with_value(&BeBigUint::from(height)) .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))?; Ok(block_headers_db @@ -178,21 +178,30 @@ impl BlockHeaderStorageOps for IDBBlockHeadersStorage { .await .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))?; - // Todo: use open_cursor with direction to optimze this process. - let res = block_headers_db - .open_cursor("ticker") - .await - .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))? + let maybe_item = block_headers_db + .cursor_builder() .only("ticker", ticker.clone()) .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))? - .collect() + // We need to provide any constraint on the `height` property + // since `ticker_height` consists of both `ticker` and `height` properties. + .bound("height", BeBigUint::from(0u64), BeBigUint::from(u64::MAX)) + // Cursor returns values from the lowest to highest key indexes. + // But we need to get the most highest height, so reverse the cursor direction. + .reverse() + .open_cursor(BlockHeaderStorageTable::TICKER_HEIGHT_INDEX) .await .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))? - .into_iter() - .map(|(_item_id, item)| item.height) - .collect::>(); + .next() + .await + .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))?; - Ok(res.into_iter().max()) + maybe_item + .map(|(_, item)| { + item.height + .to_u64() + .ok_or_else(|| BlockHeaderStorageError::get_err(&ticker, "height is too large".to_string())) + }) + .transpose() } async fn get_last_block_header_with_non_max_bits( @@ -214,25 +223,29 @@ impl BlockHeaderStorageOps for IDBBlockHeadersStorage { .await .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))?; - // Todo: use open_cursor with direction to optimze this process. - let res = block_headers_db - .open_cursor("ticker") - .await - .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))? + let mut cursor = block_headers_db + .cursor_builder() .only("ticker", ticker.clone()) .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))? - .collect() + // We need to provide any constraint on the `height` property + // since `ticker_height` consists of both `ticker` and `height` properties. + .bound("height", BeBigUint::from(0u64), BeBigUint::from(u64::MAX)) + // Cursor returns values from the lowest to highest key indexes. + // But we need to get the most highest height, so reverse the cursor direction. + .reverse() + .open_cursor(BlockHeaderStorageTable::TICKER_HEIGHT_INDEX) + .await + .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))?; + + while let Some((_item_id, header)) = cursor + .next() .await .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))? - .into_iter() - .map(|(_item_id, item)| item) - .collect::>(); - let res = res - .into_iter() - .filter_map(|e| if e.bits != max_bits { Some(e) } else { None }) - .collect::>(); - - for header in res { + { + if header.bits == max_bits { + continue; + } + let serialized = &hex::decode(header.raw_header).map_err(|e| BlockHeaderStorageError::DecodeError { coin: ticker.clone(), reason: e.to_string(), @@ -273,11 +286,18 @@ impl BlockHeaderStorageOps for IDBBlockHeadersStorage { .with_value(&ticker) .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))?; - Ok(block_headers_db + let maybe_item = block_headers_db .get_item_by_unique_multi_index(index_keys) .await - .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))? - .map(|raw| raw.1.height as i64)) + .map_err(|err| BlockHeaderStorageError::get_err(&ticker, err.to_string()))?; + + maybe_item + .map(|(_, item)| { + item.height + .to_i64() + .ok_or_else(|| BlockHeaderStorageError::get_err(&ticker, "height is too large".to_string())) + }) + .transpose() } async fn remove_headers_up_to_height(&self, to_height: u64) -> Result<(), BlockHeaderStorageError> { @@ -297,10 +317,10 @@ impl BlockHeaderStorageOps for IDBBlockHeadersStorage { .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))?; for height in 0..=to_height { - let index_keys = MultiIndex::new(BlockHeaderStorageTable::HEIGHT_TICKER_INDEX) - .with_value(&height) - .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))? + let index_keys = MultiIndex::new(BlockHeaderStorageTable::TICKER_HEIGHT_INDEX) .with_value(&ticker) + .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))? + .with_value(&BeBigUint::from(height)) .map_err(|err| BlockHeaderStorageError::table_err(&ticker, err.to_string()))?; block_headers_db diff --git a/mm2src/mm2_db/Cargo.toml b/mm2src/mm2_db/Cargo.toml index 5c07fcd407..95baa9715f 100644 --- a/mm2src/mm2_db/Cargo.toml +++ b/mm2src/mm2_db/Cargo.toml @@ -10,6 +10,7 @@ doctest = false async-trait = "0.1" common = { path = "../common" } derive_more = "0.99" +enum_from = { path = "../derives/enum_from" } futures = { version = "0.3", package = "futures", features = ["compat", "async-await", "thread-pool"] } itertools = "0.10" hex = "0.4.2" @@ -26,4 +27,4 @@ serde_json = { version = "1.0", features = ["preserve_order", "raw_value"] } wasm-bindgen = { version = "0.2.50", features = ["nightly"] } wasm-bindgen-futures = { version = "0.4.1" } wasm-bindgen-test = { version = "0.3.2" } -web-sys = { version = "0.3.55", features = ["console", "CloseEvent", "DomException", "ErrorEvent", "IdbDatabase", "IdbCursor", "IdbCursorWithValue", "IdbFactory", "IdbIndex", "IdbIndexParameters", "IdbObjectStore", "IdbObjectStoreParameters", "IdbOpenDbRequest", "IdbKeyRange", "IdbTransaction", "IdbTransactionMode", "IdbVersionChangeEvent", "MessageEvent", "WebSocket"] } +web-sys = { version = "0.3.55", features = ["console", "CloseEvent", "DomException", "ErrorEvent", "IdbDatabase", "IdbCursor", "IdbCursorWithValue", "IdbCursorDirection", "IdbFactory", "IdbIndex", "IdbIndexParameters", "IdbObjectStore", "IdbObjectStoreParameters", "IdbOpenDbRequest", "IdbKeyRange", "IdbTransaction", "IdbTransactionMode", "IdbVersionChangeEvent", "MessageEvent", "WebSocket"] } diff --git a/mm2src/mm2_db/src/indexed_db/drivers/cursor/cursor.rs b/mm2src/mm2_db/src/indexed_db/drivers/cursor/cursor.rs index 694fec6f16..0fdc26261b 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/cursor/cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/cursor/cursor.rs @@ -1,9 +1,9 @@ use super::construct_event_closure; use crate::indexed_db::db_driver::{InternalItem, ItemId}; use crate::indexed_db::BeBigUint; -use async_trait::async_trait; use common::wasm::{deserialize_from_js, serialize_to_js, stringify_js_error}; use derive_more::Display; +use enum_from::EnumFromTrait; use futures::channel::mpsc; use futures::StreamExt; use js_sys::Array; @@ -12,22 +12,23 @@ use serde_json::{self as json, Value as Json}; use std::convert::TryInto; use wasm_bindgen::prelude::*; use wasm_bindgen::JsCast; -use web_sys::{IdbCursorWithValue, IdbIndex, IdbKeyRange, IdbRequest}; +use web_sys::{IdbCursorDirection, IdbCursorWithValue, IdbIndex, IdbKeyRange, IdbRequest}; +mod empty_cursor; mod multi_key_bound_cursor; mod multi_key_cursor; mod single_key_bound_cursor; mod single_key_cursor; -pub use multi_key_bound_cursor::IdbMultiKeyBoundCursor; -pub use multi_key_cursor::IdbMultiKeyCursor; -pub use single_key_bound_cursor::IdbSingleKeyBoundCursor; -pub use single_key_cursor::IdbSingleKeyCursor; +use empty_cursor::IdbEmptyCursor; +use multi_key_bound_cursor::IdbMultiKeyBoundCursor; +use multi_key_cursor::IdbMultiKeyCursor; +use single_key_bound_cursor::IdbSingleKeyBoundCursor; +use single_key_cursor::IdbSingleKeyCursor; pub type CursorResult = Result>; -pub type DbFilter = Box (CollectItemAction, CollectCursorAction) + Send>; -#[derive(Debug, Display, PartialEq)] +#[derive(Debug, Display, EnumFromTrait, PartialEq)] pub enum CursorError { #[display( fmt = "Error serializing the '{}' value of the index field '{}' : {:?}", @@ -59,6 +60,7 @@ pub enum CursorError { )] IncorrectNumberOfKeysPerIndex { expected: usize, found: usize }, #[display(fmt = "Error occurred due to an unexpected state: {:?}", _0)] + #[from_trait(WithInternal::internal)] UnexpectedState(String), #[display(fmt = "Incorrect usage of the cursor: {:?}", description)] IncorrectUsage { description: String }, @@ -81,6 +83,13 @@ pub enum CursorBoundValue { BigUint(BeBigUint), } +#[derive(Default)] +pub struct CursorFilters { + pub(crate) only_keys: Vec<(String, Json)>, + pub(crate) bound_keys: Vec<(String, CursorBoundValue, CursorBoundValue)>, + pub(crate) reverse: bool, +} + impl From for CursorBoundValue { fn from(uint: u32) -> Self { CursorBoundValue::Uint(uint) } } @@ -160,143 +169,209 @@ impl CursorBoundValue { } #[derive(Debug, PartialEq)] -pub enum CollectCursorAction { +pub enum CursorAction { Continue, ContinueWithValue(JsValue), Stop, } #[derive(Debug, PartialEq)] -pub enum CollectItemAction { +pub enum CursorItemAction { Include, Skip, } -#[async_trait(?Send)] -pub trait CursorOps: Sized { - fn db_index(&self) -> &IdbIndex; - +pub trait CursorDriverImpl: Sized { fn key_range(&self) -> CursorResult>; - fn on_collect_iter(&mut self, key: JsValue, value: &Json) - -> CursorResult<(CollectItemAction, CollectCursorAction)>; + fn on_iteration(&mut self, key: JsValue) -> CursorResult<(CursorItemAction, CursorAction)>; +} - /// Collect items that match the specified bounds. - async fn collect(mut self) -> CursorResult> { - let (tx, mut rx) = mpsc::channel(1); +pub(crate) struct CursorDriver { + /// An actual cursor implementation. + inner: IdbCursorEnum, + cursor_request: IdbRequest, + cursor_item_rx: mpsc::Receiver>, + /// Whether we got `CursorAction::Stop` at the last iteration or not. + stopped: bool, + /// We need to hold the closures in memory till `cursor` exists. + _onsuccess_closure: Closure, + _onerror_closure: Closure, +} - let db_index = self.db_index(); - let cursor_request_result = match self.key_range()? { +impl CursorDriver { + pub(crate) fn init_cursor(db_index: IdbIndex, filters: CursorFilters) -> CursorResult { + let reverse = filters.reverse; + let inner = IdbCursorEnum::new(filters)?; + + let cursor_request_result = match inner.key_range()? { + Some(key_range) if reverse => { + db_index.open_cursor_with_range_and_direction(&key_range, IdbCursorDirection::Prev) + }, Some(key_range) => db_index.open_cursor_with_range(&key_range), + // Please note that `IndexedDb` doesn't allow to open a cursor with a direction + // but without a key range. + None if reverse => { + return MmError::err(CursorError::ErrorOpeningCursor { + description: format!("Direction cannot be specified without a range"), + }) + }, None => db_index.open_cursor(), }; let cursor_request = cursor_request_result.map_err(|e| CursorError::ErrorOpeningCursor { description: stringify_js_error(&e), })?; - let onsuccess_closure = construct_event_closure(Ok, tx.clone()); - let onerror_closure = construct_event_closure(Err, tx); + let (cursor_item_tx, cursor_item_rx) = mpsc::channel(1); + + let onsuccess_closure = construct_event_closure(Ok, cursor_item_tx.clone()); + let onerror_closure = construct_event_closure(Err, cursor_item_tx); cursor_request.set_onsuccess(Some(onsuccess_closure.as_ref().unchecked_ref())); cursor_request.set_onerror(Some(onerror_closure.as_ref().unchecked_ref())); - let mut collected_items = Vec::new(); + Ok(CursorDriver { + inner, + cursor_request, + cursor_item_rx, + stopped: false, + _onsuccess_closure: onsuccess_closure, + _onerror_closure: onerror_closure, + }) + } + + pub(crate) async fn next(&mut self) -> CursorResult> { + loop { + // Check if we got `CursorAction::Stop` at the last iteration. + if self.stopped { + return Ok(None); + } + + let event = match self.cursor_item_rx.next().await { + Some(event) => event, + None => { + self.stopped = true; + return Ok(None); + }, + }; - while let Some(event) = rx.next().await { let _cursor_event = event.map_to_mm(|e| CursorError::ErrorOpeningCursor { description: stringify_js_error(&e), })?; - let cursor = match cursor_from_request(&cursor_request)? { + let cursor = match cursor_from_request(&self.cursor_request)? { Some(cursor) => cursor, - // no more items, stop the loop - None => break, + // No more items. + None => { + self.stopped = true; + return Ok(None); + }, }; let (key, js_value) = match (cursor.key(), cursor.value()) { (Ok(key), Ok(js_value)) => (key, js_value), - // no more items, stop the loop - _ => break, + // No more items. + _ => { + self.stopped = true; + return Ok(None); + }, }; let item: InternalItem = deserialize_from_js(js_value).map_to_mm(|e| CursorError::ErrorDeserializingItem(e.to_string()))?; - let (item_action, cursor_action) = self.on_collect_iter(key, &item.item)?; - match item_action { - CollectItemAction::Include => collected_items.push(item.into_pair()), - CollectItemAction::Skip => (), - } + let (item_action, cursor_action) = self.inner.on_iteration(key)?; match cursor_action { - CollectCursorAction::Continue => cursor.continue_().map_to_mm(|e| CursorError::AdvanceError { + CursorAction::Continue => cursor.continue_().map_to_mm(|e| CursorError::AdvanceError { description: stringify_js_error(&e), })?, - CollectCursorAction::ContinueWithValue(next_value) => { + CursorAction::ContinueWithValue(next_value) => { cursor .continue_with_key(&next_value) .map_to_mm(|e| CursorError::AdvanceError { description: stringify_js_error(&e), })? }, - // don't advance the cursor, just stop the loop - CollectCursorAction::Stop => break, + // Don't advance the cursor. + // Here we set the `stopped` flag so we return `Ok(None)` at the next iteration immediately. + // This is required because `item_action` can be `CollectItemAction::Include`, + // and at this iteration we will return `Ok(Some)`. + CursorAction::Stop => self.stopped = true, } - } - Ok(collected_items) + match item_action { + CursorItemAction::Include => return Ok(Some(item.into_pair())), + // Try to fetch the next item. + CursorItemAction::Skip => (), + } + } } } -pub struct IdbCursorBuilder { - db_index: IdbIndex, +pub(crate) enum IdbCursorEnum { + Empty(IdbEmptyCursor), + SingleKey(IdbSingleKeyCursor), + SingleKeyBound(IdbSingleKeyBoundCursor), + MultiKey(IdbMultiKeyCursor), + MultiKeyBound(IdbMultiKeyBoundCursor), } -impl IdbCursorBuilder { - pub fn new(db_index: IdbIndex) -> IdbCursorBuilder { IdbCursorBuilder { db_index } } - - /// Returns a cursor that is a representation of a range that includes records - /// whose value of the `field_name` field equals to the `field_value` value. - pub fn single_key_cursor( - self, - field_name: String, - field_value: Json, - collect_filter: Option, - ) -> IdbSingleKeyCursor { - IdbSingleKeyCursor::new(self.db_index, field_name, field_value, collect_filter) - } +impl IdbCursorEnum { + fn new(cursor_filters: CursorFilters) -> CursorResult { + if cursor_filters.only_keys.len() > 1 && cursor_filters.bound_keys.is_empty() { + return Ok(IdbCursorEnum::MultiKey(IdbMultiKeyCursor::new( + cursor_filters.only_keys, + ))); + } + if !cursor_filters.bound_keys.is_empty() + && (cursor_filters.only_keys.len() + cursor_filters.bound_keys.len() > 1) + { + return Ok(IdbCursorEnum::MultiKeyBound(IdbMultiKeyBoundCursor::new( + cursor_filters.only_keys, + cursor_filters.bound_keys, + )?)); + } // otherwise we're sure that there is either one `only`, or one `bound`, or no constraint specified. + + if let Some((field_name, field_value)) = cursor_filters.only_keys.into_iter().next() { + return Ok(IdbCursorEnum::SingleKey(IdbSingleKeyCursor::new( + field_name, + field_value, + ))); + } + + if let Some((field_name, lower_bound, upper_bound)) = cursor_filters.bound_keys.into_iter().next() { + return Ok(IdbCursorEnum::SingleKeyBound(IdbSingleKeyBoundCursor::new( + field_name, + lower_bound, + upper_bound, + )?)); + } - /// Returns a cursor that is a representation of a range that includes records - /// whose value of the `field_name` field is lower than `lower_bound` and greater than `upper_bound`. - pub fn single_key_bound_cursor( - self, - field_name: String, - lower_bound: CursorBoundValue, - upper_bound: CursorBoundValue, - collect_filter: Option, - ) -> CursorResult { - IdbSingleKeyBoundCursor::new(self.db_index, field_name, lower_bound, upper_bound, collect_filter) + // There are no constraint specified. + Ok(IdbCursorEnum::Empty(IdbEmptyCursor)) } +} - /// Returns a cursor that is a representation of a range that includes records - /// whose fields have only the specified values `only_values`. - pub fn multi_key_cursor( - self, - only_values: Vec<(String, Json)>, - collect_filter: Option, - ) -> CursorResult { - IdbMultiKeyCursor::new(self.db_index, only_values, collect_filter) +impl CursorDriverImpl for IdbCursorEnum { + fn key_range(&self) -> CursorResult> { + match self { + IdbCursorEnum::Empty(empty) => empty.key_range(), + IdbCursorEnum::SingleKey(single) => single.key_range(), + IdbCursorEnum::SingleKeyBound(single_bound) => single_bound.key_range(), + IdbCursorEnum::MultiKey(multi) => multi.key_range(), + IdbCursorEnum::MultiKeyBound(multi_bound) => multi_bound.key_range(), + } } - /// Returns a cursor that is a representation of a range that includes records - /// with the multiple `only` and `bound` restrictions. - pub fn multi_key_bound_cursor( - self, - only_values: Vec<(String, Json)>, - bound_values: Vec<(String, CursorBoundValue, CursorBoundValue)>, - collect_filter: Option, - ) -> CursorResult { - IdbMultiKeyBoundCursor::new(self.db_index, only_values, bound_values, collect_filter) + fn on_iteration(&mut self, key: JsValue) -> CursorResult<(CursorItemAction, CursorAction)> { + match self { + IdbCursorEnum::Empty(empty) => empty.on_iteration(key), + IdbCursorEnum::SingleKey(single) => single.on_iteration(key), + IdbCursorEnum::SingleKeyBound(single_bound) => single_bound.on_iteration(key), + IdbCursorEnum::MultiKey(multi) => multi.on_iteration(key), + IdbCursorEnum::MultiKeyBound(multi_bound) => multi_bound.on_iteration(key), + } } } diff --git a/mm2src/mm2_db/src/indexed_db/drivers/cursor/empty_cursor.rs b/mm2src/mm2_db/src/indexed_db/drivers/cursor/empty_cursor.rs new file mode 100644 index 0000000000..441180d8bb --- /dev/null +++ b/mm2src/mm2_db/src/indexed_db/drivers/cursor/empty_cursor.rs @@ -0,0 +1,14 @@ +use super::{CursorAction, CursorDriverImpl, CursorItemAction, CursorResult}; +use wasm_bindgen::prelude::*; +use web_sys::IdbKeyRange; + +/// The representation of a range that includes all records. +pub struct IdbEmptyCursor; + +impl CursorDriverImpl for IdbEmptyCursor { + fn key_range(&self) -> CursorResult> { Ok(None) } + + fn on_iteration(&mut self, _key: JsValue) -> CursorResult<(CursorItemAction, CursorAction)> { + Ok((CursorItemAction::Include, CursorAction::Continue)) + } +} diff --git a/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_bound_cursor.rs b/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_bound_cursor.rs index 836204e94b..6a6b224fcd 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_bound_cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_bound_cursor.rs @@ -1,37 +1,29 @@ -use super::{index_key_as_array, CollectCursorAction, CollectItemAction, CursorBoundValue, CursorError, CursorOps, - CursorResult, DbFilter}; -use async_trait::async_trait; +use super::{index_key_as_array, CursorAction, CursorBoundValue, CursorDriverImpl, CursorError, CursorItemAction, + CursorResult}; use common::{deserialize_from_js, serialize_to_js, stringify_js_error}; use js_sys::Array; use mm2_err_handle::prelude::*; use serde_json::{json, Value as Json}; use wasm_bindgen::prelude::*; -use web_sys::{IdbIndex, IdbKeyRange}; +use web_sys::IdbKeyRange; /// The representation of a range that includes records /// with the multiple `only` and `bound` restrictions. /// https://developer.mozilla.org/en-US/docs/Web/API/IDBKeyRange/bound pub struct IdbMultiKeyBoundCursor { - db_index: IdbIndex, only_values: Vec<(String, Json)>, bound_values: Vec<(String, CursorBoundValue, CursorBoundValue)>, - /// An additional predicate that may be used to filter records. - collect_filter: Option, } impl IdbMultiKeyBoundCursor { pub(super) fn new( - db_index: IdbIndex, only_values: Vec<(String, Json)>, bound_values: Vec<(String, CursorBoundValue, CursorBoundValue)>, - collect_filter: Option, ) -> CursorResult { Self::check_bounds(&only_values, &bound_values)?; Ok(IdbMultiKeyBoundCursor { - db_index, only_values, bound_values, - collect_filter, }) } @@ -83,10 +75,7 @@ impl IdbMultiKeyBoundCursor { } } -#[async_trait(?Send)] -impl CursorOps for IdbMultiKeyBoundCursor { - fn db_index(&self) -> &IdbIndex { &self.db_index } - +impl CursorDriverImpl for IdbMultiKeyBoundCursor { fn key_range(&self) -> CursorResult> { let lower = Array::new(); let upper = Array::new(); @@ -119,11 +108,7 @@ impl CursorOps for IdbMultiKeyBoundCursor { /// The range `IDBKeyRange.bound([2,2], [4,4])` includes values like `[3,0]` and `[3,5]` as `[2,2] < [3,0] < [3,5] < [4,4]`, /// so we need to do additional filtering. /// For more information on why it's required, see https://stackoverflow.com/a/32976384. - fn on_collect_iter( - &mut self, - index_key: JsValue, - value: &Json, - ) -> CursorResult<(CollectItemAction, CollectCursorAction)> { + fn on_iteration(&mut self, index_key: JsValue) -> CursorResult<(CursorItemAction, CursorAction)> { let index_keys_js_array = index_key_as_array(index_key)?; let index_keys: Vec = index_keys_js_array .iter() @@ -172,8 +157,8 @@ impl CursorOps for IdbMultiKeyBoundCursor { return Ok(( // the `actual_index_value` is not in our expected bounds - CollectItemAction::Skip, - CollectCursorAction::ContinueWithValue(new_index.into()), + CursorItemAction::Skip, + CursorAction::ContinueWithValue(new_index.into()), )); } if &actual_index_value > upper_bound { @@ -192,13 +177,13 @@ impl CursorOps for IdbMultiKeyBoundCursor { return Ok(( // the `actual_index_value` is not in our expected bounds - CollectItemAction::Skip, - CollectCursorAction::ContinueWithValue(new_index.into()), + CursorItemAction::Skip, + CursorAction::ContinueWithValue(new_index.into()), )); } // otherwise there is no an index greater than actual `index`, stop the cursor - return Ok((CollectItemAction::Skip, CollectCursorAction::Stop)); + return Ok((CursorItemAction::Skip, CursorAction::Stop)); } let increased_index_key = actual_index_value.next(); @@ -209,19 +194,13 @@ impl CursorOps for IdbMultiKeyBoundCursor { idx_in_index += 1; idx_in_bounds += 1; } - - // `index_key` is in our expected bounds - if let Some(ref mut filter) = self.collect_filter { - return Ok(filter(value)); - } - Ok((CollectItemAction::Include, CollectCursorAction::Continue)) + Ok((CursorItemAction::Include, CursorAction::Continue)) } } mod tests { use super::*; use common::log::wasm_log::register_wasm_log; - use wasm_bindgen::JsCast; use wasm_bindgen_test::*; wasm_bindgen_test_configure!(run_in_browser); @@ -230,10 +209,10 @@ mod tests { fn test_in_bound_indexes(cursor: &mut IdbMultiKeyBoundCursor, input_indexes: Vec) { for input_index in input_indexes { let input_index_js_value = serialize_to_js(&input_index).unwrap(); - let result = cursor.on_collect_iter(input_index_js_value, &Json::Null); + let result = cursor.on_iteration(input_index_js_value); assert_eq!( result, - Ok((CollectItemAction::Include, CollectCursorAction::Continue)), + Ok((CursorItemAction::Include, CursorAction::Continue)), "'{}' index is expected to be in a bound", input_index ); @@ -247,19 +226,16 @@ mod tests { for (input_index, expected_next) in input_indexes { let input_index_js_value = serialize_to_js(&input_index).unwrap(); let (item_action, cursor_action) = cursor - .on_collect_iter(input_index_js_value, &Json::Null) + .on_iteration(input_index_js_value) .expect(&format!("Error due to the index '{:?}'", input_index)); let actual_next: Json = match cursor_action { - CollectCursorAction::ContinueWithValue(next_index_js_value) => { + CursorAction::ContinueWithValue(next_index_js_value) => { deserialize_from_js(next_index_js_value).expect("Error deserializing next index}") }, - action => panic!( - "Expected 'CollectCursorAction::ContinueWithValue', found '{:?}'", - action - ), + action => panic!("Expected 'CursorAction::ContinueWithValue', found '{:?}'", action), }; - assert_eq!(item_action, CollectItemAction::Skip); + assert_eq!(item_action, CursorItemAction::Skip); assert_eq!(actual_next, expected_next); } } @@ -268,10 +244,10 @@ mod tests { fn test_out_of_bound_indexes(cursor: &mut IdbMultiKeyBoundCursor, input_indexes: Vec) { for input_index in input_indexes { let input_index_js_value = serialize_to_js(&input_index).unwrap(); - let result = cursor.on_collect_iter(input_index_js_value, &Json::Null); + let result = cursor.on_iteration(input_index_js_value); assert_eq!( result, - Ok((CollectItemAction::Skip, CollectCursorAction::Stop)), + Ok((CursorItemAction::Skip, CursorAction::Stop)), "'{}' index is expected to be out of bound", input_index ); @@ -280,7 +256,7 @@ mod tests { /// This test doesn't check [`IdbMultiKeyBoundCursor::filter`]. #[wasm_bindgen_test] - fn test_on_collect_iter_multiple_only_and_bound_values() { + fn test_on_iteration_multiple_only_and_bound_values() { register_wasm_log(); let only_values = vec![ @@ -311,10 +287,7 @@ mod tests { ), ]; - // Use [`wasm_bindgen::JsCast::unchecked_from_js`] to create not-valid `IdbIndex` - // that is not used by [`IdbMultiKeyBoundCursor::on_collect_iter`] anyway. - let db_index = IdbIndex::unchecked_from_js(JsValue::NULL); - let mut cursor = IdbMultiKeyBoundCursor::new(db_index, only_values, bound_values, None).unwrap(); + let mut cursor = IdbMultiKeyBoundCursor::new(only_values, bound_values).unwrap(); ////////////////// @@ -388,7 +361,7 @@ mod tests { /// This test doesn't check [`IdbMultiKeyBoundCursor::filter`]. #[wasm_bindgen_test] - fn test_on_collect_iter_multiple_bound_values() { + fn test_on_iteration_multiple_bound_values() { register_wasm_log(); let only_values = Vec::new(); @@ -405,10 +378,7 @@ mod tests { ), ]; - // Use [`wasm_bindgen::JsCast::unchecked_from_js`] to create not-valid `IdbIndex` - // that is not used by [`IdbMultiKeyBoundCursor::on_collect_iter`] anyway. - let db_index = IdbIndex::unchecked_from_js(JsValue::NULL); - let mut cursor = IdbMultiKeyBoundCursor::new(db_index, only_values, bound_values, None).unwrap(); + let mut cursor = IdbMultiKeyBoundCursor::new(only_values, bound_values).unwrap(); ////////////////// @@ -453,7 +423,7 @@ mod tests { /// This test doesn't check [`IdbMultiKeyBoundCursor::filter`]. #[wasm_bindgen_test] - fn test_on_collect_iter_single_only_and_bound_values() { + fn test_on_iteration_single_only_and_bound_values() { register_wasm_log(); let only_values = vec![("field1".to_owned(), json!(2))]; @@ -463,10 +433,7 @@ mod tests { CursorBoundValue::Uint(5), // upper bound )]; - // Use [`wasm_bindgen::JsCast::unchecked_from_js`] to create not-valid `IdbIndex` - // that is not used by [`IdbMultiKeyBoundCursor::on_collect_iter`] anyway. - let db_index = IdbIndex::unchecked_from_js(JsValue::NULL); - let mut cursor = IdbMultiKeyBoundCursor::new(db_index, only_values, bound_values, None).unwrap(); + let mut cursor = IdbMultiKeyBoundCursor::new(only_values, bound_values).unwrap(); ////////////////// @@ -502,7 +469,7 @@ mod tests { } #[wasm_bindgen_test] - fn test_on_collect_iter_error() { + fn test_on_iteration_error() { register_wasm_log(); let only_values = vec![("field1".to_owned(), json!(2u32))]; @@ -512,10 +479,7 @@ mod tests { CursorBoundValue::Uint(5), // upper bound )]; - // Use [`wasm_bindgen::JsCast::unchecked_from_js`] to create not-valid `IdbIndex` - // that is not used by [`IdbMultiKeyBoundCursor::on_collect_iter`] anyway. - let db_index = IdbIndex::unchecked_from_js(JsValue::NULL); - let mut cursor = IdbMultiKeyBoundCursor::new(db_index, only_values, bound_values, None).unwrap(); + let mut cursor = IdbMultiKeyBoundCursor::new(only_values, bound_values).unwrap(); ////////////////// @@ -531,7 +495,7 @@ mod tests { for (input_index, expected_type) in input_indexes { let input_index_js_value = serialize_to_js(&input_index).unwrap(); let error = cursor - .on_collect_iter(input_index_js_value, &Json::Null) + .on_iteration(input_index_js_value) .expect_err(&format!("'{:?}' must lead to 'CursorError::TypeMismatch'", input_index)); match error.into_inner() { CursorError::TypeMismatch { expected, .. } => assert_eq!(expected, expected_type), @@ -549,7 +513,7 @@ mod tests { for input_index in input_indexes { let input_index_js_value = serialize_to_js(&input_index).unwrap(); let error = cursor - .on_collect_iter(input_index_js_value, &Json::Null) + .on_iteration(input_index_js_value) .expect_err(&format!("'{:?}' must lead to 'CursorError::TypeMismatch'", input_index)); match error.into_inner() { CursorError::IncorrectNumberOfKeysPerIndex { .. } => (), diff --git a/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_cursor.rs b/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_cursor.rs index c161ada34e..f6b128aa1b 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/cursor/multi_key_cursor.rs @@ -1,52 +1,23 @@ -use super::{CollectCursorAction, CollectItemAction, CursorError, CursorOps, CursorResult, DbFilter}; -use async_trait::async_trait; +use super::{CursorAction, CursorDriverImpl, CursorError, CursorItemAction, CursorResult}; use common::{serialize_to_js, stringify_js_error}; use js_sys::Array; use mm2_err_handle::prelude::*; use serde_json::Value as Json; use wasm_bindgen::prelude::*; -use web_sys::{IdbIndex, IdbKeyRange}; +use web_sys::IdbKeyRange; /// The representation of a range that includes records /// whose fields have only the specified [`IdbSingleCursor::only_values`] values. /// https://developer.mozilla.org/en-US/docs/Web/API/IDBKeyRange/only pub struct IdbMultiKeyCursor { - db_index: IdbIndex, only_values: Vec<(String, Json)>, - /// An additional predicate that can be used to filter records. - collect_filter: Option, } impl IdbMultiKeyCursor { - pub(super) fn new( - db_index: IdbIndex, - only_values: Vec<(String, Json)>, - collect_filter: Option, - ) -> CursorResult { - Self::check_only_values(&only_values)?; - Ok(IdbMultiKeyCursor { - db_index, - only_values, - collect_filter, - }) - } - - fn check_only_values(only_values: &Vec<(String, Json)>) -> CursorResult<()> { - if only_values.len() < 2 { - let description = format!( - "Incorrect usage of 'IdbMultiKeyCursor': expected more than one cursor bound, found '{}'", - only_values.len(), - ); - return MmError::err(CursorError::IncorrectUsage { description }); - } - Ok(()) - } + pub(super) fn new(only_values: Vec<(String, Json)>) -> IdbMultiKeyCursor { IdbMultiKeyCursor { only_values } } } -#[async_trait(?Send)] -impl CursorOps for IdbMultiKeyCursor { - fn db_index(&self) -> &IdbIndex { &self.db_index } - +impl CursorDriverImpl for IdbMultiKeyCursor { fn key_range(&self) -> CursorResult> { let only = Array::new(); @@ -65,14 +36,7 @@ impl CursorOps for IdbMultiKeyCursor { Ok(Some(key_range)) } - fn on_collect_iter( - &mut self, - _key: JsValue, - value: &Json, - ) -> CursorResult<(CollectItemAction, CollectCursorAction)> { - if let Some(ref mut filter) = self.collect_filter { - return Ok(filter(value)); - } - Ok((CollectItemAction::Include, CollectCursorAction::Continue)) + fn on_iteration(&mut self, _key: JsValue) -> CursorResult<(CursorItemAction, CursorAction)> { + Ok((CursorItemAction::Include, CursorAction::Continue)) } } diff --git a/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_bound_cursor.rs b/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_bound_cursor.rs index d3138f09e9..92e7ee087b 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_bound_cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_bound_cursor.rs @@ -1,40 +1,31 @@ -use super::{CollectCursorAction, CollectItemAction, CursorBoundValue, CursorError, CursorOps, CursorResult, DbFilter}; -use async_trait::async_trait; +use super::{CursorAction, CursorBoundValue, CursorDriverImpl, CursorError, CursorItemAction, CursorResult}; use common::{log::warn, stringify_js_error}; use mm2_err_handle::prelude::*; -use serde_json::Value as Json; use wasm_bindgen::prelude::*; -use web_sys::{IdbIndex, IdbKeyRange}; +use web_sys::IdbKeyRange; /// The representation of a range that includes records /// whose value of the [`IdbSingleBoundCursor::field_name`] field is lower than [`IdbSingleBoundCursor::lower_bound_value`] /// and greater than [`IdbSingleBoundCursor::upper_bound_value`]. /// https://developer.mozilla.org/en-US/docs/Web/API/IDBKeyRange/bound pub struct IdbSingleKeyBoundCursor { - db_index: IdbIndex, #[allow(dead_code)] field_name: String, lower_bound: CursorBoundValue, upper_bound: CursorBoundValue, - /// An additional predicate that may be used to filter records. - collect_filter: Option, } impl IdbSingleKeyBoundCursor { pub(super) fn new( - db_index: IdbIndex, field_name: String, lower_bound: CursorBoundValue, upper_bound: CursorBoundValue, - collect_filter: Option, ) -> CursorResult { Self::check_bounds(&lower_bound, &upper_bound)?; Ok(IdbSingleKeyBoundCursor { - db_index, field_name, lower_bound, upper_bound, - collect_filter, }) } } @@ -56,10 +47,7 @@ impl IdbSingleKeyBoundCursor { } } -#[async_trait(?Send)] -impl CursorOps for IdbSingleKeyBoundCursor { - fn db_index(&self) -> &IdbIndex { &self.db_index } - +impl CursorDriverImpl for IdbSingleKeyBoundCursor { fn key_range(&self) -> CursorResult> { let key_range = IdbKeyRange::bound(&self.lower_bound.to_js_value()?, &self.upper_bound.to_js_value()?) .map_to_mm(|e| CursorError::InvalidKeyRange { @@ -68,14 +56,7 @@ impl CursorOps for IdbSingleKeyBoundCursor { Ok(Some(key_range)) } - fn on_collect_iter( - &mut self, - _key: JsValue, - value: &Json, - ) -> CursorResult<(CollectItemAction, CollectCursorAction)> { - if let Some(ref mut filter) = self.collect_filter { - return Ok(filter(value)); - } - Ok((CollectItemAction::Include, CollectCursorAction::Continue)) + fn on_iteration(&mut self, _key: JsValue) -> CursorResult<(CursorItemAction, CursorAction)> { + Ok((CursorItemAction::Include, CursorAction::Continue)) } } diff --git a/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_cursor.rs b/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_cursor.rs index f24f0da1f8..b3d6efa476 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/cursor/single_key_cursor.rs @@ -1,46 +1,29 @@ -use super::{CollectCursorAction, CollectItemAction, CursorError, CursorOps, CursorResult, DbFilter}; -use async_trait::async_trait; -use common::{log::warn, serialize_to_js, stringify_js_error}; +use super::{CursorAction, CursorDriverImpl, CursorError, CursorItemAction, CursorResult}; +use common::{serialize_to_js, stringify_js_error}; use mm2_err_handle::prelude::*; use serde_json::Value as Json; use wasm_bindgen::prelude::*; -use web_sys::{IdbIndex, IdbKeyRange}; +use web_sys::IdbKeyRange; /// The representation of a range that includes records /// whose value of the [`IdbSingleKeyCursor::field_name`] field equals to the [`IdbSingleKeyCursor::field_value`] value. /// https://developer.mozilla.org/en-US/docs/Web/API/IDBKeyRange/only pub struct IdbSingleKeyCursor { - db_index: IdbIndex, #[allow(dead_code)] field_name: String, field_value: Json, - /// An additional predicate that may be used to filter records. - collect_filter: Option, } impl IdbSingleKeyCursor { - pub(super) fn new( - db_index: IdbIndex, - field_name: String, - field_value: Json, - filter: Option, - ) -> IdbSingleKeyCursor { - if filter.is_none() { - warn!("Consider using 'IdbObjectStoreImpl::get_items' instead of 'IdbSingleKeyCursor'"); - } + pub(super) fn new(field_name: String, field_value: Json) -> IdbSingleKeyCursor { IdbSingleKeyCursor { - db_index, field_name, field_value, - collect_filter: None, } } } -#[async_trait(?Send)] -impl CursorOps for IdbSingleKeyCursor { - fn db_index(&self) -> &IdbIndex { &self.db_index } - +impl CursorDriverImpl for IdbSingleKeyCursor { fn key_range(&self) -> CursorResult> { let js_value = serialize_to_js(&self.field_value).map_to_mm(|e| CursorError::ErrorSerializingIndexFieldValue { @@ -55,14 +38,7 @@ impl CursorOps for IdbSingleKeyCursor { Ok(Some(key_range)) } - fn on_collect_iter( - &mut self, - _key: JsValue, - value: &Json, - ) -> CursorResult<(CollectItemAction, CollectCursorAction)> { - if let Some(ref mut filter) = self.collect_filter { - return Ok(filter(value)); - } - Ok((CollectItemAction::Include, CollectCursorAction::Continue)) + fn on_iteration(&mut self, _key: JsValue) -> CursorResult<(CursorItemAction, CursorAction)> { + Ok((CursorItemAction::Include, CursorAction::Continue)) } } diff --git a/mm2src/mm2_db/src/indexed_db/drivers/object_store.rs b/mm2src/mm2_db/src/indexed_db/drivers/object_store.rs index 53c9be6104..a7b8081e9c 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/object_store.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/object_store.rs @@ -1,5 +1,4 @@ use super::{construct_event_closure, DbTransactionError, DbTransactionResult, InternalItem, ItemId}; -use crate::indexed_db::db_driver::cursor::IdbCursorBuilder; use common::{deserialize_from_js, serialize_to_js, stringify_js_error}; use futures::channel::mpsc; use futures::StreamExt; @@ -9,7 +8,7 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use wasm_bindgen::prelude::*; use wasm_bindgen::JsCast; -use web_sys::{IdbObjectStore, IdbRequest}; +use web_sys::{IdbIndex, IdbObjectStore, IdbRequest}; pub struct IdbObjectStoreImpl { pub(crate) object_store: IdbObjectStore, @@ -58,10 +57,7 @@ impl IdbObjectStoreImpl { let index = index_str.to_owned(); let index_value_js = try_serialize_index_value!(serialize_to_js(&index_value), index_str); - let db_index = match self.object_store.index(index_str) { - Ok(index) => index, - Err(_) => return MmError::err(DbTransactionError::NoSuchIndex { index }), - }; + let db_index = self.open_index(index_str)?; let get_request = match db_index.get_all_with_key(&index_value_js) { Ok(request) => request, Err(e) => { @@ -90,10 +86,7 @@ impl IdbObjectStoreImpl { let index = index_str.to_owned(); let index_value_js = try_serialize_index_value!(serialize_to_js(&index_value), index); - let db_index = match self.object_store.index(index_str) { - Ok(index) => index, - Err(_) => return MmError::err(DbTransactionError::NoSuchIndex { index }), - }; + let db_index = self.open_index(index_str)?; let get_request = match db_index.get_all_keys_with_key(&index_value_js) { Ok(request) => request, Err(e) => { @@ -141,10 +134,7 @@ impl IdbObjectStoreImpl { let index = index_str.to_owned(); let index_value_js = try_serialize_index_value!(serialize_to_js(&index_value), index); - let db_index = match self.object_store.index(index_str) { - Ok(index) => index, - Err(_) => return MmError::err(DbTransactionError::NoSuchIndex { index }), - }; + let db_index = self.open_index(index_str)?; let count_request = match db_index.count_with_key(&index_value_js) { Ok(request) => request, Err(e) => { @@ -248,16 +238,13 @@ impl IdbObjectStoreImpl { Ok(()) } - pub(crate) fn cursor_builder(&self, index_str: &str) -> DbTransactionResult { - let db_index = match self.object_store.index(index_str) { - Ok(index) => index, - Err(_) => { - return MmError::err(DbTransactionError::NoSuchIndex { - index: index_str.to_owned(), - }) - }, - }; - Ok(IdbCursorBuilder::new(db_index)) + pub(crate) fn open_index(&self, index_str: &str) -> DbTransactionResult { + match self.object_store.index(index_str) { + Ok(index) => Ok(index), + Err(_) => MmError::err(DbTransactionError::NoSuchIndex { + index: index_str.to_owned(), + }), + } } async fn wait_for_request_complete(request: &IdbRequest) -> Result { diff --git a/mm2src/mm2_db/src/indexed_db/drivers/transaction.rs b/mm2src/mm2_db/src/indexed_db/drivers/transaction.rs index aef31a4e7d..f76794b2de 100644 --- a/mm2src/mm2_db/src/indexed_db/drivers/transaction.rs +++ b/mm2src/mm2_db/src/indexed_db/drivers/transaction.rs @@ -1,6 +1,7 @@ use super::IdbObjectStoreImpl; use common::wasm::stringify_js_error; use derive_more::Display; +use enum_from::EnumFromTrait; use mm2_err_handle::prelude::*; use serde_json::Value as Json; use std::collections::HashSet; @@ -11,7 +12,7 @@ use web_sys::IdbTransaction; pub type DbTransactionResult = Result>; -#[derive(Debug, Display, PartialEq)] +#[derive(Debug, Display, EnumFromTrait, PartialEq)] pub enum DbTransactionError { #[display(fmt = "No such table '{}'", table)] NoSuchTable { table: String }, @@ -44,6 +45,7 @@ pub enum DbTransactionError { description: String, }, #[display(fmt = "Error occurred due to an unexpected state: {:?}", _0)] + #[from_trait(WithInternal::internal)] UnexpectedState(String), #[display(fmt = "Transaction was aborted")] TransactionAborted, diff --git a/mm2src/mm2_db/src/indexed_db/indexed_cursor.rs b/mm2src/mm2_db/src/indexed_db/indexed_cursor.rs index e04c622ddf..04dd7c37cb 100644 --- a/mm2src/mm2_db/src/indexed_db/indexed_cursor.rs +++ b/mm2src/mm2_db/src/indexed_db/indexed_cursor.rs @@ -18,10 +18,13 @@ //! you can use [`WithBound::bound`] along with [`WithOnly::only`]: //! ```rust //! let table = open_table_somehow(); -//! let all_rick_morty_swaps = table.open_cursor("search_index") +//! let all_rick_morty_swaps = table +//! .cursor_builder() //! .only("base_coin", "RICK", "MORTY")? //! .bound("base_coin_value", 10, 13) //! .bound("started_at", 1000000030.into(), u32::MAX.into()) +//! .open_cursor("search_index") +//! .await? //! .collect() //! .await?; //! ``` @@ -49,13 +52,12 @@ //! because ['RICK', 'MORTY', 13] < ['RICK', 'MORTY', 13, 1000000030], //! although it is expected to be within the specified bounds. -use crate::indexed_db::db_driver::cursor::{CollectCursorAction, CollectItemAction, CursorBoundValue, CursorOps, - DbFilter, IdbCursorBuilder}; +use crate::indexed_db::db_driver::cursor::CursorBoundValue; +pub(crate) use crate::indexed_db::db_driver::cursor::{CursorDriver, CursorFilters}; pub use crate::indexed_db::db_driver::cursor::{CursorError, CursorResult}; -use crate::indexed_db::{ItemId, TableSignature}; -use async_trait::async_trait; +use crate::indexed_db::{DbTable, ItemId, TableSignature}; use futures::channel::{mpsc, oneshot}; -use futures::StreamExt; +use futures::{SinkExt, StreamExt}; use mm2_err_handle::prelude::*; use serde::Serialize; use serde_json::{self as json, Value as Json}; @@ -65,102 +67,20 @@ use std::marker::PhantomData; pub(super) type DbCursorEventTx = mpsc::UnboundedSender; pub(super) type DbCursorEventRx = mpsc::UnboundedReceiver; -pub enum DbCursorEvent { - Collect { - options: CursorCollectOptions, - result_tx: oneshot::Sender>>, - }, -} - -pub enum CursorCollectOptions { - SingleKey { - field_name: String, - field_value: Json, - filter: Option, - }, - SingleKeyBound { - field_name: String, - lower_bound: CursorBoundValue, - upper_bound: CursorBoundValue, - filter: Option, - }, - MultiKey { - keys: Vec<(String, Json)>, - filter: Option, - }, - MultiKeyBound { - only_keys: Vec<(String, Json)>, - bound_keys: Vec<(String, CursorBoundValue, CursorBoundValue)>, - filter: Option, - }, -} - -pub async fn cursor_event_loop(mut rx: DbCursorEventRx, cursor_builder: IdbCursorBuilder) { - while let Some(event) = rx.next().await { - match event { - DbCursorEvent::Collect { options, result_tx } => { - on_collect_cursor_event(result_tx, options, cursor_builder).await; - return; - }, - } - } +pub struct CursorBuilder<'transaction, 'reference, Table: TableSignature> { + db_table: &'reference DbTable<'transaction, Table>, + filters: CursorFilters, } -async fn on_collect_cursor_event( - result_tx: oneshot::Sender>>, - options: CursorCollectOptions, - cursor_builder: IdbCursorBuilder, -) { - async fn on_collect_cursor_event_impl( - options: CursorCollectOptions, - cursor_builder: IdbCursorBuilder, - ) -> CursorResult> { - match options { - CursorCollectOptions::SingleKey { - field_name, - field_value, - filter, - } => { - cursor_builder - .single_key_cursor(field_name, field_value, filter) - .collect() - .await - }, - CursorCollectOptions::SingleKeyBound { - field_name, - lower_bound, - upper_bound, - filter, - } => { - cursor_builder - .single_key_bound_cursor(field_name, lower_bound, upper_bound, filter)? - .collect() - .await - }, - CursorCollectOptions::MultiKey { keys, filter } => { - cursor_builder.multi_key_cursor(keys, filter)?.collect().await - }, - CursorCollectOptions::MultiKeyBound { - only_keys, - bound_keys, - filter, - } => { - cursor_builder - .multi_key_bound_cursor(only_keys, bound_keys, filter)? - .collect() - .await - }, +impl<'transaction, 'reference, Table: TableSignature> CursorBuilder<'transaction, 'reference, Table> { + pub(crate) fn new(db_table: &'reference DbTable<'transaction, Table>) -> Self { + CursorBuilder { + db_table, + filters: CursorFilters::default(), } } - let result = on_collect_cursor_event_impl(options, cursor_builder).await; - result_tx.send(result).ok(); -} - -pub trait WithOnly: Sized { - type ResultCursor; - - fn only(self, field_name: &str, field_value: Value) -> CursorResult + pub fn only(mut self, field_name: &str, field_value: Value) -> CursorResult where Value: Serialize + fmt::Debug, { @@ -170,367 +90,93 @@ pub trait WithOnly: Sized { value: field_value_str, description: e.to_string(), })?; - Ok(self.only_json(field_name, field_value)) - } - fn only_json(self, field_name: &str, field_value: Json) -> Self::ResultCursor; -} - -pub trait WithBound: Sized { - type ResultCursor; + self.filters.only_keys.push((field_name.to_owned(), field_value)); + Ok(self) + } - fn bound(self, field_name: &str, lower_bound: Value, upper_bound: Value) -> Self::ResultCursor + pub fn bound(mut self, field_name: &str, lower_bound: Value, upper_bound: Value) -> Self where CursorBoundValue: From, { let lower_bound = CursorBoundValue::from(lower_bound); let upper_bound = CursorBoundValue::from(upper_bound); - self.bound_values(field_name, lower_bound, upper_bound) - } - - fn bound_values( - self, - field_name: &str, - lower_bound: CursorBoundValue, - upper_bound: CursorBoundValue, - ) -> Self::ResultCursor; -} - -pub trait WithFilter: Sized { - fn filter(self, filter: F) -> Self - where - F: FnMut(&Json) -> (CollectItemAction, CollectCursorAction) + Send + 'static, - { - let filter = Box::new(filter); - self.filter_boxed(filter) - } - - fn filter_boxed(self, filter: DbFilter) -> Self; -} - -#[async_trait] -pub trait CollectCursor { - async fn collect(self) -> CursorResult>; -} - -#[async_trait] -impl + Send> CollectCursor for T { - async fn collect(self) -> CursorResult> { self.collect_impl().await } -} - -#[async_trait] -pub(crate) trait CollectCursorImpl: Sized { - fn into_collect_options(self) -> CursorCollectOptions; - - fn event_tx(&self) -> DbCursorEventTx; - - async fn collect_impl(self) -> CursorResult> { - let event_tx = self.event_tx(); - let options = self.into_collect_options(); - - let (result_tx, result_rx) = oneshot::channel(); - let event = DbCursorEvent::Collect { result_tx, options }; - let items: Vec<(ItemId, Json)> = send_event_recv_response(&event_tx, event, result_rx).await?; - - items - .into_iter() - .map(|(item_id, item)| json::from_value(item).map(|item| (item_id, item))) - .map(|res| res.map_to_mm(|e| CursorError::ErrorDeserializingItem(e.to_string()))) - // Item = CursorResult<(ItemId, Table)> - .collect() - } -} - -pub struct DbEmptyCursor<'a, Table: TableSignature> { - event_tx: DbCursorEventTx, - filter: Option, - phantom: PhantomData<&'a Table>, -} - -impl<'a, Table: TableSignature> WithOnly for DbEmptyCursor<'a, Table> { - type ResultCursor = DbSingleKeyCursor<'a, Table>; - - fn only_json(self, field_name: &str, field_value: Json) -> Self::ResultCursor { - DbSingleKeyCursor { - event_tx: self.event_tx, - field_name: field_name.to_owned(), - field_value, - filter: self.filter, - phantom: PhantomData::default(), - } - } -} - -impl<'a, Table: TableSignature> WithBound for DbEmptyCursor<'a, Table> { - type ResultCursor = DbSingleKeyBoundCursor<'a, Table>; - - fn bound_values( - self, - field_name: &str, - lower_bound: CursorBoundValue, - upper_bound: CursorBoundValue, - ) -> Self::ResultCursor { - DbSingleKeyBoundCursor { - event_tx: self.event_tx, - field_name: field_name.to_owned(), - lower_bound, - upper_bound, - filter: self.filter, - phantom: PhantomData::default(), - } - } -} - -impl<'a, Table: TableSignature> WithFilter for DbEmptyCursor<'a, Table> { - fn filter_boxed(mut self, filter: DbFilter) -> Self { - self.filter = Some(filter); + self.filters + .bound_keys + .push((field_name.to_owned(), lower_bound, upper_bound)); self } -} - -impl<'a, Table: TableSignature> DbEmptyCursor<'a, Table> { - pub(super) fn new(event_tx: DbCursorEventTx) -> DbEmptyCursor<'a, Table> { - DbEmptyCursor { - event_tx, - filter: None, - phantom: PhantomData::default(), - } - } -} - -pub struct DbSingleKeyCursor<'a, Table: TableSignature> { - event_tx: DbCursorEventTx, - field_name: String, - field_value: Json, - filter: Option, - phantom: PhantomData<&'a Table>, -} - -impl<'a, Table: TableSignature> WithOnly for DbSingleKeyCursor<'a, Table> { - type ResultCursor = DbMultiKeyCursor<'a, Table>; - - fn only_json(self, field_name: &str, field_value: Json) -> Self::ResultCursor { - let keys = vec![ - (self.field_name, self.field_value), - (field_name.to_owned(), field_value), - ]; - DbMultiKeyCursor { - event_tx: self.event_tx, - keys, - filter: self.filter, - phantom: PhantomData::default(), - } - } -} - -impl<'a, Table: TableSignature> WithBound for DbSingleKeyCursor<'a, Table> { - type ResultCursor = DbMultiKeyBoundCursor<'a, Table>; - - fn bound_values( - self, - field_name: &str, - lower_bound: CursorBoundValue, - upper_bound: CursorBoundValue, - ) -> Self::ResultCursor { - let only_keys = vec![(self.field_name, self.field_value)]; - let bound_keys = vec![(field_name.to_owned(), lower_bound, upper_bound)]; - DbMultiKeyBoundCursor { - event_tx: self.event_tx, - only_keys, - bound_keys, - filter: self.filter, - phantom: PhantomData::default(), - } - } -} -impl<'a, Table: TableSignature> WithFilter for DbSingleKeyCursor<'a, Table> { - fn filter_boxed(mut self, filter: DbFilter) -> Self { - self.filter = Some(filter); + pub fn reverse(mut self) -> Self { + self.filters.reverse = true; self } -} - -#[async_trait] -impl<'a, Table: TableSignature> CollectCursorImpl
for DbSingleKeyCursor<'a, Table> { - fn into_collect_options(self) -> CursorCollectOptions { - CursorCollectOptions::SingleKey { - field_name: self.field_name, - field_value: self.field_value, - filter: self.filter, - } - } - - fn event_tx(&self) -> DbCursorEventTx { self.event_tx.clone() } -} - -/// `DbSingleKeyBoundCursor` doesn't implement `WithOnly` trait, because indexes MUST start with `only` values. -pub struct DbSingleKeyBoundCursor<'a, Table: TableSignature> { - event_tx: DbCursorEventTx, - field_name: String, - lower_bound: CursorBoundValue, - upper_bound: CursorBoundValue, - filter: Option, - phantom: PhantomData<&'a Table>, -} -impl<'a, Table: TableSignature> WithBound for DbSingleKeyBoundCursor<'a, Table> { - type ResultCursor = DbMultiKeyBoundCursor<'a, Table>; - - fn bound_values( - self, - field_name: &str, - lower_bound: CursorBoundValue, - upper_bound: CursorBoundValue, - ) -> Self::ResultCursor { - let bound_keys = vec![ - (self.field_name, self.lower_bound, self.upper_bound), - (field_name.to_owned(), lower_bound, upper_bound), - ]; - DbMultiKeyBoundCursor { - event_tx: self.event_tx, - only_keys: Vec::new(), - bound_keys, - filter: self.filter, + /// Opens a cursor by the specified `index`. + /// https://developer.mozilla.org/en-US/docs/Web/API/IDBObjectStore/openCursor + pub async fn open_cursor(self, index: &str) -> CursorResult> { + let event_tx = + self.db_table + .open_cursor(index, self.filters) + .await + .mm_err(|e| CursorError::ErrorOpeningCursor { + description: e.to_string(), + })?; + Ok(CursorIter { + event_tx, phantom: PhantomData::default(), - } - } -} - -impl<'a, Table: TableSignature> WithFilter for DbSingleKeyBoundCursor<'a, Table> { - fn filter_boxed(mut self, filter: DbFilter) -> Self { - self.filter = Some(filter); - self + }) } } -#[async_trait] -impl<'a, Table: TableSignature> CollectCursorImpl
for DbSingleKeyBoundCursor<'a, Table> { - fn into_collect_options(self) -> CursorCollectOptions { - CursorCollectOptions::SingleKeyBound { - field_name: self.field_name, - lower_bound: self.lower_bound, - upper_bound: self.upper_bound, - filter: self.filter, - } - } - - fn event_tx(&self) -> DbCursorEventTx { self.event_tx.clone() } -} - -pub struct DbMultiKeyCursor<'a, Table: TableSignature> { +pub struct CursorIter<'transaction, Table> { event_tx: DbCursorEventTx, - keys: Vec<(String, Json)>, - filter: Option, - phantom: PhantomData<&'a Table>, -} - -impl<'a, Table: TableSignature> WithOnly for DbMultiKeyCursor<'a, Table> { - type ResultCursor = DbMultiKeyCursor<'a, Table>; - - fn only_json(mut self, field_name: &str, field_value: Json) -> Self::ResultCursor { - self.keys.push((field_name.to_owned(), field_value)); - self - } + phantom: PhantomData<&'transaction Table>, } -impl<'a, Table: TableSignature> WithBound for DbMultiKeyCursor<'a, Table> { - type ResultCursor = DbMultiKeyBoundCursor<'a, Table>; - - fn bound_values( - self, - field_name: &str, - lower_bound: CursorBoundValue, - upper_bound: CursorBoundValue, - ) -> Self::ResultCursor { - DbMultiKeyBoundCursor { - event_tx: self.event_tx, - only_keys: self.keys, - bound_keys: vec![(field_name.to_owned(), lower_bound, upper_bound)], - filter: self.filter, - phantom: PhantomData::default(), - } - } -} - -impl<'a, Table: TableSignature> WithFilter for DbMultiKeyCursor<'a, Table> { - fn filter_boxed(mut self, filter: DbFilter) -> Self { - self.filter = Some(filter); - self +impl<'transaction, Table: TableSignature> CursorIter<'transaction, Table> { + /// Advances the iterator and returns the next value. + /// Please note that the items are sorted by the index keys. + pub async fn next(&mut self) -> CursorResult> { + let (result_tx, result_rx) = oneshot::channel(); + self.event_tx + .send(DbCursorEvent::NextItem { result_tx }) + .await + .map_to_mm(|e| CursorError::UnexpectedState(format!("Error sending cursor event: {e}")))?; + let maybe_item = result_rx + .await + .map_to_mm(|e| CursorError::UnexpectedState(format!("Error receiving cursor item: {e}")))??; + let (item_id, item) = match maybe_item { + Some((item_id, item)) => (item_id, item), + None => return Ok(None), + }; + let item = json::from_value(item).map_to_mm(|e| CursorError::ErrorDeserializingItem(e.to_string()))?; + Ok(Some((item_id, item))) } -} -#[async_trait] -impl<'a, Table: TableSignature> CollectCursorImpl
for DbMultiKeyCursor<'a, Table> { - fn into_collect_options(self) -> CursorCollectOptions { - CursorCollectOptions::MultiKey { - keys: self.keys, - filter: self.filter, + pub async fn collect(mut self) -> CursorResult> { + let mut result = Vec::new(); + while let Some((item_id, item)) = self.next().await? { + result.push((item_id, item)); } - } - - fn event_tx(&self) -> DbCursorEventTx { self.event_tx.clone() } -} - -/// `DbMultiKeyBoundCursor` doesn't implement `WithOnly` trait, because indexes MUST start with `only` values. -pub struct DbMultiKeyBoundCursor<'a, Table: TableSignature> { - event_tx: DbCursorEventTx, - only_keys: Vec<(String, Json)>, - bound_keys: Vec<(String, CursorBoundValue, CursorBoundValue)>, - filter: Option, - phantom: PhantomData<&'a Table>, -} - -impl<'a, Table: TableSignature> WithBound for DbMultiKeyBoundCursor<'a, Table> { - type ResultCursor = DbMultiKeyBoundCursor<'a, Table>; - - fn bound_values( - mut self, - field_name: &str, - lower_bound: CursorBoundValue, - upper_bound: CursorBoundValue, - ) -> Self::ResultCursor { - self.bound_keys.push((field_name.to_owned(), lower_bound, upper_bound)); - self + Ok(result) } } -impl<'a, Table: TableSignature> WithFilter for DbMultiKeyBoundCursor<'a, Table> { - fn filter_boxed(mut self, filter: DbFilter) -> Self { - self.filter = Some(filter); - self - } +pub enum DbCursorEvent { + NextItem { + result_tx: oneshot::Sender>>, + }, } -#[async_trait] -impl<'a, Table: TableSignature> CollectCursorImpl
for DbMultiKeyBoundCursor<'a, Table> { - fn into_collect_options(self) -> CursorCollectOptions { - CursorCollectOptions::MultiKeyBound { - only_keys: self.only_keys, - bound_keys: self.bound_keys, - filter: self.filter, +pub(crate) async fn cursor_event_loop(mut rx: DbCursorEventRx, mut cursor: CursorDriver) { + while let Some(event) = rx.next().await { + match event { + DbCursorEvent::NextItem { result_tx } => { + result_tx.send(cursor.next().await).ok(); + }, } } - - fn event_tx(&self) -> DbCursorEventTx { self.event_tx.clone() } -} - -async fn send_event_recv_response( - event_tx: &mpsc::UnboundedSender, - event: Event, - result_rx: oneshot::Receiver>, -) -> CursorResult { - if let Err(e) = event_tx.unbounded_send(event) { - let error = format!("Error sending event: {}", e); - return MmError::err(CursorError::UnexpectedState(error)); - } - match result_rx.await { - Ok(result) => result, - Err(e) => { - let error = format!("Error receiving result: {}", e); - MmError::err(CursorError::UnexpectedState(error)) - }, - } } mod tests { @@ -609,6 +255,15 @@ mod tests { } } + #[track_caller] + async fn next_item(cursor_iter: &mut CursorIter<'_, Table>) -> Option
{ + cursor_iter + .next() + .await + .expect("!CursorIter::next") + .map(|(_item_id, item)| item) + } + /// The table with `BeBigUint` parameters. #[derive(Clone, Debug, Default, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)] #[serde(deny_unknown_fields)] @@ -705,13 +360,14 @@ mod tests { // Get every item that satisfies the following [num_x, num_y] bound. let actual_items = table + .cursor_builder() + .bound("timestamp_x", num_x.clone(), num_y.clone()) .open_cursor("timestamp_x") .await - .expect("!DbTable::open_cursor") - .bound("timestamp_x", num_x.clone(), num_y.clone()) + .expect("!CursorBuilder::open_cursor") .collect() .await - .expect("!DbSingleKeyBoundCursor::collect") + .expect("!CursorIter::collect") .into_iter() // Map `(ItemId, TimestampTable)` into `BeBigUint`. .map(|(_item_id, item)| item.timestamp_x) @@ -759,14 +415,15 @@ mod tests { fill_table(&table, items).await; let mut actual_items = table + .cursor_builder() + .only("base_coin", "RICK") + .expect("!CursorBuilder::only") .open_cursor("base_coin") .await - .expect("!DbTable::open_cursor") - .only("base_coin", "RICK") - .expect("!DbEmptyCursor::only") + .expect("!CursorBuilder::open_cursor") .collect() .await - .expect("!DbSingleKeyCursor::collect") + .expect("!CursorIter::collect") .into_iter() .map(|(_item_id, item)| item) .collect::>(); @@ -812,13 +469,14 @@ mod tests { fill_table(&table, items).await; let mut actual_items = table + .cursor_builder() + .bound("rel_coin_value", 5u32, u32::MAX) .open_cursor("rel_coin_value") .await - .expect("!DbTable::open_cursor") - .bound("rel_coin_value", 5u32, u32::MAX) + .expect("!CursorBuilder::open_cursor") .collect() .await - .expect("!DbSingleKeyCursor::collect") + .expect("!CursorIter::collect") .into_iter() .map(|(_item_id, item)| item) .collect::>(); @@ -870,18 +528,19 @@ mod tests { fill_table(&table, items).await; let mut actual_items = table - .open_cursor("basecoin_basecoinvalue_startedat_index") - .await - .expect("!DbTable::open_cursor") + .cursor_builder() .only("base_coin", "RICK") - .expect("!DbEmptyCursor::only") + .expect("!CursorBuilder::only") .only("base_coin_value", 12) - .expect("!DbSingleKeyCursor::only") + .expect("!CursorBuilder::only") .only("started_at", 721) - .expect("!DbMultiKeyCursor::only") + .expect("!CursorBuilder::only") + .open_cursor("basecoin_basecoinvalue_startedat_index") + .await + .expect("!CursorBuilder::open_cursor") .collect() .await - .expect("!DbMultiKeyCursor::collect") + .expect("!CursorIter::collect") .into_iter() .map(|(_item_id, item)| item) .collect::>(); @@ -946,19 +605,20 @@ mod tests { fill_table(&table, items).await; let actual_items = table - .open_cursor("all_fields_index") - .await - .expect("!DbTable::open_cursor") + .cursor_builder() .only("base_coin", "RICK") - .expect("!DbEmptyCursor::only") + .expect("!CursorBuilder::only") .only("rel_coin", "QRC20") - .expect("!DbEmptyCursor::only") + .expect("!CursorBuilder::only") .bound("base_coin_value", 3u32, 8u32) .bound("rel_coin_value", 10u32, 12u32) .bound("started_at", 600i32, 800i32) + .open_cursor("all_fields_index") + .await + .expect("!CursorBuilder::open_cursor") .collect() .await - .expect("!DbMultiKeyCursor::collect") + .expect("!CursorIter::collect") .into_iter() .map(|(_item_id, item)| item) .collect::>(); @@ -1006,15 +666,16 @@ mod tests { fill_table(&table, items).await; let actual_items = table - .open_cursor("timestamp_xyz") - .await - .expect("!DbTable::open_cursor") + .cursor_builder() .bound("timestamp_x", BeBigUint::from(u64::MAX - 1), BeBigUint::from(u128::MAX)) .bound("timestamp_y", 0u32, 5u32) .bound("timestamp_z", BeBigUint::from(u64::MAX), BeBigUint::from(u128::MAX - 2)) + .open_cursor("timestamp_xyz") + .await + .expect("!CursorBuilder::open_cursor") .collect() .await - .expect("!DbMultiKeyCursor::collect") + .expect("!CursorIter::collect") .into_iter() .map(|(_item_id, item)| item) .collect::>(); @@ -1029,4 +690,198 @@ mod tests { assert_eq!(actual_items, expected_items); } + + #[wasm_bindgen_test] + async fn test_iter_without_constraints() { + const DB_NAME: &str = "TEST_ITER_WITHOUT_CONSTRAINTS"; + const DB_VERSION: u32 = 1; + + register_wasm_log(); + + let items = vec![ + swap_item!("uuid1", "RICK", "MORTY", 10, 3, 700), + swap_item!("uuid2", "MORTY", "KMD", 95000, 1, 721), + swap_item!("uuid3", "RICK", "XYZ", 7, u32::MAX, 1281), + swap_item!("uuid4", "RICK", "MORTY", 8, 6, 92), + ]; + + let db = IndexedDbBuilder::new(DbIdentifier::for_test(DB_NAME)) + .with_version(DB_VERSION) + .with_table::() + .build() + .await + .expect("!IndexedDb::init"); + let transaction = db.transaction().await.expect("!IndexedDb::transaction"); + let table = transaction + .table::() + .await + .expect("!DbTransaction::open_table"); + fill_table(&table, items).await; + + let mut cursor_iter = table + .cursor_builder() + .open_cursor("rel_coin_value") + .await + .expect("!CursorBuilder::open_cursor"); + + // The items must be sorted by `rel_coin_value`. + assert_eq!( + next_item(&mut cursor_iter).await, + Some(swap_item!("uuid2", "MORTY", "KMD", 95000, 1, 721)) + ); + assert_eq!( + next_item(&mut cursor_iter).await, + Some(swap_item!("uuid1", "RICK", "MORTY", 10, 3, 700)) + ); + assert_eq!( + next_item(&mut cursor_iter).await, + Some(swap_item!("uuid4", "RICK", "MORTY", 8, 6, 92)) + ); + assert_eq!( + next_item(&mut cursor_iter).await, + Some(swap_item!("uuid3", "RICK", "XYZ", 7, u32::MAX, 1281)) + ); + assert!(next_item(&mut cursor_iter).await.is_none()); + // Try to poll one more time. This should not fail but return `None`. + assert!(next_item(&mut cursor_iter).await.is_none()); + } + + #[wasm_bindgen_test] + async fn test_rev_iter_without_constraints() { + const DB_NAME: &str = "TEST_REV_ITER_WITHOUT_CONSTRAINTS"; + const DB_VERSION: u32 = 1; + + register_wasm_log(); + + let db = IndexedDbBuilder::new(DbIdentifier::for_test(DB_NAME)) + .with_version(DB_VERSION) + .with_table::() + .build() + .await + .expect("!IndexedDb::init"); + let transaction = db.transaction().await.expect("!IndexedDb::transaction"); + let table = transaction + .table::() + .await + .expect("!DbTransaction::open_table"); + + table + .cursor_builder() + .reverse() + .open_cursor("rel_coin_value") + .await + .map(|_| ()) + .expect_err( + "CursorBuilder::open_cursor should have failed because 'reverse' can be used with key range only", + ); + } + + #[wasm_bindgen_test] + async fn test_iter_single_key_bound_cursor() { + const DB_NAME: &str = "TEST_ITER_SINGLE_KEY_BOUND_CURSOR"; + const DB_VERSION: u32 = 1; + + register_wasm_log(); + + let items = vec![ + swap_item!("uuid1", "RICK", "MORTY", 10, 3, 700), + swap_item!("uuid2", "MORTY", "KMD", 95000, 1, 721), + swap_item!("uuid3", "RICK", "XYZ", 7, u32::MAX, 1281), // + + swap_item!("uuid4", "RICK", "MORTY", 8, 6, 92), // + + swap_item!("uuid5", "QRC20", "RICK", 2, 4, 721), + swap_item!("uuid6", "KMD", "MORTY", 12, 3124, 214), // + + ]; + + let db = IndexedDbBuilder::new(DbIdentifier::for_test(DB_NAME)) + .with_version(DB_VERSION) + .with_table::() + .build() + .await + .expect("!IndexedDb::init"); + let transaction = db.transaction().await.expect("!IndexedDb::transaction"); + let table = transaction + .table::() + .await + .expect("!DbTransaction::open_table"); + fill_table(&table, items).await; + + let mut cursor_iter = table + .cursor_builder() + .bound("rel_coin_value", 5u32, u32::MAX) + .open_cursor("rel_coin_value") + .await + .expect("!CursorBuilder::open_cursor"); + + // The items must be sorted by `rel_coin_value`. + assert_eq!( + next_item(&mut cursor_iter).await, + Some(swap_item!("uuid4", "RICK", "MORTY", 8, 6, 92)) + ); + assert_eq!( + next_item(&mut cursor_iter).await, + Some(swap_item!("uuid6", "KMD", "MORTY", 12, 3124, 214)) + ); + assert_eq!( + next_item(&mut cursor_iter).await, + Some(swap_item!("uuid3", "RICK", "XYZ", 7, u32::MAX, 1281)) + ); + assert!(next_item(&mut cursor_iter).await.is_none()); + // Try to poll one more time. This should not fail but return `None`. + assert!(next_item(&mut cursor_iter).await.is_none()); + } + + #[wasm_bindgen_test] + async fn test_rev_iter_single_key_bound_cursor() { + const DB_NAME: &str = "TEST_REV_ITER_SINGLE_KEY_BOUND_CURSOR"; + const DB_VERSION: u32 = 1; + + register_wasm_log(); + + let items = vec![ + swap_item!("uuid1", "RICK", "MORTY", 10, 3, 700), + swap_item!("uuid2", "MORTY", "KMD", 95000, 1, 721), + swap_item!("uuid3", "RICK", "XYZ", 7, u32::MAX, 1281), // + + swap_item!("uuid4", "RICK", "MORTY", 8, 6, 92), // + + swap_item!("uuid5", "QRC20", "RICK", 2, 4, 721), + swap_item!("uuid6", "KMD", "MORTY", 12, 3124, 214), // + + ]; + + let db = IndexedDbBuilder::new(DbIdentifier::for_test(DB_NAME)) + .with_version(DB_VERSION) + .with_table::() + .build() + .await + .expect("!IndexedDb::init"); + let transaction = db.transaction().await.expect("!IndexedDb::transaction"); + let table = transaction + .table::() + .await + .expect("!DbTransaction::open_table"); + fill_table(&table, items).await; + + let mut cursor_iter = table + .cursor_builder() + .bound("rel_coin_value", 5u32, u32::MAX) + .reverse() + .open_cursor("rel_coin_value") + .await + .expect("!CursorBuilder::open_cursor"); + + // The items must be sorted in reverse order by `rel_coin_value`. + assert_eq!( + next_item(&mut cursor_iter).await, + Some(swap_item!("uuid3", "RICK", "XYZ", 7, u32::MAX, 1281)) + ); + assert_eq!( + next_item(&mut cursor_iter).await, + Some(swap_item!("uuid6", "KMD", "MORTY", 12, 3124, 214)) + ); + assert_eq!( + next_item(&mut cursor_iter).await, + Some(swap_item!("uuid4", "RICK", "MORTY", 8, 6, 92)) + ); + assert!(next_item(&mut cursor_iter).await.is_none()); + // Try to poll one more time. This should not fail but return `None`. + assert!(next_item(&mut cursor_iter).await.is_none()); + } } diff --git a/mm2src/mm2_db/src/indexed_db/indexed_db.rs b/mm2src/mm2_db/src/indexed_db/indexed_db.rs index 897161464f..97fc4d684d 100644 --- a/mm2src/mm2_db/src/indexed_db/indexed_db.rs +++ b/mm2src/mm2_db/src/indexed_db/indexed_db.rs @@ -48,15 +48,15 @@ pub use db_driver::{DbTransactionError, DbTransactionResult, DbUpgrader, InitDbE pub use db_lock::{ConstructibleDb, DbLocked, SharedDb, WeakDb}; use db_driver::{IdbDatabaseBuilder, IdbDatabaseImpl, IdbObjectStoreImpl, IdbTransactionImpl, OnUpgradeNeededCb}; -use indexed_cursor::{cursor_event_loop, DbCursorEventTx, DbEmptyCursor}; +use indexed_cursor::{cursor_event_loop, CursorBuilder, CursorDriver, CursorError, CursorFilters, CursorResult, + DbCursorEventTx}; type DbEventTx = mpsc::UnboundedSender; type DbTransactionEventTx = mpsc::UnboundedSender; type DbTableEventTx = mpsc::UnboundedSender; pub mod cursor_prelude { - pub use crate::indexed_db::indexed_cursor::{CollectCursor, CursorError, CursorResult, WithBound, WithFilter, - WithOnly}; + pub use crate::indexed_db::indexed_cursor::{CursorError, CursorResult}; } pub trait TableSignature: DeserializeOwned + Serialize + 'static { @@ -174,21 +174,20 @@ pub struct IndexedDb { event_tx: DbEventTx, } -async fn send_event_recv_response( +async fn send_event_recv_response( event_tx: &mpsc::UnboundedSender, event: Event, - result_rx: oneshot::Receiver>, -) -> DbTransactionResult { + result_rx: oneshot::Receiver>, +) -> MmResult +where + Error: WithInternal + NotMmError, +{ if let Err(e) = event_tx.unbounded_send(event) { - let error = format!("Error sending event: {}", e); - return MmError::err(DbTransactionError::UnexpectedState(error)); + return MmError::err(Error::internal(format!("Error sending event: {}", e))); } match result_rx.await { Ok(result) => result, - Err(e) => { - let error = format!("Error receiving result: {}", e); - MmError::err(DbTransactionError::UnexpectedState(error)) - }, + Err(e) => MmError::err(Error::internal(format!("Error receiving result: {}", e))), } } @@ -232,9 +231,9 @@ impl IndexedDb { } } -pub struct DbTransaction<'a> { +pub struct DbTransaction<'transaction> { event_tx: DbTransactionEventTx, - phantom: PhantomData<&'a ()>, + phantom: PhantomData<&'transaction ()>, } impl DbTransaction<'_> { @@ -297,9 +296,9 @@ impl DbTransaction<'_> { } } -pub struct DbTable<'a, Table: TableSignature> { +pub struct DbTable<'transaction, Table: TableSignature> { event_tx: DbTableEventTx, - phantom: PhantomData<&'a Table>, + phantom: PhantomData<&'transaction Table>, } pub enum AddOrIgnoreResult { @@ -307,7 +306,7 @@ pub enum AddOrIgnoreResult { ExistAlready(ItemId), } -impl DbTable<'_, Table> { +impl<'transaction, Table: TableSignature> DbTable<'transaction, Table> { /// Adds the given item to the table. /// https://developer.mozilla.org/en-US/docs/Web/API/IDBObjectStore/add pub async fn add_item(&self, item: &Table) -> DbTransactionResult { @@ -636,16 +635,10 @@ impl DbTable<'_, Table> { send_event_recv_response(&self.event_tx, event, result_rx).await } - /// Opens a cursor by the specified `index`. - /// https://developer.mozilla.org/en-US/docs/Web/API/IDBObjectStore/openCursor - pub async fn open_cursor(&self, index: &str) -> DbTransactionResult> { - let (result_tx, result_rx) = oneshot::channel(); - let event = internal::DbTableEvent::OpenCursor { - index: index.to_owned(), - result_tx, - }; - let cursor_event_tx = send_event_recv_response(&self.event_tx, event, result_rx).await?; - Ok(DbEmptyCursor::new(cursor_event_tx)) + /// Returns a `CursorBuilder` builder. It can be used to open a cursor at the specified with specific key bounds. + /// See [`CursorBuilder::open_cursor`]. + pub fn cursor_builder<'reference>(&'reference self) -> CursorBuilder<'transaction, 'reference, Table> { + CursorBuilder::new(self) } /// Whether the transaction is aborted. @@ -655,6 +648,21 @@ impl DbTable<'_, Table> { send_event_recv_response(&self.event_tx, event, result_rx).await } + /// Opens a cursor by the specified `index`. + /// https://developer.mozilla.org/en-US/docs/Web/API/IDBObjectStore/openCursor + async fn open_cursor(&self, index: &str, filters: CursorFilters) -> CursorResult { + let (result_tx, result_rx) = oneshot::channel(); + let event = internal::DbTableEvent::OpenCursor { + index: index.to_owned(), + filters, + result_tx, + }; + let cursor_event_tx = send_event_recv_response(&self.event_tx, event, result_rx) + .await + .mm_err(|e| CursorError::UnexpectedState(e.to_string()))?; + Ok(cursor_event_tx) + } + fn deserialize_items(items: Vec<(ItemId, Json)>) -> DbTransactionResult> { items .into_iter() @@ -726,8 +734,12 @@ async fn table_event_loop(mut rx: mpsc::UnboundedReceiver { result_tx.send(Ok(table.aborted())).ok(); }, - internal::DbTableEvent::OpenCursor { index, result_tx } => { - open_cursor(&table, index, result_tx); + internal::DbTableEvent::OpenCursor { + index, + filters, + result_tx, + } => { + open_cursor(&table, index, filters, result_tx); }, } } @@ -761,18 +773,30 @@ impl MultiIndex { fn open_cursor( table: &IdbObjectStoreImpl, index: String, - result_tx: oneshot::Sender>, + filters: CursorFilters, + result_tx: oneshot::Sender>, ) { - let cursor_builder = match table.cursor_builder(&index) { - Ok(builder) => builder, + let db_index = match table.open_index(&index) { + Ok(db_index) => db_index, + Err(tr_err) => { + let cursor_err = tr_err.map(|tr_err| CursorError::ErrorOpeningCursor { + description: tr_err.to_string(), + }); + result_tx.send(Err(cursor_err)).ok(); + return; + }, + }; + let cursor = match CursorDriver::init_cursor(db_index, filters) { + Ok(cursor) => cursor, Err(e) => { result_tx.send(Err(e)).ok(); return; }, }; + let (event_tx, event_rx) = mpsc::unbounded(); - let fut = async move { cursor_event_loop(event_rx, cursor_builder).await }; + let fut = async move { cursor_event_loop(event_rx, cursor).await }; // `cursor_event_loop` will finish almost immediately once `event_tx` is dropped. spawn_local(fut); @@ -843,7 +867,8 @@ mod internal { }, OpenCursor { index: String, - result_tx: oneshot::Sender>, + filters: CursorFilters, + result_tx: oneshot::Sender>, }, } } diff --git a/mm2src/mm2_main/src/lp_swap/my_swaps_storage.rs b/mm2src/mm2_main/src/lp_swap/my_swaps_storage.rs index 6d2dc630bb..e38b8c5419 100644 --- a/mm2src/mm2_main/src/lp_swap/my_swaps_storage.rs +++ b/mm2src/mm2_main/src/lp_swap/my_swaps_storage.rs @@ -201,37 +201,41 @@ mod wasm_impl { let items = match (&filter.my_coin, &filter.other_coin) { (Some(my_coin), Some(other_coin)) => { my_swaps_table - .open_cursor("with_my_other_coins") - .await? + .cursor_builder() .only("my_coin", my_coin)? .only("other_coin", other_coin)? .bound("started_at", from_timestamp, to_timestamp) + .open_cursor("with_my_other_coins") + .await? .collect() .await? }, (Some(my_coin), None) => { my_swaps_table - .open_cursor("with_my_coin") - .await? + .cursor_builder() .only("my_coin", my_coin)? .bound("started_at", from_timestamp, to_timestamp) + .open_cursor("with_my_coin") + .await? .collect() .await? }, (None, Some(other_coin)) => { my_swaps_table - .open_cursor("with_other_coin") - .await? + .cursor_builder() .only("other_coin", other_coin)? .bound("started_at", from_timestamp, to_timestamp) + .open_cursor("with_other_coin") + .await? .collect() .await? }, (None, None) => { my_swaps_table + .cursor_builder() + .bound("started_at", from_timestamp, to_timestamp) .open_cursor("started_at") .await? - .bound("started_at", from_timestamp, to_timestamp) .collect() .await? },