openzeppelin_relayer/repositories/relayer/
relayer_redis.rs

1//! Redis-backed implementation of the RelayerRepository.
2
3use crate::models::UpdateRelayerRequest;
4use crate::models::{PaginationQuery, RelayerNetworkPolicy, RelayerRepoModel, RepositoryError};
5use crate::repositories::redis_base::RedisRepository;
6use crate::repositories::{BatchRetrievalResult, PaginatedResult, RelayerRepository, Repository};
7use async_trait::async_trait;
8use log::{debug, error, warn};
9use redis::aio::ConnectionManager;
10use redis::AsyncCommands;
11use std::fmt;
12use std::sync::Arc;
13
14const RELAYER_PREFIX: &str = "relayer";
15const RELAYER_LIST_KEY: &str = "relayer_list";
16
17#[derive(Clone)]
18pub struct RedisRelayerRepository {
19    pub client: Arc<ConnectionManager>,
20    pub key_prefix: String,
21}
22
23impl RedisRepository for RedisRelayerRepository {}
24
25impl RedisRelayerRepository {
26    pub fn new(
27        connection_manager: Arc<ConnectionManager>,
28        key_prefix: String,
29    ) -> Result<Self, RepositoryError> {
30        if key_prefix.is_empty() {
31            return Err(RepositoryError::InvalidData(
32                "Redis key prefix cannot be empty".to_string(),
33            ));
34        }
35
36        Ok(Self {
37            client: connection_manager,
38            key_prefix,
39        })
40    }
41
42    /// Generate key for relayer data: relayer:{relayer_id}
43    fn relayer_key(&self, relayer_id: &str) -> String {
44        format!("{}:{}:{}", self.key_prefix, RELAYER_PREFIX, relayer_id)
45    }
46
47    /// Generate key for relayer list: relayer_list (set of all relayer IDs)
48    fn relayer_list_key(&self) -> String {
49        format!("{}:{}", self.key_prefix, RELAYER_LIST_KEY)
50    }
51
52    /// Batch fetch relayers by IDs
53    async fn get_relayers_by_ids(
54        &self,
55        ids: &[String],
56    ) -> Result<BatchRetrievalResult<RelayerRepoModel>, RepositoryError> {
57        if ids.is_empty() {
58            debug!("No relayer IDs provided for batch fetch");
59            return Ok(BatchRetrievalResult {
60                results: vec![],
61                failed_ids: vec![],
62            });
63        }
64
65        let mut conn = self.client.as_ref().clone();
66        let keys: Vec<String> = ids.iter().map(|id| self.relayer_key(id)).collect();
67
68        debug!("Batch fetching {} relayer data", keys.len());
69
70        let values: Vec<Option<String>> = conn
71            .mget(&keys)
72            .await
73            .map_err(|e| self.map_redis_error(e, "batch_fetch_relayers"))?;
74
75        let mut relayers = Vec::new();
76        let mut failed_count = 0;
77        let mut failed_ids = Vec::new();
78        for (i, value) in values.into_iter().enumerate() {
79            match value {
80                Some(json) => {
81                    match self.deserialize_entity(&json, &ids[i], "relayer") {
82                        Ok(relayer) => relayers.push(relayer),
83                        Err(e) => {
84                            failed_count += 1;
85                            error!("Failed to deserialize relayer {}: {}", ids[i], e);
86                            failed_ids.push(ids[i].clone());
87                            // Continue processing other relayers
88                        }
89                    }
90                }
91                None => {
92                    warn!("Relayer {} not found in batch fetch", ids[i]);
93                }
94            }
95        }
96
97        if failed_count > 0 {
98            warn!(
99                "Failed to deserialize {} out of {} relayers in batch",
100                failed_count,
101                ids.len()
102            );
103        }
104
105        debug!("Successfully fetched {} relayers", relayers.len());
106        Ok(BatchRetrievalResult {
107            results: relayers,
108            failed_ids,
109        })
110    }
111}
112
113impl fmt::Debug for RedisRelayerRepository {
114    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
115        f.debug_struct("RedisRelayerRepository")
116            .field("client", &"<ConnectionManager>")
117            .field("key_prefix", &self.key_prefix)
118            .finish()
119    }
120}
121
122#[async_trait]
123impl Repository<RelayerRepoModel, String> for RedisRelayerRepository {
124    async fn create(&self, entity: RelayerRepoModel) -> Result<RelayerRepoModel, RepositoryError> {
125        if entity.id.is_empty() {
126            return Err(RepositoryError::InvalidData(
127                "Relayer ID cannot be empty".to_string(),
128            ));
129        }
130
131        if entity.name.is_empty() {
132            return Err(RepositoryError::InvalidData(
133                "Relayer name cannot be empty".to_string(),
134            ));
135        }
136
137        let mut conn = self.client.as_ref().clone();
138        let relayer_key = self.relayer_key(&entity.id);
139
140        // Check if relayer already exists
141        let exists: bool = conn
142            .exists(&relayer_key)
143            .await
144            .map_err(|e| self.map_redis_error(e, "create_relayer_exists_check"))?;
145
146        if exists {
147            return Err(RepositoryError::ConstraintViolation(format!(
148                "Relayer with ID {} already exists",
149                entity.id
150            )));
151        }
152
153        let serialized = self.serialize_entity(&entity, |r| &r.id, "relayer")?;
154
155        // Use pipeline for atomic operations
156        let mut pipe = redis::pipe();
157        pipe.atomic();
158        pipe.set(&relayer_key, &serialized);
159        pipe.sadd(self.relayer_list_key(), &entity.id);
160
161        pipe.exec_async(&mut conn)
162            .await
163            .map_err(|e| self.map_redis_error(e, "create_relayer_pipeline"))?;
164
165        debug!("Created relayer {}", entity.id);
166        Ok(entity)
167    }
168
169    async fn get_by_id(&self, id: String) -> Result<RelayerRepoModel, RepositoryError> {
170        if id.is_empty() {
171            return Err(RepositoryError::InvalidData(
172                "Relayer ID cannot be empty".to_string(),
173            ));
174        }
175
176        let mut conn = self.client.as_ref().clone();
177        let relayer_key = self.relayer_key(&id);
178
179        debug!("Fetching relayer {}", id);
180
181        let json: Option<String> = conn
182            .get(&relayer_key)
183            .await
184            .map_err(|e| self.map_redis_error(e, "get_relayer_by_id"))?;
185
186        match json {
187            Some(json) => {
188                debug!("Found relayer {}", id);
189                self.deserialize_entity(&json, &id, "relayer")
190            }
191            None => {
192                debug!("Relayer {} not found", id);
193                Err(RepositoryError::NotFound(format!(
194                    "Relayer with ID {} not found",
195                    id
196                )))
197            }
198        }
199    }
200
201    async fn list_all(&self) -> Result<Vec<RelayerRepoModel>, RepositoryError> {
202        let mut conn = self.client.as_ref().clone();
203        let relayer_list_key = self.relayer_list_key();
204
205        debug!("Listing all relayers");
206
207        let relayer_ids: Vec<String> = conn
208            .smembers(&relayer_list_key)
209            .await
210            .map_err(|e| self.map_redis_error(e, "list_all_relayers"))?;
211
212        debug!("Found {} relayers in index", relayer_ids.len());
213
214        let relayers = self.get_relayers_by_ids(&relayer_ids).await?;
215        Ok(relayers.results)
216    }
217
218    async fn list_paginated(
219        &self,
220        query: PaginationQuery,
221    ) -> Result<PaginatedResult<RelayerRepoModel>, RepositoryError> {
222        if query.page == 0 {
223            return Err(RepositoryError::InvalidData(
224                "Page number must be greater than 0".to_string(),
225            ));
226        }
227
228        if query.per_page == 0 {
229            return Err(RepositoryError::InvalidData(
230                "Per page count must be greater than 0".to_string(),
231            ));
232        }
233
234        let mut conn = self.client.as_ref().clone();
235        let relayer_list_key = self.relayer_list_key();
236
237        // Get total count
238        let total: u64 = conn
239            .scard(&relayer_list_key)
240            .await
241            .map_err(|e| self.map_redis_error(e, "list_paginated_count"))?;
242
243        if total == 0 {
244            return Ok(PaginatedResult {
245                items: vec![],
246                total: 0,
247                page: query.page,
248                per_page: query.per_page,
249            });
250        }
251
252        // Get all IDs and paginate in memory
253        let all_ids: Vec<String> = conn
254            .smembers(&relayer_list_key)
255            .await
256            .map_err(|e| self.map_redis_error(e, "list_paginated_members"))?;
257
258        let start = ((query.page - 1) * query.per_page) as usize;
259        let end = (start + query.per_page as usize).min(all_ids.len());
260
261        let page_ids = &all_ids[start..end];
262        let items = self.get_relayers_by_ids(page_ids).await?;
263
264        Ok(PaginatedResult {
265            items: items.results.clone(),
266            total,
267            page: query.page,
268            per_page: query.per_page,
269        })
270    }
271
272    async fn update(
273        &self,
274        id: String,
275        entity: RelayerRepoModel,
276    ) -> Result<RelayerRepoModel, RepositoryError> {
277        if id.is_empty() {
278            return Err(RepositoryError::InvalidData(
279                "Relayer ID cannot be empty".to_string(),
280            ));
281        }
282
283        if entity.name.is_empty() {
284            return Err(RepositoryError::InvalidData(
285                "Relayer name cannot be empty".to_string(),
286            ));
287        }
288
289        let mut conn = self.client.as_ref().clone();
290        let relayer_key = self.relayer_key(&id);
291
292        // Check if relayer exists
293        let exists: bool = conn
294            .exists(&relayer_key)
295            .await
296            .map_err(|e| self.map_redis_error(e, "update_relayer_exists_check"))?;
297
298        if !exists {
299            return Err(RepositoryError::NotFound(format!(
300                "Relayer with ID {} not found",
301                id
302            )));
303        }
304
305        // Ensure we preserve the original ID
306        let mut updated_entity = entity;
307        updated_entity.id = id.clone();
308
309        let serialized = self.serialize_entity(&updated_entity, |r| &r.id, "relayer")?;
310
311        // Use pipeline for atomic operations
312        let mut pipe = redis::pipe();
313        pipe.atomic();
314        pipe.set(&relayer_key, &serialized);
315        pipe.sadd(self.relayer_list_key(), &id);
316
317        pipe.exec_async(&mut conn)
318            .await
319            .map_err(|e| self.map_redis_error(e, "update_relayer_pipeline"))?;
320
321        debug!("Updated relayer {}", id);
322        Ok(updated_entity)
323    }
324
325    async fn delete_by_id(&self, id: String) -> Result<(), RepositoryError> {
326        if id.is_empty() {
327            return Err(RepositoryError::InvalidData(
328                "Relayer ID cannot be empty".to_string(),
329            ));
330        }
331
332        let mut conn = self.client.as_ref().clone();
333        let relayer_key = self.relayer_key(&id);
334
335        // Check if relayer exists
336        let exists: bool = conn
337            .exists(&relayer_key)
338            .await
339            .map_err(|e| self.map_redis_error(e, "delete_relayer_exists_check"))?;
340
341        if !exists {
342            return Err(RepositoryError::NotFound(format!(
343                "Relayer with ID {} not found",
344                id
345            )));
346        }
347
348        // Use pipeline for atomic operations
349        let mut pipe = redis::pipe();
350        pipe.atomic();
351        pipe.del(&relayer_key);
352        pipe.srem(self.relayer_list_key(), &id);
353
354        pipe.exec_async(&mut conn)
355            .await
356            .map_err(|e| self.map_redis_error(e, "delete_relayer_pipeline"))?;
357
358        debug!("Deleted relayer {}", id);
359        Ok(())
360    }
361
362    async fn count(&self) -> Result<usize, RepositoryError> {
363        let mut conn = self.client.as_ref().clone();
364        let relayer_list_key = self.relayer_list_key();
365
366        let count: u64 = conn
367            .scard(&relayer_list_key)
368            .await
369            .map_err(|e| self.map_redis_error(e, "count_relayers"))?;
370
371        Ok(count as usize)
372    }
373
374    async fn has_entries(&self) -> Result<bool, RepositoryError> {
375        let mut conn = self.client.as_ref().clone();
376        let relayer_list_key = self.relayer_list_key();
377
378        debug!("Checking if relayer entries exist");
379
380        let exists: bool = conn
381            .exists(&relayer_list_key)
382            .await
383            .map_err(|e| self.map_redis_error(e, "has_entries_check"))?;
384
385        debug!("Relayer entries exist: {}", exists);
386        Ok(exists)
387    }
388
389    async fn drop_all_entries(&self) -> Result<(), RepositoryError> {
390        let mut conn = self.client.as_ref().clone();
391        let relayer_list_key = self.relayer_list_key();
392
393        debug!("Dropping all relayer entries");
394
395        // Get all relayer IDs first
396        let relayer_ids: Vec<String> = conn
397            .smembers(&relayer_list_key)
398            .await
399            .map_err(|e| self.map_redis_error(e, "drop_all_entries_get_ids"))?;
400
401        if relayer_ids.is_empty() {
402            debug!("No relayer entries to drop");
403            return Ok(());
404        }
405
406        // Use pipeline for atomic operations
407        let mut pipe = redis::pipe();
408        pipe.atomic();
409
410        // Delete all individual relayer entries
411        for relayer_id in &relayer_ids {
412            let relayer_key = self.relayer_key(relayer_id);
413            pipe.del(&relayer_key);
414        }
415
416        // Delete the relayer list key
417        pipe.del(&relayer_list_key);
418
419        pipe.exec_async(&mut conn)
420            .await
421            .map_err(|e| self.map_redis_error(e, "drop_all_entries_pipeline"))?;
422
423        debug!("Dropped {} relayer entries", relayer_ids.len());
424        Ok(())
425    }
426}
427
428#[async_trait]
429impl RelayerRepository for RedisRelayerRepository {
430    async fn list_active(&self) -> Result<Vec<RelayerRepoModel>, RepositoryError> {
431        let all_relayers = self.list_all().await?;
432        let active_relayers: Vec<RelayerRepoModel> = all_relayers
433            .into_iter()
434            .filter(|relayer| !relayer.paused)
435            .collect();
436
437        debug!("Found {} active relayers", active_relayers.len());
438        Ok(active_relayers)
439    }
440
441    async fn list_by_signer_id(
442        &self,
443        signer_id: &str,
444    ) -> Result<Vec<RelayerRepoModel>, RepositoryError> {
445        let all_relayers = self.list_all().await?;
446        let relayers_with_signer: Vec<RelayerRepoModel> = all_relayers
447            .into_iter()
448            .filter(|relayer| relayer.signer_id == signer_id)
449            .collect();
450
451        debug!(
452            "Found {} relayers using signer '{}'",
453            relayers_with_signer.len(),
454            signer_id
455        );
456        Ok(relayers_with_signer)
457    }
458
459    async fn list_by_notification_id(
460        &self,
461        notification_id: &str,
462    ) -> Result<Vec<RelayerRepoModel>, RepositoryError> {
463        let all_relayers = self.list_all().await?;
464        let relayers_with_notification: Vec<RelayerRepoModel> = all_relayers
465            .into_iter()
466            .filter(|relayer| {
467                relayer
468                    .notification_id
469                    .as_ref()
470                    .is_some_and(|id| id == notification_id)
471            })
472            .collect();
473
474        debug!(
475            "Found {} relayers using notification '{}'",
476            relayers_with_notification.len(),
477            notification_id
478        );
479        Ok(relayers_with_notification)
480    }
481
482    async fn partial_update(
483        &self,
484        id: String,
485        update: UpdateRelayerRequest,
486    ) -> Result<RelayerRepoModel, RepositoryError> {
487        // First get the current relayer
488        let mut relayer = self.get_by_id(id.clone()).await?;
489
490        // Apply the partial update
491        if let Some(paused) = update.paused {
492            relayer.paused = paused;
493        }
494
495        // Update the relayer
496        self.update(id, relayer).await
497    }
498
499    async fn enable_relayer(
500        &self,
501        relayer_id: String,
502    ) -> Result<RelayerRepoModel, RepositoryError> {
503        // First get the current relayer
504        let mut relayer = self.get_by_id(relayer_id.clone()).await?;
505
506        // Update the system_disabled flag
507        relayer.system_disabled = false;
508
509        // Update the relayer
510        self.update(relayer_id, relayer).await
511    }
512
513    async fn disable_relayer(
514        &self,
515        relayer_id: String,
516    ) -> Result<RelayerRepoModel, RepositoryError> {
517        // First get the current relayer
518        let mut relayer = self.get_by_id(relayer_id.clone()).await?;
519
520        // Update the system_disabled flag
521        relayer.system_disabled = true;
522
523        // Update the relayer
524        self.update(relayer_id, relayer).await
525    }
526
527    async fn update_policy(
528        &self,
529        id: String,
530        policy: RelayerNetworkPolicy,
531    ) -> Result<RelayerRepoModel, RepositoryError> {
532        // First get the current relayer
533        let mut relayer = self.get_by_id(id.clone()).await?;
534
535        // Update the policy
536        relayer.policies = policy;
537
538        // Update the relayer
539        self.update(id, relayer).await
540    }
541}
542
543#[cfg(test)]
544mod tests {
545    use super::*;
546    use crate::models::{NetworkType, RelayerEvmPolicy, RelayerNetworkPolicy};
547    use redis::aio::ConnectionManager;
548    use std::sync::Arc;
549
550    fn create_test_relayer(id: &str) -> RelayerRepoModel {
551        RelayerRepoModel {
552            id: id.to_string(),
553            name: format!("Test Relayer {}", id),
554            network: "ethereum".to_string(),
555            paused: false,
556            network_type: NetworkType::Evm,
557            signer_id: "test-signer".to_string(),
558            policies: RelayerNetworkPolicy::Evm(RelayerEvmPolicy::default()),
559            address: "0x742d35Cc6634C0532925a3b844Bc454e4438f44e".to_string(),
560            notification_id: None,
561            system_disabled: false,
562            custom_rpc_urls: None,
563        }
564    }
565
566    fn create_test_relayer_with_pause(id: &str, paused: bool) -> RelayerRepoModel {
567        let mut relayer = create_test_relayer(id);
568        relayer.paused = paused;
569        relayer
570    }
571
572    async fn setup_test_repo() -> RedisRelayerRepository {
573        let redis_url =
574            std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379/".to_string());
575        let client = redis::Client::open(redis_url).expect("Failed to create Redis client");
576        let connection_manager = ConnectionManager::new(client)
577            .await
578            .expect("Failed to create Redis connection manager");
579
580        RedisRelayerRepository::new(Arc::new(connection_manager), "test".to_string())
581            .expect("Failed to create Redis relayer repository")
582    }
583
584    #[ignore = "Requires active Redis instance"]
585    #[tokio::test]
586    async fn test_new_repository_creation() {
587        let repo = setup_test_repo().await;
588        assert_eq!(repo.key_prefix, "test");
589    }
590
591    #[ignore = "Requires active Redis instance"]
592    #[tokio::test]
593    async fn test_new_repository_empty_prefix_fails() {
594        let redis_url =
595            std::env::var("REDIS_URL").unwrap_or_else(|_| "redis://127.0.0.1:6379/".to_string());
596        let client = redis::Client::open(redis_url).expect("Failed to create Redis client");
597        let connection_manager = ConnectionManager::new(client)
598            .await
599            .expect("Failed to create Redis connection manager");
600
601        let result = RedisRelayerRepository::new(Arc::new(connection_manager), "".to_string());
602        assert!(matches!(result, Err(RepositoryError::InvalidData(_))));
603    }
604
605    #[ignore = "Requires active Redis instance"]
606    #[tokio::test]
607    async fn test_key_generation() {
608        let repo = setup_test_repo().await;
609
610        let relayer_key = repo.relayer_key("test-relayer");
611        assert_eq!(relayer_key, "test:relayer:test-relayer");
612
613        let list_key = repo.relayer_list_key();
614        assert_eq!(list_key, "test:relayer_list");
615    }
616
617    #[ignore = "Requires active Redis instance"]
618    #[tokio::test]
619    async fn test_serialize_deserialize_relayer() {
620        let repo = setup_test_repo().await;
621        let relayer = create_test_relayer("test-relayer");
622
623        let serialized = repo
624            .serialize_entity(&relayer, |r| &r.id, "relayer")
625            .unwrap();
626        let deserialized: RelayerRepoModel = repo
627            .deserialize_entity(&serialized, &relayer.id, "relayer")
628            .unwrap();
629
630        assert_eq!(relayer.id, deserialized.id);
631        assert_eq!(relayer.name, deserialized.name);
632        assert_eq!(relayer.network, deserialized.network);
633        assert_eq!(relayer.paused, deserialized.paused);
634        assert_eq!(relayer.network_type, deserialized.network_type);
635        assert_eq!(relayer.signer_id, deserialized.signer_id);
636        assert_eq!(relayer.address, deserialized.address);
637        assert_eq!(relayer.notification_id, deserialized.notification_id);
638        assert_eq!(relayer.system_disabled, deserialized.system_disabled);
639        assert_eq!(relayer.custom_rpc_urls, deserialized.custom_rpc_urls);
640    }
641
642    #[ignore = "Requires active Redis instance"]
643    #[tokio::test]
644    async fn test_create_relayer() {
645        let repo = setup_test_repo().await;
646        let relayer_id = uuid::Uuid::new_v4().to_string();
647        let relayer = create_test_relayer(&relayer_id);
648
649        let result = repo.create(relayer.clone()).await;
650        assert!(result.is_ok());
651
652        let created_relayer = result.unwrap();
653        assert_eq!(created_relayer.id, relayer_id);
654        assert_eq!(created_relayer.name, relayer.name);
655    }
656
657    #[ignore = "Requires active Redis instance"]
658    #[tokio::test]
659    async fn test_get_relayer() {
660        let repo = setup_test_repo().await;
661        let relayer_id = uuid::Uuid::new_v4().to_string();
662        let relayer = create_test_relayer(&relayer_id);
663
664        repo.create(relayer.clone()).await.unwrap();
665
666        let retrieved = repo.get_by_id(relayer_id).await.unwrap();
667        assert_eq!(retrieved.id, relayer.id);
668        assert_eq!(retrieved.name, relayer.name);
669    }
670
671    #[ignore = "Requires active Redis instance"]
672    #[tokio::test]
673    async fn test_list_all_relayers() {
674        let repo = setup_test_repo().await;
675        let relayer1_id = uuid::Uuid::new_v4().to_string();
676        let relayer2_id = uuid::Uuid::new_v4().to_string();
677        let relayer1 = create_test_relayer(&relayer1_id);
678        let relayer2 = create_test_relayer(&relayer2_id);
679
680        repo.create(relayer1).await.unwrap();
681        repo.create(relayer2).await.unwrap();
682
683        let all_relayers = repo.list_all().await.unwrap();
684        assert!(all_relayers.len() >= 2);
685    }
686
687    #[ignore = "Requires active Redis instance"]
688    #[tokio::test]
689    async fn test_list_active_relayers() {
690        let repo = setup_test_repo().await;
691        let relayer1_id = uuid::Uuid::new_v4().to_string();
692        let relayer2_id = uuid::Uuid::new_v4().to_string();
693        let relayer1 = create_test_relayer_with_pause(&relayer1_id, false);
694        let relayer2 = create_test_relayer_with_pause(&relayer2_id, true);
695
696        repo.create(relayer1).await.unwrap();
697        repo.create(relayer2).await.unwrap();
698
699        let active_relayers = repo.list_active().await.unwrap();
700        // Should have at least 1 active relayer
701        assert!(!active_relayers.is_empty());
702        // All returned relayers should be active
703        assert!(active_relayers.iter().all(|r| !r.paused));
704    }
705
706    #[ignore = "Requires active Redis instance"]
707    #[tokio::test]
708    async fn test_count_relayers() {
709        let repo = setup_test_repo().await;
710        let relayer_id = uuid::Uuid::new_v4().to_string();
711        let relayer = create_test_relayer(&relayer_id);
712
713        repo.create(relayer).await.unwrap();
714
715        let count = repo.count().await.unwrap();
716        assert!(count >= 1);
717    }
718
719    #[ignore = "Requires active Redis instance"]
720    #[tokio::test]
721    async fn test_get_nonexistent_relayer() {
722        let repo = setup_test_repo().await;
723
724        let result = repo.get_by_id("nonexistent-relayer".to_string()).await;
725        assert!(matches!(result, Err(RepositoryError::NotFound(_))));
726    }
727
728    #[ignore = "Requires active Redis instance"]
729    #[tokio::test]
730    async fn test_duplicate_relayer_creation() {
731        let repo = setup_test_repo().await;
732        let relayer_id = uuid::Uuid::new_v4().to_string();
733        let relayer = create_test_relayer(&relayer_id);
734
735        repo.create(relayer.clone()).await.unwrap();
736
737        let duplicate_result = repo.create(relayer).await;
738        assert!(matches!(
739            duplicate_result,
740            Err(RepositoryError::ConstraintViolation(_))
741        ));
742    }
743
744    #[ignore = "Requires active Redis instance"]
745    #[tokio::test]
746    async fn test_update_relayer() {
747        let repo = setup_test_repo().await;
748        let relayer_id = uuid::Uuid::new_v4().to_string();
749        let relayer = create_test_relayer(&relayer_id);
750
751        repo.create(relayer.clone()).await.unwrap();
752
753        let mut updated_relayer = relayer.clone();
754        updated_relayer.name = "Updated Relayer Name".to_string();
755
756        let result = repo.update(relayer.id.clone(), updated_relayer).await;
757        assert!(result.is_ok());
758
759        let updated = result.unwrap();
760        assert_eq!(updated.name, "Updated Relayer Name");
761        assert_eq!(updated.id, relayer.id);
762    }
763
764    #[ignore = "Requires active Redis instance"]
765    #[tokio::test]
766    async fn test_delete_relayer() {
767        let repo = setup_test_repo().await;
768        let relayer_id = uuid::Uuid::new_v4().to_string();
769        let relayer = create_test_relayer(&relayer_id);
770
771        repo.create(relayer.clone()).await.unwrap();
772
773        let delete_result = repo.delete_by_id(relayer.id.clone()).await;
774        assert!(delete_result.is_ok());
775
776        let get_result = repo.get_by_id(relayer.id).await;
777        assert!(matches!(get_result, Err(RepositoryError::NotFound(_))));
778    }
779
780    #[ignore = "Requires active Redis instance"]
781    #[tokio::test]
782    async fn test_list_paginated() {
783        let repo = setup_test_repo().await;
784        let relayer1_id = uuid::Uuid::new_v4().to_string();
785        let relayer2_id = uuid::Uuid::new_v4().to_string();
786        let relayer1 = create_test_relayer(&relayer1_id);
787        let relayer2 = create_test_relayer(&relayer2_id);
788
789        repo.create(relayer1).await.unwrap();
790        repo.create(relayer2).await.unwrap();
791
792        let query = PaginationQuery {
793            page: 1,
794            per_page: 10,
795        };
796
797        let result = repo.list_paginated(query).await.unwrap();
798        assert!(result.total >= 2);
799        assert_eq!(result.page, 1);
800        assert_eq!(result.per_page, 10);
801    }
802
803    #[ignore = "Requires active Redis instance"]
804    #[tokio::test]
805    async fn test_partial_update_relayer() {
806        let repo = setup_test_repo().await;
807        let relayer_id = uuid::Uuid::new_v4().to_string();
808        let relayer = create_test_relayer(&relayer_id);
809
810        repo.create(relayer.clone()).await.unwrap();
811
812        let update = UpdateRelayerRequest {
813            paused: Some(true),
814            ..Default::default()
815        };
816        let result = repo.partial_update(relayer.id.clone(), update).await;
817        assert!(result.is_ok());
818
819        let updated = result.unwrap();
820        assert_eq!(updated.id, relayer.id);
821        assert!(updated.paused);
822    }
823
824    #[ignore = "Requires active Redis instance"]
825    #[tokio::test]
826    async fn test_enable_disable_relayer() {
827        let repo = setup_test_repo().await;
828        let relayer_id = uuid::Uuid::new_v4().to_string();
829        let relayer = create_test_relayer(&relayer_id);
830
831        repo.create(relayer.clone()).await.unwrap();
832
833        // Test disable
834        let disabled = repo.disable_relayer(relayer.id.clone()).await.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
