file_engine/
lib.rs

1mod error;
2#[cfg(feature = "operations")]
3mod handle;
4// `sync`/`compress` already imply `operations` via Cargo.toml, but
5// `watch` deliberately doesn't (it never touches the
6// Profiler/Planner/Dispatcher pipeline) — this needs its own condition
7// rather than reusing the `operations` feature alone, or `watch`-only
8// builds fail to find this module at all.
9#[cfg(any(feature = "operations", feature = "watch"))]
10mod operations;
11// Also intended for `sync`'s `diff.rs` once that's wired up to use it
12// too (dev-docs/design/filesystem-detection.md, item 6) — currently only
13// called from `profiler::validate`.
14#[cfg(feature = "operations")]
15mod paths;
16#[cfg(feature = "operations")]
17mod planner;
18#[cfg(feature = "operations")]
19mod profiler;
20#[cfg(feature = "operations")]
21mod progress;
22#[cfg(feature = "watch")]
23mod watch_event;
24#[cfg(feature = "watch")]
25mod watch_handle;
26
27pub use error::{Error, Result};
28#[cfg(feature = "operations")]
29pub use handle::Handle;
30#[cfg(feature = "operations")]
31pub use progress::Progress;
32#[cfg(feature = "watch")]
33pub use watch_event::{WatchEvent, WatchEventKind};
34#[cfg(feature = "watch")]
35pub use watch_handle::WatchHandle;
36
37#[cfg(feature = "operations")]
38pub use operations::CopyBuilder;
39#[cfg(feature = "compress")]
40pub use operations::{CompressBuilder, CompressFormat};
41#[cfg(feature = "operations")]
42pub use operations::MoveBuilder;
43#[cfg(feature = "sync")]
44pub use operations::{SyncBuilder, SyncOutcome};
45#[cfg(feature = "watch")]
46pub use operations::WatchBuilder;
47
48pub struct FileEngine;
49
50impl Default for FileEngine {
51    fn default() -> Self {
52        Self::new()
53    }
54}
55
56impl FileEngine {
57    pub fn new() -> Self {
58        FileEngine
59    }
60
61    #[cfg(feature = "operations")]
62    pub fn copy(
63        &self,
64        source: impl Into<std::path::PathBuf>,
65        dest: impl Into<std::path::PathBuf>,
66    ) -> CopyBuilder {
67        CopyBuilder::new(source, dest)
68    }
69
70    #[cfg(feature = "operations")]
71    pub fn move_path(
72        &self,
73        source: impl Into<std::path::PathBuf>,
74        dest: impl Into<std::path::PathBuf>,
75    ) -> MoveBuilder {
76        MoveBuilder::new(source, dest)
77    }
78
79    #[cfg(feature = "watch")]
80    pub fn watch(&self, path: impl Into<std::path::PathBuf>) -> WatchBuilder {
81        WatchBuilder::new(path)
82    }
83
84    #[cfg(feature = "sync")]
85    pub fn sync(
86        &self,
87        source: impl Into<std::path::PathBuf>,
88        dest: impl Into<std::path::PathBuf>,
89    ) -> SyncBuilder {
90        SyncBuilder::new(source, dest)
91    }
92
93    #[cfg(feature = "compress")]
94    pub fn compress(
95        &self,
96        source: impl Into<std::path::PathBuf>,
97        dest: impl Into<std::path::PathBuf>,
98    ) -> CompressBuilder {
99        CompressBuilder::new(source, dest)
100    }
101}
102
103#[cfg(all(test, feature = "operations", feature = "sync", feature = "compress"))]
104mod tests {
105    use std::fs;
106
107    use tempfile::tempdir;
108    use tokio_stream::StreamExt;
109
110    use super::*;
111
112    /// Confirms the event sequence contract from
113    /// dev-docs/design/handle-progress.md's `.start()` test list: a `Started`
114    /// before any `EntryStarted`, its `entries_total` matching what
115    /// actually ran, and one terminal event (`EntryCompleted` or
116    /// `EntryFailed`) per entry.
117    fn assert_well_formed(events: &[Progress], expected_entries: usize) {
118        let started_at = events.iter().position(|e| matches!(e, Progress::Started { .. }));
119        assert!(started_at.is_some(), "expected a Started event");
120
121        if let Some(first_entry_started) = events.iter().position(|e| matches!(e, Progress::EntryStarted { .. })) {
122            assert!(started_at.unwrap() < first_entry_started, "Started must come before any EntryStarted");
123        }
124
125        let entries_total = events
126            .iter()
127            .find_map(|e| match e {
128                Progress::Started { entries_total, .. } => Some(*entries_total),
129                _ => None,
130            })
131            .unwrap();
132        assert_eq!(entries_total, expected_entries);
133
134        let terminal_count = events
135            .iter()
136            .filter(|e| matches!(e, Progress::EntryCompleted { .. } | Progress::EntryFailed { .. }))
137            .count();
138        assert_eq!(terminal_count, expected_entries);
139    }
140
141    #[tokio::test]
142    async fn copy_end_to_end_through_the_public_api() {
143        let src_dir = tempdir().unwrap();
144        let dest_dir = tempdir().unwrap();
145        fs::write(src_dir.path().join("a.txt"), b"hello").unwrap();
146
147        let engine = FileEngine::new();
148        let mut handle = engine.copy(src_dir.path(), dest_dir.path()).start().unwrap();
149
150        let mut events = Vec::new();
151        while let Some(event) = handle.progress().next().await {
152            events.push(event);
153        }
154
155        let outcome = handle.await.unwrap();
156
157        assert_eq!(outcome.succeeded.len(), 1);
158        assert_eq!(fs::read(dest_dir.path().join("a.txt")).unwrap(), b"hello");
159        assert_well_formed(&events, 1);
160    }
161
162    #[tokio::test]
163    async fn move_end_to_end_through_the_public_api() {
164        // Both paths under one tempdir, guaranteeing the same filesystem
165        // so this exercises the atomic-rename fast path (matches
166        // move_path.rs's own same-filesystem test).
167        let root = tempdir().unwrap();
168        let src_file = root.path().join("a.txt");
169        let dest_file = root.path().join("dst.txt");
170        fs::write(&src_file, b"hello").unwrap();
171
172        let engine = FileEngine::new();
173        let handle = engine.move_path(root.path().join("a.txt"), dest_file.clone()).start().unwrap();
174        let outcome = handle.await.unwrap();
175
176        // Fast path enumerates no entries — matches move_path.rs's tests.
177        assert!(outcome.succeeded.is_empty());
178        assert!(!src_file.exists());
179        assert_eq!(fs::read(&dest_file).unwrap(), b"hello");
180    }
181
182    #[tokio::test]
183    async fn sync_end_to_end_through_the_public_api() {
184        let src_dir = tempdir().unwrap();
185        let dest_dir = tempdir().unwrap();
186        fs::write(src_dir.path().join("new.txt"), b"new").unwrap();
187        fs::write(dest_dir.path().join("orphan.txt"), b"stale").unwrap();
188
189        let engine = FileEngine::new();
190        let handle = engine.sync(src_dir.path(), dest_dir.path()).start().unwrap();
191        let outcome = handle.await.unwrap();
192
193        assert_eq!(outcome.copy.succeeded.len(), 1);
194        assert_eq!(outcome.delete.succeeded.len(), 1);
195        assert_eq!(fs::read(dest_dir.path().join("new.txt")).unwrap(), b"new");
196        assert!(!dest_dir.path().join("orphan.txt").exists());
197    }
198
199    #[tokio::test]
200    async fn compress_end_to_end_through_the_public_api() {
201        let src_dir = tempdir().unwrap();
202        let out_dir = tempdir().unwrap();
203        fs::write(src_dir.path().join("a.txt"), b"a").unwrap();
204        let dest = out_dir.path().join("archive.zip");
205
206        let engine = FileEngine::new();
207        let mut handle = engine.compress(src_dir.path(), &dest).start().unwrap();
208
209        let mut events = Vec::new();
210        while let Some(event) = handle.progress().next().await {
211            events.push(event);
212        }
213
214        let outcome = handle.await.unwrap();
215
216        assert_eq!(outcome.succeeded.len(), 1);
217        assert!(dest.exists());
218        assert_well_formed(&events, 1);
219    }
220}
file_engine/lib.rs

file_engine/
lib.rs
