openzeppelin_relayer/repositories/relayer/
relayer_redis.rs

1//! Redis-backed implementation of the RelayerRepository.
2
3use crate::models::UpdateRelayerRequest;
4use crate::models::{
5    DisabledReason, PaginationQuery, RelayerNetworkPolicy, RelayerRepoModel, RepositoryError,
6};
7use crate::repositories::redis_base::RedisRepository;
8use crate::repositories::{BatchRetrievalResult, PaginatedResult, RelayerRepository, Repository};
9use async_trait::async_trait;
10use redis::aio::ConnectionManager;
11use redis::AsyncCommands;
12use std::fmt;
13use std::sync::Arc;
14use tracing::{debug, error, warn};
15
16const RELAYER_PREFIX: &str = "relayer";
17const RELAYER_LIST_KEY: &str = "relayer_list";
18
19#[derive(Clone)]
20pub struct RedisRelayerRepository {
21    pub client: Arc<ConnectionManager>,
22    pub key_prefix: String,
23}
24
25impl RedisRepository for RedisRelayerRepository {}
26
27impl RedisRelayerRepository {
28    pub fn new(
29        connection_manager: Arc<ConnectionManager>,
30        key_prefix: String,
31    ) -> Result<Self, RepositoryError> {
32        if key_prefix.is_empty() {
33            return Err(RepositoryError::InvalidData(
34                "Redis key prefix cannot be empty".to_string(),
35            ));
36        }
37
38        Ok(Self {
39            client: connection_manager,
40            key_prefix,
41        })
42    }
43
44    /// Generate key for relayer data: relayer:{relayer_id}
45    fn relayer_key(&self, relayer_id: &str) -> String {
46        format!("{}:{}:{}", self.key_prefix, RELAYER_PREFIX, relayer_id)
47    }
48
49    /// Generate key for relayer list: relayer_list (set of all relayer IDs)
50    fn relayer_list_key(&self) -> String {
51        format!("{}:{}", self.key_prefix, RELAYER_LIST_KEY)
52    }
53
54    /// Batch fetch relayers by IDs
55    async fn get_relayers_by_ids(
56        &self,
57        ids: &[String],
58    ) -> Result<BatchRetrievalResult<RelayerRepoModel>, RepositoryError> {
59        if ids.is_empty() {
60            debug!("no relayer IDs provided for batch fetch");
61            return Ok(BatchRetrievalResult {
62                results: vec![],
63                failed_ids: vec![],
64            });
65        }
66
67        let mut conn = self.client.as_ref().clone();
68        let keys: Vec<String> = ids.iter().map(|id| self.relayer_key(id)).collect();
69
70        debug!(count = %keys.len(), "batch fetching relayer data");
71
72        let values: Vec<Option<String>> = conn
73            .mget(&keys)
74            .await
75            .map_err(|e| self.map_redis_error(e, "batch_fetch_relayers"))?;
76
77        let mut relayers = Vec::new();
78        let mut failed_count = 0;
79        let mut failed_ids = Vec::new();
80        for (i, value) in values.into_iter().enumerate() {
81            match value {
82                Some(json) => {
83                    match self.deserialize_entity(&json, &ids[i], "relayer") {
84                        Ok(relayer) => relayers.push(relayer),
85                        Err(e) => {
86                            failed_count += 1;
87                            error!(relayer_id = %ids[i], error = %e, "failed to deserialize relayer");
88                            failed_ids.push(ids[i].clone());
89                            // Continue processing other relayers
90                        }
91                    }
92                }
93                None => {
94                    warn!(relayer_id = %ids[i], "relayer not found in batch fetch");
95                }
96            }
97        }
98
99        if failed_count > 0 {
100            warn!(failed_count = %failed_count, total_count = %ids.len(), "failed to deserialize relayers in batch");
101        }
102
103        debug!(count = %relayers.len(), "successfully fetched relayers");
104        Ok(BatchRetrievalResult {
105            results: relayers,
106            failed_ids,
107        })
108    }
109}
110
111impl fmt::Debug for RedisRelayerRepository {
112    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
113        f.debug_struct("RedisRelayerRepository")
114            .field("client", &"<ConnectionManager>")
115            .field("key_prefix", &self.key_prefix)
116            .finish()
117    }
118}
119
120#[async_trait]
121impl Repository<RelayerRepoModel, String> for RedisRelayerRepository {
122    async fn create(&self, entity: RelayerRepoModel) -> Result<RelayerRepoModel, RepositoryError> {
123        if entity.id.is_empty() {
124            return Err(RepositoryError::InvalidData(
125                "Relayer ID cannot be empty".to_string(),
126            ));
127        }
128
129        if entity.name.is_empty() {
130            return Err(RepositoryError::InvalidData(
131                "Relayer name cannot be empty".to_string(),
132            ));
133        }
134
135        let mut conn = self.client.as_ref().clone();
136        let relayer_key = self.relayer_key(&entity.id);
137
138        // Check if relayer already exists
139        let exists: bool = conn
140            .exists(&relayer_key)
141            .await
142            .map_err(|e| self.map_redis_error(e, "create_relayer_exists_check"))?;
143
144        if exists {
145            return Err(RepositoryError::ConstraintViolation(format!(
146                "Relayer with ID {} already exists",
147                entity.id
148            )));
149        }
150
151        let serialized = self.serialize_entity(&entity, |r| &r.id, "relayer")?;
152
153        // Use pipeline for atomic operations
154        let mut pipe = redis::pipe();
155        pipe.atomic();
156        pipe.set(&relayer_key, &serialized);
157        pipe.sadd(self.relayer_list_key(), &entity.id);
158
159        pipe.exec_async(&mut conn)
160            .await
161            .map_err(|e| self.map_redis_error(e, "create_relayer_pipeline"))?;
162
163        debug!(relayer_id = %entity.id, "created relayer");
164        Ok(entity)
165    }
166
167    async fn get_by_id(&self, id: String) -> Result<RelayerRepoModel, RepositoryError> {
168        if id.is_empty() {
169            return Err(RepositoryError::InvalidData(
170                "Relayer ID cannot be empty".to_string(),
171            ));
172        }
173
174        let mut conn = self.client.as_ref().clone();
175        let relayer_key = self.relayer_key(&id);
176
177        debug!(relayer_id = %id, "fetching relayer");
178
179        let json: Option<String> = conn
180            .get(&relayer_key)
181            .await
182            .map_err(|e| self.map_redis_error(e, "get_relayer_by_id"))?;
183
184        match json {
185            Some(json) => {
186                debug!(relayer_id = %id, "found relayer");
187                self.deserialize_entity(&json, &id, "relayer")
188            }
189            None => {
190                debug!(relayer_id = %id, "relayer not found");
191                Err(RepositoryError::NotFound(format!(
192                    "Relayer with ID {} not found",
193                    id
194                )))
195            }
196        }
197    }
198
199    async fn list_all(&self) -> Result<Vec<RelayerRepoModel>, RepositoryError> {
200        let mut conn = self.client.as_ref().clone();
201        let relayer_list_key = self.relayer_list_key();
202
203        debug!("listing all relayers");
204
205        let relayer_ids: Vec<String> = conn
206            .smembers(&relayer_list_key)
207            .await
208            .map_err(|e| self.map_redis_error(e, "list_all_relayers"))?;
209
210        debug!(count = %relayer_ids.len(), "found relayers in index");
211
212        let relayers = self.get_relayers_by_ids(&relayer_ids).await?;
213        Ok(relayers.results)
214    }
215
216    async fn list_paginated(
217        &self,
218        query: PaginationQuery,
219    ) -> Result<PaginatedResult<RelayerRepoModel>, RepositoryError> {
220        if query.page == 0 {
221            return Err(RepositoryError::InvalidData(
222                "Page number must be greater than 0".to_string(),
223            ));
224        }
225
226        if query.per_page == 0 {
227            return Err(RepositoryError::InvalidData(
228                "Per page count must be greater than 0".to_string(),
229            ));
230        }
231
232        let mut conn = self.client.as_ref().clone();
233        let relayer_list_key = self.relayer_list_key();
234
235        // Get total count
236        let total: u64 = conn
237            .scard(&relayer_list_key)
238            .await
239            .map_err(|e| self.map_redis_error(e, "list_paginated_count"))?;
240
241        if total == 0 {
242            return Ok(PaginatedResult {
243                items: vec![],
244                total: 0,
245                page: query.page,
246                per_page: query.per_page,
247            });
248        }
249
250        // Get all IDs and paginate in memory
251        let all_ids: Vec<String> = conn
252            .smembers(&relayer_list_key)
253            .await
254            .map_err(|e| self.map_redis_error(e, "list_paginated_members"))?;
255
256        let start = ((query.page - 1) * query.per_page) as usize;
257        let end = (start + query.per_page as usize).min(all_ids.len());
258
259        let page_ids = &all_ids[start..end];
260        let items = self.get_relayers_by_ids(page_ids).await?;
261
262        Ok(PaginatedResult {
263            items: items.results.clone(),
264            total,
265            page: query.page,
266            per_page: query.per_page,
267        })
268    }
269
270    async fn update(
271        &self,
272        id: String,
273        entity: RelayerRepoModel,
274    ) -> Result<RelayerRepoModel, RepositoryError> {
275        if id.is_empty() {
276            return Err(RepositoryError::InvalidData(
277                "Relayer ID cannot be empty".to_string(),
278            ));
279        }
280
281        if entity.name.is_empty() {
282            return Err(RepositoryError::InvalidData(
283                "Relayer name cannot be empty".to_string(),
284            ));
285        }
286
287        let mut conn = self.client.as_ref().clone();
288        let relayer_key = self.relayer_key(&id);
289
290        // Check if relayer exists
291        let exists: bool = conn
292            .exists(&relayer_key)
293            .await
294            .map_err(|e| self.map_redis_error(e, "update_relayer_exists_check"))?;
295
296        if !exists {
297            return Err(RepositoryError::NotFound(format!(
298                "Relayer with ID {} not found",
299                id
300            )));
301        }
302
303        // Ensure we preserve the original ID
304        let mut updated_entity = entity;
305        updated_entity.id = id.clone();
306
307        let serialized = self.serialize_entity(&updated_entity, |r| &r.id, "relayer")?;
308
309        // Use pipeline for atomic operations
310        let mut pipe = redis::pipe();
311        pipe.atomic();
312        pipe.set(&relayer_key, &serialized);
313        pipe.sadd(self.relayer_list_key(), &id);
314
315        pipe.exec_async(&mut conn)
316            .await
317            .map_err(|e| self.map_redis_error(e, "update_relayer_pipeline"))?;
318
319        debug!(relayer_id = %id, "updated relayer");
320        Ok(updated_entity)
321    }
322
323    async fn delete_by_id(&self, id: String) -> Result<(), RepositoryError> {
324        if id.is_empty() {
325            return Err(RepositoryError::InvalidData(
326                "Relayer ID cannot be empty".to_string(),
327            ));
328        }
329
330        let mut conn = self.client.as_ref().clone();
331        let relayer_key = self.relayer_key(&id);
332
333        // Check if relayer exists
334        let exists: bool = conn
335            .exists(&relayer_key)
336            .await
337            .map_err(|e| self.map_redis_error(e, "delete_relayer_exists_check"))?;
338
339        if !exists {
340            return Err(RepositoryError::NotFound(format!(
341                "Relayer with ID {} not found",
342                id
343            )));
344        }
345
346        // Use pipeline for atomic operations
347        let mut pipe = redis::pipe();
348        pipe.atomic();
349        pipe.del(&relayer_key);
350        pipe.srem(self.relayer_list_key(), &id);
351
352        pipe.exec_async(&mut conn)
353            .await
354            .map_err(|e| self.map_redis_error(e, "delete_relayer_pipeline"))?;
355
356        debug!(relayer_id = %id, "deleted relayer");
357        Ok(())
358    }
359
360    async fn count(&self) -> Result<usize, RepositoryError> {
361        let mut conn = self.client.as_ref().clone();
362        let relayer_list_key = self.relayer_list_key();
363
364        let count: u64 = conn
365            .scard(&relayer_list_key)
366            .await
367            .map_err(|e| self.map_redis_error(e, "count_relayers"))?;
368
369        Ok(count as usize)
370    }
371
372    async fn has_entries(&self) -> Result<bool, RepositoryError> {
373        let mut conn = self.client.as_ref().clone();
374        let relayer_list_key = self.relayer_list_key();
375
376        debug!("checking if relayer entries exist");
377
378        let exists: bool = conn
379            .exists(&relayer_list_key)
380            .await
381            .map_err(|e| self.map_redis_error(e, "has_entries_check"))?;
382
383        debug!(exists = %exists, "relayer entries exist");
384        Ok(exists)
385    }
386
387    async fn drop_all_entries(&self) -> Result<(), RepositoryError> {
388        let mut conn = self.client.as_ref().clone();
389        let relayer_list_key = self.relayer_list_key();
390
391        debug!("dropping all relayer entries");
392
393        // Get all relayer IDs first
394        let relayer_ids: Vec<String> = conn
395            .smembers(&relayer_list_key)
396            .await
397            .map_err(|e| self.map_redis_error(e, "drop_all_entries_get_ids"))?;
398
399        if relayer_ids.is_empty() {
400            debug!("no relayer entries to drop");
401            return Ok(());
402        }
403
404        // Use pipeline for atomic operations
405        let mut pipe = redis::pipe();
406        pipe.atomic();
407
408        // Delete all individual relayer entries
409        for relayer_id in &relayer_ids {
410            let relayer_key = self.relayer_key(relayer_id);
411            pipe.del(&relayer_key);
412        }
413
414        // Delete the relayer list key
415        pipe.del(&relayer_list_key);
416
417        pipe.exec_async(&mut conn)
418            .await
419            .map_err(|e| self.map_redis_error(e, "drop_all_entries_pipeline"))?;
420
421        debug!(count = %relayer_ids.len(), "dropped relayer entries");
422        Ok(())
423    }
424}
425
426#[async_trait]
427impl RelayerRepository for RedisRelayerRepository {
428    async fn list_active(&self) -> Result<Vec<RelayerRepoModel>, RepositoryError> {
429        let all_relayers = self.list_all().await?;
430        let active_relayers: Vec<RelayerRepoModel> = all_relayers
431            .into_iter()
432            .filter(|relayer| !relayer.paused)
433            .collect();
434
435        debug!(count = %active_relayers.len(), "found active relayers");
436        Ok(active_relayers)
437    }
438
439    async fn list_by_signer_id(
440        &self,
441        signer_id: &str,
442    ) -> Result<Vec<RelayerRepoModel>, RepositoryError> {
443        let all_relayers = self.list_all().await?;
444        let relayers_with_signer: Vec<RelayerRepoModel> = all_relayers
445            .into_iter()
446            .filter(|relayer| relayer.signer_id == signer_id)
447            .collect();
448
449        debug!(count = %relayers_with_signer.len(), signer_id = %signer_id, "found relayers using signer");
450        Ok(relayers_with_signer)
451    }
452
453    async fn list_by_notification_id(
454        &self,
455        notification_id: &str,
456    ) -> Result<Vec<RelayerRepoModel>, RepositoryError> {
457        let all_relayers = self.list_all().await?;
458        let relayers_with_notification: Vec<RelayerRepoModel> = all_relayers
459            .into_iter()
460            .filter(|relayer| {
461                relayer
462                    .notification_id
463                    .as_ref()
464                    .is_some_and(|id| id == notification_id)
465            })
466            .collect();
467
468        debug!(count = %relayers_with_notification.len(), notification_id = %notification_id, "found relayers using notification");
469        Ok(relayers_with_notification)
470    }
471
472    async fn partial_update(
473        &self,
474        id: String,
475        update: UpdateRelayerRequest,
476    ) -> Result<RelayerRepoModel, RepositoryError> {
477        // First get the current relayer
478        let mut relayer = self.get_by_id(id.clone()).await?;
479
480        // Apply the partial update
481        if let Some(paused) = update.paused {
482            relayer.paused = paused;
483        }
484
485        // Update the relayer
486        self.update(id, relayer).await
487    }
488
489    async fn enable_relayer(
490        &self,
491        relayer_id: String,
492    ) -> Result<RelayerRepoModel, RepositoryError> {
493        // First get the current relayer
494        let mut relayer = self.get_by_id(relayer_id.clone()).await?;
495
496        // Update the system_disabled flag and clear reason
497        relayer.system_disabled = false;
498        relayer.disabled_reason = None;
499
500        // Update the relayer
501        self.update(relayer_id, relayer).await
502    }
503
504    async fn disable_relayer(
505        &self,
506        relayer_id: String,
507        reason: DisabledReason,
508    ) -> Result<RelayerRepoModel, RepositoryError> {
509        // First get the current relayer
510        let mut relayer = self.get_by_id(relayer_id.clone()).await?;
511
512        // Update the system_disabled flag and set reason
513        relayer.system_disabled = true;
514        relayer.disabled_reason = Some(reason);
515
516        // Update the relayer
517        self.update(relayer_id, relayer).await
518    }
519
520    async fn update_policy(
521        &self,
522        id: String,
523        policy: RelayerNetworkPolicy,
524    ) -> Result<RelayerRepoModel, RepositoryError> {
525        // First get the current relayer
526        let mut relayer = self.get_by_id(id.clone()).await?;
527
528        // Update the policy
529        relayer.policies = policy;
530
531        // Update the relayer
532        self.update(id, relayer).await
533    }
534}
535
536#[cfg(test)]
537mod tests {
538    use super::*;
539    use crate::models::{NetworkType, RelayerEvmPolicy, RelayerNetworkPolicy};
540    use redis::aio::ConnectionManager;
541    use std::sync::Arc;
542
543    fn create_test_relayer(id: &str) -> RelayerRepoModel {
544        RelayerRepoModel {
545            id: id.to_string(),
546            name: format!("Test Relayer {}", id),
547            network: "ethereum".to_string(),
548            paused: false,
549            network_type: NetworkType::Evm,
550            signer_id: "test-signer".to_string(),
551            policies: RelayerNetworkPolicy::Evm(RelayerEvmPolicy::default()),
552            address: "0x742d35Cc6634C0532925a3b844Bc454e4438f44e".to_string(),
553            notification_id: None,
554            system_disabled: false,
555            disabled_reason: None,
556            custom_rpc_urls: None,
557        }
558    }
559
560    fn create_test_relayer_with_pause(id: &str, paused: bool) -> RelayerRepoModel {
561        let mut relayer = create_test_relayer(id);
562        relayer.paused = paused;
563        relayer
564    }
565
566    async fn setup_test_repo() -> RedisRelayerRepository {
567        let redis_url =
568            std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379/".to_string());
569        let client = redis::Client::open(redis_url).expect("Failed to create Redis client");
570        let connection_manager = ConnectionManager::new(client)
571            .await
572            .expect("Failed to create Redis connection manager");
573
574        RedisRelayerRepository::new(Arc::new(connection_manager), "test".to_string())
575            .expect("Failed to create Redis relayer repository")
576    }
577
578    #[ignore = "Requires active Redis instance"]
579    #[tokio::test]
580    async fn test_new_repository_creation() {
581        let repo = setup_test_repo().await;
582        assert_eq!(repo.key_prefix, "test");
583    }
584
585    #[ignore = "Requires active Redis instance"]
586    #[tokio::test]
587    async fn test_new_repository_empty_prefix_fails() {
588        let redis_url =
589            std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379/".to_string());
590        let client = redis::Client::open(redis_url).expect("Failed to create Redis client");
591        let connection_manager = ConnectionManager::new(client)
592            .await
593            .expect("Failed to create Redis connection manager");
594
595        let result = RedisRelayerRepository::new(Arc::new(connection_manager), "".to_string());
596        assert!(matches!(result, Err(RepositoryError::InvalidData(_))));
597    }
598
599    #[ignore = "Requires active Redis instance"]
600    #[tokio::test]
601    async fn test_key_generation() {
602        let repo = setup_test_repo().await;
603
604        let relayer_key = repo.relayer_key("test-relayer");
605        assert_eq!(relayer_key, "test:relayer:test-relayer");
606
607        let list_key = repo.relayer_list_key();
608        assert_eq!(list_key, "test:relayer_list");
609    }
610
611    #[ignore = "Requires active Redis instance"]
612    #[tokio::test]
613    async fn test_serialize_deserialize_relayer() {
614        let repo = setup_test_repo().await;
615        let relayer = create_test_relayer("test-relayer");
616
617        let serialized = repo
618            .serialize_entity(&relayer, |r| &r.id, "relayer")
619            .unwrap();
620        let deserialized: RelayerRepoModel = repo
621            .deserialize_entity(&serialized, &relayer.id, "relayer")
622            .unwrap();
623
624        assert_eq!(relayer.id, deserialized.id);
625        assert_eq!(relayer.name, deserialized.name);
626        assert_eq!(relayer.network, deserialized.network);
627        assert_eq!(relayer.paused, deserialized.paused);
628        assert_eq!(relayer.network_type, deserialized.network_type);
629        assert_eq!(relayer.signer_id, deserialized.signer_id);
630        assert_eq!(relayer.address, deserialized.address);
631        assert_eq!(relayer.notification_id, deserialized.notification_id);
632        assert_eq!(relayer.system_disabled, deserialized.system_disabled);
633        assert_eq!(relayer.custom_rpc_urls, deserialized.custom_rpc_urls);
634    }
635
636    #[ignore = "Requires active Redis instance"]
637    #[tokio::test]
638    async fn test_create_relayer() {
639        let repo = setup_test_repo().await;
640        let relayer_id = uuid::Uuid::new_v4().to_string();
641        let relayer = create_test_relayer(&relayer_id);
642
643        let result = repo.create(relayer.clone()).await;
644        assert!(result.is_ok());
645
646        let created_relayer = result.unwrap();
647        assert_eq!(created_relayer.id, relayer_id);
648        assert_eq!(created_relayer.name, relayer.name);
649    }
650
651    #[ignore = "Requires active Redis instance"]
652    #[tokio::test]
653    async fn test_get_relayer() {
654        let repo = setup_test_repo().await;
655        let relayer_id = uuid::Uuid::new_v4().to_string();
656        let relayer = create_test_relayer(&relayer_id);
657
658        repo.create(relayer.clone()).await.unwrap();
659
660        let retrieved = repo.get_by_id(relayer_id).await.unwrap();
661        assert_eq!(retrieved.id, relayer.id);
662        assert_eq!(retrieved.name, relayer.name);
663    }
664
665    #[ignore = "Requires active Redis instance"]
666    #[tokio::test]
667    async fn test_list_all_relayers() {
668        let repo = setup_test_repo().await;
669        let relayer1_id = uuid::Uuid::new_v4().to_string();
670        let relayer2_id = uuid::Uuid::new_v4().to_string();
671        let relayer1 = create_test_relayer(&relayer1_id);
672        let relayer2 = create_test_relayer(&relayer2_id);
673
674        repo.create(relayer1).await.unwrap();
675        repo.create(relayer2).await.unwrap();
676
677        let all_relayers = repo.list_all().await.unwrap();
678        assert!(all_relayers.len() >= 2);
679    }
680
681    #[ignore = "Requires active Redis instance"]
682    #[tokio::test]
683    async fn test_list_active_relayers() {
684        let repo = setup_test_repo().await;
685        let relayer1_id = uuid::Uuid::new_v4().to_string();
686        let relayer2_id = uuid::Uuid::new_v4().to_string();
687        let relayer1 = create_test_relayer_with_pause(&relayer1_id, false);
688        let relayer2 = create_test_relayer_with_pause(&relayer2_id, true);
689
690        repo.create(relayer1).await.unwrap();
691        repo.create(relayer2).await.unwrap();
692
693        let active_relayers = repo.list_active().await.unwrap();
694        // Should have at least 1 active relayer
695        assert!(!active_relayers.is_empty());
696        // All returned relayers should be active
697        assert!(active_relayers.iter().all(|r| !r.paused));
698    }
699
700    #[ignore = "Requires active Redis instance"]
701    #[tokio::test]
702    async fn test_count_relayers() {
703        let repo = setup_test_repo().await;
704        let relayer_id = uuid::Uuid::new_v4().to_string();
705        let relayer = create_test_relayer(&relayer_id);
706
707        repo.create(relayer).await.unwrap();
708
709        let count = repo.count().await.unwrap();
710        assert!(count >= 1);
711    }
712
713    #[ignore = "Requires active Redis instance"]
714    #[tokio::test]
715    async fn test_get_nonexistent_relayer() {
716        let repo = setup_test_repo().await;
717
718        let result = repo.get_by_id("nonexistent-relayer".to_string()).await;
719        assert!(matches!(result, Err(RepositoryError::NotFound(_))));
720    }
721
722    #[ignore = "Requires active Redis instance"]
723    #[tokio::test]
724    async fn test_duplicate_relayer_creation() {
725        let repo = setup_test_repo().await;
726        let relayer_id = uuid::Uuid::new_v4().to_string();
727        let relayer = create_test_relayer(&relayer_id);
728
729        repo.create(relayer.clone()).await.unwrap();
730
731        let duplicate_result = repo.create(relayer).await;
732        assert!(matches!(
733            duplicate_result,
734            Err(RepositoryError::ConstraintViolation(_))
735        ));
736    }
737
738    #[ignore = "Requires active Redis instance"]
739    #[tokio::test]
740    async fn test_update_relayer() {
741        let repo = setup_test_repo().await;
742        let relayer_id = uuid::Uuid::new_v4().to_string();
743        let relayer = create_test_relayer(&relayer_id);
744
745        repo.create(relayer.clone()).await.unwrap();
746
747        let mut updated_relayer = relayer.clone();
748        updated_relayer.name = "Updated Relayer Name".to_string();
749
750        let result = repo.update(relayer.id.clone(), updated_relayer).await;
751        assert!(result.is_ok());
752
753        let updated = result.unwrap();
754        assert_eq!(updated.name, "Updated Relayer Name");
755        assert_eq!(updated.id, relayer.id);
756    }
757
758    #[ignore = "Requires active Redis instance"]
759    #[tokio::test]
760    async fn test_delete_relayer() {
761        let repo = setup_test_repo().await;
762        let relayer_id = uuid::Uuid::new_v4().to_string();
763        let relayer = create_test_relayer(&relayer_id);
764
765        repo.create(relayer.clone()).await.unwrap();
766
767        let delete_result = repo.delete_by_id(relayer.id.clone()).await;
768        assert!(delete_result.is_ok());
769
770        let get_result = repo.get_by_id(relayer.id).await;
771        assert!(matches!(get_result, Err(RepositoryError::NotFound(_))));
772    }
773
774    #[ignore = "Requires active Redis instance"]
775    #[tokio::test]
776    async fn test_list_paginated() {
777        let repo = setup_test_repo().await;
778        let relayer1_id = uuid::Uuid::new_v4().to_string();
779        let relayer2_id = uuid::Uuid::new_v4().to_string();
780        let relayer1 = create_test_relayer(&relayer1_id);
781        let relayer2 = create_test_relayer(&relayer2_id);
782
783        repo.create(relayer1).await.unwrap();
784        repo.create(relayer2).await.unwrap();
785
786        let query = PaginationQuery {
787            page: 1,
788            per_page: 10,
789        };
790
791        let result = repo.list_paginated(query).await.unwrap();
792        assert!(result.total >= 2);
793        assert_eq!(result.page, 1);
794        assert_eq!(result.per_page, 10);
795    }
796
797    #[ignore = "Requires active Redis instance"]
798    #[tokio::test]
799    async fn test_partial_update_relayer() {
800        let repo = setup_test_repo().await;
801        let relayer_id = uuid::Uuid::new_v4().to_string();
802        let relayer = create_test_relayer(&relayer_id);
803
804        repo.create(relayer.clone()).await.unwrap();
805
806        let update = UpdateRelayerRequest {
807            paused: Some(true),
808            ..Default::default()
809        };
810        let result = repo.partial_update(relayer.id.clone(), update).await;
811        assert!(result.is_ok());
812
813        let updated = result.unwrap();
814        assert_eq!(updated.id, relayer.id);
815        assert!(updated.paused);
816    }
817
818    #[ignore = "Requires active Redis instance"]
819    #[tokio::test]
820    async fn test_enable_disable_relayer() {
821        let repo = setup_test_repo().await;
822        let relayer_id = uuid::Uuid::new_v4().to_string();
823        let relayer = create_test_relayer(&relayer_id);
824
825        repo.create(relayer.clone()).await.unwrap();
826
827        // Test disable
828        let disabled = repo
829            .disable_relayer(
830                relayer.id.clone(),
831                DisabledReason::BalanceCheckFailed("test reason".to_string()),
832            )
833            .await
834            .unwrap();
835        assert!(disabled.system_disabled);
836
837        // Test enable
838        let enabled = repo.enable_relayer(relayer.id.clone()).await.unwrap();
839        assert!(!enabled.system_disabled);
840    }
841
842    #[ignore = "Requires active Redis instance"]
843    #[tokio::test]
844    async fn test_update_policy() {
845        let repo = setup_test_repo().await;
846        let relayer_id = uuid::Uuid::new_v4().to_string();
847        let relayer = create_test_relayer(&relayer_id);
848
849        repo.create(relayer.clone()).await.unwrap();
850
851        let new_policy = RelayerNetworkPolicy::Evm(RelayerEvmPolicy {
852            gas_price_cap: Some(50_000_000_000),
853            whitelist_receivers: Some(vec!["0x123".to_string()]),
854            eip1559_pricing: Some(true),
855            private_transactions: Some(true),
856            min_balance: Some(1000000000000000000),
857            gas_limit_estimation: Some(true),
858        });
859
860        let result = repo.update_policy(relayer.id.clone(), new_policy).await;
861        assert!(result.is_ok());
862
863        let updated = result.unwrap();
864        if let RelayerNetworkPolicy::Evm(evm_policy) = updated.policies {
865            assert_eq!(evm_policy.gas_price_cap, Some(50_000_000_000));
866            assert_eq!(
867                evm_policy.whitelist_receivers,
868                Some(vec!["0x123".to_string()])
869            );
870            assert_eq!(evm_policy.eip1559_pricing, Some(true));
871            assert!(evm_policy.private_transactions.unwrap_or(false));
872            assert_eq!(evm_policy.min_balance, Some(1000000000000000000));
873        } else {
874            panic!("Expected EVM policy");
875        }
876    }
877
878    #[ignore = "Requires active Redis instance"]
879    #[tokio::test]
880    async fn test_debug_implementation() {
881        let repo = setup_test_repo().await;
882        let debug_str = format!("{:?}", repo);
883        assert!(debug_str.contains("RedisRelayerRepository"));
884        assert!(debug_str.contains("key_prefix"));
885    }
886
887    #[ignore = "Requires active Redis instance"]
888    #[tokio::test]
889    async fn test_error_handling_empty_id() {
890        let repo = setup_test_repo().await;
891
892        let create_result = repo
893            .create(RelayerRepoModel {
894                id: "".to_string(),
895                ..create_test_relayer("test")
896            })
897            .await;
898        assert!(matches!(
899            create_result,
900            Err(RepositoryError::InvalidData(_))
901        ));
902
903        let get_result = repo.get_by_id("".to_string()).await;
904        assert!(matches!(get_result, Err(RepositoryError::InvalidData(_))));
905
906        let update_result = repo
907            .update("".to_string(), create_test_relayer("test"))
908            .await;
909        assert!(matches!(
910            update_result,
911            Err(RepositoryError::InvalidData(_))
912        ));
913
914        let delete_result = repo.delete_by_id("".to_string()).await;
915        assert!(matches!(
916            delete_result,
917            Err(RepositoryError::InvalidData(_))
918        ));
919    }
920
921    #[ignore = "Requires active Redis instance"]
922    #[tokio::test]
923    async fn test_error_handling_empty_name() {
924        let repo = setup_test_repo().await;
925
926        let create_result = repo
927            .create(RelayerRepoModel {
928                name: "".to_string(),
929                ..create_test_relayer("test")
930            })
931            .await;
932        assert!(matches!(
933            create_result,
934            Err(RepositoryError::InvalidData(_))
935        ));
936    }
937
938    #[ignore = "Requires active Redis instance"]
939    #[tokio::test]
940    async fn test_pagination_validation() {
941        let repo = setup_test_repo().await;
942
943        let invalid_page = PaginationQuery {
944            page: 0,
945            per_page: 10,
946        };
947        let result = repo.list_paginated(invalid_page).await;
948        assert!(matches!(result, Err(RepositoryError::InvalidData(_))));
949
950        let invalid_per_page = PaginationQuery {
951            page: 1,
952            per_page: 0,
953        };
954        let result = repo.list_paginated(invalid_per_page).await;
955        assert!(matches!(result, Err(RepositoryError::InvalidData(_))));
956    }
957
958    #[ignore = "Requires active Redis instance"]
959    #[tokio::test]
960    async fn test_update_nonexistent_relayer() {
961        let repo = setup_test_repo().await;
962        let relayer = create_test_relayer("nonexistent-relayer");
963
964        let result = repo
965            .update("nonexistent-relayer".to_string(), relayer)
966            .await;
967        assert!(matches!(result, Err(RepositoryError::NotFound(_))));
968    }
969
970    #[ignore = "Requires active Redis instance"]
971    #[tokio::test]
972    async fn test_delete_nonexistent_relayer() {
973        let repo = setup_test_repo().await;
974
975        let result = repo.delete_by_id("nonexistent-relayer".to_string()).await;
976        assert!(matches!(result, Err(RepositoryError::NotFound(_))));
977    }
978
979    #[tokio::test]
980    #[ignore = "Requires active Redis instance"]
981    async fn test_has_entries() {
982        let repo = setup_test_repo().await;
983        assert!(!repo.has_entries().await.unwrap());
984
985        let relayer_id = uuid::Uuid::new_v4().to_string();
986        let relayer = create_test_relayer(&relayer_id);
987        repo.create(relayer.clone()).await.unwrap();
988        assert!(repo.has_entries().await.unwrap());
989    }
990
991    #[tokio::test]
992    #[ignore = "Requires active Redis instance"]
993    async fn test_drop_all_entries() {
994        let repo = setup_test_repo().await;
995        let relayer_id = uuid::Uuid::new_v4().to_string();
996        let relayer = create_test_relayer(&relayer_id);
997        repo.create(relayer.clone()).await.unwrap();
998        assert!(repo.has_entries().await.unwrap());
999
1000        repo.drop_all_entries().await.unwrap();
1001        assert!(!repo.has_entries().await.unwrap());
1002    }
1003
1004    #[ignore = "Requires active Redis instance"]
1005    #[tokio::test]
1006    async fn test_list_by_signer_id() {
1007        let repo = setup_test_repo().await;
1008
1009        let relayer1_id = uuid::Uuid::new_v4().to_string();
1010        let relayer2_id = uuid::Uuid::new_v4().to_string();
1011        let relayer3_id = uuid::Uuid::new_v4().to_string();
1012        let signer1_id = uuid::Uuid::new_v4().to_string();
1013        let signer2_id = uuid::Uuid::new_v4().to_string();
1014
1015        let mut relayer1 = create_test_relayer(&relayer1_id);
1016        relayer1.signer_id = signer1_id.clone();
1017        repo.create(relayer1).await.unwrap();
1018
1019        let mut relayer2 = create_test_relayer(&relayer2_id);
1020
1021        relayer2.signer_id = signer2_id.clone();
1022        repo.create(relayer2).await.unwrap();
1023
1024        let mut relayer3 = create_test_relayer(&relayer3_id);
1025        relayer3.signer_id = signer1_id.clone();
1026        repo.create(relayer3).await.unwrap();
1027
1028        let result = repo.list_by_signer_id(&signer1_id).await.unwrap();
1029        assert_eq!(result.len(), 2);
1030        let ids: Vec<_> = result.iter().map(|r| r.id.clone()).collect();
1031        assert!(ids.contains(&relayer1_id));
1032        assert!(ids.contains(&relayer3_id));
1033
1034        let result = repo.list_by_signer_id(&signer2_id).await.unwrap();
1035        assert_eq!(result.len(), 1);
1036
1037        let result = repo.list_by_signer_id("nonexistent").await.unwrap();
1038        assert_eq!(result.len(), 0);
1039    }
1040
1041    #[ignore = "Requires active Redis instance"]
1042    #[tokio::test]
1043    async fn test_list_by_notification_id() {
1044        let repo = setup_test_repo().await;
1045
1046        let relayer1_id = uuid::Uuid::new_v4().to_string();
1047        let mut relayer1 = create_test_relayer(&relayer1_id);
1048        relayer1.notification_id = Some("notif1".to_string());
1049        repo.create(relayer1).await.unwrap();
1050
1051        let relayer2_id = uuid::Uuid::new_v4().to_string();
1052        let mut relayer2 = create_test_relayer(&relayer2_id);
1053        relayer2.notification_id = Some("notif2".to_string());
1054        repo.create(relayer2).await.unwrap();
1055
1056        let relayer3_id = uuid::Uuid::new_v4().to_string();
1057        let mut relayer3 = create_test_relayer(&relayer3_id);
1058        relayer3.notification_id = Some("notif1".to_string());
1059        repo.create(relayer3).await.unwrap();
1060
1061        let relayer4_id = uuid::Uuid::new_v4().to_string();
1062        let mut relayer4 = create_test_relayer(&relayer4_id);
1063        relayer4.notification_id = None;
1064        repo.create(relayer4).await.unwrap();
1065
1066        let result = repo.list_by_notification_id("notif1").await.unwrap();
1067        assert_eq!(result.len(), 2);
1068        let ids: Vec<_> = result.iter().map(|r| r.id.clone()).collect();
1069        assert!(ids.contains(&relayer1_id));
1070        assert!(ids.contains(&relayer3_id));
1071
1072        let result = repo.list_by_notification_id("notif2").await.unwrap();
1073        assert_eq!(result.len(), 1);
1074
1075        let result = repo.list_by_notification_id("nonexistent").await.unwrap();
1076        assert_eq!(result.len(), 0);
1077    }
1078}
openzeppelin_relayer/repositories/relayer/relayer_redis.rs

openzeppelin_relayer/repositories/relayer/
relayer_redis.rs
