openzeppelin_relayer/repositories/transaction/
mod.rs

1//! Transaction Repository Module
2//!
3//! This module provides the transaction repository layer for the OpenZeppelin Relayer service.
4//! It implements the Repository pattern to abstract transaction data persistence operations,
5//! supporting both in-memory and Redis-backed storage implementations.
6//!
7//! ## Features
8//!
9//! - **CRUD Operations**: Create, read, update, and delete transactions
10//! - **Specialized Queries**: Find transactions by relayer ID, status, and nonce
11//! - **Pagination Support**: Efficient paginated listing of transactions
12//! - **Status Management**: Update transaction status and timestamps
13//! - **Partial Updates**: Support for partial transaction updates
14//! - **Network Data**: Manage transaction network-specific data
15//!
16//! ## Repository Implementations
17//!
18//! - [`InMemoryTransactionRepository`]: Fast in-memory storage for testing/development
19//! - [`RedisTransactionRepository`]: Redis-backed storage for production environments
20//!
21mod transaction_in_memory;
22mod transaction_redis;
23
24pub use transaction_in_memory::*;
25pub use transaction_redis::*;
26
27use crate::{
28    models::{
29        NetworkTransactionData, TransactionRepoModel, TransactionStatus, TransactionUpdateRequest,
30    },
31    repositories::{BatchDeleteResult, TransactionDeleteRequest, *},
32    utils::RedisConnections,
33};
34use async_trait::async_trait;
35use eyre::Result;
36use std::sync::Arc;
37
38/// A trait defining transaction repository operations
39#[async_trait]
40pub trait TransactionRepository: Repository<TransactionRepoModel, String> {
41    /// Returns underlying storage Redis connections when available.
42    ///
43    /// In-memory implementations return `None`.
44    fn connection_info(&self) -> Option<(Arc<RedisConnections>, String)> {
45        None
46    }
47
48    /// Find transactions by relayer ID with pagination
49    async fn find_by_relayer_id(
50        &self,
51        relayer_id: &str,
52        query: PaginationQuery,
53    ) -> Result<PaginatedResult<TransactionRepoModel>, RepositoryError>;
54
55    /// Find transactions by relayer ID and status(es).
56    ///
57    /// Results are sorted by created_at descending (newest first).
58    async fn find_by_status(
59        &self,
60        relayer_id: &str,
61        statuses: &[TransactionStatus],
62    ) -> Result<Vec<TransactionRepoModel>, RepositoryError>;
63
64    /// Find transactions by relayer ID and status(es) with pagination.
65    ///
66    /// Results are sorted by timestamp:
67    /// - For Confirmed transactions: sorted by confirmed_at (on-chain confirmation order)
68    /// - For all other statuses: sorted by created_at (queue/processing order)
69    ///
70    /// The `oldest_first` parameter controls sort direction:
71    /// - `false` (default): newest first (descending) - for displaying recent transactions
72    /// - `true`: oldest first (ascending) - for FIFO queue processing
73    ///
74    /// For multi-status queries, transactions are merged and sorted using the same rules,
75    /// ensuring consistent ordering across different statuses.
76    async fn find_by_status_paginated(
77        &self,
78        relayer_id: &str,
79        statuses: &[TransactionStatus],
80        query: PaginationQuery,
81        oldest_first: bool,
82    ) -> Result<PaginatedResult<TransactionRepoModel>, RepositoryError>;
83
84    /// Like [`find_by_status_paginated`], but when `exclude_canceled` is true,
85    /// transactions flagged `is_canceled` are removed BEFORE counting and pagination,
86    /// so the returned page and `total` stay consistent (no short/empty pages).
87    ///
88    /// Used by the listing API to hide cancelled-in-progress transactions from
89    /// active-status queries; internal callers use the unfiltered variant so they
90    /// still see the still-active replacement NOOP.
91    async fn find_by_status_paginated_filtered(
92        &self,
93        relayer_id: &str,
94        statuses: &[TransactionStatus],
95        query: PaginationQuery,
96        oldest_first: bool,
97        exclude_canceled: bool,
98    ) -> Result<PaginatedResult<TransactionRepoModel>, RepositoryError>;
99
100    /// Find a transaction by relayer ID and nonce
101    async fn find_by_nonce(
102        &self,
103        relayer_id: &str,
104        nonce: u64,
105    ) -> Result<Option<TransactionRepoModel>, RepositoryError>;
106
107    /// Returns the transaction status for each nonce in `[from_nonce, to_nonce)`.
108    ///
109    /// For each nonce, returns `Some(status)` if a transaction exists at that slot,
110    /// or `None` if the slot is empty. Implementations should batch I/O where possible
111    /// (e.g., Redis MGET) to minimize round trips.
112    async fn get_nonce_occupancy(
113        &self,
114        relayer_id: &str,
115        from_nonce: u64,
116        to_nonce: u64,
117    ) -> Result<Vec<(u64, Option<TransactionStatus>)>, RepositoryError>;
118
119    /// Update the status of a transaction
120    async fn update_status(
121        &self,
122        tx_id: String,
123        status: TransactionStatus,
124    ) -> Result<TransactionRepoModel, RepositoryError>;
125
126    /// Partially update a transaction
127    async fn partial_update(
128        &self,
129        tx_id: String,
130        update: TransactionUpdateRequest,
131    ) -> Result<TransactionRepoModel, RepositoryError>;
132
133    /// Repairs stale Redis status-index entries whose indexed status diverged from
134    /// the persisted transaction body.
135    ///
136    /// Backends where status indexes cannot diverge from transaction bodies should
137    /// use the default no-op implementation.
138    async fn reconcile_stale_status_indexes(
139        &self,
140        _relayer_id: &str,
141    ) -> Result<usize, RepositoryError> {
142        Ok(0)
143    }
144
145    /// Update the network data of a transaction
146    async fn update_network_data(
147        &self,
148        tx_id: String,
149        network_data: NetworkTransactionData,
150    ) -> Result<TransactionRepoModel, RepositoryError>;
151
152    /// Set the sent_at timestamp of a transaction
153    async fn set_sent_at(
154        &self,
155        tx_id: String,
156        sent_at: String,
157    ) -> Result<TransactionRepoModel, RepositoryError>;
158
159    /// Atomically increments status-check failure counters using the latest stored metadata.
160    async fn increment_status_check_failures(
161        &self,
162        tx_id: String,
163    ) -> Result<TransactionRepoModel, RepositoryError>;
164
165    /// Atomically resets consecutive status-check failures to zero while preserving other counters.
166    async fn reset_status_check_consecutive_failures(
167        &self,
168        tx_id: String,
169    ) -> Result<TransactionRepoModel, RepositoryError>;
170
171    /// Atomically sets `sent_at` and increments Stellar insufficient-fee retries.
172    async fn record_stellar_insufficient_fee_retry(
173        &self,
174        tx_id: String,
175        sent_at: String,
176    ) -> Result<TransactionRepoModel, RepositoryError>;
177
178    /// Atomically sets `sent_at` and increments Stellar try-again-later retries.
179    async fn record_stellar_try_again_later_retry(
180        &self,
181        tx_id: String,
182        sent_at: String,
183    ) -> Result<TransactionRepoModel, RepositoryError>;
184
185    /// Set the confirmed_at timestamp of a transaction
186    async fn set_confirmed_at(
187        &self,
188        tx_id: String,
189        confirmed_at: String,
190    ) -> Result<TransactionRepoModel, RepositoryError>;
191
192    /// Count transactions by status(es) without fetching full transaction data.
193    /// This is an optimized O(1) operation in Redis using ZCARD.
194    async fn count_by_status(
195        &self,
196        relayer_id: &str,
197        statuses: &[TransactionStatus],
198    ) -> Result<u64, RepositoryError>;
199
200    /// Delete multiple transactions by their IDs in a single batch operation.
201    ///
202    /// This is more efficient than calling `delete_by_id` multiple times as it
203    /// reduces the number of round-trips to the storage backend.
204    ///
205    /// Note: This method requires fetching transaction data first to clean up indexes.
206    /// If you already have transaction data, use `delete_by_requests` instead for
207    /// better performance.
208    ///
209    /// # Arguments
210    /// * `ids` - List of transaction IDs to delete
211    ///
212    /// # Returns
213    /// * `BatchDeleteResult` containing the count of successful deletions and any failures
214    async fn delete_by_ids(&self, ids: Vec<String>) -> Result<BatchDeleteResult, RepositoryError>;
215
216    /// Delete multiple transactions using pre-extracted data.
217    ///
218    /// This is the most efficient batch delete method as it doesn't require
219    /// re-fetching transaction data. Use this when you already have the transaction
220    /// data (e.g., from a previous query).
221    ///
222    /// # Arguments
223    /// * `requests` - List of delete requests containing transaction data needed for cleanup
224    ///
225    /// # Returns
226    /// * `BatchDeleteResult` containing the count of successful deletions and any failures
227    async fn delete_by_requests(
228        &self,
229        requests: Vec<TransactionDeleteRequest>,
230    ) -> Result<BatchDeleteResult, RepositoryError>;
231}
232
233#[cfg(test)]
234mockall::mock! {
235  pub TransactionRepository {}
236
237  #[async_trait]
238  impl Repository<TransactionRepoModel, String> for TransactionRepository {
239      async fn create(&self, entity: TransactionRepoModel) -> Result<TransactionRepoModel, RepositoryError>;
240      async fn get_by_id(&self, id: String) -> Result<TransactionRepoModel, RepositoryError>;
241      async fn list_all(&self) -> Result<Vec<TransactionRepoModel>, RepositoryError>;
242      async fn list_paginated(&self, query: PaginationQuery) -> Result<PaginatedResult<TransactionRepoModel>, RepositoryError>;
243      async fn update(&self, id: String, entity: TransactionRepoModel) -> Result<TransactionRepoModel, RepositoryError>;
244      async fn delete_by_id(&self, id: String) -> Result<(), RepositoryError>;
245      async fn count(&self) -> Result<usize, RepositoryError>;
246      async fn has_entries(&self) -> Result<bool, RepositoryError>;
247      async fn drop_all_entries(&self) -> Result<(), RepositoryError>;
248  }
249
250  #[async_trait]
251  impl TransactionRepository for TransactionRepository {
252      fn connection_info(&self) -> Option<(Arc<RedisConnections>, String)>;
253      async fn find_by_relayer_id(&self, relayer_id: &str, query: PaginationQuery) -> Result<PaginatedResult<TransactionRepoModel>, RepositoryError>;
254      async fn find_by_status(&self, relayer_id: &str, statuses: &[TransactionStatus]) -> Result<Vec<TransactionRepoModel>, RepositoryError>;
255      async fn find_by_status_paginated(&self, relayer_id: &str, statuses: &[TransactionStatus], query: PaginationQuery, oldest_first: bool) -> Result<PaginatedResult<TransactionRepoModel>, RepositoryError>;
256      async fn find_by_status_paginated_filtered(&self, relayer_id: &str, statuses: &[TransactionStatus], query: PaginationQuery, oldest_first: bool, exclude_canceled: bool) -> Result<PaginatedResult<TransactionRepoModel>, RepositoryError>;
257      async fn find_by_nonce(&self, relayer_id: &str, nonce: u64) -> Result<Option<TransactionRepoModel>, RepositoryError>;
258      async fn get_nonce_occupancy(&self, relayer_id: &str, from_nonce: u64, to_nonce: u64) -> Result<Vec<(u64, Option<TransactionStatus>)>, RepositoryError>;
259      async fn update_status(&self, tx_id: String, status: TransactionStatus) -> Result<TransactionRepoModel, RepositoryError>;
260      async fn partial_update(&self, tx_id: String, update: TransactionUpdateRequest) -> Result<TransactionRepoModel, RepositoryError>;
261      async fn reconcile_stale_status_indexes(&self, relayer_id: &str) -> Result<usize, RepositoryError>;
262      async fn update_network_data(&self, tx_id: String, network_data: NetworkTransactionData) -> Result<TransactionRepoModel, RepositoryError>;
263      async fn set_sent_at(&self, tx_id: String, sent_at: String) -> Result<TransactionRepoModel, RepositoryError>;
264      async fn increment_status_check_failures(&self, tx_id: String) -> Result<TransactionRepoModel, RepositoryError>;
265      async fn reset_status_check_consecutive_failures(&self, tx_id: String) -> Result<TransactionRepoModel, RepositoryError>;
266      async fn record_stellar_insufficient_fee_retry(&self, tx_id: String, sent_at: String) -> Result<TransactionRepoModel, RepositoryError>;
267      async fn record_stellar_try_again_later_retry(&self, tx_id: String, sent_at: String) -> Result<TransactionRepoModel, RepositoryError>;
268      async fn set_confirmed_at(&self, tx_id: String, confirmed_at: String) -> Result<TransactionRepoModel, RepositoryError>;
269      async fn count_by_status(&self, relayer_id: &str, statuses: &[TransactionStatus]) -> Result<u64, RepositoryError>;
270      async fn delete_by_ids(&self, ids: Vec<String>) -> Result<BatchDeleteResult, RepositoryError>;
271      async fn delete_by_requests(&self, requests: Vec<TransactionDeleteRequest>) -> Result<BatchDeleteResult, RepositoryError>;
272  }
273}
274
275/// Enum wrapper for different transaction repository implementations
276#[derive(Debug, Clone)]
277pub enum TransactionRepositoryStorage {
278    InMemory(InMemoryTransactionRepository),
279    Redis(RedisTransactionRepository),
280}
281
282impl TransactionRepositoryStorage {
283    pub fn new_in_memory() -> Self {
284        Self::InMemory(InMemoryTransactionRepository::new())
285    }
286    pub fn new_redis(
287        connections: Arc<RedisConnections>,
288        key_prefix: String,
289    ) -> Result<Self, RepositoryError> {
290        Ok(Self::Redis(RedisTransactionRepository::new(
291            connections,
292            key_prefix,
293        )?))
294    }
295
296    /// Returns underlying Redis connections if this is a persistent storage backend.
297    ///
298    /// This is useful for operations that need direct storage access, such as
299    /// distributed locking and health checks.
300    ///
301    /// # Returns
302    /// * `Some((connections, key_prefix))` - If using persistent Redis storage
303    /// * `None` - If using in-memory storage
304    pub fn connection_info(&self) -> Option<(Arc<RedisConnections>, &str)> {
305        match self {
306            TransactionRepositoryStorage::InMemory(_) => None,
307            TransactionRepositoryStorage::Redis(repo) => {
308                Some((repo.connections.clone(), &repo.key_prefix))
309            }
310        }
311    }
312
313    /// Returns key prefix used by persistent storage backends.
314    pub fn key_prefix(&self) -> Option<&str> {
315        match self {
316            TransactionRepositoryStorage::InMemory(_) => None,
317            TransactionRepositoryStorage::Redis(repo) => Some(&repo.key_prefix),
318        }
319    }
320}
321
322#[async_trait]
323impl TransactionRepository for TransactionRepositoryStorage {
324    fn connection_info(&self) -> Option<(Arc<RedisConnections>, String)> {
325        TransactionRepositoryStorage::connection_info(self)
326            .map(|(connections, key_prefix)| (connections, key_prefix.to_string()))
327    }
328
329    async fn find_by_relayer_id(
330        &self,
331        relayer_id: &str,
332        query: PaginationQuery,
333    ) -> Result<PaginatedResult<TransactionRepoModel>, RepositoryError> {
334        match self {
335            TransactionRepositoryStorage::InMemory(repo) => {
336                repo.find_by_relayer_id(relayer_id, query).await
337            }
338            TransactionRepositoryStorage::Redis(repo) => {
339                repo.find_by_relayer_id(relayer_id, query).await
340            }
341        }
342    }
343
344    async fn find_by_status(
345        &self,
346        relayer_id: &str,
347        statuses: &[TransactionStatus],
348    ) -> Result<Vec<TransactionRepoModel>, RepositoryError> {
349        match self {
350            TransactionRepositoryStorage::InMemory(repo) => {
351                repo.find_by_status(relayer_id, statuses).await
352            }
353            TransactionRepositoryStorage::Redis(repo) => {
354                repo.find_by_status(relayer_id, statuses).await
355            }
356        }
357    }
358
359    async fn find_by_status_paginated(
360        &self,
361        relayer_id: &str,
362        statuses: &[TransactionStatus],
363        query: PaginationQuery,
364        oldest_first: bool,
365    ) -> Result<PaginatedResult<TransactionRepoModel>, RepositoryError> {
366        match self {
367            TransactionRepositoryStorage::InMemory(repo) => {
368                repo.find_by_status_paginated(relayer_id, statuses, query, oldest_first)
369                    .await
370            }
371            TransactionRepositoryStorage::Redis(repo) => {
372                repo.find_by_status_paginated(relayer_id, statuses, query, oldest_first)
373                    .await
374            }
375        }
376    }
377
378    async fn find_by_status_paginated_filtered(
379        &self,
380        relayer_id: &str,
381        statuses: &[TransactionStatus],
382        query: PaginationQuery,
383        oldest_first: bool,
384        exclude_canceled: bool,
385    ) -> Result<PaginatedResult<TransactionRepoModel>, RepositoryError> {
386        match self {
387            TransactionRepositoryStorage::InMemory(repo) => {
388                repo.find_by_status_paginated_filtered(
389                    relayer_id,
390                    statuses,
391                    query,
392                    oldest_first,
393                    exclude_canceled,
394                )
395                .await
396            }
397            TransactionRepositoryStorage::Redis(repo) => {
398                repo.find_by_status_paginated_filtered(
399                    relayer_id,
400                    statuses,
401                    query,
402                    oldest_first,
403                    exclude_canceled,
404                )
405                .await
406            }
407        }
408    }
409
410    async fn find_by_nonce(
411        &self,
412        relayer_id: &str,
413        nonce: u64,
414    ) -> Result<Option<TransactionRepoModel>, RepositoryError> {
415        match self {
416            TransactionRepositoryStorage::InMemory(repo) => {
417                repo.find_by_nonce(relayer_id, nonce).await
418            }
419            TransactionRepositoryStorage::Redis(repo) => {
420                repo.find_by_nonce(relayer_id, nonce).await
421            }
422        }
423    }
424
425    async fn get_nonce_occupancy(
426        &self,
427        relayer_id: &str,
428        from_nonce: u64,
429        to_nonce: u64,
430    ) -> Result<Vec<(u64, Option<TransactionStatus>)>, RepositoryError> {
431        match self {
432            TransactionRepositoryStorage::InMemory(repo) => {
433                repo.get_nonce_occupancy(relayer_id, from_nonce, to_nonce)
434                    .await
435            }
436            TransactionRepositoryStorage::Redis(repo) => {
437                repo.get_nonce_occupancy(relayer_id, from_nonce, to_nonce)
438                    .await
439            }
440        }
441    }
442
443    async fn update_status(
444        &self,
445        tx_id: String,
446        status: TransactionStatus,
447    ) -> Result<TransactionRepoModel, RepositoryError> {
448        match self {
449            TransactionRepositoryStorage::InMemory(repo) => repo.update_status(tx_id, status).await,
450            TransactionRepositoryStorage::Redis(repo) => repo.update_status(tx_id, status).await,
451        }
452    }
453
454    async fn partial_update(
455        &self,
456        tx_id: String,
457        update: TransactionUpdateRequest,
458    ) -> Result<TransactionRepoModel, RepositoryError> {
459        match self {
460            TransactionRepositoryStorage::InMemory(repo) => {
461                repo.partial_update(tx_id, update).await
462            }
463            TransactionRepositoryStorage::Redis(repo) => repo.partial_update(tx_id, update).await,
464        }
465    }
466
467    async fn reconcile_stale_status_indexes(
468        &self,
469        relayer_id: &str,
470    ) -> Result<usize, RepositoryError> {
471        match self {
472            TransactionRepositoryStorage::InMemory(repo) => {
473                repo.reconcile_stale_status_indexes(relayer_id).await
474            }
475            TransactionRepositoryStorage::Redis(repo) => {
476                repo.reconcile_stale_status_indexes(relayer_id).await
477            }
478        }
479    }
480
481    async fn update_network_data(
482        &self,
483        tx_id: String,
484        network_data: NetworkTransactionData,
485    ) -> Result<TransactionRepoModel, RepositoryError> {
486        match self {
487            TransactionRepositoryStorage::InMemory(repo) => {
488                repo.update_network_data(tx_id, network_data).await
489            }
490            TransactionRepositoryStorage::Redis(repo) => {
491                repo.update_network_data(tx_id, network_data).await
492            }
493        }
494    }
495
496    async fn set_sent_at(
497        &self,
498        tx_id: String,
499        sent_at: String,
500    ) -> Result<TransactionRepoModel, RepositoryError> {
501        match self {
502            TransactionRepositoryStorage::InMemory(repo) => repo.set_sent_at(tx_id, sent_at).await,
503            TransactionRepositoryStorage::Redis(repo) => repo.set_sent_at(tx_id, sent_at).await,
504        }
505    }
506
507    async fn increment_status_check_failures(
508        &self,
509        tx_id: String,
510    ) -> Result<TransactionRepoModel, RepositoryError> {
511        match self {
512            TransactionRepositoryStorage::InMemory(repo) => {
513                repo.increment_status_check_failures(tx_id).await
514            }
515            TransactionRepositoryStorage::Redis(repo) => {
516                repo.increment_status_check_failures(tx_id).await
517            }
518        }
519    }
520
521    async fn reset_status_check_consecutive_failures(
522        &self,
523        tx_id: String,
524    ) -> Result<TransactionRepoModel, RepositoryError> {
525        match self {
526            TransactionRepositoryStorage::InMemory(repo) => {
527                repo.reset_status_check_consecutive_failures(tx_id).await
528            }
529            TransactionRepositoryStorage::Redis(repo) => {
530                repo.reset_status_check_consecutive_failures(tx_id).await
531            }
532        }
533    }
534
535    async fn record_stellar_insufficient_fee_retry(
536        &self,
537        tx_id: String,
538        sent_at: String,
539    ) -> Result<TransactionRepoModel, RepositoryError> {
540        match self {
541            TransactionRepositoryStorage::InMemory(repo) => {
542                repo.record_stellar_insufficient_fee_retry(tx_id, sent_at)
543                    .await
544            }
545            TransactionRepositoryStorage::Redis(repo) => {
546                repo.record_stellar_insufficient_fee_retry(tx_id, sent_at)
547                    .await
548            }
549        }
550    }
551
552    async fn record_stellar_try_again_later_retry(
553        &self,
554        tx_id: String,
555        sent_at: String,
556    ) -> Result<TransactionRepoModel, RepositoryError> {
557        match self {
558            TransactionRepositoryStorage::InMemory(repo) => {
559                repo.record_stellar_try_again_later_retry(tx_id, sent_at)
560                    .await
561            }
562            TransactionRepositoryStorage::Redis(repo) => {
563                repo.record_stellar_try_again_later_retry(tx_id, sent_at)
564                    .await
565            }
566        }
567    }
568
569    async fn set_confirmed_at(
570        &self,
571        tx_id: String,
572        confirmed_at: String,
573    ) -> Result<TransactionRepoModel, RepositoryError> {
574        match self {
575            TransactionRepositoryStorage::InMemory(repo) => {
576                repo.set_confirmed_at(tx_id, confirmed_at).await
577            }
578            TransactionRepositoryStorage::Redis(repo) => {
579                repo.set_confirmed_at(tx_id, confirmed_at).await
580            }
581        }
582    }
583
584    async fn count_by_status(
585        &self,
586        relayer_id: &str,
587        statuses: &[TransactionStatus],
588    ) -> Result<u64, RepositoryError> {
589        match self {
590            TransactionRepositoryStorage::InMemory(repo) => {
591                repo.count_by_status(relayer_id, statuses).await
592            }
593            TransactionRepositoryStorage::Redis(repo) => {
594                repo.count_by_status(relayer_id, statuses).await
595            }
596        }
597    }
598
599    async fn delete_by_ids(&self, ids: Vec<String>) -> Result<BatchDeleteResult, RepositoryError> {
600        match self {
601            TransactionRepositoryStorage::InMemory(repo) => repo.delete_by_ids(ids).await,
602            TransactionRepositoryStorage::Redis(repo) => repo.delete_by_ids(ids).await,
603        }
604    }
605
606    async fn delete_by_requests(
607        &self,
608        requests: Vec<TransactionDeleteRequest>,
609    ) -> Result<BatchDeleteResult, RepositoryError> {
610        match self {
611            TransactionRepositoryStorage::InMemory(repo) => repo.delete_by_requests(requests).await,
612            TransactionRepositoryStorage::Redis(repo) => repo.delete_by_requests(requests).await,
613        }
614    }
615}
616
617#[async_trait]
618impl Repository<TransactionRepoModel, String> for TransactionRepositoryStorage {
619    async fn create(
620        &self,
621        entity: TransactionRepoModel,
622    ) -> Result<TransactionRepoModel, RepositoryError> {
623        match self {
624            TransactionRepositoryStorage::InMemory(repo) => repo.create(entity).await,
625            TransactionRepositoryStorage::Redis(repo) => repo.create(entity).await,
626        }
627    }
628
629    async fn get_by_id(&self, id: String) -> Result<TransactionRepoModel, RepositoryError> {
630        match self {
631            TransactionRepositoryStorage::InMemory(repo) => repo.get_by_id(id).await,
632            TransactionRepositoryStorage::Redis(repo) => repo.get_by_id(id).await,
633        }
634    }
635
636    async fn list_all(&self) -> Result<Vec<TransactionRepoModel>, RepositoryError> {
637        match self {
638            TransactionRepositoryStorage::InMemory(repo) => repo.list_all().await,
639            TransactionRepositoryStorage::Redis(repo) => repo.list_all().await,
640        }
641    }
642
643    async fn list_paginated(
644        &self,
645        query: PaginationQuery,
646    ) -> Result<PaginatedResult<TransactionRepoModel>, RepositoryError> {
647        match self {
648            TransactionRepositoryStorage::InMemory(repo) => repo.list_paginated(query).await,
649            TransactionRepositoryStorage::Redis(repo) => repo.list_paginated(query).await,
650        }
651    }
652
653    async fn update(
654        &self,
655        id: String,
656        entity: TransactionRepoModel,
657    ) -> Result<TransactionRepoModel, RepositoryError> {
658        match self {
659            TransactionRepositoryStorage::InMemory(repo) => repo.update(id, entity).await,
660            TransactionRepositoryStorage::Redis(repo) => repo.update(id, entity).await,
661        }
662    }
663
664    async fn delete_by_id(&self, id: String) -> Result<(), RepositoryError> {
665        match self {
666            TransactionRepositoryStorage::InMemory(repo) => repo.delete_by_id(id).await,
667            TransactionRepositoryStorage::Redis(repo) => repo.delete_by_id(id).await,
668        }
669    }
670
671    async fn count(&self) -> Result<usize, RepositoryError> {
672        match self {
673            TransactionRepositoryStorage::InMemory(repo) => repo.count().await,
674            TransactionRepositoryStorage::Redis(repo) => repo.count().await,
675        }
676    }
677
678    async fn has_entries(&self) -> Result<bool, RepositoryError> {
679        match self {
680            TransactionRepositoryStorage::InMemory(repo) => repo.has_entries().await,
681            TransactionRepositoryStorage::Redis(repo) => repo.has_entries().await,
682        }
683    }
684
685    async fn drop_all_entries(&self) -> Result<(), RepositoryError> {
686        match self {
687            TransactionRepositoryStorage::InMemory(repo) => repo.drop_all_entries().await,
688            TransactionRepositoryStorage::Redis(repo) => repo.drop_all_entries().await,
689        }
690    }
691}
692
693#[cfg(test)]
694mod tests {
695    use chrono::Utc;
696    use color_eyre::Result;
697    use deadpool_redis::{Config, Runtime};
698
699    use super::*;
700    use crate::models::{
701        EvmTransactionData, NetworkTransactionData, TransactionStatus, TransactionUpdateRequest,
702    };
703    use crate::repositories::PaginationQuery;
704    use crate::utils::mocks::mockutils::create_mock_transaction;
705
706    fn create_test_transaction(id: &str, relayer_id: &str) -> TransactionRepoModel {
707        let mut transaction = create_mock_transaction();
708        transaction.id = id.to_string();
709        transaction.relayer_id = relayer_id.to_string();
710        transaction
711    }
712
713    fn create_test_transaction_with_status(
714        id: &str,
715        relayer_id: &str,
716        status: TransactionStatus,
717    ) -> TransactionRepoModel {
718        let mut transaction = create_test_transaction(id, relayer_id);
719        transaction.status = status;
720        transaction
721    }
722
723    fn create_test_transaction_with_nonce(
724        id: &str,
725        relayer_id: &str,
726        nonce: u64,
727    ) -> TransactionRepoModel {
728        let mut transaction = create_test_transaction(id, relayer_id);
729        if let NetworkTransactionData::Evm(ref mut evm_data) = transaction.network_data {
730            evm_data.nonce = Some(nonce);
731        }
732        transaction
733    }
734
735    fn create_test_update_request() -> TransactionUpdateRequest {
736        TransactionUpdateRequest {
737            status: Some(TransactionStatus::Sent),
738            status_reason: Some("Test reason".to_string()),
739            sent_at: Some(Utc::now().to_string()),
740            confirmed_at: None,
741            network_data: None,
742            priced_at: None,
743            hashes: Some(vec!["test_hash".to_string()]),
744            noop_count: None,
745            is_canceled: None,
746            delete_at: None,
747            metadata: None,
748        }
749    }
750
751    #[tokio::test]
752    async fn test_new_in_memory() {
753        let storage = TransactionRepositoryStorage::new_in_memory();
754
755        match storage {
756            TransactionRepositoryStorage::InMemory(_) => {
757                // Success - verify it's the InMemory variant
758            }
759            TransactionRepositoryStorage::Redis(_) => {
760                panic!("Expected InMemory variant, got Redis");
761            }
762        }
763    }
764
765    #[tokio::test]
766    async fn test_connection_info_returns_none_for_in_memory() {
767        let storage = TransactionRepositoryStorage::new_in_memory();
768
769        // In-memory storage should return None for connection_info
770        assert!(storage.connection_info().is_none());
771    }
772
773    #[tokio::test]
774    #[ignore = "Requires active Redis instance"]
775    async fn test_connection_info_returns_some_for_redis() -> Result<()> {
776        let redis_url = std::env::var("REDIS_TEST_URL")
777            .unwrap_or_else(|_| "redis://127.0.0.1:6379".to_string());
778        let cfg = Config::from_url(&redis_url);
779        let pool = Arc::new(
780            cfg.builder()
781                .map_err(|e| eyre::eyre!("Failed to create Redis pool builder: {}", e))?
782                .max_size(16)
783                .runtime(Runtime::Tokio1)
784                .build()
785                .map_err(|e| eyre::eyre!("Failed to build Redis pool: {}", e))?,
786        );
787        let connections = Arc::new(RedisConnections::new_single_pool(pool.clone()));
788        let key_prefix = "test_prefix".to_string();
789
790        let storage = TransactionRepositoryStorage::new_redis(connections, key_prefix.clone())?;
791
792        let (returned_connection, returned_prefix) = storage
793            .connection_info()
794            .expect("Expected Redis connection info");
795
796        assert!(Arc::ptr_eq(&pool, returned_connection.primary()));
797        assert_eq!(returned_prefix, key_prefix);
798
799        Ok(())
800    }
801
802    #[tokio::test]
803    async fn test_create_in_memory() -> Result<()> {
804        let storage = TransactionRepositoryStorage::new_in_memory();
805        let transaction = create_test_transaction("test-tx", "test-relayer");
806
807        let created = storage.create(transaction.clone()).await?;
808        assert_eq!(created.id, transaction.id);
809        assert_eq!(created.relayer_id, transaction.relayer_id);
810        assert_eq!(created.status, transaction.status);
811
812        Ok(())
813    }
814
815    #[tokio::test]
816    async fn test_get_by_id_in_memory() -> Result<()> {
817        let storage = TransactionRepositoryStorage::new_in_memory();
818        let transaction = create_test_transaction("test-tx", "test-relayer");
819
820        // Create transaction first
821        storage.create(transaction.clone()).await?;
822
823        // Get by ID
824        let retrieved = storage.get_by_id("test-tx".to_string()).await?;
825        assert_eq!(retrieved.id, transaction.id);
826        assert_eq!(retrieved.relayer_id, transaction.relayer_id);
827        assert_eq!(retrieved.status, transaction.status);
828
829        Ok(())
830    }
831
832    #[tokio::test]
833    async fn test_get_by_id_not_found_in_memory() -> Result<()> {
834        let storage = TransactionRepositoryStorage::new_in_memory();
835
836        let result = storage.get_by_id("non-existent".to_string()).await;
837        assert!(result.is_err());
838
839        Ok(())
840    }
841
842    #[tokio::test]
843    async fn test_list_all_in_memory() -> Result<()> {
844        let storage = TransactionRepositoryStorage::new_in_memory();
845
846        // Initially empty
847        let transactions = storage.list_all().await?;
848        assert!(transactions.is_empty());
849
850        // Add transactions
851        let tx1 = create_test_transaction("tx-1", "relayer-1");
852        let tx2 = create_test_transaction("tx-2", "relayer-2");
853
854        storage.create(tx1.clone()).await?;
855        storage.create(tx2.clone()).await?;
856
857        let all_transactions = storage.list_all().await?;
858        assert_eq!(all_transactions.len(), 2);
859
860        let ids: Vec<&str> = all_transactions.iter().map(|t| t.id.as_str()).collect();
861        assert!(ids.contains(&"tx-1"));
862        assert!(ids.contains(&"tx-2"));
863
864        Ok(())
865    }
866
867    #[tokio::test]
868    async fn test_list_paginated_in_memory() -> Result<()> {
869        let storage = TransactionRepositoryStorage::new_in_memory();
870
871        // Add test transactions
872        for i in 1..=5 {
873            let tx = create_test_transaction(&format!("tx-{i}"), "test-relayer");
874            storage.create(tx).await?;
875        }
876
877        // Test pagination
878        let query = PaginationQuery {
879            page: 1,
880            per_page: 2,
881        };
882        let page = storage.list_paginated(query).await?;
883
884        assert_eq!(page.items.len(), 2);
885        assert_eq!(page.total, 5);
886        assert_eq!(page.page, 1);
887        assert_eq!(page.per_page, 2);
888
889        // Test second page
890        let query2 = PaginationQuery {
891            page: 2,
892            per_page: 2,
893        };
894        let page2 = storage.list_paginated(query2).await?;
895
896        assert_eq!(page2.items.len(), 2);
897        assert_eq!(page2.total, 5);
898        assert_eq!(page2.page, 2);
899        assert_eq!(page2.per_page, 2);
900
901        Ok(())
902    }
903
904    #[tokio::test]
905    async fn test_update_in_memory() -> Result<()> {
906        let storage = TransactionRepositoryStorage::new_in_memory();
907        let transaction = create_test_transaction("test-tx", "test-relayer");
908
909        // Create transaction first
910        storage.create(transaction.clone()).await?;
911
912        // Update it
913        let mut updated_transaction = transaction.clone();
914        updated_transaction.status = TransactionStatus::Sent;
915        updated_transaction.status_reason = Some("Updated reason".to_string());
916
917        let result = storage
918            .update("test-tx".to_string(), updated_transaction.clone())
919            .await?;
920        assert_eq!(result.id, "test-tx");
921        assert_eq!(result.status, TransactionStatus::Sent);
922        assert_eq!(result.status_reason, Some("Updated reason".to_string()));
923
924        // Verify the update persisted
925        let retrieved = storage.get_by_id("test-tx".to_string()).await?;
926        assert_eq!(retrieved.status, TransactionStatus::Sent);
927        assert_eq!(retrieved.status_reason, Some("Updated reason".to_string()));
928
929        Ok(())
930    }
931
932    #[tokio::test]
933    async fn test_update_not_found_in_memory() -> Result<()> {
934        let storage = TransactionRepositoryStorage::new_in_memory();
935        let transaction = create_test_transaction("non-existent", "test-relayer");
936
937        let result = storage
938            .update("non-existent".to_string(), transaction)
939            .await;
940        assert!(result.is_err());
941
942        Ok(())
943    }
944
945    #[tokio::test]
946    async fn test_delete_by_id_in_memory() -> Result<()> {
947        let storage = TransactionRepositoryStorage::new_in_memory();
948        let transaction = create_test_transaction("test-tx", "test-relayer");
949
950        // Create transaction first
951        storage.create(transaction.clone()).await?;
952
953        // Verify it exists
954        let retrieved = storage.get_by_id("test-tx".to_string()).await?;
955        assert_eq!(retrieved.id, "test-tx");
956
957        // Delete it
958        storage.delete_by_id("test-tx".to_string()).await?;
959
960        // Verify it's gone
961        let result = storage.get_by_id("test-tx".to_string()).await;
962        assert!(result.is_err());
963
964        Ok(())
965    }
966
967    #[tokio::test]
968    async fn test_delete_by_id_not_found_in_memory() -> Result<()> {
969        let storage = TransactionRepositoryStorage::new_in_memory();
970
971        let result = storage.delete_by_id("non-existent".to_string()).await;
972        assert!(result.is_err());
973
974        Ok(())
975    }
976
977    #[tokio::test]
978    async fn test_count_in_memory() -> Result<()> {
979        let storage = TransactionRepositoryStorage::new_in_memory();
980
981        // Initially empty
982        let count = storage.count().await?;
983        assert_eq!(count, 0);
984
985        // Add transactions
986        let tx1 = create_test_transaction("tx-1", "relayer-1");
987        let tx2 = create_test_transaction("tx-2", "relayer-2");
988
989        storage.create(tx1).await?;
990        let count_after_one = storage.count().await?;
991        assert_eq!(count_after_one, 1);
992
993        storage.create(tx2).await?;
994        let count_after_two = storage.count().await?;
995        assert_eq!(count_after_two, 2);
996
997        // Delete one
998        storage.delete_by_id("tx-1".to_string()).await?;
999        let count_after_delete = storage.count().await?;
1000        assert_eq!(count_after_delete, 1);
1001
1002        Ok(())
1003    }
1004
1005    #[tokio::test]
1006    async fn test_has_entries_in_memory() -> Result<()> {
1007        let storage = TransactionRepositoryStorage::new_in_memory();
1008
1009        // Initially empty
1010        let has_entries = storage.has_entries().await?;
1011        assert!(!has_entries);
1012
1013        // Add transaction
1014        let transaction = create_test_transaction("test-tx", "test-relayer");
1015        storage.create(transaction).await?;
1016
1017        let has_entries_after_create = storage.has_entries().await?;
1018        assert!(has_entries_after_create);
1019
1020        // Delete transaction
1021        storage.delete_by_id("test-tx".to_string()).await?;
1022
1023        let has_entries_after_delete = storage.has_entries().await?;
1024        assert!(!has_entries_after_delete);
1025
1026        Ok(())
1027    }
1028
1029    #[tokio::test]
1030    async fn test_drop_all_entries_in_memory() -> Result<()> {
1031        let storage = TransactionRepositoryStorage::new_in_memory();
1032
1033        // Add multiple transactions
1034        for i in 1..=5 {
1035            let tx = create_test_transaction(&format!("tx-{i}"), "test-relayer");
1036            storage.create(tx).await?;
1037        }
1038
1039        // Verify they exist
1040        let count_before = storage.count().await?;
1041        assert_eq!(count_before, 5);
1042
1043        let has_entries_before = storage.has_entries().await?;
1044        assert!(has_entries_before);
1045
1046        // Drop all entries
1047        storage.drop_all_entries().await?;
1048
1049        // Verify they're gone
1050        let count_after = storage.count().await?;
1051        assert_eq!(count_after, 0);
1052
1053        let has_entries_after = storage.has_entries().await?;
1054        assert!(!has_entries_after);
1055
1056        let all_transactions = storage.list_all().await?;
1057        assert!(all_transactions.is_empty());
1058
1059        Ok(())
1060    }
1061
1062    #[tokio::test]
1063    async fn test_find_by_relayer_id_in_memory() -> Result<()> {
1064        let storage = TransactionRepositoryStorage::new_in_memory();
1065
1066        // Add transactions for different relayers
1067        let tx1 = create_test_transaction("tx-1", "relayer-1");
1068        let tx2 = create_test_transaction("tx-2", "relayer-1");
1069        let tx3 = create_test_transaction("tx-3", "relayer-2");
1070
1071        storage.create(tx1).await?;
1072        storage.create(tx2).await?;
1073        storage.create(tx3).await?;
1074
1075        // Find by relayer ID
1076        let query = PaginationQuery {
1077            page: 1,
1078            per_page: 10,
1079        };
1080        let result = storage.find_by_relayer_id("relayer-1", query).await?;
1081
1082        assert_eq!(result.items.len(), 2);
1083        assert_eq!(result.total, 2);
1084
1085        // Verify all transactions belong to relayer-1
1086        for tx in result.items {
1087            assert_eq!(tx.relayer_id, "relayer-1");
1088        }
1089
1090        Ok(())
1091    }
1092
1093    #[tokio::test]
1094    async fn test_find_by_status_in_memory() -> Result<()> {
1095        let storage = TransactionRepositoryStorage::new_in_memory();
1096
1097        // Add transactions with different statuses
1098        let tx1 =
1099            create_test_transaction_with_status("tx-1", "relayer-1", TransactionStatus::Pending);
1100        let tx2 = create_test_transaction_with_status("tx-2", "relayer-1", TransactionStatus::Sent);
1101        let tx3 =
1102            create_test_transaction_with_status("tx-3", "relayer-1", TransactionStatus::Pending);
1103        let tx4 =
1104            create_test_transaction_with_status("tx-4", "relayer-2", TransactionStatus::Pending);
1105
1106        storage.create(tx1).await?;
1107        storage.create(tx2).await?;
1108        storage.create(tx3).await?;
1109        storage.create(tx4).await?;
1110
1111        // Find by status
1112        let statuses = vec![TransactionStatus::Pending];
1113        let result = storage.find_by_status("relayer-1", &statuses).await?;
1114
1115        assert_eq!(result.len(), 2);
1116
1117        // Verify all transactions have Pending status and belong to relayer-1
1118        for tx in result {
1119            assert_eq!(tx.status, TransactionStatus::Pending);
1120            assert_eq!(tx.relayer_id, "relayer-1");
1121        }
1122
1123        Ok(())
1124    }
1125
1126    #[tokio::test]
1127    async fn test_find_by_nonce_in_memory() -> Result<()> {
1128        let storage = TransactionRepositoryStorage::new_in_memory();
1129
1130        // Add transactions with different nonces
1131        let tx1 = create_test_transaction_with_nonce("tx-1", "relayer-1", 10);
1132        let tx2 = create_test_transaction_with_nonce("tx-2", "relayer-1", 20);
1133        let tx3 = create_test_transaction_with_nonce("tx-3", "relayer-2", 10);
1134
1135        storage.create(tx1).await?;
1136        storage.create(tx2).await?;
1137        storage.create(tx3).await?;
1138
1139        // Find by nonce
1140        let result = storage.find_by_nonce("relayer-1", 10).await?;
1141
1142        assert!(result.is_some());
1143        let found_tx = result.unwrap();
1144        assert_eq!(found_tx.id, "tx-1");
1145        assert_eq!(found_tx.relayer_id, "relayer-1");
1146
1147        // Check EVM nonce
1148        if let NetworkTransactionData::Evm(evm_data) = found_tx.network_data {
1149            assert_eq!(evm_data.nonce, Some(10));
1150        }
1151
1152        // Test not found
1153        let not_found = storage.find_by_nonce("relayer-1", 99).await?;
1154        assert!(not_found.is_none());
1155
1156        Ok(())
1157    }
1158
1159    #[tokio::test]
1160    async fn test_update_status_in_memory() -> Result<()> {
1161        let storage = TransactionRepositoryStorage::new_in_memory();
1162        let transaction = create_test_transaction("test-tx", "test-relayer");
1163
1164        // Create transaction first
1165        storage.create(transaction).await?;
1166
1167        // Update status
1168        let updated = storage
1169            .update_status("test-tx".to_string(), TransactionStatus::Sent)
1170            .await?;
1171
1172        assert_eq!(updated.id, "test-tx");
1173        assert_eq!(updated.status, TransactionStatus::Sent);
1174
1175        // Verify the update persisted
1176        let retrieved = storage.get_by_id("test-tx".to_string()).await?;
1177        assert_eq!(retrieved.status, TransactionStatus::Sent);
1178
1179        Ok(())
1180    }
1181
1182    #[tokio::test]
1183    async fn test_partial_update_in_memory() -> Result<()> {
1184        let storage = TransactionRepositoryStorage::new_in_memory();
1185        let transaction = create_test_transaction("test-tx", "test-relayer");
1186
1187        // Create transaction first
1188        storage.create(transaction).await?;
1189
1190        // Partial update
1191        let update_request = create_test_update_request();
1192        let updated = storage
1193            .partial_update("test-tx".to_string(), update_request)
1194            .await?;
1195
1196        assert_eq!(updated.id, "test-tx");
1197        assert_eq!(updated.status, TransactionStatus::Sent);
1198        assert_eq!(updated.status_reason, Some("Test reason".to_string()));
1199        assert!(updated.sent_at.is_some());
1200        assert_eq!(updated.hashes, vec!["test_hash".to_string()]);
1201
1202        Ok(())
1203    }
1204
1205    #[tokio::test]
1206    async fn test_update_network_data_in_memory() -> Result<()> {
1207        let storage = TransactionRepositoryStorage::new_in_memory();
1208        let transaction = create_test_transaction("test-tx", "test-relayer");
1209
1210        // Create transaction first
1211        storage.create(transaction).await?;
1212
1213        // Update network data
1214        let new_evm_data = EvmTransactionData {
1215            nonce: Some(42),
1216            gas_limit: Some(21000),
1217            ..Default::default()
1218        };
1219        let new_network_data = NetworkTransactionData::Evm(new_evm_data);
1220
1221        let updated = storage
1222            .update_network_data("test-tx".to_string(), new_network_data)
1223            .await?;
1224
1225        assert_eq!(updated.id, "test-tx");
1226        if let NetworkTransactionData::Evm(evm_data) = updated.network_data {
1227            assert_eq!(evm_data.nonce, Some(42));
1228            assert_eq!(evm_data.gas_limit, Some(21000));
1229        } else {
1230            panic!("Expected EVM network data");
1231        }
1232
1233        Ok(())
1234    }
1235
1236    #[tokio::test]
1237    async fn test_set_sent_at_in_memory() -> Result<()> {
1238        let storage = TransactionRepositoryStorage::new_in_memory();
1239        let transaction = create_test_transaction("test-tx", "test-relayer");
1240
1241        // Create transaction first
1242        storage.create(transaction).await?;
1243
1244        // Set sent_at
1245        let sent_at = Utc::now().to_string();
1246        let updated = storage
1247            .set_sent_at("test-tx".to_string(), sent_at.clone())
1248            .await?;
1249
1250        assert_eq!(updated.id, "test-tx");
1251        assert_eq!(updated.sent_at, Some(sent_at));
1252
1253        Ok(())
1254    }
1255
1256    #[tokio::test]
1257    async fn test_set_confirmed_at_in_memory() -> Result<()> {
1258        let storage = TransactionRepositoryStorage::new_in_memory();
1259        let transaction = create_test_transaction("test-tx", "test-relayer");
1260
1261        // Create transaction first
1262        storage.create(transaction).await?;
1263
1264        // Set confirmed_at
1265        let confirmed_at = Utc::now().to_string();
1266        let updated = storage
1267            .set_confirmed_at("test-tx".to_string(), confirmed_at.clone())
1268            .await?;
1269
1270        assert_eq!(updated.id, "test-tx");
1271        assert_eq!(updated.confirmed_at, Some(confirmed_at));
1272
1273        Ok(())
1274    }
1275
1276    #[tokio::test]
1277    async fn test_create_duplicate_id_in_memory() -> Result<()> {
1278        let storage = TransactionRepositoryStorage::new_in_memory();
1279        let transaction = create_test_transaction("duplicate-id", "test-relayer");
1280
1281        // Create first transaction
1282        storage.create(transaction.clone()).await?;
1283
1284        // Try to create another with same ID - should fail
1285        let result = storage.create(transaction.clone()).await;
1286        assert!(result.is_err());
1287
1288        Ok(())
1289    }
1290
1291    #[tokio::test]
1292    async fn test_workflow_in_memory() -> Result<()> {
1293        let storage = TransactionRepositoryStorage::new_in_memory();
1294
1295        // 1. Start with empty storage
1296        assert!(!storage.has_entries().await?);
1297        assert_eq!(storage.count().await?, 0);
1298
1299        // 2. Create transaction
1300        let transaction = create_test_transaction("workflow-test", "test-relayer");
1301        let created = storage.create(transaction.clone()).await?;
1302        assert_eq!(created.id, "workflow-test");
1303
1304        // 3. Verify it exists
1305        assert!(storage.has_entries().await?);
1306        assert_eq!(storage.count().await?, 1);
1307
1308        // 4. Retrieve it
1309        let retrieved = storage.get_by_id("workflow-test".to_string()).await?;
1310        assert_eq!(retrieved.id, "workflow-test");
1311
1312        // 5. Update status
1313        let updated = storage
1314            .update_status("workflow-test".to_string(), TransactionStatus::Sent)
1315            .await?;
1316        assert_eq!(updated.status, TransactionStatus::Sent);
1317
1318        // 6. Verify update
1319        let retrieved_updated = storage.get_by_id("workflow-test".to_string()).await?;
1320        assert_eq!(retrieved_updated.status, TransactionStatus::Sent);
1321
1322        // 7. Delete it
1323        storage.delete_by_id("workflow-test".to_string()).await?;
1324
1325        // 8. Verify it's gone
1326        assert!(!storage.has_entries().await?);
1327        assert_eq!(storage.count().await?, 0);
1328
1329        let result = storage.get_by_id("workflow-test".to_string()).await;
1330        assert!(result.is_err());
1331
1332        Ok(())
1333    }
1334
1335    #[tokio::test]
1336    async fn test_multiple_relayers_workflow() -> Result<()> {
1337        let storage = TransactionRepositoryStorage::new_in_memory();
1338
1339        // Add transactions for multiple relayers
1340        let tx1 =
1341            create_test_transaction_with_status("tx-1", "relayer-1", TransactionStatus::Pending);
1342        let tx2 = create_test_transaction_with_status("tx-2", "relayer-1", TransactionStatus::Sent);
1343        let tx3 =
1344            create_test_transaction_with_status("tx-3", "relayer-2", TransactionStatus::Pending);
1345
1346        storage.create(tx1).await?;
1347        storage.create(tx2).await?;
1348        storage.create(tx3).await?;
1349
1350        // Test find_by_relayer_id
1351        let query = PaginationQuery {
1352            page: 1,
1353            per_page: 10,
1354        };
1355        let relayer1_txs = storage.find_by_relayer_id("relayer-1", query).await?;
1356        assert_eq!(relayer1_txs.items.len(), 2);
1357
1358        // Test find_by_status
1359        let pending_txs = storage
1360            .find_by_status("relayer-1", &[TransactionStatus::Pending])
1361            .await?;
1362        assert_eq!(pending_txs.len(), 1);
1363        assert_eq!(pending_txs[0].id, "tx-1");
1364
1365        // Test count remains accurate
1366        assert_eq!(storage.count().await?, 3);
1367
1368        Ok(())
1369    }
1370
1371    #[tokio::test]
1372    async fn test_pagination_edge_cases_in_memory() -> Result<()> {
1373        let storage = TransactionRepositoryStorage::new_in_memory();
1374
1375        // Test pagination with empty storage
1376        let query = PaginationQuery {
1377            page: 1,
1378            per_page: 10,
1379        };
1380        let page = storage.list_paginated(query).await?;
1381        assert_eq!(page.items.len(), 0);
1382        assert_eq!(page.total, 0);
1383        assert_eq!(page.page, 1);
1384        assert_eq!(page.per_page, 10);
1385
1386        // Add one transaction
1387        let transaction = create_test_transaction("single-tx", "test-relayer");
1388        storage.create(transaction).await?;
1389
1390        // Test pagination with single item
1391        let query = PaginationQuery {
1392            page: 1,
1393            per_page: 10,
1394        };
1395        let page = storage.list_paginated(query).await?;
1396        assert_eq!(page.items.len(), 1);
1397        assert_eq!(page.total, 1);
1398        assert_eq!(page.page, 1);
1399        assert_eq!(page.per_page, 10);
1400
1401        // Test pagination with page beyond total
1402        let query = PaginationQuery {
1403            page: 3,
1404            per_page: 10,
1405        };
1406        let page = storage.list_paginated(query).await?;
1407        assert_eq!(page.items.len(), 0);
1408        assert_eq!(page.total, 1);
1409        assert_eq!(page.page, 3);
1410        assert_eq!(page.per_page, 10);
1411
1412        Ok(())
1413    }
1414
1415    #[tokio::test]
1416    async fn test_find_by_relayer_id_pagination() -> Result<()> {
1417        let storage = TransactionRepositoryStorage::new_in_memory();
1418
1419        // Add many transactions for one relayer
1420        for i in 1..=10 {
1421            let tx = create_test_transaction(&format!("tx-{i}"), "test-relayer");
1422            storage.create(tx).await?;
1423        }
1424
1425        // Test first page
1426        let query = PaginationQuery {
1427            page: 1,
1428            per_page: 3,
1429        };
1430        let page1 = storage.find_by_relayer_id("test-relayer", query).await?;
1431        assert_eq!(page1.items.len(), 3);
1432        assert_eq!(page1.total, 10);
1433        assert_eq!(page1.page, 1);
1434        assert_eq!(page1.per_page, 3);
1435
1436        // Test second page
1437        let query = PaginationQuery {
1438            page: 2,
1439            per_page: 3,
1440        };
1441        let page2 = storage.find_by_relayer_id("test-relayer", query).await?;
1442        assert_eq!(page2.items.len(), 3);
1443        assert_eq!(page2.total, 10);
1444        assert_eq!(page2.page, 2);
1445        assert_eq!(page2.per_page, 3);
1446
1447        Ok(())
1448    }
1449
1450    #[tokio::test]
1451    async fn test_find_by_multiple_statuses() -> Result<()> {
1452        let storage = TransactionRepositoryStorage::new_in_memory();
1453
1454        // Add transactions with different statuses
1455        let tx1 =
1456            create_test_transaction_with_status("tx-1", "test-relayer", TransactionStatus::Pending);
1457        let tx2 =
1458            create_test_transaction_with_status("tx-2", "test-relayer", TransactionStatus::Sent);
1459        let tx3 = create_test_transaction_with_status(
1460            "tx-3",
1461            "test-relayer",
1462            TransactionStatus::Confirmed,
1463        );
1464        let tx4 =
1465            create_test_transaction_with_status("tx-4", "test-relayer", TransactionStatus::Failed);
1466
1467        storage.create(tx1).await?;
1468        storage.create(tx2).await?;
1469        storage.create(tx3).await?;
1470        storage.create(tx4).await?;
1471
1472        // Find by multiple statuses
1473        let statuses = vec![TransactionStatus::Pending, TransactionStatus::Sent];
1474        let result = storage.find_by_status("test-relayer", &statuses).await?;
1475
1476        assert_eq!(result.len(), 2);
1477
1478        // Verify all transactions have the correct statuses
1479        let found_statuses: Vec<TransactionStatus> =
1480            result.iter().map(|tx| tx.status.clone()).collect();
1481        assert!(found_statuses.contains(&TransactionStatus::Pending));
1482        assert!(found_statuses.contains(&TransactionStatus::Sent));
1483
1484        Ok(())
1485    }
1486
1487    #[tokio::test]
1488    async fn test_record_stellar_try_again_later_retry_in_memory() -> Result<()> {
1489        let storage = TransactionRepositoryStorage::new_in_memory();
1490        let mut transaction = create_test_transaction("test-tx", "test-relayer");
1491        transaction.status = TransactionStatus::Sent;
1492        storage.create(transaction).await?;
1493
1494        let sent_at = "2025-03-18T10:00:00Z".to_string();
1495        let updated = storage
1496            .record_stellar_try_again_later_retry("test-tx".to_string(), sent_at.clone())
1497            .await?;
1498
1499        assert_eq!(updated.id, "test-tx");
1500        assert_eq!(updated.sent_at, Some(sent_at));
1501        let meta = updated.metadata.expect("metadata should be set");
1502        assert_eq!(meta.try_again_later_retries, 1);
1503        assert_eq!(meta.consecutive_failures, 0);
1504        assert_eq!(meta.total_failures, 0);
1505        assert_eq!(meta.insufficient_fee_retries, 0);
1506
1507        Ok(())
1508    }
1509
1510    #[tokio::test]
1511    async fn test_record_stellar_try_again_later_retry_accumulates_in_memory() -> Result<()> {
1512        let storage = TransactionRepositoryStorage::new_in_memory();
1513        let mut transaction = create_test_transaction("test-tx", "test-relayer");
1514        transaction.status = TransactionStatus::Sent;
1515        storage.create(transaction).await?;
1516
1517        storage
1518            .record_stellar_try_again_later_retry(
1519                "test-tx".to_string(),
1520                "2025-03-18T10:00:00Z".to_string(),
1521            )
1522            .await?;
1523
1524        let updated = storage
1525            .record_stellar_try_again_later_retry(
1526                "test-tx".to_string(),
1527                "2025-03-18T10:01:00Z".to_string(),
1528            )
1529            .await?;
1530
1531        assert_eq!(updated.sent_at.as_deref(), Some("2025-03-18T10:01:00Z"));
1532        let meta = updated.metadata.unwrap();
1533        assert_eq!(meta.try_again_later_retries, 2);
1534
1535        Ok(())
1536    }
1537
1538    #[tokio::test]
1539    async fn test_record_stellar_try_again_later_retry_noop_on_final_state_in_memory() -> Result<()>
1540    {
1541        let storage = TransactionRepositoryStorage::new_in_memory();
1542        let mut transaction = create_test_transaction("test-tx", "test-relayer");
1543        transaction.status = TransactionStatus::Confirmed;
1544        transaction.sent_at = Some("old-time".to_string());
1545        storage.create(transaction).await?;
1546
1547        let result = storage
1548            .record_stellar_try_again_later_retry("test-tx".to_string(), "new-time".to_string())
1549            .await?;
1550
1551        assert_eq!(result.sent_at.as_deref(), Some("old-time"));
1552        assert!(result.metadata.is_none());
1553
1554        Ok(())
1555    }
1556
1557    #[tokio::test]
1558    async fn test_record_stellar_try_again_later_retry_not_found_in_memory() -> Result<()> {
1559        let storage = TransactionRepositoryStorage::new_in_memory();
1560
1561        let result = storage
1562            .record_stellar_try_again_later_retry(
1563                "nonexistent".to_string(),
1564                "2025-03-18T10:00:00Z".to_string(),
1565            )
1566            .await;
1567
1568        assert!(matches!(result, Err(RepositoryError::NotFound(_))));
1569
1570        Ok(())
1571    }
1572}
openzeppelin_relayer/repositories/transaction/mod.rs

openzeppelin_relayer/repositories/transaction/
mod.rs
