chore: deduplicated side effects code

should also fix the fact that some posts miss context

src/server/inbox.rs

@@ -4,7 +4,7 @@ use sea_orm::{sea_query::Expr, ActiveValue::{Set, NotSet}, ColumnTrait, Conditio
 
 use crate::{errors::{LoggableError, UpubError}, model, server::{addresser::Addresser, builders::AnyQuery, normalizer::Normalizer}};
 
-use super::{fetcher::Fetcher, Context};
+use super::{fetcher::Fetcher, Context, side_effects::SideEffects};
 
 
 #[axum::async_trait]
@@ -18,6 +18,11 @@ impl apb::server::Inbox for Context {
 			tracing::error!("refusing to process activity without embedded object: {}", serde_json::to_string_pretty(&activity).unwrap());
 			return Err(UpubError::unprocessable());
 		};
+		if let Some(reply) = object_node.in_reply_to().id() {
+			if let Err(e) = self.fetch_object(&reply).await {
+				tracing::warn!("failed fetching replies for received object: {e}");
+			}
+		}
 		let activity_model = self.insert_activity(activity, Some(server.clone())).await?;
 		let object_model = self.insert_object(object_node, Some(server)).await?;
 		let expanded_addressing = self.expand_addressing(activity_model.addressed()).await?;
@@ -39,19 +44,7 @@ impl apb::server::Inbox for Context {
 		}
 
 		let activity_model = self.insert_activity(activity, Some(server)).await?;
-		let like = model::like::ActiveModel {
-			internal: NotSet,
-			actor: Set(internal_uid),
-			object: Set(obj.internal),
-			activity: Set(activity_model.internal),
-			published: Set(activity_model.published),
-		};
-		model::like::Entity::insert(like).exec(self.db()).await?;
-		model::object::Entity::update_many()
-			.col_expr(model::object::Column::Likes, Expr::col(model::object::Column::Likes).add(1))
-			.filter(model::object::Column::Internal.eq(obj.internal))
-			.exec(self.db())
-			.await?;
+		self.process_like(internal_uid, obj.internal, activity_model.internal, activity_model.published).await?;
 		let mut expanded_addressing = self.expand_addressing(activity_model.addressed()).await?;
 		if expanded_addressing.is_empty() { // WHY MASTODON!!!!!!!
 			expanded_addressing.push(
@@ -244,10 +237,9 @@ impl apb::server::Inbox for Context {
 
 	async fn undo(&self, server: String, activity: serde_json::Value) -> crate::Result<()> {
 		let uid = activity.actor().id().ok_or_else(UpubError::bad_request)?;
+		let internal_uid = model::actor::Entity::ap_to_internal(&uid, self.db()).await?;
 		// TODO in theory we could work with just object_id but right now only accept embedded
 		let undone_activity = activity.object().extract().ok_or_else(UpubError::bad_request)?;
-		let activity_type = undone_activity.activity_type().ok_or_else(UpubError::bad_request)?;
-		let undone_object_id = undone_activity.object().id().ok_or_else(UpubError::bad_request)?;
 		let undone_activity_author = undone_activity.actor().id().ok_or_else(UpubError::bad_request)?;
 
 		// can't undo activities from remote actors!
@@ -255,50 +247,19 @@ impl apb::server::Inbox for Context {
 			return Err(UpubError::forbidden());
 		};
 
-		self.insert_activity(activity.clone(), Some(server)).await?;
+		let activity_model = self.insert_activity(activity.clone(), Some(server)).await?;
 
-		match activity_type {
-			apb::ActivityType::Like => {
-				let internal_uid = model::actor::Entity::ap_to_internal(&uid, self.db()).await?;
-				let internal_oid = model::object::Entity::ap_to_internal(&undone_object_id, self.db()).await?;
-				model::like::Entity::delete_many()
-					.filter(
-						Condition::all()
-							.add(model::like::Column::Actor.eq(internal_uid))
-							.add(model::like::Column::Object.eq(internal_oid))
-					)
-					.exec(self.db())
-					.await?;
-				model::object::Entity::update_many()
-					.filter(model::object::Column::Internal.eq(internal_oid))
-					.col_expr(model::object::Column::Likes, Expr::col(model::object::Column::Likes).sub(1))
-					.exec(self.db())
-					.await?;
-			},
-			apb::ActivityType::Follow => {
-				let undone_aid = undone_activity.id().ok_or_else(UpubError::bad_request)?;
-				let internal_aid = model::activity::Entity::ap_to_internal(undone_aid, self.db()).await?;
-				model::relation::Entity::delete_many()
-					.filter(model::relation::Column::Activity.eq(internal_aid))
-					.exec(self.db())
-					.await?;
-				model::actor::Entity::update_many()
-					.filter(model::actor::Column::Id.eq(&undone_object_id))
-					.col_expr(model::actor::Column::FollowersCount, Expr::col(model::actor::Column::FollowersCount).sub(1))
-					.exec(self.db())
-					.await?;
-			},
-			_ => {
-				tracing::error!("received 'Undo' for unimplemented activity: {}", serde_json::to_string_pretty(&activity).unwrap());
-				return Err(StatusCode::NOT_IMPLEMENTED.into());
-			},
-		}
+		let targets = self.expand_addressing(activity.addressed()).await?;
+		self.process_undo(internal_uid, activity).await?;
+
+		self.address_to(Some(activity_model.internal), None, &targets).await?;
 
 		Ok(())
 	}
 	
 	async fn announce(&self, server: String, activity: serde_json::Value) -> crate::Result<()> {
 		let uid = activity.actor().id().ok_or_else(|| UpubError::field("actor"))?;
+		let actor = self.fetch_user(&uid).await?;
 		let internal_uid = model::actor::Entity::ap_to_internal(&uid, self.db()).await?;
 		let announced_id = activity.object().id().ok_or_else(|| UpubError::field("object"))?;
 		let activity_model = self.insert_activity(activity.clone(), Some(server)).await?;

src/server/mod.rs

@@ -10,5 +10,6 @@ pub mod auth;
 pub mod builders;
 pub mod httpsign;
 pub mod normalizer;
+pub mod side_effects;
 
 pub use context::Context;

src/server/normalizer.rs

@@ -53,14 +53,12 @@ impl Normalizer for super::Context {
 
 		// update replies counter
 		if let Set(Some(ref in_reply_to)) = object_model.in_reply_to {
-			if self.fetch_object(in_reply_to).await.is_ok() {
 			model::object::Entity::update_many()
 				.filter(model::object::Column::Id.eq(in_reply_to))
 				.col_expr(model::object::Column::Replies, Expr::col(model::object::Column::Replies).add(1))
 				.exec(self.db())
 				.await?;
 		}
-		}
 		// update statuses counter
 		if let Some(object_author) = uid {
 			model::actor::Entity::update_many()

src/server/outbox.rs

@@ -2,9 +2,9 @@ use apb::{target::Addressed, Activity, ActivityMut, Base, BaseMut, Node, Object,
 use reqwest::StatusCode;
 use sea_orm::{sea_query::Expr, ActiveValue::{Set, NotSet, Unchanged}, ColumnTrait, EntityTrait, IntoActiveModel, QueryFilter, QuerySelect, SelectColumns};
 
-use crate::{errors::UpubError, model};
+use crate::{errors::UpubError, model, routes::activitypub::jsonld::LD};
 
-use super::{addresser::Addresser, builders::AnyQuery, fetcher::Fetcher, normalizer::Normalizer, Context};
+use super::{addresser::Addresser, builders::AnyQuery, fetcher::Fetcher, normalizer::Normalizer, side_effects::SideEffects, Context};
 
 
 #[axum::async_trait]
@@ -14,13 +14,34 @@ impl apb::server::Outbox for Context {
 	type Activity = serde_json::Value;
 
 	async fn create_note(&self, uid: String, object: serde_json::Value) -> crate::Result<String> {
-		// TODO regex hell, here i come...
-		let re = regex::Regex::new(r"@(.+)@([^ ]+)").expect("failed compiling regex pattern");
+		self.create(
+			uid,
+			serde_json::Value::new_object()
+				.set_activity_type(Some(apb::ActivityType::Create))
+				.set_to(object.to())
+				.set_bto(object.bto())
+				.set_cc(object.cc())
+				.set_bcc(object.bcc())
+				.set_object(Node::object(object))
+		).await
+	}
+
+	async fn create(&self, uid: String, activity: serde_json::Value) -> crate::Result<String> {
+		let Some(object) = activity.object().extract() else {
+			return Err(UpubError::bad_request());
+		};
+
 		let raw_oid = uuid::Uuid::new_v4().to_string();
 		let oid = self.oid(&raw_oid);
 		let aid = self.aid(&uuid::Uuid::new_v4().to_string());
-		let activity_targets = object.addressed();
+		let activity_targets = activity.addressed();
 
+		if let Some(reply) = object.in_reply_to().id() {
+			self.fetch_object(&reply).await?;
+		}
+
+		// TODO regex hell here i come...
+		let re = regex::Regex::new(r"@(.+)@([^ ]+)").expect("failed compiling regex pattern");
 		let mut content = object.content().map(|x| x.to_string());
 		if let Some(c) = content {
 			let mut tmp = mdhtml::safe_markdown(&c);
@@ -40,70 +61,24 @@ impl apb::server::Outbox for Context {
 			content = Some(tmp);
 		}
 
-		let object_model = self.insert_object(
-			object
-				.set_id(Some(&oid))
-				.set_attributed_to(Node::link(uid.clone()))
-				.set_content(content.as_deref())
-				.set_published(Some(chrono::Utc::now()))
-				.set_url(Node::maybe_link(self.cfg().instance.frontend.as_ref().map(|x| format!("{x}/objects/{raw_oid}")))),
-			Some(self.domain().to_string()),
-		).await?;
-
-		let activity_model = model::activity::ActiveModel {
-			internal: NotSet,
-			id: Set(aid.clone()),
-			activity_type: Set(apb::ActivityType::Create),
-			actor: Set(uid.clone()),
-			object: Set(Some(oid.clone())),
-			target: Set(None),
-			cc: Set(object_model.cc.clone()),
-			bcc: Set(object_model.bcc.clone()),
-			to: Set(object_model.to.clone()),
-			bto: Set(object_model.bto.clone()),
-			published: Set(object_model.published),
-		};
-
-		model::activity::Entity::insert(activity_model).exec(self.db()).await?;
-
-		self.dispatch(&uid, activity_targets, &aid, Some(&oid)).await?;
-
-		Ok(aid)
-	}
-
-	async fn create(&self, uid: String, activity: serde_json::Value) -> crate::Result<String> {
-		let Some(object) = activity.object().extract() else {
-			return Err(UpubError::bad_request());
-		};
-
-		let raw_oid = uuid::Uuid::new_v4().to_string();
-		let oid = self.oid(&raw_oid);
-		let aid = self.aid(&uuid::Uuid::new_v4().to_string());
-		let activity_targets = activity.addressed();
-
 		self.insert_object(
 			object
 				.set_id(Some(&oid))
 				.set_attributed_to(Node::link(uid.clone()))
 				.set_published(Some(chrono::Utc::now()))
-				.set_to(activity.to())
-				.set_bto(activity.bto())
-				.set_cc(activity.cc())
-				.set_bcc(activity.bcc())
+				.set_content(content.as_deref())
 				.set_url(Node::maybe_link(self.cfg().instance.frontend.as_ref().map(|x| format!("{x}/objects/{raw_oid}")))),
 			Some(self.domain().to_string()),
 		).await?;
 
-		let activity_model = model::activity::ActiveModel::new(
-			&activity
+		self.insert_activity(
+			activity
 				.set_id(Some(&aid))
 				.set_actor(Node::link(uid.clone()))
 				.set_object(Node::link(oid.clone()))
-				.set_published(Some(chrono::Utc::now()))
-		)?;
-
-		model::activity::Entity::insert(activity_model.into_active_model())
-			.exec(self.db()).await?;
+				.set_published(Some(chrono::Utc::now())),
+			Some(self.domain().to_string()),
+		).await?;
 
 		self.dispatch(&uid, activity_targets, &aid, Some(&oid)).await?;
 		Ok(aid)
@@ -132,20 +107,7 @@ impl apb::server::Outbox for Context {
 			Some(self.domain().to_string()),
 		).await?;
 
-		let like_model = model::like::ActiveModel {
-			internal: NotSet,
-			actor: Set(internal_uid),
-			object: Set(obj_model.internal),
-			activity: Set(activity_model.internal),
-			published: Set(chrono::Utc::now()),
-		};
-
-		model::like::Entity::insert(like_model).exec(self.db()).await?;
-		model::object::Entity::update_many()
-			.col_expr(model::object::Column::Likes, Expr::col(model::object::Column::Likes).add(1))
-			.filter(model::object::Column::Internal.eq(obj_model.internal))
-			.exec(self.db())
-			.await?;
+		self.process_like(internal_uid, obj_model.internal, activity_model.internal, chrono::Utc::now()).await?;
 
 		self.dispatch(&uid, activity_targets, &aid, None).await?;
 
@@ -275,7 +237,7 @@ impl apb::server::Outbox for Context {
 
 	async fn undo(&self, uid: String, activity: serde_json::Value) -> crate::Result<String> {
 		let aid = self.aid(&uuid::Uuid::new_v4().to_string());
-		let activity_targets = activity.addressed();
+		let internal_uid = model::actor::Entity::ap_to_internal(&uid, self.db()).await?;
 		let old_aid = activity.object().id().ok_or_else(UpubError::bad_request)?;
 		let old_activity = model::activity::Entity::find_by_ap_id(&old_aid)
 			.one(self.db())
@@ -284,40 +246,20 @@ impl apb::server::Outbox for Context {
 		if old_activity.actor != uid {
 			return Err(UpubError::forbidden());
 		}
-		let activity_object = old_activity.object.ok_or_else(UpubError::bad_request)?;
-		let actor_internal = model::actor::Entity::ap_to_internal(&old_activity.actor, self.db()).await?;
 
-		let activity_model = model::activity::ActiveModel::new(
-			&activity
+		let activity_model = self.insert_activity(
+			activity.clone()
 				.set_id(Some(&aid))
 				.set_actor(Node::link(uid.clone()))
-				.set_published(Some(chrono::Utc::now()))
-		)?;
-		model::activity::Entity::insert(activity_model.into_active_model())
-			.exec(self.db())
-			.await?;
+				.set_published(Some(chrono::Utc::now())),
+			Some(self.domain().to_string())
+		).await?;
 
-		match old_activity.activity_type {
-			apb::ActivityType::Like => {
-				let object_internal = model::object::Entity::ap_to_internal(&activity_object, self.db()).await?;
-				model::like::Entity::delete_many()
-					.filter(model::like::Column::Actor.eq(actor_internal))
-					.filter(model::like::Column::Object.eq(object_internal))
-					.exec(self.db())
-					.await?;
-			},
-			apb::ActivityType::Follow => {
-				let target_internal = model::actor::Entity::ap_to_internal(&activity_object, self.db()).await?;
-				model::relation::Entity::delete_many()
-					.filter(model::relation::Column::Follower.eq(actor_internal))
-					.filter(model::relation::Column::Following.eq(target_internal))
-					.exec(self.db())
-					.await?;
-			},
-			t => tracing::error!("extra side effects for activity {t:?} not implemented"),
-		}
+		let targets = self.expand_addressing(activity.addressed()).await?;
+		self.process_undo(internal_uid, activity).await?;
 
-		self.dispatch(&uid, activity_targets, &aid, None).await?;
+		self.address_to(Some(activity_model.internal), None, &targets).await?;
+		self.deliver_to(&activity_model.id, &uid, &targets).await?;
 
 		Ok(aid)
 	}

src/server/side_effects.rs

@@ -0,0 +1,78 @@
+use reqwest::StatusCode;
+use sea_orm::{sea_query::Expr, ActiveValue::{NotSet, Set}, ColumnTrait, Condition, EntityTrait, QueryFilter};
+
+use crate::{errors::UpubError, model};
+
+#[axum::async_trait]
+pub trait SideEffects {
+	async fn process_like(&self, who: i64, what: i64, with: i64, when: chrono::DateTime<chrono::Utc>) -> crate::Result<()>;
+	async fn process_undo(&self, who: i64, activity: impl apb::Activity) -> crate::Result<()>;
+}
+
+#[axum::async_trait]
+impl SideEffects for super::Context {
+	async fn process_like(&self, who: i64, what: i64, with: i64, when: chrono::DateTime<chrono::Utc>) -> crate::Result<()> {
+		let like = model::like::ActiveModel {
+			internal: NotSet,
+			actor: Set(who),
+			object: Set(what),
+			activity: Set(with),
+			published: Set(when),
+		};
+		model::like::Entity::insert(like).exec(self.db()).await?;
+		model::object::Entity::update_many()
+			.col_expr(model::object::Column::Likes, Expr::col(model::object::Column::Likes).add(1))
+			.filter(model::object::Column::Internal.eq(what))
+			.exec(self.db())
+			.await?;
+
+		Ok(())
+	}
+
+	async fn process_undo(&self, who: i64, activity: impl apb::Activity) -> crate::Result<()> {
+		let undone_object_id = activity.object().id().ok_or_else(UpubError::bad_request)?;
+		match activity.activity_type() {
+			Some(apb::ActivityType::Like) => {
+				let internal_oid = model::object::Entity::ap_to_internal(&undone_object_id, self.db()).await?;
+				model::like::Entity::delete_many()
+					.filter(
+						Condition::all()
+							.add(model::like::Column::Actor.eq(who))
+							.add(model::like::Column::Object.eq(internal_oid))
+					)
+					.exec(self.db())
+					.await?;
+				model::object::Entity::update_many()
+					.filter(model::object::Column::Internal.eq(internal_oid))
+					.col_expr(model::object::Column::Likes, Expr::col(model::object::Column::Likes).sub(1))
+					.exec(self.db())
+					.await?;
+			},
+			Some(apb::ActivityType::Follow) => {
+				let undone_aid = activity.object().id().ok_or_else(UpubError::bad_request)?;
+				let internal_aid = model::activity::Entity::ap_to_internal(&undone_aid, self.db()).await?;
+				model::relation::Entity::delete_many()
+					.filter(model::relation::Column::Activity.eq(internal_aid))
+					.exec(self.db())
+					.await?;
+				model::actor::Entity::update_many()
+					.filter(model::actor::Column::Internal.eq(who))
+					.col_expr(model::actor::Column::FollowingCount, Expr::col(model::actor::Column::FollowingCount).sub(1))
+					.exec(self.db())
+					.await?;
+				model::actor::Entity::update_many()
+					.filter(model::actor::Column::Id.eq(&undone_object_id))
+					.col_expr(model::actor::Column::FollowersCount, Expr::col(model::actor::Column::FollowersCount).sub(1))
+					.exec(self.db())
+					.await?;
+			},
+			t => {
+				tracing::error!("received 'Undo' for unimplemented activity type: {t:?}");
+				return Err(StatusCode::NOT_IMPLEMENTED.into());
+			},
+		}
+		
+
+		Ok(())
+	}
+}