cal_redis/

cache.rs

use crate::local_cache::{build_account_identifiers_locally, build_region_identifiers_locally};
use crate::proxy::{get_proxies, get_proxy_by_id};
use crate::redis_cache::RedisCache;
use crate::{
    get_account_by_id, get_accounts, get_asset, get_assets, get_ddi, get_ddis, get_device,
    get_devices, get_hook, get_hooks, get_region_by_id, get_regions, get_trunk, get_trunk_and_ddi,
    get_trunks,
};
use cal_core::device::device::DeviceStruct;
use cal_core::{AccountLite, Asset, FlowState, Hook, Proxy, RedisEvent, Region, Trunk, DDI};
use moka::sync::Cache;
use redis::aio::MultiplexedConnection;
use redis::{AsyncCommands, PushKind, RedisError, Value};
use std::collections::HashMap;
use std::env;
use std::time::Duration;
use tokio::sync::mpsc::unbounded_channel;

/// Main cache structure that combines local in-memory caching with remote Redis storage.
///
/// Provides a unified interface for accessing data with automatic fallback to Redis
/// when items aren't found in the local cache.
#[derive(Clone)]
pub struct CallableCache {
    pub local_cache: LocalCache,
    pub remote_cache: RedisCache,
}

/// In-memory cache for storing frequently accessed data.
///
/// Uses Moka cache for efficient memory usage with TTL-based expiration.
#[derive(Clone)]
pub struct LocalCache {
    /// Cache for flow state objects (longer TTL)
    pub flow_state: Cache<String, FlowState>,
    /// Cache for region objects
    pub regions: Cache<String, Region>,
    /// Cache for region identifier mappings
    pub region_idents: Cache<String, String>,
    /// Cache for account objects
    pub accounts: Cache<String, AccountLite>,
    /// Cache for account identifier mappings
    pub account_idents: Cache<String, String>,
    /// Cache for device objects
    pub devices: Cache<String, DeviceStruct>,
    /// Cache for trunk objects
    pub trunks: Cache<String, Trunk>,
    /// Cache for DDI objects
    pub ddis: Cache<String, DDI>,
    /// Cache for hook objects
    pub hooks: Cache<String, Hook>,
    /// Cache for asset objects
    pub assets: Cache<String, Asset>,
    /// Cache for trunk_ddi identifier mappings
    pub trunk_ddi_idents: Cache<String, String>,
    pub proxies: Cache<String, Proxy>,
}

impl CallableCache {
    /// Creates a new CallableCache instance with local and remote caches.
    ///
    /// # Returns
    /// * `CallableCache` - The new cache instance
    pub async fn new() -> CallableCache {
        let (remote_cache, local_cache) = create_pool().await;
        CallableCache {
            local_cache,
            remote_cache,
        }
    }

    /// Retrieves a region by its ID, first checking the local cache and then Redis.
    ///
    /// # Arguments
    /// * `id` - Region ID to look up
    ///
    /// # Returns
    /// * `Result<Option<Region>, RedisError>` - Region if found, None if not found, or a Redis error
    pub async fn get_region_by_id(self, id: &str) -> Result<Option<Region>, RedisError> {
        // Try to get the resolved ID from the local cache
        if let Some(resolved_id) = self.local_cache.region_idents.get(id) {
            // If we have a resolved ID, try to get the region from the local cache
            if let Some(region) = self.local_cache.regions.get(resolved_id.as_str()) {
                return Ok(Some(region));
            }

            // If not in local cache, fetch from Redis using the resolved ID
            match get_region_by_id(self.remote_cache.connection.clone(), &resolved_id).await? {
                Some(region) => {
                    // Cache the retrieved region for future use
                    build_region_identifiers_locally(self.local_cache, &region.clone());
                    Ok(Some(region))
                }
                None => Ok(None),
            }
        } else {
            // If region identifier not in local cache, try Redis
            match get_region_by_id(self.remote_cache.connection.clone(), id).await? {
                Some(region) => {
                    // Cache the retrieved region for future use
                    build_region_identifiers_locally(self.local_cache, &region.clone());
                    Ok(Some(region))
                }
                None => Ok(None),
            }
        }
    }

