observer_storage/
storage.rs

1//! 会话存储原语：落盘函数，全部吃 `&Path`，零 tauri 依赖。
2//!
3//! 供 [`tauri_plugin_observer`]（Local 模式 self-obs 落盘）与
4//! [`observer_server`](../../observer-server)（HTTP 落盘）共用。
5//!
6//! 格式：`recordings/<sessionId>/{session.json, windows.jsonl, segments/<label>#<n>.jsonl[.gz]}`。
7//! P9 起 segments 可选 gzip 落盘（`.jsonl.gz`），读路径按扩展名透明解压。
8
9use std::fs::{self, OpenOptions};
10use std::io::{Read, Write};
11use std::path::Path;
12use std::time::{SystemTime, UNIX_EPOCH};
13
14use flate2::read::MultiGzDecoder;
15use flate2::write::GzEncoder;
16use flate2::Compression;
17use serde_json::Value;
18
19pub fn now_ms() -> i64 {
20    SystemTime::now()
21        .duration_since(UNIX_EPOCH)
22        .map(|d| d.as_millis() as i64)
23        .unwrap_or(0)
24}
25
26/// segment 落盘选项。`gzip=true` 写 `.jsonl.gz`（每批一个 gzip member，可追加）。
27#[derive(Clone, Copy, Default)]
28pub struct WriteOpts {
29    pub gzip: bool,
30}
31
32impl WriteOpts {
33    pub fn plain() -> Self {
34        Self { gzip: false }
35    }
36    pub fn gzip() -> Self {
37        Self { gzip: true }
38    }
39}
40
41pub fn append_lifecycle(dir: &Path, evt: Value) -> Result<(), String> {
42    let path = dir.join("windows.jsonl");
43    let mut f = OpenOptions::new()
44        .create(true)
45        .append(true)
46        .open(&path)
47        .map_err(|e| e.to_string())?;
48    writeln!(f, "{}", evt).map_err(|e| e.to_string())
49}
50
51/// 追加事件到 segment 文件（plain JSONL，向后兼容）。
52pub fn append_events_file(dir: &Path, segment_id: &str, events: &[Value]) -> Result<(), String> {
53    append_events_file_with(dir, segment_id, events, &WriteOpts::plain())
54}
55
56/// 追加事件到 segment 文件，按 [`WriteOpts`] 决定是否 gzip。
57///
58/// gzip 模式下每次调用写一个独立 gzip member（concatenate 到同一文件），
59/// 读侧用 [`read_segment_events`] 经 [`MultiGzDecoder`] 透明解压全部 member。
60pub fn append_events_file_with(
61    dir: &Path,
62    segment_id: &str,
63    events: &[Value],
64    opts: &WriteOpts,
65) -> Result<(), String> {
66    let seg_dir = dir.join("segments");
67    fs::create_dir_all(&seg_dir).map_err(|e| e.to_string())?;
68    let fname = if opts.gzip {
69        format!("{}.jsonl.gz", segment_id)
70    } else {
71        format!("{}.jsonl", segment_id)
72    };
73    let path = seg_dir.join(&fname);
74    let f = OpenOptions::new()
75        .create(true)
76        .append(true)
77        .open(&path)
78        .map_err(|e| e.to_string())?;
79    if opts.gzip {
80        let mut enc = GzEncoder::new(f, Compression::default());
81        for e in events {
82            writeln!(enc, "{}", e).map_err(|e| e.to_string())?;
83        }
84        enc.finish().map_err(|e| e.to_string())?;
85    } else {
86        let mut f = f;
87        for e in events {
88            writeln!(f, "{}", e).map_err(|e| e.to_string())?;
89        }
90    }
91    Ok(())
92}
93
94/// 读 segment 文件事件（gzip 感知：`.jsonl.gz` 用 [`MultiGzDecoder`]，`.jsonl` 直读）。
95/// 文件不存在或单行解析失败被跳过。
96pub fn read_segment_events(path: &Path) -> Vec<Value> {
97    let Some(raw) = read_segment_text(path) else {
98        return Vec::new();
99    };
100    raw.lines()
101        .filter(|l| !l.trim().is_empty())
102        .filter_map(|l| serde_json::from_str(l).ok())
103        .collect()
104}
105
106/// 由文件名还原 segmentId：`web#1.jsonl.gz` / `web#1.jsonl` -> `web#1`。
107pub fn segment_id_from_filename(name: &str) -> &str {
108    name.strip_suffix(".jsonl.gz")
109        .or_else(|| name.strip_suffix(".jsonl"))
110        .unwrap_or(name)
111}
112
113fn read_segment_text(path: &Path) -> Option<String> {
114    let is_gz = path.extension().and_then(|e| e.to_str()) == Some("gz");
115    if is_gz {
116        let f = fs::File::open(path).ok()?;
117        let mut dec = MultiGzDecoder::new(f);
118        let mut s = String::new();
119        dec.read_to_string(&mut s).ok()?;
120        Some(s)
121    } else {
122        fs::read_to_string(path).ok()
123    }
124}
125
126/// 建会话目录并写 session.json（meta 由调用方组装好）。
127pub fn create_session(dir: &Path, meta: Value) -> Result<(), String> {
128    fs::create_dir_all(dir.join("segments")).map_err(|e| e.to_string())?;
129    fs::write(dir.join("session.json"), meta.to_string()).map_err(|e| e.to_string())
130}
131
132/// 结束会话：读回 session.json 补 endedAt 后覆写。
133pub fn finalize_session(dir: &Path, ended_at: i64) -> Result<(), String> {
134    let path = dir.join("session.json");
135    let mut v = serde_json::from_str::<Value>(
136        &fs::read_to_string(&path).map_err(|e| e.to_string())?,
137    )
138    .map_err(|e| e.to_string())?;
139    if let Some(obj) = v.as_object_mut() {
140        obj.insert("endedAt".into(), serde_json::json!(ended_at));
141    }
142    fs::write(path, v.to_string()).map_err(|e| e.to_string())
143}
144
145#[cfg(test)]
146mod tests {
147    use super::*;
148    use serde_json::json;
149
150    #[test]
151    fn append_events_writes_jsonl() {
152        let dir = tempfile::tempdir().unwrap();
153        append_events_file(
154            dir.path(),
155            "main#0",
156            &[json!({ "type": 2, "timestamp": 1 }), json!({ "type": 6, "timestamp": 2 })],
157        )
158        .unwrap();
159        let lines: Vec<Value> = fs::read_to_string(dir.path().join("segments/main#0.jsonl"))
160            .unwrap()
161            .lines()
162            .filter(|l| !l.is_empty())
163            .map(|l| serde_json::from_str(l).unwrap())
164            .collect();
165        assert_eq!(lines.len(), 2);
166        assert_eq!(lines[1]["type"], 6);
167    }
168
169    /// gzip 追加：两次调用 = 两个 concatenate gzip member，读侧 MultiGzDecoder 还原全部。
170    #[test]
171    fn append_events_gzip_roundtrip() {
172        let dir = tempfile::tempdir().unwrap();
173        let seg = "web#1";
174        append_events_file_with(
175            dir.path(),
176            seg,
177            &[json!({ "type": 2, "timestamp": 1 })],
178            &WriteOpts::gzip(),
179        )
180        .unwrap();
181        append_events_file_with(
182            dir.path(),
183            seg,
184            &[json!({ "type": 6, "timestamp": 2 })],
185            &WriteOpts::gzip(),
186        )
187        .unwrap();
188
189        let path = dir.path().join("segments/web#1.jsonl.gz");
190        assert!(path.exists(), ".gz 文件应存在");
191        assert!(
192            !dir.path().join("segments/web#1.jsonl").exists(),
193            "不应同时存在 plain 文件"
194        );
195
196        let events = read_segment_events(&path);
197        assert_eq!(events.len(), 2);
198        assert_eq!(events[0]["type"], 2);
199        assert_eq!(events[1]["type"], 6);
200    }
201
202    /// 读路径同时支持 plain 与 gz，按扩展名分派。
203    #[test]
204    fn read_segment_events_plain_and_gz() {
205        let dir = tempfile::tempdir().unwrap();
206        append_events_file(dir.path(), "a#0", &[json!({ "type": 2, "timestamp": 1 })]).unwrap();
207        let plain = read_segment_events(&dir.path().join("segments/a#0.jsonl"));
208        assert_eq!(plain.len(), 1);
209
210        append_events_file_with(
211            dir.path(),
212            "b#0",
213            &[json!({ "type": 2, "timestamp": 2 })],
214            &WriteOpts::gzip(),
215        )
216        .unwrap();
217        let gz = read_segment_events(&dir.path().join("segments/b#0.jsonl.gz"));
218        assert_eq!(gz.len(), 1);
219        assert_eq!(gz[0]["timestamp"], 2);
220    }
221
222    #[test]
223    fn segment_id_from_filename_strips_extensions() {
224        assert_eq!(segment_id_from_filename("web#1.jsonl.gz"), "web#1");
225        assert_eq!(segment_id_from_filename("web#1.jsonl"), "web#1");
226        assert_eq!(segment_id_from_filename("web#1"), "web#1");
227    }
228}
observer_storage/storage.rs

observer_storage/
storage.rs
