mirror_log/
stage.rs

1use chrono::{DateTime, TimeZone, Utc};
2use std::fs;
3use std::path::Path;
4use uuid::Uuid;
5
6#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
7pub struct StagedEvent {
8    pub id: String,
9    pub source: String,
10    pub content: String,
11    pub meta: Option<String>,
12    pub timestamp: i64,
13}
14
15impl StagedEvent {
16    pub fn new(source: &str, content: &str, meta: Option<&str>) -> Self {
17        let now = Utc::now().timestamp();
18        let id = Uuid::new_v4().to_string();
19
20        Self {
21            id,
22            source: source.to_string(),
23            content: content.to_string(),
24            meta: meta.map(|s| s.to_string()),
25            timestamp: now,
26        }
27    }
28
29    /// Return the event timestamp as a `DateTime<Utc>`.
30    pub fn timestamp_utc(&self) -> DateTime<Utc> {
31        Utc.timestamp_opt(self.timestamp, 0).unwrap()
32    }
33
34    /// Persist this staged event to disk as a JSON file.
35    pub fn save_to_file(&self, staging_dir: &Path) -> Result<(), std::io::Error> {
36        let filename = format!("{}.json", self.id);
37        let path = staging_dir.join(filename);
38
39        fs::create_dir_all(staging_dir)?;
40        let json = serde_json::to_string_pretty(self)?;
41
42        fs::write(path, json)?;
43        Ok(())
44    }
45
46    /// Load a staged event by its ID from the staging directory.
47    pub fn load_from_file(id: &str, staging_dir: &Path) -> Result<Self, std::io::Error> {
48        let filename = format!("{}.json", id);
49        let path = staging_dir.join(filename);
50
51        let json = fs::read_to_string(path)?;
52        let event: StagedEvent = serde_json::from_str(&json)?;
53        Ok(event)
54    }
55
56    /// Load a staged event from an arbitrary file path.
57    pub fn from_file(path: &Path) -> Result<Self, std::io::Error> {
58        let json = fs::read_to_string(path)?;
59        let event: StagedEvent = serde_json::from_str(&json)
60            .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
61        Ok(event)
62    }
63
64    /// Load all staged events from a directory.
65    pub fn load_all(staging_dir: &Path) -> Result<Vec<Self>, std::io::Error> {
66        let mut events = Vec::new();
67
68        if !staging_dir.exists() {
69            return Ok(events);
70        }
71
72        for entry in fs::read_dir(staging_dir)? {
73            let path = entry?.path();
74            if path.extension() == Some(std::ffi::OsStr::new("json")) {
75                match Self::from_file(&path) {
76                    Ok(event) => events.push(event),
77                    Err(e) => eprintln!("Failed to parse staging event {}: {}", path.display(), e),
78                }
79            }
80        }
81
82        events.sort_by_key(|e| e.timestamp);
83        Ok(events)
84    }
85
86    /// Remove the staging file for this event.
87    pub fn remove_file(&self, staging_dir: &Path) -> Result<(), std::io::Error> {
88        let filename = format!("{}.json", self.id);
89        let path = staging_dir.join(filename);
90        fs::remove_file(path)
91    }
92}
mirror_log/stage.rs

mirror_log/
stage.rs