    /// Retrieves an account by trunk IP and DDI, first checking the local cache and then Redis.
    ///
    /// This function:
    /// 1. Checks if the trunk-DDI combination exists in the local cache
    /// 2. If found, retrieves the account ID and uses it to look up the account
    /// 3. If not found in local cache, falls back to Redis lookup
    /// 4. Caches the result for future use
    ///
    /// # Arguments
    /// * `ddi_id` - DDI identifier (typically a phone number)
    /// * `trunk_ip` - IP address of the trunk
    ///
    /// # Returns
    /// * `Result<Option<AccountLite>, RedisError>` - Account if found, None if not found, or a Redis error
    pub async fn get_account_by_trunk_ddi(
        self,
        ddi_id: &str,
        trunk_ip: &str,
    ) -> Result<Option<AccountLite>, RedisError> {
        // Create a composite key for the trunk-DDI combination
        let trunk_ddi_key = format!("{}:{}", trunk_ip, ddi_id);

        // Try to get the account ID from the local trunk-DDI cache
        if let Some(account_id) = self.local_cache.trunk_ddi_idents.get(&trunk_ddi_key) {
            // If we have an account ID, try to get the account from the local account cache
            if let Some(account) = self.local_cache.accounts.get(account_id.as_str()) {
                return Ok(Some(account.clone()));
            }

            // If account not in local cache, but we have the ID, fetch from Redis
            match get_account_by_id(self.remote_cache.connection.clone(), account_id.as_str())
                .await?
            {
                Some(account) => {
                    // Cache the retrieved account for future use
                    build_account_identifiers_locally(self.local_cache, &account.clone());
                    Ok(Some(account))
                }
                None => Ok(None),
            }
        } else {
            // If trunk-DDI combination not in local cache, try Redis
            match get_trunk_and_ddi(self.remote_cache.connection.clone(), ddi_id, trunk_ip).await? {
                Some(account) => {
                    // Cache the retrieved account and trunk-DDI mapping for future use
                    build_account_identifiers_locally(self.local_cache.clone(), &account.clone());

                    // Add the trunk-DDI mapping to local cache
                    self.local_cache
                        .trunk_ddi_idents
                        .insert(trunk_ddi_key, account.id.clone());

                    Ok(Some(account))
                }
                None => Ok(None),
            }
        }
    }

    /// Retrieves an account by its ID, first checking the local cache and then Redis.
    ///
    /// # Arguments
    /// * `id` - Account ID to look up
    ///
    /// # Returns
    /// * `Result<Option<AccountLite>, RedisError>` - Account if found, None if not found, or a Redis error
    pub async fn get_account_by_id(self, id: &str) -> Result<Option<AccountLite>, RedisError> {
        // Try to get the resolved ID from the local cache
        if let Some(resolved_id) = self.local_cache.account_idents.get(id) {
            // If we have a resolved ID, try to get the account from the local cache
            if let Some(account) = self.local_cache.accounts.get(resolved_id.as_str()) {
                return Ok(Some(account));
            }

            // If not in local cache, fetch from Redis using the resolved ID
            match get_account_by_id(self.remote_cache.connection.clone(), &resolved_id).await? {
                Some(account) => {
                    // Cache the retrieved account for future use
                    build_account_identifiers_locally(self.local_cache, &account.clone());
                    Ok(Some(account))
                }
                None => Ok(None),
            }
        } else {
            // If account identifier not in local cache, try Redis
            match get_account_by_id(self.remote_cache.connection.clone(), id).await? {
                Some(account) => {
                    // Cache the retrieved account for future use
                    build_account_identifiers_locally(self.local_cache, &account.clone());
                    Ok(Some(account))
                }
                None => Ok(None),
            }
        }
    }

    /// Retrieves all regions from Redis.
    ///
    /// # Returns
    /// * `Result<Vec<Region>, RedisError>` - Vec of Regions
    pub async fn get_regions(self) -> Result<Vec<Region>, RedisError> {
        let regions = get_regions(self.remote_cache.connection).await?;

        for region in &regions {
            build_region_identifiers_locally(self.local_cache.clone(), region);
        }

        Ok(regions)
    }

