dynamo_es/
view_repository.rs

1use std::marker::PhantomData;
2
3use aws_sdk_dynamodb::primitives::Blob;
4use aws_sdk_dynamodb::types::{AttributeValue, Put, TransactWriteItem};
5use cqrs_es::persist::{PersistenceError, ViewContext, ViewRepository};
6use cqrs_es::{Aggregate, View};
7
8use crate::helpers::{att_as_number, att_as_value, commit_transactions, load_dynamo_view};
9
10/// A DynamoDb backed view repository for use in backing a `GenericQuery`.
11pub struct DynamoViewRepository<V, A> {
12    _phantom: PhantomData<(V, A)>,
13    view_name: String,
14    client: aws_sdk_dynamodb::client::Client,
15}
16
17impl<V, A> DynamoViewRepository<V, A>
18where
19    V: View<A>,
20    A: Aggregate,
21{
22    /// Creates a new `DynamoViewRepository` that will store serialized views in a DynamoDb table named
23    /// identically to the `view_name` value provided. This table should be created by the user
24    /// before using this query repository (see `Makefile` for `create-table` command
25    /// line example).
26    ///
27    /// ```
28    /// # use cqrs_es::doc::MyAggregate;
29    /// # use cqrs_es::persist::doc::MyView;
30    /// use aws_sdk_dynamodb::Client;
31    /// use dynamo_es::DynamoViewRepository;
32    ///
33    /// fn configure_view_repo(client: Client) -> DynamoViewRepository<MyView,MyAggregate> {
34    ///     DynamoViewRepository::new("my_view_table", client)
35    /// }
36    /// ```
37    pub fn new(view_name: &str, client: aws_sdk_dynamodb::client::Client) -> Self {
38        Self {
39            _phantom: PhantomData,
40            view_name: view_name.to_string(),
41            client,
42        }
43    }
44}
45
46impl<V, A> ViewRepository<V, A> for DynamoViewRepository<V, A>
47where
48    V: View<A>,
49    A: Aggregate,
50{
51    async fn load(&self, view_id: &str) -> Result<Option<V>, PersistenceError> {
52        let query_result = load_dynamo_view(&self.client, &self.view_name, view_id).await?;
53        let query_items = match query_result.items {
54            None => return Ok(None),
55            Some(items) => items,
56        };
57        let query_item = match query_items.first() {
58            None => return Ok(None),
59            Some(item) => item,
60        };
61        let payload = att_as_value(query_item, "Payload")?;
62        let view: V = serde_json::from_value(payload)?;
63        Ok(Some(view))
64    }
65
66    async fn load_with_context(
67        &self,
68        view_id: &str,
69    ) -> Result<Option<(V, ViewContext)>, PersistenceError> {
70        let query_result = load_dynamo_view(&self.client, &self.view_name, view_id).await?;
71        let query_items = match query_result.items {
72            None => return Ok(None),
73            Some(items) => items,
74        };
75        let query_item = match query_items.first() {
76            None => {
77                let view = V::default();
78                let context = ViewContext::new(view_id.to_string(), 0);
79                return Ok(Some((view, context)));
80            }
81            Some(item) => item,
82        };
83        let version = att_as_number(query_item, "ViewVersion")?;
84        let payload = att_as_value(query_item, "Payload")?;
85        let view: V = serde_json::from_value(payload)?;
86        let context = ViewContext::new(view_id.to_string(), version as i64);
87        Ok(Some((view, context)))
88    }
89
90    async fn update_view(&self, view: V, context: ViewContext) -> Result<(), PersistenceError> {
91        let view_id = AttributeValue::S(String::from(&context.view_instance_id));
92        let expected_view_version = AttributeValue::N(context.version.to_string());
93        let view_version = AttributeValue::N((context.version + 1).to_string());
94        let payload_blob = serde_json::to_vec(&view).unwrap();
95        let payload = AttributeValue::B(Blob::new(payload_blob));
96        let transaction = TransactWriteItem::builder()
97            .put(Put::builder()
98                .table_name(&self.view_name)
99                .item("ViewId", view_id)
100                .item("ViewVersion", view_version)
101                .item("Payload", payload)
102                .condition_expression("attribute_not_exists(ViewVersion) OR (ViewVersion  = :expected_view_version)")
103                .expression_attribute_values(":expected_view_version", expected_view_version)
104                .build()
105                .map_err(|e|PersistenceError::UnknownError(Box::new(e)))?)
106            .build();
107        commit_transactions(&self.client, vec![transaction]).await?;
108        Ok(())
109    }
110}
111
112#[cfg(test)]
113mod test {
114    use cqrs_es::persist::{ViewContext, ViewRepository};
115
116    use crate::testing::tests::{
117        test_dynamodb_client, Created, TestAggregate, TestEvent, TestView,
118    };
119    use crate::DynamoViewRepository;
120
121    #[tokio::test]
122    async fn test_valid_view_repository() {
123        let repo = DynamoViewRepository::<TestView, TestAggregate>::new(
124            "TestViewTable",
125            test_dynamodb_client().await,
126        );
127        let test_view_id = uuid::Uuid::new_v4().to_string();
128
129        let view = TestView {
130            events: vec![TestEvent::Created(Created {
131                id: "just a test event for this view".to_string(),
132            })],
133        };
134        repo.update_view(view.clone(), ViewContext::new(test_view_id.to_string(), 0))
135            .await
136            .unwrap();
137        let (found, context) = repo
138            .load_with_context(&test_view_id)
139            .await
140            .unwrap()
141            .unwrap();
142        assert_eq!(found, view);
143
144        let updated_view = TestView {
145            events: vec![TestEvent::Created(Created {
146                id: "a totally different view".to_string(),
147            })],
148        };
149        repo.update_view(updated_view.clone(), context)
150            .await
151            .unwrap();
152        let found_option = repo.load(&test_view_id).await.unwrap();
153        let found = found_option.unwrap();
154
155        assert_eq!(found, updated_view);
156    }
157}
dynamo_es/view_repository.rs

dynamo_es/
view_repository.rs
