faucet_source_mongodb_cdc/
state.rs

1//! State key derivation and resume-token bookmark serialization.
2//!
3//! Bookmark shape (round-trips through `serde_json::Value`):
4//!
5//! ```json
6//! { "resume_token": { "_data": "8264..." } }
7//! ```
8
9use crate::config::Scope;
10use faucet_core::FaucetError;
11use mongodb::bson::{self, Bson};
12use mongodb::change_stream::event::ResumeToken;
13use serde::{Deserialize, Serialize};
14use serde_json::Value;
15
16/// State key for a given watch scope.
17///
18/// `cluster` → `mongodb-cdc:cluster`
19/// `database` → `mongodb-cdc:db:<database>`
20/// `collection` → `mongodb-cdc:coll:<database>.<collection>`
21pub fn state_key(scope: &Scope) -> String {
22    match scope {
23        Scope::Cluster => "mongodb-cdc:cluster".to_string(),
24        Scope::Database { database } => format!("mongodb-cdc:db:{database}"),
25        Scope::Collection {
26            database,
27            collection,
28        } => {
29            format!("mongodb-cdc:coll:{database}.{collection}")
30        }
31    }
32}
33
34/// Durable bookmark wrapping a resume token.
35#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
36pub struct Bookmark {
37    /// The opaque resume token as relaxed extended JSON (e.g. `{"_data": "..."}`).
38    pub resume_token: Value,
39}
40
41impl Bookmark {
42    /// Build a bookmark from a driver `ResumeToken`.
43    pub fn from_token(token: &ResumeToken) -> Result<Self, FaucetError> {
44        let bson = bson::to_bson(token)
45            .map_err(|e| FaucetError::State(format!("mongodb-cdc resume token serialize: {e}")))?;
46        Ok(Self {
47            resume_token: bson.into_relaxed_extjson(),
48        })
49    }
50
51    /// Parse a bookmark previously emitted by `to_value`.
52    pub fn from_value(v: Value) -> Result<Self, FaucetError> {
53        serde_json::from_value(v)
54            .map_err(|e| FaucetError::State(format!("mongodb-cdc bookmark parse: {e}")))
55    }
56
57    /// Serialize for the state store.
58    pub fn to_value(&self) -> Result<Value, FaucetError> {
59        serde_json::to_value(self)
60            .map_err(|e| FaucetError::State(format!("mongodb-cdc bookmark serialize: {e}")))
61    }
62
63    /// Decode the stored resume token back into a driver `ResumeToken`.
64    pub fn to_token(&self) -> Result<ResumeToken, FaucetError> {
65        let bson: Bson = bson::to_bson(&self.resume_token)
66            .map_err(|e| FaucetError::State(format!("mongodb-cdc resume token to bson: {e}")))?;
67        bson::from_bson(bson)
68            .map_err(|e| FaucetError::State(format!("mongodb-cdc resume token decode: {e}")))
69    }
70}
71
72#[cfg(test)]
73mod tests {
74    use super::*;
75    use serde_json::json;
76
77    #[test]
78    fn state_key_cluster() {
79        assert_eq!(state_key(&Scope::Cluster), "mongodb-cdc:cluster");
80    }
81
82    #[test]
83    fn state_key_database() {
84        assert_eq!(
85            state_key(&Scope::Database {
86                database: "app".into()
87            }),
88            "mongodb-cdc:db:app"
89        );
90    }
91
92    #[test]
93    fn state_key_collection() {
94        assert_eq!(
95            state_key(&Scope::Collection {
96                database: "app".into(),
97                collection: "users".into()
98            }),
99            "mongodb-cdc:coll:app.users"
100        );
101    }
102
103    #[test]
104    fn bookmark_value_round_trip() {
105        let b = Bookmark {
106            resume_token: json!({ "_data": "8264AB" }),
107        };
108        let v = b.to_value().unwrap();
109        let parsed = Bookmark::from_value(v).unwrap();
110        assert_eq!(parsed, b);
111    }
112
113    #[test]
114    fn bookmark_token_round_trip() {
115        let b = Bookmark {
116            resume_token: json!({ "_data": "8264AB00" }),
117        };
118        let token = b.to_token().unwrap();
119        let b2 = Bookmark::from_token(&token).unwrap();
120        assert_eq!(b2.resume_token["_data"], json!("8264AB00"));
121    }
122
123    #[test]
124    fn from_value_rejects_garbage() {
125        assert!(Bookmark::from_value(json!({})).is_err());
126        assert!(Bookmark::from_value(json!("bare")).is_err());
127    }
128}
faucet_source_mongodb_cdc/state.rs

faucet_source_mongodb_cdc/
state.rs
