Handle incoming Add(Person) and Remove(Person) activities

src/activitypub/activity.rs

@@ -95,7 +95,14 @@ pub struct Activity {
 
     pub actor: String,
     pub object: Value,
+
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub target: Option<Value>,
+
+    #[serde(skip_serializing_if = "Option::is_none")]
     pub to: Option<Value>,
+
+    #[serde(skip_serializing_if = "Option::is_none")]
     pub cc: Option<Value>,
 }
 
@@ -115,6 +122,7 @@ pub fn create_activity(
         activity_type: activity_type.to_string(),
         actor: actor_id,
         object: serde_json::to_value(object).unwrap(),
+        target: None,
         to: Some(json!(primary_audience)),
         cc: Some(json!(secondary_audience)),
     }

src/activitypub/handlers/add.rs

@@ -0,0 +1,37 @@
+use tokio_postgres::GenericClient;
+
+use crate::activitypub::{
+    activity::Activity,
+    fetcher::helpers::ImportError,
+    identifiers::parse_local_actor_id,
+    receiver::find_object_id,
+    vocabulary::PERSON,
+};
+use crate::config::Config;
+use crate::errors::ValidationError;
+use crate::models::profiles::queries::get_profile_by_actor_id;
+use crate::models::relationships::queries::subscribe_opt;
+use crate::models::users::queries::get_user_by_name;
+use super::HandlerResult;
+
+pub async fn handle_add(
+    config: &Config,
+    db_client: &impl GenericClient,
+    activity: Activity,
+) -> HandlerResult {
+    let actor_profile = get_profile_by_actor_id(
+        db_client,
+        &activity.actor,
+    ).await?;
+    let actor = actor_profile.actor_json.ok_or(ImportError::LocalObject)?;
+    let object_id = find_object_id(&activity.object)?;
+    let username = parse_local_actor_id(&config.instance_url(), &object_id)?;
+    let user = get_user_by_name(db_client, &username).await?;
+    let target_value = activity.target.ok_or(ValidationError("target is missing"))?;
+    let target_id = find_object_id(&target_value)?;
+    if Some(target_id) == actor.subscribers {
+        subscribe_opt(db_client, &user.id, &actor_profile.id).await?;
+        return Ok(Some(PERSON));
+    };
+    Ok(None)
+}

src/activitypub/handlers/mod.rs

@@ -4,12 +4,14 @@ use super::fetcher::helpers::ImportError;
 pub type HandlerResult = Result<Option<&'static str>, ImportError>;
 
 pub mod accept_follow;
+pub mod add;
 pub mod announce;
 pub mod create_note;
 pub mod delete;
 pub mod follow;
 pub mod like;
 pub mod reject_follow;
+pub mod remove;
 pub mod undo;
 pub mod undo_follow;
 pub mod update_note;

src/activitypub/handlers/remove.rs

@@ -0,0 +1,52 @@
+use tokio_postgres::GenericClient;
+
+use crate::activitypub::{
+    activity::Activity,
+    fetcher::helpers::ImportError,
+    identifiers::parse_local_actor_id,
+    receiver::find_object_id,
+    vocabulary::PERSON,
+};
+use crate::config::Config;
+use crate::errors::{DatabaseError, ValidationError};
+use crate::models::notifications::queries::{
+    create_subscription_expiration_notification,
+};
+use crate::models::profiles::queries::get_profile_by_actor_id;
+use crate::models::relationships::queries::unsubscribe;
+use crate::models::users::queries::get_user_by_name;
+use super::HandlerResult;
+
+pub async fn handle_remove(
+    config: &Config,
+    db_client: &impl GenericClient,
+    activity: Activity,
+) -> HandlerResult {
+    let actor_profile = get_profile_by_actor_id(
+        db_client,
+        &activity.actor,
+    ).await?;
+    let actor = actor_profile.actor_json.ok_or(ImportError::LocalObject)?;
+    let object_id = find_object_id(&activity.object)?;
+    let username = parse_local_actor_id(&config.instance_url(), &object_id)?;
+    let user = get_user_by_name(db_client, &username).await?;
+    let target_value = activity.target.ok_or(ValidationError("target is missing"))?;
+    let target_id = find_object_id(&target_value)?;
+    if Some(target_id) == actor.subscribers {
+        // actor is recipient, user is sender
+        match unsubscribe(db_client, &user.id, &actor_profile.id).await {
+            Ok(_) => {
+                create_subscription_expiration_notification(
+                    db_client,
+                    &actor_profile.id,
+                    &user.id,
+                ).await?;
+                return Ok(Some(PERSON));
+            },
+            // Ignore removal if relationship does not exist
+            Err(DatabaseError::NotFound(_)) => return Ok(None),
+            Err(other_error) => return Err(other_error.into()),
+        };
+    };
+    Ok(None)
+}

src/activitypub/receiver.rs

@@ -10,11 +10,13 @@ use super::activity::{Activity, Object};
 use super::fetcher::helpers::import_post;
 use super::handlers::{
     accept_follow::handle_accept_follow,
+    add::handle_add,
     announce::handle_announce,
     delete::handle_delete,
     follow::handle_follow,
     like::handle_like,
     reject_follow::handle_reject_follow,
+    remove::handle_remove,
     undo::handle_undo,
     undo_follow::handle_undo_follow,
     update_note::handle_update_note,
@@ -193,6 +195,14 @@ pub async fn receive_activity(
             require_actor_signature(&activity.actor, &signer_id)?;
             handle_update_person(db_client, &config.media_dir(), activity).await?
         },
+        (ADD, _) => {
+            require_actor_signature(&activity.actor, &signer_id)?;
+            handle_add(config, db_client, activity).await?
+        },
+        (REMOVE, _) => {
+            require_actor_signature(&activity.actor, &signer_id)?;
+            handle_remove(config, db_client, activity).await?
+        },
         _ => {
             log::warn!("activity type is not supported: {}", activity_raw);
             return Ok(());