pharos_native/runtime/

cache.rs

//! Runtime-owned Redis cache orchestration.

use std::path::Path;
use std::sync::{OnceLock, RwLock};
use std::time::{SystemTime, UNIX_EPOCH};

use crate::config::{self, ServeConfig};

use super::cache_key;

#[cfg(feature = "redis")]
use pharos_runtime_integrations::redis::{
    RedisConnection, RedisConnectionFactory, RedisError, RedisRuntimeConfig, RedisStatus,
};
#[cfg(feature = "redis")]
use std::cell::RefCell;

#[derive(Debug, Clone, Copy, Default, Eq, PartialEq)]
pub struct CachedSessionRecord {
    pub user_id: i64,
    pub expires_at_epoch: u64,
}

#[derive(Debug, Clone, Default)]
struct RuntimeCacheState {
    enabled: bool,
    cache_sessions_enabled: bool,
    cache_template_source_enabled: bool,
    cache_route_match_enabled: bool,
    cache_config_allowlist_enabled: bool,
    cache_prefix: String,
    status: String,
    #[cfg(feature = "redis")]
    next_connect_after_epoch: u64,
    #[cfg(feature = "redis")]
    current_backoff_seconds: u64,
    route_match_generation: u64,
    #[cfg(feature = "redis")]
    factory: Option<RedisConnectionFactory>,
}

static RUNTIME_CACHE_STATE: OnceLock<RwLock<RuntimeCacheState>> = OnceLock::new();

#[cfg(feature = "redis")]
thread_local! {
    static REDIS_CONNECTION: RefCell<Option<RedisConnection>> = const { RefCell::new(None) };
}

pub fn install_runtime_cache(config_record: &ServeConfig) {
    let state_lock = RUNTIME_CACHE_STATE.get_or_init(|| RwLock::new(RuntimeCacheState::default()));
    let mut state = match state_lock.write() {
        Ok(state) => state,
        Err(_) => return,
    };

    #[cfg(not(feature = "redis"))]
    {
        let _ = config_record;
        *state = RuntimeCacheState {
            enabled: false,
            cache_sessions_enabled: false,
            cache_template_source_enabled: false,
            cache_route_match_enabled: false,
            cache_config_allowlist_enabled: false,
            cache_prefix: "pharos".to_owned(),
            status: "disabled".to_owned(),
            route_match_generation: 0,
        };
    }

    #[cfg(feature = "redis")]
    {
        let runtime_config = runtime_config_from_serve_config(config_record);
        let factory = RedisConnectionFactory::new(runtime_config.clone()).ok();
        *state = RuntimeCacheState {
            enabled: runtime_config.enabled && factory.is_some(),
            cache_sessions_enabled: runtime_config.enabled && runtime_config.cache_sessions_enabled,
            cache_template_source_enabled: runtime_config.enabled
                && runtime_config.cache_template_source_enabled,
            cache_route_match_enabled: runtime_config.enabled
                && runtime_config.cache_route_match_enabled,
            cache_config_allowlist_enabled: runtime_config.enabled
                && runtime_config.cache_config_allowlist_enabled,
            cache_prefix: runtime_config.cache_prefix.clone(),
            status: if runtime_config.enabled && factory.is_some() {
                RedisStatus::Connecting.as_str().to_owned()
            } else {
                RedisStatus::Disabled.as_str().to_owned()
            },
            next_connect_after_epoch: 0,
            current_backoff_seconds: runtime_config.backoff_seconds.max(1),
            route_match_generation: 0,
            factory,
        };
        REDIS_CONNECTION.with(|cell| {
            cell.borrow_mut().take();
        });
    }
}

pub fn redis_status() -> String {
    cache_state_read(|state| state.status.clone()).unwrap_or_else(|| "disabled".to_owned())
}

pub fn probe_runtime_cache() {
    #[cfg(feature = "redis")]
    {
        let _ = with_connection(|_, _| Ok(()));
    }
}

