cortiq-gateway/src/state.rs

//! Shared gateway state accessible across all request handlers.
//!
//! The config-derived part (`Live`) lives behind an `ArcSwap`, which allows
//! hot-reloading the model pool, routing table, and router connection without
//! a restart: the admin API validates the new config, builds a new `Live`,
//! atomically writes the TOML, and swaps the pointer.

use crate::cache::SemanticCache;
use crate::config::Config;
use crate::registry::Registry;
use crate::router_client::{RouterClient, RouterLastStatus};
use crate::routing::RoutingTable;
use crate::secrets::SecretStore;
use crate::stats::Stats;
use arc_swap::ArcSwap;
use std::sync::Arc;

#[derive(Clone)]
pub struct SharedState(pub Arc<AppState>);

/// The portion of state that is rebuilt whenever the config changes.
pub struct Live {
    pub cfg: Config,
    pub registry: Registry,
    pub routing: RoutingTable,
    pub router: RouterClient,
}

impl Live {
    pub fn build(
        cfg: Config,
        secrets: &SecretStore,
        router_status: RouterLastStatus,
    ) -> anyhow::Result<Self> {
        let router = RouterClient::new(&cfg.router, secrets, router_status)?;
        let registry = Registry::from_config(&cfg, secrets)?;
        let routing = RoutingTable::from_config(&cfg);
        Ok(Self {
            cfg,
            registry,
            routing,
            router,
        })
    }

    /// Local-first fallback targets, in order: the managed local CMF model (if
    /// configured), then the routing default, then any registered model — only
    /// registered ids are returned. Used whenever the router is bypassed
    /// (local-only / disabled) or unavailable, so CMF models serve with no router.
    pub fn local_candidates(&self) -> Vec<String> {
        let mut v = Vec::new();
        for s in self.cfg.cmf.effective_servers() {
            v.push(s.id);
        }
        let def = self.routing.default_model().to_string();
        if !def.is_empty() && !v.contains(&def) {
            v.push(def);
        }
        v.retain(|id| self.registry.get(id).is_some());
        if v.is_empty() {
            if let Some(a) = self.registry.any_id() {
                v.push(a);
            }
        }
        v
    }
}

/// Fixed-window per-key rate limiter (minute buckets). Small and honest:
/// at minute boundaries a burst can briefly see up to 2× the limit, which is
/// fine for office-grade fairness and costs one HashMap lookup per request.
#[derive(Default)]
pub struct RateWindows {
    inner: std::sync::Mutex<std::collections::HashMap<String, (u64, u32)>>,
}

impl RateWindows {
    pub fn allow(&self, key: &str, per_min: u32) -> bool {
        let minute = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs()
            / 60;
        let mut g = self.inner.lock().unwrap();
        let e = g.entry(key.to_string()).or_insert((minute, 0));
        if e.0 != minute {
            *e = (minute, 0);
        }
        if e.1 >= per_min {
            return false;
        }
        e.1 += 1;
        true
    }
    /// Seconds until the current minute window rolls over.
    pub fn retry_after_secs(&self) -> u64 {
        60 - (std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_secs()
            % 60)
    }
}

pub struct AppState {
    pub live: ArcSwap<Live>,
    pub stats: Arc<Stats>,
    pub promotion: Arc<crate::promotion::Promotion>,
    pub imports: Arc<crate::import::JobStore>,
    /// Image/video generation jobs (cortiq imagine / animate).
    pub media: Arc<crate::media::MediaStore>,
    /// One-click local model benchmarks (cortiq bench --json).
    pub bench: Arc<crate::bench::BenchStore>,
    /// Host metrics sampler (CPU / RAM / disk) for the dashboard.
    pub sysmon: Arc<crate::sysmon::SysMon>,
    /// Newest cortiq-gateway version on crates.io, when newer than ours.
    pub gateway_update: Arc<std::sync::Mutex<Option<String>>>,
    /// Admin console reachable from other machines without a token — the
    /// footgun banner flag (set at startup, shown by the SPA).
    pub admin_open_public: std::sync::atomic::AtomicBool,
    /// Managed local CMF model server (install/update/spawn lifecycle + status).
    pub cmf: Arc<crate::cmf_runtime::CmfRuntime>,
    pub cache: Arc<SemanticCache>,
    /// Per-API-key fixed-window rate limiter.
    pub rate: RateWindows,
    pub secrets: SecretStore,
    pub config_path: String,
    pub pipeline: crate::pipeline::Pipeline,
    /// Outcome of the most recent router call (survives config reloads) —
    /// lets the admin panel distinguish a bad/expired key from a down router.
    pub router_status: RouterLastStatus,
}

