mongo_es/
view_repository.rs

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

mongo_es/
view_repository.rs