pub fn session_cache_lookup(
    app_id: &str,
    session_token_hash: &str,
    now_epoch_seconds: u64,
) -> Option<CachedSessionRecord> {
    if !session_cache_enabled() {
        return None;
    }
    let key = prefix_runtime_key(cache_key::session_value(app_id, session_token_hash)?)?;
    let text = get_value(&key)?;
    let record = decode_session_record(&text)?;
    (record.user_id > 0 && record.expires_at_epoch > now_epoch_seconds).then_some(record)
}

pub fn session_cache_store(
    app_id: &str,
    user_id: i64,
    session_token_hash: &str,
    expires_at_epoch: u64,
) {
    if !session_cache_enabled() || user_id <= 0 || expires_at_epoch <= now_epoch_seconds() {
        return;
    }
    let Some(key) = cache_key::session_value(app_id, session_token_hash) else {
        return;
    };
    let Some(index_key) = cache_key::session_user_index(app_id, user_id) else {
        return;
    };
    let Some(key) = prefix_runtime_key(key) else {
        return;
    };
    let Some(index_key) = prefix_runtime_key(index_key) else {
        return;
    };
    let ttl = expires_at_epoch.saturating_sub(now_epoch_seconds()).max(1);
    let value = encode_session_record(CachedSessionRecord {
        user_id,
        expires_at_epoch,
    });
    let _ = set_value(&key, &value, ttl);
    let _ = set_add(&index_key, &key, ttl);
}

pub fn session_cache_revoke_token(app_id: &str, session_token_hash: &str) {
    if !session_cache_enabled() {
        return;
    }
    let Some(key) = cache_key::session_value(app_id, session_token_hash) else {
        return;
    };
    let Some(key) = prefix_runtime_key(key) else {
        return;
    };
    let _ = delete_value(&key);
}

pub fn session_cache_revoke_user(app_id: &str, user_id: i64) {
    if !session_cache_enabled() || user_id <= 0 {
        return;
    }
    let Some(index_key) = cache_key::session_user_index(app_id, user_id) else {
        return;
    };
    let Some(index_key) = prefix_runtime_key(index_key) else {
        return;
    };
    let keys = set_members(&index_key);
    if !keys.is_empty() {
        let borrowed = keys.iter().map(String::as_str).collect::<Vec<_>>();
        let _ = delete_many(&borrowed);
    }
    let _ = delete_value(&index_key);
}

pub fn route_match_lookup(path: &str) -> Option<String> {
    if !route_match_cache_enabled() {
        return None;
    }
    let generation = cache_state_read(|state| state.route_match_generation).unwrap_or(0);
    let key = prefix_runtime_key(cache_key::route_match(&format!("{generation}:{path}"))?)?;
    get_value(&key)
}

pub fn route_match_store(path: &str, app_id: &str) {
    if !route_match_cache_enabled() || app_id.is_empty() {
        return;
    }
    let generation = cache_state_read(|state| state.route_match_generation).unwrap_or(0);
    let Some(key) = cache_key::route_match(&format!("{generation}:{path}")) else {
        return;
    };
    let Some(key) = prefix_runtime_key(key) else {
        return;
    };
    let _ = set_value(&key, app_id, 300);
}

pub fn invalidate_route_match_namespace() {
    let Some(state) = RUNTIME_CACHE_STATE.get() else {
        return;
    };
    if let Ok(mut state) = state.write() {
        state.route_match_generation = state.route_match_generation.saturating_add(1);
    }
}

pub fn template_source_lookup(app_root: &Path, relative_path: &str) -> Option<String> {
    if !template_source_cache_enabled() {
        return None;
    }
    let app_root = app_root.to_string_lossy();
    let key = prefix_runtime_key(cache_key::template_source(&app_root, relative_path)?)?;
    get_value(&key)
}

