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,

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,

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));
             (