faucet-source-singer 1.3.3

//! Configuration types for the Singer tap bridge source.
//!
//! No I/O or protocol logic lives here — just the serde/`JsonSchema` config
//! surface, following the connector-crate convention.

pub use faucet_common_singer::InheritEnv;
use faucet_core::JsonSchema;
use faucet_core::Value;
use serde::{Deserialize, Serialize};

/// What to do when the tap emits a line that is not valid Singer JSON.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema, Default)]
#[serde(rename_all = "snake_case")]
pub enum MalformedPolicy {
    /// Log the offending line at `warn` and continue (default).
    #[default]
    Skip,
    /// Abort the run with a [`faucet_core::FaucetError::Source`].
    Fail,
}

/// Configuration for the Singer tap bridge source.
///
/// Runs an external [Singer](https://www.singer.io/) tap executable and adapts
/// its stdout message stream into faucet records. **v0 is single-stream only**:
/// exactly the stream named by [`stream`](Self::stream) is emitted; RECORD
/// messages for other streams are ignored.
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct SingerSourceConfig {
    /// The tap executable to run (looked up on `PATH`, or an absolute path),
    /// e.g. `tap-github` or `/opt/taps/tap-csv`.
    #[schemars(extend("x-faucet-resource" = "command"))]
    pub executable: String,

    /// Extra positional/flag arguments appended after faucet's own
    /// `--config` / `--catalog` / `--state` flags.
    #[serde(default)]
    pub args: Vec<String>,

    /// The tap's configuration object. faucet's loader has already resolved any
    /// `${secret:…}` references before this reaches the source; it is
    /// materialized to a private temp file and passed as `--config`.
    #[serde(default)]
    pub tap_config: Value,

    /// Optional Singer catalog object, passed as `--catalog` when present.
    /// Selects streams/fields; required by most SDK-based taps.
    #[serde(default)]
    pub catalog: Option<Value>,

    /// The single stream to emit records for. RECORD messages for any other
    /// stream are dropped (v0 is single-stream).
    pub stream: String,

    /// State-store key for the resume bookmark. Defaults to
    /// `singer:{executable}:{stream}` when unset.
    #[serde(default)]
    pub state_key: Option<String>,

    /// Flush the current page whenever the tap emits a STATE message (so the
    /// pipeline can checkpoint at the tap's own commit points). Default `true`.
    #[serde(default = "default_true")]
    pub flush_on_state: bool,

    /// Abort the run if no line arrives from the tap within this many seconds
    /// (default 3600). `null` waits indefinitely — a stalled tap then hangs
    /// the run.
    #[serde(default = "default_idle_timeout_secs")]
    pub idle_timeout_secs: Option<u64>,

    /// Longest tap output line accepted, in bytes (default 64 MiB). A longer
    /// line — a huge RECORD, or binary output with no newline — fails the run
    /// instead of exhausting memory.
    #[serde(default = "default_max_line_bytes")]
    pub max_line_bytes: usize,

    /// How to handle a line that is not valid Singer JSON. Default
    /// [`MalformedPolicy::Skip`].
    #[serde(default)]
    pub on_malformed: MalformedPolicy,

    /// Which of faucet's environment variables the tap receives: `true`
    /// (default) the whole environment, `false` only `PATH`, `HOME`, `LANG`,
    /// `LC_ALL` and `TMPDIR`, or a list of variable names passed on top of
    /// that baseline. Set it to keep faucet's own credentials away from the tap.
    #[serde(default)]
    pub inherit_env: InheritEnv,
}

fn default_true() -> bool {
    true
}

fn default_idle_timeout_secs() -> Option<u64> {
    Some(3600)
}

fn default_max_line_bytes() -> usize {
    faucet_common_singer::DEFAULT_MAX_LINE_BYTES
}

impl SingerSourceConfig {
    /// Convenience constructor for the two required fields; other fields take
    /// their serde defaults.
    pub fn new(executable: impl Into<String>, stream: impl Into<String>) -> Self {
        Self {
            executable: executable.into(),
            args: Vec::new(),
            tap_config: Value::Object(Default::default()),
            catalog: None,
            stream: stream.into(),
            state_key: None,
            flush_on_state: true,
            idle_timeout_secs: default_idle_timeout_secs(),
            max_line_bytes: default_max_line_bytes(),
            on_malformed: MalformedPolicy::Skip,
            inherit_env: InheritEnv::default(),
        }
    }

