ActivePush/src/app.rs

//! Shared application state.

use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::atomic::{AtomicI64, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Instant;

use arc_swap::ArcSwap;
use rand::RngCore;

use crate::backend::dav::DavClient;
use crate::backend::{ImapPool, Session};
use crate::config::Settings;
use crate::db::{self, Db};
use crate::logging::LogRing;
use crate::metrics::{AuthGuard, Metrics};

#[derive(Debug, Clone)]
pub struct Boot {
    pub data_dir: PathBuf,
    pub eas_listen: SocketAddr,
    pub admin_listen: SocketAddr,
    pub allow_remote_admin: bool,
    pub admin_tls: Option<(String, String)>,
    pub eas_tls: Option<(String, String)>,
}

pub struct App {
    pub db: Db,
    pub boot: Boot,
    settings: ArcSwap<Settings>,
    pub settings_rev: AtomicI64,
    pub metrics: Metrics,
    pub auth: AuthGuard,
    pub pool: Arc<ImapPool>,
    pub logs: Arc<LogRing>,
    pub started: Instant,
    pub started_ts: i64,
    http: Mutex<(i64, Option<reqwest::Client>, Option<reqwest::Client>)>,
    /// Random per-process key for HMACs (credential cache keys).
    pub secret: [u8; 32],
    loops: Mutex<std::collections::HashMap<String, (u32, i64)>>,
    pub shutdown: tokio_util::sync::CancellationToken,
    /// Collection ids a device still uses but that do not exist here (e.g. left over from
    /// Z-Push). They are sent as FolderHierarchy Delete on the next FolderSync.
    stale: Mutex<std::collections::HashMap<i64, std::collections::BTreeSet<String>>>,
    /// Last Autodiscover requests with headers (no secrets) and bodies, for the panel.
    ad_trace: Mutex<std::collections::VecDeque<serde_json::Value>>,
}

impl App {
    pub fn ad_trace_push(&self, v: serde_json::Value) {
        let mut t = self.ad_trace.lock().unwrap();
        if t.len() >= 30 {
            t.pop_front();
        }
        t.push_back(v);
    }

    pub fn ad_trace_list(&self) -> Vec<serde_json::Value> {
        self.ad_trace
            .lock()
            .unwrap()
            .iter()
            .rev()
            .cloned()
            .collect()
    }

    pub async fn new(boot: Boot, logs: Arc<LogRing>) -> anyhow::Result<Arc<App>> {
        let db = db::open(&boot.data_dir).await?;
        let (settings, rev) = db::load_settings(&db).await?;
        if let Err(e) = settings.validate() {
            tracing::warn!("stored settings do not validate ({e}); running with them anyway");
        }
        logs.set_cap(settings.logging.ring_size);
        let mut secret = [0u8; 32];
        rand::rngs::OsRng.fill_bytes(&mut secret);
        Ok(Arc::new(App {
            db,
            boot,
            settings: ArcSwap::from_pointee(settings),
            settings_rev: AtomicI64::new(rev),
            metrics: Metrics::new(),
            auth: AuthGuard::default(),
            pool: Arc::new(ImapPool::new(secret)),
            logs,
            started: Instant::now(),
            started_ts: crate::util::now(),
            http: Mutex::new((-1, None, None)),
            secret,
            loops: Mutex::new(std::collections::HashMap::new()),
            shutdown: tokio_util::sync::CancellationToken::new(),
            stale: Mutex::new(std::collections::HashMap::new()),
            ad_trace: Mutex::new(std::collections::VecDeque::new()),
        }))
    }

    pub fn settings(&self) -> Arc<Settings> {
        self.settings.load_full()
    }

    pub async fn update_settings(&self, s: Settings) -> anyhow::Result<i64> {
        let rev = db::save_settings(&self.db, &s).await?;
        self.logs.set_cap(s.logging.ring_size);
        self.settings.store(Arc::new(s));
        self.settings_rev.store(rev, Ordering::SeqCst);
        // backend endpoints may have changed
        self.pool.clear();
        self.auth.clear_cache();
        Ok(rev)
    }

    fn http_clients(&self, s: &Settings) -> (Option<reqwest::Client>, Option<reqwest::Client>) {
        let rev = self.settings_rev.load(Ordering::SeqCst);
        let mut g = self.http.lock().unwrap();
        if g.0 != rev {
            let cal = s
                .caldav
                .enabled
                .then(|| DavClient::build_http(s.caldav.verify_tls, s.caldav.timeout_secs).ok())
                .flatten();
            let card = s
                .carddav
                .enabled
                .then(|| DavClient::build_http(s.carddav.verify_tls, s.carddav.timeout_secs).ok())
                .flatten();
            *g = (rev, cal, card);
        }
        (g.1.clone(), g.2.clone())
    }

    pub fn stale_add(&self, device: i64, collection: &str) {
        if collection.is_empty() || collection.len() > 64 {
            return;
        }
        let mut g = self.stale.lock().unwrap();
        let set = g.entry(device).or_default();
        if set.len() < 500 {
            set.insert(collection.to_string());
        }
    }

    pub fn stale_take(&self, device: i64) -> Vec<String> {
        self.stale
            .lock()
            .unwrap()
            .remove(&device)
            .map(|s| s.into_iter().collect())
            .unwrap_or_default()
    }

    /// Count how often a (device, collection, old sync key) was retried within the last hour.
    pub fn loop_hit(&self, key: &str) -> u32 {
        let now = crate::util::now();
        let mut g = self.loops.lock().unwrap();
        g.retain(|_, (_, t)| *t > now - 3600);
        let e = g.entry(key.to_string()).or_insert((0, now));
        e.0 += 1;
        e.1 = now;
        e.0
    }

    pub fn session(&self, login: &str, password: &str) -> Session {
        let s = self.settings();
        let (cal, card) = self.http_clients(&s);
        Session::new(s, self.pool.clone(), cal, card, login, password)
    }
}