federate reject activity and add edit_conflicts api endpoint

src/api.rs

@@ -3,7 +3,7 @@ use crate::error::MyResult;
 use crate::federation::activities::create_article::CreateArticle;
 use crate::federation::activities::update_article::UpdateArticle;
 use crate::federation::objects::article::DbArticle;
-use crate::federation::objects::edit::{DbEdit, EditVersion};
+use crate::federation::objects::edit::{ApubEdit, DbEdit, EditVersion};
 use crate::federation::objects::instance::DbInstance;
 use activitypub_federation::config::Data;
 use activitypub_federation::fetch::object_id::ObjectId;
@@ -23,6 +23,7 @@ pub fn api_routes() -> Router {
             "/article",
             get(get_article).post(create_article).patch(edit_article),
         )
+        .route("/edit_conflicts", get(edit_conflicts))
         .route("/resolve_instance", get(resolve_instance))
         .route("/resolve_article", get(resolve_article))
         .route("/instance", get(get_local_instance))
@@ -96,8 +97,7 @@ async fn edit_article(
     } else {
         UpdateArticle::send_to_origin(
             edit,
-            // TODO: should be dereference(), but then article is refetched which breaks test_edit_conflict()
-            original_article.instance.dereference_local(&data).await?,
+            original_article.instance.dereference(&data).await?,
             &data,
         )
         .await?;
@@ -168,3 +168,10 @@ async fn follow_instance(
     data.local_instance().follow(&instance, &data).await?;
     Ok(())
 }
+
+#[debug_handler]
+async fn edit_conflicts(data: Data<DatabaseHandle>) -> MyResult<Json<Vec<ApubEdit>>> {
+    let lock = data.conflicts.lock().unwrap();
+    let conflicts = lock.clone();
+    Ok(Json(conflicts))
+}

src/database.rs

@@ -1,4 +1,5 @@
 use crate::federation::objects::article::DbArticle;
+use crate::federation::objects::edit::ApubEdit;
 use crate::federation::objects::instance::DbInstance;
 use std::collections::HashMap;
 use std::sync::{Arc, Mutex};
@@ -9,6 +10,7 @@ pub type DatabaseHandle = Arc<Database>;
 pub struct Database {
     pub instances: Mutex<HashMap<Url, DbInstance>>,
     pub articles: Mutex<HashMap<Url, DbArticle>>,
+    pub conflicts: Mutex<Vec<ApubEdit>>,
 }
 
 impl Database {

src/federation/activities/mod.rs

@@ -1,4 +1,5 @@
 pub mod accept;
 pub mod create_article;
 pub mod follow;
+pub mod reject;
 pub mod update_article;

src/federation/activities/reject.rs

@@ -0,0 +1,72 @@
+use crate::database::DatabaseHandle;
+use crate::error::MyResult;
+use crate::federation::objects::edit::ApubEdit;
+use crate::federation::objects::instance::DbInstance;
+use crate::utils::generate_activity_id;
+use activitypub_federation::kinds::activity::RejectType;
+use activitypub_federation::{
+    config::Data, fetch::object_id::ObjectId, protocol::helpers::deserialize_one_or_many,
+    traits::ActivityHandler,
+};
+
+use serde::{Deserialize, Serialize};
+use url::Url;
+
+#[derive(Deserialize, Serialize, Debug)]
+#[serde(rename_all = "camelCase")]
+pub struct RejectEdit {
+    pub actor: ObjectId<DbInstance>,
+    #[serde(deserialize_with = "deserialize_one_or_many")]
+    pub to: Vec<Url>,
+    pub object: ApubEdit,
+    #[serde(rename = "type")]
+    pub kind: RejectType,
+    pub id: Url,
+}
+
+impl RejectEdit {
+    pub async fn send(
+        edit: ApubEdit,
+        user_instance: DbInstance,
+        data: &Data<DatabaseHandle>,
+    ) -> MyResult<()> {
+        let local_instance = data.local_instance();
+        let id = generate_activity_id(local_instance.ap_id.inner())?;
+        let reject = RejectEdit {
+            actor: local_instance.ap_id.clone(),
+            to: vec![user_instance.ap_id.into_inner()],
+            object: edit,
+            kind: Default::default(),
+            id,
+        };
+        local_instance
+            .send(reject, vec![user_instance.inbox], data)
+            .await?;
+        Ok(())
+    }
+}
+
+#[async_trait::async_trait]
+impl ActivityHandler for RejectEdit {
+    type DataType = DatabaseHandle;
+    type Error = crate::error::Error;
+
+    fn id(&self) -> &Url {
+        &self.id
+    }
+
+    fn actor(&self) -> &Url {
+        self.actor.inner()
+    }
+
+    async fn verify(&self, _data: &Data<Self::DataType>) -> Result<(), Self::Error> {
+        Ok(())
+    }
+
+    async fn receive(self, data: &Data<Self::DataType>) -> Result<(), Self::Error> {
+        // TODO: cant convert this to DbEdit as it tries to apply patch and fails
+        let mut lock = data.conflicts.lock().unwrap();
+        lock.push(self.object);
+        Ok(())
+    }
+}

src/federation/activities/update_article.rs

@@ -12,6 +12,7 @@ use activitypub_federation::{
     traits::{ActivityHandler, Object},
 };
 
+use crate::federation::activities::reject::RejectEdit;
 use serde::{Deserialize, Serialize};
 use url::Url;
 
@@ -85,28 +86,32 @@ impl ActivityHandler for UpdateArticle {
     }
 
     async fn receive(self, data: &Data<Self::DataType>) -> Result<(), Self::Error> {
-        let article_local = {
-            DbEdit::from_json(self.object.clone(), data).await?;
-            let lock = data.articles.lock().unwrap();
-            let article = lock.get(self.object.object.inner()).unwrap();
-            article.local
-        };
-
-        if article_local {
-            // No need to wrap in announce, we can construct a new activity as all important info
-            // is in the object and result fields.
-            let local_instance = data.local_instance();
-            let id = generate_activity_id(local_instance.ap_id.inner())?;
-            let update = UpdateArticle {
-                actor: local_instance.ap_id.clone(),
-                to: local_instance.follower_ids(),
-                object: self.object,
-                kind: Default::default(),
-                id,
+        if DbEdit::from_json(self.object.clone(), data).await.is_ok() {
+            let article_local = {
+                let lock = data.articles.lock().unwrap();
+                let article = lock.get(self.object.object.inner()).unwrap();
+                article.local
             };
-            data.local_instance()
-                .send_to_followers(update, data)
-                .await?;
+
+            if article_local {
+                // No need to wrap in announce, we can construct a new activity as all important info
+                // is in the object and result fields.
+                let local_instance = data.local_instance();
+                let id = generate_activity_id(local_instance.ap_id.inner())?;
+                let update = UpdateArticle {
+                    actor: local_instance.ap_id.clone(),
+                    to: local_instance.follower_ids(),
+                    object: self.object,
+                    kind: Default::default(),
+                    id,
+                };
+                data.local_instance()
+                    .send_to_followers(update, data)
+                    .await?;
+            }
+        } else {
+            let user_instance = self.actor.dereference(data).await?;
+            RejectEdit::send(self.object.clone(), user_instance, data).await?;
         }
 
         Ok(())

src/federation/mod.rs

@@ -35,6 +35,7 @@ pub async fn federation_config(hostname: &str) -> Result<FederationConfig<Databa
             local_instance,
         )])),
         articles: Mutex::new(HashMap::new()),
+        conflicts: Mutex::new(vec![]),
     });
     let config = FederationConfig::builder()
         .domain(hostname)

