observer_storage/
retention.rs

1//! 会话保留策略：超龄/超量清理。
2//!
3//! 供 observer-server 后台清扫线程（多租户 per-tenant）与 console 本地模式（单租户）
4//! 共用。纯逻辑，吃 `&Path`，零 tauri 依赖。
5
6use std::fs;
7use std::path::Path;
8
9use serde::{Deserialize, Serialize};
10
11use crate::bundle::list_sessions;
12use crate::storage::now_ms;
13
14/// 保留策略：任一字段为 None 表示不限该维度。
15#[derive(Clone, Debug, Default, Serialize, Deserialize)]
16#[serde(rename_all = "camelCase")]
17pub struct RetentionPolicy {
18    /// 会话 endedAt 超过 max_age_days 被清理（未结束的会话不清理）。
19    pub max_age_days: Option<u32>,
20    /// 按 startedAt 倒序保留 max_sessions 条，余者清理。
21    pub max_sessions: Option<u32>,
22}
23
24impl RetentionPolicy {
25    pub fn is_empty(&self) -> bool {
26        self.max_age_days.is_none() && self.max_sessions.is_none()
27    }
28}
29
30/// 按策略清理 root 下的会话：先超龄，再超量。返回移除数量。
31///
32/// - 超龄：`endedAt < now - max_age_days` 删除（无 endedAt 的活跃会话跳过）。
33/// - 超量：剩余按 startedAt 倒序，保留前 max_sessions 条，余者删除。
34///
35/// 策略为空时直接返回 0，不做扫描。
36pub fn enforce_retention(root: &Path, policy: &RetentionPolicy) -> Result<usize, String> {
37    if policy.is_empty() {
38        return Ok(0);
39    }
40    let mut sessions = list_sessions(root);
41    let now = now_ms();
42    let mut removed = 0usize;
43
44    // 超龄清理
45    if let Some(days) = policy.max_age_days {
46        let cutoff = now - (days as i64) * 86_400_000;
47        sessions.retain(|s| {
48            let keep = s["endedAt"].as_i64().map_or(true, |t| t >= cutoff);
49            if !keep {
50                if let Some(id) = s["id"].as_str() {
51                    let _ = fs::remove_dir_all(root.join(id));
52                    removed += 1;
53                }
54            }
55            keep
56        });
57    }
58
59    // 超量清理：按 startedAt 倒序，保留前 max_sessions
60    if let Some(max) = policy.max_sessions {
61        sessions.sort_by(|a, b| b["startedAt"].as_i64().cmp(&a["startedAt"].as_i64()));
62        for s in sessions.iter().skip(max as usize) {
63            if let Some(id) = s["id"].as_str() {
64                let _ = fs::remove_dir_all(root.join(id));
65                removed += 1;
66            }
67        }
68    }
69
70    Ok(removed)
71}
72
73#[cfg(test)]
74mod tests {
75    use super::*;
76    use serde_json::json;
77    use tempfile::tempdir;
78
79    fn make_session(root: &Path, id: &str, started: i64, ended: Option<i64>) {
80        let dir = root.join(id);
81        fs::create_dir_all(dir.join("segments")).unwrap();
82        let mut s = json!({ "id": id, "startedAt": started });
83        if let Some(e) = ended {
84            s["endedAt"] = json!(e);
85        }
86        fs::write(dir.join("session.json"), s.to_string()).unwrap();
87    }
88
89    /// 超龄会话被清，活跃会话（无 endedAt）与未超龄保留。
90    #[test]
91    fn enforces_max_age() {
92        let root = tempdir().unwrap();
93        let now = now_ms();
94        make_session(root.path(), "old", now - 86_400_000 * 40, Some(now - 86_400_000 * 40));
95        make_session(root.path(), "fresh", now - 1000, Some(now - 1000));
96        make_session(root.path(), "active", now - 2000, None); // 活跃，不清理
97
98        let removed = enforce_retention(
99            root.path(),
100            &RetentionPolicy {
101                max_age_days: Some(30),
102                max_sessions: None,
103            },
104        )
105        .unwrap();
106        assert_eq!(removed, 1);
107        assert!(!root.path().join("old").exists());
108        assert!(root.path().join("fresh").exists());
109        assert!(root.path().join("active").exists());
110    }
111
112    /// 超量按 startedAt 倒序淘汰。
113    #[test]
114    fn enforces_max_sessions() {
115        let root = tempdir().unwrap();
116        make_session(root.path(), "s1", 1000, Some(1100));
117        make_session(root.path(), "s2", 2000, Some(2100));
118        make_session(root.path(), "s3", 3000, Some(3100));
119
120        let removed = enforce_retention(
121            root.path(),
122            &RetentionPolicy {
123                max_age_days: None,
124                max_sessions: Some(2),
125            },
126        )
127        .unwrap();
128        assert_eq!(removed, 1);
129        // 保留最新两条 s3/s2，淘汰 s1
130        assert!(!root.path().join("s1").exists());
131        assert!(root.path().join("s2").exists());
132        assert!(root.path().join("s3").exists());
133    }
134
135    /// 空策略不扫描、不删。
136    #[test]
137    fn empty_policy_noop() {
138        let root = tempdir().unwrap();
139        make_session(root.path(), "s1", 1000, Some(1100));
140        let removed = enforce_retention(root.path(), &RetentionPolicy::default()).unwrap();
141        assert_eq!(removed, 0);
142        assert!(root.path().join("s1").exists());
143    }
144}
observer_storage/retention.rs

observer_storage/
retention.rs
