Add startup routine to fix orphaned objects

fixes #145, related to #142

.sqlx/query-053c17f3b54ae3e153137926115486eb19a801bd73a74230bcf72a9a7254824a.json

@@ -0,0 +1,38 @@
+{
+  "db_name": "SQLite",
+  "query": "\n                SELECT principal, cal_id, id, (deleted_at IS NOT NULL) AS \"deleted: bool\"\n                    FROM calendarobjects\n                    WHERE (principal, cal_id, id) NOT IN (\n                        SELECT DISTINCT principal, cal_id, object_id FROM calendarobjectchangelog\n                    )\n                ;\n                ",
+  "describe": {
+    "columns": [
+      {
+        "name": "principal",
+        "ordinal": 0,
+        "type_info": "Text"
+      },
+      {
+        "name": "cal_id",
+        "ordinal": 1,
+        "type_info": "Text"
+      },
+      {
+        "name": "id",
+        "ordinal": 2,
+        "type_info": "Text"
+      },
+      {
+        "name": "deleted: bool",
+        "ordinal": 3,
+        "type_info": "Integer"
+      }
+    ],
+    "parameters": {
+      "Right": 0
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "053c17f3b54ae3e153137926115486eb19a801bd73a74230bcf72a9a7254824a"
+}

.sqlx/query-c138b1143ac04af4930266ffae0990e82005911c11a683ad565e92335e085f4d.json

@@ -0,0 +1,38 @@
+{
+  "db_name": "SQLite",
+  "query": "\n                SELECT principal, addressbook_id, id, (deleted_at IS NOT NULL) AS \"deleted: bool\"\n                    FROM addressobjects\n                    WHERE (principal, addressbook_id, id) NOT IN (\n                        SELECT DISTINCT principal, addressbook_id, object_id FROM addressobjectchangelog\n                    )\n                ;\n                ",
+  "describe": {
+    "columns": [
+      {
+        "name": "principal",
+        "ordinal": 0,
+        "type_info": "Text"
+      },
+      {
+        "name": "addressbook_id",
+        "ordinal": 1,
+        "type_info": "Text"
+      },
+      {
+        "name": "id",
+        "ordinal": 2,
+        "type_info": "Text"
+      },
+      {
+        "name": "deleted: bool",
+        "ordinal": 3,
+        "type_info": "Integer"
+      }
+    ],
+    "parameters": {
+      "Right": 0
+    },
+    "nullable": [
+      false,
+      false,
+      false,
+      false
+    ]
+  },
+  "hash": "c138b1143ac04af4930266ffae0990e82005911c11a683ad565e92335e085f4d"
+}

crates/store_sqlite/src/addressbook_store.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};
 
 #[derive(Debug, Clone)]
 struct AddressObjectRow {
@@ -32,6 +32,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,
@@ -90,9 +144,9 @@ impl SqliteAddressbookStore {
 
     async fn _update_addressbook<'e, E: Executor<'e, Database = Sqlite>>(
         executor: E,
-        principal: String,
+        principal: &str,
-        id: String,
+        id: &str,
-        addressbook: Addressbook,
+        addressbook: &Addressbook,
     ) -> Result<(), rustical_store::Error> {
         let result = sqlx::query!(
             r#"UPDATE addressbooks SET principal = ?, id = ?, displayname = ?, description = ?, push_topic = ?
@@ -116,7 +170,7 @@ impl SqliteAddressbookStore {
 
     async fn _insert_addressbook<'e, E: Executor<'e, Database = Sqlite>>(
         executor: E,
-        addressbook: Addressbook,
+        addressbook: &Addressbook,
     ) -> Result<(), rustical_store::Error> {
         sqlx::query!(
             r#"INSERT INTO addressbooks (principal, id, displayname, description, push_topic)
@@ -283,9 +337,9 @@ impl SqliteAddressbookStore {
 
     async fn _put_object<'e, E: Executor<'e, Database = Sqlite>>(
         executor: E,
-        principal: String,
+        principal: &str,
-        addressbook_id: String,
+        addressbook_id: &str,
-        object: AddressObject,
+        object: &AddressObject,
         overwrite: bool,
     ) -> Result<(), rustical_store::Error> {
         let (object_id, vcf) = (object.get_id(), object.get_vcf());
@@ -397,7 +451,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]
@@ -405,7 +459,7 @@ impl AddressbookStore for SqliteAddressbookStore {
         &self,
         addressbook: Addressbook,
     ) -> Result<(), rustical_store::Error> {
-        Self::_insert_addressbook(&self.db, addressbook).await
+        Self::_insert_addressbook(&self.db, &addressbook).await
     }
 
     #[instrument]
@@ -521,14 +575,7 @@ impl AddressbookStore for SqliteAddressbookStore {
 
         let object_id = object.get_id().to_owned();
 
-        Self::_put_object(
+        Self::_put_object(&mut *tx, &principal, &addressbook_id, &object, overwrite).await?;
-            &mut *tx,
-            principal.clone(),
-            addressbook_id.clone(),
-            object,
-            overwrite,
-        )
-        .await?;
 
         let sync_token = log_object_operation(
             &mut tx,
@@ -659,21 +706,44 @@ impl AddressbookStore for SqliteAddressbookStore {
             return Err(Error::AlreadyExists);
         }
         if existing.is_none() {
-            Self::_insert_addressbook(&mut *tx, addressbook.clone()).await?;
+            Self::_insert_addressbook(&mut *tx, &addressbook).await?;
         }
 
+        let mut sync_token = None;
         for object in objects {
             Self::_put_object(
                 &mut *tx,
-                addressbook.principal.clone(),
+                &addressbook.principal,
-                addressbook.id.clone(),
+                &addressbook.id,
-                object,
+                &object,
                 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(())
     }
 }
@@ -685,7 +755,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,
     }
@@ -700,7 +770,8 @@ async fn log_object_operation(
         addressbook_id
     )
     .fetch_one(&mut **tx)
-    .await?;
+    .await
+    .map_err(crate::Error::from)?;
 
     sqlx::query!(
         r#"
@@ -714,6 +785,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,
+        principal: &str,
-        cal_id: String,
+        cal_id: &str,
-        object: CalendarObject,
+        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(
+            Self::_put_object(&mut *tx, &calendar.principal, &calendar.id, &object, false).await?;
-                &mut *tx,
+
-                calendar.principal.clone(),
+            sync_token = Some(
-                calendar.id.clone(),
+                log_object_operation(
-                object,
+                    &mut tx,
-                false,
+                    &calendar.principal,
-            )
+                    &calendar.id,
-            .await?;
+                    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(
+        Self::_put_object(&mut *tx, &principal, &cal_id, &object, overwrite).await?;
-            &mut *tx,
-            principal.clone(),
-            cal_id.clone(),
-            object,
-            overwrite,
-        )
-        .await?;
 
         let sync_token = log_object_operation(
             &mut tx,

src/main.rs

@@ -69,7 +69,9 @@ async fn get_data_stores(
             let (send, recv) = tokio::sync::mpsc::channel(1000);
 
             let addressbook_store = Arc::new(SqliteAddressbookStore::new(db.clone(), send.clone()));
+            addressbook_store.repair_orphans().await?;
             let cal_store = Arc::new(SqliteCalendarStore::new(db.clone(), send));
+            cal_store.repair_orphans().await?;
             let subscription_store = Arc::new(SqliteStore::new(db.clone()));
             let principal_store = Arc::new(SqlitePrincipalStore::new(db));
             (