faucet_source_singer/
message.rs

1//! Singer message model + line parser.
2//!
3//! Pure logic: turns one line of a tap's stdout into a [`SingerMessage`]. The
4//! process runner applies the [`MalformedPolicy`](crate::config::MalformedPolicy)
5//! to any [`Err`] returned here.
6//!
7//! Singer message reference: <https://hub.meltano.com/singer/spec/>.
8
9use faucet_core::Value;
10
11/// A parsed Singer protocol message.
12///
13/// v0 models the three messages the bridge acts on plus a catch-all
14/// [`Other`](SingerMessage::Other) arm for messages that are logged and
15/// skipped (`ACTIVATE_VERSION`, `BATCH`, …).
16#[derive(Debug, Clone, PartialEq)]
17pub enum SingerMessage {
18    /// A data record for `stream`.
19    Record { stream: String, record: Value },
20    /// A JSON-Schema declaration for `stream`.
21    Schema {
22        stream: String,
23        schema: Value,
24        key_properties: Option<Vec<String>>,
25        bookmark_properties: Option<Vec<String>>,
26    },
27    /// A resume-checkpoint blob. `value` is opaque to the bridge — it is passed
28    /// back to the tap verbatim on resume via `--state`.
29    State { value: Value },
30    /// Any other message type (e.g. `ACTIVATE_VERSION`, `BATCH`): recognized,
31    /// logged, and skipped in v0.
32    Other { message_type: String },
33}
34
35/// Parse one line of tap stdout into a [`SingerMessage`].
36///
37/// Returns `Err(reason)` for a blank line, non-JSON, a non-object payload, a
38/// missing/`non-string `type`, or a well-typed message missing a required field
39/// (e.g. a RECORD with no `record`). The caller decides whether that aborts the
40/// run or is skipped, per the malformed policy.
41pub fn parse_line(line: &str) -> Result<SingerMessage, String> {
42    let trimmed = line.trim();
43    if trimmed.is_empty() {
44        return Err("blank line".to_string());
45    }
46    let value: Value = serde_json::from_str(trimmed).map_err(|e| format!("not valid JSON: {e}"))?;
47    let obj = value
48        .as_object()
49        .ok_or_else(|| "message is not a JSON object".to_string())?;
50    let msg_type = obj
51        .get("type")
52        .and_then(Value::as_str)
53        .ok_or_else(|| "missing or non-string `type` field".to_string())?;
54
55    match msg_type {
56        "RECORD" => {
57            let stream = obj
58                .get("stream")
59                .and_then(Value::as_str)
60                .ok_or_else(|| "RECORD missing `stream`".to_string())?
61                .to_string();
62            let record = obj
63                .get("record")
64                .ok_or_else(|| "RECORD missing `record`".to_string())?
65                .clone();
66            if !record.is_object() {
67                return Err("RECORD `record` is not an object".to_string());
68            }
69            Ok(SingerMessage::Record { stream, record })
70        }
71        "SCHEMA" => {
72            let stream = obj
73                .get("stream")
74                .and_then(Value::as_str)
75                .ok_or_else(|| "SCHEMA missing `stream`".to_string())?
76                .to_string();
77            let schema = obj
78                .get("schema")
79                .ok_or_else(|| "SCHEMA missing `schema`".to_string())?
80                .clone();
81            Ok(SingerMessage::Schema {
82                stream,
83                schema,
84                key_properties: string_array(obj.get("key_properties")),
85                bookmark_properties: string_array(obj.get("bookmark_properties")),
86            })
87        }
88        "STATE" => {
89            let value = obj
90                .get("value")
91                .ok_or_else(|| "STATE missing `value`".to_string())?
92                .clone();
93            Ok(SingerMessage::State { value })
94        }
95        other => Ok(SingerMessage::Other {
96            message_type: other.to_string(),
97        }),
98    }
99}
100
101/// Extract an optional `["a","b"]` string array from a JSON field.
102fn string_array(v: Option<&Value>) -> Option<Vec<String>> {
103    v.and_then(Value::as_array).map(|arr| {
104        arr.iter()
105            .filter_map(|x| x.as_str().map(str::to_string))
106            .collect()
107    })
108}
109
110#[cfg(test)]
111mod tests {
112    use super::*;
113    use serde_json::json;
114
115    #[test]
116    fn parses_record() {
117        let m = parse_line(r#"{"type":"RECORD","stream":"s","record":{"id":1}}"#).unwrap();
118        assert_eq!(
119            m,
120            SingerMessage::Record {
121                stream: "s".into(),
122                record: json!({"id": 1})
123            }
124        );
125    }
126
127    #[test]
128    fn parses_schema_with_key_props() {
129        let m = parse_line(
130            r#"{"type":"SCHEMA","stream":"s","schema":{"type":"object"},"key_properties":["id"],"bookmark_properties":["updated_at"]}"#,
131        )
132        .unwrap();
133        match m {
134            SingerMessage::Schema {
135                stream,
136                key_properties,
137                bookmark_properties,
138                ..
139            } => {
140                assert_eq!(stream, "s");
141                assert_eq!(key_properties, Some(vec!["id".to_string()]));
142                assert_eq!(bookmark_properties, Some(vec!["updated_at".to_string()]));
143            }
144            _ => panic!("expected SCHEMA"),
145        }
146    }
147
148    #[test]
149    fn parses_state() {
150        let m = parse_line(r#"{"type":"STATE","value":{"bookmarks":{"s":{"v":5}}}}"#).unwrap();
151        assert_eq!(
152            m,
153            SingerMessage::State {
154                value: json!({"bookmarks":{"s":{"v":5}}})
155            }
156        );
157    }
158
159    #[test]
160    fn activate_version_and_batch_are_other() {
161        assert_eq!(
162            parse_line(r#"{"type":"ACTIVATE_VERSION","stream":"s","version":1}"#).unwrap(),
163            SingerMessage::Other {
164                message_type: "ACTIVATE_VERSION".into()
165            }
166        );
167        assert_eq!(
168            parse_line(r#"{"type":"BATCH","stream":"s"}"#).unwrap(),
169            SingerMessage::Other {
170                message_type: "BATCH".into()
171            }
172        );
173    }
174
175    #[test]
176    fn malformed_variants_error() {
177        assert!(parse_line("").is_err()); // blank
178        assert!(parse_line("not json").is_err()); // non-json
179        assert!(parse_line("[1,2,3]").is_err()); // not an object
180        assert!(parse_line(r#"{"stream":"s"}"#).is_err()); // no type
181        assert!(parse_line(r#"{"type":"RECORD","stream":"s"}"#).is_err()); // no record
182        assert!(parse_line(r#"{"type":"RECORD","record":{"id":1}}"#).is_err()); // no stream
183        assert!(parse_line(r#"{"type":"RECORD","stream":"s","record":5}"#).is_err()); // record not obj
184        assert!(parse_line(r#"{"type":"STATE"}"#).is_err()); // no value
185        assert!(parse_line(r#"{"type":"SCHEMA","stream":"s"}"#).is_err()); // no schema
186    }
187}
faucet_source_singer/message.rs

faucet_source_singer/
message.rs
