Merge pull request #2 from crodas/tags

Tags and better filter

src/main.rs

@@ -4,7 +4,7 @@ use actix_web::{
 };
 use serde::{Deserialize, Serialize};
 use serde_json::json;
-use verax::{AccountId, AnyAmount, AnyId, Asset, RevId, Status, Type};
+use verax::{AccountId, AnyAmount, AnyId, Asset, Filter, RevId, Status, Tag, Type};
 
 #[derive(Deserialize)]
 pub struct Movement {
@@ -19,7 +19,7 @@ pub struct Deposit {
     #[serde(flatten)]
     pub amount: AnyAmount,
     pub memo: String,
-    pub tags: Vec<String>,
+    pub tags: Vec<Tag>,
     pub status: Status,
 }
 
@@ -41,7 +41,7 @@ impl Deposit {
         Ok(if !self.tags.is_empty() {
             ledger
                 ._inner
-                .set_tags(&zdeposit.id, self.tags, "Update tags".to_owned())
+                .set_tags(zdeposit.revision_id, self.tags, "Update tags".to_owned())
                 .await?
         } else {
             zdeposit
@@ -66,7 +66,7 @@ pub struct UpdateTransaction {
 impl UpdateTransaction {
     pub async fn to_ledger_transaction(
         self,
-        id: &RevId,
+        id: RevId,
         ledger: &Ledger,
     ) -> Result<verax::Transaction, verax::Error> {
         ledger
@@ -132,63 +132,73 @@ async fn get_balance(info: web::Path<AccountId>, ledger: web::Data<Ledger>) -> i
 
 #[get("/{id}")]
 async fn get_info(info: web::Path<AnyId>, ledger: web::Data<Ledger>) -> impl Responder {
-    let result = match info.0 {
-        AnyId::Account(account_id) => ledger
-            ._inner
-            .get_transactions(
-                &account_id,
-                vec![Type::Deposit, Type::Withdrawal, Type::Transaction],
-            )
-            .await
-            .map(|transactions| HttpResponse::Ok().json(transactions)),
-        AnyId::Revision(rev_id) => ledger
-            ._inner
-            .get_transaction_by_revision(&rev_id)
-            .await
-            .map(|rev| {
+    let (cache_for_ever, filter) = match info.0 {
+        AnyId::Account(account_id) => (
+            false,
+            Filter {
+                accounts: vec![account_id],
+                kind: vec![Type::Deposit, Type::Withdrawal, Type::Transaction],
+                ..Default::default()
+            },
+        ),
+        AnyId::Revision(rev_id) => (
+            true,
+            Filter {
+                revisions: vec![rev_id],
+                limit: 1,
+                ..Default::default()
+            },
+        ),
+
+        AnyId::Transaction(transaction_id) => (
+            false,
+            Filter {
+                ids: vec![transaction_id],
+                limit: 1,
+                ..Default::default()
+            },
+        ),
+
+        AnyId::Payment(payment_id) => {
+            let _ = ledger
+                ._inner
+                .get_payment_info(&payment_id)
+                .await
+                .map(|tx| HttpResponse::Ok().json(tx));
+
+            todo!()
+        }
+    };
+
+    let limit = filter.limit;
+
+    ledger
+        ._inner
+        .get_transactions(filter)
+        .await
+        .map(|results| {
+            let json_response = if limit == 1 {
+                serde_json::to_value(&results[0])
+            } else {
+                serde_json::to_value(&results)
+            }
+            .unwrap();
+
+            if cache_for_ever {
                 HttpResponse::Ok()
                     .header(
                         "Cache-Control",
                         "public, max-age=31536000, s-maxage=31536000, immutable",
                     )
                     .header("Vary", "Accept-Encoding")
-                    .json(rev)
-            }),
-
-        AnyId::Transaction(transaction_id) => ledger
-            ._inner
-            .get_transaction(&transaction_id)
-            .await
-            .map(|tx| {
-                if ledger
-                    ._inner
-                    .get_status_manager()
-                    .is_final(&tx.revision.status)
-                {
-                    HttpResponse::Ok()
-                        .header(
-                            "Cache-Control",
-                            "public, max-age=31536000, s-maxage=31536000, immutable",
-                        )
-                        .header("Vary", "Accept-Encoding")
-                        .json(tx)
-                } else {
-                    HttpResponse::Ok().json(tx)
-                }
-            }),
-        AnyId::Payment(payment_id) => ledger
-            ._inner
-            .get_payment_info(&payment_id)
-            .await
-            .map(|tx| HttpResponse::Ok().json(tx)),
-    };
-
-    match result {
-        Ok(x) => x,
-        Err(err) => {
+                    .json(json_response)
+            } else {
+                HttpResponse::Ok().json(json_response)
+            }
+        })
+        .map_err(|err| {
             HttpResponse::InternalServerError().json(json!({ "text": err.to_string(), "err": err}))
-        }
-    }
+        })
 }
 
 #[post("/deposit")]
@@ -226,7 +236,7 @@ async fn update_status(
 ) -> impl Responder {
     match item
         .into_inner()
-        .to_ledger_transaction(&info.0, &ledger)
+        .to_ledger_transaction(info.0, &ledger)
         .await
     {
         Ok(tx) => HttpResponse::Accepted().json(tx),

utxo/src/error.rs

@@ -8,6 +8,10 @@ pub enum Error {
     #[error("Transaction: {0}")]
     Transaction(#[from] transaction::Error),
 
+    /// Transaction not found
+    #[error("Transaction not found")]
+    TxNotFound,
+
     /// An internal conversion error
     #[error("Conversion overflow: {0}")]
     Overflow(String),

utxo/src/filter.rs

@@ -0,0 +1,149 @@
+use crate::{AccountId, RevId, Tag, TxId, Type};
+use chrono::{DateTime, Utc};
+use serde::{Deserialize, Serialize};
+
+/// Filter transactions
+///
+/// All this filters options are AND, meaning that an object must match all the
+/// requirements in order to be included in the resultset.
+#[derive(Clone, Debug, Serialize, Deserialize, Default)]
+pub struct Filter {
+    /// List of transaction IDs to query
+    #[serde(skip_serializing_if = "Vec::is_empty")]
+    pub ids: Vec<TxId>,
+    /// List of revisions to query
+    #[serde(skip_serializing_if = "Vec::is_empty")]
+    pub revisions: Vec<RevId>,
+    /// List of accounts to query their transactions
+    #[serde(skip_serializing_if = "Vec::is_empty")]
+    pub accounts: Vec<AccountId>,
+    /// List of transaction types-kind
+    #[serde(rename = "type", skip_serializing_if = "Vec::is_empty")]
+    pub typ: Vec<Type>,
+    /// List of transactions by tags
+    #[serde(skip_serializing_if = "Vec::is_empty")]
+    pub tags: Vec<Tag>,
+    /// List transactions newer than this timestamp
+    #[serde(
+        default,
+        with = "option_ts_seconds",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub since: Option<DateTime<Utc>>,
+    /// List transactions upto this timestamp
+    #[serde(
+        default,
+        with = "option_ts_seconds",
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub until: Option<DateTime<Utc>>,
+    /// Limit for transactions
+    #[serde(default, skip_serializing_if = "is_zero")]
+    pub limit: usize,
+    /// Skip the first `skip` transactions
+    #[serde(default, skip_serializing_if = "is_zero")]
+    pub skip: usize,
+}
+
+impl Filter {
+    /// Adds a given kind to the filter
+    pub fn kind(mut self, kind: Type) -> Self {
+        self.typ.push(kind);
+        self
+    }
+
+    /// Adds a given tag to the filter
+    pub fn account(mut self, account: AccountId) -> Self {
+        self.accounts.push(account);
+        self
+    }
+
+    /// Adds until timestamp to the filter
+    pub fn until(mut self, time: DateTime<Utc>) -> Self {
+        self.until = Some(time);
+        self
+    }
+
+    /// Adds since timestamp to the filter
+    pub fn since(mut self, time: DateTime<Utc>) -> Self {
+        self.since = Some(time);
+        self
+    }
+}
+
+impl From<Vec<Tag>> for Filter {
+    fn from(value: Vec<Tag>) -> Self {
+        Filter {
+            tags: value,
+            ..Default::default()
+        }
+    }
+}
+
+impl From<TxId> for Filter {
+    fn from(value: TxId) -> Self {
+        Filter {
+            ids: vec![value],
+            ..Default::default()
+        }
+    }
+}
+
+impl From<RevId> for Filter {
+    fn from(value: RevId) -> Self {
+        Filter {
+            revisions: vec![value],
+            ..Default::default()
+        }
+    }
+}
+
+impl From<AccountId> for Filter {
+    fn from(value: AccountId) -> Self {
+        Filter {
+            accounts: vec![value],
+            ..Default::default()
+        }
+    }
+}
+
+#[inline]
+fn is_zero(number: &usize) -> bool {
+    *number == 0
+}
+
+pub(crate) mod option_ts_seconds {
+    use chrono::{DateTime, LocalResult, TimeZone, Utc};
+    use serde::{de, Deserialize, Deserializer, Serializer};
+
+    pub fn serialize<S>(date: &Option<DateTime<Utc>>, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        match date {
+            Some(date) => serializer.serialize_i64(date.timestamp()),
+            None => serializer.serialize_none(),
+        }
+    }
+
+    pub fn deserialize<'de, D>(deserializer: D) -> Result<Option<DateTime<Utc>>, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        let timestamp: Option<i64> = Deserialize::deserialize(deserializer)?;
+        let timestamp = if let Some(timestamp) = timestamp {
+            timestamp
+        } else {
+            return Ok(None);
+        };
+
+        match Utc.timestamp_opt(timestamp, 0) {
+            LocalResult::None => Err(de::Error::custom("Invalid timestamp")),
+            LocalResult::Ambiguous(min, max) => Err(de::Error::custom(format!(
+                "Timestamp {0} is ambiguous (min={1}, max={2})",
+                timestamp, min, max
+            ))),
+            LocalResult::Single(ts) => Ok(Some(ts)),
+        }
+    }
+}

utxo/src/ledger.rs

@@ -1,7 +1,7 @@
 use crate::{
     amount::AmountCents, config::Config, status::StatusManager, storage::Storage,
-    transaction::Type, AccountId, Amount, Error, PaymentFrom, PaymentId, RevId, Status,
-    Transaction, TxId,
+    transaction::Type, AccountId, Amount, Error, Filter, PaymentFrom, PaymentId, RevId, Status,
+    Tag, Transaction, TxId,
 };
 use std::{cmp::Ordering, collections::HashMap, sync::Arc};
 
@@ -267,45 +267,26 @@ where
         todo!()
     }
 
-    /// Returns the transaction object by a given revision id
-    pub async fn get_transaction_by_revision(
-        &self,
-        revision_id: &RevId,
-    ) -> Result<Transaction, Error> {
-        Ok(self
-            .config
-            .storage
-            .get_transaction(revision_id.into())
-            .await?)
-    }
-
     /// Returns the transaction object by a given transaction id
-    pub async fn get_transaction(&self, transaction_id: &TxId) -> Result<Transaction, Error> {
-        Ok(self
-            .config
+    pub async fn get_transaction(&self, transaction_id: TxId) -> Result<Transaction, Error> {
+        let filter = Filter {
+            ids: vec![transaction_id],
+            limit: 1,
+            ..Default::default()
+        };
+        self.config
             .storage
-            .get_transaction(transaction_id.clone().into())
-            .await?)
+            .find(filter)
+            .await?
+            .pop()
+            .ok_or(Error::TxNotFound)
     }
 
     /// Returns all transactions from a given account. It can be optionally be
     /// sorted by transaction type. The transactions are sorted from newest to
     /// oldest.
-    pub async fn get_transactions(
-        &self,
-        account_id: &AccountId,
-        types: Vec<Type>,
-    ) -> Result<Vec<Transaction>, Error> {
-        let types = if types.is_empty() {
-            vec![Type::Transaction, Type::Deposit, Type::Withdrawal]
-        } else {
-            types
-        };
-        Ok(self
-            .config
-            .storage
-            .get_transactions(account_id, &types, &[])
-            .await?)
+    pub async fn get_transactions(&self, filter: Filter) -> Result<Vec<Transaction>, Error> {
+        Ok(self.config.storage.find(filter).await?)
     }
 
     /// Returns the status manager
@@ -316,15 +297,22 @@ where
     /// Updates a transaction and updates their tags to this given set
     pub async fn set_tags(
         &self,
-        transaction_id: &TxId,
-        tags: Vec<String>,
+        revision_id: RevId,
+        tags: Vec<Tag>,
         reason: String,
     ) -> Result<Transaction, Error> {
+        let filter = Filter {
+            revisions: vec![revision_id],
+            limit: 1,
+            ..Default::default()
+        };
         Ok(self
             .config
             .storage
-            .get_transaction(transaction_id.into())
+            .find(filter)
             .await?
+            .pop()
+            .ok_or(Error::TxNotFound)?
             .set_tags(&self.config, tags, reason)
             .await?)
     }
@@ -333,15 +321,22 @@ where
     /// new transaction object is returned, otherwise an error is returned.
     pub async fn change_status(
         &self,
-        revision_id: &RevId,
+        revision_id: RevId,
         new_status: Status,
         reason: String,
     ) -> Result<Transaction, Error> {
+        let filter = Filter {
+            revisions: vec![revision_id],
+            limit: 1,
+            ..Default::default()
+        };
         Ok(self
             .config
             .storage
-            .get_transaction(revision_id.into())
+            .find(filter)
             .await?
+            .pop()
+            .ok_or(Error::TxNotFound)?
             .change_status(&self.config, new_status, reason)
             .await?)
     }

utxo/src/lib.rs

@@ -27,6 +27,7 @@ mod amount;
 mod asset;
 mod config;
 mod error;
+mod filter;
 mod id;
 mod ledger;
 mod payment;
@@ -44,6 +45,7 @@ pub use self::{
     amount::{Amount, AnyAmount, HumanAmount},
     asset::Asset,
     error::Error,
+    filter::Filter,
     id::*,
     ledger::Ledger,
     payment::PaymentFrom,

utxo/src/storage/cache/batch.rs

@@ -2,7 +2,7 @@ use super::Storage;
 use crate::{
     payment::PaymentTo,
     storage::{Batch, Error, ReceivedPaymentStatus},
-    AccountId, BaseTx, PaymentId, RevId, Revision, TxId,
+    AccountId, BaseTx, PaymentId, RevId, Revision, Tag, TxId,
 };
 use std::{collections::HashMap, marker::PhantomData, sync::Arc};
 
@@ -173,10 +173,13 @@ where
     async fn tag_transaction(
         &mut self,
         transaction_id: &TxId,
-        tags: &[String],
+        base_tx: &BaseTx,
+        tags: &[Tag],
     ) -> Result<(), Error> {
         self.to_invalidate
             .insert(Ids::Transaction(transaction_id.clone()), ());
-        self.inner.tag_transaction(transaction_id, tags).await
+        self.inner
+            .tag_transaction(transaction_id, base_tx, tags)
+            .await
     }
 }

utxo/src/storage/cache/mod.rs

@@ -1,8 +1,8 @@
 //! Cache storage implementation.
 use crate::{
     amount::AmountCents,
-    storage::{self, Error, FilterId},
-    AccountId, Amount, Asset, BaseTx, PaymentFrom, PaymentId, RevId, Revision, TxId, Type,
+    storage::{self, Error},
+    AccountId, Amount, Asset, BaseTx, Filter, PaymentFrom, PaymentId, RevId, Revision, TxId,
 };
 use std::{collections::HashMap, sync::Arc};
 use tokio::sync::RwLock;
@@ -29,8 +29,6 @@ where
 pub struct Storage {
     payments: SharedHashMap<PaymentId, PaymentFrom>,
     balances: SharedHashMap<AccountId, Vec<Amount>>,
-    transactions: SharedHashMap<TxId, BaseTx>,
-    revisions: SharedHashMap<RevId, Revision>,
     revisions_by_transactions: SharedHashMap<TxId, Vec<RevId>>,
 }
 
@@ -79,59 +77,6 @@ where
         }
     }
 
-    async fn get_transaction_and_revision(
-        &self,
-        id: FilterId,
-    ) -> Result<(BaseTx, Revision), Error> {
-        let tx_cache = self.storage.transactions.read().await;
-        let rev_by_tx_cache = self.storage.revisions_by_transactions.read().await;
-        let rev_cache = self.storage.revisions.read().await;
-        match &id {
-            FilterId::LatestRevision(id) => {
-                if let Some(Some(Some(Some((tx, rev))))) = tx_cache.get(id).map(|tx| {
-                    rev_by_tx_cache.get(id).map(|revs| {
-                        revs.last().map(|last_rev| {
-                            rev_cache.get(last_rev).map(|rev| (tx.clone(), rev.clone()))
-                        })
-                    })
-                }) {
-                    return Ok((tx, rev));
-                }
-            }
-            FilterId::RevisionId(rev_id) => {
-                if let Some(Some((tx, rev))) = rev_cache.get(rev_id).map(|revision| {
-                    tx_cache
-                        .get(revision.transaction_id())
-                        .map(|tx| (tx.clone(), revision.clone()))
-                }) {
-                    return Ok((tx, rev));
-                }
-            }
-        }
-        drop(tx_cache);
-        drop(rev_by_tx_cache);
-        drop(rev_cache);
-
-        let (base_tx, rev) = self.inner.get_transaction_and_revision(id).await?;
-
-        self.storage.transactions.write().await.insert(
-            base_tx
-                .id()
-                .map_err(|e| Error::Encoding(e.to_string()))?
-                .clone(),
-            base_tx.clone(),
-        );
-
-        self.storage.revisions.write().await.insert(
-            rev.rev_id()
-                .map_err(|e| Error::Encoding(e.to_string()))?
-                .clone(),
-            rev.clone(),
-        );
-
-        Ok((base_tx, rev))
-    }
-
     async fn get_revisions(&self, transaction_id: &TxId) -> Result<Vec<RevId>, Error> {
         let cache = self.storage.revisions_by_transactions.read().await;
         if let Some(revisions) = cache.get(transaction_id) {
@@ -148,6 +93,13 @@ where
         }
     }
 
+    async fn find_base_tx_and_revision(
+        &self,
+        filter: Filter,
+    ) -> Result<Vec<(BaseTx, Revision)>, Error> {
+        self.inner.find_base_tx_and_revision(filter).await
+    }
+
     async fn get_negative_unspent_payments(
         &self,
         account: &AccountId,
@@ -168,17 +120,6 @@ where
             .get_positive_unspent_payments(account, asset, target_amount)
             .await
     }
-
-    async fn get_transactions_with_latest_revision(
-        &self,
-        account: &AccountId,
-        types: &[Type],
-        tags: &[String],
-    ) -> Result<Vec<(BaseTx, Revision)>, Error> {
-        self.inner
-            .get_transactions_with_latest_revision(account, types, tags)
-            .await
-    }
 }
 
 #[cfg(test)]

utxo/src/storage/cursor.rs

@@ -0,0 +1,102 @@
+//! Cursor implementation
+use crate::{AccountId, BaseTx, Filter, RevId, Revision, Tag, TxId, Type};
+
+#[derive(Debug, Clone)]
+/// The primary filter is used to filter the transactions before applying the other filters. Think
+/// of it like which key is used to filter transactions quickly before applying the other filters.
+pub enum PrimaryFilter {
+    /// By transaction ID
+    TxId(Vec<TxId>),
+    /// By revision ID
+    Revision(Vec<RevId>),
+    /// By accounts
+    Account(Vec<AccountId>),
+    /// By transaction type
+    Type(Vec<Type>),
+    /// By tags
+    Tags(Vec<Tag>),
+    /// By transaction status
+    Stream,
+}
+
+#[derive(Debug, Clone)]
+/// The cursor
+pub struct Cursor {
+    /// The primary filter
+    pub primary_filter: PrimaryFilter,
+    /// The secondary filter
+    pub filter: Filter,
+}
+
+impl From<Filter> for Cursor {
+    fn from(mut filter: Filter) -> Self {
+        let primary_filter = if !filter.revisions.is_empty() {
+            PrimaryFilter::Revision(std::mem::take(&mut filter.revisions))
+        } else if !filter.ids.is_empty() {
+            PrimaryFilter::TxId(std::mem::take(&mut filter.ids))
+        } else if !filter.accounts.is_empty() {
+            PrimaryFilter::Account(std::mem::take(&mut filter.accounts))
+        } else if !filter.typ.is_empty() {
+            PrimaryFilter::Type(std::mem::take(&mut filter.typ))
+        } else if !filter.tags.is_empty() {
+            PrimaryFilter::Tags(std::mem::take(&mut filter.tags))
+        } else {
+            PrimaryFilter::Stream
+        };
+
+        filter.ids.sort();
+        filter.revisions.sort();
+        filter.accounts.sort();
+        filter.typ.sort();
+        filter.tags.sort();
+
+        Self {
+            primary_filter,
+            filter,
+        }
+    }
+}
+
+impl Cursor {
+    /// Check if the base transaction and the revision matches the current cursor filter
+    pub fn matches(&self, base: &BaseTx, revision: &Revision) -> bool {
+        if !self.filter.ids.is_empty()
+            && self
+                .filter
+                .ids
+                .binary_search(&revision.transaction_id)
+                .is_err()
+        {
+            return false;
+        }
+
+        if !self.filter.revisions.is_empty()
+            && self
+                .filter
+                .revisions
+                .binary_search(&revision.rev_id().expect("vv"))
+                .is_err()
+        {
+            return false;
+        }
+
+        if !self.filter.typ.is_empty() && self.filter.typ.binary_search(&base.typ).is_err() {
+            return false;
+        }
+
+        if !self.filter.tags.is_empty() {
+            let mut found = false;
+            for tag in revision.tags.iter() {
+                if self.filter.tags.binary_search(&tag).is_ok() {
+                    found = true;
+                    break;
+                }
+            }
+            if !found {
+                return false;
+            }
+        }
+
+        true
+    }
+}

utxo/src/storage/mod.rs

@@ -1,15 +1,18 @@
 //! Storage layer trait
 use crate::{
     amount::AmountCents, payment::PaymentTo, transaction::Type, AccountId, Amount, Asset, BaseTx,
-    PaymentFrom, PaymentId, RevId, Revision, Transaction, TxId,
+    Filter, PaymentFrom, PaymentId, RevId, Revision, Tag, Transaction, TxId,
 };
 //use chrono::{DateTime, Utc};
 use serde::Serialize;
 
 pub mod cache;
+mod cursor;
+
 #[cfg(any(feature = "sqlite", test))]
 pub mod sqlite;
 pub use self::cache::Cache;
+pub use self::cursor::{Cursor, PrimaryFilter};
 #[cfg(any(feature = "sqlite", test))]
 pub use self::sqlite::SQLite;
 
@@ -68,48 +71,6 @@ impl TryFrom<u32> for ReceivedPaymentStatus {
     }
 }
 
-/// IDs to query transactions
-#[derive(Debug, Clone)]
-pub enum FilterId {
-    /// By transaction ID with the latest revision
-    LatestRevision(TxId),
-    /// A transaction at a given revision
-    RevisionId(RevId),
-}
-
-impl From<TxId> for FilterId {
-    fn from(id: TxId) -> Self {
-        FilterId::LatestRevision(id)
-    }
-}
-
-impl From<&TxId> for FilterId {
-    fn from(id: &TxId) -> Self {
-        FilterId::LatestRevision(id.clone())
-    }
-}
-
-impl From<RevId> for FilterId {
-    fn from(id: RevId) -> Self {
-        FilterId::RevisionId(id)
-    }
-}
-
-impl From<&RevId> for FilterId {
-    fn from(id: &RevId) -> Self {
-        FilterId::RevisionId(id.clone())
-    }
-}
-
-impl ToString for FilterId {
-    fn to_string(&self) -> String {
-        match self {
-            FilterId::LatestRevision(id) => id.to_string(),
-            FilterId::RevisionId(id) => id.to_string(),
-        }
-    }
-}
-
 #[derive(thiserror::Error, Debug, Serialize)]
 /// Storage error
 pub enum Error {
@@ -235,7 +196,8 @@ pub trait Batch<'a> {
     async fn tag_transaction(
         &mut self,
         transaction_id: &TxId,
-        tags: &[String],
+        base_tx: &BaseTx,
+        tags: &[Tag],
     ) -> Result<(), Error>;
 
     /// Creates a relationship between an account and a transaction.
@@ -327,63 +289,20 @@ pub trait Storage {
         Ok(payments)
     }
 
-    /*
-    ///
-    async fn get_revision(&self, revision_id: &TxId) -> Result<Transaction, Error>;
-
-    ///
-    async fn get_all_revisions(
-        &self,
-        transaction_id: &TxId,
-    ) -> Result<Vec<(TxId, String, DateTime<Utc>)>, Error>;
-    */
-
-    /// Returns a transaction by id, the transaction is returned as a tuple of base transaction and
-    /// the current revision
-    async fn get_transaction_and_revision(
+    /// Find transactions with a given filter
+    async fn find_base_tx_and_revision(
         &self,
-        transaction_id: FilterId,
-    ) -> Result<(BaseTx, Revision), Error>;
+        mut filter: Filter,
+    ) -> Result<Vec<(BaseTx, Revision)>, Error>;
 
     /// List all revisions for a given transaction
     async fn get_revisions(&self, transaction_id: &TxId) -> Result<Vec<RevId>, Error>;
 
-    /// Returns a list of a transactions for a given account (and optionally
-    /// filter by types). The list of transactions is expected to be sorted by
-    /// date desc (newest transactions first),
-    async fn get_transactions_with_latest_revision(
-        &self,
-        account: &AccountId,
-        types: &[Type],
-        tags: &[String],
-    ) -> Result<Vec<(BaseTx, Revision)>, Error>;
-
     /// Returns a revision with a transaction object by id
-    async fn get_transaction(&self, id: FilterId) -> Result<Transaction, Error> {
-        let (base_tx, revision) = self.get_transaction_and_revision(id).await?;
-        (
-            self.get_revisions(&revision.transaction_id).await?,
-            base_tx,
-            revision,
-        )
-            .try_into()
-            .map_err(|e: crate::transaction::Error| Error::Encoding(e.to_string()))
-    }
-
-    /// Get list of transactions
-    async fn get_transactions(
-        &self,
-        account: &AccountId,
-        types: &[Type],
-        tags: &[String],
-    ) -> Result<Vec<Transaction>, Error> {
-        let mut transactions = Vec::new();
-        for (base_tx, revision) in self
-            .get_transactions_with_latest_revision(account, types, tags)
-            .await?
-            .into_iter()
-        {
-            transactions.push(
+    async fn find(&self, filter: Filter) -> Result<Vec<Transaction>, Error> {
+        let mut results = vec![];
+        for (base_tx, revision) in self.find_base_tx_and_revision(filter).await?.into_iter() {
+            results.push(
                 (
                     self.get_revisions(&revision.transaction_id).await?,
                     base_tx,
@@ -393,7 +312,8 @@ pub trait Storage {
                     .map_err(|e: crate::transaction::Error| Error::Encoding(e.to_string()))?,
             );
         }
-        Ok(transactions)
+
+        Ok(results)
     }
 }
 
@@ -402,7 +322,7 @@ pub mod test {
     use std::collections::HashMap;
 
     use super::*;
-    use crate::{config::Config, status::StatusManager, Transaction};
+    use crate::{config::Config, status::StatusManager, Ledger, Transaction};
     use rand::Rng;
 
     #[macro_export]
@@ -427,6 +347,7 @@ pub mod test {
             $crate::storage_unit_test!(does_not_spend_unspendable_payments);
             $crate::storage_unit_test!(spend_spendable_payments);
             $crate::storage_unit_test!(relate_account_to_transaction);
+            $crate::storage_unit_test!(find_transactions_by_tags);
             $crate::storage_unit_test!(not_spendable_new_payments_not_spendable);
         };
     }
@@ -501,7 +422,11 @@ pub mod test {
             .await
             .expect("update tx");
         storing.rollback().await.expect("rollback");
-        assert!(storage.get_transaction((&deposit.id).into()).await.is_err());
+        assert!(storage
+            .find(deposit.id.clone().into())
+            .await
+            .expect("result")
+            .is_empty());
 
         let mut storing = storage.begin().await.expect("valid tx");
         storing
@@ -521,7 +446,10 @@ pub mod test {
             .await
             .expect("update tx");
         storing.commit().await.expect("commit");
-        assert!(storage.get_transaction((&deposit.id).into()).await.is_ok());
+        assert_eq!(
+            storage.find(deposit.id.into()).await.expect("result").len(),
+            1
+        );
     }
 
     pub async fn transaction_not_available_until_commit<T>(storage: T)
@@ -555,9 +483,17 @@ pub mod test {
             )
             .await
             .expect("update tx");
-        assert!(storage.get_transaction((&deposit.id).into()).await.is_err());
+        assert!(storage
+            .find(deposit.id.clone().into())
+            .await
+            .expect("results")
+            .is_empty());
         storing.rollback().await.expect("rollback");
-        assert!(storage.get_transaction((&deposit.id).into()).await.is_err());
+        assert!(storage
+            .find(deposit.id.clone().into())
+            .await
+            .expect("results")
+            .is_empty());
 
         let mut storing = storage.begin().await.expect("valid tx");
         storing
@@ -577,7 +513,14 @@ pub mod test {
             .await
             .expect("update tx");
         storing.commit().await.expect("commit");
-        assert!(storage.get_transaction((&deposit.id).into()).await.is_ok());
+        assert_eq!(
+            1,
+            storage
+                .find(deposit.id.into())
+                .await
+                .expect("results")
+                .len()
+        );
     }
 
     pub async fn does_not_update_spent_payments<T>(storage: T)
@@ -648,6 +591,94 @@ pub mod test {
         }
     }
 
+    pub async fn find_transactions_by_tags<T>(storage: T)
+    where
+        T: Storage + Send + Sync,
+    {
+        let ledger = Ledger::new(Config {
+            storage,
+            status: Default::default(),
+        });
+
+        for i in 0..10 {
+            let usd: Asset = "USD/2".parse().expect("valid asset");
+            let account = format!("account-{}", i).parse().expect("valid account");
+
+            let deposit = ledger
+                .deposit(
+                    &account,
+                    usd.from_human("100.99").expect("valid amount"),
+                    "settled".into(),
+                    format!("test deposit {}", i),
+                )
+                .await
+                .expect("valid deposit");
+
+            if i % 2 == 0 {
+                ledger
+                    .set_tags(
+                        deposit.revision_id,
+                        vec![
+                            "even".parse().expect("valid tag"),
+                            "even".parse().expect("valid tag"),
+                            "all".parse().expect("valid tag"),
+                        ],
+                        "add tags".to_owned(),
+                    )
+                    .await
+                    .expect("tag tx");
+            } else {
+                ledger
+                    .set_tags(
+                        deposit.revision_id,
+                        vec![
+                            "odd".parse().expect("valid tag"),
+                            "all".parse().expect("valid tag"),
+                        ],
+                        "add tags".to_owned(),
+                    )
+                    .await
+                    .expect("tag tx");
+            }
+        }
+
+        assert_eq!(
+            5,
+            ledger
+                .get_transactions(Filter {
+                    tags: vec!["even".parse().expect("valid tag")],
+                    ..Default::default()
+                })
+                .await
+                .expect("valid filter")
+                .len()
+        );
+
+        assert_eq!(
+            5,
+            ledger
+                .get_transactions(Filter {
+                    tags: vec!["odd".parse().expect("valid tag")],
+                    ..Default::default()
+                })
+                .await
+                .expect("valid filter")
+                .len()
+        );
+
+        assert_eq!(
+            10,
+            ledger
+                .get_transactions(Filter {
+                    tags: vec!["all".parse().expect("valid tag")],
+                    ..Default::default()
+                })
+                .await
+                .expect("valid filter")
+                .len()
+        );
+    }
+
     pub async fn spend_spendable_payments<T>(storage: T)
     where
         T: Storage + Send + Sync,
@@ -877,10 +908,15 @@ pub mod test {
 
         batch.commit().await.expect("valid commit");
 
+        let filter = Filter {
+            typ: vec![Type::Deposit],
+            ..Default::default()
+        };
+
         assert_eq!(
             1,
             storage
-                .get_transactions(&account1, &[Type::Deposit], &[])
+                .find(filter.clone().account(account1.clone()))
                 .await
                 .expect("valid tx")
                 .len()
@@ -888,7 +924,7 @@ pub mod test {
         assert_eq!(
             1,
             storage
-                .get_transactions(&account2, &[Type::Deposit], &[])
+                .find(filter.clone().account(account2.clone()))
                 .await
                 .expect("valid tx")
                 .len()
@@ -896,16 +932,20 @@ pub mod test {
         assert_eq!(
             0,
             storage
-                .get_transactions(&account3, &[Type::Deposit], &[])
+                .find(filter.account(account3.clone()))
                 .await
                 .expect("valid tx")
                 .len()
         );
-        for account in &[account1, account2, account3] {
+        for account in [account1, account2, account3] {
             assert_eq!(
                 0,
                 storage
-                    .get_transactions(account, &[Type::Withdrawal], &[])
+                    .find(Filter {
+                        typ: vec![Type::Withdrawal],
+                        accounts: vec![account],
+                        ..Default::default()
+                    })
                     .await
                     .expect("valid tx")
                     .len()

utxo/src/storage/sqlite/batch.rs

@@ -1,7 +1,7 @@
 use crate::{
     payment::PaymentTo,
     storage::{self, to_bytes, Error, ReceivedPaymentStatus},
-    AccountId, BaseTx, PaymentId, RevId, Revision, TxId, Type,
+    AccountId, BaseTx, PaymentId, RevId, Revision, Tag, TxId, Type,
 };
 use sqlx::{Row, Sqlite, Transaction as SqlxTransaction};
 use std::{marker::PhantomData, num::TryFromIntError};
@@ -309,9 +309,28 @@ impl<'a> storage::Batch<'a> for Batch<'a> {
 
     async fn tag_transaction(
         &mut self,
-        _transaction_id: &TxId,
-        _tags: &[String],
+        transaction_id: &TxId,
+        base_tx: &BaseTx,
+        tags: &[Tag],
     ) -> Result<(), Error> {
-        todo!()
+        sqlx::query(r#"DELETE FROM "transactions_by_tags" WHERE "transaction_id" = ? "#)
+            .bind(transaction_id.to_string())
+            .execute(&mut *self.inner)
+            .await
+            .map_err(|e| Error::Storage(e.to_string()))?;
+
+        for tag in tags {
+            sqlx::query(r#"INSERT INTO "transactions_by_tags"("transaction_id", "tag", "created_at") VALUES(?, ?, ?)
+                ON CONFLICT("transaction_id", "tag", "created_at") DO NOTHING
+                "#)
+                .bind(transaction_id.to_string())
+                .bind(tag.as_str())
+                .bind(base_tx.created_at.timestamp() as i64)
+                .execute(&mut *self.inner)
+                .await
+                .map_err(|e| Error::Storage(e.to_string()))?;
+        }
+
+        Ok(())
     }
 }

utxo/src/storage/sqlite/mod.rs

@@ -1,9 +1,10 @@
 //! SQLite storage layer for Verax
+use super::{Cursor, PrimaryFilter, ReceivedPaymentStatus};
 use crate::{
     amount::AmountCents,
-    storage::{Error, FilterId, Storage},
-    transaction::{Revision, Type},
-    AccountId, Amount, Asset, BaseTx, PaymentFrom, PaymentId, RevId, TxId,
+    storage::{Error, Storage},
+    transaction::Revision,
+    AccountId, Amount, Asset, BaseTx, Filter, PaymentFrom, PaymentId, RevId, TxId,
 };
 use borsh::from_slice;
 use futures::TryStreamExt;
@@ -15,13 +16,238 @@ mod batch;
 pub use batch::Batch;
 pub use sqlx::sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions};
 
-use super::ReceivedPaymentStatus;
-
 /// SQLite storage layer for Verax
 pub struct SQLite {
     db: sqlx::SqlitePool,
 }
 
+/// Receives a cursor and does the initial loading of transactions that may match the requested
+/// filter.
+///
+/// The function will load the full transaction of candidates and return them.
+#[allow(warnings)]
+async fn find_candidates<'e, 'c: 'e, E>(
+    executor: E,
+    cursor: &mut Cursor,
+    limit: usize,
+) -> Result<Vec<(BaseTx, Revision)>, Error>
+where
+    E: sqlx::Executor<'c, Database = sqlx::Sqlite>,
+{
+    let since = if let Some(since) = cursor.filter.since {
+        format!("AND created_at <= {}", since.timestamp())
+    } else {
+        "".to_owned()
+    };
+    let until = if let Some(until) = cursor.filter.until {
+        format!("AND created_at >= {}", until.timestamp())
+    } else {
+        "".to_owned()
+    };
+
+    let rows = match &cursor.primary_filter {
+        PrimaryFilter::TxId(ids) => {
+            let sql = format!(
+                r#"
+                    SELECT
+                        "bt"."blob",
+                        "b"."blob"
+                    FROM
+                        "transactions" as "t",
+                        "base_transactions" as "bt",
+                        "revisions" as "b"
+                    WHERE
+                        "t"."transaction_id" IN ({})
+                        AND "t"."revision_id" = "b"."revision_id"
+                        AND "t"."transaction_id" = "bt"."transaction_id"
+                        {} {}
+                    ORDER BY "t"."created_at" DESC
+                    LIMIT {} OFFSET {}
+                    "#,
+                "?,".repeat(ids.len()).trim_end_matches(","),
+                since,
+                until,
+                limit,
+                cursor.filter.skip
+            );
+            let mut query = sqlx::query(&sql);
+            for id in ids.iter() {
+                query = query.bind(id.to_string());
+            }
+            query
+                .bind(limit as i64)
+                .bind(cursor.filter.skip as i64)
+                .fetch_all(executor)
+                .await
+        }
+        PrimaryFilter::Revision(rev_ids) => {
+            let sql = format!(
+                r#"
+                    SELECT
+                        "bt"."blob",
+                        "b"."blob"
+                    FROM
+                        "base_transactions" as "bt",
+                        "revisions" as "b"
+                    WHERE
+                        "b"."revision_id" IN ({})
+                        AND "b"."transaction_id" = "bt"."transaction_id"
+                        {} {}
+                    ORDER BY "b"."created_at" DESC
+                    LIMIT {} OFFSET {}
+                    "#,
+                "?,".repeat(rev_ids.len()).trim_end_matches(","),
+                since,
+                until,
+                limit,
+                cursor.filter.skip
+            );
+            let mut query = sqlx::query(&sql);
+            for id in rev_ids.iter() {
+                query = query.bind(id.to_string());
+            }
+            query.fetch_all(executor).await
+        }
+        PrimaryFilter::Account(account_ids) => {
+            let sql = format!(
+                r#"
+                    SELECT
+                        "bt"."blob",
+                        "b"."blob"
+                    FROM
+                        "transaction_accounts" as "ta",
+                        "transactions" as "t",
+                        "base_transactions" as "bt",
+                        "revisions" as "b"
+                    WHERE
+                        "ta"."account_id" IN ({})
+                        AND "t"."transaction_id" = "ta"."transaction_id"
+                        AND "t"."revision_id" = "b"."revision_id"
+                        AND "t"."transaction_id" = "bt"."transaction_id"
+                        {} {}
+                    ORDER BY "ta"."created_at" DESC
+                    LIMIT {} OFFSET {}
+                    "#,
+                "?,".repeat(account_ids.len()).trim_end_matches(","),
+                since,
+                until,
+                limit,
+                cursor.filter.skip
+            );
+            let mut query = sqlx::query(&sql);
+            for id in account_ids.iter() {
+                query = query.bind(id.to_string());
+            }
+            query.fetch_all(executor).await
+        }
+        PrimaryFilter::Type(types) => {
+            let sql = format!(
+                r#"
+                    SELECT
+                        "bt"."blob",
+                        "b"."blob"
+                    FROM
+                        "transactions" as "t",
+                        "base_transactions" as "bt",
+                        "revisions" as "b"
+                    WHERE
+                        "bt"."type" IN ({})
+                        AND "t"."revision_id" = "b"."revision_id"
+                        AND "t"."transaction_id" = "bt"."transaction_id"
+                        {} {}
+                    ORDER BY "bt"."created_at" DESC
+                    LIMIT {} OFFSET {}
+                    "#,
+                "?,".repeat(types.len()).trim_end_matches(","),
+                since,
+                until,
+                limit,
+                cursor.filter.skip
+            );
+            let mut query = sqlx::query(&sql);
+            for id in types.iter() {
+                query = query.bind(id.to_string());
+            }
+            query.fetch_all(executor).await
+        }
+        PrimaryFilter::Tags(tags) => {
+            let sql = format!(
+                r#"
+                    SELECT
+                        "bt"."blob",
+                        "b"."blob"
+                    FROM
+                        "transactions" as "t",
+                        "base_transactions" as "bt",
+                        "revisions" as "b"
+                    WHERE
+                        "t"."transaction_id" IN (SELECT DISTINCT "transaction_id" FROM "transactions_by_tags" WHERE "tag" IN ({}) ORDER BY "created_at" DESC)
+                        AND "t"."revision_id" = "b"."revision_id"
+                        AND "t"."transaction_id" = "bt"."transaction_id"
+                        {} {}
+                    ORDER BY "t"."created_at" DESC
+                    LIMIT {} OFFSET {}
+                    "#,
+                "?,".repeat(tags.len()).trim_end_matches(","),
+                since,
+                until,
+                limit,
+                cursor.filter.skip
+            );
+
+            let mut query = sqlx::query(&sql);
+            for id in tags.iter() {
+                query = query.bind(id.to_string());
+            }
+            query.fetch_all(executor).await
+        }
+        PrimaryFilter::Stream => {
+            let sql = format!(
+                r#"
+                    SELECT
+                        "bt"."blob",
+                        "b"."blob"
+                    FROM
+                        "transactions" as "t",
+                        "base_transactions" as "bt",
+                        "revisions" as "b"
+                    WHERE
+                        "t"."revision_id" = "b"."revision_id"
+                        AND "t"."transaction_id" = "bt"."transaction_id"
+                        {} {}
+                    ORDER BY "t"."created_at" DESC
+                    LIMIT {} OFFSET {}
+                    "#,
+                since, until, limit, cursor.filter.skip
+            );
+            sqlx::query(&sql).fetch_all(executor).await
+        }
+    };
+
+    let results = rows
+        .map_err(|e| Error::Storage(e.to_string()))?
+        .into_iter()
+        .map(|row| {
+            let base_tx = row
+                .try_get::<Vec<u8>, usize>(0)
+                .map_err(|e| Error::Storage(e.to_string()))?;
+
+            let revision = row
+                .try_get::<Vec<u8>, usize>(1)
+                .map_err(|e| Error::Storage(e.to_string()))?;
+
+            Ok((
+                from_slice::<BaseTx>(&base_tx)?,
+                from_slice::<Revision>(&revision)?,
+            ))
+        })
+        .collect::<Result<Vec<_>, Error>>()?;
+
+    cursor.filter.skip += results.len();
+
+    Ok(results)
+}
+
 impl SQLite {
     /// Creates a new Verax SQLite storage layer. It takes an instance of the asset_manager
     pub fn new(db: sqlx::SqlitePool) -> Self {
@@ -59,6 +285,13 @@ impl SQLite {
             "blob" BINARY NOT NULL,
             "created_at" DATETIME DEFAULT CURRENT_TIMESTAMP
         );
+        CREATE TABLE IF NOT EXISTS "transactions_by_tags" (
+            "transaction_id" VARCHAR(67),
+            "tag" VARCHAR(250) NOT NULL,
+            "created_at" DATETIME DEFAULT CURRENT_TIMESTAMP,
+            PRIMARY KEY ("tag", "created_at", "transaction_id")
+        );
+        CREATE INDEX IF NOT EXISTS "transaction_id_and_tags" ON "transactions_by_tags" ("transaction_id");
         CREATE TABLE IF NOT EXISTS "payments" (
             "payment_id" VARCHAR(80) NOT NULL PRIMARY KEY,
             "to" VARCHAR(64) NOT NULL,
@@ -305,156 +538,66 @@ impl Storage for SQLite {
 
         Ok(sqlx::query(
             r#"
-            SELECT "revision_id" FROM "revisions" WHERE "transaction_id" = ? ORDER BY "created_at" ASC
+            SELECT
+                "revision_id"
+            FROM
+                "revisions"
+            WHERE
+                "transaction_id" = ?
+            ORDER BY "created_at" ASC
             "#,
         )
         .bind(transaction_id.to_string())
         .fetch_all(&mut *conn)
         .await
         .map_err(|e| Error::Storage(e.to_string()))?
-        .into_iter().map(|x| {
-              x.try_get::<String, usize>(0)
+        .into_iter()
+        .map(|x| {
+            x.try_get::<String, usize>(0)
                 .map(|x| RevId::from_str(&x))
                 .map_err(|e| Error::Storage(e.to_string()))
-        }).collect::<Result<Result<Vec<_>, _>, _>>()??)
+        })
+        .collect::<Result<Result<Vec<_>, _>, _>>()??)
     }
 
-    async fn get_transaction_and_revision(
+    async fn find_base_tx_and_revision(
         &self,
-        id: FilterId,
-    ) -> Result<(BaseTx, Revision), Error> {
+        filter: Filter,
+    ) -> Result<Vec<(BaseTx, Revision)>, Error> {
         let mut conn = self
             .db
             .acquire()
             .await
             .map_err(|e| Error::Storage(e.to_string()))?;
-        let row = sqlx::query(match id {
-            FilterId::LatestRevision(_) => {
-                r#"
-            SELECT
-                "bt"."blob",
-                "b"."blob"
-            FROM
-                "transactions" as "t",
-                "base_transactions" as "bt",
-                "revisions" as "b"
-            WHERE
-                "t"."transaction_id" = ?
-                AND "t"."revision_id" = "b"."revision_id"
-                AND "t"."transaction_id" = "bt"."transaction_id"
-        "#
-            }
-            FilterId::RevisionId(_) => {
-                r#"
-            SELECT
-                "bt"."blob",
-                "b"."blob"
-            FROM
-                "base_transactions" as "bt",
-                "revisions" as "b"
-            WHERE
-                "b"."revision_id" = ?
-                AND "b"."transaction_id" = "bt"."transaction_id"
-                "#
-            }
-        })
-        .bind(id.to_string())
-        .fetch_one(&mut *conn)
-        .await
-        .map_err(|e| Error::Storage(e.to_string()))?;
 
-        let transaction = row
-            .try_get::<Vec<u8>, usize>(0)
-            .map_err(|e| Error::Storage(e.to_string()))?;
+        let mut cursor: Cursor = filter.into();
 
-        let revision = row
-            .try_get::<Vec<u8>, usize>(1)
-            .map_err(|e| Error::Storage(e.to_string()))?;
+        let candidate_limits = 20;
+        let mut results = vec![];
 
-        Ok((
-            from_slice::<BaseTx>(&transaction)?,
-            from_slice::<Revision>(&revision)?,
-        ))
-    }
-
-    async fn get_transactions_with_latest_revision(
-        &self,
-        account: &AccountId,
-        types: &[Type],
-        _tags: &[String],
-    ) -> Result<Vec<(BaseTx, Revision)>, Error> {
-        let mut conn = self
-            .db
-            .acquire()
-            .await
-            .map_err(|e| Error::Storage(e.to_string()))?;
+        loop {
+            let candidates = find_candidates(&mut *conn, &mut cursor, candidate_limits).await?;
+            // if there are fewer candidates than requested it means there are no more resultset
+            let no_more_candidates = candidates.len() != candidate_limits;
 
-        let sql = if types.is_empty() {
-            r#"SELECT
-                "t"."transaction_id",
-                "t"."revision_id",
-                "bt"."blob",
-                "b"."blob"
-            FROM
-                "transaction_accounts" as "ta",
-                "base_transactions" as "bt",
-                "transactions" as "t",
-                "revisions" as "b"
-            WHERE
-                "ta"."account_id" = ?
-                AND "t"."transaction_id" = "ta"."transaction_id"
-                AND "t"."revision_id" = "b"."revision_id"
-                AND "t"."transaction_id" = "bt"."transaction_id"
-            ORDER BY "ta"."id" DESC"#
-                .to_owned()
-        } else {
-            let types = types
+            let mut iteration_result = candidates
                 .into_iter()
-                .map(|t| u32::from(t).to_string())
-                .collect::<Vec<_>>()
-                .join(",");
-            format!(
-                r#"SELECT
-                    "t"."transaction_id",
-                    "t"."revision_id",
-                    "bt"."blob",
-                    "b"."blob"
-                FROM
-                    "transaction_accounts" as "ta",
-                    "base_transactions" as "bt",
-                    "transactions" as "t",
-                    "revisions" as "b"
-                WHERE
-                    "account_id" = ?
-                    AND "t"."transaction_id" = "ta"."transaction_id"
-                    AND "t"."revision_id" = "b"."revision_id"
-                    AND "t"."transaction_id" = "bt"."transaction_id"
-                    AND "ta"."type" IN ({types})
-                ORDER BY "ta"."id" DESC"#,
-            )
-        };
-
-        sqlx::query(&sql)
-            .bind(account.to_string())
-            .fetch_all(&mut *conn)
-            .await
-            .map_err(|e| Error::Storage(e.to_string()))?
-            .into_iter()
-            .map(|row| {
-                let base_tx = row
-                    .try_get::<Vec<u8>, usize>(2)
-                    .map_err(|e| Error::Storage(e.to_string()))?;
-
-                let revision = row
-                    .try_get::<Vec<u8>, usize>(3)
-                    .map_err(|e| Error::Storage(e.to_string()))?;
-
-                Ok((
-                    from_slice::<BaseTx>(&base_tx)?,
-                    from_slice::<Revision>(&revision)?,
-                ))
-            })
-            .collect::<Result<Vec<_>, _>>()
+                .filter(|(base, revision)| cursor.matches(base, revision))
+                .collect::<Vec<_>>();
+
+            results.append(&mut iteration_result);
+
+            if no_more_candidates
+                || (cursor.filter.limit != 0 && cursor.filter.limit >= results.len())
+            {
+                break;
+            }
+        }
+
+        if cursor.filter.limit != 0 {
+            results.truncate(cursor.filter.limit);
+        }
+        Ok(results)
     }
 }
 

utxo/src/tests/deposit.rs

@@ -25,7 +25,7 @@ async fn pending_deposit_and_failure() {
         .is_empty());
 
     ledger
-        .change_status(&rev_id, "failed".into(), "failed due test".to_owned())
+        .change_status(rev_id, "failed".into(), "failed due test".to_owned())
         .await
         .expect("valid tx");
 
@@ -130,7 +130,7 @@ async fn balance_decreases_while_pending_spending_and_confirm() {
     assert!(ledger.get_balance(&fee).await.expect("balance").is_empty());
 
     ledger
-        .change_status(&rev_id, "settled".into(), "ready".to_owned())
+        .change_status(rev_id, "settled".into(), "ready".to_owned())
         .await
         .expect("valid tx");
 
@@ -187,7 +187,7 @@ async fn balance_decreases_while_pending_spending_and_cancel() {
     assert!(ledger.get_balance(&fee).await.expect("balance").is_empty());
 
     ledger
-        .change_status(&rev_id, "cancelled".into(), "cancelled by test".to_owned())
+        .change_status(rev_id, "cancelled".into(), "cancelled by test".to_owned())
         .await
         .expect("valid tx");
 
@@ -238,7 +238,7 @@ async fn balance_decreases_while_pending_spending_and_failed() {
     assert!(ledger.get_balance(&fee).await.expect("balance").is_empty());
 
     let new_rev_id = ledger
-        .change_status(&rev_id, "processing".into(), "processing now".to_owned())
+        .change_status(rev_id, "processing".into(), "processing now".to_owned())
         .await
         .expect("valid tx")
         .revision_id;
@@ -254,7 +254,7 @@ async fn balance_decreases_while_pending_spending_and_failed() {
         "Transaction: Error in status: Invalid transition from processing to cancelled".to_owned(),
         ledger
             .change_status(
-                &new_rev_id,
+                new_rev_id.clone(),
                 "cancelled".into(),
                 "cancelled by user".to_owned()
             )
@@ -271,7 +271,7 @@ async fn balance_decreases_while_pending_spending_and_failed() {
     assert!(ledger.get_balance(&fee).await.expect("balance").is_empty());
 
     ledger
-        .change_status(&new_rev_id, "failed".into(), "it has failed".to_owned())
+        .change_status(new_rev_id, "failed".into(), "it has failed".to_owned())
         .await
         .expect("valid");
 

utxo/src/tests/tx.rs

@@ -1,5 +1,5 @@
 use super::{deposit, get_asset_manager_and_ledger};
-use crate::{AccountId, Asset, Type};
+use crate::{AccountId, Asset, Filter, Type};
 
 #[tokio::test]
 async fn multi_account_transfers() {
@@ -39,18 +39,23 @@ async fn multi_account_transfers() {
         ledger.get_balance(&target).await.expect("balance")
     );
 
-    let rev_id = ledger
-        .get_transactions(&accounts[0], vec![Type::Exchange])
-        .await
-        .expect("valid")[0]
+    let filter = Filter {
+        accounts: vec![accounts[0].clone()],
+        typ: vec![Type::Exchange],
+        ..Default::default()
+    };
+
+    let rev_id = ledger.get_transactions(filter).await.expect("valid")[0]
         .revision_id
         .clone();
 
     for _ in &accounts {
-        let txs = ledger
-            .get_transactions(&accounts[0], vec![Type::Exchange])
-            .await
-            .expect("valid");
+        let filter = Filter {
+            accounts: vec![accounts[0].clone()],
+            typ: vec![Type::Exchange],
+            ..Default::default()
+        };
+        let txs = ledger.get_transactions(filter).await.expect("valid");
         assert_eq!(1, txs.len());
         assert_eq!(rev_id, txs[0].revision_id);
     }

utxo/src/tests/withdrawal.rs

@@ -198,7 +198,7 @@ async fn cancelled_withdrawal() {
     );
 
     ledger
-        .change_status(&rev_id, "cancelled".into(), "cancelled by test".to_owned())
+        .change_status(rev_id, "cancelled".into(), "cancelled by test".to_owned())
         .await
         .expect("valid tx");
 

utxo/src/transaction/mod.rs

@@ -3,7 +3,7 @@ use crate::{
     payment::PaymentTo,
     status::InternalStatus,
     storage::{Batch, ReceivedPaymentStatus, Storage},
-    AccountId, Amount, PaymentFrom, RevId, Status, TxId,
+    AccountId, Amount, MaxLengthString, PaymentFrom, RevId, Status, TxId,
 };
 use chrono::{DateTime, TimeZone, Utc};
 use serde::{Deserialize, Serialize};
@@ -16,6 +16,9 @@ mod typ;
 
 pub use self::{base_tx::BaseTx, error::Error, revision::Revision, typ::Type};
 
+/// Tag definition
+pub type Tag = MaxLengthString<64>;
+
 pub(crate) fn to_ts_microseconds<W: std::io::Write>(
     dt: &DateTime<Utc>,
     writer: &mut W,
@@ -172,7 +175,7 @@ impl Transaction {
     pub async fn set_tags<S>(
         self,
         config: &Config<S>,
-        new_tags: Vec<String>,
+        new_tags: Vec<Tag>,
         reason: String,
     ) -> Result<Self, Error>
     where
@@ -307,22 +310,24 @@ impl Transaction {
 
         let (created_updated, spent_updated) =
             match config.status.internal_type(&self.revision.status) {
-                InternalStatus::Reverted => batch
-                    .update_transaction_payments(
-                        &self.id,
-                        ReceivedPaymentStatus::Failed,
-                        ReceivedPaymentStatus::Spendable,
-                    )
-                    .await
-                    .expect("foo0"),
-                InternalStatus::Spendable => batch
-                    .update_transaction_payments(
-                        &self.id,
-                        ReceivedPaymentStatus::Spendable,
-                        ReceivedPaymentStatus::Spent,
-                    )
-                    .await
-                    .expect("foo1"),
+                InternalStatus::Reverted => {
+                    batch
+                        .update_transaction_payments(
+                            &self.id,
+                            ReceivedPaymentStatus::Failed,
+                            ReceivedPaymentStatus::Spendable,
+                        )
+                        .await?
+                }
+                InternalStatus::Spendable => {
+                    batch
+                        .update_transaction_payments(
+                            &self.id,
+                            ReceivedPaymentStatus::Spendable,
+                            ReceivedPaymentStatus::Spent,
+                        )
+                        .await?
+                }
                 _ => (self.creates.len(), self.spends.len()),
             };
 
@@ -337,8 +342,7 @@ impl Transaction {
                     ReceivedPaymentStatus::Spendable,
                     ReceivedPaymentStatus::Spent,
                 )
-                .await
-                .expect("foo2");
+                .await?;
         }
 
         batch
@@ -346,6 +350,10 @@ impl Transaction {
             .await?;
 
         batch
+            .tag_transaction(&self.id, &self.transaction, &self.revision.tags)
+            .await?;
+
+        batch
             .update_transaction_revision(
                 &self.id,
                 &self.revision_id,

utxo/src/transaction/revision.rs

@@ -1,4 +1,4 @@
-use crate::{transaction::Error, RevId, Status, TxId};
+use crate::{transaction::Error, RevId, Status, Tag, TxId};
 use chrono::{serde::ts_milliseconds, DateTime, Utc};
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
@@ -18,7 +18,7 @@ pub struct Revision {
     pub changelog: String,
 
     /// Vector of tags associated with a transaction
-    pub tags: Vec<String>,
+    pub tags: Vec<Tag>,
 
     /// Transaction status
     pub status: Status,
@@ -38,7 +38,7 @@ impl Revision {
         transaction_id: TxId,
         previous: Option<RevId>,
         changelog: String,
-        tags: Vec<String>,
+        tags: Vec<Tag>,
         status: Status,
     ) -> Self {
         Self {

utxo/src/transaction/typ.rs

@@ -11,6 +11,9 @@ pub enum Error {
     Clone,
     Copy,
     PartialEq,
+    PartialOrd,
+    Ord,
+    Eq,
     Serialize,
     Deserialize,
     borsh::BorshDeserialize,