    /// Retrieves all accounts from Redis.
    ///
    /// # Returns
    /// * `Result<Vec<AccountLite>, RedisError>` - Vec of AccountLite
    pub async fn get_accounts(self) -> Result<Vec<AccountLite>, RedisError> {
        let accounts = get_accounts(self.remote_cache.connection).await?;

        for account in &accounts {
            build_account_identifiers_locally(self.local_cache.clone(), account);
        }

        Ok(accounts)
    }

    /// Retrieves a string value from Redis.
    ///
    /// # Arguments
    /// * `key` - Redis key
    ///
    /// # Returns
    /// * `Result<Option<String>, RedisError>` - String value if found, None if not found, or a Redis error
    pub async fn get_str(self, key: &str) -> Result<Option<String>, RedisError> {
        get_str(self.remote_cache.connection, key).await
    }

    /// Retrieves a field from a Redis hash.
    ///
    /// # Arguments
    /// * `key` - Redis hash key
    /// * `field` - Field to retrieve
    ///
    /// # Returns
    /// * `Result<Option<String>, RedisError>` - Field value if found, None if not found, or a Redis error
    pub async fn get_hash(self, key: &str, field: &str) -> Result<Option<String>, RedisError> {
        get_hash(self.remote_cache.connection, key, field).await
    }

    /// Retrieves all devices for an account.
    ///
    /// # Arguments
    /// * `account_id` - Account ID
    ///
    /// # Returns
    /// * `Result<Vec<DeviceStruct>, RedisError>` - List of devices or a Redis error
    pub async fn get_devices(self, account_id: &str) -> Result<Vec<DeviceStruct>, RedisError> {
        // Get from Redis and update local cache
        let devices = get_devices(self.remote_cache.connection.clone(), account_id).await?;

        // Cache each device locally for future use
        for device in &devices {
            let cache_key = format!("{}:{}", account_id, device.id);
            self.local_cache.devices.insert(cache_key, device.clone());
        }

        Ok(devices)
    }

    /// Retrieves a specific device for an account.
    ///
    /// # Arguments
    /// * `account_id` - Account ID
    /// * `device_id` - Device ID to retrieve
    ///
    /// # Returns
    /// * `Result<Option<DeviceStruct>, RedisError>` - The device if found, None if not found, or a Redis error
    pub async fn get_device(
        self,
        account_id: &str,
        device_id: &str,
    ) -> Result<Option<DeviceStruct>, RedisError> {
        // Try local cache first
        let cache_key = format!("{}:{}", account_id, device_id);
        if let Some(device) = self.local_cache.devices.get(&cache_key) {
            return Ok(Some(device));
        }

        // If not in local cache, try Redis
        match get_device(self.remote_cache.connection.clone(), account_id, device_id).await? {
            Some(device) => {
                // Cache for future use
                self.local_cache.devices.insert(cache_key, device.clone());
                Ok(Some(device))
            }
            None => Ok(None),
        }
    }

    /// Retrieves all trunks for an account.
    ///
    /// # Arguments
    /// * `account_id` - Account ID
    ///
    /// # Returns
    /// * `Result<Vec<Trunk>, RedisError>` - List of trunks or a Redis error
    pub async fn get_trunks(self, account_id: &str) -> Result<Vec<Trunk>, RedisError> {
        // Get from Redis and update local cache
        let trunks = get_trunks(self.remote_cache.connection.clone(), account_id).await?;

        // Cache each trunk locally for future use
        for trunk in &trunks {
            let cache_key = format!("{}:{}", account_id, trunk.id);
            self.local_cache.trunks.insert(cache_key, trunk.clone());
        }

        Ok(trunks)
    }

