dynamo_es/
view_repository.rs

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

dynamo_es/
view_repository.rs
