faucet_source_singer/config.rs
1//! Configuration types for the Singer tap bridge source.
2//!
3//! No I/O or protocol logic lives here — just the serde/`JsonSchema` config
4//! surface, following the connector-crate convention.
5
6use faucet_core::JsonSchema;
7use faucet_core::Value;
8use serde::{Deserialize, Serialize};
9
10/// What to do when the tap emits a line that is not valid Singer JSON.
11#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema, Default)]
12#[serde(rename_all = "snake_case")]
13pub enum MalformedPolicy {
14 /// Log the offending line at `warn` and continue (default).
15 #[default]
16 Skip,
17 /// Abort the run with a [`faucet_core::FaucetError::Source`].
18 Fail,
19}
20
21/// Configuration for the Singer tap bridge source.
22///
23/// Runs an external [Singer](https://www.singer.io/) tap executable and adapts
24/// its stdout message stream into faucet records. **v0 is single-stream only**:
25/// exactly the stream named by [`stream`](Self::stream) is emitted; RECORD
26/// messages for other streams are ignored.
27#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
28pub struct SingerSourceConfig {
29 /// The tap executable to run (looked up on `PATH`, or an absolute path),
30 /// e.g. `tap-github` or `/opt/taps/tap-csv`.
31 pub executable: String,
32
33 /// Extra positional/flag arguments appended after faucet's own
34 /// `--config` / `--catalog` / `--state` flags.
35 #[serde(default)]
36 pub args: Vec<String>,
37
38 /// The tap's configuration object. faucet's loader has already resolved any
39 /// `${secret:…}` references before this reaches the source; it is
40 /// materialized to a private temp file and passed as `--config`.
41 #[serde(default)]
42 pub tap_config: Value,
43
44 /// Optional Singer catalog object, passed as `--catalog` when present.
45 /// Selects streams/fields; required by most SDK-based taps.
46 #[serde(default)]
47 pub catalog: Option<Value>,
48
49 /// The single stream to emit records for. RECORD messages for any other
50 /// stream are dropped (v0 is single-stream).
51 pub stream: String,
52
53 /// State-store key for the resume bookmark. Defaults to
54 /// `singer:{executable}:{stream}` when unset.
55 #[serde(default)]
56 pub state_key: Option<String>,
57
58 /// Flush the current page whenever the tap emits a STATE message (so the
59 /// pipeline can checkpoint at the tap's own commit points). Default `true`.
60 #[serde(default = "default_true")]
61 pub flush_on_state: bool,
62
63 /// Abort the run if no line arrives from the tap within this many seconds.
64 /// `None` (default) waits indefinitely.
65 #[serde(default)]
66 pub idle_timeout_secs: Option<u64>,
67
68 /// How to handle a line that is not valid Singer JSON. Default
69 /// [`MalformedPolicy::Skip`].
70 #[serde(default)]
71 pub on_malformed: MalformedPolicy,
72}
73
74fn default_true() -> bool {
75 true
76}
77
78impl SingerSourceConfig {
79 /// Convenience constructor for the two required fields; other fields take
80 /// their serde defaults.
81 pub fn new(executable: impl Into<String>, stream: impl Into<String>) -> Self {
82 Self {
83 executable: executable.into(),
84 args: Vec::new(),
85 tap_config: Value::Object(Default::default()),
86 catalog: None,
87 stream: stream.into(),
88 state_key: None,
89 flush_on_state: true,
90 idle_timeout_secs: None,
91 on_malformed: MalformedPolicy::Skip,
92 }
93 }
94
95 /// The effective state-store key: the configured `state_key`, or the
96 /// derived default `singer:{executable}:{stream}`.
97 ///
98 /// The derived key is sanitized so a path-style `executable` (e.g.
99 /// `/opt/taps/tap-csv`) still yields a valid state key — `validate_state_key`
100 /// only permits `[A-Za-z0-9_\-:.]`, so any other character (notably `/`) is
101 /// replaced with `_`.
102 pub fn effective_state_key(&self) -> String {
103 self.state_key
104 .clone()
105 .unwrap_or_else(|| format!("singer:{}:{}", sanitize(&self.executable), self.stream))
106 }
107}
108
109/// Replace characters not allowed in a state key with `_`.
110fn sanitize(s: &str) -> String {
111 s.chars()
112 .map(|c| {
113 if c.is_ascii_alphanumeric() || matches!(c, '_' | '-' | '.') {
114 c
115 } else {
116 '_'
117 }
118 })
119 .collect()
120}
121
122#[cfg(test)]
123mod tests {
124 use super::*;
125
126 #[test]
127 fn defaults_apply() {
128 let cfg: SingerSourceConfig =
129 serde_json::from_value(serde_json::json!({"executable": "tap-x", "stream": "s"}))
130 .unwrap();
131 assert!(cfg.flush_on_state);
132 assert_eq!(cfg.on_malformed, MalformedPolicy::Skip);
133 assert!(cfg.args.is_empty());
134 assert!(cfg.catalog.is_none());
135 assert!(cfg.idle_timeout_secs.is_none());
136 }
137
138 #[test]
139 fn default_state_key_derivation() {
140 let cfg = SingerSourceConfig::new("tap-github", "issues");
141 assert_eq!(cfg.effective_state_key(), "singer:tap-github:issues");
142 let cfg2 = SingerSourceConfig {
143 state_key: Some("custom".into()),
144 ..SingerSourceConfig::new("tap-github", "issues")
145 };
146 assert_eq!(cfg2.effective_state_key(), "custom");
147 }
148
149 #[test]
150 fn derived_state_key_is_valid_even_for_path_executable() {
151 let cfg = SingerSourceConfig::new("/opt/taps/tap-csv", "s");
152 let key = cfg.effective_state_key();
153 assert_eq!(key, "singer:_opt_taps_tap-csv:s");
154 // must satisfy the core state-key validator
155 faucet_core::state::validate_state_key(&key).expect("derived key must be valid");
156 }
157
158 #[test]
159 fn malformed_policy_parses_snake_case() {
160 let p: MalformedPolicy = serde_json::from_value(serde_json::json!("fail")).unwrap();
161 assert_eq!(p, MalformedPolicy::Fail);
162 }
163}