trx-rs/src/trx-server/src/history_store.rs

// SPDX-FileCopyrightText: 2026 Stan Grams <sjg@haxx.space>
//
// SPDX-License-Identifier: GPL-2.0-or-later

//! Persistent decode history storage for trx-server using pickledb.
//!
//! History for all decoder types (AIS, VDES, APRS, CW, FT8, WSPR) is
//! serialised as JSON arrays to `~/.local/cache/trx-rs/history.db` and
//! loaded back on startup, preserving up to 24 hours of decodes across
//! trx-server restarts.  Each rig's keys are prefixed with the rig id
//! (e.g. `"default.ais"`) so multi-rig setups don't collide.

use std::collections::VecDeque;
use std::path::PathBuf;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use pickledb::{PickleDb, PickleDbDumpPolicy, SerializationMethod};
use serde::{de::DeserializeOwned, Deserialize, Serialize};

use trx_core::decode::{
    AisMessage, AprsPacket, CwEvent, Ft8Message, LrptImage, SstvMessage, VdesMessage, WefaxMessage,
    WsprMessage,
};

use crate::audio::DecoderHistories;

const HISTORY_RETENTION_MS: i64 = 24 * 60 * 60 * 1_000;

#[derive(Serialize, Deserialize)]
struct StoredEntry<T> {
    ts_ms: i64,
    data: T,
}

fn now_unix_ms() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis()
        .try_into()
        .unwrap_or(i64::MAX)
}

pub fn db_path() -> PathBuf {
    let base = dirs::cache_dir().unwrap_or_else(|| PathBuf::from("."));
    base.join("trx-rs").join("history.db")
}

/// Open (or create) the history database at the canonical cache path.
pub fn open_db() -> PickleDb {
    let path = db_path();
    if let Some(parent) = path.parent() {
        let _ = std::fs::create_dir_all(parent);
    }
    PickleDb::load(
        &path,
        PickleDbDumpPolicy::DumpUponRequest,
        SerializationMethod::Json,
    )
    .unwrap_or_else(|_| {
        PickleDb::new(
            path,
            PickleDbDumpPolicy::DumpUponRequest,
            SerializationMethod::Json,
        )
    })
}

fn load_key<T: DeserializeOwned>(db: &PickleDb, key: &str) -> Vec<(Instant, T)> {
    let now_ms = now_unix_ms();
    let cutoff_ms = now_ms - HISTORY_RETENTION_MS;

    let entries: Vec<StoredEntry<T>> = db.get(key).unwrap_or_default();

    entries
        .into_iter()
        .filter(|e| e.ts_ms >= cutoff_ms)
        .map(|e| {
            let age_ms = now_ms.saturating_sub(e.ts_ms).max(0) as u64;
            // checked_sub returns None when age exceeds system uptime; treat as
            // brand-new so the entry stays visible for another 24 h.
            let instant = Instant::now()
                .checked_sub(Duration::from_millis(age_ms))
                .unwrap_or_else(Instant::now);
            (instant, e.data)
        })
        .collect()
}

fn save_key<T: Clone + Serialize>(db: &mut PickleDb, key: &str, deque: &VecDeque<(Instant, T)>) {
    let now_ms = now_unix_ms();
    let entries: Vec<StoredEntry<T>> = deque
        .iter()
        .map(|(inst, data)| StoredEntry {
            ts_ms: now_ms - inst.elapsed().as_millis() as i64,
            data: data.clone(),
        })
        .collect();
    let _ = db.set(key, &entries);
}

/// Populate `histories` from the database using `rig_id`-prefixed keys.
pub fn load_all(db: &PickleDb, rig_id: &str, histories: &Arc<DecoderHistories>) {
    let k = |suffix: &str| format!("{}.{}", rig_id, suffix);

    if let Ok(mut h) = histories.ais.lock() {
        for e in load_key::<AisMessage>(db, &k("ais")) {
            h.push_back(e);
        }
    }
    if let Ok(mut h) = histories.vdes.lock() {
        for e in load_key::<VdesMessage>(db, &k("vdes")) {
            h.push_back(e);
        }
    }
    if let Ok(mut h) = histories.aprs.lock() {
        for e in load_key::<AprsPacket>(db, &k("aprs")) {
            h.push_back(e);
        }
    }
    if let Ok(mut h) = histories.hf_aprs.lock() {
        for e in load_key::<AprsPacket>(db, &k("hf_aprs")) {
            h.push_back(e);
        }
    }
    if let Ok(mut h) = histories.cw.lock() {
        for e in load_key::<CwEvent>(db, &k("cw")) {
            h.push_back(e);
        }
    }
    if let Ok(mut h) = histories.ft8.lock() {
        for e in load_key::<Ft8Message>(db, &k("ft8")) {
            h.push_back(e);
        }
    }
    if let Ok(mut h) = histories.ft4.lock() {
        for e in load_key::<Ft8Message>(db, &k("ft4")) {
            h.push_back(e);
        }
    }
    if let Ok(mut h) = histories.ft2.lock() {
        for e in load_key::<Ft8Message>(db, &k("ft2")) {
            h.push_back(e);
        }
    }
    if let Ok(mut h) = histories.wspr.lock() {
        for e in load_key::<WsprMessage>(db, &k("wspr")) {
            h.push_back(e);
        }
    }
    if let Ok(mut h) = histories.lrpt.lock() {
        for e in load_key::<LrptImage>(db, &k("lrpt")) {
            h.push_back(e);
        }
    }
    if let Ok(mut h) = histories.wefax.lock() {
        for e in load_key::<WefaxMessage>(db, &k("wefax")) {
            h.push_back(e);
        }
    }
    if let Ok(mut h) = histories.sstv.lock() {
        for e in load_key::<SstvMessage>(db, &k("sstv")) {
            h.push_back(e);
        }
    }
    histories.recalculate_total_count();
}

