mithril_persistence/database/repository/
cardano_transaction_repository.rs

1use std::ops::Range;
2use std::sync::Arc;
3
4use anyhow::Context;
5
6use mithril_common::StdResult;
7use mithril_common::crypto_helper::MKTreeNode;
8use mithril_common::entities::{
9    BlockHash, BlockNumber, BlockRange, CardanoBlockWithTransactions, CardanoTransaction,
10    ChainPoint, SlotNumber, TransactionHash,
11};
12
13use crate::database::query::{
14    DeleteBlockRangeRootQuery, DeleteCardanoBlockAndTransactionQuery,
15    DeleteLegacyBlockRangeRootQuery, GetBlockRangeRootQuery, GetCardanoBlockQuery,
16    GetCardanoBlockTransactionsQuery, GetCardanoTransactionQuery, GetLegacyBlockRangeRootQuery,
17    InsertBlockRangeRootQuery, InsertCardanoBlockQuery, InsertCardanoTransactionQuery,
18    InsertLegacyBlockRangeRootQuery,
19};
20use crate::database::record::{
21    BlockRangeRootRecord, CardanoBlockRecord, CardanoBlockTransactionsRecord,
22    CardanoTransactionRecord, IntoRecords, StorableCardanoTransactionRecord,
23};
24use crate::sqlite::{
25    ConnectionExtensions, OptimizeMode, SqliteCleaner, SqliteConnection, SqliteConnectionPool,
26};
27
28/// ## Cardano transaction repository
29///
30/// This is a business oriented layer to perform actions on the database through
31/// queries.
32pub struct CardanoTransactionRepository {
33    connection_pool: Arc<SqliteConnectionPool>,
34}
35
36impl CardanoTransactionRepository {
37    /// Instantiate service
38    pub fn new(connection_pool: Arc<SqliteConnectionPool>) -> Self {
39        Self { connection_pool }
40    }
41
42    /// Perform database optimization (see [SqliteCleaner::optimize] for the applied optimizations)
43    pub fn optimize(&self) -> StdResult<()> {
44        let connection = self.connection_pool.connection()?;
45        SqliteCleaner::optimize(&connection, OptimizeMode::Default)
46            .with_context(|| "Failed to optimize database")?;
47        self.connection_pool
48            .renew_connections()
49            .with_context(|| "Failed to renew connection pool")?;
50
51        Ok(())
52    }
53
54    /// Return all the [CardanoTransactionRecord]s in the database.
55    pub async fn get_all_transactions(&self) -> StdResult<Vec<CardanoTransactionRecord>> {
56        self.connection_pool
57            .connection()?
58            .fetch_collect(GetCardanoTransactionQuery::all())
59    }
60
61    /// Return all the [CardanoBlockRecord]s in the database.
62    pub async fn get_all_blocks(&self) -> StdResult<Vec<CardanoBlockRecord>> {
63        self.connection_pool
64            .connection()?
65            .fetch_collect(GetCardanoBlockQuery::all())
66    }
67
68    /// Return all the [CardanoTransactionRecord]s in the database where block number is in the
69    /// given range.
70    pub async fn get_transactions_in_range_blocks(
71        &self,
72        range: Range<BlockNumber>,
73    ) -> StdResult<Vec<CardanoTransactionRecord>> {
74        self.connection_pool
75            .connection()?
76            .fetch_collect(GetCardanoTransactionQuery::between_blocks(range))
77    }
78
79    /// Return all the [CardanoTransactionRecord]s in the database where block number is in the
80    /// given range.
81    pub async fn get_blocks_with_transactions_in_range_blocks(
82        &self,
83        range: Range<BlockNumber>,
84    ) -> StdResult<Vec<CardanoBlockTransactionsRecord>> {
85        self.connection_pool
86            .connection()?
87            .fetch_collect(GetCardanoBlockTransactionsQuery::between_blocks(range))
88    }
89
90    /// Return the [CardanoBlockRecord] for the given block hash.
91    pub async fn get_block<T: Into<BlockHash>>(
92        &self,
93        block_hash: T,
94    ) -> StdResult<Option<CardanoBlockRecord>> {
95        self.connection_pool
96            .connection()?
97            .fetch_first(GetCardanoBlockQuery::by_block_hash(&block_hash.into()))
98    }
99
100    /// Return the [CardanoTransactionRecord] for the given transaction hash.
101    pub async fn get_transaction<T: Into<TransactionHash>>(
102        &self,
103        transaction_hash: T,
104    ) -> StdResult<Option<CardanoTransactionRecord>> {
105        self.connection_pool.connection()?.fetch_first(
106            GetCardanoTransactionQuery::by_transaction_hash(transaction_hash),
107        )
108    }
109
110    /// Create new [CardanoTransactionRecord]s in the database.
111    pub async fn create_block_and_transactions(
112        &self,
113        blocks_with_transactions: Vec<CardanoBlockWithTransactions>,
114    ) -> StdResult<()> {
115        let connection = self.connection_pool.connection()?;
116
117        self.create_block_and_transactions_with_connection(&connection, blocks_with_transactions)
118            .await
119    }
120
121    /// Create new [CardanoBlock] [CardanoTransactionRecord]s in the database.
122    async fn create_block_and_transactions_with_connection(
123        &self,
124        connection: &SqliteConnection,
125        blocks_with_transactions: Vec<CardanoBlockWithTransactions>,
126    ) -> StdResult<()> {
127        if blocks_with_transactions.is_empty() {
128            return Ok(());
129        }
130        let (blocks_records, mut transactions_records) = blocks_with_transactions.into_records();
131        connection.apply(InsertCardanoBlockQuery::insert_many(blocks_records)?)?;
132
133        while !transactions_records.is_empty() {
134            let chunk: Vec<_> = transactions_records
135                .drain(
136                    ..transactions_records
137                        .len()
138                        .min(StorableCardanoTransactionRecord::MAX_PER_INSERT),
139                )
140                .collect();
141            connection.apply(InsertCardanoTransactionQuery::insert_many(chunk)?)?;
142        }
143
144        Ok(())
145    }
146
147    /// Create new [BlockRangeRootRecord]s in the database.
148    pub async fn create_block_range_roots<T: Into<BlockRangeRootRecord>>(
149        &self,
150        block_ranges: Vec<T>,
151    ) -> StdResult<Vec<BlockRangeRootRecord>> {
152        let records: Vec<BlockRangeRootRecord> =
153            block_ranges.into_iter().map(|tx| tx.into()).collect();
154        let connection = self.connection_pool.connection()?;
155
156        connection.fetch_collect(InsertBlockRangeRootQuery::insert_many(records)?)
157    }
158
159    /// Create new legacy [BlockRangeRootRecord]s in the database.
160    pub async fn create_legacy_block_range_roots<T: Into<BlockRangeRootRecord>>(
161        &self,
162        block_ranges: Vec<T>,
163    ) -> StdResult<Vec<BlockRangeRootRecord>> {
164        let records: Vec<BlockRangeRootRecord> =
165            block_ranges.into_iter().map(|tx| tx.into()).collect();
166        let connection = self.connection_pool.connection()?;
167
168        connection.fetch_collect(InsertLegacyBlockRangeRootQuery::insert_many(records)?)
169    }
170
171    /// Get the highest [ChainPoint] of the cardano transactions stored in the database.
172    pub async fn get_transaction_highest_chain_point(&self) -> StdResult<Option<ChainPoint>> {
173        let first_transaction_with_highest_block_number = self
174            .connection_pool
175            .connection()?
176            .fetch_first(GetCardanoBlockQuery::with_highest_block_number())?;
177
178        Ok(first_transaction_with_highest_block_number.map(|record| {
179            ChainPoint::new(record.slot_number, record.block_number, record.block_hash)
180        }))
181    }
182
183    /// Get the threshold [BlockNumber] to prune the blocks and transactions below.
184    ///
185    /// To ensure that all new and legacy blocks range roots were computed and stored: This threshold
186    /// is the minimum of the highest block number of the block range roots and the highest block
187    /// number of the legacy block range roots.
188    pub async fn get_prune_blocks_threshold(&self) -> StdResult<Option<BlockNumber>> {
189        let highest: Option<i64> = self.connection_pool.connection()?.query_single_cell(
190            r#"
191select coalesce(min(max_new.highest, max_legacy.highest), max_new.highest, max_legacy.highest)
192from (select max(start) as highest from block_range_root) max_new,
193     (select max(start) as highest from block_range_root_legacy) max_legacy;"#,
194            &[],
195        )?;
196        highest
197            .map(u64::try_from)
198            .transpose()
199            .map(|num| num.map(BlockNumber))
200            .with_context(||
201                format!("Integer field max(start) (value={highest:?}) is incompatible with u64 representation.")
202            )
203    }
204
205    /// Retrieve all the Block Range Roots in database for which their start number is below the given
206    /// block number.
207    pub async fn retrieve_block_range_roots_up_to(
208        &self,
209        block_number: BlockNumber,
210    ) -> StdResult<Box<dyn Iterator<Item = (BlockRange, MKTreeNode)> + '_>> {
211        let block_range_roots = self
212            .connection_pool
213            .connection()?
214            .fetch(GetBlockRangeRootQuery::contains_or_below_block_number(
215                block_number,
216            ))?
217            .map(|record| -> (BlockRange, MKTreeNode) { record.into() })
218            .collect::<Vec<_>>(); // TODO: remove this collect to return the iterator directly
219
220        Ok(Box::new(block_range_roots.into_iter()))
221    }
222
223    /// Retrieve all the legacy Block Range Roots in database for which their start number is below the given
224    /// block number.
225    pub async fn retrieve_legacy_block_range_roots_up_to(
226        &self,
227        block_number: BlockNumber,
228    ) -> StdResult<Box<dyn Iterator<Item = (BlockRange, MKTreeNode)> + '_>> {
229        let block_range_roots = self
230            .connection_pool
231            .connection()?
232            .fetch(GetLegacyBlockRangeRootQuery::contains_or_below_block_number(block_number))?
233            .map(|record| -> (BlockRange, MKTreeNode) { record.into() })
234            .collect::<Vec<_>>(); // TODO: remove this collect to return the iterator directly
235
236        Ok(Box::new(block_range_roots.into_iter()))
237    }
238
239    /// Retrieve the block range root with the highest bounds in the database.
240    pub async fn retrieve_highest_block_range_root(
241        &self,
242    ) -> StdResult<Option<BlockRangeRootRecord>> {
243        self.connection_pool
244            .connection()?
245            .fetch_first(GetBlockRangeRootQuery::highest())
246    }
247
248    /// Retrieve the legacy block range root with the highest bounds in the database.
249    pub async fn retrieve_highest_legacy_block_range_root(
250        &self,
251    ) -> StdResult<Option<BlockRangeRootRecord>> {
252        self.connection_pool
253            .connection()?
254            .fetch_first(GetLegacyBlockRangeRootQuery::highest())
255    }
256
257    /// Retrieve all the [CardanoTransaction] in database.
258    pub async fn get_all(&self) -> StdResult<Vec<CardanoTransaction>> {
259        let records = self
260            .connection_pool
261            .connection()?
262            .fetch(GetCardanoTransactionQuery::all())?
263            .map(|record| record.into())
264            .collect();
265
266        Ok(records)
267    }
268
269    /// Retrieve all the [BlockRangeRootRecord] in database.
270    pub fn get_all_block_range_root(&self) -> StdResult<Vec<BlockRangeRootRecord>> {
271        self.connection_pool
272            .connection()?
273            .fetch_collect(GetBlockRangeRootQuery::all())
274    }
275
276    /// Retrieve all the legacy [BlockRangeRootRecord] in database.
277    pub fn get_all_legacy_block_range_root(&self) -> StdResult<Vec<BlockRangeRootRecord>> {
278        self.connection_pool
279            .connection()?
280            .fetch_collect(GetLegacyBlockRangeRootQuery::all())
281    }
282
283    /// Store the given transactions in the database.
284    ///
285    /// The storage is done in chunks to avoid exceeding sqlite binding limitations.
286    pub async fn store_blocks_and_transactions(
287        &self,
288        blocks_with_transactions: Vec<CardanoBlockWithTransactions>,
289    ) -> StdResult<()> {
290        let mut remaining_blocks = blocks_with_transactions;
291
292        while !remaining_blocks.is_empty() {
293            let connection = self.connection_pool.connection()?;
294            let transaction = connection.begin_transaction()?;
295
296            let chunk: Vec<_> = remaining_blocks
297                .drain(..remaining_blocks.len().min(CardanoBlockRecord::MAX_PER_INSERT))
298                .collect();
299
300            self.create_block_and_transactions_with_connection(&connection, chunk)
301                .await
302                .with_context(
303                    || "CardanoTransactionRepository can not store blocks and transactions",
304                )?;
305
306            transaction.commit()?;
307        }
308
309        Ok(())
310    }
311
312    /// Get the closest block number above a given slot number
313    pub async fn get_closest_block_number_above_slot_number(
314        &self,
315        slot_number: SlotNumber,
316    ) -> StdResult<Option<BlockNumber>> {
317        let query = GetCardanoBlockQuery::with_highest_block_number_below_slot_number(slot_number);
318        let record = self.connection_pool.connection()?.fetch_first(query)?;
319
320        Ok(record.map(|r| r.block_number))
321    }
322
323    /// Get the [CardanoTransactionRecord] for the given transaction hashes, up to a block number
324    pub async fn get_transaction_by_hashes<T: Into<TransactionHash>>(
325        &self,
326        hashes: Vec<T>,
327        up_to: BlockNumber,
328    ) -> StdResult<Vec<CardanoTransactionRecord>> {
329        let query = GetCardanoTransactionQuery::by_transaction_hashes(
330            hashes.into_iter().map(Into::into).collect(),
331            up_to,
332        );
333        self.connection_pool.connection()?.fetch_collect(query)
334    }
335
336    /// Get the [CardanoTransactionRecord] for the given block ranges.
337    pub async fn get_transaction_by_block_ranges(
338        &self,
339        block_ranges: Vec<BlockRange>,
340    ) -> StdResult<Vec<CardanoTransactionRecord>> {
341        let mut transactions = vec![];
342        // Make one query per block range to optimize throughput as asking multiple block ranges at once
343        // made SQLite quickly collapse (see PR #1723)
344        for block_range in block_ranges {
345            let block_range_transactions: Vec<CardanoTransactionRecord> = self
346                .connection_pool
347                .connection()?
348                .fetch_collect(GetCardanoTransactionQuery::between_blocks(block_range))?;
349            transactions.extend(block_range_transactions);
350        }
351
352        Ok(transactions)
353    }
354
355    /// Get the [CardanoBlockTransactionsRecord] for the given block ranges.
356    pub async fn get_blocks_with_transactions_by_block_ranges(
357        &self,
358        block_ranges: Vec<BlockRange>,
359    ) -> StdResult<Vec<CardanoBlockTransactionsRecord>> {
360        let mut blocks_with_transactions = vec![];
361        // Make one query per block range to optimize throughput as asking multiple block ranges at once
362        // made SQLite quickly collapse (see PR #1723)
363        for block_range in block_ranges {
364            let block_range_transactions: Vec<CardanoBlockTransactionsRecord> =
365                self.connection_pool.connection()?.fetch_collect(
366                    GetCardanoBlockTransactionsQuery::between_blocks(block_range),
367                )?;
368            blocks_with_transactions.extend(block_range_transactions);
369        }
370
371        Ok(blocks_with_transactions)
372    }
373
374    /// Prune the transactions older than the given number of blocks (based on the block range root
375    /// stored).
376    pub async fn prune_transaction(&self, number_of_blocks_to_keep: BlockNumber) -> StdResult<()> {
377        if let Some(highest_block_range_start) = self.get_prune_blocks_threshold().await? {
378            let threshold = highest_block_range_start - number_of_blocks_to_keep;
379            let query =
380                DeleteCardanoBlockAndTransactionQuery::below_block_number_threshold(threshold)?;
381
382            let connection = self.connection_pool.connection()?;
383            connection.fetch_first(query)?;
384        }
385
386        Ok(())
387    }
388
389    /// Remove blocks, transactions, and block range roots that are in a rolled-back fork
390    ///
391    /// * Remove blocks and transactions with a block number strictly greater than the given block number
392    /// * Remove block range roots that have lower bound range strictly above the given block number
393    pub async fn remove_rolled_back_transactions_and_block_range_by_block_number(
394        &self,
395        block_number: BlockNumber,
396    ) -> StdResult<()> {
397        let connection = self.connection_pool.connection()?;
398        let transaction = connection.begin_transaction()?;
399
400        connection.fetch_first(
401            DeleteCardanoBlockAndTransactionQuery::above_block_number_threshold(block_number)?,
402        )?;
403        connection.fetch_first(
404            DeleteBlockRangeRootQuery::contains_or_above_block_number_threshold(block_number)?,
405        )?;
406        connection.fetch_first(
407            DeleteLegacyBlockRangeRootQuery::contains_or_above_block_number_threshold(
408                block_number,
409            )?,
410        )?;
411
412        transaction.commit()?;
413        Ok(())
414    }
415
416    /// Remove blocks, transactions, and block range roots that are in a rolled-back fork
417    ///
418    /// * Remove blocks and transactions with the closest block number strictly greater than the given slot number if it exists
419    /// * Remove block range roots that have lower bound range strictly above the aforementioned block number
420    pub async fn remove_rolled_back_blocks_transactions_and_block_range_by_slot_number(
421        &self,
422        slot_number: SlotNumber,
423    ) -> StdResult<()> {
424        if let Some(block_number) =
425            self.get_closest_block_number_above_slot_number(slot_number).await?
426        {
427            self.remove_rolled_back_transactions_and_block_range_by_block_number(block_number)
428                .await?;
429        }
430
431        Ok(())
432    }
433}
434
435#[cfg(test)]
436mod tests {
437    use mithril_common::temp_dir_create;
438
439    use crate::database::query::GetLegacyBlockRangeRootQuery;
440    use crate::database::test_helper::cardano_tx_db_connection_builder;
441
442    use super::*;
443
444    #[tokio::test]
445    async fn repository_create_and_get_blocks_and_transactions() {
446        let temp_dir = temp_dir_create!();
447        let repository = CardanoTransactionRepository::new(Arc::new(
448            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
449        ));
450
451        repository
452            .create_block_and_transactions(vec![
453                CardanoBlockWithTransactions::new(
454                    "block_hash-123",
455                    BlockNumber(10),
456                    SlotNumber(50),
457                    vec!["tx_hash-123", "tx_hash-456"],
458                ),
459                CardanoBlockWithTransactions::new(
460                    "block_hash-789",
461                    BlockNumber(11),
462                    SlotNumber(51),
463                    vec!["tx_hash-789"],
464                ),
465            ])
466            .await
467            .unwrap();
468
469        {
470            let block_result = repository.get_block("block_hash-123").await.unwrap();
471            assert_eq!(
472                Some(CardanoBlockRecord {
473                    block_hash: "block_hash-123".to_string(),
474                    block_number: BlockNumber(10),
475                    slot_number: SlotNumber(50),
476                }),
477                block_result
478            );
479
480            let transaction_result = repository.get_transaction("tx_hash-123").await.unwrap();
481            assert_eq!(
482                Some(CardanoTransactionRecord {
483                    transaction_hash: "tx_hash-123".to_string(),
484                    block_number: BlockNumber(10),
485                    slot_number: SlotNumber(50),
486                    block_hash: "block_hash-123".to_string(),
487                }),
488                transaction_result
489            );
490        }
491        {
492            let transaction_result = repository.get_transaction("not-exist").await.unwrap();
493            assert_eq!(None, transaction_result);
494        }
495    }
496
497    #[tokio::test]
498    async fn repository_create_and_get_blocks_and_transactions_dont_fail_if_empty_or_no_transactions()
499     {
500        let temp_dir = temp_dir_create!();
501        let repository = CardanoTransactionRepository::new(Arc::new(
502            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
503        ));
504
505        repository.create_block_and_transactions(vec![]).await.unwrap();
506        repository
507            .create_block_and_transactions(vec![CardanoBlockWithTransactions::new(
508                "block_hash-10",
509                BlockNumber(10),
510                SlotNumber(50),
511                Vec::<String>::new(),
512            )])
513            .await
514            .unwrap();
515    }
516
517    #[tokio::test]
518    async fn repository_get_transaction_by_hashes() {
519        let temp_dir = temp_dir_create!();
520        let repository = CardanoTransactionRepository::new(Arc::new(
521            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
522        ));
523
524        repository
525            .create_block_and_transactions(vec![
526                CardanoBlockWithTransactions::new(
527                    "block_hash-10",
528                    BlockNumber(10),
529                    SlotNumber(50),
530                    vec!["tx_hash-123", "tx_hash-456", "tx_hash-789"],
531                ),
532                CardanoBlockWithTransactions::new(
533                    "block_hash-101",
534                    BlockNumber(101),
535                    SlotNumber(100),
536                    vec!["tx_hash-000"],
537                ),
538            ])
539            .await
540            .unwrap();
541
542        {
543            let transactions = repository
544                .get_transaction_by_hashes(vec!["tx_hash-123", "tx_hash-789"], BlockNumber(100))
545                .await
546                .unwrap();
547
548            assert_eq!(
549                vec![
550                    CardanoTransactionRecord::new(
551                        "tx_hash-123",
552                        BlockNumber(10),
553                        SlotNumber(50),
554                        "block_hash-10"
555                    ),
556                    CardanoTransactionRecord::new(
557                        "tx_hash-789",
558                        BlockNumber(10),
559                        SlotNumber(50),
560                        "block_hash-10"
561                    ),
562                ],
563                transactions
564            );
565        }
566        {
567            let transactions = repository
568                .get_transaction_by_hashes(
569                    vec!["tx_hash-123", "tx_hash-789", "tx_hash-000"],
570                    BlockNumber(100),
571                )
572                .await
573                .unwrap();
574
575            assert_eq!(
576                vec![
577                    CardanoTransactionRecord::new(
578                        "tx_hash-123",
579                        BlockNumber(10),
580                        SlotNumber(50),
581                        "block_hash-10"
582                    ),
583                    CardanoTransactionRecord::new(
584                        "tx_hash-789",
585                        BlockNumber(10),
586                        SlotNumber(50),
587                        "block_hash-10"
588                    ),
589                ],
590                transactions
591            );
592        }
593        {
594            let transactions = repository
595                .get_transaction_by_hashes(
596                    vec!["tx_hash-123", "tx_hash-789", "tx_hash-000"],
597                    BlockNumber(101),
598                )
599                .await
600                .unwrap();
601
602            assert_eq!(
603                vec![
604                    CardanoTransactionRecord::new(
605                        "tx_hash-123",
606                        BlockNumber(10),
607                        SlotNumber(50),
608                        "block_hash-10"
609                    ),
610                    CardanoTransactionRecord::new(
611                        "tx_hash-789",
612                        BlockNumber(10),
613                        SlotNumber(50),
614                        "block_hash-10"
615                    ),
616                    CardanoTransactionRecord::new(
617                        "tx_hash-000",
618                        BlockNumber(101),
619                        SlotNumber(100),
620                        "block_hash-101"
621                    ),
622                ],
623                transactions
624            );
625        }
626        {
627            let transactions = repository
628                .get_transaction_by_hashes(vec!["not-exist".to_string()], BlockNumber(100))
629                .await
630                .unwrap();
631
632            assert_eq!(Vec::<CardanoTransactionRecord>::new(), transactions);
633        }
634    }
635
636    #[tokio::test]
637    async fn repository_create_ignore_further_blocks_when_exists() {
638        let temp_dir = temp_dir_create!();
639        let repository = CardanoTransactionRepository::new(Arc::new(
640            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
641        ));
642        let base_block = CardanoBlockWithTransactions::new(
643            "block_hash-1",
644            BlockNumber(10),
645            SlotNumber(50),
646            vec![""],
647        );
648        let expected_record: CardanoBlockRecord = base_block.clone().into();
649
650        repository
651            .create_block_and_transactions(vec![base_block.clone()])
652            .await
653            .unwrap();
654
655        // Same block number - ignore changes
656        {
657            repository
658                .create_block_and_transactions(vec![CardanoBlockWithTransactions {
659                    block_hash: "block_hash-1-new".to_string(),
660                    slot_number: base_block.slot_number + 10,
661                    ..base_block.clone()
662                }])
663                .await
664                .unwrap();
665            let block_result = repository.get_block("block_hash-1").await.unwrap();
666
667            assert_eq!(Some(expected_record.clone()), block_result);
668        }
669        // Same slot number - ignore changes
670        {
671            repository
672                .create_block_and_transactions(vec![CardanoBlockWithTransactions {
673                    block_hash: "block_hash-1-new".to_string(),
674                    block_number: base_block.block_number + 10,
675                    ..base_block.clone()
676                }])
677                .await
678                .unwrap();
679            let block_result = repository.get_block("block_hash-1").await.unwrap();
680
681            assert_eq!(Some(expected_record.clone()), block_result);
682        }
683        // Same block hash - ignore changes
684        {
685            repository
686                .create_block_and_transactions(vec![CardanoBlockWithTransactions {
687                    block_number: base_block.block_number + 10,
688                    slot_number: base_block.slot_number,
689                    ..base_block.clone()
690                }])
691                .await
692                .unwrap();
693            let block_result = repository.get_block("block_hash-1").await.unwrap();
694
695            assert_eq!(Some(expected_record.clone()), block_result);
696        }
697    }
698
699    #[tokio::test]
700    async fn repository_create_ignore_further_transactions_when_exists() {
701        let temp_dir = temp_dir_create!();
702        let repository = CardanoTransactionRepository::new(Arc::new(
703            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
704        ));
705        let base_block_with_txs = CardanoBlockWithTransactions::new(
706            "block_hash-1",
707            BlockNumber(10),
708            SlotNumber(50),
709            vec!["tx_hash-1"],
710        );
711
712        repository
713            .create_block_and_transactions(vec![base_block_with_txs.clone()])
714            .await
715            .unwrap();
716        repository
717            .create_block_and_transactions(vec![base_block_with_txs.clone()])
718            .await
719            .unwrap();
720        let transactions_result = repository.get_all().await.unwrap();
721
722        assert_eq!(
723            vec![CardanoTransaction::new(
724                "tx_hash-1".to_string(),
725                BlockNumber(10),
726                SlotNumber(50),
727                "block_hash-1".to_string(),
728            )],
729            transactions_result
730        );
731    }
732
733    #[tokio::test]
734    async fn repository_create_blocks_and_transactions_and_get_stored_them_individually() {
735        let temp_dir = temp_dir_create!();
736        let repository = CardanoTransactionRepository::new(Arc::new(
737            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
738        ));
739
740        repository
741            .create_block_and_transactions(vec![CardanoBlockWithTransactions::new(
742                "block_hash-123",
743                BlockNumber(10),
744                SlotNumber(50),
745                vec!["tx_hash-123", "tx_hash-456"],
746            )])
747            .await
748            .unwrap();
749
750        let block_result = repository.get_block("block_hash-123").await.unwrap();
751        assert_eq!(
752            Some(CardanoBlockRecord {
753                block_hash: "block_hash-123".to_string(),
754                block_number: BlockNumber(10),
755                slot_number: SlotNumber(50),
756            }),
757            block_result
758        );
759
760        let transaction_result = repository.get_transaction("tx_hash-123").await.unwrap();
761        assert_eq!(
762            Some(CardanoTransactionRecord {
763                transaction_hash: "tx_hash-123".to_string(),
764                block_number: BlockNumber(10),
765                slot_number: SlotNumber(50),
766                block_hash: "block_hash-123".to_string(),
767            }),
768            transaction_result
769        );
770
771        let transaction_result = repository.get_transaction("tx_hash-456").await.unwrap();
772        assert_eq!(
773            Some(CardanoTransactionRecord {
774                transaction_hash: "tx_hash-456".to_string(),
775                block_number: BlockNumber(10),
776                slot_number: SlotNumber(50),
777                block_hash: "block_hash-123".to_string(),
778            }),
779            transaction_result
780        );
781    }
782
783    #[tokio::test]
784    async fn repository_store_blocks_and_transactions_bulk_insert_1_000_blocks_with_500_transactions_per_block()
785     {
786        let temp_dir = temp_dir_create!();
787        let repository = CardanoTransactionRepository::new(Arc::new(
788            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
789        ));
790
791        let blocks_to_insert: Vec<_> = (0..1_000)
792            .map(|i| {
793                CardanoBlockWithTransactions::new(
794                    format!("block_hash-{i}"),
795                    BlockNumber(i),
796                    SlotNumber(i * 5),
797                    (0..500).map(|j| format!("tx_hash-{i}-{j}")).collect(),
798                )
799            })
800            .collect();
801
802        repository
803            .store_blocks_and_transactions(blocks_to_insert)
804            .await
805            .unwrap();
806
807        let all_blocks = repository.get_all_blocks().await.unwrap();
808        assert_eq!(1_000, all_blocks.len());
809        let all_transactions = repository.get_all_transactions().await.unwrap();
810        assert_eq!(1_000 * 500, all_transactions.len());
811    }
812
813    #[tokio::test]
814    async fn repository_get_all_stored_blocks() {
815        let temp_dir = temp_dir_create!();
816        let repository = CardanoTransactionRepository::new(Arc::new(
817            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
818        ));
819
820        repository
821            .create_block_and_transactions(vec![
822                CardanoBlockWithTransactions::new(
823                    "block_hash-1",
824                    BlockNumber(10),
825                    SlotNumber(50),
826                    Vec::<String>::new(),
827                ),
828                CardanoBlockWithTransactions::new(
829                    "block_hash-2",
830                    BlockNumber(11),
831                    SlotNumber(51),
832                    Vec::<String>::new(),
833                ),
834            ])
835            .await
836            .unwrap();
837
838        let transactions_result = repository.get_all_blocks().await.unwrap();
839        assert_eq!(
840            vec![
841                CardanoBlockRecord::new(
842                    "block_hash-1".to_string(),
843                    BlockNumber(10),
844                    SlotNumber(50),
845                ),
846                CardanoBlockRecord::new(
847                    "block_hash-2".to_string(),
848                    BlockNumber(11),
849                    SlotNumber(51),
850                )
851            ],
852            transactions_result
853        )
854    }
855
856    #[tokio::test]
857    async fn repository_get_all_stored_transactions() {
858        let temp_dir = temp_dir_create!();
859        let repository = CardanoTransactionRepository::new(Arc::new(
860            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
861        ));
862
863        repository
864            .create_block_and_transactions(vec![CardanoBlockWithTransactions::new(
865                "block_hash-123",
866                BlockNumber(10),
867                SlotNumber(50),
868                vec!["tx_hash-123", "tx_hash-456"],
869            )])
870            .await
871            .unwrap();
872
873        let transactions_result = repository.get_all_transactions().await.unwrap();
874        assert_eq!(
875            vec![
876                CardanoTransactionRecord::new(
877                    "tx_hash-123".to_string(),
878                    BlockNumber(10),
879                    SlotNumber(50),
880                    "block_hash-123".to_string(),
881                ),
882                CardanoTransactionRecord::new(
883                    "tx_hash-456".to_string(),
884                    BlockNumber(10),
885                    SlotNumber(50),
886                    "block_hash-123".to_string(),
887                )
888            ],
889            transactions_result
890        )
891    }
892
893    #[tokio::test]
894    async fn repository_get_highest_chain_point_without_blocks_in_db() {
895        let temp_dir = temp_dir_create!();
896        let repository = CardanoTransactionRepository::new(Arc::new(
897            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
898        ));
899
900        let highest_beacon = repository.get_transaction_highest_chain_point().await.unwrap();
901        assert_eq!(None, highest_beacon);
902    }
903
904    #[tokio::test]
905    async fn repository_get_highest_chain_point_with_blocks_in_db() {
906        let temp_dir = temp_dir_create!();
907        let repository = CardanoTransactionRepository::new(Arc::new(
908            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
909        ));
910
911        repository
912            .create_block_and_transactions(vec![
913                CardanoBlockWithTransactions::new(
914                    "block_hash-1",
915                    BlockNumber(10),
916                    SlotNumber(50),
917                    Vec::<String>::new(),
918                ),
919                CardanoBlockWithTransactions::new(
920                    "block_hash-2",
921                    BlockNumber(100),
922                    SlotNumber(150),
923                    Vec::<String>::new(),
924                ),
925            ])
926            .await
927            .unwrap();
928
929        let highest_beacon = repository.get_transaction_highest_chain_point().await.unwrap();
930        assert_eq!(
931            Some(ChainPoint::new(
932                SlotNumber(150),
933                BlockNumber(100),
934                "block_hash-2"
935            )),
936            highest_beacon
937        );
938    }
939
940    #[tokio::test]
941    async fn repository_get_transactions_in_range_blocks() {
942        let temp_dir = temp_dir_create!();
943        let repository = CardanoTransactionRepository::new(Arc::new(
944            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
945        ));
946
947        let blocks = vec![
948            CardanoBlockWithTransactions::new(
949                "block_hash-1",
950                BlockNumber(10),
951                SlotNumber(50),
952                vec!["tx_hash-1"],
953            ),
954            CardanoBlockWithTransactions::new(
955                "block_hash-2",
956                BlockNumber(11),
957                SlotNumber(51),
958                vec!["tx_hash-2"],
959            ),
960            CardanoBlockWithTransactions::new(
961                "block_hash-3",
962                BlockNumber(12),
963                SlotNumber(52),
964                vec!["tx_hash-3"],
965            ),
966        ];
967        repository
968            .create_block_and_transactions(blocks.clone())
969            .await
970            .unwrap();
971        let expected_transactions: Vec<CardanoTransactionRecord> = blocks
972            .into_iter()
973            .flat_map(|b| b.into_transactions())
974            .map(Into::into)
975            .collect();
976
977        {
978            let transaction_result = repository
979                .get_transactions_in_range_blocks(BlockNumber(0)..BlockNumber(10))
980                .await
981                .unwrap();
982            assert_eq!(Vec::<CardanoTransactionRecord>::new(), transaction_result);
983        }
984        {
985            let transaction_result = repository
986                .get_transactions_in_range_blocks(BlockNumber(13)..BlockNumber(21))
987                .await
988                .unwrap();
989            assert_eq!(Vec::<CardanoTransactionRecord>::new(), transaction_result);
990        }
991        {
992            let transaction_result = repository
993                .get_transactions_in_range_blocks(BlockNumber(9)..BlockNumber(12))
994                .await
995                .unwrap();
996            assert_eq!(expected_transactions[0..=1].to_vec(), transaction_result);
997        }
998        {
999            let transaction_result = repository
1000                .get_transactions_in_range_blocks(BlockNumber(10)..BlockNumber(13))
1001                .await
1002                .unwrap();
1003            assert_eq!(expected_transactions.clone(), transaction_result);
1004        }
1005        {
1006            let transaction_result = repository
1007                .get_transactions_in_range_blocks(BlockNumber(11)..BlockNumber(14))
1008                .await
1009                .unwrap();
1010            assert_eq!(expected_transactions[1..=2].to_vec(), transaction_result);
1011        }
1012    }
1013
1014    #[tokio::test]
1015    async fn repository_get_blocks_with_transactions_in_range_blocks() {
1016        let temp_dir = temp_dir_create!();
1017        let repository = CardanoTransactionRepository::new(Arc::new(
1018            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
1019        ));
1020
1021        let blocks = vec![
1022            CardanoBlockWithTransactions::new(
1023                "block_hash-1",
1024                BlockNumber(10),
1025                SlotNumber(50),
1026                vec!["tx_hash-1"],
1027            ),
1028            CardanoBlockWithTransactions::new(
1029                "block_hash-2",
1030                BlockNumber(11),
1031                SlotNumber(51),
1032                vec!["tx_hash-2"],
1033            ),
1034            CardanoBlockWithTransactions::new(
1035                "block_hash-3",
1036                BlockNumber(12),
1037                SlotNumber(52),
1038                vec!["tx_hash-3"],
1039            ),
1040        ];
1041        repository
1042            .create_block_and_transactions(blocks.clone())
1043            .await
1044            .unwrap();
1045        let expected_blocks: Vec<CardanoBlockTransactionsRecord> =
1046            blocks.into_iter().map(Into::into).collect();
1047
1048        {
1049            let blocks = repository
1050                .get_blocks_with_transactions_in_range_blocks(BlockNumber(0)..BlockNumber(10))
1051                .await
1052                .unwrap();
1053            assert_eq!(Vec::<CardanoBlockTransactionsRecord>::new(), blocks);
1054        }
1055        {
1056            let blocks = repository
1057                .get_blocks_with_transactions_in_range_blocks(BlockNumber(13)..BlockNumber(21))
1058                .await
1059                .unwrap();
1060            assert_eq!(Vec::<CardanoBlockTransactionsRecord>::new(), blocks);
1061        }
1062        {
1063            let blocks = repository
1064                .get_blocks_with_transactions_in_range_blocks(BlockNumber(9)..BlockNumber(12))
1065                .await
1066                .unwrap();
1067            assert_eq!(expected_blocks[0..=1].to_vec(), blocks);
1068        }
1069        {
1070            let blocks = repository
1071                .get_blocks_with_transactions_in_range_blocks(BlockNumber(10)..BlockNumber(13))
1072                .await
1073                .unwrap();
1074            assert_eq!(expected_blocks.clone(), blocks);
1075        }
1076        {
1077            let blocks = repository
1078                .get_blocks_with_transactions_in_range_blocks(BlockNumber(11)..BlockNumber(14))
1079                .await
1080                .unwrap();
1081            assert_eq!(expected_blocks[1..=2].to_vec(), blocks);
1082        }
1083    }
1084
1085    #[tokio::test]
1086    async fn repository_get_transactions_by_block_ranges() {
1087        let temp_dir = temp_dir_create!();
1088        let repository = CardanoTransactionRepository::new(Arc::new(
1089            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
1090        ));
1091
1092        let blocks = vec![
1093            CardanoBlockWithTransactions::new(
1094                "block_hash-1",
1095                BlockNumber(10),
1096                SlotNumber(50),
1097                vec!["tx_hash-1"],
1098            ),
1099            CardanoBlockWithTransactions::new(
1100                "block_hash-2",
1101                BlockNumber(11),
1102                SlotNumber(51),
1103                vec!["tx_hash-2"],
1104            ),
1105            CardanoBlockWithTransactions::new(
1106                "block_hash-3",
1107                BlockNumber(20),
1108                SlotNumber(52),
1109                vec!["tx_hash-3"],
1110            ),
1111            CardanoBlockWithTransactions::new(
1112                "block_hash-4",
1113                BlockNumber(31),
1114                SlotNumber(53),
1115                vec!["tx_hash-4"],
1116            ),
1117            CardanoBlockWithTransactions::new(
1118                "block_hash-5",
1119                BlockNumber(35),
1120                SlotNumber(54),
1121                vec!["tx_hash-5"],
1122            ),
1123            CardanoBlockWithTransactions::new(
1124                "block_hash-6",
1125                BlockNumber(46),
1126                SlotNumber(55),
1127                vec!["tx_hash-6"],
1128            ),
1129        ];
1130        repository
1131            .create_block_and_transactions(blocks.clone())
1132            .await
1133            .unwrap();
1134        let expected_transactions: Vec<CardanoTransactionRecord> = blocks
1135            .into_iter()
1136            .flat_map(|b| b.into_transactions())
1137            .map(Into::into)
1138            .collect();
1139
1140        {
1141            let transaction_result = repository
1142                .get_transaction_by_block_ranges(vec![BlockRange::from_block_number(BlockNumber(
1143                    100,
1144                ))])
1145                .await
1146                .unwrap();
1147            assert_eq!(Vec::<CardanoTransactionRecord>::new(), transaction_result);
1148        }
1149        {
1150            let transaction_result = repository
1151                .get_transaction_by_block_ranges(vec![BlockRange::from_block_number(BlockNumber(
1152                    0,
1153                ))])
1154                .await
1155                .unwrap();
1156            assert_eq!(expected_transactions[0..=1].to_vec(), transaction_result);
1157        }
1158        {
1159            let transaction_result = repository
1160                .get_transaction_by_block_ranges(vec![
1161                    BlockRange::from_block_number(BlockNumber(0)),
1162                    BlockRange::from_block_number(BlockNumber(15)),
1163                ])
1164                .await
1165                .unwrap();
1166            assert_eq!(expected_transactions[0..=2].to_vec(), transaction_result);
1167        }
1168        {
1169            let transaction_result = repository
1170                .get_transaction_by_block_ranges(vec![
1171                    BlockRange::from_block_number(BlockNumber(0)),
1172                    BlockRange::from_block_number(BlockNumber(30)),
1173                ])
1174                .await
1175                .unwrap();
1176            assert_eq!(
1177                [
1178                    expected_transactions[0..=1].to_vec(),
1179                    expected_transactions[3..=4].to_vec()
1180                ]
1181                .concat(),
1182                transaction_result
1183            );
1184        }
1185    }
1186
1187    #[tokio::test]
1188    async fn repository_get_blocks_with_transactions_by_block_ranges() {
1189        let temp_dir = temp_dir_create!();
1190        let repository = CardanoTransactionRepository::new(Arc::new(
1191            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
1192        ));
1193
1194        let blocks = vec![
1195            CardanoBlockWithTransactions::new(
1196                "block_hash-1",
1197                BlockNumber(10),
1198                SlotNumber(50),
1199                vec!["tx_hash-1"],
1200            ),
1201            CardanoBlockWithTransactions::new(
1202                "block_hash-2",
1203                BlockNumber(11),
1204                SlotNumber(51),
1205                vec!["tx_hash-2"],
1206            ),
1207            CardanoBlockWithTransactions::new(
1208                "block_hash-3",
1209                BlockNumber(20),
1210                SlotNumber(52),
1211                vec!["tx_hash-3"],
1212            ),
1213            CardanoBlockWithTransactions::new(
1214                "block_hash-4",
1215                BlockNumber(31),
1216                SlotNumber(53),
1217                vec!["tx_hash-4"],
1218            ),
1219            CardanoBlockWithTransactions::new(
1220                "block_hash-5",
1221                BlockNumber(35),
1222                SlotNumber(54),
1223                vec!["tx_hash-5"],
1224            ),
1225            CardanoBlockWithTransactions::new(
1226                "block_hash-6",
1227                BlockNumber(46),
1228                SlotNumber(55),
1229                vec!["tx_hash-6"],
1230            ),
1231        ];
1232        repository
1233            .create_block_and_transactions(blocks.clone())
1234            .await
1235            .unwrap();
1236        let expected_blocks: Vec<CardanoBlockTransactionsRecord> =
1237            blocks.into_iter().map(Into::into).collect();
1238
1239        {
1240            let transaction_result = repository
1241                .get_blocks_with_transactions_by_block_ranges(vec![BlockRange::from_block_number(
1242                    BlockNumber(100),
1243                )])
1244                .await
1245                .unwrap();
1246            assert_eq!(
1247                Vec::<CardanoBlockTransactionsRecord>::new(),
1248                transaction_result
1249            );
1250        }
1251        {
1252            let transaction_result = repository
1253                .get_blocks_with_transactions_by_block_ranges(vec![BlockRange::from_block_number(
1254                    BlockNumber(0),
1255                )])
1256                .await
1257                .unwrap();
1258            assert_eq!(expected_blocks[0..=1].to_vec(), transaction_result);
1259        }
1260        {
1261            let transaction_result = repository
1262                .get_blocks_with_transactions_by_block_ranges(vec![
1263                    BlockRange::from_block_number(BlockNumber(0)),
1264                    BlockRange::from_block_number(BlockNumber(15)),
1265                ])
1266                .await
1267                .unwrap();
1268            assert_eq!(expected_blocks[0..=2].to_vec(), transaction_result);
1269        }
1270        {
1271            let transaction_result = repository
1272                .get_blocks_with_transactions_by_block_ranges(vec![
1273                    BlockRange::from_block_number(BlockNumber(0)),
1274                    BlockRange::from_block_number(BlockNumber(30)),
1275                ])
1276                .await
1277                .unwrap();
1278            assert_eq!(
1279                [expected_blocks[0..=1].to_vec(), expected_blocks[3..=4].to_vec()].concat(),
1280                transaction_result
1281            );
1282        }
1283    }
1284
1285    #[tokio::test]
1286    async fn repository_get_closest_block_number_by_slot_number() {
1287        let temp_dir = temp_dir_create!();
1288        let repository = CardanoTransactionRepository::new(Arc::new(
1289            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
1290        ));
1291
1292        let blocks = vec![
1293            CardanoBlockWithTransactions::new(
1294                "block_hash-1",
1295                BlockNumber(100),
1296                SlotNumber(500),
1297                Vec::<String>::new(),
1298            ),
1299            CardanoBlockWithTransactions::new(
1300                "block_hash-2",
1301                BlockNumber(101),
1302                SlotNumber(501),
1303                Vec::<String>::new(),
1304            ),
1305        ];
1306        repository
1307            .create_block_and_transactions(blocks.clone())
1308            .await
1309            .unwrap();
1310
1311        let transaction_block_number_retrieved = repository
1312            .get_closest_block_number_above_slot_number(SlotNumber(500))
1313            .await
1314            .unwrap();
1315
1316        assert_eq!(transaction_block_number_retrieved, Some(BlockNumber(100)));
1317    }
1318
1319    #[tokio::test]
1320    async fn repository_store_legacy_block_range() {
1321        let temp_dir = temp_dir_create!();
1322        let repository = CardanoTransactionRepository::new(Arc::new(
1323            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
1324        ));
1325
1326        repository
1327            .create_legacy_block_range_roots(vec![
1328                (
1329                    BlockRange::from_block_number(BlockNumber(0)),
1330                    MKTreeNode::from_hex("AAAA").unwrap(),
1331                ),
1332                (
1333                    BlockRange::from_block_number(BlockRange::LENGTH),
1334                    MKTreeNode::from_hex("BBBB").unwrap(),
1335                ),
1336            ])
1337            .await
1338            .unwrap();
1339
1340        let connection = repository.connection_pool.connection().unwrap();
1341        let records: Vec<BlockRangeRootRecord> =
1342            connection.fetch_collect(GetLegacyBlockRangeRootQuery::all()).unwrap();
1343        assert_eq!(
1344            vec![
1345                BlockRangeRootRecord {
1346                    range: BlockRange::from_block_number(BlockNumber(0)),
1347                    merkle_root: MKTreeNode::from_hex("AAAA").unwrap(),
1348                },
1349                BlockRangeRootRecord {
1350                    range: BlockRange::from_block_number(BlockRange::LENGTH),
1351                    merkle_root: MKTreeNode::from_hex("BBBB").unwrap(),
1352                }
1353            ],
1354            records
1355        );
1356    }
1357
1358    #[tokio::test]
1359    async fn repository_store_legacy_block_range_with_existing_hash_doesnt_erase_existing_data() {
1360        let temp_dir = temp_dir_create!();
1361        let repository = CardanoTransactionRepository::new(Arc::new(
1362            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
1363        ));
1364        let range = BlockRange::from_block_number(BlockNumber(0));
1365
1366        repository
1367            .create_legacy_block_range_roots(vec![(
1368                range.clone(),
1369                MKTreeNode::from_hex("AAAA").unwrap(),
1370            )])
1371            .await
1372            .unwrap();
1373        repository
1374            .create_legacy_block_range_roots(vec![(
1375                range.clone(),
1376                MKTreeNode::from_hex("BBBB").unwrap(),
1377            )])
1378            .await
1379            .unwrap();
1380
1381        let record: Vec<BlockRangeRootRecord> = repository
1382            .connection_pool
1383            .connection()
1384            .unwrap()
1385            .fetch_collect(GetLegacyBlockRangeRootQuery::all())
1386            .unwrap();
1387        assert_eq!(
1388            vec![BlockRangeRootRecord {
1389                range,
1390                merkle_root: MKTreeNode::from_hex("AAAA").unwrap()
1391            }],
1392            record
1393        );
1394    }
1395
1396    #[tokio::test]
1397    async fn repository_retrieve_legacy_block_range_roots_up_to() {
1398        let temp_dir = temp_dir_create!();
1399        let repository = CardanoTransactionRepository::new(Arc::new(
1400            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
1401        ));
1402
1403        let block_range_roots = vec![
1404            (
1405                BlockRange::from_block_number(BlockNumber(15)),
1406                MKTreeNode::from_hex("AAAA").unwrap(),
1407            ),
1408            (
1409                BlockRange::from_block_number(BlockNumber(30)),
1410                MKTreeNode::from_hex("BBBB").unwrap(),
1411            ),
1412            (
1413                BlockRange::from_block_number(BlockNumber(45)),
1414                MKTreeNode::from_hex("CCCC").unwrap(),
1415            ),
1416        ];
1417        repository
1418            .create_legacy_block_range_roots(block_range_roots.clone())
1419            .await
1420            .unwrap();
1421
1422        let retrieved_block_ranges = repository
1423            .retrieve_legacy_block_range_roots_up_to(BlockNumber(45))
1424            .await
1425            .unwrap();
1426        assert_eq!(
1427            block_range_roots[0..2].to_vec(),
1428            retrieved_block_ranges.collect::<Vec<_>>()
1429        );
1430    }
1431
1432    #[tokio::test]
1433    async fn repository_retrieve_highest_legacy_block_range_roots() {
1434        let temp_dir = temp_dir_create!();
1435        let repository = CardanoTransactionRepository::new(Arc::new(
1436            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
1437        ));
1438
1439        let block_range_roots = vec![
1440            BlockRangeRootRecord {
1441                range: BlockRange::from_block_number(BlockNumber(15)),
1442                merkle_root: MKTreeNode::from_hex("AAAA").unwrap(),
1443            },
1444            BlockRangeRootRecord {
1445                range: BlockRange::from_block_number(BlockNumber(30)),
1446                merkle_root: MKTreeNode::from_hex("BBBB").unwrap(),
1447            },
1448            BlockRangeRootRecord {
1449                range: BlockRange::from_block_number(BlockNumber(45)),
1450                merkle_root: MKTreeNode::from_hex("CCCC").unwrap(),
1451            },
1452        ];
1453        repository
1454            .create_legacy_block_range_roots(block_range_roots.clone())
1455            .await
1456            .unwrap();
1457
1458        let retrieved_block_range =
1459            repository.retrieve_highest_legacy_block_range_root().await.unwrap();
1460        assert_eq!(block_range_roots.last().cloned(), retrieved_block_range);
1461    }
1462
1463    #[tokio::test]
1464    async fn repository_store_block_range() {
1465        let temp_dir = temp_dir_create!();
1466        let repository = CardanoTransactionRepository::new(Arc::new(
1467            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
1468        ));
1469
1470        repository
1471            .create_block_range_roots(vec![
1472                (
1473                    BlockRange::from_block_number(BlockNumber(0)),
1474                    MKTreeNode::from_hex("AAAA").unwrap(),
1475                ),
1476                (
1477                    BlockRange::from_block_number(BlockRange::LENGTH),
1478                    MKTreeNode::from_hex("BBBB").unwrap(),
1479                ),
1480            ])
1481            .await
1482            .unwrap();
1483
1484        let connection = repository.connection_pool.connection().unwrap();
1485        let records: Vec<BlockRangeRootRecord> =
1486            connection.fetch_collect(GetBlockRangeRootQuery::all()).unwrap();
1487        assert_eq!(
1488            vec![
1489                BlockRangeRootRecord {
1490                    range: BlockRange::from_block_number(BlockNumber(0)),
1491                    merkle_root: MKTreeNode::from_hex("AAAA").unwrap(),
1492                },
1493                BlockRangeRootRecord {
1494                    range: BlockRange::from_block_number(BlockRange::LENGTH),
1495                    merkle_root: MKTreeNode::from_hex("BBBB").unwrap(),
1496                }
1497            ],
1498            records
1499        );
1500    }
1501
1502    #[tokio::test]
1503    async fn repository_store_block_range_with_existing_hash_doesnt_erase_existing_data() {
1504        let temp_dir = temp_dir_create!();
1505        let repository = CardanoTransactionRepository::new(Arc::new(
1506            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
1507        ));
1508        let range = BlockRange::from_block_number(BlockNumber(0));
1509
1510        repository
1511            .create_block_range_roots(vec![(range.clone(), MKTreeNode::from_hex("AAAA").unwrap())])
1512            .await
1513            .unwrap();
1514        repository
1515            .create_block_range_roots(vec![(range.clone(), MKTreeNode::from_hex("BBBB").unwrap())])
1516            .await
1517            .unwrap();
1518
1519        let record: Vec<BlockRangeRootRecord> = repository
1520            .connection_pool
1521            .connection()
1522            .unwrap()
1523            .fetch_collect(GetBlockRangeRootQuery::all())
1524            .unwrap();
1525        assert_eq!(
1526            vec![BlockRangeRootRecord {
1527                range,
1528                merkle_root: MKTreeNode::from_hex("AAAA").unwrap()
1529            }],
1530            record
1531        );
1532    }
1533
1534    #[tokio::test]
1535    async fn repository_retrieve_block_range_roots_up_to() {
1536        let temp_dir = temp_dir_create!();
1537        let repository = CardanoTransactionRepository::new(Arc::new(
1538            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
1539        ));
1540        let block_range_roots = vec![
1541            (
1542                BlockRange::from_block_number(BlockNumber(15)),
1543                MKTreeNode::from_hex("AAAA").unwrap(),
1544            ),
1545            (
1546                BlockRange::from_block_number(BlockNumber(30)),
1547                MKTreeNode::from_hex("BBBB").unwrap(),
1548            ),
1549            (
1550                BlockRange::from_block_number(BlockNumber(45)),
1551                MKTreeNode::from_hex("CCCC").unwrap(),
1552            ),
1553        ];
1554        repository
1555            .create_block_range_roots(block_range_roots.clone())
1556            .await
1557            .unwrap();
1558
1559        let retrieved_block_ranges = repository
1560            .retrieve_block_range_roots_up_to(BlockNumber(45))
1561            .await
1562            .unwrap();
1563        assert_eq!(
1564            block_range_roots[0..2].to_vec(),
1565            retrieved_block_ranges.collect::<Vec<_>>()
1566        );
1567    }
1568
1569    #[tokio::test]
1570    async fn repository_retrieve_highest_block_range_roots() {
1571        let temp_dir = temp_dir_create!();
1572        let repository = CardanoTransactionRepository::new(Arc::new(
1573            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
1574        ));
1575        let block_range_roots = vec![
1576            BlockRangeRootRecord {
1577                range: BlockRange::from_block_number(BlockNumber(15)),
1578                merkle_root: MKTreeNode::from_hex("AAAA").unwrap(),
1579            },
1580            BlockRangeRootRecord {
1581                range: BlockRange::from_block_number(BlockNumber(30)),
1582                merkle_root: MKTreeNode::from_hex("BBBB").unwrap(),
1583            },
1584            BlockRangeRootRecord {
1585                range: BlockRange::from_block_number(BlockNumber(45)),
1586                merkle_root: MKTreeNode::from_hex("CCCC").unwrap(),
1587            },
1588        ];
1589        repository
1590            .create_block_range_roots(block_range_roots.clone())
1591            .await
1592            .unwrap();
1593
1594        let retrieved_block_range = repository.retrieve_highest_block_range_root().await.unwrap();
1595        assert_eq!(block_range_roots.last().cloned(), retrieved_block_range);
1596    }
1597
1598    #[tokio::test]
1599    async fn repository_prune_blocks_and_transactions() {
1600        let temp_dir = temp_dir_create!();
1601        let repository = CardanoTransactionRepository::new(Arc::new(
1602            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
1603        ));
1604
1605        let blocks = vec![
1606            CardanoBlockWithTransactions::new(
1607                "block_hash-1",
1608                BlockNumber(24),
1609                SlotNumber(50),
1610                vec!["tx_hash-1"],
1611            ),
1612            CardanoBlockWithTransactions::new(
1613                "block_hash-2",
1614                BlockNumber(25),
1615                SlotNumber(51),
1616                vec!["tx_hash-2"],
1617            ),
1618            CardanoBlockWithTransactions::new(
1619                "block_hash-3",
1620                BlockNumber(26),
1621                SlotNumber(52),
1622                vec!["tx_hash-3", "tx_hash-4"],
1623            ),
1624        ];
1625        repository.create_block_and_transactions(blocks).await.unwrap();
1626        // Use by 'prune_transaction' to get the block_range of the highest block number
1627        repository
1628            .create_legacy_block_range_roots(vec![(
1629                BlockRange::from_block_number(BlockNumber(45)),
1630                MKTreeNode::from_hex("BBBB").unwrap(),
1631            )])
1632            .await
1633            .unwrap();
1634
1635        let stored_transactions = repository.get_all_transactions().await.unwrap();
1636        assert_eq!(4, stored_transactions.len());
1637        let stored_blocks = repository.get_all_blocks().await.unwrap();
1638        assert_eq!(3, stored_blocks.len());
1639
1640        // Pruning with a number of block to keep greater than the highest block range start should
1641        // do nothing.
1642        repository.prune_transaction(BlockNumber(10_000_000)).await.unwrap();
1643        let stored_transactions = repository.get_all_transactions().await.unwrap();
1644        assert_eq!(4, stored_transactions.len());
1645        let stored_blocks = repository.get_all_blocks().await.unwrap();
1646        assert_eq!(3, stored_blocks.len());
1647
1648        // Since the highest block range start is 45, pruning with 20 should remove transactions
1649        // with a block number strictly below 25.
1650        repository.prune_transaction(BlockNumber(20)).await.unwrap();
1651        let transaction_result = repository
1652            .get_transactions_in_range_blocks(BlockNumber(0)..BlockNumber(25))
1653            .await
1654            .unwrap();
1655        assert_eq!(Vec::<CardanoTransactionRecord>::new(), transaction_result);
1656
1657        let transaction_result = repository
1658            .get_transactions_in_range_blocks(BlockNumber(25)..BlockNumber(1000))
1659            .await
1660            .unwrap();
1661        assert_eq!(3, transaction_result.len());
1662
1663        let stored_blocks = repository.get_all_blocks().await.unwrap();
1664        assert_eq!(2, stored_blocks.len());
1665    }
1666
1667    #[tokio::test]
1668    async fn get_prune_blocks_threshold() {
1669        let temp_dir = temp_dir_create!();
1670        let repository = CardanoTransactionRepository::new(Arc::new(
1671            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
1672        ));
1673
1674        // Nothing stored
1675        {
1676            let highest = repository.get_prune_blocks_threshold().await.unwrap();
1677            assert_eq!(None, highest);
1678        }
1679        // Add two block range roots in the "new" table but none in the "legacy"; select the highest of the two in the "new" table
1680        {
1681            repository
1682                .create_block_range_roots(vec![
1683                    (
1684                        BlockRange::from_block_number(BlockNumber(45)),
1685                        MKTreeNode::from_hex("AAAA").unwrap(),
1686                    ),
1687                    (
1688                        BlockRange::from_block_number(BlockNumber(60)),
1689                        MKTreeNode::from_hex("BBBB").unwrap(),
1690                    ),
1691                ])
1692                .await
1693                .unwrap();
1694            let highest = repository.get_prune_blocks_threshold().await.unwrap();
1695            assert_eq!(Some(BlockNumber(60)), highest);
1696        }
1697        // Add a first block range in the "legacy" table, but below than existing in the "new" table; the highest should be the new block range
1698        {
1699            repository
1700                .create_legacy_block_range_roots(vec![(
1701                    BlockRange::from_block_number(BlockNumber(30)),
1702                    MKTreeNode::from_hex("DDDD").unwrap(),
1703                )])
1704                .await
1705                .unwrap();
1706
1707            let highest = repository.get_prune_blocks_threshold().await.unwrap();
1708            assert_eq!(Some(BlockNumber(30)), highest);
1709        }
1710        // Add a second block range in the "legacy" table, but higher than existing in the "new" table; the highest should be change back to the new block range
1711        {
1712            repository
1713                .create_legacy_block_range_roots(vec![(
1714                    BlockRange::from_block_number(BlockNumber(75)),
1715                    MKTreeNode::from_hex("CCCC").unwrap(),
1716                )])
1717                .await
1718                .unwrap();
1719
1720            let highest = repository.get_prune_blocks_threshold().await.unwrap();
1721            assert_eq!(Some(BlockNumber(60)), highest);
1722        }
1723    }
1724
1725    #[tokio::test]
1726    async fn remove_blocks_transactions_and_block_ranges_greater_than_given_block_number() {
1727        let temp_dir = temp_dir_create!();
1728        let repository = CardanoTransactionRepository::new(Arc::new(
1729            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
1730        ));
1731
1732        let blocks = vec![
1733            CardanoBlockWithTransactions::new(
1734                "block_hash-1",
1735                BlockRange::LENGTH,
1736                SlotNumber(50),
1737                vec!["tx_hash-1", "tx_hash-2"],
1738            ),
1739            CardanoBlockWithTransactions::new(
1740                "block_hash-2",
1741                BlockRange::LENGTH * 3,
1742                SlotNumber(51),
1743                vec!["tx_hash-3", "tx_hash-4"],
1744            ),
1745            CardanoBlockWithTransactions::new(
1746                "block_hash-3",
1747                BlockRange::LENGTH * 3 + 1,
1748                SlotNumber(52),
1749                vec!["tx_hash-5", "tx_hash-6"],
1750            ),
1751        ];
1752        repository.create_block_and_transactions(blocks).await.unwrap();
1753        repository
1754            .create_legacy_block_range_roots(vec![
1755                (
1756                    BlockRange::from_block_number(BlockRange::LENGTH),
1757                    MKTreeNode::from_hex("AAAA").unwrap(),
1758                ),
1759                (
1760                    BlockRange::from_block_number(BlockRange::LENGTH * 2),
1761                    MKTreeNode::from_hex("AAAA").unwrap(),
1762                ),
1763                (
1764                    BlockRange::from_block_number(BlockRange::LENGTH * 3),
1765                    MKTreeNode::from_hex("AAAA").unwrap(),
1766                ),
1767            ])
1768            .await
1769            .unwrap();
1770        repository
1771            .create_block_range_roots(vec![
1772                (
1773                    BlockRange::from_block_number(BlockNumber(0)),
1774                    MKTreeNode::from_hex("9999").unwrap(),
1775                ),
1776                (
1777                    BlockRange::from_block_number(BlockRange::LENGTH),
1778                    MKTreeNode::from_hex("AAAA").unwrap(),
1779                ),
1780                (
1781                    BlockRange::from_block_number(BlockRange::LENGTH * 2),
1782                    MKTreeNode::from_hex("AAAA").unwrap(),
1783                ),
1784                (
1785                    BlockRange::from_block_number(BlockRange::LENGTH * 3),
1786                    MKTreeNode::from_hex("AAAA").unwrap(),
1787                ),
1788            ])
1789            .await
1790            .unwrap();
1791
1792        repository
1793            .remove_rolled_back_transactions_and_block_range_by_block_number(BlockRange::LENGTH * 3)
1794            .await
1795            .unwrap();
1796        assert_eq!(4, repository.get_all_transactions().await.unwrap().len());
1797        assert_eq!(
1798            2,
1799            repository.get_all_legacy_block_range_root().unwrap().len()
1800        );
1801        assert_eq!(3, repository.get_all_block_range_root().unwrap().len());
1802        assert_eq!(2, repository.get_all_blocks().await.unwrap().len());
1803    }
1804
1805    #[tokio::test]
1806    async fn remove_rolled_back_blocks_transactions_and_block_range_by_slot_number() {
1807        fn transaction_record(
1808            block_number: BlockNumber,
1809            slot_number: SlotNumber,
1810            tx_hash: &str,
1811        ) -> CardanoTransactionRecord {
1812            CardanoTransactionRecord::new(
1813                tx_hash,
1814                block_number,
1815                slot_number,
1816                format!("block_hash-{block_number}"),
1817            )
1818        }
1819
1820        let temp_dir = temp_dir_create!();
1821        let repository = CardanoTransactionRepository::new(Arc::new(
1822            cardano_tx_db_connection_builder(&temp_dir).build_pool(1).unwrap(),
1823        ));
1824
1825        let blocks = vec![
1826            CardanoBlockWithTransactions::new(
1827                "block_hash-10",
1828                BlockNumber(10),
1829                SlotNumber(50),
1830                vec!["tx_hash-1"],
1831            ),
1832            CardanoBlockWithTransactions::new(
1833                "block_hash-13",
1834                BlockNumber(13),
1835                SlotNumber(52),
1836                vec!["tx_hash-2"],
1837            ),
1838            CardanoBlockWithTransactions::new(
1839                "block_hash-101",
1840                BlockNumber(101),
1841                SlotNumber(100),
1842                vec!["tx_hash-3", "tx_hash-4"],
1843            ),
1844        ];
1845        repository.create_block_and_transactions(blocks).await.unwrap();
1846
1847        {
1848            repository
1849                .remove_rolled_back_blocks_transactions_and_block_range_by_slot_number(SlotNumber(
1850                    110,
1851                ))
1852                .await
1853                .expect("Failed to remove rolled back transactions");
1854
1855            let transactions = repository.get_all_transactions().await.unwrap();
1856            assert_eq!(
1857                vec![
1858                    transaction_record(BlockNumber(10), SlotNumber(50), "tx_hash-1"),
1859                    transaction_record(BlockNumber(13), SlotNumber(52), "tx_hash-2"),
1860                    transaction_record(BlockNumber(101), SlotNumber(100), "tx_hash-3"),
1861                    transaction_record(BlockNumber(101), SlotNumber(100), "tx_hash-4"),
1862                ],
1863                transactions
1864            );
1865        }
1866
1867        {
1868            repository
1869                .remove_rolled_back_blocks_transactions_and_block_range_by_slot_number(SlotNumber(
1870                    53,
1871                ))
1872                .await
1873                .expect("Failed to remove rolled back transactions");
1874
1875            let transactions = repository.get_all_transactions().await.unwrap();
1876            assert_eq!(
1877                vec![
1878                    transaction_record(BlockNumber(10), SlotNumber(50), "tx_hash-1"),
1879                    transaction_record(BlockNumber(13), SlotNumber(52), "tx_hash-2"),
1880                ],
1881                transactions
1882            );
1883        }
1884
1885        {
1886            repository
1887                .remove_rolled_back_blocks_transactions_and_block_range_by_slot_number(SlotNumber(
1888                    51,
1889                ))
1890                .await
1891                .expect("Failed to remove rolled back transactions");
1892
1893            let transactions = repository.get_all_transactions().await.unwrap();
1894            assert_eq!(
1895                vec![transaction_record(BlockNumber(10), SlotNumber(50), "tx_hash-1")],
1896                transactions
1897            );
1898        }
1899    }
1900}
mithril_persistence/database/repository/cardano_transaction_repository.rs

mithril_persistence/database/repository/
cardano_transaction_repository.rs
