hypercall_api/upstash/

sources.rs

use anyhow::Result;
use async_trait::async_trait;
use axum::body::Bytes;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant, SystemTime};

use super::codec::{compress_snapshot_payload, version_upstash_key};
use super::publisher::{system_time_to_millis, UpstashSnapshotSource};
use crate::caches::MarqueeMoversCache;

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

fn parse_key_env(env_name: &str, default: &str) -> Result<String> {
    let raw = match std::env::var(env_name) {
        Ok(v) if v.trim().is_empty() => {
            return Err(anyhow::anyhow!(
                "{env_name} is set but empty; either provide a valid key or unset it"
            ));
        }
        Ok(v) => v,
        Err(_) => default.to_string(),
    };
    version_upstash_key(&raw)
        .map_err(|e| anyhow::anyhow!("{env_name} must resolve to a non-empty value: {e}"))
}

fn parse_ttl_env(env_name: &str, default: u64) -> u64 {
    match std::env::var(env_name).ok().map(|v| v.trim().to_string()) {
        Some(v) if !v.is_empty() => match v.parse::<u64>() {
            Ok(n) if n > 0 => n,
            Ok(_) => {
                tracing::warn!("{env_name}=0 is invalid, using default {default}");
                default
            }
            Err(e) => {
                tracing::warn!(
                    "{env_name}={v:?} is not a valid u64 ({e}), using default {default}"
                );
                default
            }
        },
        _ => default,
    }
}

// ===========================================================================
// StandardSnapshotSource — one struct for all sync snapshot caches
// ===========================================================================

pub struct StandardSnapshotSource {
    name: &'static str,
    redis_key: String,
    ttl_seconds: u64,
    last_built_at_ms: Mutex<Option<u64>>,
    get_snapshot: Box<dyn Fn() -> (Bytes, SystemTime) + Send + Sync>,
}

impl StandardSnapshotSource {
    pub fn new(
        name: &'static str,
        key_env: &str,
        key_default: &str,
        ttl_env: &str,
        ttl_default: u64,
        get_snapshot: Box<dyn Fn() -> (Bytes, SystemTime) + Send + Sync>,
    ) -> Result<Self> {
        Ok(Self {
            name,
            redis_key: parse_key_env(key_env, key_default)?,
            ttl_seconds: parse_ttl_env(ttl_env, ttl_default),
            last_built_at_ms: Mutex::new(None),
            get_snapshot,
        })
    }
}

#[async_trait]
impl UpstashSnapshotSource for StandardSnapshotSource {
    fn name(&self) -> &'static str {
        self.name
    }
    fn key(&self) -> &str {
        &self.redis_key
    }
    fn ttl_seconds(&self) -> u64 {
        self.ttl_seconds
    }

    async fn next_payload(&self) -> Result<Option<(Vec<u8>, u64)>> {
        let (payload, built_at) = (self.get_snapshot)();
        let built_at_ms = system_time_to_millis(built_at)?;
        {
            let last = self.last_built_at_ms.lock().unwrap();
            if *last == Some(built_at_ms) {
                return Ok(None);
            }
        }
        let compressed = compress_snapshot_payload(payload.as_ref())?;
        Ok(Some((compressed, built_at_ms)))
    }

    fn on_success(&self, built_at_ms: u64, elapsed: Duration) {
        {
            let mut last = self.last_built_at_ms.lock().unwrap();
            *last = Some(built_at_ms);
        }
        let n = self.name;
        metrics::counter!(format!("ht_{n}_upstash_publish_total"), "status" => "success")
            .increment(1);
        metrics::histogram!(format!("ht_{n}_upstash_publish_seconds"))
            .record(elapsed.as_secs_f64());
        metrics::gauge!(format!("ht_{n}_upstash_last_success_timestamp"))
            .set(built_at_ms as f64 / 1000.0);
    }

    fn on_skip(&self) {
        let n = self.name;
        metrics::counter!(format!("ht_{n}_upstash_publish_total"), "status" => "skipped_unchanged")
            .increment(1);
    }

    fn on_error(&self) {
        let n = self.name;
        metrics::counter!(format!("ht_{n}_upstash_publish_total"), "status" => "error")
            .increment(1);
    }
}

// ===========================================================================
// MarqueeMoversSource — async refresh + rate limiting
// ===========================================================================

const MARQUEE_MOVERS_INTERVAL: Duration = Duration::from_secs(15);

pub struct MarqueeMoversSource {
    cache: Arc<MarqueeMoversCache>,
    redis_key: String,
    ttl_seconds: u64,
    last_built_at_ms: Mutex<Option<u64>>,
    last_refresh: Mutex<Option<Instant>>,
}

impl MarqueeMoversSource {
    pub fn new(cache: Arc<MarqueeMoversCache>) -> Result<Self> {
        Ok(Self {
            cache,
            redis_key: parse_key_env("MARQUEE_MOVERS_UPSTASH_KEY", "marquee:movers:latest")?,
            ttl_seconds: parse_ttl_env("MARQUEE_MOVERS_UPSTASH_TTL_SECONDS", 60),
            last_built_at_ms: Mutex::new(None),
            last_refresh: Mutex::new(None),
        })
    }
}

#[async_trait]
impl UpstashSnapshotSource for MarqueeMoversSource {
    fn name(&self) -> &'static str {
        "marquee_movers"
    }
    fn key(&self) -> &str {
        &self.redis_key
    }
    fn ttl_seconds(&self) -> u64 {
        self.ttl_seconds
    }

    async fn next_payload(&self) -> Result<Option<(Vec<u8>, u64)>> {
        {
            let mut last = self.last_refresh.lock().unwrap();
            if let Some(t) = *last {
                if t.elapsed() < MARQUEE_MOVERS_INTERVAL {
                    return Ok(None);
                }
            }
            *last = Some(Instant::now());
        }

        self.cache.refresh().await;

        let (payload, built_at) = self.cache.published_response().await;
        let built_at_ms = system_time_to_millis(built_at)?;
        {
            let last = self.last_built_at_ms.lock().unwrap();
            if *last == Some(built_at_ms) {
                return Ok(None);
            }
        }
        let compressed = compress_snapshot_payload(payload.as_ref())?;
        Ok(Some((compressed, built_at_ms)))
    }

    fn on_success(&self, built_at_ms: u64, elapsed: Duration) {
        {
            let mut last = self.last_built_at_ms.lock().unwrap();
            *last = Some(built_at_ms);
        }
        metrics::counter!("ht_marquee_movers_upstash_publish_total", "status" => "success")
            .increment(1);
        metrics::histogram!("ht_marquee_movers_upstash_publish_seconds")
            .record(elapsed.as_secs_f64());
        metrics::gauge!("ht_marquee_movers_upstash_last_success_timestamp")
            .set(built_at_ms as f64 / 1000.0);
    }

    fn on_skip(&self) {
        metrics::counter!("ht_marquee_movers_upstash_publish_total", "status" => "skipped_unchanged")
            .increment(1);
    }

    fn on_error(&self) {
        metrics::counter!("ht_marquee_movers_upstash_publish_total", "status" => "error")
            .increment(1);
    }

    fn min_interval(&self) -> Option<Duration> {
        Some(MARQUEE_MOVERS_INTERVAL)
    }
}