ravenrs_report_server/
store.rs

1//! Keeps layouts, jobs, history and documents as files, so they can be backed
2//! up and read with ordinary tools. Every name is validated before it touches
3//! a path, and writes go through a temporary file so a crash leaves no half file.
4
5use std::fs;
6use std::io;
7use std::path::{Path, PathBuf};
8use std::sync::Arc;
9use std::sync::atomic::{AtomicU64, Ordering};
10
11use jiff::Timestamp;
12use serde::de::DeserializeOwned;
13use serde::{Deserialize, Serialize};
14
15use crate::job::Job;
16
17/// Whether `name` is safe as a single path component: letters, digits, `.`,
18/// `_` and `-`, starting with a letter or digit.
19pub fn valid_name(name: &str) -> bool {
20    let mut chars = name.chars();
21    name.len() <= 64
22        && chars.next().is_some_and(|c| c.is_ascii_alphanumeric())
23        && chars.all(|c| c.is_ascii_alphanumeric() || matches!(c, '.' | '_' | '-'))
24}
25
26/// Whether a file name is safe to serve.
27fn valid_document_name(name: &str) -> bool {
28    !name.is_empty()
29        && name.len() <= 255
30        && !name.contains(['/', '\\', '\0'])
31        && !name.starts_with('.')
32        && name.chars().all(|c| !c.is_control())
33}
34
35/// What happened in a run.
36#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
37#[serde(rename_all = "camelCase")]
38pub enum RunStatus {
39    /// Everything was delivered.
40    Success,
41    /// The report had no rows and the job skips empty reports.
42    SkippedEmpty,
43    /// Some deliveries worked and others did not, even after the retries.
44    Partial,
45    /// Nothing was delivered.
46    Failed,
47    /// The run was too late to happen (the server was down), or the previous one was still running.
48    Missed,
49}
50
51/// A delivery made in a run: who received what.
52#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
53#[serde(rename_all = "camelCase")]
54pub struct Delivered {
55    /// `email` or `folder`.
56    pub kind: String,
57    /// The recipients, or the folder.
58    pub target: String,
59    /// The files.
60    pub files: Vec<String>,
61    /// When it happened.
62    pub at: Timestamp,
63}
64
65/// One run of a job, as the history keeps it.
66#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
67#[serde(rename_all = "camelCase")]
68pub struct RunRecord {
69    /// Identifies the run within its job; later runs sort after earlier ones.
70    pub id: String,
71    /// The job.
72    pub job: String,
73    /// The layout version rendered.
74    pub layout_version: Option<u32>,
75    /// When the run was due.
76    pub scheduled_for: Timestamp,
77    /// When it started.
78    pub started_at: Timestamp,
79    /// When it ended.
80    pub finished_at: Timestamp,
81    /// Whether it was started by hand.
82    pub manual: bool,
83    /// The outcome.
84    pub status: RunStatus,
85    /// How many times it was attempted.
86    pub attempts: u32,
87    /// How many rows the report printed.
88    pub rows: Option<usize>,
89    /// What was delivered.
90    pub delivered: Vec<Delivered>,
91    /// Why deliveries or the render failed, in the last attempt.
92    pub errors: Vec<String>,
93    /// Problems in the report itself (expression errors, missing fields). The document is still produced.
94    pub report_warnings: Vec<String>,
95    /// The files kept on the server.
96    pub documents: Vec<String>,
97}
98
99/// When a job runs next. Reset when the job changes.
100#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
101#[serde(rename_all = "camelCase")]
102pub struct JobState {
103    /// The next run; `None` when the schedule has none left or it was not computed yet.
104    pub next_run: Option<Timestamp>,
105}
106
107/// A stored layout version.
108#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
109#[serde(rename_all = "camelCase")]
110pub struct LayoutVersion {
111    /// The version number, counting from 1.
112    pub version: u32,
113    /// Its size in bytes.
114    pub size: u64,
115}
116
117/// The server's files. Clones share the same folder.
118#[derive(Clone, Debug)]
119pub struct Store {
120    root: Arc<PathBuf>,
121    /// Makes temporary files and run ids unique within this process.
122    counter: Arc<AtomicU64>,
123}
124
125impl Store {
126    /// Opens the store, creating its folders.
127    pub fn open(root: &Path) -> io::Result<Self> {
128        for folder in ["layouts", "jobs", "runs", "documents", "tmp"] {
129            fs::create_dir_all(root.join(folder))?;
130        }
131        Ok(Self { root: Arc::new(root.to_path_buf()), counter: Arc::new(AtomicU64::new(0)) })
132    }
133
134    fn write_atomic(&self, path: &Path, bytes: &[u8]) -> io::Result<()> {
135        if let Some(parent) = path.parent() {
136            fs::create_dir_all(parent)?;
137        }
138        let n = self.counter.fetch_add(1, Ordering::Relaxed);
139        let temporary = self.root.join("tmp").join(format!("{}-{n}", std::process::id()));
140        fs::write(&temporary, bytes)?;
141        fs::rename(&temporary, path)
142    }
143
144    fn write_json<T: Serialize>(&self, path: &Path, value: &T) -> io::Result<()> {
145        let bytes = serde_json::to_vec_pretty(value).map_err(io::Error::other)?;
146        self.write_atomic(path, &bytes)
147    }
148
149    fn read_json<T: DeserializeOwned>(path: &Path) -> io::Result<Option<T>> {
150        match fs::read(path) {
151            Ok(bytes) => serde_json::from_slice(&bytes).map(Some).map_err(io::Error::other),
152            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(None),
153            Err(e) => Err(e),
154        }
155    }
156
157    fn check(name: &str) -> io::Result<()> {
158        if valid_name(name) {
159            Ok(())
160        } else {
161            Err(io::Error::new(io::ErrorKind::InvalidInput, format!("'{name}' is not a valid name")))
162        }
163    }
164
165    /// The names in a folder that are valid names, sorted.
166    fn names_in(folder: &Path) -> io::Result<Vec<String>> {
167        let mut names = Vec::new();
168        let entries = match fs::read_dir(folder) {
169            Ok(entries) => entries,
170            Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(names),
171            Err(e) => return Err(e),
172        };
173        for entry in entries {
174            if let Some(name) = entry?.file_name().to_str() {
175                names.push(name.to_owned());
176            }
177        }
178        names.sort();
179        Ok(names)
180    }
181
182    fn layout_dir(&self, name: &str) -> PathBuf {
183        self.root.join("layouts").join(name)
184    }
185
186    /// The names of the stored layouts.
187    pub fn layouts(&self) -> io::Result<Vec<String>> {
188        Ok(Self::names_in(&self.root.join("layouts"))?.into_iter().filter(|n| valid_name(n)).collect())
189    }
190
191    /// A layout's versions, oldest first.
192    pub fn layout_versions(&self, name: &str) -> io::Result<Vec<LayoutVersion>> {
193        Self::check(name)?;
194        let dir = self.layout_dir(name);
195        let mut versions = Vec::new();
196        for file in Self::names_in(&dir)? {
197            if let Some(version) = file.strip_suffix(".rvreport.json").and_then(|v| v.parse().ok()) {
198                versions.push(LayoutVersion { version, size: fs::metadata(dir.join(&file))?.len() });
199            }
200        }
201        versions.sort_by_key(|v| v.version);
202        Ok(versions)
203    }
204
205    /// Saves a layout. Saving what the latest version already holds adds no
206    /// version. Returns the version that holds it.
207    pub fn save_layout(&self, name: &str, json: &str) -> io::Result<u32> {
208        Self::check(name)?;
209        if let Some((latest, current)) = self.layout(name, None)?
210            && current == json
211        {
212            return Ok(latest);
213        }
214        let next = self.layout_versions(name)?.last().map_or(1, |v| v.version + 1);
215        self.write_atomic(&self.layout_dir(name).join(format!("{next}.rvreport.json")), json.as_bytes())?;
216        Ok(next)
217    }
218
219    /// A layout's text with its version: the latest, or the one asked for.
220    pub fn layout(&self, name: &str, version: Option<u32>) -> io::Result<Option<(u32, String)>> {
221        Self::check(name)?;
222        let version = match version {
223            Some(version) => version,
224            None => match self.layout_versions(name)?.last() {
225                Some(latest) => latest.version,
226                None => return Ok(None),
227            },
228        };
229        match fs::read_to_string(self.layout_dir(name).join(format!("{version}.rvreport.json"))) {
230            Ok(text) => Ok(Some((version, text))),
231            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(None),
232            Err(e) => Err(e),
233        }
234    }
235
236    /// Deletes a layout and all its versions. Returns whether it existed.
237    pub fn delete_layout(&self, name: &str) -> io::Result<bool> {
238        Self::check(name)?;
239        let dir = self.layout_dir(name);
240        if !dir.exists() {
241            return Ok(false);
242        }
243        fs::remove_dir_all(dir)?;
244        Ok(true)
245    }
246
247    fn job_path(&self, id: &str) -> PathBuf {
248        self.root.join("jobs").join(format!("{id}.job.json"))
249    }
250
251    fn state_path(&self, id: &str) -> PathBuf {
252        self.root.join("jobs").join(format!("{id}.state.json"))
253    }
254
255    /// The ids of the stored jobs.
256    pub fn job_ids(&self) -> io::Result<Vec<String>> {
257        Ok(Self::names_in(&self.root.join("jobs"))?
258            .iter()
259            .filter_map(|file| file.strip_suffix(".job.json"))
260            .filter(|id| valid_name(id))
261            .map(str::to_owned)
262            .collect())
263    }
264
265    /// Saves a job. Its next run is computed again.
266    pub fn save_job(&self, id: &str, job: &Job) -> io::Result<()> {
267        Self::check(id)?;
268        self.write_json(&self.job_path(id), job)?;
269        match fs::remove_file(self.state_path(id)) {
270            Err(e) if e.kind() != io::ErrorKind::NotFound => Err(e),
271            _ => Ok(()),
272        }
273    }
274
275    /// A job by id.
276    pub fn job(&self, id: &str) -> io::Result<Option<Job>> {
277        Self::check(id)?;
278        Self::read_json(&self.job_path(id))
279    }
280
281    /// Deletes a job, with its history and kept documents. Returns whether it existed.
282    pub fn delete_job(&self, id: &str) -> io::Result<bool> {
283        Self::check(id)?;
284        let existed = self.job_path(id).exists();
285        for file in [self.job_path(id), self.state_path(id)] {
286            match fs::remove_file(file) {
287                Err(e) if e.kind() != io::ErrorKind::NotFound => return Err(e),
288                _ => {}
289            }
290        }
291        for folder in ["runs", "documents"] {
292            match fs::remove_dir_all(self.root.join(folder).join(id)) {
293                Err(e) if e.kind() != io::ErrorKind::NotFound => return Err(e),
294                _ => {}
295            }
296        }
297        Ok(existed)
298    }
299
300    /// A job's schedule state.
301    pub fn job_state(&self, id: &str) -> io::Result<JobState> {
302        Self::check(id)?;
303        Ok(Self::read_json(&self.state_path(id))?.unwrap_or_default())
304    }
305
306    /// Saves a job's schedule state.
307    pub fn save_job_state(&self, id: &str, state: &JobState) -> io::Result<()> {
308        Self::check(id)?;
309        self.write_json(&self.state_path(id), state)
310    }
311
312    /// A new run id, later than the ones before it.
313    pub fn new_run_id(&self, now: Timestamp) -> String {
314        let n = self.counter.fetch_add(1, Ordering::Relaxed) % 1_000_000;
315        format!("{}-{n:06}", now.strftime("%Y%m%dT%H%M%S"))
316    }
317
318    /// Adds a run to a job's history, dropping the oldest past `limit`.
319    pub fn save_run(&self, run: &RunRecord, limit: usize) -> io::Result<()> {
320        Self::check(&run.job)?;
321        Self::check(&run.id)?;
322        let dir = self.root.join("runs").join(&run.job);
323        self.write_json(&dir.join(format!("{}.json", run.id)), run)?;
324        let files = Self::names_in(&dir)?;
325        for old in files.iter().take(files.len().saturating_sub(limit)) {
326            fs::remove_file(dir.join(old))?;
327        }
328        Ok(())
329    }
330
331    /// A job's runs, newest first, at most `limit`.
332    pub fn runs(&self, job: &str, limit: usize) -> io::Result<Vec<RunRecord>> {
333        Self::check(job)?;
334        let dir = self.root.join("runs").join(job);
335        let mut runs = Vec::new();
336        for file in Self::names_in(&dir)?.iter().rev().take(limit) {
337            if let Some(run) = Self::read_json(&dir.join(file))? {
338                runs.push(run);
339            }
340        }
341        Ok(runs)
342    }
343
344    /// One run of a job.
345    pub fn run(&self, job: &str, id: &str) -> io::Result<Option<RunRecord>> {
346        Self::check(job)?;
347        Self::check(id)?;
348        Self::read_json(&self.root.join("runs").join(job).join(format!("{id}.json")))
349    }
350
351    /// Keeps a run's file on the server.
352    pub fn save_document(&self, job: &str, run: &str, file: &str, bytes: &[u8]) -> io::Result<()> {
353        Self::check(job)?;
354        Self::check(run)?;
355        if !valid_document_name(file) {
356            return Err(io::Error::new(io::ErrorKind::InvalidInput, "invalid file name"));
357        }
358        self.write_atomic(&self.root.join("documents").join(job).join(run).join(file), bytes)
359    }
360
361    /// A kept file.
362    pub fn document(&self, job: &str, run: &str, file: &str) -> io::Result<Option<Vec<u8>>> {
363        Self::check(job)?;
364        Self::check(run)?;
365        if !valid_document_name(file) {
366            return Ok(None);
367        }
368        match fs::read(self.root.join("documents").join(job).join(run).join(file)) {
369            Ok(bytes) => Ok(Some(bytes)),
370            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(None),
371            Err(e) => Err(e),
372        }
373    }
374
375    /// Deletes a job's kept files older than `cutoff`. Returns how many runs lost theirs.
376    pub fn purge_documents(&self, job: &str, cutoff: Timestamp) -> io::Result<usize> {
377        Self::check(job)?;
378        let dir = self.root.join("documents").join(job);
379        let mut purged = 0;
380        for run in Self::names_in(&dir)? {
381            let Ok(record) =
382                Self::read_json::<RunRecord>(&self.root.join("runs").join(job).join(format!("{run}.json")))
383            else {
384                continue;
385            };
386            // A run the history no longer has is older than anything it keeps.
387            if record.is_none_or(|r| r.started_at < cutoff) {
388                fs::remove_dir_all(dir.join(run))?;
389                purged += 1;
390            }
391        }
392        Ok(purged)
393    }
394}
395
396/// A plain record of what a job's file listing returns over the API.
397#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
398#[serde(rename_all = "camelCase")]
399pub struct JobSummary {
400    /// The id.
401    pub id: String,
402    /// The job.
403    pub job: Job,
404    /// Its next run, when known.
405    pub next_run: Option<Timestamp>,
406}
407
408#[cfg(test)]
409mod tests {
410    use super::*;
411
412    fn open() -> Result<(tempfile::TempDir, Store), io::Error> {
413        let dir = tempfile::tempdir()?;
414        let store = Store::open(dir.path())?;
415        Ok((dir, store))
416    }
417
418    #[test]
419    fn names_that_could_leave_a_folder_are_refused() {
420        for bad in ["", "..", ".", "../x", "a/b", "a\\b", ".hidden", "a\0", "a b", &"x".repeat(65)] {
421            assert!(!valid_name(bad), "{bad:?}");
422        }
423        for good in ["a", "sales", "Sales-2026_v1.2", "20261101T120000-000001"] {
424            assert!(valid_name(good), "{good}");
425        }
426    }
427
428    #[test]
429    fn every_entry_point_refuses_a_bad_name() -> io::Result<()> {
430        let (_dir, store) = open()?;
431        assert!(store.save_layout("../x", "{}").is_err());
432        assert!(store.layout("../x", None).is_err());
433        assert!(store.delete_layout("..").is_err());
434        assert!(store.layout_versions("../..").is_err());
435        assert!(store.job("../x").is_err());
436        assert!(store.delete_job("../x").is_err());
437        assert!(store.runs("../x", 1).is_err());
438        assert!(store.run("x", "../y").is_err());
439        assert!(store.document("x", "../y", "z").is_err());
440        assert_eq!(store.document("x", "y", "../../secret")?, None);
441        assert!(store.save_document("x", "y", "../z", b"").is_err());
442        Ok(())
443    }
444
445    #[test]
446    fn layouts_keep_versions_and_skip_identical_saves() -> io::Result<()> {
447        let (_dir, store) = open()?;
448        assert_eq!(store.layout("a", None)?, None);
449        assert_eq!(store.save_layout("a", "one")?, 1);
450        assert_eq!(store.save_layout("a", "one")?, 1);
451        assert_eq!(store.save_layout("a", "two")?, 2);
452        assert_eq!(store.layout("a", None)?, Some((2, "two".to_owned())));
453        assert_eq!(store.layout("a", Some(1))?, Some((1, "one".to_owned())));
454        assert_eq!(store.layout("a", Some(9))?, None);
455        assert_eq!(store.layouts()?, ["a"]);
456        assert_eq!(store.layout_versions("a")?.len(), 2);
457        assert!(store.delete_layout("a")?);
458        assert!(!store.delete_layout("a")?);
459        Ok(())
460    }
461
462    #[test]
463    fn version_ten_comes_after_nine() -> io::Result<()> {
464        let (_dir, store) = open()?;
465        for n in 1..=11 {
466            store.save_layout("a", &n.to_string())?;
467        }
468        assert_eq!(store.layout("a", None)?, Some((11, "11".to_owned())));
469        Ok(())
470    }
471}
ravenrs_report_server/store.rs

ravenrs_report_server/
store.rs
