Merge branch 'main' into feature/birthday-calendar

crates/store_sqlite/Cargo.toml

@@ -1,6 +1,7 @@
 [package]
 name = "rustical_store_sqlite"
 version.workspace = true
+rust-version.workspace = true
 edition.workspace = true
 description.workspace = true
 repository.workspace = true
@@ -15,12 +16,12 @@ rstest.workspace = true
 
 [dependencies]
 tokio.workspace = true
-rustical_store = { workspace = true }
-async-trait = { workspace = true }
-serde = { workspace = true }
-sqlx = { workspace = true }
-thiserror = { workspace = true }
-tracing = { workspace = true }
+rustical_store.workspace = true
+async-trait.workspace = true
+serde.workspace = true
+sqlx.workspace = true
+thiserror.workspace = true
+tracing.workspace = true
 derive_more.workspace = true
 chrono.workspace = true
 password-auth.workspace = true

crates/store_sqlite/src/addressbook_store/mod.rs

@@ -9,7 +9,7 @@ use rustical_store::{
 };
 use sqlx::{Acquire, Executor, Sqlite, SqlitePool, Transaction};
 use tokio::sync::mpsc::Sender;
-use tracing::{error, instrument};
+use tracing::{error, instrument, warn};
 
 pub mod birthday_calendar;
 