pub fn template_source_store(app_root: &Path, relative_path: &str, template_text: &str) {
    if !template_source_cache_enabled() || template_text.is_empty() {
        return;
    }
    let app_root = app_root.to_string_lossy();
    let Some(key) = cache_key::template_source(&app_root, relative_path) else {
        return;
    };
    let Some(key) = prefix_runtime_key(key) else {
        return;
    };
    let _ = set_value(&key, template_text, 60);
}

pub fn dynamic_config_allowlist_value(
    app_root: &Path,
    runtime_conf_path: Option<&Path>,
    key: &str,
) -> Option<String> {
    if key.is_empty() {
        return None;
    }
    let runtime_conf_str = runtime_conf_path.map(|path| path.to_string_lossy().into_owned());
    if config_allowlist_cache_enabled() && config_key_allowlisted(key) {
        let app_root_str = app_root.to_string_lossy();
        let cache_key = prefix_runtime_key(cache_key::config_allowlist(
            &app_root_str,
            runtime_conf_str.as_deref(),
            key,
        )?)?;
        if let Some(value) = get_value(&cache_key) {
            return (!value.is_empty()).then_some(value);
        }
        let value = config::dynamic_config_effective_value(app_root, runtime_conf_path, key)?;
        let _ = set_value(&cache_key, &value, 30);
        return Some(value);
    }
    config::dynamic_config_effective_value(app_root, runtime_conf_path, key)
}

fn session_cache_enabled() -> bool {
    cache_state_read(|state| state.enabled && state.cache_sessions_enabled).unwrap_or(false)
}

fn template_source_cache_enabled() -> bool {
    cache_state_read(|state| state.enabled && state.cache_template_source_enabled).unwrap_or(false)
}

fn route_match_cache_enabled() -> bool {
    cache_state_read(|state| state.enabled && state.cache_route_match_enabled).unwrap_or(false)
}

fn config_allowlist_cache_enabled() -> bool {
    cache_state_read(|state| state.enabled && state.cache_config_allowlist_enabled).unwrap_or(false)
}

fn config_key_allowlisted(key: &str) -> bool {
    matches!(
        key,
        "environment"
            | "base_path"
            | "base_url"
            | "payment_providers"
            | "stripe_client_id"
            | "stripe_client_secret_env"
            | "stripe_webhook_secret_env"
            | "payment_provider_state_key_env"
            | "payment_provider_secret_key_env"
            | "payment_provider_secret_key_version"
            | "payment_provider_secret_previous_key_version"
            | "payment_provider_secret_previous_key_env"
            | "payment_provider_allow_loopback_http"
            | "payment_provider_allow_plaintext_local"
            | "stripe_oauth_authorize_url"
            | "stripe_oauth_token_url"
            | "stripe_account_api_base_url"
            | "stripe_oauth_deauthorize_url"
    )
}

fn encode_session_record(record: CachedSessionRecord) -> String {
    format!("{}|{}", record.user_id, record.expires_at_epoch)
}

fn decode_session_record(value: &str) -> Option<CachedSessionRecord> {
    let mut fields = value.split('|');
    let user_id = fields.next()?.parse::<i64>().ok()?;
    let expires_at_epoch = fields.next()?.parse::<u64>().ok()?;
    Some(CachedSessionRecord {
        user_id,
        expires_at_epoch,
    })
}

fn now_epoch_seconds() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|value| value.as_secs())
        .unwrap_or(0)
}

fn cache_state_read<T>(reader: impl FnOnce(&RuntimeCacheState) -> T) -> Option<T> {
    let state = RUNTIME_CACHE_STATE.get()?;
    let state = state.read().ok()?;
    Some(reader(&state))
}

fn prefix_runtime_key(key: String) -> Option<String> {
    let prefix = cache_state_read(|state| state.cache_prefix.clone()).unwrap_or_default();
    (!prefix.is_empty() && !key.is_empty()).then(|| format!("{prefix}:{key}"))
}