src/federation/objects/edit.rs

@@ -1,5 +1,6 @@
 use crate::database::DatabaseHandle;
 use crate::error::{Error, MyResult};
+
 use crate::federation::objects::article::DbArticle;
 use activitypub_federation::config::Data;
 use activitypub_federation::fetch::object_id::ObjectId;
@@ -103,17 +104,21 @@ impl Object for DbEdit {
             version: json.version,
             local: false,
         };
-        let mut lock = data.articles.lock().unwrap();
-        let article = lock.get_mut(edit.article_id.inner()).unwrap();
-        article.edits.push(edit.clone());
+        let article_read = {
+            let lock = data.articles.lock().unwrap();
+            lock.get(edit.article_id.inner()).unwrap().clone()
+        };
         let patch = Patch::from_str(&edit.diff)?;
         // Dont apply the edit if we already fetched an update Article version.
         // TODO: this assumes that we always receive edits in the correct order, probably need to
         //       include the parent for each edit
-        if article.latest_version != edit.version {
-            article.text = apply(&article.text, &patch)?;
-        }
-
+        //if article_read.latest_version != edit.version {
+        let applied = apply(&article_read.text, &patch)?;
+        let mut lock = data.articles.lock().unwrap();
+        let article = lock.get_mut(edit.article_id.inner()).unwrap();
+        article.edits.push(edit.clone());
+        article.text = applied;
         Ok(edit)