    /// Retrieves a specific trunk for an account.
    ///
    /// # Arguments
    /// * `account_id` - Account ID
    /// * `trunk_id` - Trunk ID to retrieve
    ///
    /// # Returns
    /// * `Result<Option<Trunk>, RedisError>` - The trunk if found, None if not found, or a Redis error
    pub async fn get_trunk(
        self,
        account_id: &str,
        trunk_id: &str,
    ) -> Result<Option<Trunk>, RedisError> {
        // Try local cache first
        let cache_key = format!("{}:{}", account_id, trunk_id);
        if let Some(trunk) = self.local_cache.trunks.get(&cache_key) {
            return Ok(Some(trunk));
        }

        // If not in local cache, try Redis
        match get_trunk(self.remote_cache.connection.clone(), account_id, trunk_id).await? {
            Some(trunk) => {
                // Cache for future use
                self.local_cache.trunks.insert(cache_key, trunk.clone());
                Ok(Some(trunk))
            }
            None => Ok(None),
        }
    }

    /// Retrieves all DDIs for an account.
    ///
    /// # Arguments
    /// * `account_id` - Account ID
    ///
    /// # Returns
    /// * `Result<Vec<DDI>, RedisError>` - List of DDIs or a Redis error
    pub async fn get_ddis(self, account_id: &str) -> Result<Vec<DDI>, RedisError> {
        // Get from Redis and update local cache
        let ddis = get_ddis(self.remote_cache.connection.clone(), account_id).await?;

        // Cache each DDI locally for future use
        for ddi in &ddis {
            let cache_key = format!("{}:{}", account_id, ddi.id);
            self.local_cache.ddis.insert(cache_key, ddi.clone());
        }

        Ok(ddis)
    }

    /// Retrieves a specific DDI for an account.
    ///
    /// # Arguments
    /// * `account_id` - Account ID
    /// * `ddi_id` - DDI ID to retrieve
    ///
    /// # Returns
    /// * `Result<Option<DDI>, RedisError>` - The DDI if found, None if not found, or a Redis error
    pub async fn get_ddi(self, account_id: &str, ddi_id: &str) -> Result<Option<DDI>, RedisError> {
        // Try local cache first
        let cache_key = format!("{}:{}", account_id, ddi_id);
        if let Some(ddi) = self.local_cache.ddis.get(&cache_key) {
            return Ok(Some(ddi));
        }

        // If not in local cache, try Redis
        match get_ddi(self.remote_cache.connection.clone(), account_id, ddi_id).await? {
            Some(ddi) => {
                // Cache for future use
                self.local_cache.ddis.insert(cache_key, ddi.clone());
                Ok(Some(ddi))
            }
            None => Ok(None),
        }
    }

    /// Retrieves all hooks for an account.
    ///
    /// # Arguments
    /// * `account_id` - Account ID
    ///
    /// # Returns
    /// * `Result<Vec<Hook>, RedisError>` - List of hooks or a Redis error
    pub async fn get_hooks(self, account_id: &str) -> Result<Vec<Hook>, RedisError> {
        // Get from Redis and update local cache
        let hooks = get_hooks(self.remote_cache.connection.clone(), account_id).await?;

        // Cache each hook locally for future use
        for hook in &hooks {
            let cache_key = format!("{}:{}", account_id, hook.id);
            self.local_cache.hooks.insert(cache_key, hook.clone());
        }

        Ok(hooks)
    }

    /// Retrieves a specific hook for an account.
    ///
    /// # Arguments
    /// * `account_id` - Account ID
    /// * `hook_id` - Hook ID to retrieve
    ///
    /// # Returns
    /// * `Result<Option<Hook>, RedisError>` - The hook if found, None if not found, or a Redis error
    pub async fn get_hook(
        self,
        account_id: &str,
        hook_id: &str,
    ) -> Result<Option<Hook>, RedisError> {
        // Try local cache first
        let cache_key = format!("{}:{}", account_id, hook_id);
        if let Some(hook) = self.local_cache.hooks.get(&cache_key) {
            return Ok(Some(hook));
        }

        // If not in local cache, try Redis
        match get_hook(self.remote_cache.connection.clone(), account_id, hook_id).await? {
            Some(hook) => {
                // Cache for future use
                self.local_cache.hooks.insert(cache_key, hook.clone());
                Ok(Some(hook))
            }
            None => Ok(None),
        }
    }

