observer-storage 0.2.0

本地优先前端观测平台 · 纯存储层(落盘 + annotations + bundle 契约)
Documentation
//! 会话保留策略:超龄/孤儿/超量清理(带保护名单变体)。
//!
//! 供 observer-server 后台清扫线程(多租户 per-tenant)、console 本地模式(单租户)
//! 与 tauri-plugin-observer Local 清扫线程(P23)共用。纯逻辑,吃 `&Path`,零 tauri 依赖。

use std::collections::HashSet;
use std::fs;
use std::path::Path;

use serde::{Deserialize, Serialize};

use crate::bundle::{list_sessions, validate_session_id};
use crate::storage::now_ms;

/// 保留策略:任一字段为 None 表示不限该维度。
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", default)]
pub struct RetentionPolicy {
    /// 会话 endedAt 超过 max_age_days 被清理(未结束的会话不在本路径——孤儿见
    /// [`RetentionPolicy::orphan_max_age_days`])。
    pub max_age_days: Option<u32>,
    /// 按 startedAt 倒序保留 max_sessions 条,余者清理。
    pub max_sessions: Option<u32>,
    /// 孤儿清理(P23):无 endedAt 且 startedAt 超过 orphan_max_age_days 的会话被清理,
    /// 覆盖崩溃/强杀/未登出直接退出产生的无收口会话(活跃会话由 protect 名单豁免)。
    /// None = 不清孤儿(0.1.x 语义)。
    pub orphan_max_age_days: Option<u32>,
}

impl RetentionPolicy {
    pub fn is_empty(&self) -> bool {
        self.max_age_days.is_none()
            && self.max_sessions.is_none()
            && self.orphan_max_age_days.is_none()
    }
}

/// `Some(0)` 视同 `None`:策略生效首启 30s 内删光存量不可能是合法意图,防宿主笔误。
fn effective(v: Option<u32>) -> Option<u32> {
    v.filter(|&x| x > 0)
}

/// 删除纪律:meta.id 先过 [`validate_session_id`](id 取自 session.json,防 `../`
/// 路径穿越与 id≠目录名误删),`remove_dir_all` 实际成功才算删除。
fn remove_session_dir(root: &Path, id: &str) -> bool {
    if !validate_session_id(id) {
        return false;
    }
    fs::remove_dir_all(root.join(id)).is_ok()
}

/// 按策略清理 root 下的会话,`protect` 名单中的会话 id 三条路径均跳过(如插件侧
/// 活跃会话,P23)。返回实际删除数(删除失败不计,如 Windows 文件锁,下轮重试)。
///
/// - 超龄:`endedAt < now - max_age_days` 删除(无 endedAt 的会话不在本路径)。
/// - 孤儿:无 endedAt 且 `startedAt < now - orphan_max_age_days` 删除(起点不明保守保留)。
/// - 超量:非 protect 会话按 startedAt 倒序保留前 max_sessions 条(protect 不占名额),
///   余者删除。
///
/// 纪律:`Some(0)` 一律视同 `None`;meta.id 非法跳过不删不计数;session.json 缺失/
/// 损坏的目录不在扫描结果内,三条路径均不可见(已知边界)。策略为空(三维全
/// None/0)直接返回 0,不做扫描。
pub fn enforce_retention_protect(
    root: &Path,
    policy: &RetentionPolicy,
    protect: &[&str],
) -> Result<usize, String> {
    let max_age = effective(policy.max_age_days);
    let orphan_age = effective(policy.orphan_max_age_days);
    let max_sessions = effective(policy.max_sessions);
    if max_age.is_none() && orphan_age.is_none() && max_sessions.is_none() {
        return Ok(0);
    }
    let protected: HashSet<&str> = protect.iter().copied().collect();
    let mut sessions = list_sessions(root);
    let now = now_ms();
    let mut removed = 0usize;

    // 超龄清理(只认 endedAt)
    if let Some(days) = max_age {
        let cutoff = now - (days as i64) * 86_400_000;
        sessions.retain(|s| {
            let Some(id) = s["id"].as_str() else {
                return true;
            };
            if protected.contains(id) {
                return true;
            }
            let overage = matches!(s["endedAt"].as_i64(), Some(t) if t < cutoff);
            if overage && remove_session_dir(root, id) {
                removed += 1;
                return false;
            }
            true
        });
    }

    // 孤儿清理(无 endedAt 且 startedAt 超龄)
    if let Some(days) = orphan_age {
        let cutoff = now - (days as i64) * 86_400_000;
        sessions.retain(|s| {
            let Some(id) = s["id"].as_str() else {
                return true;
            };
            if protected.contains(id) {
                return true;
            }
            if s["endedAt"].as_i64().is_some() {
                return true; // 有收口归超龄/超量路径
            }
            let Some(started) = s["startedAt"].as_i64() else {
                return true;
            };
            if started < cutoff && remove_session_dir(root, id) {
                removed += 1;
                return false;
            }
            true
        });
    }

    // 超量清理:非 protect 集合按 startedAt 倒序,保留前 max_sessions
    if let Some(max) = max_sessions {
        let mut eligible: Vec<&serde_json::Value> = sessions
            .iter()
            .filter(|s| s["id"].as_str().map_or(true, |id| !protected.contains(id)))
            .collect();
        eligible.sort_by(|a, b| b["startedAt"].as_i64().cmp(&a["startedAt"].as_i64()));
        for s in eligible.iter().skip(max as usize) {
            if let Some(id) = s["id"].as_str() {
                if remove_session_dir(root, id) {
                    removed += 1;
                }
            }
        }
    }

    Ok(removed)
}

