Copy cardano tx & block range roots database from aggregator to persistence

In order to share them between the aggregator & signer.

Some changes were made:
* all business traits implementation were removed
* most of the business traits methods are still there, but
  directly on the repository and may have name and signatures
  changes (to return the record instead of the business type most
  of the time).
* "get_all" are now public and not just test only as it would be
  cumbersome to add a 'test_only' feature flags just for that.

Author: Alenar

internal/mithril-persistence/src/database/cardano_transaction_migration.rs

@@ -0,0 +1,90 @@
+//! Migration module for cardano transactions store
+//!
+use crate::database::SqlMigration;
+
+/// Get all the migrations required by this version of the software.
+/// There shall be one migration per database version. There could be several
+/// statements per migration.
+pub fn get_migrations() -> Vec<SqlMigration> {
+    vec![
+        // Migration 1
+        // Add the `cardano_tx` table.
+        SqlMigration::new(
+            1,
+            r#"
+create table cardano_tx (
+    transaction_hash        text      not null,
+    block_number            integer   not null,
+    immutable_file_number   integer   not null,
+    primary key (transaction_hash)
+);
+
+create unique index cardano_tx_immutable_file_number_index on cardano_tx(immutable_file_number);
+"#,
+        ),
+        // Migration 2
+        // Fix the `cardano_tx` table index on immutable_file number, incorrectly marked as unique.
+        SqlMigration::new(
+            2,
+            r#"
+-- remove all data from the cardano tx table since a lot of transactions where missing for each
+-- block and we rely on their insert order.
+delete from cardano_tx;
+
+drop index cardano_tx_immutable_file_number_index;
+create index cardano_tx_immutable_file_number_index on cardano_tx(immutable_file_number);
+
+vacuum;
+"#,
+        ),
+        // Migration 3
+        // Add `slot_number` and `block_hash` columns to `cardano_tx`.
+        SqlMigration::new(
+            3,
+            r#"
+-- remove all data from the cardano tx table since the new columns are mandatory
+delete from cardano_tx;
+
+alter table cardano_tx add column slot_number integer not null;
+alter table cardano_tx add column block_hash text not null;
+
+vacuum;
+        "#,
+        ),
+        // Migration 4
+        // Add index on `block_number` column of `cardano_tx` table
+        SqlMigration::new(
+            4,
+            r#"
+create index block_number_index on cardano_tx(block_number);
+"#,
+        ),
+        // Migration 5
+        // Add `block_range_root` table
+        SqlMigration::new(
+            5,
+            r#"
+create table block_range_root (
+    start         integer   not null,
+    end           integer   not null,
+    merkle_root   text      not null,
+    primary key (start, end)
+);
+"#,
+        ),
+        // Migration 6
+        // Add composite index on `block_number/transaction_hash` column of `cardano_tx` table
+        // Truncate `block_range_root` table after changing the order of retrieval of the transactions
+        SqlMigration::new(
+            6,
+            r#"
+create index block_number_transaction_hash_index on cardano_tx(block_number, transaction_hash);
+
+-- remove all data from the block_range_root table since the order used to create them has changed
+delete from block_range_root;
+
+vacuum;
+"#,
+        ),
+    ]
+}

internal/mithril-persistence/src/database/mod.rs

@@ -1,13 +1,35 @@
 //! database module.
 //! This module contains providers and entities shared between all application types.
 
+pub mod cardano_transaction_migration;
 mod db_version;
+pub(crate) mod provider;
+pub mod record;
+pub mod repository;
 mod signed_entity_hydrator;
 mod version_checker;
 
-/// Database version.
-pub type DbVersion = i64;
-
 pub use db_version::*;
 pub use signed_entity_hydrator::SignedEntityTypeHydrator;
 pub use version_checker::{DatabaseVersionChecker, SqlMigration};
+
+/// Database version.
+pub type DbVersion = i64;
+
+#[cfg(test)]
+pub mod test_helper {
+    use sqlite::ConnectionThreadSafe;
+
+    use mithril_common::StdResult;
+
+    use crate::sqlite::{ConnectionBuilder, ConnectionOptions};
+
+    /// In-memory sqlite database without foreign key support with cardano db migrations applied
+    pub fn cardano_tx_db_connection() -> StdResult<ConnectionThreadSafe> {
+        let connection = ConnectionBuilder::open_memory()
+            .with_options(&[ConnectionOptions::ForceDisableForeignKeys])
+            .with_migrations(crate::database::cardano_transaction_migration::get_migrations())
+            .build()?;
+        Ok(connection)
+    }
+}

internal/mithril-persistence/src/database/provider/block_range_root/get_block_range_root.rs