    /// Retrieves all assets for an account.
    ///
    /// # Arguments
    /// * `account_id` - Account ID
    ///
    /// # Returns
    /// * `Result<Vec<Asset>, RedisError>` - List of assets or a Redis error
    pub async fn get_assets(self, account_id: &str) -> Result<Vec<Asset>, RedisError> {
        // Get from Redis and update local cache
        let assets = get_assets(self.remote_cache.connection.clone(), account_id).await?;

        // Cache each asset locally for future use
        for asset in &assets {
            let cache_key = format!("{}:{}", account_id, asset.id);
            self.local_cache.assets.insert(cache_key, asset.clone());
        }

        Ok(assets)
    }

    /// Retrieves a specific asset for an account.
    ///
    /// # Arguments
    /// * `account_id` - Account ID
    /// * `asset_id` - Asset ID to retrieve
    ///
    /// # Returns
    /// * `Result<Option<Asset>, RedisError>` - The asset if found, None if not found, or a Redis error
    pub async fn get_asset(
        self,
        account_id: &str,
        asset_id: &str,
    ) -> Result<Option<Asset>, RedisError> {
        // Try local cache first
        let cache_key = format!("{}:{}", account_id, asset_id);
        if let Some(asset) = self.local_cache.assets.get(&cache_key) {
            return Ok(Some(asset));
        }

        // If not in local cache, try Redis
        match get_asset(self.remote_cache.connection.clone(), account_id, asset_id).await? {
            Some(asset) => {
                // Cache for future use
                self.local_cache.assets.insert(cache_key, asset.clone());
                Ok(Some(asset))
            }
            None => Ok(None),
        }
    }

    /// Retrieves a proxy by its ID, first checking the local cache and then Redis.
    ///
    /// # Arguments
    /// * `id` - Proxy ID to look up
    ///
    /// # Returns
    /// * `Result<Option<Proxy>, RedisError>` - Proxy if found, None if not found, or a Redis error
    pub async fn get_proxy_by_id(&mut self, id: &str) -> Result<Option<Proxy>, RedisError> {
        if let Some(proxy) = self.local_cache.proxies.get(id) {
            Ok(Some(proxy))
        } else {
            // If proxy identifier not in local cache, try Redis
            match get_proxy_by_id(self.remote_cache.connection.clone(), id).await? {
                Some(proxy) => {
                    self.local_cache
                        .proxies
                        .insert(id.to_string(), proxy.clone());
                    Ok(Some(proxy))
                }
                None => Ok(None),
            }
        }
    }
    
    pub async fn get_proxies(&mut self) -> Result<Vec<Proxy>, RedisError> {
        let proxies = get_proxies(self.remote_cache.connection.clone()).await?;
        self.local_cache.proxies.invalidate_all();
        for proxy in &proxies {
            self.local_cache.proxies.insert(proxy.id.clone(), proxy.clone());
        }
        Ok(proxies)
    }
}

/// Gets the Redis host from the environment variable.
///
/// # Returns
/// * `String` - Redis host URL
///
/// # Panics
/// Panics if the environment variable is not set
fn get_redis_host() -> String {
    let name = "CAL_VALKEY_HOST";
    env::var(name).unwrap_or_else(|_| panic!("${} is not set", name))
}