/// Flush `histories` to the database under `rig_id`-prefixed keys and sync.
///
/// Each history's mutex is held only long enough to clone the data out,
/// so serialization (which may be slow) never blocks concurrent readers.
pub fn flush_all(db: &mut PickleDb, rig_id: &str, histories: &Arc<DecoderHistories>) {
    let k = |suffix: &str| format!("{}.{}", rig_id, suffix);

    if let Ok(h) = histories.ais.lock() {
        let snapshot = h.clone();
        drop(h);
        save_key(db, &k("ais"), &snapshot);
    }
    if let Ok(h) = histories.vdes.lock() {
        let snapshot = h.clone();
        drop(h);
        save_key(db, &k("vdes"), &snapshot);
    }
    if let Ok(h) = histories.aprs.lock() {
        let snapshot = h.clone();
        drop(h);
        save_key(db, &k("aprs"), &snapshot);
    }
    if let Ok(h) = histories.hf_aprs.lock() {
        let snapshot = h.clone();
        drop(h);
        save_key(db, &k("hf_aprs"), &snapshot);
    }
    if let Ok(h) = histories.cw.lock() {
        let snapshot = h.clone();
        drop(h);
        save_key(db, &k("cw"), &snapshot);
    }
    if let Ok(h) = histories.ft8.lock() {
        let snapshot = h.clone();
        drop(h);
        save_key(db, &k("ft8"), &snapshot);
    }
    if let Ok(h) = histories.ft4.lock() {
        let snapshot = h.clone();
        drop(h);
        save_key(db, &k("ft4"), &snapshot);
    }
    if let Ok(h) = histories.ft2.lock() {
        let snapshot = h.clone();
        drop(h);
        save_key(db, &k("ft2"), &snapshot);
    }
    if let Ok(h) = histories.wspr.lock() {
        let snapshot = h.clone();
        drop(h);
        save_key(db, &k("wspr"), &snapshot);
    }
    if let Ok(h) = histories.lrpt.lock() {
        let snapshot = h.clone();
        drop(h);
        save_key(db, &k("lrpt"), &snapshot);
    }
    if let Ok(h) = histories.wefax.lock() {
        let snapshot = h.clone();
        drop(h);
        save_key(db, &k("wefax"), &snapshot);
    }
    if let Ok(h) = histories.sstv.lock() {
        let snapshot = h.clone();
        drop(h);
        save_key(db, &k("sstv"), &snapshot);
    }
    let _ = db.dump();
}

fn flush_all_rigs(db: &Mutex<PickleDb>, rig_histories: &[(String, Arc<DecoderHistories>)]) {
    let Ok(mut guard) = db.lock() else {
        tracing::warn!("history database mutex poisoned; skipping periodic flush");
        return;
    };
    for (rig_id, histories) in rig_histories {
        flush_all(&mut guard, rig_id, histories);
    }
}