@@ -0,0 +1,38 @@
+use mithril_common::entities::BlockNumber;
+use sqlite::Value;
+
+use crate::database::record::BlockRangeRootRecord;
+use crate::sqlite::{Provider, SourceAlias, SqLiteEntity, SqliteConnection, WhereCondition};
+
+/// Simple queries to retrieve [BlockRangeRootRecord] from the sqlite database.
+pub struct GetBlockRangeRootProvider<'client> {
+    connection: &'client SqliteConnection,
+}
+
+impl<'client> GetBlockRangeRootProvider<'client> {
+    /// Create a new instance
+    pub fn new(connection: &'client SqliteConnection) -> Self {
+        Self { connection }
+    }
+
+    pub fn get_up_to_block_number_condition(&self, block_number: BlockNumber) -> WhereCondition {
+        WhereCondition::new("end < ?*", vec![Value::Integer(block_number as i64)])
+    }
+}
+
+impl crate::sqlite::GetAllCondition for GetBlockRangeRootProvider<'_> {}
+
+impl<'client> Provider<'client> for GetBlockRangeRootProvider<'client> {
+    type Entity = BlockRangeRootRecord;
+
+    fn get_connection(&'client self) -> &'client SqliteConnection {
+        self.connection
+    }
+
+    fn get_definition(&self, condition: &str) -> String {
+        let aliases = SourceAlias::new(&[("{:block_range_root:}", "block_range_root")]);
+        let projection = Self::Entity::get_projection().expand(aliases);
+
+        format!("select {projection} from block_range_root where {condition} order by start, end")
+    }
+}

internal/mithril-persistence/src/database/provider/block_range_root/get_interval_without_block_range_provider.rs

@@ -0,0 +1,40 @@
+use crate::database::record::IntervalWithoutBlockRangeRootRecord;
+use crate::sqlite::{Provider, SourceAlias, SqLiteEntity, SqliteConnection, WhereCondition};
+
+/// Query that return the interval of block numbers that does not have a block range root.
+pub struct GetIntervalWithoutBlockRangeRootProvider<'client> {
+    connection: &'client SqliteConnection,
+}
+
+impl<'client> GetIntervalWithoutBlockRangeRootProvider<'client> {
+    /// Create a new instance
+    pub fn new(connection: &'client SqliteConnection) -> Self {
+        Self { connection }
+    }
+
+    pub fn get_interval_without_block_range_condition(&self) -> WhereCondition {
+        WhereCondition::default()
+    }
+}
+
+impl<'client> Provider<'client> for GetIntervalWithoutBlockRangeRootProvider<'client> {
+    type Entity = IntervalWithoutBlockRangeRootRecord;
+
+    fn get_connection(&'client self) -> &'client SqliteConnection {
+        self.connection
+    }
+
+    fn get_definition(&self, condition: &str) -> String {
+        let aliases = SourceAlias::new(&[
+            ("{:interval_start:}", "interval_start"),
+            ("{:interval_end:}", "interval_end"),
+        ]);
+        let projection = Self::Entity::get_projection().expand(aliases);
+
+        format!(
+            "select {projection} from \
+        (select max(end) as start from block_range_root where {condition}) as interval_start, \
+        (select max(block_number) as end from cardano_tx where {condition}) as interval_end"
+        )
+    }
+}

internal/mithril-persistence/src/database/provider/block_range_root/insert_block_range.rs

@@ -0,0 +1,63 @@
+use std::iter::repeat;
+
+use sqlite::Value;
+
+use mithril_common::StdResult;
+
+use crate::database::record::BlockRangeRootRecord;
+use crate::sqlite::{Provider, SourceAlias, SqLiteEntity, SqliteConnection, WhereCondition};
+
+/// Query to insert [BlockRangeRootRecord] in the sqlite database
+pub struct InsertBlockRangeRootProvider<'client> {
+    connection: &'client SqliteConnection,
+}
+
+impl<'client> InsertBlockRangeRootProvider<'client> {
+    /// Create a new instance
+    pub fn new(connection: &'client SqliteConnection) -> Self {
+        Self { connection }
+    }
+
+    /// Condition to insert multiples records.
+    pub fn get_insert_many_condition(
+        &self,
+        block_range_records: Vec<BlockRangeRootRecord>,
+    ) -> StdResult<WhereCondition> {
+        let columns = "(start, end, merkle_root)";
+        let values_columns: Vec<&str> = repeat("(?*, ?*, ?*)")
+            .take(block_range_records.len())
+            .collect();
+
+        let values: StdResult<Vec<Value>> =
+            block_range_records
+                .into_iter()
+                .try_fold(vec![], |mut vec, record| {
+                    vec.append(&mut vec![
+                        Value::Integer(record.range.start.try_into()?),
+                        Value::Integer(record.range.end.try_into()?),
+                        Value::String(record.merkle_root.to_hex()),
+                    ]);
+                    Ok(vec)
+                });
+
+        Ok(WhereCondition::new(
+            format!("{columns} values {}", values_columns.join(", ")).as_str(),
+            values?,
+        ))
+    }
+}
+
+impl<'client> Provider<'client> for InsertBlockRangeRootProvider<'client> {
+    type Entity = BlockRangeRootRecord;
+
+    fn get_connection(&'client self) -> &'client SqliteConnection {
+        self.connection
+    }
+
+    fn get_definition(&self, condition: &str) -> String {
+        let aliases = SourceAlias::new(&[("{:block_range_root:}", "block_range_root")]);
+        let projection = Self::Entity::get_projection().expand(aliases);
+
+        format!("insert or ignore into block_range_root {condition} returning {projection}")
+    }
+}