/// Creates Redis and local cache instances and sets up the event subscription.
///
/// # Returns
/// * `(RedisCache, LocalCache)` - The created cache instances
///
/// # Panics
/// Panics if Redis connection fails or if subscription fails
pub async fn create_pool() -> (RedisCache, LocalCache) {
    // Set up Redis connection
    let client = redis::Client::open(get_redis_host()).expect("Failed to connect to Redis");

    let (tx, mut rx) = unbounded_channel();
    let config = redis::AsyncConnectionConfig::new().set_push_sender(tx);
    let mut con = client
        .get_multiplexed_async_connection_with_config(&config)
        .await
        .expect("Failed to get Redis connection");

    con.subscribe(&["cal:events"])
        .await
        .expect("Failed to subscribe to events");

    // Create cache configuration function to avoid repetition
    fn create_cache<T: Clone + Send + Sync + 'static>(
        capacity_mb: u64,
        ttl_secs: u64,
    ) -> Cache<String, T> {
        Cache::builder()
            .max_capacity(capacity_mb * 1024 * 1024)
            .time_to_live(Duration::from_secs(ttl_secs))
            .build()
    }

    // Initialize all caches with appropriate sizes
    let local_cache = LocalCache {
        flow_state: create_cache::<FlowState>(50, 60 * 60 * 4),
        regions: create_cache::<Region>(5, 60),
        region_idents: create_cache::<String>(5, 60),
        accounts: create_cache::<AccountLite>(50, 60),
        account_idents: create_cache::<String>(5, 60),
        devices: create_cache::<DeviceStruct>(20, 60),
        trunks: create_cache::<Trunk>(10, 60),
        ddis: create_cache::<DDI>(10, 60),
        hooks: create_cache::<Hook>(10, 60),
        assets: create_cache::<Asset>(10, 60),
        trunk_ddi_idents: create_cache::<String>(10, 60),
        proxies: create_cache::<Proxy>(10, 60 * 60),
    };

    let local_cache_cloned = local_cache.clone();

    // Spawn event listener task
    tokio::spawn(async move {
        while let Some(push_info) = rx.recv().await {
            if push_info.kind != PushKind::Message {
                continue;
            }

            if let Some(Value::BulkString(bytes)) = push_info.data.get(1) {
                if let Ok(event_str) = String::from_utf8(bytes.clone()) {
                    if let Ok(event) = serde_json::from_str::<RedisEvent>(&event_str) {
                        handle_redis_event(event, &local_cache_cloned);
                    }
                }
            }
        }
    });

    (RedisCache { connection: con }, local_cache)
}

/// Handles Redis pubsub events by updating the local cache appropriately.
///
/// # Arguments
/// * `event` - The Redis event to handle
/// * `cache` - The local cache to update
fn handle_redis_event(event: RedisEvent, cache: &LocalCache) {
    match event {
        RedisEvent::AccountUpdate(update) => {
            build_account_identifiers_locally(cache.clone(), &update.payload.into());
        }
        RedisEvent::AccountDelete(delete) => {
            cache.accounts.invalidate(&delete.id);
        }
        RedisEvent::AccountCreate(create) => {
            build_account_identifiers_locally(cache.clone(), &create.payload.into());
        }
        RedisEvent::AccountSync => {
            cache.accounts.invalidate_all();
            cache.account_idents.invalidate_all();
        }
        RedisEvent::RegionUpdate(update) => {
            cache
                .regions
                .insert(update.payload.id.to_string(), update.payload);
        }
        RedisEvent::RegionDelete(delete) => {
            cache.regions.invalidate(&delete.id);
        }
        RedisEvent::RegionCreate(create) => {
            build_region_identifiers_locally(cache.clone(), &create.payload);
        }
        RedisEvent::RegionSync => {
            cache.regions.invalidate_all();
            cache.region_idents.invalidate_all();
        }
    }
}

/// Retrieves a string value directly from Redis.
///
/// # Arguments
/// * `con` - Redis connection
/// * `key` - Redis key
///
/// # Returns
/// * `Result<Option<String>, RedisError>` - String value if found, None if not found, or a Redis error
pub async fn get_str(
    mut con: MultiplexedConnection,
    key: &str,
) -> Result<Option<String>, RedisError> {
    con.get(key).await
}

/// Retrieves a field from a Redis hash directly.
///
/// # Arguments
/// * `con` - Redis connection
/// * `key` - Redis hash key
/// * `field` - Field to retrieve
///
/// # Returns
/// * `Result<Option<String>, RedisError>` - Field value if found, None if not found, or a Redis error
pub async fn get_hash(
    mut con: MultiplexedConnection,
    key: &str,
    field: &str,
) -> Result<Option<String>, RedisError> {
    con.hget(key, field).await
}

/// Retrieves all values from a Redis hash.
///
/// # Arguments
/// * `con` - Redis connection
/// * `key` - Redis hash key
///
/// # Returns
/// * `Result<HashMap<String, String>, RedisError>` - Map of field names to values, or a Redis error
pub async fn get_hash_all_values(
    mut con: MultiplexedConnection,
    key: &str,
) -> Result<HashMap<String, String>, RedisError> {
    con.hvals(key).await
}