faucet_source_redshift/
config.rs

1//! Amazon Redshift source configuration.
2
3use faucet_common_redshift::RedshiftConnection;
4use faucet_core::{DEFAULT_BATCH_SIZE, FaucetError};
5use schemars::JsonSchema;
6use serde::{Deserialize, Serialize};
7use serde_json::Value;
8
9/// How the Redshift source replicates rows across runs.
10///
11/// Serializes as `{ type: full }` or
12/// `{ type: incremental, column: "...", initial_value: ... }`.
13#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema, Default, PartialEq)]
14#[serde(tag = "type", rename_all = "snake_case")]
15pub enum RedshiftReplication {
16    /// Every run re-fetches the full result set (default).
17    #[default]
18    Full,
19    /// Only rows whose `column` is strictly greater than the stored bookmark
20    /// (or `initial_value` on the first run) are emitted. If the SQL contains
21    /// the literal token `${bookmark}`, it is replaced with a positional bind
22    /// parameter so Redshift filters server-side (efficient); the source also
23    /// filters client-side as a correctness backstop. The new maximum of
24    /// `column` is persisted on the final page.
25    Incremental {
26        /// Column whose value is the replication cursor (e.g. `updated_at`).
27        column: String,
28        /// Lower bound used on the first run, before any bookmark is stored.
29        initial_value: Value,
30    },
31}
32
33fn default_max_connections() -> u32 {
34    10
35}
36
37fn default_batch_size() -> usize {
38    DEFAULT_BATCH_SIZE
39}
40
41/// Configuration for the Amazon Redshift query source.
42#[derive(Clone, Debug, Serialize, Deserialize, JsonSchema)]
43pub struct RedshiftSourceConfig {
44    /// Connection block (host / port / database / user / credentials / tls),
45    /// flattened to the config top level.
46    #[serde(flatten)]
47    pub connection: RedshiftConnection,
48    /// SQL query to execute. May contain `${field.path}` parent-context tokens
49    /// (resolved to positional binds at runtime) and, for incremental
50    /// replication, a `${bookmark}` token.
51    pub query: String,
52    /// Positional bind parameters for the query, applied in `$1, $2, …` order
53    /// before any context- or bookmark-derived values. Defaults to empty.
54    #[serde(default)]
55    pub params: Vec<Value>,
56    /// Maximum number of connections in the pool. Defaults to 10.
57    #[serde(default = "default_max_connections")]
58    pub max_connections: u32,
59    /// Records per emitted [`StreamPage`](faucet_core::StreamPage). Rows are
60    /// drained from the `sqlx` cursor and yielded whenever the buffer reaches
61    /// this size. Defaults to [`DEFAULT_BATCH_SIZE`].
62    ///
63    /// `batch_size = 0` is the "no batching" sentinel: the cursor is fully
64    /// drained and the entire result set is emitted in a single page.
65    #[serde(default = "default_batch_size")]
66    pub batch_size: usize,
67    /// Replication mode. Defaults to [`RedshiftReplication::Full`].
68    #[serde(default)]
69    pub replication: RedshiftReplication,
70    /// Explicit state-store key for the incremental bookmark. When unset, a key
71    /// is derived from the host / database and a query fingerprint.
72    #[serde(default, skip_serializing_if = "Option::is_none")]
73    pub state_key: Option<String>,
74}
75
76impl RedshiftSourceConfig {
77    /// Validate the config; returns a human-readable error if invalid.
78    pub fn validate(&self) -> Result<(), FaucetError> {
79        if self.query.trim().is_empty() {
80            return Err(FaucetError::Config(
81                "redshift: `query` must not be empty".into(),
82            ));
83        }
84        if let RedshiftReplication::Incremental { column, .. } = &self.replication
85            && column.trim().is_empty()
86        {
87            return Err(FaucetError::Config(
88                "redshift: incremental replication `column` must not be empty".into(),
89            ));
90        }
91        faucet_core::validate_batch_size(self.batch_size)?;
92        Ok(())
93    }
94}
95
96#[cfg(test)]
97mod tests {
98    use super::*;
99    use faucet_common_redshift::RedshiftConnection;
100    use serde_json::json;
101
102    fn base() -> RedshiftSourceConfig {
103        RedshiftSourceConfig {
104            connection: RedshiftConnection::new("host", "db", "user", "pw"),
105            query: "SELECT * FROM events".into(),
106            params: Vec::new(),
107            max_connections: default_max_connections(),
108            batch_size: DEFAULT_BATCH_SIZE,
109            replication: RedshiftReplication::Full,
110            state_key: None,
111        }
112    }
113
114    #[test]
115    fn valid_config_passes() {
116        base().validate().unwrap();
117    }
118
119    #[test]
120    fn rejects_empty_query() {
121        let mut c = base();
122        c.query = "  ".into();
123        assert!(c.validate().is_err());
124    }
125
126    #[test]
127    fn rejects_empty_incremental_column() {
128        let mut c = base();
129        c.replication = RedshiftReplication::Incremental {
130            column: "".into(),
131            initial_value: json!(0),
132        };
133        assert!(c.validate().is_err());
134    }
135
136    #[test]
137    fn rejects_oversized_batch() {
138        let mut c = base();
139        c.batch_size = faucet_core::MAX_BATCH_SIZE + 1;
140        assert!(c.validate().is_err());
141    }
142
143    #[test]
144    fn deserializes_flattened_connection_and_defaults() {
145        let json = r#"{
146            "host": "h",
147            "database": "db",
148            "user": "u",
149            "credentials": {"type": "password", "config": {"password": "pw"}},
150            "query": "SELECT 1"
151        }"#;
152        let c: RedshiftSourceConfig = serde_json::from_str(json).unwrap();
153        assert_eq!(c.connection.host, "h");
154        assert_eq!(c.connection.port, faucet_common_redshift::DEFAULT_PORT);
155        assert_eq!(c.max_connections, 10);
156        assert_eq!(c.batch_size, DEFAULT_BATCH_SIZE);
157        assert!(matches!(c.replication, RedshiftReplication::Full));
158        c.validate().unwrap();
159    }
160
161    #[test]
162    fn deserializes_incremental_replication() {
163        let json = r#"{
164            "host": "h",
165            "database": "db",
166            "user": "u",
167            "credentials": {"type": "password", "config": {"password": "pw"}},
168            "query": "SELECT * FROM t WHERE ts > ${bookmark}",
169            "replication": {"type": "incremental", "column": "ts", "initial_value": "2026-01-01"},
170            "state_key": "my-key"
171        }"#;
172        let c: RedshiftSourceConfig = serde_json::from_str(json).unwrap();
173        match &c.replication {
174            RedshiftReplication::Incremental {
175                column,
176                initial_value,
177            } => {
178                assert_eq!(column, "ts");
179                assert_eq!(initial_value, &json!("2026-01-01"));
180            }
181            _ => panic!("expected incremental"),
182        }
183        assert_eq!(c.state_key.as_deref(), Some("my-key"));
184    }
185}
faucet_source_redshift/config.rs

faucet_source_redshift/
config.rs
