mongo_es/
view_repository.rs

1use cqrs_es::persist::{PersistenceError, ViewContext, ViewRepository};
2use cqrs_es::{Aggregate, View};
3use mongodb::bson::{self, doc, Document};
4
5use crate::utils::load_view;
6
7pub struct MongoViewRepository<V, A> {
8    _phantom: std::marker::PhantomData<(V, A)>,
9    view_name: String,
10    client: mongodb::Client,
11}
12
13impl<V, A> MongoViewRepository<V, A>
14where
15    V: View<A>,
16    A: Aggregate,
17{
18    pub fn new(view_name: &str, client: mongodb::Client) -> Self {
19        Self {
20            _phantom: Default::default(),
21            view_name: view_name.to_string(),
22            client,
23        }
24    }
25}
26
27impl<V, A> ViewRepository<V, A> for MongoViewRepository<V, A>
28where
29    V: View<A>,
30    A: Aggregate,
31{
32    async fn load(&self, view_id: &str) -> Result<Option<V>, PersistenceError> {
33        let result = load_view(&self.client, &self.view_name, view_id).await?;
34        let document = match result {
35            Some(item) => item,
36            None => return Ok(None),
37        };
38
39        let payload = bson::from_bson(document.get("payload").unwrap().clone()).unwrap();
40        let view: V = serde_json::from_value(payload)?;
41        Ok(Some(view))
42    }
43
44    async fn load_with_context(
45        &self,
46        view_id: &str,
47    ) -> Result<Option<(V, ViewContext)>, PersistenceError> {
48        let result = load_view(&self.client, &self.view_name, view_id).await?;
49        let document = match result {
50            Some(item) => item,
51            None => return Ok(None),
52        };
53
54        let version = document.get_i64("version").unwrap_or(0);
55        let payload = bson::from_bson(document.get("payload").unwrap().clone()).unwrap();
56        let view: V = serde_json::from_value(payload)?;
57        let context = ViewContext::new(view_id.to_string(), version);
58        Ok(Some((view, context)))
59    }
60
61    async fn update_view(&self, view: V, context: ViewContext) -> Result<(), PersistenceError> {
62        let collection = self
63            .client
64            .default_database()
65            .expect("Default database not configured")
66            .collection::<Document>(&self.view_name);
67
68        let view_id = context.view_instance_id;
69
70        let filter = doc! { "view_id": &view_id };
71        let update = doc! {
72            "$set": {
73                "payload": bson::to_bson(&view).unwrap(),
74                "version": context.version + 1,
75            }
76        };
77
78        let res = collection
79            .update_one(filter, update)
80            .upsert(true)
81            .await
82            .expect("Failed to update view");
83
84        log::debug!(
85            "Modified {} documents in `{}` collection",
86            res.modified_count,
87            &self.view_name
88        );
89
90        Ok(())
91    }
92}
93
94#[cfg(test)]
95mod tests {
96    use cqrs_es::doc::{Customer, CustomerEvent};
97    use cqrs_es::persist::{ViewContext, ViewRepository};
98
99    use crate::utils::tests::{mongodb_client, CustomerView};
100    use crate::MongoViewRepository;
101
102    #[tokio::test]
103    async fn test_view_repository() {
104        let repository =
105            MongoViewRepository::<CustomerView, Customer>::new("test_view", mongodb_client().await);
106
107        let test_view_id = uuid::Uuid::new_v4().to_string();
108
109        let view = CustomerView {
110            events: vec![CustomerEvent::NameAdded {
111                name: "Ferris".to_string(),
112            }],
113        };
114
115        repository
116            .update_view(view.clone(), ViewContext::new(test_view_id.to_string(), 0))
117            .await
118            .unwrap();
119
120        let found = repository.load(&test_view_id).await.unwrap().unwrap();
121
122        assert_eq!(found, view);
123    }
124}
mongo_es/view_repository.rs

mongo_es/
view_repository.rs
