sqlite_es/
view_repository.rs

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

sqlite_es/
view_repository.rs