#[cfg(feature = "redis")]
fn runtime_config_from_serve_config(config_record: &ServeConfig) -> RedisRuntimeConfig {
    RedisRuntimeConfig {
        enabled: config_record.redis_enabled,
        host: config_record
            .redis_host
            .clone()
            .filter(|value| !value.is_empty())
            .unwrap_or_else(|| "127.0.0.1".to_owned()),
        port: config_record
            .redis_port
            .as_deref()
            .and_then(|value| value.parse::<u16>().ok())
            .unwrap_or(6379),
        auth_user: config_record
            .redis_auth_user
            .clone()
            .filter(|value| !value.is_empty())
            .unwrap_or_else(|| "default".to_owned()),
        auth_token_env: config_record
            .redis_auth_token_env
            .clone()
            .unwrap_or_default(),
        db_index: config_record
            .redis_db_index
            .as_deref()
            .and_then(|value| value.parse::<i64>().ok())
            .unwrap_or(0),
        connect_timeout_ms: config_record
            .redis_connect_timeout_ms
            .as_deref()
            .and_then(|value| value.parse::<u64>().ok())
            .unwrap_or(500),
        read_timeout_ms: config_record
            .redis_read_timeout_ms
            .as_deref()
            .and_then(|value| value.parse::<u64>().ok())
            .unwrap_or(300),
        cache_prefix: config_record
            .redis_cache_prefix
            .clone()
            .filter(|value| !value.is_empty())
            .unwrap_or_else(|| "pharos".to_owned()),
        backoff_seconds: config_record
            .redis_backoff_seconds
            .as_deref()
            .and_then(|value| value.parse::<u64>().ok())
            .unwrap_or(30),
        backoff_max_seconds: config_record
            .redis_backoff_max_seconds
            .as_deref()
            .and_then(|value| value.parse::<u64>().ok())
            .unwrap_or(300),
        tls_mode: config_record
            .redis_tls_mode
            .clone()
            .filter(|value| !value.is_empty())
            .unwrap_or_else(|| "disable".to_owned()),
        tls_root_cert: config_record
            .redis_tls_root_cert
            .clone()
            .unwrap_or_default(),
        tls_cert: config_record.redis_tls_cert.clone().unwrap_or_default(),
        tls_key: config_record.redis_tls_key.clone().unwrap_or_default(),
        require_peer_name_match: config_record.redis_require_peer_name_match,
        cache_sessions_enabled: config_record.redis_cache_sessions_enabled,
        cache_template_source_enabled: config_record.redis_cache_template_source_enabled,
        cache_route_match_enabled: config_record.redis_cache_route_match_enabled,
        cache_config_allowlist_enabled: config_record.redis_cache_config_allowlist_enabled,
    }
}

#[cfg(feature = "redis")]
fn with_connection<T>(
    action: impl FnOnce(&RedisConnectionFactory, &mut RedisConnection) -> Result<T, RedisError>,
) -> Option<T> {
    let factory = cache_state_read(|state| {
        if !state.enabled || now_epoch_seconds() < state.next_connect_after_epoch {
            None
        } else {
            state.factory.clone()
        }
    })
    .flatten()?;

    REDIS_CONNECTION.with(|cell| {
        if cell.borrow().is_none() {
            update_status(RedisStatus::Connecting, 0, None);
            let mut connection = factory.connect().ok()?;
            factory.ping(&mut connection).ok()?;
            *cell.borrow_mut() = Some(connection);
            update_status(
                RedisStatus::Active,
                factory.config().backoff_seconds.max(1),
                Some(0),
            );
        }
        let result = {
            let mut borrow = cell.borrow_mut();
            let connection = borrow.as_mut()?;
            action(&factory, connection)
        };
        match result {
            Ok(value) => Some(value),
            Err(_) => {
                cell.borrow_mut().take();
                schedule_backoff(&factory);
                None
            }
        }
    })
}

#[cfg(not(feature = "redis"))]
fn get_value(_key: &str) -> Option<String> {
    None
}

#[cfg(feature = "redis")]
fn get_value(key: &str) -> Option<String> {
    with_connection(|factory, connection| factory.get_text(connection, key)).flatten()
}