/// Map the user-facing `[shadow]` config to the promotion-table tuning.
fn promotion_cfg(s: &crate::config::ShadowCfg) -> crate::promotion::PromotionCfg {
    crate::promotion::PromotionCfg {
        enabled: s.enabled && !s.local_model_id.is_empty(),
        window: s.window,
        n_min: s.n_min,
        promote_lb: s.promote_lb,
        soak: s.soak,
        file: if s.file.is_empty() {
            None
        } else {
            Some(s.file.clone())
        },
        ..Default::default()
    }
}

impl std::ops::Deref for SharedState {
    type Target = AppState;
    fn deref(&self) -> &Self::Target {
        &self.0
    }
}

impl SharedState {
    pub fn build(cfg: Config, config_path: String) -> anyhow::Result<Self> {
        let secrets_path = sibling(&config_path, "secrets.toml");
        let secrets = SecretStore::load(&secrets_path);
        let stats = Stats::new(&cfg.stats);
        let promotion = crate::promotion::Promotion::new(promotion_cfg(&cfg.shadow));
        let imports = crate::import::JobStore::new();
        let media = crate::media::MediaStore::new();
        let bench = crate::bench::BenchStore::new();
        // jobs survive restarts: files under <data>/, reconciled on load
        let data = crate::config::data_dir();
        let _ = std::fs::create_dir_all(&data);
        imports.attach_persistence(data.join("jobs-import.json"));
        media.attach_persistence(data.join("jobs-media.json"));
        let sysmon = crate::sysmon::spawn(cfg.cmf.models_dir.clone());
        let gateway_update: Arc<std::sync::Mutex<Option<String>>> =
            Arc::new(std::sync::Mutex::new(None));
        {
            // check crates.io for a newer gateway once at start and twice a day
            let slot = gateway_update.clone();
            tokio::spawn(async move {
                loop {
                    if let Some(latest) = crate::cmf_runtime::latest_gateway_version().await {
                        let newer =
                            crate::cmf_runtime::version_lt(env!("CARGO_PKG_VERSION"), &latest);
                        *slot.lock().unwrap() = newer.then_some(latest);
                    }
                    tokio::time::sleep(std::time::Duration::from_secs(12 * 3600)).await;
                }
            });
        }
        let cmf = crate::cmf_runtime::CmfRuntime::new();
        let cache = SemanticCache::new(&cfg.cache);
        let router_status = RouterLastStatus::default();
        let live = Live::build(cfg, &secrets, router_status.clone())?;
        Ok(SharedState(Arc::new(AppState {
            live: ArcSwap::from_pointee(live),
            stats,
            promotion,
            imports,
            media,
            bench,
            sysmon,
            gateway_update,
            admin_open_public: std::sync::atomic::AtomicBool::new(false),
            cmf,
            cache,
            rate: RateWindows::default(),
            secrets,
            config_path,
            pipeline: crate::pipeline::Pipeline::new(),
            router_status,
        })))
    }

    /// Current snapshot of the rebuildable state (safe to hold across await points).
    pub fn live(&self) -> Arc<Live> {
        self.0.live.load_full()
    }
}

impl AppState {
    /// Replace the config entirely: validate → build → write to disk → swap.
    /// If any step fails, the active state and file are left unchanged.
    pub fn reload(&self, new_cfg: Config) -> anyhow::Result<()> {
        new_cfg.validate()?;
        let live = Live::build(new_cfg.clone(), &self.secrets, self.router_status.clone())?;
        new_cfg.save(&self.config_path)?;
        self.live.store(Arc::new(live));

        let cmf_rt = self.cmf.clone();
        let cmf_cfg = new_cfg.cmf.clone();
        tokio::spawn(async move {
            cmf_rt.stop().await;
            crate::cmf_runtime::manage(cmf_rt, cmf_cfg).await;
        });

        Ok(())
    }

    /// Rebuild `Live` from the current config without writing to disk
    /// (e.g. after a secret change — only provider keys changed).
    pub fn rebuild(&self) -> anyhow::Result<()> {
        let cfg = self.live.load_full().cfg.clone();
        let live = Live::build(cfg, &self.secrets, self.router_status.clone())?;
        self.live.store(Arc::new(live));
        Ok(())
    }
}

/// Path to file `name` next to `path` (in the same directory).
fn sibling(path: &str, name: &str) -> String {
    std::path::Path::new(path)
        .parent()
        .map(|d| d.join(name))
        .unwrap_or_else(|| std::path::PathBuf::from(name))
        .to_string_lossy()
        .into_owned()
}