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
105
106
107
108
109
110
111
112

// File: cal-redis/src/helpers.rs - Updated session methods

use async_trait::async_trait;
use redis::{AsyncCommands, RedisError};
use cal_core::authentication::SessionData;
use cal_core::agent::AgentStatus;
use crate::CallableCache;

/// Trait for Redis operations that can be mixed into CallableCache
#[async_trait]
pub trait RedisHelpers {
    // Session Management with SessionData
    async fn set_session(&self, account_id: &str, session_data: &SessionData) -> Result<(), RedisError>;
    async fn get_session(&self, account_id: &str, session_id: &str) -> Result<Option<SessionData>, RedisError>;
    async fn create_session(&self, account_id: &str, user_id: &str, agent_id: &str, ttl_seconds: u64) -> Result<SessionData, RedisError>;
    async fn refresh_session(&self, account_id: &str, session_id: &str, ttl_seconds: u64) -> Result<Option<SessionData>, RedisError>;
    async fn delete_session(&self, account_id: &str, session_id: &str) -> Result<(), RedisError>;
    async fn get_active_sessions(&self, account_id: &str) -> Result<Vec<SessionData>, RedisError>;
    async fn cleanup_expired_sessions(&self, account_id: &str) -> Result<u32, RedisError>;

    // Agent Management
    async fn set_agent_status(&self, account_id: &str, status: &AgentStatus) -> Result<(), RedisError>;
    async fn get_agent_status(&self, account_id: &str, user_id: &str) -> Result<Option<AgentStatus>, RedisError>;
    async fn add_connected_agent(&self, account_id: &str, user_id: &str, agent_id: &str, ws_connection_id: &str) -> Result<(), RedisError>;
    async fn remove_connected_agent(&self, account_id: &str, user_id: &str, agent_id: &str, ws_connection_id: &str) -> Result<(), RedisError>;
    async fn get_connected_agents(&self, account_id: &str) -> Result<Vec<String>, RedisError>;

    // Queue Management
    async fn queue_add(&self, account_id: &str, queue_name: &str, item: &str, priority: f64) -> Result<(), RedisError>;
    async fn queue_pop(&self, account_id: &str, queue_name: &str) -> Result<Option<String>, RedisError>;
    async fn queue_length(&self, account_id: &str, queue_name: &str) -> Result<usize, RedisError>;

    // Pub/Sub
    async fn publish_account_event(&self, account_id: &str, event: &cal_core::RedisEvent) -> Result<(), RedisError>;
}

#[async_trait]
impl RedisHelpers for CallableCache {
    // Session operations
    async fn set_session(&self, account_id: &str, session_data: &SessionData) -> Result<(), RedisError> {
        self.session_set(account_id, session_data).await
    }

    async fn get_session(&self, account_id: &str, session_id: &str) -> Result<Option<SessionData>, RedisError> {
        self.session_get(account_id, session_id).await
    }

    async fn create_session(&self, account_id: &str, user_id: &str, agent_id: &str, ttl_seconds: u64) -> Result<SessionData, RedisError> {
        self.session_create(account_id, user_id, agent_id, ttl_seconds).await
    }

    async fn refresh_session(&self, account_id: &str, session_id: &str, ttl_seconds: u64) -> Result<Option<SessionData>, RedisError> {
        self.session_refresh(account_id, session_id, ttl_seconds).await
    }

    async fn delete_session(&self, account_id: &str, session_id: &str) -> Result<(), RedisError> {
        self.session_delete(account_id, session_id).await
    }

    async fn get_active_sessions(&self, account_id: &str) -> Result<Vec<SessionData>, RedisError> {
        self.session_get_active(account_id).await
    }

    async fn cleanup_expired_sessions(&self, account_id: &str) -> Result<u32, RedisError> {
        self.session_cleanup_expired(account_id).await
    }

    // Agent operations
    async fn set_agent_status(&self, account_id: &str, status: &AgentStatus) -> Result<(), RedisError> {
        self.agent_status_set(account_id, status).await
    }

    async fn get_agent_status(&self, account_id: &str, user_id: &str) -> Result<Option<AgentStatus>, RedisError> {
        self.agent_status_get(account_id, user_id).await
    }

    async fn add_connected_agent(&self, account_id: &str, user_id: &str, agent_id: &str, ws_connection_id: &str) -> Result<(), RedisError> {
        self.add_connected_agent(account_id, user_id, agent_id, ws_connection_id).await
    }

    async fn remove_connected_agent(&self, account_id: &str, user_id: &str, agent_id: &str, ws_connection_id: &str) -> Result<(), RedisError> {
        self.remove_connected_agent(account_id, user_id, agent_id, ws_connection_id).await
    }

    async fn get_connected_agents(&self, account_id: &str) -> Result<Vec<String>, RedisError> {
        self.get_connected_agents(account_id).await
    }

    // Queue operations
    async fn queue_add(&self, account_id: &str, queue_name: &str, item: &str, priority: f64) -> Result<(), RedisError> {
        let key = crate::constants::QueueKeys::queue(account_id, queue_name);
        let mut con = self.redis_connection();
        con.zadd(&key, item, priority).await
    }

    async fn queue_pop(&self, account_id: &str, queue_name: &str) -> Result<Option<String>, RedisError> {
        let key = crate::constants::QueueKeys::queue(account_id, queue_name);
        let mut con = self.redis_connection();
        let result: Vec<String> = con.zpopmin(&key, 1).await?;
        Ok(result.into_iter().next())
    }

    async fn queue_length(&self, account_id: &str, queue_name: &str) -> Result<usize, RedisError> {
        let key = crate::constants::QueueKeys::queue(account_id, queue_name);
        let mut con = self.redis_connection();
        con.zcard(&key).await
    }

    async fn publish_account_event(&self, account_id: &str, event: &cal_core::RedisEvent) -> Result<(), RedisError> {
        self.publish_account_event(account_id, event).await
    }
}