matrix_lite_rs/storage/

redis_storage.rs

use super::Storage;
use crate::Result;
use crate::types::{Event, Room, RoomId, UserId, DeviceId, Session, Device, EventId};
use async_trait::async_trait;
use redis::{aio::ConnectionManager, AsyncCommands, Client};

/// Redis storage implementation for Matrix
pub struct RedisStorage {
    client: ConnectionManager,
}

impl RedisStorage {
    /// Create a new Redis storage instance
    pub async fn new(redis_url: &str) -> Result<Self> {
        let client = Client::open(redis_url)?;
        let connection = ConnectionManager::new(client).await?;
        Ok(Self { client: connection })
    }

    /// Create storage with an existing connection
    pub fn with_connection(client: ConnectionManager) -> Self {
        Self { client }
    }

    // Key generation helpers
    fn event_key(event_id: &EventId) -> String {
        format!("matrix:event:{}", event_id.as_str())
    }

    fn room_events_key(room_id: &RoomId) -> String {
        format!("matrix:room:{}:events", room_id.as_str())
    }

    fn room_key(room_id: &RoomId) -> String {
        format!("matrix:room:{}", room_id.as_str())
    }

    fn user_rooms_key(user_id: &UserId) -> String {
        format!("matrix:user:{}:rooms", user_id.as_str())
    }

    fn session_key(access_token: &str) -> String {
        format!("matrix:session:{}", access_token)
    }

    fn device_key(user_id: &UserId, device_id: &DeviceId) -> String {
        format!("matrix:device:{}:{}", user_id.as_str(), device_id.as_str())
    }

    fn user_devices_key(user_id: &UserId) -> String {
        format!("matrix:user:{}:devices", user_id.as_str())
    }

    fn encryption_key_key(user_id: &UserId, device_id: &DeviceId) -> String {
        format!("matrix:encryption:{}:{}", user_id.as_str(), device_id.as_str())
    }

    fn read_marker_key(user_id: &UserId, room_id: &RoomId) -> String {
        format!("matrix:read:{}:{}", user_id.as_str(), room_id.as_str())
    }

    fn unread_count_key(user_id: &UserId, room_id: &RoomId) -> String {
        format!("matrix:unread:{}:{}", user_id.as_str(), room_id.as_str())
    }
}

#[async_trait]
impl Storage for RedisStorage {
    async fn store_event(&self, event: &Event) -> Result<()> {
        let mut conn = self.client.clone();

        // Serialize event to JSON
        let event_json = serde_json::to_string(event)?;

        // Store event data
        let event_key = Self::event_key(&event.event_id);
        conn.set::<_, _, ()>(&event_key, &event_json).await?;

        // Add to room's event list (sorted by timestamp)
        let room_events_key = Self::room_events_key(&event.room_id);
        let timestamp = event.timestamp.timestamp_millis() as f64;
        conn.zadd::<_, _, _, ()>(&room_events_key, event.event_id.as_str(), timestamp).await?;

        Ok(())
    }

    async fn get_room_events(&self, room_id: &RoomId, limit: usize) -> Result<Vec<Event>> {
        let mut conn = self.client.clone();

        let room_events_key = Self::room_events_key(room_id);

        // Get the last N event IDs (newest first)
        let event_ids: Vec<String> = conn.zrevrange(&room_events_key, 0, (limit - 1) as isize).await?;

        let mut events = Vec::new();

        for event_id_str in event_ids {
            let event_id = EventId::new(event_id_str);
            let event_key = Self::event_key(&event_id);

            if let Some(event_json) = conn.get::<_, Option<String>>(&event_key).await? {
                let event: Event = serde_json::from_str(&event_json)?;
                events.push(event);
            }
        }

        // Reverse to get chronological order (oldest first)
        events.reverse();

        Ok(events)
    }

    async fn store_room(&self, room: &Room) -> Result<()> {
        let mut conn = self.client.clone();

        let room_json = serde_json::to_string(room)?;
        let room_key = Self::room_key(&room.room_id);

        // Store room data
        conn.set::<_, _, ()>(&room_key, &room_json).await?;

        // Add to each member's room list
        for member in &room.members {
            let user_rooms_key = Self::user_rooms_key(&member.user_id);
            conn.sadd::<_, _, ()>(&user_rooms_key, room.room_id.as_str()).await?;
        }

        Ok(())
    }

    async fn get_room(&self, room_id: &RoomId) -> Result<Option<Room>> {
        let mut conn = self.client.clone();
        let room_key = Self::room_key(room_id);

        let room_json: Option<String> = conn.get(&room_key).await?;

        match room_json {
            Some(json) => {
                let room: Room = serde_json::from_str(&json)?;
                Ok(Some(room))
            }
            None => Ok(None),
        }
    }

    async fn get_user_rooms(&self, user_id: &UserId) -> Result<Vec<Room>> {
        let mut conn = self.client.clone();
        let user_rooms_key = Self::user_rooms_key(user_id);

        // Get all room IDs for this user
        let room_ids: Vec<String> = conn.smembers(&user_rooms_key).await?;

        let mut rooms = Vec::new();

        for room_id_str in room_ids {
            let room_id = RoomId::new(room_id_str);
            if let Some(room) = self.get_room(&room_id).await? {
                rooms.push(room);
            }
        }

        Ok(rooms)
    }