/// 兼容保留(0.1.x 签名):等价 protect 空名单。既有调用方(server 清扫线程 /
/// console 单租户)策略不含 [`RetentionPolicy::orphan_max_age_days`](serde default
/// None)→ 行为不变,仅删除纪律(防呆/校验/实际计数)随变体生效。
pub fn enforce_retention(root: &Path, policy: &RetentionPolicy) -> Result<usize, String> {
    enforce_retention_protect(root, policy, &[])
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;
    use tempfile::tempdir;

    fn make_session(root: &Path, id: &str, started: i64, ended: Option<i64>) {
        let dir = root.join(id);
        fs::create_dir_all(dir.join("segments")).unwrap();
        let mut s = json!({ "id": id, "startedAt": started });
        if let Some(e) = ended {
            s["endedAt"] = json!(e);
        }
        fs::write(dir.join("session.json"), s.to_string()).unwrap();
    }

    /// 超龄会话被清,活跃会话(无 endedAt)与未超龄保留。
    #[test]
    fn enforces_max_age() {
        let root = tempdir().unwrap();
        let now = now_ms();
        make_session(root.path(), "900000000002", now - 86_400_000 * 40, Some(now - 86_400_000 * 40));
        make_session(root.path(), "900000000003", now - 1000, Some(now - 1000));
        make_session(root.path(), "900000000004", now - 2000, None); // 活跃,不清理

        let removed = enforce_retention(
            root.path(),
            &RetentionPolicy {
                max_age_days: Some(30),
                max_sessions: None,
                orphan_max_age_days: None,
            },
        )
        .unwrap();
        assert_eq!(removed, 1);
        assert!(!root.path().join("900000000002").exists());
        assert!(root.path().join("900000000003").exists());
        assert!(root.path().join("900000000004").exists());
    }

    /// 超量按 startedAt 倒序淘汰。
    #[test]
    fn enforces_max_sessions() {
        let root = tempdir().unwrap();
        make_session(root.path(), "1001", 1000, Some(1100));
        make_session(root.path(), "1002", 2000, Some(2100));
        make_session(root.path(), "1003", 3000, Some(3100));

        let removed = enforce_retention(
            root.path(),
            &RetentionPolicy {
                max_age_days: None,
                max_sessions: Some(2),
                orphan_max_age_days: None,
            },
        )
        .unwrap();
        assert_eq!(removed, 1);
        // 保留最新两条 s3/s2,淘汰 s1
        assert!(!root.path().join("1001").exists());
        assert!(root.path().join("1002").exists());
        assert!(root.path().join("1003").exists());
    }

    /// 空策略不扫描、不删。
    #[test]
    fn empty_policy_noop() {
        let root = tempdir().unwrap();
        make_session(root.path(), "1001", 1000, Some(1100));
        let removed = enforce_retention(root.path(), &RetentionPolicy::default()).unwrap();
        assert_eq!(removed, 0);
        assert!(root.path().join("1001").exists());
    }

    /// protect 名单中的已收口超龄会话保留(超龄路径被 protect 收窄)。
    #[test]
    fn protect_skips_in_max_age() {
        let root = tempdir().unwrap();
        let now = now_ms();
        make_session(root.path(), "900000000001", now - 86_400_000 * 40, Some(now - 86_400_000 * 40));
        make_session(root.path(), "900000000002", now - 86_400_000 * 40, Some(now - 86_400_000 * 40));

        let policy = RetentionPolicy {
            max_age_days: Some(30),
            max_sessions: None,
            orphan_max_age_days: None,
        };
        let removed = enforce_retention_protect(root.path(), &policy, &["900000000001"]).unwrap();
        assert_eq!(removed, 1);
        assert!(root.path().join("900000000001").exists());
        assert!(!root.path().join("900000000002").exists());
    }

    /// 超量路径 protect 不占名额:受保护的最老会话保留,顺位多删一条。
    #[test]
    fn protect_skips_in_max_sessions() {
        let root = tempdir().unwrap();
        make_session(root.path(), "1001", 1000, Some(1100));
        make_session(root.path(), "1002", 2000, Some(2100));
        make_session(root.path(), "1003", 3000, Some(3100));
        make_session(root.path(), "1004", 4000, Some(4100));

        let policy = RetentionPolicy {
            max_age_days: None,
            max_sessions: Some(2),
            orphan_max_age_days: None,
        };
        let removed = enforce_retention_protect(root.path(), &policy, &["1001"]).unwrap();
        // eligible = s2/s3/s4(s1 受保护不占名额),保留 s4/s3,删 s2
        assert_eq!(removed, 1);
        assert!(root.path().join("1001").exists());
        assert!(!root.path().join("1002").exists());
        assert!(root.path().join("1003").exists());
        assert!(root.path().join("1004").exists());
    }

    /// 孤儿:无 endedAt 且 startedAt 超龄可清(orphan_max_age_days 显式开启)。
    #[test]
    fn orphan_swept_by_started_at() {
        let root = tempdir().unwrap();
        let now = now_ms();
        make_session(root.path(), "800000000001", now - 86_400_000 * 40, None);
        make_session(root.path(), "800000000002", now - 86_400_000 * 40, Some(now - 86_400_000 * 40));

        let policy = RetentionPolicy {
            max_age_days: None,
            max_sessions: None,
            orphan_max_age_days: Some(30),
        };
        let removed = enforce_retention_protect(root.path(), &policy, &[]).unwrap();
        assert_eq!(removed, 1);
        assert!(!root.path().join("800000000001").exists());
        assert!(root.path().join("800000000002").exists()); // 有 endedAt 不归孤儿路径
    }

    /// 孤儿路径同样受 protect 豁免(宿主合法长活会话,P22 D5)。
    #[test]
    fn orphan_skips_protected() {
        let root = tempdir().unwrap();
        let now = now_ms();
        make_session(root.path(), "800000000003", now - 86_400_000 * 40, None);

        let policy = RetentionPolicy {
            max_age_days: None,
            max_sessions: None,
            orphan_max_age_days: Some(30),
        };
        let removed = enforce_retention_protect(root.path(), &policy, &["800000000003"]).unwrap();
        assert_eq!(removed, 0);
        assert!(root.path().join("800000000003").exists());
    }

    /// orphan 字段 None = 0.1.x 语义:无 endedAt 一律保留(legacy 回归锚)。
    #[test]
    fn orphan_none_keeps_legacy_semantics() {
        let root = tempdir().unwrap();
        let now = now_ms();
        make_session(root.path(), "800000000001", now - 86_400_000 * 40, None);

        let policy = RetentionPolicy {
            max_age_days: Some(30),
            max_sessions: None,
            orphan_max_age_days: None,
        };
        let removed = enforce_retention_protect(root.path(), &policy, &[]).unwrap();
        assert_eq!(removed, 0);
        assert!(root.path().join("800000000001").exists());
    }

    /// `Some(0)` 三维一律视同 None:不删任何东西(防宿主笔误)。
    #[test]
    fn zero_values_treated_as_noop() {
        let root = tempdir().unwrap();
        let now = now_ms();
        make_session(root.path(), "900000000002", now - 86_400_000 * 40, Some(now - 86_400_000 * 40));
        make_session(root.path(), "800000000001", now - 86_400_000 * 40, None);

        let policy = RetentionPolicy {
            max_age_days: Some(0),
            max_sessions: Some(0),
            orphan_max_age_days: Some(0),
        };
        let removed = enforce_retention_protect(root.path(), &policy, &[]).unwrap();
        assert_eq!(removed, 0);
        assert!(root.path().join("900000000002").exists());
        assert!(root.path().join("800000000001").exists());
    }

    /// meta.id 非法(路径穿越 / 与目录名不符)跳过不删不计数。
    #[test]
    fn invalid_session_id_skipped() {
        let root = tempdir().unwrap();
        let now = now_ms();
        let dir = root.path().join("dir-a");
        fs::create_dir_all(dir.join("segments")).unwrap();
        fs::write(
            dir.join("session.json"),
            json!({
                "id": "../evil",
                "startedAt": now - 86_400_000 * 40,
                "endedAt": now - 86_400_000 * 40
            })
            .to_string(),
        )
        .unwrap();

        let policy = RetentionPolicy {
            max_age_days: Some(30),
            max_sessions: None,
            orphan_max_age_days: None,
        };
        let removed = enforce_retention_protect(root.path(), &policy, &[]).unwrap();
        assert_eq!(removed, 0);
        assert!(root.path().join("dir-a").exists());
    }

    /// 原函数 = protect 空名单(委托等价回归锚)。
    #[test]
    fn delegate_empty_protect_equivalent() {
        let root = tempdir().unwrap();
        let now = now_ms();
        make_session(root.path(), "900000000002", now - 86_400_000 * 40, Some(now - 86_400_000 * 40));

        let policy = RetentionPolicy {
            max_age_days: Some(30),
            max_sessions: None,
            orphan_max_age_days: None,
        };
        let removed = enforce_retention(root.path(), &policy).unwrap();
        assert_eq!(removed, 1);
        assert!(!root.path().join("900000000002").exists());
    }
}