@@ -34,6 +34,60 @@ pub struct SqliteAddressbookStore {
 }
 
 impl SqliteAddressbookStore {
+    // Commit "orphaned" objects to the changelog table
+    pub async fn repair_orphans(&self) -> Result<(), Error> {
+        struct Row {
+            principal: String,
+            addressbook_id: String,
+            id: String,
+            deleted: bool,
+        }
+
+        let mut tx = self
+            .db
+            .begin_with(BEGIN_IMMEDIATE)
+            .await
+            .map_err(crate::Error::from)?;
+
+        let rows = sqlx::query_as!(
+            Row,
+            r#"
+                SELECT principal, addressbook_id, id, (deleted_at IS NOT NULL) AS "deleted: bool"
+                    FROM addressobjects
+                    WHERE (principal, addressbook_id, id) NOT IN (
+                        SELECT DISTINCT principal, addressbook_id, object_id FROM addressobjectchangelog
+                    )
+                ;
+                "#,
+        )
+        .fetch_all(&mut *tx)
+        .await
+        .map_err(crate::Error::from)?;
+
+        for row in rows {
+            let operation = if row.deleted {
+                ChangeOperation::Delete
+            } else {
+                ChangeOperation::Add
+            };
+            warn!(
+                "Commiting orphaned addressbook object ({},{},{}), deleted={}",
+                &row.principal, &row.addressbook_id, &row.id, &row.deleted
+            );
+            log_object_operation(
+                &mut tx,
+                &row.principal,
+                &row.addressbook_id,
+                &row.id,
+                operation,
+            )
+            .await?;
+        }
+        tx.commit().await.map_err(crate::Error::from)?;
+
+        Ok(())
+    }
+
     async fn _get_addressbook<'e, E: Executor<'e, Database = Sqlite>>(
         executor: E,
         principal: &str,
@@ -92,9 +146,9 @@ impl SqliteAddressbookStore {
 
     async fn _update_addressbook<'e, E: Executor<'e, Database = Sqlite>>(
         executor: E,
-        principal: String,
-        id: String,
-        addressbook: Addressbook,
+        principal: &str,
+        id: &str,
+        addressbook: &Addressbook,
     ) -> Result<(), rustical_store::Error> {
         let result = sqlx::query!(
             r#"UPDATE addressbooks SET principal = ?, id = ?, displayname = ?, description = ?, push_topic = ?
@@ -399,7 +453,7 @@ impl AddressbookStore for SqliteAddressbookStore {
         id: String,
         addressbook: Addressbook,
     ) -> Result<(), rustical_store::Error> {
-        Self::_update_addressbook(&self.db, principal, id, addressbook).await
+        Self::_update_addressbook(&self.db, &principal, &id, &addressbook).await
     }
 
     #[instrument]
@@ -631,7 +685,7 @@ impl AddressbookStore for SqliteAddressbookStore {
                 .await?
                 .push_topic,
         }) {
-            error!("Push notification about deleted addressbook failed: {err}");
+            error!("Push notification about restored addressbook object failed: {err}");
         }
 
         Ok(())
@@ -665,6 +719,7 @@ impl AddressbookStore for SqliteAddressbookStore {
             Self::_insert_addressbook(&mut *tx, &addressbook).await?;
         }
 
+        let mut sync_token = None;
         for object in objects {
             Self::_put_object(
                 &mut *tx,
@@ -674,9 +729,31 @@ impl AddressbookStore for SqliteAddressbookStore {
                 false,
             )
             .await?;
+
+            sync_token = Some(
+                log_object_operation(
+                    &mut tx,
+                    &addressbook.principal,
+                    &addressbook.id,
+                    object.get_id(),
+                    ChangeOperation::Add,
+                )
+                .await?,
+            );
         }
 
         tx.commit().await.map_err(crate::Error::from)?;
+        if let Some(sync_token) = sync_token
+            && let Err(err) = self.sender.try_send(CollectionOperation {
+                data: CollectionOperationInfo::Content { sync_token },
+                topic: self
+                    .get_addressbook(&addressbook.principal, &addressbook.id, true)
+                    .await?
+                    .push_topic,
+            })
+        {
+            error!("Push notification about imported addressbook failed: {err}");
+        }
         Ok(())
     }
 }
@@ -688,7 +765,7 @@ async fn log_object_operation(
     addressbook_id: &str,
     object_id: &str,
     operation: ChangeOperation,
-) -> Result<String, sqlx::Error> {
+) -> Result<String, Error> {
     struct Synctoken {
         synctoken: i64,
     }
@@ -703,7 +780,8 @@ async fn log_object_operation(
         addressbook_id
     )
     .fetch_one(&mut **tx)
-    .await?;
+    .await
+    .map_err(crate::Error::from)?;
 
     sqlx::query!(
         r#"
@@ -717,6 +795,7 @@ async fn log_object_operation(
         operation
     )
     .execute(&mut **tx)
-    .await?;
+    .await
+    .map_err(crate::Error::from)?;
     Ok(format_synctoken(synctoken))
 }

crates/store_sqlite/src/calendar_store.rs

@@ -11,7 +11,7 @@ use rustical_store::{CollectionOperation, CollectionOperationInfo};
 use sqlx::types::chrono::NaiveDateTime;
 use sqlx::{Acquire, Executor, Sqlite, SqlitePool, Transaction};
 use tokio::sync::mpsc::Sender;
-use tracing::{error, instrument};
+use tracing::{error, instrument, warn};
 
 #[derive(Debug, Clone)]
 struct CalendarObjectRow {
@@ -94,6 +94,53 @@ pub struct SqliteCalendarStore {
 }
 
 impl SqliteCalendarStore {
+    // Commit "orphaned" objects to the changelog table
+    pub async fn repair_orphans(&self) -> Result<(), Error> {
+        struct Row {
+            principal: String,
+            cal_id: String,
+            id: String,
+            deleted: bool,
+        }
+
+        let mut tx = self
+            .db
+            .begin_with(BEGIN_IMMEDIATE)
+            .await
+            .map_err(crate::Error::from)?;
+
+        let rows = sqlx::query_as!(
+            Row,
+            r#"
+                SELECT principal, cal_id, id, (deleted_at IS NOT NULL) AS "deleted: bool"
+                    FROM calendarobjects
+                    WHERE (principal, cal_id, id) NOT IN (
+                        SELECT DISTINCT principal, cal_id, object_id FROM calendarobjectchangelog
+                    )
+                ;
+                "#,
+        )
+        .fetch_all(&mut *tx)
+        .await
+        .map_err(crate::Error::from)?;
+
+        for row in rows {
+            let operation = if row.deleted {
+                ChangeOperation::Delete
+            } else {
+                ChangeOperation::Add
+            };
+            warn!(
+                "Commiting orphaned calendar object ({},{},{}), deleted={}",
+                &row.principal, &row.cal_id, &row.id, &row.deleted
+            );
+            log_object_operation(&mut tx, &row.principal, &row.cal_id, &row.id, operation).await?;
+        }
+        tx.commit().await.map_err(crate::Error::from)?;
+
+        Ok(())
+    }
+
     async fn _get_calendar<'e, E: Executor<'e, Database = Sqlite>>(
         executor: E,
         principal: &str,
@@ -351,9 +398,9 @@ impl SqliteCalendarStore {
     #[instrument]
     async fn _put_object<'e, E: Executor<'e, Database = Sqlite>>(
         executor: E,
-        principal: String,
-        cal_id: String,
-        object: CalendarObject,
+        principal: &str,
+        cal_id: &str,
+        object: &CalendarObject,
         overwrite: bool,
     ) -> Result<(), Error> {
         let (object_id, uid, ics) = (object.get_id(), object.get_uid(), object.get_ics());
@@ -600,18 +647,35 @@ impl CalendarStore for SqliteCalendarStore {
             Self::_insert_calendar(&mut *tx, calendar.clone()).await?;
         }
 
+        let mut sync_token = None;
         for object in objects {
-            Self::_put_object(
-                &mut *tx,
-                calendar.principal.clone(),
-                calendar.id.clone(),
-                object,
-                false,
-            )
-            .await?;
+            Self::_put_object(&mut *tx, &calendar.principal, &calendar.id, &object, false).await?;
+
+            sync_token = Some(
+                log_object_operation(
+                    &mut tx,
+                    &calendar.principal,
+                    &calendar.id,
+                    object.get_id(),
+                    ChangeOperation::Add,
+                )
+                .await?,
+            );
         }
 
         tx.commit().await.map_err(crate::Error::from)?;
+
+        if let Some(sync_token) = sync_token
+            && let Err(err) = self.sender.try_send(CollectionOperation {
+                data: CollectionOperationInfo::Content { sync_token },
+                topic: self
+                    .get_calendar(&calendar.principal, &calendar.id, true)
+                    .await?
+                    .push_topic,
+            })
+        {
+            error!("Push notification about imported calendar failed: {err}");
+        }
         Ok(())
     }
 
@@ -689,14 +753,7 @@ impl CalendarStore for SqliteCalendarStore {
             return Err(Error::ReadOnly);
         }
 
-        Self::_put_object(
-            &mut *tx,
-            principal.clone(),
-            cal_id.clone(),
-            object,
-            overwrite,
-        )
-        .await?;
+        Self::_put_object(&mut *tx, &principal, &cal_id, &object, overwrite).await?;
 
         let sync_token = log_object_operation(
             &mut tx,
@@ -774,7 +831,7 @@ impl CalendarStore for SqliteCalendarStore {
             data: CollectionOperationInfo::Content { sync_token },
             topic: self.get_calendar(principal, cal_id, true).await?.push_topic,
         }) {
-            error!("Push notification about deleted calendar failed: {err}");
+            error!("Push notification about restored calendar object failed: {err}");
         }
         Ok(())
     }
@@ -795,6 +852,7 @@ impl CalendarStore for SqliteCalendarStore {
 }
 
 // Logs an operation to the events
+// TODO: Log multiple updates
 async fn log_object_operation(
     tx: &mut Transaction<'_, Sqlite>,
     principal: &str,

crates/store_sqlite/src/tests/addressbook_store.rs

@@ -0,0 +1,81 @@
+#[cfg(test)]
+mod tests {
+    use crate::{addressbook_store::SqliteAddressbookStore, tests::get_test_addressbook_store};
+    use rstest::rstest;
+    use rustical_store::{Addressbook, AddressbookStore};
+
+    #[rstest]
+    #[tokio::test]
+    async fn test_addressbook_store(
+        #[from(get_test_addressbook_store)]
+        #[future]
+        addr_store: SqliteAddressbookStore,
+    ) {
+        let addr_store = addr_store.await;
+
+        let cal = Addressbook {
+            id: "addr".to_string(),
+            principal: "fake-user".to_string(),
+            displayname: None,
+            description: None,
+            deleted_at: None,
+            synctoken: 0,
+            push_topic: "alskdj".to_string(),
+        };
+
+        assert!(
+            addr_store.insert_addressbook(cal).await.is_err(),
+            "This should fail due to the user not existing "
+        );
+
+        let addr = Addressbook {
+            id: "addr".to_string(),
+            principal: "user".to_string(),
+            displayname: None,
+            description: None,
+            deleted_at: None,
+            synctoken: 0,
+            push_topic: "alskdj".to_string(),
+        };
+
+        addr_store.insert_addressbook(addr.clone()).await.unwrap();
+
+        assert_eq!(
+            addr_store
+                .get_addressbook("user", "addr", false)
+                .await
+                .unwrap(),
+            addr
+        );
+
+        addr_store
+            .delete_addressbook("user", "addr", true)
+            .await
+            .unwrap();
+
+        let Err(err) = addr_store.get_addressbook("user", "addr", false).await else {
+            panic!()
+        };
+        assert!(err.is_not_found());
+
+        addr_store
+            .get_addressbook("user", "addr", true)
+            .await
+            .unwrap();
+
+        addr_store
+            .restore_addressbook("user", "addr")
+            .await
+            .unwrap();
+
+        addr_store
+            .delete_addressbook("user", "addr", false)
+            .await
+            .unwrap();
+
+        let Err(err) = addr_store.get_addressbook("user", "addr", true).await else {
+            panic!()
+        };
+        assert!(err.is_not_found());
+    }
+}

crates/store_sqlite/src/tests/calendar_store.rs

@@ -0,0 +1,76 @@
+#[cfg(test)]
+mod tests {
+    use crate::{calendar_store::SqliteCalendarStore, tests::get_test_calendar_store};
+    use rstest::rstest;
+    use rustical_store::{Calendar, CalendarMetadata, CalendarStore};
+
+    #[rstest]
+    #[tokio::test]
+    async fn test_calendar_store(
+        #[from(get_test_calendar_store)]
+        #[future]
+        cal_store: SqliteCalendarStore,
+    ) {
+        let cal_store = cal_store.await;
+
+        let cal = Calendar {
+            principal: "fake-user".to_string(),
+            timezone_id: None,
+            deleted_at: None,
+            meta: CalendarMetadata::default(),
+            id: "cal".to_string(),
+            synctoken: 0,
+            subscription_url: None,
+            push_topic: "alskdj".to_string(),
+            components: vec![],
+        };
+
+        assert!(
+            cal_store.insert_calendar(cal).await.is_err(),
+            "This should fail due to the user not existing "
+        );
+
+        let cal = Calendar {
+            principal: "user".to_string(),
+            timezone_id: None,
+            deleted_at: None,
+            meta: CalendarMetadata::default(),
+            id: "cal".to_string(),
+            synctoken: 0,
+            subscription_url: None,
+            push_topic: "alskdj".to_string(),
+            components: vec![],
+        };
+
+        cal_store.insert_calendar(cal.clone()).await.unwrap();
+
+        assert_eq!(
+            cal_store.get_calendar("user", "cal", false).await.unwrap(),
+            cal
+        );
+
+        cal_store
+            .delete_calendar("user", "cal", true)
+            .await
+            .unwrap();
+
+        let Err(err) = cal_store.get_calendar("user", "cal", false).await else {
+            panic!()
+        };
+        assert!(err.is_not_found());
+
+        cal_store.get_calendar("user", "cal", true).await.unwrap();
+
+        cal_store.restore_calendar("user", "cal").await.unwrap();
+
+        cal_store
+            .delete_calendar("user", "cal", false)
+            .await
+            .unwrap();
+
+        let Err(err) = cal_store.get_calendar("user", "cal", true).await else {
+            panic!()
+        };
+        assert!(err.is_not_found());
+    }
+}

crates/store_sqlite/src/tests/mod.rs

@@ -8,6 +8,9 @@ use tokio::sync::OnceCell;
 
 static DB: OnceCell<SqlitePool> = OnceCell::const_new();
 
+mod addressbook_store;
+mod calendar_store;
+
 async fn get_test_db() -> SqlitePool {
     DB.get_or_init(async || {
         let db = SqlitePool::connect("sqlite::memory:").await.unwrap();