faucet_source_sftp/
config.rs

1//! SFTP source configuration.
2
3use faucet_common_sftp::SftpConnectionConfig;
4use faucet_core::DEFAULT_BATCH_SIZE;
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7
8/// Format of the remote files read by the SFTP source.
9#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, JsonSchema, PartialEq, Eq)]
10#[serde(rename_all = "snake_case")]
11pub enum SftpFormat {
12    /// Each line in the file is a separate JSON record (the default).
13    /// Streamed line-by-line with bounded memory.
14    #[default]
15    Jsonl,
16    /// The entire file is a JSON array of records. Buffered fully per file
17    /// (the closing `]` is required to validate the structure), then chunked.
18    JsonArray,
19    /// Each file becomes a single record with `"path"` and `"content"` fields.
20    RawText,
21}
22
23/// Configuration for the SFTP source connector.
24#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
25pub struct SftpSourceConfig {
26    /// Shared SFTP connection settings (host, port, username, auth, host-key
27    /// policy). Flattened, so its fields sit at the top level of the config.
28    #[serde(flatten)]
29    pub connection: SftpConnectionConfig,
30    /// Remote path to read: a directory whose files are listed and streamed,
31    /// or a single file.
32    pub path: String,
33    /// Optional filename glob (`*` / `?`) applied to the basenames of files in
34    /// a directory listing. Ignored when `path` points at a single file.
35    #[serde(default)]
36    pub glob: Option<String>,
37    /// Format of the files to read (default: `jsonl`).
38    #[serde(default)]
39    pub format: SftpFormat,
40    /// Records per emitted [`StreamPage`](faucet_core::StreamPage). For
41    /// `jsonl` / `raw_text`, files are decoded incrementally and a page is
42    /// yielded whenever the buffer reaches this size (bounded memory). For
43    /// `json_array`, each file is buffered fully, then its records are chunked
44    /// into pages of this size. Defaults to [`DEFAULT_BATCH_SIZE`].
45    ///
46    /// `batch_size = 0` is the "no batching" sentinel: one page is emitted per
47    /// file.
48    #[serde(default = "default_batch_size")]
49    pub batch_size: usize,
50}
51
52fn default_batch_size() -> usize {
53    DEFAULT_BATCH_SIZE
54}
55
56impl SftpSourceConfig {
57    /// Build a source config from a connection and a remote path, with default
58    /// format (`jsonl`) and batch size.
59    pub fn new(connection: SftpConnectionConfig, path: impl Into<String>) -> Self {
60        Self {
61            connection,
62            path: path.into(),
63            glob: None,
64            format: SftpFormat::default(),
65            batch_size: DEFAULT_BATCH_SIZE,
66        }
67    }
68
69    /// Set the filename glob filter.
70    pub fn glob(mut self, glob: impl Into<String>) -> Self {
71        self.glob = Some(glob.into());
72        self
73    }
74
75    /// Set the file format.
76    pub fn format(mut self, format: SftpFormat) -> Self {
77        self.format = format;
78        self
79    }
80
81    /// Set the per-page record count.
82    pub fn with_batch_size(mut self, batch_size: usize) -> Self {
83        self.batch_size = batch_size;
84        self
85    }
86}
87
88/// Match `name` against a `*` / `?` glob `pattern`.
89///
90/// `*` matches any run of characters (including empty); `?` matches exactly one
91/// character. All other characters match literally. Pure and allocation-light
92/// (linear backtracking), used to filter directory listings by basename.
93pub fn glob_match(pattern: &str, name: &str) -> bool {
94    let p: Vec<char> = pattern.chars().collect();
95    let n: Vec<char> = name.chars().collect();
96    let (mut pi, mut ni) = (0usize, 0usize);
97    // Position to backtrack to on a `*` mismatch.
98    let (mut star, mut star_n) = (None, 0usize);
99
100    while ni < n.len() {
101        if pi < p.len() && (p[pi] == '?' || p[pi] == n[ni]) {
102            pi += 1;
103            ni += 1;
104        } else if pi < p.len() && p[pi] == '*' {
105            star = Some(pi);
106            star_n = ni;
107            pi += 1;
108        } else if let Some(s) = star {
109            pi = s + 1;
110            star_n += 1;
111            ni = star_n;
112        } else {
113            return false;
114        }
115    }
116    // Consume trailing `*`s.
117    while pi < p.len() && p[pi] == '*' {
118        pi += 1;
119    }
120    pi == p.len()
121}
122
123#[cfg(test)]
124mod tests {
125    use super::*;
126    use faucet_common_sftp::SftpConnectionConfig;
127
128    fn conn() -> SftpConnectionConfig {
129        SftpConnectionConfig::with_password("h", "u", "p")
130    }
131
132    #[test]
133    fn defaults() {
134        let cfg = SftpSourceConfig::new(conn(), "/data");
135        assert_eq!(cfg.path, "/data");
136        assert!(cfg.glob.is_none());
137        assert_eq!(cfg.format, SftpFormat::Jsonl);
138        assert_eq!(cfg.batch_size, DEFAULT_BATCH_SIZE);
139    }
140
141    #[test]
142    fn format_default_is_jsonl() {
143        assert_eq!(SftpFormat::default(), SftpFormat::Jsonl);
144    }
145
146    #[test]
147    fn deserializes_flat_shape() {
148        let json = r#"{
149            "host": "sftp.example.com",
150            "port": 2222,
151            "username": "user",
152            "type": "password",
153            "config": { "password": "secret" },
154            "path": "/incoming",
155            "glob": "*.jsonl",
156            "format": "jsonl",
157            "batch_size": 250
158        }"#;
159        let cfg: SftpSourceConfig = serde_json::from_str(json).unwrap();
160        assert_eq!(cfg.connection.host, "sftp.example.com");
161        assert_eq!(cfg.connection.port, 2222);
162        assert_eq!(cfg.path, "/incoming");
163        assert_eq!(cfg.glob.as_deref(), Some("*.jsonl"));
164        assert_eq!(cfg.batch_size, 250);
165    }
166
167    #[test]
168    fn batch_size_zero_is_valid_sentinel() {
169        let cfg = SftpSourceConfig::new(conn(), "/d").with_batch_size(0);
170        assert_eq!(cfg.batch_size, 0);
171        assert!(faucet_core::validate_batch_size(cfg.batch_size).is_ok());
172    }
173
174    #[test]
175    fn format_parses_all_variants() {
176        for (s, want) in [
177            ("jsonl", SftpFormat::Jsonl),
178            ("json_array", SftpFormat::JsonArray),
179            ("raw_text", SftpFormat::RawText),
180        ] {
181            let got: SftpFormat = serde_json::from_str(&format!("\"{s}\"")).unwrap();
182            assert_eq!(got, want);
183        }
184    }
185
186    #[test]
187    fn glob_literal_and_wildcards() {
188        assert!(glob_match("*.jsonl", "orders.jsonl"));
189        assert!(glob_match("*.jsonl", ".jsonl"));
190        assert!(!glob_match("*.jsonl", "orders.json"));
191        assert!(glob_match("data-?.csv", "data-1.csv"));
192        assert!(!glob_match("data-?.csv", "data-12.csv"));
193        assert!(glob_match("*", "anything"));
194        assert!(glob_match("exact.txt", "exact.txt"));
195        assert!(!glob_match("exact.txt", "other.txt"));
196        assert!(glob_match("a*b*c", "axxbyyc"));
197        assert!(!glob_match("a*b*c", "axxbyy"));
198    }
199}
faucet_source_sftp/config.rs

faucet_source_sftp/
config.rs