+        //}
     }
 }

src/federation/routes.rs

@@ -13,6 +13,7 @@ use activitypub_federation::traits::{ActivityHandler, Collection};
 use axum::extract::Path;
 
 use crate::federation::activities::create_article::CreateArticle;
+use crate::federation::activities::reject::RejectEdit;
 use crate::federation::activities::update_article::UpdateArticle;
 use crate::federation::objects::article::ApubArticle;
 use crate::federation::objects::articles_collection::{ArticleCollection, DbArticleCollection};
@@ -85,6 +86,7 @@ pub enum InboxActivities {
     Accept(Accept),
     CreateArticle(CreateArticle),
     UpdateArticle(UpdateArticle),
+    RejectEdit(RejectEdit),
 }
 
 #[debug_handler]

tests/common.rs

@@ -66,9 +66,19 @@ impl TestData {
     }
 }
 
+pub const TEST_ARTICLE_DEFAULT_TEXT: &str = "empty\n";
+
 pub async fn create_article(hostname: &str, title: String) -> MyResult<DbArticle> {
-    let create_form = CreateArticleData { title };
-    post(hostname, "article", &create_form).await
+    let create_form = CreateArticleData {
+        title: title.clone(),
+    };
+    let article: DbArticle = post(hostname, "article", &create_form).await?;
+    // create initial edit to ensure that conflicts are generated (there are no conflicts on empty file)
+    let edit_form = EditArticleData {
+        ap_id: article.ap_id,
+        new_text: TEST_ARTICLE_DEFAULT_TEXT.to_string(),
+    };
+    edit_article(hostname, &title, &edit_form).await
 }
 
 pub async fn get_article(hostname: &str, title: &str) -> MyResult<DbArticle> {
@@ -91,8 +101,7 @@ pub async fn edit_article(
     let get_article = GetArticleData {
         title: title.to_string(),
     };
-    let updated_article: DbArticle =
-        get_query(hostname, &"article".to_string(), Some(get_article)).await?;
+    let updated_article: DbArticle = get_query(hostname, "article", Some(get_article)).await?;
     Ok(updated_article)
 }
 

tests/test.rs

@@ -4,13 +4,16 @@ mod common;
 
 use crate::common::{
     create_article, edit_article, follow_instance, get_article, get_query, TestData,
+    TEST_ARTICLE_DEFAULT_TEXT,
 };
 use common::get;
 use fediwiki::api::{EditArticleData, ResolveObject};
 use fediwiki::error::MyResult;
 use fediwiki::federation::objects::article::DbArticle;
+use fediwiki::federation::objects::edit::ApubEdit;
 use fediwiki::federation::objects::instance::DbInstance;
 use serial_test::serial;
+
 use url::Url;
 
 #[tokio::test]
@@ -31,7 +34,7 @@ async fn test_create_read_and_edit_article() -> MyResult<()> {
     // now article can be read
     let get_res = get_article(data.hostname_alpha, &create_res.title).await?;
     assert_eq!(title, get_res.title);
-    assert!(get_res.text.is_empty());
+    assert_eq!(TEST_ARTICLE_DEFAULT_TEXT, get_res.text);
     assert!(get_res.local);
 
     // edit article
@@ -41,7 +44,7 @@ async fn test_create_read_and_edit_article() -> MyResult<()> {
     };
     let edit_res = edit_article(data.hostname_alpha, &create_res.title, &edit_form).await?;
     assert_eq!(edit_form.new_text, edit_res.text);
-    assert_eq!(1, edit_res.edits.len());
+    assert_eq!(2, edit_res.edits.len());
 
     data.stop()
 }
@@ -57,7 +60,7 @@ async fn test_follow_instance() -> MyResult<()> {
     let beta_instance: DbInstance = get(data.hostname_beta, "instance").await?;
     assert_eq!(0, beta_instance.followers.len());
 
-    follow_instance(data.hostname_alpha, &data.hostname_beta).await?;
+    follow_instance(data.hostname_alpha, data.hostname_beta).await?;
 
     // check that follow was federated
     let beta_instance: DbInstance = get(data.hostname_beta, "instance").await?;
@@ -78,13 +81,13 @@ async fn test_synchronize_articles() -> MyResult<()> {
     let title = "Manu_Chao".to_string();
     let create_res = create_article(data.hostname_alpha, title.clone()).await?;
     assert_eq!(title, create_res.title);
-    assert_eq!(0, create_res.edits.len());
+    assert_eq!(1, create_res.edits.len());
     assert!(create_res.local);
 
     // edit the article
     let edit_form = EditArticleData {
         ap_id: create_res.ap_id.clone(),
-        new_text: "Lorem Ipsum 2".to_string(),
+        new_text: "Lorem Ipsum 2\n".to_string(),
     };
     edit_article(data.hostname_alpha, &title, &edit_form).await?;
 
@@ -103,7 +106,7 @@ async fn test_synchronize_articles() -> MyResult<()> {
     let get_res = get_article(data.hostname_beta, &create_res.title).await?;
     assert_eq!(create_res.ap_id, get_res.ap_id);
     assert_eq!(title, get_res.title);
-    assert_eq!(1, get_res.edits.len());
+    assert_eq!(2, get_res.edits.len());
     assert_eq!(edit_form.new_text, get_res.text);
     assert!(!get_res.local);
 
@@ -126,7 +129,7 @@ async fn test_edit_local_article() -> MyResult<()> {
     // article should be federated to alpha
     let get_res = get_article(data.hostname_alpha, &create_res.title).await?;
     assert_eq!(create_res.title, get_res.title);
-    assert_eq!(0, get_res.edits.len());
+    assert_eq!(1, get_res.edits.len());
     assert!(!get_res.local);
     assert_eq!(create_res.text, get_res.text);
 
@@ -137,7 +140,7 @@ async fn test_edit_local_article() -> MyResult<()> {
     };
     let edit_res = edit_article(data.hostname_beta, &create_res.title, &edit_form).await?;
     assert_eq!(edit_res.text, edit_form.new_text);
-    assert_eq!(edit_res.edits.len(), 1);
+    assert_eq!(edit_res.edits.len(), 2);
     assert!(edit_res.edits[0]
         .id
         .to_string()
@@ -146,7 +149,7 @@ async fn test_edit_local_article() -> MyResult<()> {
     // edit should be federated to alpha
     let get_res = get_article(data.hostname_alpha, &edit_res.title).await?;
     assert_eq!(edit_res.title, get_res.title);
-    assert_eq!(edit_res.edits.len(), 1);
+    assert_eq!(edit_res.edits.len(), 2);
     assert_eq!(edit_res.text, get_res.text);
 
     data.stop()
@@ -169,7 +172,7 @@ async fn test_edit_remote_article() -> MyResult<()> {
     // article should be federated to alpha and gamma
     let get_res = get_article(data.hostname_alpha, &title).await?;
     assert_eq!(create_res.title, get_res.title);
-    assert_eq!(0, get_res.edits.len());
+    assert_eq!(1, get_res.edits.len());
     assert!(!get_res.local);
 
     let get_res = get_article(data.hostname_gamma, &title).await?;
@@ -181,8 +184,8 @@ async fn test_edit_remote_article() -> MyResult<()> {
         new_text: "Lorem Ipsum 2".to_string(),
     };
     let edit_res = edit_article(data.hostname_alpha, &title, &edit_form).await?;
-    assert_eq!(edit_res.text, edit_form.new_text);
-    assert_eq!(edit_res.edits.len(), 1);
+    assert_eq!(edit_form.new_text, edit_res.text);
+    assert_eq!(2, edit_res.edits.len());
     assert!(!edit_res.local);
     assert!(edit_res.edits[0]
         .id
@@ -192,12 +195,12 @@ async fn test_edit_remote_article() -> MyResult<()> {
     // edit should be federated to beta and gamma
     let get_res = get_article(data.hostname_alpha, &title).await?;
     assert_eq!(edit_res.title, get_res.title);
-    assert_eq!(edit_res.edits.len(), 1);
+    assert_eq!(edit_res.edits.len(), 2);
     assert_eq!(edit_res.text, get_res.text);
 
     let get_res = get_article(data.hostname_gamma, &title).await?;
     assert_eq!(edit_res.title, get_res.title);
-    assert_eq!(edit_res.edits.len(), 1);
+    assert_eq!(edit_res.edits.len(), 2);
     assert_eq!(edit_res.text, get_res.text);
 
     data.stop()
@@ -227,13 +230,13 @@ async fn test_edit_conflict() -> MyResult<()> {
     // alpha edits article
     let edit_form = EditArticleData {
         ap_id: create_res.ap_id.clone(),
-        new_text: "Lorem Ipsum".to_string(),
+        new_text: "Lorem Ipsum\n".to_string(),
     };
     let edit_res = edit_article(data.hostname_alpha, &create_res.title, &edit_form).await?;
     assert_eq!(edit_res.text, edit_form.new_text);
-    assert_eq!(edit_res.edits.len(), 1);
+    assert_eq!(2, edit_res.edits.len());
     assert!(!edit_res.local);
-    assert!(edit_res.edits[0]
+    assert!(edit_res.edits[1]
         .id
         .to_string()
         .starts_with(&edit_res.ap_id.to_string()));
@@ -242,14 +245,18 @@ async fn test_edit_conflict() -> MyResult<()> {
     // not be updated with this conflicting version, instead user needs to handle the conflict
     let edit_form = EditArticleData {
         ap_id: create_res.ap_id,
-        new_text: "aaaa".to_string(),
+        new_text: "aaaa\n".to_string(),
     };
     let edit_res = edit_article(data.hostname_gamma, &create_res.title, &edit_form).await?;
-    assert_eq!(create_res.text, edit_res.text);
-    assert_eq!(0, edit_res.edits.len());
+    assert_ne!(edit_form.new_text, edit_res.text);
+    assert_eq!(2, edit_res.edits.len());
     assert!(!edit_res.local);
 
-    // TODO: need to federate the conflict as `Reject` and then resolve it
+    let conflicts: Vec<ApubEdit> =
+        get_query(data.hostname_gamma, "edit_conflicts", None::<()>).await?;
+    assert_eq!(1, conflicts.len());
+
+    // TODO: need a way to mark conflict as resolved, maybe opt param on edit endpoint
 
     data.stop()
 }