database: add "add_or_update_webmention" operation

This is an operation that atomically adds or updates a webmention cite
attached to a post. This is used so a database backend can optimize
for it (for example, using a transaction or shifting the JSON
modification operation to the database)

1 files changed, 39 insertions, 0 deletions

diff --git a/kittybox-rs/src/database/postgres/mod.rs b/kittybox-rs/src/database/postgres/mod.rs
index b9a21c3..b1a03b1 100644
--- a/kittybox-rs/src/database/postgres/mod.rs
+++ b/kittybox-rs/src/database/postgres/mod.rs
@@ -130,6 +130,45 @@ impl Storage for PostgresStorage {
             .map(|_| ())
     }
 
+    async fn add_or_update_webmention(&self, target: &str, mention_type: MentionType, mention: serde_json::Value) -> Result<()> {
+        let mut txn = self.db.begin().await?;
+
+        let (uid, mut post) = sqlx::query_as::<_, (String, serde_json::Value)>("SELECT uid, mf2 FROM kittybox.mf2_json WHERE uid = $1 OR mf2['properties']['url'] ? $1 FOR UPDATE")
+            .bind(target)
+            .fetch_optional(&mut *txn)
+            .await?
+            .ok_or(StorageError::from_static(
+                ErrorKind::NotFound,
+                "The specified post wasn't found in the database."
+            ))?;
+
+        let key: &'static str = match mention_type {
+            MentionType::Reply => "reply",
+            MentionType::Like => "like",
+            MentionType::Repost => "repost",
+            MentionType::Bookmark => "bookmark",
+            MentionType::Mention => "mention",
+        };
+        let mention_uid = mention["properties"]["uid"][0].clone();
+        if let Some(values) = post["properties"][key].as_array_mut() {
+            for value in values.iter_mut() {
+                if value["properties"]["uid"][0] == mention_uid {
+                    *value = mention;
+                    break;
+                }
+            }
+        } else {
+            post["properties"][key] = serde_json::Value::Array(vec![mention]);
+        }
+
+        sqlx::query("UPDATE kittybox.mf2_json SET mf2 = $2 WHERE uid = $1")
+            .bind(uid)
+            .bind(post)
+            .execute(&mut *txn)
+            .await?;
+
+        txn.commit().await.map_err(Into::into)
+    }
     #[tracing::instrument(skip(self))]
     async fn update_post(&self, url: &'_ str, update: MicropubUpdate) -> Result<()> {
         tracing::debug!("Updating post {}", url);