hermes_core/directories/
directory.rs

1//! Async Directory abstraction for IO operations
2//!
3//! Supports network, local filesystem, and in-memory storage.
4//! All reads are async to minimize blocking on network latency.
5
6use async_trait::async_trait;
7use parking_lot::RwLock;
8use std::collections::HashMap;
9use std::io;
10use std::ops::Range;
11use std::path::{Path, PathBuf};
12use std::sync::Arc;
13
14/// Callback type for lazy range reading
15#[cfg(not(target_arch = "wasm32"))]
16pub type RangeReadFn = Arc<
17    dyn Fn(
18            Range<u64>,
19        )
20            -> std::pin::Pin<Box<dyn std::future::Future<Output = io::Result<OwnedBytes>> + Send>>
21        + Send
22        + Sync,
23>;
24
25#[cfg(target_arch = "wasm32")]
26pub type RangeReadFn = Arc<
27    dyn Fn(
28        Range<u64>,
29    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = io::Result<OwnedBytes>>>>,
30>;
31
32/// Unified file handle for both inline (mmap/RAM) and lazy (HTTP/filesystem) access.
33///
34/// Replaces the previous `FileSlice`, `LazyFileHandle`, and `LazyFileSlice` types.
35/// - **Inline**: data is available synchronously (mmap, RAM). Sync reads via `read_bytes_range_sync`.
36/// - **Lazy**: data is fetched on-demand via async callback (HTTP, filesystem).
37///
38/// Use `.slice()` to create sub-range views (zero-copy for Inline, offset-adjusted for Lazy).
39#[derive(Clone)]
40pub struct FileHandle {
41    inner: FileHandleInner,
42}
43
44#[derive(Clone)]
45enum FileHandleInner {
46    /// Data available inline — sync reads possible (mmap, RAM)
47    Inline {
48        data: OwnedBytes,
49        offset: u64,
50        len: u64,
51    },
52    /// Data fetched on-demand via async callback (HTTP, filesystem)
53    Lazy {
54        read_fn: RangeReadFn,
55        offset: u64,
56        len: u64,
57    },
58}
59
60impl std::fmt::Debug for FileHandle {
61    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
62        match &self.inner {
63            FileHandleInner::Inline { len, offset, .. } => f
64                .debug_struct("FileHandle::Inline")
65                .field("offset", offset)
66                .field("len", len)
67                .finish(),
68            FileHandleInner::Lazy { len, offset, .. } => f
69                .debug_struct("FileHandle::Lazy")
70                .field("offset", offset)
71                .field("len", len)
72                .finish(),
73        }
74    }
75}
76
77impl FileHandle {
78    /// Create an inline file handle from owned bytes (mmap, RAM).
79    /// Sync reads are available.
80    pub fn from_bytes(data: OwnedBytes) -> Self {
81        let len = data.len() as u64;
82        Self {
83            inner: FileHandleInner::Inline {
84                data,
85                offset: 0,
86                len,
87            },
88        }
89    }
90
91    /// Create an empty file handle.
92    pub fn empty() -> Self {
93        Self::from_bytes(OwnedBytes::empty())
94    }
95
96    /// Create a lazy file handle from an async range-read callback.
97    /// Only async reads are available.
98    pub fn lazy(len: u64, read_fn: RangeReadFn) -> Self {
99        Self {
100            inner: FileHandleInner::Lazy {
101                read_fn,
102                offset: 0,
103                len,
104            },
105        }
106    }
107
108    /// Total length in bytes.
109    #[inline]
110    pub fn len(&self) -> u64 {
111        match &self.inner {
112            FileHandleInner::Inline { len, .. } => *len,
113            FileHandleInner::Lazy { len, .. } => *len,
114        }
115    }
116
117    /// Check if empty.
118    #[inline]
119    pub fn is_empty(&self) -> bool {
120        self.len() == 0
121    }
122
123    /// Whether synchronous reads are available (inline/mmap data).
124    #[inline]
125    pub fn is_sync(&self) -> bool {
126        matches!(&self.inner, FileHandleInner::Inline { .. })
127    }
128
129    /// Create a sub-range view. Zero-copy for Inline, offset-adjusted for Lazy.
130    pub fn slice(&self, range: Range<u64>) -> Self {
131        match &self.inner {
132            FileHandleInner::Inline { data, offset, len } => {
133                let new_offset = offset + range.start;
134                let new_len = range.end - range.start;
135                debug_assert!(
136                    new_offset + new_len <= offset + len,
137                    "slice out of bounds: {}+{} > {}+{}",
138                    new_offset,
139                    new_len,
140                    offset,
141                    len
142                );
143                Self {
144                    inner: FileHandleInner::Inline {
145                        data: data.clone(),
146                        offset: new_offset,
147                        len: new_len,
148                    },
149                }
150            }
151            FileHandleInner::Lazy {
152                read_fn,
153                offset,
154                len,
155            } => {
156                let new_offset = offset + range.start;
157                let new_len = range.end - range.start;
158                debug_assert!(
159                    new_offset + new_len <= offset + len,
160                    "slice out of bounds: {}+{} > {}+{}",
161                    new_offset,
162                    new_len,
163                    offset,
164                    len
165                );
166                Self {
167                    inner: FileHandleInner::Lazy {
168                        read_fn: Arc::clone(read_fn),
169                        offset: new_offset,
170                        len: new_len,
171                    },
172                }
173            }
174        }
175    }
176
177    /// Advise the kernel about the access pattern for a byte range of this handle.
178    ///
179    /// Only effective for Inline handles backed by mmap; no-op for Lazy
180    /// handles (HTTP, filesystem callbacks) and heap-backed data.
181    #[cfg(feature = "native")]
182    pub fn madvise_range(&self, range: Range<u64>, advice: libc::c_int) {
183        if let FileHandleInner::Inline { data, offset, len } = &self.inner {
184            let end = range.end.min(*len);
185            if range.start >= end {
186                return;
187            }
188            let start = (*offset + range.start) as usize;
189            let end = (*offset + end) as usize;
190            data.madvise_range(start..end, advice);
191        }
192    }
193
194    /// Async range read — works for both Inline and Lazy.
195    pub async fn read_bytes_range(&self, range: Range<u64>) -> io::Result<OwnedBytes> {
196        match &self.inner {
197            FileHandleInner::Inline { data, offset, len } => {
198                if range.end > *len {
199                    return Err(io::Error::new(
200                        io::ErrorKind::InvalidInput,
201                        format!("Range {:?} out of bounds (len: {})", range, len),
202                    ));
203                }
204                let start = (*offset + range.start) as usize;
205                let end = (*offset + range.end) as usize;
206                Ok(data.slice(start..end))
207            }
208            FileHandleInner::Lazy {
209                read_fn,
210                offset,
211                len,
212            } => {
213                if range.end > *len {
214                    return Err(io::Error::new(
215                        io::ErrorKind::InvalidInput,
216                        format!("Range {:?} out of bounds (len: {})", range, len),
217                    ));
218                }
219                let abs_start = offset + range.start;
220                let abs_end = offset + range.end;
221                (read_fn)(abs_start..abs_end).await
222            }
223        }
224    }
225
226    /// Read all bytes.
227    pub async fn read_bytes(&self) -> io::Result<OwnedBytes> {
228        self.read_bytes_range(0..self.len()).await
229    }
230
231    /// Synchronous range read — only works for Inline handles.
232    /// Returns `Err` if the handle is Lazy.
233    #[inline]
234    pub fn read_bytes_range_sync(&self, range: Range<u64>) -> io::Result<OwnedBytes> {
235        match &self.inner {
236            FileHandleInner::Inline { data, offset, len } => {
237                if range.end > *len {
238                    return Err(io::Error::new(
239                        io::ErrorKind::InvalidInput,
240                        format!("Range {:?} out of bounds (len: {})", range, len),
241                    ));
242                }
243                let start = (*offset + range.start) as usize;
244                let end = (*offset + range.end) as usize;
245                Ok(data.slice(start..end))
246            }
247            FileHandleInner::Lazy { .. } => Err(io::Error::new(
248                io::ErrorKind::Unsupported,
249                "Synchronous read not available on lazy file handle",
250            )),
251        }
252    }
253
254    /// Synchronous read of all bytes — only works for Inline handles.
255    #[inline]
256    pub fn read_bytes_sync(&self) -> io::Result<OwnedBytes> {
257        self.read_bytes_range_sync(0..self.len())
258    }
259}
260
261/// Backing store for OwnedBytes — supports both heap Vec and mmap.
262#[derive(Clone)]
263enum SharedBytes {
264    Vec(Arc<Vec<u8>>),
265    #[cfg(feature = "native")]
266    Mmap(Arc<memmap2::Mmap>),
267}
268
269impl SharedBytes {
270    #[inline]
271    fn as_bytes(&self) -> &[u8] {
272        match self {
273            SharedBytes::Vec(v) => v.as_slice(),
274            #[cfg(feature = "native")]
275            SharedBytes::Mmap(m) => m.as_ref(),
276        }
277    }
278}
279
280impl std::fmt::Debug for SharedBytes {
281    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
282        match self {
283            SharedBytes::Vec(v) => write!(f, "Vec(len={})", v.len()),
284            #[cfg(feature = "native")]
285            SharedBytes::Mmap(m) => write!(f, "Mmap(len={})", m.len()),
286        }
287    }
288}
289
290/// Owned bytes with cheap cloning (Arc-backed)
291///
292/// Supports two backing stores:
293/// - `Vec<u8>` for owned data (RamDirectory, FsDirectory, decompressed blocks)
294/// - `Mmap` for zero-copy memory-mapped files (MmapDirectory, native only)
295#[derive(Debug, Clone)]
296pub struct OwnedBytes {
297    data: SharedBytes,
298    range: Range<usize>,
299}
300
301impl OwnedBytes {
302    pub fn new(data: Vec<u8>) -> Self {
303        let len = data.len();
304        Self {
305            data: SharedBytes::Vec(Arc::new(data)),
306            range: 0..len,
307        }
308    }
309
310    pub fn empty() -> Self {
311        Self {
312            data: SharedBytes::Vec(Arc::new(Vec::new())),
313            range: 0..0,
314        }
315    }
316
317    /// Create from a pre-existing Arc<Vec<u8>> with a sub-range.
318    /// Used by RamDirectory and CachingDirectory to share data without copying.
319    pub(crate) fn from_arc_vec(data: Arc<Vec<u8>>, range: Range<usize>) -> Self {
320        Self {
321            data: SharedBytes::Vec(data),
322            range,
323        }
324    }
325
326    /// Create from a memory-mapped file (zero-copy).
327    #[cfg(feature = "native")]
328    pub(crate) fn from_mmap(mmap: Arc<memmap2::Mmap>) -> Self {
329        let len = mmap.len();
330        Self {
331            data: SharedBytes::Mmap(mmap),
332            range: 0..len,
333        }
334    }
335
336    /// Create from a memory-mapped file with a sub-range (zero-copy).
337    #[cfg(feature = "native")]
338    pub(crate) fn from_mmap_range(mmap: Arc<memmap2::Mmap>, range: Range<usize>) -> Self {
339        Self {
340            data: SharedBytes::Mmap(mmap),
341            range,
342        }
343    }
344
345    pub fn len(&self) -> usize {
346        self.range.len()
347    }
348
349    pub fn is_empty(&self) -> bool {
350        self.range.is_empty()
351    }
352
353    pub fn slice(&self, range: Range<usize>) -> Self {
354        let start = self.range.start + range.start;
355        let end = self.range.start + range.end;
356        Self {
357            data: self.data.clone(),
358            range: start..end,
359        }
360    }
361
362    pub fn as_slice(&self) -> &[u8] {
363        &self.data.as_bytes()[self.range.clone()]
364    }
365
366    /// Returns `true` if the backing store is a memory-mapped file.
367    ///
368    /// Used to guard `madvise` calls: `MADV_DONTNEED` on heap memory
369    /// zeroes pages on Linux and corrupts allocator metadata.
370    #[cfg(feature = "native")]
371    #[inline]
372    pub fn is_mmap(&self) -> bool {
373        matches!(self.data, SharedBytes::Mmap(_))
374    }
375
376    /// Advise the kernel about the access pattern for these bytes.
377    ///
378    /// No-op unless the backing store is mmap (heap memory must never be
379    /// madvised: `MADV_DONTNEED` on heap zeroes pages and corrupts allocator
380    /// metadata) or the range is empty.
381    #[cfg(feature = "native")]
382    pub fn madvise(&self, advice: libc::c_int) {
383        self.madvise_range(0..self.len(), advice);
384    }
385
386    /// Advise the kernel about the access pattern for a sub-range.
387    ///
388    /// The range is relative to these bytes. Same mmap-only guard as
389    /// [`Self::madvise`]. The pointer is aligned down to a page boundary
390    /// as required by `madvise`.
391    #[cfg(feature = "native")]
392    pub fn madvise_range(&self, range: Range<usize>, advice: libc::c_int) {
393        if !self.is_mmap() {
394            return;
395        }
396        let slice = &self.as_slice()[range];
397        if slice.is_empty() {
398            return;
399        }
400        let ptr = slice.as_ptr();
401        let len = slice.len();
402        let page_size = 4096usize;
403        let aligned_ptr = (ptr as usize) & !(page_size - 1);
404        let aligned_len = len + (ptr as usize - aligned_ptr);
405        unsafe {
406            libc::madvise(aligned_ptr as *mut libc::c_void, aligned_len, advice);
407        }
408    }
409
410    pub fn to_vec(&self) -> Vec<u8> {
411        self.as_slice().to_vec()
412    }
413}
414
415impl AsRef<[u8]> for OwnedBytes {
416    fn as_ref(&self) -> &[u8] {
417        self.as_slice()
418    }
419}
420
421impl std::ops::Deref for OwnedBytes {
422    type Target = [u8];
423
424    fn deref(&self) -> &Self::Target {
425        self.as_slice()
426    }
427}
428
429/// Async directory trait for reading index files
430#[cfg(not(target_arch = "wasm32"))]
431#[async_trait]
432pub trait Directory: Send + Sync + 'static {
433    /// Check if a file exists
434    async fn exists(&self, path: &Path) -> io::Result<bool>;
435
436    /// Get file size
437    async fn file_size(&self, path: &Path) -> io::Result<u64>;
438
439    /// Open a file for reading (loads entire file into an inline FileHandle)
440    async fn open_read(&self, path: &Path) -> io::Result<FileHandle>;
441
442    /// Read a specific byte range from a file (optimized for network)
443    async fn read_range(&self, path: &Path, range: Range<u64>) -> io::Result<OwnedBytes>;
444
445    /// List files in directory
446    async fn list_files(&self, prefix: &Path) -> io::Result<Vec<PathBuf>>;
447
448    /// Open a file handle that fetches ranges on demand.
449    /// For mmap directories this returns an Inline handle (sync-capable).
450    /// For HTTP/filesystem directories this returns a Lazy handle.
451    async fn open_lazy(&self, path: &Path) -> io::Result<FileHandle>;
452}
453
454/// Async directory trait for reading index files (WASM version - no Send requirement)
455#[cfg(target_arch = "wasm32")]
456#[async_trait(?Send)]
457pub trait Directory: 'static {
458    /// Check if a file exists
459    async fn exists(&self, path: &Path) -> io::Result<bool>;
460
461    /// Get file size
462    async fn file_size(&self, path: &Path) -> io::Result<u64>;
463
464    /// Open a file for reading (loads entire file into an inline FileHandle)
465    async fn open_read(&self, path: &Path) -> io::Result<FileHandle>;
466
467    /// Read a specific byte range from a file (optimized for network)
468    async fn read_range(&self, path: &Path, range: Range<u64>) -> io::Result<OwnedBytes>;
469
470    /// List files in directory
471    async fn list_files(&self, prefix: &Path) -> io::Result<Vec<PathBuf>>;
472
473    /// Open a file handle that fetches ranges on demand.
474    async fn open_lazy(&self, path: &Path) -> io::Result<FileHandle>;
475}
476
477/// A writer for incrementally writing data to a directory file.
478///
479/// Avoids buffering entire files in memory during merge. File-backed
480/// directories write directly to disk; memory directories collect to Vec.
481pub trait StreamingWriter: io::Write + Send {
482    /// Finalize the write, making data available for reading.
483    fn finish(self: Box<Self>) -> io::Result<()>;
484
485    /// Bytes written so far.
486    fn bytes_written(&self) -> u64;
487}
488
489/// StreamingWriter backed by Vec<u8>, finalized via DirectoryWriter::write.
490/// Used as default/fallback and for RamDirectory.
491struct BufferedStreamingWriter {
492    path: PathBuf,
493    buffer: Vec<u8>,
494    /// Callback to write the buffer to the directory on finish.
495    /// We store the files Arc directly for RamDirectory.
496    files: Arc<RwLock<HashMap<PathBuf, Arc<Vec<u8>>>>>,
497}
498
499impl io::Write for BufferedStreamingWriter {
500    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
501        self.buffer.extend_from_slice(buf);
502        Ok(buf.len())
503    }
504
505    fn flush(&mut self) -> io::Result<()> {
506        Ok(())
507    }
508}
509
510impl StreamingWriter for BufferedStreamingWriter {
511    fn finish(self: Box<Self>) -> io::Result<()> {
512        self.files.write().insert(self.path, Arc::new(self.buffer));
513        Ok(())
514    }
515
516    fn bytes_written(&self) -> u64 {
517        self.buffer.len() as u64
518    }
519}
520
521/// Buffer size for FileStreamingWriter (8 MB).
522/// Large enough to coalesce millions of tiny writes (e.g. per-vector doc_id writes)
523/// into efficient sequential I/O.
524#[cfg(feature = "native")]
525const FILE_STREAMING_BUF_SIZE: usize = 8 * 1024 * 1024;
526
527/// StreamingWriter backed by a buffered std::fs::File for filesystem directories.
528#[cfg(feature = "native")]
529pub(crate) struct FileStreamingWriter {
530    pub(crate) file: io::BufWriter<std::fs::File>,
531    pub(crate) written: u64,
532}
533
534#[cfg(feature = "native")]
535impl FileStreamingWriter {
536    pub(crate) fn new(file: std::fs::File) -> Self {
537        Self {
538            file: io::BufWriter::with_capacity(FILE_STREAMING_BUF_SIZE, file),
539            written: 0,
540        }
541    }
542}
543
544#[cfg(feature = "native")]
545impl io::Write for FileStreamingWriter {
546    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
547        let n = self.file.write(buf)?;
548        self.written += n as u64;
549        Ok(n)
550    }
551
552    fn flush(&mut self) -> io::Result<()> {
553        self.file.flush()
554    }
555}
556
557#[cfg(feature = "native")]
558impl StreamingWriter for FileStreamingWriter {
559    fn finish(self: Box<Self>) -> io::Result<()> {
560        let file = self.file.into_inner().map_err(|e| e.into_error())?;
561        file.sync_all()?;
562        Ok(())
563    }
564
565    fn bytes_written(&self) -> u64 {
566        self.written
567    }
568}
569
570/// Async directory trait for writing index files
571#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
572#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
573pub trait DirectoryWriter: Directory {
574    /// Create/overwrite a file with data
575    async fn write(&self, path: &Path, data: &[u8]) -> io::Result<()>;
576
577    /// Delete a file
578    async fn delete(&self, path: &Path) -> io::Result<()>;
579
580    /// Atomic rename
581    async fn rename(&self, from: &Path, to: &Path) -> io::Result<()>;
582
583    /// Sync all pending writes
584    async fn sync(&self) -> io::Result<()>;
585
586    /// Create a streaming writer for incremental file writes.
587    /// Call finish() on the returned writer to finalize.
588    async fn streaming_writer(&self, path: &Path) -> io::Result<Box<dyn StreamingWriter>>;
589}
590
591/// In-memory directory for testing and small indexes
592#[derive(Debug, Default)]
593pub struct RamDirectory {
594    files: Arc<RwLock<HashMap<PathBuf, Arc<Vec<u8>>>>>,
595}
596
597impl Clone for RamDirectory {
598    fn clone(&self) -> Self {
599        Self {
600            files: Arc::clone(&self.files),
601        }
602    }
603}
604
605impl RamDirectory {
606    pub fn new() -> Self {
607        Self::default()
608    }
609
610    /// Synchronous file listing (for serialization).
611    pub fn list_files_sync(&self, prefix: &Path) -> io::Result<Vec<PathBuf>> {
612        let files = self.files.read();
613        Ok(files
614            .keys()
615            .filter(|p| p.starts_with(prefix))
616            .cloned()
617            .collect())
618    }
619
620    /// Synchronous file read (for serialization).
621    pub fn read_file_sync(&self, path: &Path) -> io::Result<Vec<u8>> {
622        let files = self.files.read();
623        files
624            .get(path)
625            .map(|data| data.as_ref().clone())
626            .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "File not found"))
627    }
628
629    /// Synchronous file write (for deserialization).
630    pub fn write_sync(&self, path: &Path, data: &[u8]) -> io::Result<()> {
631        self.files
632            .write()
633            .insert(path.to_path_buf(), Arc::new(data.to_vec()));
634        Ok(())
635    }
636}
637
638#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
639#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
640impl Directory for RamDirectory {
641    async fn exists(&self, path: &Path) -> io::Result<bool> {
642        Ok(self.files.read().contains_key(path))
643    }
644
645    async fn file_size(&self, path: &Path) -> io::Result<u64> {
646        self.files
647            .read()
648            .get(path)
649            .map(|data| data.len() as u64)
650            .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "File not found"))
651    }
652
653    async fn open_read(&self, path: &Path) -> io::Result<FileHandle> {
654        let files = self.files.read();
655        let data = files
656            .get(path)
657            .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "File not found"))?;
658
659        Ok(FileHandle::from_bytes(OwnedBytes::from_arc_vec(
660            Arc::clone(data),
661            0..data.len(),
662        )))
663    }
664
665    async fn read_range(&self, path: &Path, range: Range<u64>) -> io::Result<OwnedBytes> {
666        let files = self.files.read();
667        let data = files
668            .get(path)
669            .ok_or_else(|| io::Error::new(io::ErrorKind::NotFound, "File not found"))?;
670
671        let start = range.start as usize;
672        let end = range.end as usize;
673
674        if end > data.len() {
675            return Err(io::Error::new(
676                io::ErrorKind::InvalidInput,
677                "Range out of bounds",
678            ));
679        }
680
681        Ok(OwnedBytes::from_arc_vec(Arc::clone(data), start..end))
682    }
683
684    async fn list_files(&self, prefix: &Path) -> io::Result<Vec<PathBuf>> {
685        let files = self.files.read();
686        Ok(files
687            .keys()
688            .filter(|p| p.starts_with(prefix))
689            .cloned()
690            .collect())
691    }
692
693    async fn open_lazy(&self, path: &Path) -> io::Result<FileHandle> {
694        // RAM data is always available synchronously — return Inline handle
695        self.open_read(path).await
696    }
697}
698
699#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
700#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
701impl DirectoryWriter for RamDirectory {
702    async fn write(&self, path: &Path, data: &[u8]) -> io::Result<()> {
703        self.files
704            .write()
705            .insert(path.to_path_buf(), Arc::new(data.to_vec()));
706        Ok(())
707    }
708
709    async fn delete(&self, path: &Path) -> io::Result<()> {
710        self.files.write().remove(path);
711        Ok(())
712    }
713
714    async fn rename(&self, from: &Path, to: &Path) -> io::Result<()> {
715        let mut files = self.files.write();
716        if let Some(data) = files.remove(from) {
717            files.insert(to.to_path_buf(), data);
718        }
719        Ok(())
720    }
721
722    async fn sync(&self) -> io::Result<()> {
723        Ok(())
724    }
725
726    async fn streaming_writer(&self, path: &Path) -> io::Result<Box<dyn StreamingWriter>> {
727        Ok(Box::new(BufferedStreamingWriter {
728            path: path.to_path_buf(),
729            buffer: Vec::new(),
730            files: Arc::clone(&self.files),
731        }))
732    }
733}
734
735/// Local filesystem directory with async IO via tokio
736#[cfg(feature = "native")]
737#[derive(Debug, Clone)]
738pub struct FsDirectory {
739    root: PathBuf,
740}
741
742#[cfg(feature = "native")]
743impl FsDirectory {
744    pub fn new(root: impl AsRef<Path>) -> Self {
745        Self {
746            root: root.as_ref().to_path_buf(),
747        }
748    }
749
750    fn resolve(&self, path: &Path) -> PathBuf {
751        self.root.join(path)
752    }
753}
754
755#[cfg(feature = "native")]
756#[async_trait]
757impl Directory for FsDirectory {
758    async fn exists(&self, path: &Path) -> io::Result<bool> {
759        let full_path = self.resolve(path);
760        Ok(tokio::fs::try_exists(&full_path).await.unwrap_or(false))
761    }
762
763    async fn file_size(&self, path: &Path) -> io::Result<u64> {
764        let full_path = self.resolve(path);
765        let metadata = tokio::fs::metadata(&full_path).await?;
766        Ok(metadata.len())
767    }
768
769    async fn open_read(&self, path: &Path) -> io::Result<FileHandle> {
770        let full_path = self.resolve(path);
771        let data = tokio::fs::read(&full_path).await?;
772        Ok(FileHandle::from_bytes(OwnedBytes::new(data)))
773    }
774
775    async fn read_range(&self, path: &Path, range: Range<u64>) -> io::Result<OwnedBytes> {
776        use tokio::io::{AsyncReadExt, AsyncSeekExt};
777
778        let full_path = self.resolve(path);
779        let mut file = tokio::fs::File::open(&full_path).await?;
780
781        file.seek(std::io::SeekFrom::Start(range.start)).await?;
782
783        let len = (range.end - range.start) as usize;
784        let mut buffer = vec![0u8; len];
785        file.read_exact(&mut buffer).await?;
786
787        Ok(OwnedBytes::new(buffer))
788    }
789
790    async fn list_files(&self, prefix: &Path) -> io::Result<Vec<PathBuf>> {
791        let full_path = self.resolve(prefix);
792        let mut entries = tokio::fs::read_dir(&full_path).await?;
793        let mut files = Vec::new();
794
795        while let Some(entry) = entries.next_entry().await? {
796            if entry.file_type().await?.is_file() {
797                files.push(entry.path().strip_prefix(&self.root).unwrap().to_path_buf());
798            }
799        }
800
801        Ok(files)
802    }
803
804    async fn open_lazy(&self, path: &Path) -> io::Result<FileHandle> {
805        let full_path = self.resolve(path);
806        let metadata = tokio::fs::metadata(&full_path).await?;
807        let file_size = metadata.len();
808
809        let read_fn: RangeReadFn = Arc::new(move |range: Range<u64>| {
810            let full_path = full_path.clone();
811            Box::pin(async move {
812                use tokio::io::{AsyncReadExt, AsyncSeekExt};
813
814                let mut file = tokio::fs::File::open(&full_path).await?;
815                file.seek(std::io::SeekFrom::Start(range.start)).await?;
816
817                let len = (range.end - range.start) as usize;
818                let mut buffer = vec![0u8; len];
819                file.read_exact(&mut buffer).await?;
820
821                Ok(OwnedBytes::new(buffer))
822            })
823        });
824
825        Ok(FileHandle::lazy(file_size, read_fn))
826    }
827}
828
829#[cfg(feature = "native")]
830#[async_trait]
831impl DirectoryWriter for FsDirectory {
832    async fn write(&self, path: &Path, data: &[u8]) -> io::Result<()> {
833        let full_path = self.resolve(path);
834
835        // Ensure parent directory exists
836        if let Some(parent) = full_path.parent() {
837            tokio::fs::create_dir_all(parent).await?;
838        }
839
840        tokio::fs::write(&full_path, data).await
841    }
842
843    async fn delete(&self, path: &Path) -> io::Result<()> {
844        let full_path = self.resolve(path);
845        tokio::fs::remove_file(&full_path).await
846    }
847
848    async fn rename(&self, from: &Path, to: &Path) -> io::Result<()> {
849        let from_path = self.resolve(from);
850        let to_path = self.resolve(to);
851        tokio::fs::rename(&from_path, &to_path).await
852    }
853
854    async fn sync(&self) -> io::Result<()> {
855        // fsync the directory
856        let dir = std::fs::File::open(&self.root)?;
857        dir.sync_all()?;
858        Ok(())
859    }
860
861    async fn streaming_writer(&self, path: &Path) -> io::Result<Box<dyn StreamingWriter>> {
862        let full_path = self.resolve(path);
863        if let Some(parent) = full_path.parent() {
864            tokio::fs::create_dir_all(parent).await?;
865        }
866        let file = std::fs::File::create(&full_path)?;
867        Ok(Box::new(FileStreamingWriter::new(file)))
868    }
869}
870
871/// Caching wrapper for any Directory - caches file reads
872pub struct CachingDirectory<D: Directory> {
873    inner: D,
874    cache: RwLock<HashMap<PathBuf, Arc<Vec<u8>>>>,
875    max_cached_bytes: usize,
876    current_bytes: RwLock<usize>,
877}
878
879impl<D: Directory> CachingDirectory<D> {
880    pub fn new(inner: D, max_cached_bytes: usize) -> Self {
881        Self {
882            inner,
883            cache: RwLock::new(HashMap::new()),
884            max_cached_bytes,
885            current_bytes: RwLock::new(0),
886        }
887    }
888
889    fn try_cache(&self, path: &Path, data: &[u8]) {
890        let mut current = self.current_bytes.write();
891        if *current + data.len() <= self.max_cached_bytes {
892            self.cache
893                .write()
894                .insert(path.to_path_buf(), Arc::new(data.to_vec()));
895            *current += data.len();
896        }
897    }
898}
899
900#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
901#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
902impl<D: Directory> Directory for CachingDirectory<D> {
903    async fn exists(&self, path: &Path) -> io::Result<bool> {
904        if self.cache.read().contains_key(path) {
905            return Ok(true);
906        }
907        self.inner.exists(path).await
908    }
909
910    async fn file_size(&self, path: &Path) -> io::Result<u64> {
911        if let Some(data) = self.cache.read().get(path) {
912            return Ok(data.len() as u64);
913        }
914        self.inner.file_size(path).await
915    }
916
917    async fn open_read(&self, path: &Path) -> io::Result<FileHandle> {
918        // Check cache first
919        if let Some(data) = self.cache.read().get(path) {
920            return Ok(FileHandle::from_bytes(OwnedBytes::from_arc_vec(
921                Arc::clone(data),
922                0..data.len(),
923            )));
924        }
925
926        // Read from inner and potentially cache
927        let handle = self.inner.open_read(path).await?;
928        let bytes = handle.read_bytes().await?;
929
930        self.try_cache(path, bytes.as_slice());
931
932        Ok(FileHandle::from_bytes(bytes))
933    }
934
935    async fn read_range(&self, path: &Path, range: Range<u64>) -> io::Result<OwnedBytes> {
936        // Check cache first
937        if let Some(data) = self.cache.read().get(path) {
938            let start = range.start as usize;
939            let end = range.end as usize;
940            return Ok(OwnedBytes::from_arc_vec(Arc::clone(data), start..end));
941        }
942
943        self.inner.read_range(path, range).await
944    }
945
946    async fn list_files(&self, prefix: &Path) -> io::Result<Vec<PathBuf>> {
947        self.inner.list_files(prefix).await
948    }
949
950    async fn open_lazy(&self, path: &Path) -> io::Result<FileHandle> {
951        // For caching directory, delegate to inner - caching happens at read_range level
952        self.inner.open_lazy(path).await
953    }
954}
955
956#[cfg(test)]
957mod tests {
958    use super::*;
959
960    #[tokio::test]
961    async fn test_ram_directory() {
962        let dir = RamDirectory::new();
963
964        // Write file
965        dir.write(Path::new("test.txt"), b"hello world")
966            .await
967            .unwrap();
968
969        // Check exists
970        assert!(dir.exists(Path::new("test.txt")).await.unwrap());
971        assert!(!dir.exists(Path::new("nonexistent.txt")).await.unwrap());
972
973        // Read file
974        let slice = dir.open_read(Path::new("test.txt")).await.unwrap();
975        let data = slice.read_bytes().await.unwrap();
976        assert_eq!(data.as_slice(), b"hello world");
977
978        // Read range
979        let range_data = dir.read_range(Path::new("test.txt"), 0..5).await.unwrap();
980        assert_eq!(range_data.as_slice(), b"hello");
981
982        // Delete
983        dir.delete(Path::new("test.txt")).await.unwrap();
984        assert!(!dir.exists(Path::new("test.txt")).await.unwrap());
985    }
986
987    #[tokio::test]
988    async fn test_file_handle() {
989        let data = OwnedBytes::new(b"hello world".to_vec());
990        let handle = FileHandle::from_bytes(data);
991
992        assert_eq!(handle.len(), 11);
993        assert!(handle.is_sync());
994
995        let sub = handle.slice(0..5);
996        let bytes = sub.read_bytes().await.unwrap();
997        assert_eq!(bytes.as_slice(), b"hello");
998
999        let sub2 = handle.slice(6..11);
1000        let bytes2 = sub2.read_bytes().await.unwrap();
1001        assert_eq!(bytes2.as_slice(), b"world");
1002
1003        // Sync reads work on inline handles
1004        let sync_bytes = handle.read_bytes_range_sync(0..5).unwrap();
1005        assert_eq!(sync_bytes.as_slice(), b"hello");
1006    }
1007
1008    #[tokio::test]
1009    async fn test_owned_bytes() {
1010        let bytes = OwnedBytes::new(vec![1, 2, 3, 4, 5]);
1011
1012        assert_eq!(bytes.len(), 5);
1013        assert_eq!(bytes.as_slice(), &[1, 2, 3, 4, 5]);
1014
1015        let sliced = bytes.slice(1..4);
1016        assert_eq!(sliced.as_slice(), &[2, 3, 4]);
1017
1018        // Original unchanged
1019        assert_eq!(bytes.as_slice(), &[1, 2, 3, 4, 5]);
1020    }
1021}
hermes_core/directories/directory.rs

hermes_core/directories/
directory.rs