#[cfg(not(feature = "redis"))]
fn set_value(_key: &str, _value: &str, _ttl_seconds: u64) -> Option<()> {
    None
}

#[cfg(feature = "redis")]
fn set_value(key: &str, value: &str, ttl_seconds: u64) -> Option<()> {
    with_connection(|factory, connection| {
        factory.set_text_ttl(connection, key, value, ttl_seconds)?;
        Ok(())
    })
}

#[cfg(not(feature = "redis"))]
fn delete_value(_key: &str) -> Option<()> {
    None
}

#[cfg(feature = "redis")]
fn delete_value(key: &str) -> Option<()> {
    with_connection(|factory, connection| {
        factory.delete(connection, key)?;
        Ok(())
    })
}

#[cfg(not(feature = "redis"))]
fn delete_many(_keys: &[&str]) -> Option<()> {
    None
}

#[cfg(feature = "redis")]
fn delete_many(keys: &[&str]) -> Option<()> {
    with_connection(|factory, connection| {
        factory.delete_many(connection, keys.iter().copied())?;
        Ok(())
    })
}

#[cfg(not(feature = "redis"))]
fn set_add(_key: &str, _member: &str, _ttl_seconds: u64) -> Option<()> {
    None
}

#[cfg(feature = "redis")]
fn set_add(key: &str, member: &str, ttl_seconds: u64) -> Option<()> {
    with_connection(|factory, connection| {
        factory.set_add(connection, key, member)?;
        factory.set_expire(connection, key, ttl_seconds)?;
        Ok(())
    })
}

#[cfg(not(feature = "redis"))]
fn set_members(_key: &str) -> Vec<String> {
    Vec::new()
}

#[cfg(feature = "redis")]
fn set_members(key: &str) -> Vec<String> {
    with_connection(|factory, connection| factory.set_members(connection, key)).unwrap_or_default()
}

#[cfg(feature = "redis")]
fn schedule_backoff(factory: &RedisConnectionFactory) {
    let backoff = cache_state_read(|state| state.current_backoff_seconds)
        .unwrap_or(factory.config().backoff_seconds.max(1));
    let next_backoff = backoff
        .saturating_mul(2)
        .min(factory.config().backoff_max_seconds.max(backoff));
    update_status(
        RedisStatus::Backoff,
        next_backoff,
        Some(now_epoch_seconds().saturating_add(backoff)),
    );
}

#[cfg(feature = "redis")]
fn update_status(status: RedisStatus, backoff_seconds: u64, next_connect_after_epoch: Option<u64>) {
    let Some(state) = RUNTIME_CACHE_STATE.get() else {
        return;
    };
    let Ok(mut state) = state.write() else {
        return;
    };
    state.status = status.as_str().to_owned();
    state.current_backoff_seconds = backoff_seconds.max(1);
    if let Some(next_connect_after_epoch) = next_connect_after_epoch {
        state.next_connect_after_epoch = next_connect_after_epoch;
    }
}

#[cfg(test)]
mod tests {
    use super::config_key_allowlisted;

    #[test]
    fn payment_provider_endpoint_config_keys_are_allowlisted_for_runtime_cache() {
        for key in [
            "payment_providers",
            "stripe_client_id",
            "stripe_client_secret_env",
            "stripe_webhook_secret_env",
            "payment_provider_state_key_env",
            "payment_provider_secret_key_env",
            "payment_provider_secret_key_version",
            "payment_provider_secret_previous_key_version",
            "payment_provider_secret_previous_key_env",
            "payment_provider_allow_loopback_http",
            "payment_provider_allow_plaintext_local",
            "stripe_oauth_authorize_url",
            "stripe_oauth_token_url",
            "stripe_account_api_base_url",
            "stripe_oauth_deauthorize_url",
        ] {
            assert!(config_key_allowlisted(key), "{key} must be allowlisted");
        }
    }
}