faucet_common_singer/
message.rs

1//! Singer message model, line parser, and line encoders.
2//!
3//! Pure logic: [`parse_line`] turns one line of a tap's (or target's) stdout
4//! into a [`SingerMessage`]; the `write_*` encoders append one newline-terminated
5//! Singer message to a byte buffer for a target's stdin.
6//!
7//! Singer message reference: <https://hub.meltano.com/singer/spec/>.
8
9use faucet_core::Value;
10use serde::Serialize;
11
12/// A parsed Singer protocol message.
13///
14/// v0 models the three messages the bridge acts on plus a catch-all
15/// [`Other`](SingerMessage::Other) arm for messages that are logged and
16/// skipped (`ACTIVATE_VERSION`, `BATCH`, …).
17#[derive(Debug, Clone, PartialEq)]
18pub enum SingerMessage {
19    /// A data record for `stream`.
20    Record { stream: String, record: Value },
21    /// A JSON-Schema declaration for `stream`.
22    Schema {
23        stream: String,
24        schema: Value,
25        key_properties: Option<Vec<String>>,
26        bookmark_properties: Option<Vec<String>>,
27    },
28    /// A resume-checkpoint blob. `value` is opaque to the bridge — it is passed
29    /// back to the tap verbatim on resume via `--state`.
30    State { value: Value },
31    /// Any other message type (e.g. `ACTIVATE_VERSION`, `BATCH`): recognized,
32    /// logged, and skipped in v0.
33    Other { message_type: String },
34}
35
36/// Parse one line of tap stdout into a [`SingerMessage`].
37///
38/// Returns `Err(reason)` for a blank line, non-JSON, a non-object payload, a
39/// missing/`non-string `type`, or a well-typed message missing a required field
40/// (e.g. a RECORD with no `record`). The caller decides whether that aborts the
41/// run or is skipped, per the malformed policy.
42pub fn parse_line(line: &str) -> Result<SingerMessage, String> {
43    let trimmed = line.trim();
44    if trimmed.is_empty() {
45        return Err("blank line".to_string());
46    }
47    let value: Value =
48        faucet_core::fidelity::from_str(trimmed).map_err(|e| format!("not valid JSON: {e}"))?;
49    let obj = value
50        .as_object()
51        .ok_or_else(|| "message is not a JSON object".to_string())?;
52    let msg_type = obj
53        .get("type")
54        .and_then(Value::as_str)
55        .ok_or_else(|| "missing or non-string `type` field".to_string())?;
56
57    match msg_type {
58        "RECORD" => {
59            let stream = obj
60                .get("stream")
61                .and_then(Value::as_str)
62                .ok_or_else(|| "RECORD missing `stream`".to_string())?
63                .to_string();
64            let record = obj
65                .get("record")
66                .ok_or_else(|| "RECORD missing `record`".to_string())?
67                .clone();
68            if !record.is_object() {
69                return Err("RECORD `record` is not an object".to_string());
70            }
71            Ok(SingerMessage::Record { stream, record })
72        }
73        "SCHEMA" => {
74            let stream = obj
75                .get("stream")
76                .and_then(Value::as_str)
77                .ok_or_else(|| "SCHEMA missing `stream`".to_string())?
78                .to_string();
79            let schema = obj
80                .get("schema")
81                .ok_or_else(|| "SCHEMA missing `schema`".to_string())?
82                .clone();
83            Ok(SingerMessage::Schema {
84                stream,
85                schema,
86                key_properties: string_array(obj.get("key_properties")),
87                bookmark_properties: string_array(obj.get("bookmark_properties")),
88            })
89        }
90        "STATE" => {
91            let value = obj
92                .get("value")
93                .ok_or_else(|| "STATE missing `value`".to_string())?
94                .clone();
95            Ok(SingerMessage::State { value })
96        }
97        other => Ok(SingerMessage::Other {
98            message_type: other.to_string(),
99        }),
100    }
101}
102
103#[derive(Serialize)]
104struct SchemaLine<'a> {
105    #[serde(rename = "type")]
106    kind: &'static str,
107    stream: &'a str,
108    schema: &'a Value,
109    key_properties: &'a [String],
110}
111
112#[derive(Serialize)]
113struct RecordLine<'a> {
114    #[serde(rename = "type")]
115    kind: &'static str,
116    stream: &'a str,
117    record: &'a Value,
118    #[serde(skip_serializing_if = "Option::is_none")]
119    version: Option<i64>,
120}
121
122#[derive(Serialize)]
123struct StateLine<'a> {
124    #[serde(rename = "type")]
125    kind: &'static str,
126    value: &'a Value,
127}
128
129#[derive(Serialize)]
130struct ActivateVersionLine<'a> {
131    #[serde(rename = "type")]
132    kind: &'static str,
133    stream: &'a str,
134    version: i64,
135}
136
137fn push_line<T: Serialize>(buf: &mut Vec<u8>, line: &T) {
138    serde_json::to_writer(&mut *buf, line).expect("Singer message serialization is infallible");
139    buf.push(b'\n');
140}
141
142/// Append a `SCHEMA` message for `stream` to `buf`.
143pub fn write_schema(buf: &mut Vec<u8>, stream: &str, schema: &Value, key_properties: &[String]) {
144    push_line(
145        buf,
146        &SchemaLine {
147            kind: "SCHEMA",
148            stream,
149            schema,
150            key_properties,
151        },
152    );
153}
154
155/// Append a `RECORD` message to `buf`. `version` tags the record with a table
156/// version for `ACTIVATE_VERSION`-driven full refreshes.
157pub fn write_record(buf: &mut Vec<u8>, stream: &str, record: &Value, version: Option<i64>) {
158    push_line(
159        buf,
160        &RecordLine {
161            kind: "RECORD",
162            stream,
163            record,
164            version,
165        },
166    );
167}
168
169/// Append a `STATE` message carrying `value` to `buf`.
170pub fn write_state(buf: &mut Vec<u8>, value: &Value) {
171    push_line(
172        buf,
173        &StateLine {
174            kind: "STATE",
175            value,
176        },
177    );
178}
179
180/// Append an `ACTIVATE_VERSION` message to `buf`: the target makes `version`
181/// the live version of `stream` and discards rows from older versions.
182pub fn write_activate_version(buf: &mut Vec<u8>, stream: &str, version: i64) {
183    push_line(
184        buf,
185        &ActivateVersionLine {
186            kind: "ACTIVATE_VERSION",
187            stream,
188            version,
189        },
190    );
191}
192
193/// Extract an optional `["a","b"]` string array from a JSON field.
194fn string_array(v: Option<&Value>) -> Option<Vec<String>> {
195    v.and_then(Value::as_array).map(|arr| {
196        arr.iter()
197            .filter_map(|x| x.as_str().map(str::to_string))
198            .collect()
199    })
200}
201
202#[cfg(test)]
203mod tests {
204    use super::*;
205    use serde_json::json;
206
207    #[test]
208    fn parses_record() {
209        let m = parse_line(r#"{"type":"RECORD","stream":"s","record":{"id":1}}"#).unwrap();
210        assert_eq!(
211            m,
212            SingerMessage::Record {
213                stream: "s".into(),
214                record: json!({"id": 1})
215            }
216        );
217    }
218
219    #[test]
220    fn parses_schema_with_key_props() {
221        let m = parse_line(
222            r#"{"type":"SCHEMA","stream":"s","schema":{"type":"object"},"key_properties":["id"],"bookmark_properties":["updated_at"]}"#,
223        )
224        .unwrap();
225        match m {
226            SingerMessage::Schema {
227                stream,
228                key_properties,
229                bookmark_properties,
230                ..
231            } => {
232                assert_eq!(stream, "s");
233                assert_eq!(key_properties, Some(vec!["id".to_string()]));
234                assert_eq!(bookmark_properties, Some(vec!["updated_at".to_string()]));
235            }
236            _ => panic!("expected SCHEMA"),
237        }
238    }
239
240    #[test]
241    fn parses_state() {
242        let m = parse_line(r#"{"type":"STATE","value":{"bookmarks":{"s":{"v":5}}}}"#).unwrap();
243        assert_eq!(
244            m,
245            SingerMessage::State {
246                value: json!({"bookmarks":{"s":{"v":5}}})
247            }
248        );
249    }
250
251    #[test]
252    fn activate_version_and_batch_are_other() {
253        assert_eq!(
254            parse_line(r#"{"type":"ACTIVATE_VERSION","stream":"s","version":1}"#).unwrap(),
255            SingerMessage::Other {
256                message_type: "ACTIVATE_VERSION".into()
257            }
258        );
259        assert_eq!(
260            parse_line(r#"{"type":"BATCH","stream":"s"}"#).unwrap(),
261            SingerMessage::Other {
262                message_type: "BATCH".into()
263            }
264        );
265    }
266
267    #[test]
268    fn malformed_variants_error() {
269        assert!(parse_line("").is_err()); // blank
270        assert!(parse_line("not json").is_err()); // non-json
271        assert!(parse_line("[1,2,3]").is_err()); // not an object
272        assert!(parse_line(r#"{"stream":"s"}"#).is_err()); // no type
273        assert!(parse_line(r#"{"type":"RECORD","stream":"s"}"#).is_err()); // no record
274        assert!(parse_line(r#"{"type":"RECORD","record":{"id":1}}"#).is_err()); // no stream
275        assert!(parse_line(r#"{"type":"RECORD","stream":"s","record":5}"#).is_err()); // record not obj
276        assert!(parse_line(r#"{"type":"STATE"}"#).is_err()); // no value
277        assert!(parse_line(r#"{"type":"SCHEMA","stream":"s"}"#).is_err()); // no schema
278    }
279    #[test]
280    fn encoders_round_trip_through_the_parser() {
281        let mut buf = Vec::new();
282        write_schema(
283            &mut buf,
284            "s",
285            &json!({"type": "object"}),
286            &["id".to_string()],
287        );
288        write_record(&mut buf, "s", &json!({"id": 1}), None);
289        write_record(&mut buf, "s", &json!({"id": 2}), Some(7));
290        write_state(&mut buf, &json!({"k": 1}));
291        write_activate_version(&mut buf, "s", 7);
292        let text = String::from_utf8(buf).unwrap();
293        let lines: Vec<&str> = text.lines().collect();
294        assert_eq!(lines.len(), 5);
295        assert!(text.ends_with('\n'));
296        assert_eq!(
297            parse_line(lines[0]).unwrap(),
298            SingerMessage::Schema {
299                stream: "s".into(),
300                schema: json!({"type": "object"}),
301                key_properties: Some(vec!["id".into()]),
302                bookmark_properties: None,
303            }
304        );
305        assert_eq!(
306            parse_line(lines[1]).unwrap(),
307            SingerMessage::Record {
308                stream: "s".into(),
309                record: json!({"id": 1})
310            }
311        );
312        assert!(!lines[1].contains("version"));
313        let v: Value = serde_json::from_str(lines[2]).unwrap();
314        assert_eq!(v["version"], json!(7));
315        assert_eq!(
316            parse_line(lines[3]).unwrap(),
317            SingerMessage::State {
318                value: json!({"k": 1})
319            }
320        );
321        let v: Value = serde_json::from_str(lines[4]).unwrap();
322        assert_eq!(
323            v,
324            json!({"type": "ACTIVATE_VERSION", "stream": "s", "version": 7})
325        );
326    }
327}
faucet_common_singer/message.rs

faucet_common_singer/
message.rs
