mysql_es/
view_repository.rs

1use std::marker::PhantomData;
2
3use async_trait::async_trait;
4use cqrs_es::persist::{PersistenceError, ViewContext, ViewRepository};
5use cqrs_es::{Aggregate, View};
6use sqlx::mysql::MySqlRow;
7use sqlx::{MySql, Pool, Row};
8
9use crate::error::MysqlAggregateError;
10
11/// A mysql backed query repository for use in backing a `GenericQuery`.
12pub struct MysqlViewRepository<V, A> {
13    insert_sql: String,
14    update_sql: String,
15    select_sql: String,
16    pool: Pool<MySql>,
17    _phantom: PhantomData<(V, A)>,
18}
19
20impl<V, A> MysqlViewRepository<V, A>
21where
22    V: View<A>,
23    A: Aggregate,
24{
25    /// Creates a new `MysqlViewRepository` that will store serialized views in a MySql table named
26    /// identically to the `view_name` value provided. This table should be created by the user
27    /// before using this query repository (see `/db/init.sql` sql initialization file).
28    ///
29    /// ```
30    /// # use cqrs_es::doc::MyAggregate;
31    /// # use cqrs_es::persist::doc::MyView;
32    /// use sqlx::{MySql, Pool};
33    /// use mysql_es::MysqlViewRepository;
34    ///
35    /// fn configure_view_repo(pool: Pool<MySql>) -> MysqlViewRepository<MyView,MyAggregate> {
36    ///     MysqlViewRepository::new("my_view_table", pool)
37    /// }
38    /// ```
39    pub fn new(view_name: &str, pool: Pool<MySql>) -> Self {
40        let insert_sql = format!(
41            "INSERT INTO {} (payload, version, view_id) VALUES ( ?, ?, ? )",
42            view_name
43        );
44        let update_sql = format!(
45            "UPDATE {} SET payload= ? , version= ? WHERE view_id= ?",
46            view_name
47        );
48        let select_sql = format!("SELECT version,payload FROM {} WHERE view_id= ?", view_name);
49        Self {
50            insert_sql,
51            update_sql,
52            select_sql,
53            pool,
54            _phantom: Default::default(),
55        }
56    }
57}
58
59#[async_trait]
60impl<V, A> ViewRepository<V, A> for MysqlViewRepository<V, A>
61where
62    V: View<A>,
63    A: Aggregate,
64{
65    async fn load(&self, view_id: &str) -> Result<Option<V>, PersistenceError> {
66        let row: Option<MySqlRow> = sqlx::query(&self.select_sql)
67            .bind(view_id)
68            .fetch_optional(&self.pool)
69            .await
70            .map_err(MysqlAggregateError::from)?;
71        match row {
72            None => Ok(None),
73            Some(row) => {
74                let view = serde_json::from_value(row.get("payload"))?;
75                Ok(Some(view))
76            }
77        }
78    }
79
80    async fn load_with_context(
81        &self,
82        view_id: &str,
83    ) -> Result<Option<(V, ViewContext)>, PersistenceError> {
84        let row: Option<MySqlRow> = sqlx::query(&self.select_sql)
85            .bind(view_id)
86            .fetch_optional(&self.pool)
87            .await
88            .map_err(MysqlAggregateError::from)?;
89        match row {
90            None => Ok(None),
91            Some(row) => {
92                let version = row.get("version");
93                let view = serde_json::from_value(row.get("payload"))?;
94                let view_context = ViewContext::new(view_id.to_string(), version);
95                Ok(Some((view, view_context)))
96            }
97        }
98    }
99
100    async fn update_view(&self, view: V, context: ViewContext) -> Result<(), PersistenceError> {
101        let sql = match context.version {
102            0 => &self.insert_sql,
103            _ => &self.update_sql,
104        };
105        let version = context.version + 1;
106        let payload = serde_json::to_value(&view).map_err(MysqlAggregateError::from)?;
107        sqlx::query(sql.as_str())
108            .bind(payload)
109            .bind(version)
110            .bind(context.view_instance_id)
111            .execute(&self.pool)
112            .await
113            .map_err(MysqlAggregateError::from)?;
114        Ok(())
115    }
116}
117
118#[cfg(test)]
119mod test {
120    use crate::testing::tests::{
121        Created, TestAggregate, TestEvent, TestView, TEST_CONNECTION_STRING,
122    };
123    use crate::{default_mysql_pool, MysqlViewRepository};
124    use cqrs_es::persist::{ViewContext, ViewRepository};
125
126    #[tokio::test]
127    async fn test_valid_view_repository() {
128        let pool = default_mysql_pool(TEST_CONNECTION_STRING).await;
129        let repo = MysqlViewRepository::<TestView, TestAggregate>::new("test_view", pool.clone());
130        let test_view_id = uuid::Uuid::new_v4().to_string();
131
132        let view = TestView {
133            events: vec![TestEvent::Created(Created {
134                id: "just a test event for this view".to_string(),
135            })],
136        };
137        repo.update_view(view.clone(), ViewContext::new(test_view_id.to_string(), 0))
138            .await
139            .unwrap();
140        let (found, context) = repo
141            .load_with_context(&test_view_id)
142            .await
143            .unwrap()
144            .unwrap();
145        assert_eq!(found, view);
146
147        let updated_view = TestView {
148            events: vec![TestEvent::Created(Created {
149                id: "a totally different view".to_string(),
150            })],
151        };
152        repo.update_view(updated_view.clone(), context)
153            .await
154            .unwrap();
155        let found_option = repo.load(&test_view_id).await.unwrap();
156        let found = found_option.unwrap();
157
158        assert_eq!(found, updated_view);
159    }
160}
mysql_es/view_repository.rs

mysql_es/
view_repository.rs
