domi_server/events/
writer.rs

1use std::{
2    fs::{File, OpenOptions},
3    io::Write,
4    path::{Path, PathBuf},
5    time::{SystemTime, UNIX_EPOCH},
6};
7
8use fs2::FileExt;
9
10use crate::events::event::Event;
11
12const DEFAULT_SIZE_CAP: u64 = 50 * 1024 * 1024;
13
14#[derive(Debug)]
15#[non_exhaustive]
16pub enum WriteError {
17    Io(std::io::Error),
18    LockBusy,
19}
20
21impl std::fmt::Display for WriteError {
22    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
23        match self {
24            WriteError::Io(e) => write!(f, "io: {e}"),
25            WriteError::LockBusy => write!(f, "events.lock is held by another writer"),
26        }
27    }
28}
29
30impl std::error::Error for WriteError {}
31
32impl From<std::io::Error> for WriteError {
33    fn from(e: std::io::Error) -> Self { WriteError::Io(e) }
34}
35
36#[derive(Debug)]
37pub struct Rotation {
38    pub from: PathBuf,
39    pub to: PathBuf,
40}
41
42#[derive(Debug)]
43pub enum FileShape {
44    Empty,
45    V2,
46    Legacy,
47    MalformedJson,
48}
49
50pub struct EventWriter {
51    path: PathBuf,
52    size_cap: u64,
53}
54
55impl EventWriter {
56    pub fn new(path: impl Into<PathBuf>) -> Self {
57        Self { path: path.into(), size_cap: DEFAULT_SIZE_CAP }
58    }
59
60    pub fn with_size_cap(mut self, bytes: u64) -> Self {
61        self.size_cap = bytes;
62        self
63    }
64
65    pub fn file_shape(path: &Path) -> std::io::Result<FileShape> {
66        match File::open(path) {
67            Ok(f) => {
68                let mut reader = std::io::BufReader::new(f);
69                let mut first = String::new();
70                use std::io::Read;
71                if reader.read_to_string(&mut first)? == 0 {
72                    return Ok(FileShape::Empty);
73                }
74                let first_line = first.lines().next().unwrap_or("");
75                if first_line.trim().is_empty() {
76                    return Ok(FileShape::Empty);
77                }
78                // JSON syntax check first: distinguishes MalformedJson (syntax error)
79                // from Legacy (parses but not a v2 Event).
80                if serde_json::from_str::<serde_json::Value>(first_line).is_err() {
81                    return Ok(FileShape::MalformedJson);
82                }
83                match serde_json::from_str::<Event>(first_line) {
84                    Ok(e) if e.v == 2 => Ok(FileShape::V2),
85                    Ok(_) => Ok(FileShape::Legacy),
86                    Err(_) => Ok(FileShape::Legacy),
87                }
88            }
89            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(FileShape::Empty),
90            Err(e) => Err(e),
91        }
92    }
93
94    fn rotate_filename() -> PathBuf {
95        let secs = SystemTime::now().duration_since(UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0);
96        let stamp = Self::format_utc_iso8601_compact(secs);
97        PathBuf::from(format!("events-{stamp}.jsonl"))
98    }
99
100    /// Compact UTC timestamp for filenames: `YYYY-MM-DDTHH-MM-SSZ`. Colons
101    /// are replaced with dashes because some filesystems disallow them.
102    /// Origin is the Unix epoch (1970-01-01T00-00-00Z).
103    fn format_utc_iso8601_compact(secs: u64) -> String {
104        use chrono::{DateTime, Utc};
105        let dt: DateTime<Utc> = DateTime::from_timestamp(secs as i64, 0).unwrap_or_else(Utc::now);
106        let formatted = dt.format("%Y-%m-%dT%H-%M-%SZ").to_string();
107        formatted.replace(':', "-")
108    }
109
110    fn lock_path_for(path: &Path) -> PathBuf {
111        let mut p = path.to_path_buf();
112        let file = p.file_name().unwrap_or_else(|| std::ffi::OsStr::new("events.jsonl")).to_os_string();
113        p.set_file_name(format!("{}.lock", file.to_string_lossy()));
114        p
115    }
116
117    pub fn write(&self, event: &Event) -> Result<(), WriteError> {
118        if event.doc.is_empty() {
119            return Err(WriteError::Io(std::io::Error::new(
120                std::io::ErrorKind::InvalidInput,
121                "doc must be non-empty",
122            )));
123        }
124        let lock_path = Self::lock_path_for(&self.path);
125        if let Some(parent) = lock_path.parent() {
126            std::fs::create_dir_all(parent)?;
127        }
128        let lock_file = OpenOptions::new().create(true).append(true).open(&lock_path)?;
129        lock_file.try_lock_exclusive().map_err(|_| WriteError::LockBusy)?;
130
131        let result = (|| -> Result<(), WriteError> {
132            if let Ok(meta) = std::fs::metadata(&self.path) {
133                if meta.len() >= self.size_cap {
134                    let _ = self.rotate_internal();
135                }
136            }
137            let line = serde_json::to_string(event)
138                .map_err(|e| WriteError::Io(std::io::Error::new(std::io::ErrorKind::Other, e.to_string())))?;
139            let mut line = line;
140            line.push('\n');
141            let mut f = OpenOptions::new().create(true).append(true).open(&self.path)?;
142            f.write_all(line.as_bytes())?;
143            f.sync_all()?;
144            Ok(())
145        })();
146
147        let _ = FileExt::unlock(&lock_file);
148        result
149    }
150
151    pub fn rotate(&self) -> Result<Rotation, WriteError> {
152        self.rotate_internal()
153    }
154
155    fn rotate_internal(&self) -> Result<Rotation, WriteError> {
156        if !self.path.exists() {
157            return Ok(Rotation { from: self.path.clone(), to: self.path.clone() });
158        }
159        let to = self.path.with_file_name(Self::rotate_filename());
160        std::fs::rename(&self.path, &to)?;
161        Ok(Rotation { from: self.path.clone(), to })
162    }
163}
164
165#[cfg(test)]
166mod tests {
167    use super::*;
168    use crate::events::event::{EventData, Kind, Rect, Source, Target};
169    use tempfile::tempdir;
170
171    fn ev(kind: Kind, data: EventData, body: &str) -> Event {
172        Event {
173            v: 2,
174            id: ulid::Ulid::from_string("01H8XZQ5K2J9Z9Q4X5Y6Z7XYZ1").unwrap(),
175            ts: chrono::DateTime::parse_from_rfc3339("2026-07-05T18:21:00Z").unwrap().with_timezone(&chrono::Utc),
176            src: Source::DomiJs,
177            doc: "onboarding-v2".into(),
178            kind,
179            target: Target {
180                id: Some("btn-save".into()),
181                selector: None,
182                rect: Rect { x: 0.0, y: 0.0, w: 1.0, h: 1.0 },
183            },
184            data,
185        }
186    }
187
188    fn write_n(w: &EventWriter, n: usize) {
189        for _ in 0..n {
190            w.write(&ev(
191                Kind::RailAdd,
192                EventData::RailAdd { body: "x".into(), target_id: None },
193                "x",
194            )).unwrap();
195        }
196    }
197
198    #[test]
199    fn write_appends_one_line() {
200        let dir = tempdir().unwrap();
201        let path = dir.path().join("events.jsonl");
202        let w = EventWriter::new(&path);
203        write_n(&w, 1);
204        let body = std::fs::read_to_string(&path).unwrap();
205        assert_eq!(body.lines().count(), 1);
206        assert!(body.ends_with('\n'));
207    }
208
209    #[test]
210    fn jsonl_round_trip_three_events() {
211        let dir = tempdir().unwrap();
212        let path = dir.path().join("events.jsonl");
213        let w = EventWriter::new(&path);
214        write_n(&w, 3);
215        let bytes = std::fs::read(&path).unwrap();
216        let mut de = serde_json::Deserializer::from_reader(&bytes[..]).into_iter::<Event>();
217        let mut count = 0;
218        while de.next().is_some() { count += 1; }
219        assert_eq!(count, 3);
220    }
221
222    #[test]
223    fn rotation_on_size_cap() {
224        let dir = tempdir().unwrap();
225        let path = dir.path().join("events.jsonl");
226        let w = EventWriter::new(&path).with_size_cap(50);
227        write_n(&w, 10);
228        let entries: Vec<_> = std::fs::read_dir(dir.path()).unwrap().filter_map(Result::ok).collect();
229        assert!(entries.len() >= 2, "expected rotation, got {:?}",
230            entries.iter().map(|e| e.file_name()).collect::<Vec<_>>());
231    }
232
233    #[test]
234    fn rotate_renames_file() {
235        let dir = tempdir().unwrap();
236        let path = dir.path().join("events.jsonl");
237        let w = EventWriter::new(&path);
238        write_n(&w, 1);
239        let rotated = w.rotate().unwrap();
240        assert!(!path.exists());
241        assert!(rotated.to.exists());
242    }
243
244    #[test]
245    fn file_shape_detects_legacy() {
246        let dir = tempdir().unwrap();
247        let path = dir.path().join("events.jsonl");
248        std::fs::write(&path, r#"{"id":"a","selector":"b","text":"c"}"#).unwrap();
249        assert!(matches!(EventWriter::file_shape(&path).unwrap(), FileShape::Legacy));
250    }
251
252    #[test]
253    fn file_shape_detects_v2() {
254        let dir = tempdir().unwrap();
255        let path = dir.path().join("events.jsonl");
256        std::fs::write(
257            &path,
258            r#"{"v":2,"id":"01H8XZQ5K2J9Z9Q4X5Y6Z7XYZ0","ts":"2026-07-05T18:21:00Z","src":"domi.js","doc":"x","kind":"click","target":{"id":null,"selector":null,"rect":{"x":0,"y":0,"w":0,"h":0}},"data":{}}"#,
259        ).unwrap();
260        assert!(matches!(EventWriter::file_shape(&path).unwrap(), FileShape::V2));
261    }
262
263    #[test]
264    fn lock_busy_when_held() {
265        let dir = tempdir().unwrap();
266        let path = dir.path().join("events.jsonl");
267        let _ = std::fs::File::create(&path).unwrap(); // touch
268        let lock_path = EventWriter::lock_path_for(&path);
269        let lock_file = std::fs::OpenOptions::new().create(true).append(true).open(&lock_path).unwrap();
270        lock_file.lock_exclusive().unwrap();
271
272        let w = EventWriter::new(&path);
273        let event = ev(
274            Kind::Click,
275            EventData::Click { value: None },
276            "x",
277        );
278        let err = w.write(&event).unwrap_err();
279        assert!(matches!(err, WriteError::LockBusy), "expected LockBusy, got {err:?}");
280
281        // release
282        let _ = fs2::FileExt::unlock(&lock_file);
283    }
284}
domi_server/events/writer.rs

domi_server/events/
writer.rs
