link_r/source/
fs.rs

1// Author: support@lituus.io
2// Copyright: lituus-io, all rights reserved
3//
4//! The local-filesystem source: index a directory tree (offline / testing).
5//!
6//! Walks a directory, yielding one [`Page`] per indexable file with a `file://`
7//! URL. Reads are lazy (per stream item) and synchronous, so this needs no async
8//! runtime or `async-stream`.
9
10use crate::error::{Error, Result};
11use crate::payload::DocPayload;
12use crate::resource::{Page, Resource, ResourceKind, SourceRef};
13use crate::source::Source;
14use bytes::Bytes;
15use futures::StreamExt;
16use std::path::PathBuf;
17use url::Url;
18use walkdir::WalkDir;
19
20/// A source over a local directory tree.
21#[derive(Clone, Copy, Debug, Default)]
22pub struct FsSource;
23
24impl FsSource {
25    /// Create a filesystem source.
26    #[must_use]
27    pub fn new() -> Self {
28        Self
29    }
30}
31
32fn read_page(path: &PathBuf) -> Result<Page<'static>> {
33    let kind = ResourceKind::from_path(&path.to_string_lossy());
34    let bytes = std::fs::read(path)?;
35    let url = Url::from_file_path(path)
36        .map_err(|()| Error::invalid_url(format!("not an absolute path: {}", path.display())))?;
37    Ok(Page::new(
38        Resource::new(url).with_kind(kind),
39        DocPayload::Owned(Bytes::from(bytes)),
40    ))
41}
42
43impl Source for FsSource {
44    type Pages<'a>
45        = impl futures::Stream<Item = Result<Page<'a>>> + Send + 'a
46    where
47        Self: 'a;
48
49    fn kind(&self) -> &'static str {
50        "fs"
51    }
52
53    fn discover<'a>(&'a self, root: &'a SourceRef) -> Self::Pages<'a> {
54        // Collect the file list eagerly (cheap), read contents lazily per item.
55        let paths: Vec<PathBuf> = match root {
56            SourceRef::Fs { root } => WalkDir::new(root)
57                .into_iter()
58                .filter_map(std::result::Result::ok)
59                .filter(|e| e.file_type().is_file())
60                .map(|e| e.path().to_path_buf())
61                .filter(|p| ResourceKind::from_path(&p.to_string_lossy()).is_indexable())
62                .collect(),
63            SourceRef::Http { .. } => Vec::new(),
64        };
65
66        let wrong_kind = !matches!(root, SourceRef::Fs { .. });
67        futures::stream::iter(paths)
68            .map(|path| read_page(&path))
69            .chain(futures::stream::iter(wrong_kind.then(|| {
70                Err(Error::crawl("fs", "FsSource requires an Fs SourceRef"))
71            })))
72    }
73}
74
75#[cfg(test)]
76mod tests {
77    use super::*;
78    use futures::StreamExt;
79    use std::io::Write;
80
81    fn write_file(dir: &std::path::Path, name: &str, contents: &str) {
82        let path = dir.join(name);
83        if let Some(parent) = path.parent() {
84            std::fs::create_dir_all(parent).unwrap();
85        }
86        let mut f = std::fs::File::create(path).unwrap();
87        f.write_all(contents.as_bytes()).unwrap();
88    }
89
90    #[tokio::test]
91    async fn walks_directory_and_yields_indexable_files() {
92        let tmp = tempfile::tempdir().unwrap();
93        write_file(tmp.path(), "a.md", "# A\n\nalpha content");
94        write_file(tmp.path(), "sub/b.txt", "bravo content");
95        write_file(tmp.path(), "image.png", "not indexable");
96
97        let source = FsSource::new();
98        let root = SourceRef::fs(tmp.path());
99        let stream = source.discover(&root);
100        futures::pin_mut!(stream);
101
102        let mut kinds = Vec::new();
103        while let Some(item) = stream.next().await {
104            let page = item.unwrap();
105            kinds.push(page.resource.kind);
106            assert_eq!(page.resource.url.scheme(), "file");
107        }
108        // png is filtered out as non-indexable; .md and .txt remain.
109        assert_eq!(kinds.len(), 2);
110        assert!(kinds.contains(&ResourceKind::Markdown));
111        assert!(kinds.contains(&ResourceKind::Text));
112    }
113
114    #[tokio::test]
115    async fn rejects_non_fs_source_ref() {
116        let source = FsSource::new();
117        let root = SourceRef::http("https://x.dev/").unwrap();
118        let stream = source.discover(&root);
119        futures::pin_mut!(stream);
120        let first = stream.next().await.unwrap();
121        assert!(first.is_err());
122    }
123}
link_r/source/fs.rs

link_r/source/
fs.rs
