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

sqlite_es/
view_repository.rs