    async fn store_session(&self, session: &Session) -> Result<()> {
        let mut conn = self.client.clone();

        let session_json = serde_json::to_string(session)?;
        let session_key = Self::session_key(&session.access_token);

        // Store session with optional expiry
        if let Some(expires_at) = session.expires_at {
            let ttl = (expires_at - session.created_at).num_seconds() as u64;
            conn.set_ex::<_, _, ()>(&session_key, &session_json, ttl).await?;
        } else {
            conn.set::<_, _, ()>(&session_key, &session_json).await?;
        }

        Ok(())
    }

    async fn get_session(&self, access_token: &str) -> Result<Option<Session>> {
        let mut conn = self.client.clone();
        let session_key = Self::session_key(access_token);

        let session_json: Option<String> = conn.get(&session_key).await?;

        match session_json {
            Some(json) => {
                let session: Session = serde_json::from_str(&json)?;
                Ok(Some(session))
            }
            None => Ok(None),
        }
    }

    async fn delete_session(&self, access_token: &str) -> Result<()> {
        let mut conn = self.client.clone();
        let session_key = Self::session_key(access_token);

        conn.del::<_, ()>(&session_key).await?;

        Ok(())
    }

    async fn store_device(&self, device: &Device) -> Result<()> {
        let mut conn = self.client.clone();

        let device_json = serde_json::to_string(device)?;
        let device_key = Self::device_key(&device.user_id, &device.device_id);

        // Store device
        conn.set::<_, _, ()>(&device_key, &device_json).await?;

        // Add to user's device list
        let user_devices_key = Self::user_devices_key(&device.user_id);
        conn.sadd::<_, _, ()>(&user_devices_key, device.device_id.as_str()).await?;

        Ok(())
    }

    async fn get_device(&self, user_id: &UserId, device_id: &DeviceId) -> Result<Option<Device>> {
        let mut conn = self.client.clone();
        let device_key = Self::device_key(user_id, device_id);

        let device_json: Option<String> = conn.get(&device_key).await?;

        match device_json {
            Some(json) => {
                let device: Device = serde_json::from_str(&json)?;
                Ok(Some(device))
            }
            None => Ok(None),
        }
    }

    async fn get_user_devices(&self, user_id: &UserId) -> Result<Vec<Device>> {
        let mut conn = self.client.clone();
        let user_devices_key = Self::user_devices_key(user_id);

        let device_ids: Vec<String> = conn.smembers(&user_devices_key).await?;

        let mut devices = Vec::new();

        for device_id_str in device_ids {
            let device_id = DeviceId::new(device_id_str);
            if let Some(device) = self.get_device(user_id, &device_id).await? {
                devices.push(device);
            }
        }

        Ok(devices)
    }

    async fn delete_device(&self, user_id: &UserId, device_id: &DeviceId) -> Result<()> {
        let mut conn = self.client.clone();

        let device_key = Self::device_key(user_id, device_id);
        let user_devices_key = Self::user_devices_key(user_id);

        // Delete device data
        conn.del::<_, ()>(&device_key).await?;

        // Remove from user's device list
        conn.srem::<_, _, ()>(&user_devices_key, device_id.as_str()).await?;

        Ok(())
    }

    async fn store_encryption_key(&self, user_id: &UserId, device_id: &DeviceId, key: &str) -> Result<()> {
        let mut conn = self.client.clone();
        let key_key = Self::encryption_key_key(user_id, device_id);

        conn.set::<_, _, ()>(&key_key, key).await?;

        Ok(())
    }

    async fn get_encryption_key(&self, user_id: &UserId, device_id: &DeviceId) -> Result<Option<String>> {
        let mut conn = self.client.clone();
        let key_key = Self::encryption_key_key(user_id, device_id);

        let key: Option<String> = conn.get(&key_key).await?;

        Ok(key)
    }

    async fn mark_read(&self, user_id: &UserId, room_id: &RoomId, event_id: &str) -> Result<()> {
        let mut conn = self.client.clone();

        let read_marker_key = Self::read_marker_key(user_id, room_id);
        let unread_count_key = Self::unread_count_key(user_id, room_id);

        // Store read marker
        conn.set::<_, _, ()>(&read_marker_key, event_id).await?;

        // Reset unread count
        conn.set::<_, _, ()>(&unread_count_key, 0).await?;

        Ok(())
    }

    async fn get_unread_count(&self, user_id: &UserId, room_id: &RoomId) -> Result<usize> {
        let mut conn = self.client.clone();
        let unread_count_key = Self::unread_count_key(user_id, room_id);

        let count: Option<usize> = conn.get(&unread_count_key).await?;

        Ok(count.unwrap_or(0))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use chrono::Utc;

    // Tests will be implemented with comprehensive coverage
    #[tokio::test]
    async fn test_storage_creation() {
        // This is a placeholder - real tests will need Redis instance
        assert!(true);
    }

    #[tokio::test]
    async fn test_event_storage() {
        // Will test event storage and retrieval
        assert!(true);
    }

    #[tokio::test]
    async fn test_room_management() {
        // Will test room creation, updates, and membership
        assert!(true);
    }

    #[tokio::test]
    async fn test_session_management() {
        // Will test session creation, validation, and expiry
        assert!(true);
    }

    #[tokio::test]
    async fn test_device_management() {
        // Will test device registration and removal
        assert!(true);
    }

    #[tokio::test]
    async fn test_encryption_keys() {
        // Will test key storage and retrieval
        assert!(true);
    }

    #[tokio::test]
    async fn test_read_markers() {
        // Will test read markers and unread counts
        assert!(true);
    }
}