cal-redis 0.1.80

Callable Redis Implementation
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104

// File: cal-redis/src/cache/queue.rs

use super::CallableCache;
use cal_core::QueueGroup;
use redis::{AsyncCommands, RedisError};

impl CallableCache {
    /// Store a QueueGroup
    pub async fn queue_group_set(&self, queue: &QueueGroup) -> Result<(), RedisError> {
        println!("[CallableCache::queue_group_set] Setting queue group - Account: {}, Name: {}, Type: {}",
                 queue.account_id, queue.name, queue.get_type());
        let key = crate::constants::QueueKeys::queue(&queue.account_id, &queue.name);
        let json = serde_json::to_string(queue)
            .map_err(super::helpers::serde_to_redis_error)?;

        let mut con = self.redis_connection();
        con.set(&key, json).await?;

        // Track in active queues
        con.sadd(&crate::constants::QueueKeys::active(&queue.account_id), &queue.name).await?;

        // Store metadata
        let meta_key = crate::constants::QueueKeys::metadata(&queue.account_id);
        let meta = serde_json::json!({
            "name": queue.name,
            "type": queue.get_type(),
            "device_id": queue.device_id,
            "entry_count": queue.entries.len()
        });
        con.hset(&meta_key, &queue.name, meta.to_string()).await?;

        Ok(())
    }

    /// Get a QueueGroup
    pub async fn queue_group_get(&self, account_id: &str, queue_name: &str) -> Result<Option<QueueGroup>, RedisError> {
        println!("[CallableCache::queue_group_get] Getting queue group - Account: {}, Name: {}", account_id, queue_name);
        let key = crate::constants::QueueKeys::queue(account_id, queue_name);

        let mut con = self.redis_connection();
        match con.get::<_, Option<String>>(&key).await? {
            Some(json) => {
                let queue: QueueGroup = serde_json::from_str(&json)
                    .map_err(super::helpers::serde_to_redis_error)?;
                // Validate account_id
                if queue.account_id != account_id {
                    println!("[CallableCache::queue_group_get] Account ID mismatch! Expected: {}, Got: {}",
                             account_id, queue.account_id);
                    return Ok(None);
                }
                println!("[CallableCache::queue_group_get] Found queue group: {} entries", queue.entries.len());
                Ok(Some(queue))
            }
            None => {
                println!("[CallableCache::queue_group_get] Queue group not found");
                Ok(None)
            }
        }
    }

    /// Delete a QueueGroup
    pub async fn queue_group_delete(&self, account_id: &str, queue_name: &str) -> Result<(), RedisError> {
        println!("[CallableCache::queue_group_delete] Deleting queue group - Account: {}, Name: {}", account_id, queue_name);
        let key = crate::constants::QueueKeys::queue(account_id, queue_name);
        let mut con = self.redis_connection();

        // Remove queue data
        con.del(&key).await?;

        // Remove from active set
        con.srem(&crate::constants::QueueKeys::active(account_id), queue_name).await?;

        // Remove metadata
        let meta_key = crate::constants::QueueKeys::metadata(account_id);
        con.hdel(&meta_key, queue_name).await?;

        Ok(())
    }

    /// Get all active queue names for an account
    pub async fn queue_group_list(&self, account_id: &str) -> Result<Vec<String>, RedisError> {
        println!("[CallableCache::queue_group_list] Getting all queue names for account: {}", account_id);
        let mut con = self.redis_connection();
        let names: Vec<String> = con.smembers(&crate::constants::QueueKeys::active(account_id)).await?;
        println!("[CallableCache::queue_group_list] Found {} queue names", names.len());
        Ok(names)
    }

    /// Get all QueueGroups for an account
    pub async fn queue_groups_get_all(&self, account_id: &str) -> Result<Vec<QueueGroup>, RedisError> {
        println!("[CallableCache::queue_groups_get_all] Getting all queue groups for account: {}", account_id);
        let queue_names = self.queue_group_list(account_id).await?;
        let mut queues = Vec::new();

        for name in queue_names {
            if let Some(queue) = self.queue_group_get(account_id, &name).await? {
                queues.push(queue);
            }
        }

        println!("[CallableCache::queue_groups_get_all] Returning {} queue groups", queues.len());
        Ok(queues)
    }
}