/// Spawn a Tokio task that flushes all rigs' histories to disk every 60 seconds.
///
/// Snapshot cloning, JSON serialization, and disk I/O run on Tokio's blocking
/// pool so a large history database cannot stall an async runtime worker.
pub fn spawn_flush_task(
    db: Arc<Mutex<PickleDb>>,
    rig_histories: Vec<(String, Arc<DecoderHistories>)>,
) {
    tokio::spawn(async move {
        let rig_histories = Arc::new(rig_histories);
        let mut interval = tokio::time::interval(Duration::from_secs(60));
        interval.tick().await; // consume the immediate first tick
        loop {
            interval.tick().await;
            let db = Arc::clone(&db);
            let rig_histories = Arc::clone(&rig_histories);
            if let Err(err) = tokio::task::spawn_blocking(move || {
                flush_all_rigs(&db, rig_histories.as_slice());
            })
            .await
            {
                tracing::warn!(error = %err, "history flush worker failed");
            }
        }
    });
}

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

    #[test]
    fn now_unix_ms_returns_positive() {
        let ms = now_unix_ms();
        // Should be well past epoch (year 2020+).
        assert!(ms > 1_577_836_800_000);
    }

    #[test]
    fn stored_entry_roundtrip_serde() {
        let entry = StoredEntry {
            ts_ms: 1_700_000_000_000i64,
            data: "test message".to_string(),
        };
        let json = serde_json::to_string(&entry).unwrap();
        let decoded: StoredEntry<String> = serde_json::from_str(&json).unwrap();
        assert_eq!(decoded.ts_ms, 1_700_000_000_000);
        assert_eq!(decoded.data, "test message");
    }

    #[test]
    fn save_and_load_key_roundtrip() {
        let dir = std::env::temp_dir().join("trx_history_test");
        let _ = std::fs::create_dir_all(&dir);
        let db_file = dir.join("test.db");
        let mut db = PickleDb::new(
            &db_file,
            PickleDbDumpPolicy::DumpUponRequest,
            SerializationMethod::Json,
        );

        let mut deque = VecDeque::new();
        deque.push_back((Instant::now(), "entry_a".to_string()));
        deque.push_back((Instant::now(), "entry_b".to_string()));

        save_key(&mut db, "test_key", &deque);
        let loaded: Vec<(Instant, String)> = load_key(&db, "test_key");

        assert_eq!(loaded.len(), 2);
        assert_eq!(loaded[0].1, "entry_a");
        assert_eq!(loaded[1].1, "entry_b");

        let _ = std::fs::remove_file(&db_file);
        let _ = std::fs::remove_dir(&dir);
    }

    #[test]
    fn load_key_filters_expired_entries() {
        let dir = std::env::temp_dir().join("trx_history_test_expired");
        let _ = std::fs::create_dir_all(&dir);
        let db_file = dir.join("test.db");
        let mut db = PickleDb::new(
            &db_file,
            PickleDbDumpPolicy::DumpUponRequest,
            SerializationMethod::Json,
        );

        // Manually insert an entry with an old timestamp.
        let entries = vec![
            StoredEntry {
                ts_ms: 1_000, // Way in the past
                data: "old".to_string(),
            },
            StoredEntry {
                ts_ms: now_unix_ms(), // Current
                data: "fresh".to_string(),
            },
        ];
        let _ = db.set("expiry_test", &entries);

        let loaded: Vec<(Instant, String)> = load_key(&db, "expiry_test");
        assert_eq!(loaded.len(), 1);
        assert_eq!(loaded[0].1, "fresh");

        let _ = std::fs::remove_file(&db_file);
        let _ = std::fs::remove_dir(&dir);
    }

    #[test]
    fn load_key_missing_returns_empty() {
        let dir = std::env::temp_dir().join("trx_history_test_missing");
        let _ = std::fs::create_dir_all(&dir);
        let db_file = dir.join("test.db");
        let db = PickleDb::new(
            &db_file,
            PickleDbDumpPolicy::DumpUponRequest,
            SerializationMethod::Json,
        );

        let loaded: Vec<(Instant, String)> = load_key(&db, "nonexistent");
        assert!(loaded.is_empty());

        let _ = std::fs::remove_file(&db_file);
        let _ = std::fs::remove_dir(&dir);
    }

    #[test]
    fn flush_and_load_all_restores_previously_omitted_histories() {
        let db_file = std::env::temp_dir().join(format!(
            "trx_history_all_{}_{}.db",
            std::process::id(),
            now_unix_ms()
        ));
        let mut db = PickleDb::new(
            &db_file,
            PickleDbDumpPolicy::DumpUponRequest,
            SerializationMethod::Json,
        );
        let source = DecoderHistories::new();
        let now = Instant::now();

        source.hf_aprs.lock().unwrap().push_back((
            now,
            AprsPacket {
                rig_id: Some("rig-a".into()),
                ts_ms: Some(now_unix_ms()),
                src_call: "TEST".into(),
                dest_call: "APRS".into(),
                path: String::new(),
                info: "history".into(),
                info_bytes: Vec::new(),
                packet_type: "position".into(),
                crc_ok: true,
                lat: None,
                lon: None,
                symbol_table: None,
                symbol_code: None,
            },
        ));
        for (queue, mode) in [(&source.ft4, "FT4"), (&source.ft2, "FT2")] {
            queue.lock().unwrap().push_back((
                now,
                Ft8Message {
                    rig_id: Some("rig-a".into()),
                    ts_ms: now_unix_ms(),
                    snr_db: -10.0,
                    dt_s: 0.1,
                    freq_hz: 1_000.0,
                    message: mode.into(),
                },
            ));
        }
        source.lrpt.lock().unwrap().push_back((
            now,
            LrptImage {
                rig_id: Some("rig-a".into()),
                pass_start_ms: now_unix_ms(),
                pass_end_ms: now_unix_ms(),
                mcu_count: 1,
                path: "/tmp/lrpt.png".into(),
                ts_ms: Some(now_unix_ms()),
                satellite: None,
                channels: None,
                geo_bounds: None,
                ground_track: None,
            },
        ));

        flush_all(&mut db, "rig-a", &source);
        let restored = DecoderHistories::new();
        load_all(&db, "rig-a", &restored);

        assert_eq!(restored.hf_aprs.lock().unwrap().len(), 1);
        assert_eq!(restored.ft4.lock().unwrap().len(), 1);
        assert_eq!(restored.ft2.lock().unwrap().len(), 1);
        assert_eq!(restored.lrpt.lock().unwrap().len(), 1);
        assert_eq!(restored.estimated_total_count(), 4);

        let _ = std::fs::remove_file(db_file);
    }
}