redevplugin_worker_sdk/
fs.rs

1use crate::api;
2use crate::error::{Error, ErrorCode, Result};
3use crate::resource::{Handle, IO_FLAG_EOF, MAX_IO_CHUNK_BYTES};
4use serde::{Deserialize, Serialize};
5use std::sync::atomic::{AtomicU64, Ordering};
6
7static TEMP_SEQUENCE: AtomicU64 = AtomicU64::new(1);
8const ATOMIC_TEMP_ATTEMPTS: usize = 16;
9
10#[derive(Debug, Clone, Default, Serialize)]
11pub struct OpenOptions {
12    pub read: bool,
13    pub write: bool,
14    pub create: bool,
15    pub create_new: bool,
16    pub truncate: bool,
17    pub append: bool,
18    #[serde(skip_serializing_if = "Option::is_none")]
19    pub mode: Option<u32>,
20}
21
22#[derive(Debug, Clone, Deserialize)]
23pub struct FileStat {
24    pub uri: String,
25    pub kind: FileKind,
26    pub size: u64,
27    pub mode: u32,
28    pub modified_unix_ms: i64,
29    #[serde(default)]
30    pub created_unix_ms: Option<i64>,
31}
32
33#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
34#[serde(rename_all = "snake_case")]
35pub enum FileKind {
36    File,
37    Directory,
38    Symlink,
39    Other,
40    #[serde(other)]
41    Unknown,
42}
43
44#[derive(Debug, Clone, Deserialize)]
45pub struct DirectoryEntry {
46    pub name: String,
47    pub uri: String,
48    pub kind: FileKind,
49}
50
51#[derive(Debug, Clone, Deserialize)]
52pub struct DirectoryPage {
53    #[serde(default)]
54    pub entries: Vec<DirectoryEntry>,
55    pub eof: bool,
56}
57
58#[derive(Debug, Clone, Deserialize)]
59pub struct WatchEvent {
60    pub sequence: u64,
61    pub kind: WatchKind,
62    pub uri: String,
63    #[serde(default)]
64    pub previous_uri: Option<String>,
65}
66
67#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
68#[serde(rename_all = "snake_case")]
69pub enum WatchKind {
70    Create,
71    Change,
72    Delete,
73    Rename,
74    Overflow,
75    #[serde(other)]
76    Unknown,
77}
78
79#[derive(Debug, Clone, Deserialize)]
80pub struct MountInfo {
81    pub id: String,
82    pub uri: String,
83    pub read_only: bool,
84}
85
86#[derive(Debug, Clone, Deserialize)]
87pub struct Mounts {
88    #[serde(default)]
89    pub mounts: Vec<MountInfo>,
90}
91
92#[derive(Deserialize)]
93struct HandleResult {
94    handle: u64,
95}
96
97#[derive(Serialize)]
98struct URIArguments<'a> {
99    uri: &'a str,
100}
101
102pub struct File {
103    handle: Handle,
104}
105
106impl File {
107    pub fn open(uri: &str, options: OpenOptions) -> Result<Self> {
108        #[derive(Serialize)]
109        struct Arguments<'a> {
110            uri: &'a str,
111            options: OpenOptions,
112        }
113        let opened: HandleResult = api::call("fs.open", &Arguments { uri, options })?;
114        Ok(Self {
115            handle: Handle::new(opened.handle)?,
116        })
117    }
118
119    pub fn read(&mut self, capacity: usize) -> Result<(Vec<u8>, u32)> {
120        self.handle.read(capacity)
121    }
122
123    pub fn write_all(&mut self, bytes: &[u8]) -> Result<()> {
124        for chunk in bytes.chunks(MAX_IO_CHUNK_BYTES) {
125            self.handle.write(chunk, 0)?;
126        }
127        Ok(())
128    }
129
130    pub fn seek(&mut self, offset: i64, whence: u32) -> Result<u64> {
131        self.handle.seek(offset, whence)
132    }
133
134    pub fn sync(&mut self) -> Result<()> {
135        #[derive(Serialize)]
136        struct Arguments {
137            handle: u64,
138        }
139        let _: serde_json::Value = api::call(
140            "fs.sync",
141            &Arguments {
142                handle: self.handle.id(),
143            },
144        )?;
145        Ok(())
146    }
147
148    pub fn close(mut self) -> Result<()> {
149        self.handle.close()
150    }
151}
152
153pub struct Directory {
154    handle: Handle,
155}
156
157impl Directory {
158    pub fn open(uri: &str) -> Result<Self> {
159        let opened: HandleResult = api::call("fs.read_dir.open", &URIArguments { uri })?;
160        Ok(Self {
161            handle: Handle::new(opened.handle)?,
162        })
163    }
164
165    pub fn next(&mut self, limit: u16) -> Result<DirectoryPage> {
166        #[derive(Serialize)]
167        struct Arguments {
168            handle: u64,
169            limit: u16,
170        }
171        api::call(
172            "fs.read_dir.next",
173            &Arguments {
174                handle: self.handle.id(),
175                limit,
176            },
177        )
178    }
179
180    pub fn close(mut self) -> Result<()> {
181        self.handle.close()
182    }
183}
184
185pub struct Watch {
186    handle: Handle,
187}
188
189impl Watch {
190    pub fn open(uri: &str) -> Result<Self> {
191        #[derive(Serialize)]
192        struct Arguments<'a> {
193            uri: &'a str,
194            recursive: bool,
195        }
196        let opened: HandleResult = api::call(
197            "fs.watch",
198            &Arguments {
199                uri,
200                recursive: false,
201            },
202        )?;
203        Ok(Self {
204            handle: Handle::new(opened.handle)?,
205        })
206    }
207
208    pub fn next(&mut self, timeout_ms: u32) -> Result<WatchEvent> {
209        #[derive(Serialize)]
210        struct Arguments {
211            handle: u64,
212            timeout_ms: u32,
213        }
214        api::call(
215            "fs.watch_next",
216            &Arguments {
217                handle: self.handle.id(),
218                timeout_ms,
219            },
220        )
221    }
222
223    pub fn close(mut self) -> Result<()> {
224        self.handle.close()
225    }
226}
227
228pub fn mounts() -> Result<Mounts> {
229    api::call("fs.mounts", &serde_json::json!({}))
230}
231
232pub fn stat(uri: &str, follow_symlinks: bool) -> Result<FileStat> {
233    #[derive(Serialize)]
234    struct Arguments<'a> {
235        uri: &'a str,
236        follow_symlinks: bool,
237    }
238    api::call(
239        "fs.stat",
240        &Arguments {
241            uri,
242            follow_symlinks,
243        },
244    )
245}
246
247pub fn read_file(uri: &str) -> Result<Vec<u8>> {
248    let mut file = File::open(
249        uri,
250        OpenOptions {
251            read: true,
252            ..OpenOptions::default()
253        },
254    )?;
255    let mut result = Vec::new();
256    loop {
257        let (chunk, flags) = file.read(MAX_IO_CHUNK_BYTES)?;
258        if chunk.is_empty() && flags & IO_FLAG_EOF == 0 {
259            return Err(Error::internal("file read made no progress"));
260        }
261        result.extend_from_slice(&chunk);
262        if flags & IO_FLAG_EOF != 0 {
263            file.close()?;
264            return Ok(result);
265        }
266    }
267}
268
269pub fn read_text(uri: &str) -> Result<String> {
270    String::from_utf8(read_file(uri)?)
271        .map_err(|_| Error::internal("file content is not valid UTF-8"))
272}
273
274pub fn write_file(uri: &str, bytes: &[u8], atomic: bool) -> Result<()> {
275    if !atomic {
276        return write_direct(uri, bytes, false);
277    }
278    for _ in 0..ATOMIC_TEMP_ATTEMPTS {
279        let temporary = atomic_temporary_uri(uri)?;
280        match write_direct(&temporary, bytes, true) {
281            Ok(()) => {
282                let result = rename(&temporary, uri, true);
283                if result.is_err() {
284                    let _ = remove(&temporary, false);
285                }
286                return result;
287            }
288            Err(error) if error.code == ErrorCode::AlreadyExists => continue,
289            Err(error) => {
290                let _ = remove(&temporary, false);
291                return Err(error);
292            }
293        }
294    }
295    Err(Error::internal(
296        "could not allocate a unique same-directory atomic write file",
297    ))
298}
299
300pub fn write_text(uri: &str, text: &str, atomic: bool) -> Result<()> {
301    write_file(uri, text.as_bytes(), atomic)
302}
303
304fn write_direct(uri: &str, bytes: &[u8], create_new: bool) -> Result<()> {
305    let mut file = File::open(
306        uri,
307        OpenOptions {
308            write: true,
309            create: !create_new,
310            create_new,
311            truncate: !create_new,
312            mode: Some(0o600),
313            ..OpenOptions::default()
314        },
315    )?;
316    file.write_all(bytes)?;
317    file.sync()?;
318    file.close()
319}
320
321pub fn remove(uri: &str, recursive: bool) -> Result<()> {
322    #[derive(Serialize)]
323    struct Arguments<'a> {
324        uri: &'a str,
325        recursive: bool,
326    }
327    let _: serde_json::Value = api::call("fs.remove", &Arguments { uri, recursive })?;
328    Ok(())
329}
330
331pub fn mkdir(uri: &str, recursive: bool, mode: u32) -> Result<()> {
332    #[derive(Serialize)]
333    struct Arguments<'a> {
334        uri: &'a str,
335        recursive: bool,
336        mode: u32,
337    }
338    let _: serde_json::Value = api::call(
339        "fs.mkdir",
340        &Arguments {
341            uri,
342            recursive,
343            mode,
344        },
345    )?;
346    Ok(())
347}
348
349pub fn rename(from: &str, to: &str, overwrite: bool) -> Result<()> {
350    transfer("fs.rename", from, to, overwrite)
351}
352
353pub fn copy(from: &str, to: &str, overwrite: bool) -> Result<()> {
354    transfer("fs.copy", from, to, overwrite)
355}
356
357fn transfer(operation: &str, from: &str, to: &str, overwrite: bool) -> Result<()> {
358    #[derive(Serialize)]
359    struct Arguments<'a> {
360        from: &'a str,
361        to: &'a str,
362        overwrite: bool,
363    }
364    let _: serde_json::Value = api::call(
365        operation,
366        &Arguments {
367            from,
368            to,
369            overwrite,
370        },
371    )?;
372    Ok(())
373}
374
375pub fn set_times(uri: &str, accessed_unix_ms: i64, modified_unix_ms: i64) -> Result<()> {
376    #[derive(Serialize)]
377    struct Arguments<'a> {
378        uri: &'a str,
379        accessed_unix_ms: i64,
380        modified_unix_ms: i64,
381    }
382    let _: serde_json::Value = api::call(
383        "fs.set_times",
384        &Arguments {
385            uri,
386            accessed_unix_ms,
387            modified_unix_ms,
388        },
389    )?;
390    Ok(())
391}
392
393fn atomic_temporary_uri(uri: &str) -> Result<String> {
394    let (directory, name) = uri
395        .rsplit_once('/')
396        .filter(|(_, name)| !name.is_empty())
397        .ok_or_else(|| Error::from_abi_status(-1))?;
398    let sequence = TEMP_SEQUENCE.fetch_add(1, Ordering::Relaxed);
399    Ok(format!(
400        "{directory}/.{name}.redevplugin-{sequence:016x}.tmp"
401    ))
402}
403
404#[cfg(test)]
405mod tests {
406    use super::*;
407
408    #[test]
409    fn atomic_temporary_file_stays_in_the_same_directory() {
410        let temporary = atomic_temporary_uri("redevfs://workspace/src/data.bin").unwrap();
411        assert!(temporary.starts_with("redevfs://workspace/src/.data.bin.redevplugin-"));
412        assert!(temporary.ends_with(".tmp"));
413    }
414}
redevplugin_worker_sdk/fs.rs

redevplugin_worker_sdk/
fs.rs