internal/mithril-persistence/src/database/provider/block_range_root/mod.rs

@@ -0,0 +1,7 @@
+mod get_block_range_root;
+mod get_interval_without_block_range_provider;
+mod insert_block_range;
+
+pub use get_block_range_root::*;
+pub use get_interval_without_block_range_provider::*;
+pub use insert_block_range::*;

internal/mithril-persistence/src/database/provider/cardano_transaction/get_cardano_transaction.rs

@@ -0,0 +1,91 @@
+use std::ops::Range;
+
+use sqlite::Value;
+
+use mithril_common::entities::{BlockNumber, BlockRange, TransactionHash};
+
+use crate::database::record::CardanoTransactionRecord;
+use crate::sqlite::{
+    GetAllCondition, Provider, SourceAlias, SqLiteEntity, SqliteConnection, WhereCondition,
+};
+
+/// Simple queries to retrieve [CardanoTransaction] from the sqlite database.
+pub struct GetCardanoTransactionProvider<'client> {
+    connection: &'client SqliteConnection,
+}
+
+impl<'client> GetCardanoTransactionProvider<'client> {
+    /// Create a new instance
+    pub fn new(connection: &'client SqliteConnection) -> Self {
+        Self { connection }
+    }
+
+    // Useful in test and probably in the future.
+    pub fn get_transaction_hash_condition(
+        &self,
+        transaction_hash: &TransactionHash,
+    ) -> WhereCondition {
+        WhereCondition::new(
+            "transaction_hash = ?*",
+            vec![Value::String(transaction_hash.to_owned())],
+        )
+    }
+
+    pub fn get_transaction_hashes_condition(
+        &self,
+        transactions_hashes: Vec<TransactionHash>,
+    ) -> WhereCondition {
+        let hashes_values = transactions_hashes.into_iter().map(Value::String).collect();
+
+        WhereCondition::where_in("transaction_hash", hashes_values)
+    }
+
+    pub fn get_transaction_block_ranges_condition(
+        &self,
+        block_ranges: Vec<BlockRange>,
+    ) -> WhereCondition {
+        let mut where_condition = WhereCondition::default();
+        for block_range in block_ranges {
+            where_condition = where_condition.or_where(WhereCondition::new(
+                "(block_number >= ?* and block_number < ?*)",
+                vec![
+                    Value::Integer(block_range.start as i64),
+                    Value::Integer(block_range.end as i64),
+                ],
+            ))
+        }
+
+        where_condition
+    }
+
+    pub fn get_transaction_between_blocks_condition(
+        &self,
+        range: Range<BlockNumber>,
+    ) -> WhereCondition {
+        WhereCondition::new(
+            "block_number >= ?*",
+            vec![Value::Integer(range.start as i64)],
+        )
+        .and_where(WhereCondition::new(
+            "block_number < ?*",
+            vec![Value::Integer(range.end as i64)],
+        ))
+    }
+}
+
+impl<'client> Provider<'client> for GetCardanoTransactionProvider<'client> {
+    type Entity = CardanoTransactionRecord;
+
+    fn get_connection(&'client self) -> &'client SqliteConnection {
+        self.connection
+    }
+
+    fn get_definition(&self, condition: &str) -> String {
+        let aliases = SourceAlias::new(&[("{:cardano_tx:}", "cardano_tx")]);
+        let projection = Self::Entity::get_projection().expand(aliases);
+
+        format!("select {projection} from cardano_tx where {condition} order by block_number, transaction_hash")
+    }
+}
+
+impl GetAllCondition for GetCardanoTransactionProvider<'_> {}