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

openzeppelin_relayer/repositories/relayer/
relayer_redis.rs