    /// The effective state-store key: the configured `state_key`, or the
    /// derived default `singer:{executable}:{stream}`.
    ///
    /// The executable is sanitized so a path-style one (e.g.
    /// `/opt/taps/tap-csv`) still yields a valid key: every character outside
    /// `[A-Za-z0-9_.-]` becomes `_`. The stream name is kept as is whenever the
    /// key is valid with it (`validate_state_key` allows `[A-Za-z0-9_.:/-]`
    /// with no empty or dot-led `/` segments), so `public/orders` keeps its
    /// key. Otherwise it is sanitized the same way (keeping `:`) and a short
    /// hash of the raw name is appended, so `Sheet 1` and `Sheet_1` never
    /// share a key.
    pub fn effective_state_key(&self) -> String {
        if let Some(key) = &self.state_key {
            return key.clone();
        }
        let prefix = format!("singer:{}:", sanitize(&self.executable));
        let raw = format!("{prefix}{}", self.stream);
        if faucet_core::state::validate_state_key(&raw).is_ok() {
            return raw;
        }
        let suffix = format!(
            "-{:08x}",
            faucet_core::shard::shard_hash(&self.stream) >> 32
        );
        let room = MAX_STATE_KEY_LEN.saturating_sub(prefix.len() + suffix.len());
        let stream: String = self
            .stream
            .chars()
            .map(|c| if c == ':' { c } else { sanitize_char(c) })
            .take(room)
            .collect();
        format!("{prefix}{stream}{suffix}")
    }
}

/// The longest key `validate_state_key` accepts.
const MAX_STATE_KEY_LEN: usize = 256;

/// Replace characters not allowed in a state key with `_`.
fn sanitize(s: &str) -> String {
    s.chars().map(sanitize_char).collect()
}

fn sanitize_char(c: char) -> char {
    if c.is_ascii_alphanumeric() || matches!(c, '_' | '-' | '.') {
        c
    } else {
        '_'
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn defaults_apply() {
        let cfg: SingerSourceConfig =
            serde_json::from_value(serde_json::json!({"executable": "tap-x", "stream": "s"}))
                .unwrap();
        assert!(cfg.flush_on_state);
        assert_eq!(cfg.on_malformed, MalformedPolicy::Skip);
        assert!(cfg.args.is_empty());
        assert!(cfg.catalog.is_none());
        assert_eq!(cfg.idle_timeout_secs, Some(3600));
        assert_eq!(cfg.max_line_bytes, 64 * 1024 * 1024);
    }

    #[test]
    fn default_state_key_derivation() {
        let cfg = SingerSourceConfig::new("tap-github", "issues");
        assert_eq!(cfg.effective_state_key(), "singer:tap-github:issues");
        let cfg2 = SingerSourceConfig {
            state_key: Some("custom".into()),
            ..SingerSourceConfig::new("tap-github", "issues")
        };
        assert_eq!(cfg2.effective_state_key(), "custom");
    }

    #[test]
    fn derived_state_key_is_valid_even_for_path_executable() {
        let cfg = SingerSourceConfig::new("/opt/taps/tap-csv", "s");
        let key = cfg.effective_state_key();
        assert_eq!(key, "singer:_opt_taps_tap-csv:s");
        // must satisfy the core state-key validator
        faucet_core::state::validate_state_key(&key).expect("derived key must be valid");
    }

    #[test]
    fn derived_state_key_is_valid_for_any_stream_name() {
        let key = |stream: &str| SingerSourceConfig::new("tap-x", stream).effective_state_key();
        for (stream, expected) in [
            ("public/orders", "singer:tap-x:public/orders"),
            ("Sheet_1", "singer:tap-x:Sheet_1"),
            ("a:b.c-d_e", "singer:tap-x:a:b.c-d_e"),
        ] {
            assert_eq!(key(stream), expected, "a valid name keeps its key");
        }
        let spaced = key("Sheet 1");
        assert!(spaced.starts_with("singer:tap-x:Sheet_1-"), "{spaced}");
        assert_ne!(spaced, key("Sheet_1"));
        assert_ne!(spaced, key("Sheet/1 "));
        assert_ne!(key("a//b"), key("a__b"));
        let long = key(&format!("{} x", "s".repeat(400)));
        assert!(long.len() <= 256, "{}", long.len());
        for stream in ["Sheet 1", "a//b", "/lead", ".hidden", "ünï", "", "x y"] {
            let got = key(stream);
            faucet_core::state::validate_state_key(&got)
                .unwrap_or_else(|e| panic!("{stream:?} -> {got}: {e}"));
        }
        faucet_core::state::validate_state_key(&long).expect("long name");
    }

    #[test]
    fn malformed_policy_parses_snake_case() {
        let p: MalformedPolicy = serde_json::from_value(serde_json::json!("fail")).unwrap();
        assert_eq!(p, MalformedPolicy::Fail);
    }
}