src/cli/update.rs

@@ -16,10 +16,9 @@ pub async fn update_users(ctx: crate::server::Context, days: i64) -> crate::Resu
 
 		while let Some(user) = stream.try_next().await? {
 			if ctx.is_local(&user.id) { continue }
-			match ctx.pull(&user.id).await.map(|x| x.actor()) {
+			match ctx.pull_user(&user.id).await {
 				Err(e) => tracing::warn!("could not update user {}: {e}", user.id),
-				Ok(Err(e)) => tracing::warn!("could not update user {}: {e}", user.id),
+				Ok(doc) => match crate::model::actor::ActiveModel::new(&doc) {
-				Ok(Ok(doc)) => match crate::model::actor::ActiveModel::new(&doc) {
 					Ok(mut u) => {
 						u.internal = Set(user.internal);
 						u.updated = Set(chrono::Utc::now());

src/errors.rs

@@ -28,9 +28,6 @@ pub enum UpubError {
 	#[error("invalid base64 string: {0:?}")]
 	Base64(#[from] base64::DecodeError),
 
-	#[error("type mismatch on object: expected {0:?}, found {1:?}")]
-	Mismatch(apb::ObjectType, apb::ObjectType),
-
 	// TODO this isn't really an error but i need to redirect from some routes so this allows me to
 	// keep the type hints on the return type, still what the hell!!!!
 	#[error("redirecting to {0}")]
@@ -111,15 +108,6 @@ impl axum::response::IntoResponse for UpubError {
 					"description": format!("missing required field from request: '{}'", x.0),
 				}))
 			).into_response(),
-			UpubError::Mismatch(expected, found) => (
-				axum::http::StatusCode::UNPROCESSABLE_ENTITY,
-				axum::Json(serde_json::json!({
-					"error": "type",
-					"expected": expected.as_ref().to_string(),
-					"found": found.as_ref().to_string(),
-					"description": self.to_string(),
-				}))
-			).into_response(),
 			_ => (
 				StatusCode::INTERNAL_SERVER_ERROR,
 				axum::Json(serde_json::json!({

src/server/context.rs

@@ -21,10 +21,9 @@ struct ContextInner {
 	actor: model::actor::Model,
 	instance: model::instance::Model,
 	pkey: String,
-	#[allow(unused)] relay: Relays,
+	relay: Relays,
 }
 
-#[allow(unused)]
 pub struct Relays {
 	sources: BTreeSet<String>,
 	sinks: BTreeSet<String>,
@@ -179,7 +178,6 @@ impl Context {
 			.await
 	}
 
-	#[allow(unused)]
 	pub fn is_relay(&self, id: &str) -> bool {
 		self.0.relay.sources.contains(id) || self.0.relay.sinks.contains(id)
 	}

src/server/fetcher.rs

@@ -9,57 +9,20 @@ use crate::{errors::UpubError, model, VERSION};
 
 use super::{addresser::Addresser, httpsign::HttpSignature, normalizer::Normalizer, Context};
 
-pub enum PullResult<T> {
-	Actor(T),
-	Activity(T),
-	Object(T),
-}
-
-impl PullResult<serde_json::Value> {
-	pub fn actor(self) -> crate::Result<serde_json::Value> {
-		match self {
-			Self::Actor(x) => Ok(x),
-			Self::Activity(x) => Err(UpubError::Mismatch(apb::ObjectType::Actor(apb::ActorType::Person), x.object_type().unwrap_or(apb::ObjectType::Activity(apb::ActivityType::Activity)))),
-			Self::Object(x) => Err(UpubError::Mismatch(apb::ObjectType::Actor(apb::ActorType::Person), x.object_type().unwrap_or(apb::ObjectType::Object))),
-		}
-	}
-
-	pub fn activity(self) -> crate::Result<serde_json::Value> {
-		match self {
-			Self::Actor(x) => Err(UpubError::Mismatch(apb::ObjectType::Activity(apb::ActivityType::Activity), x.object_type().unwrap_or(apb::ObjectType::Actor(apb::ActorType::Person)))),
-			Self::Activity(x) => Ok(x),
-			Self::Object(x) => Err(UpubError::Mismatch(apb::ObjectType::Activity(apb::ActivityType::Activity), x.object_type().unwrap_or(apb::ObjectType::Object))),
-		}
-	}
-
-	pub fn object(self) -> crate::Result<serde_json::Value> {
-		match self {
-			Self::Actor(x) => Err(UpubError::Mismatch(apb::ObjectType::Object, x.object_type().unwrap_or(apb::ObjectType::Actor(apb::ActorType::Person)))),
-			Self::Activity(x) => Err(UpubError::Mismatch(apb::ObjectType::Object, x.object_type().unwrap_or(apb::ObjectType::Activity(apb::ActivityType::Activity)))),
-			Self::Object(x) => Ok(x),
-		}
-	}
-}
-
 #[axum::async_trait]
 pub trait Fetcher {
-	async fn pull(&self, id: &str) -> crate::Result<PullResult<serde_json::Value>>;
 	async fn webfinger(&self, user: &str, host: &str) -> crate::Result<String>;
 
 	async fn fetch_domain(&self, domain: &str) -> crate::Result<model::instance::Model>;
 
 	async fn fetch_user(&self, id: &str) -> crate::Result<model::actor::Model>;
-	async fn resolve_user(&self, actor: serde_json::Value) -> crate::Result<model::actor::Model>;
+	async fn pull_user(&self, id: &str) -> crate::Result<serde_json::Value>;
+
+	async fn fetch_object(&self, id: &str) -> crate::Result<model::object::Model>;
+	async fn pull_object(&self, id: &str) -> crate::Result<serde_json::Value>;
 
 	async fn fetch_activity(&self, id: &str) -> crate::Result<model::activity::Model>;
-	async fn resolve_activity(&self, activity: serde_json::Value) -> crate::Result<model::activity::Model>;
+	async fn pull_activity(&self, id: &str) -> crate::Result<serde_json::Value>;
-
-	async fn fetch_object(&self, id: &str) -> crate::Result<model::object::Model> { self.fetch_object_r(id, 0).await }
-	#[allow(unused)] async fn resolve_object(&self, object: serde_json::Value) -> crate::Result<model::object::Model> { self.resolve_object_r(object, 0).await }
-
-	async fn fetch_object_r(&self, id: &str, depth: i32) -> crate::Result<model::object::Model>;
-	async fn resolve_object_r(&self, object: serde_json::Value, depth: i32) -> crate::Result<model::object::Model>;
-
 
 	async fn fetch_thread(&self, id: &str) -> crate::Result<()>;
 
@@ -122,28 +85,6 @@ pub trait Fetcher {
 
 #[axum::async_trait]
 impl Fetcher for Context {
-	async fn pull(&self, id: &str) -> crate::Result<PullResult<serde_json::Value>> {
-		let _domain = self.fetch_domain(&Context::server(id)).await?;
-
-		let document = Self::request(
-			Method::GET, id, None,
-			&format!("https://{}", self.domain()), self.pkey(), self.domain(),
-		)
-			.await?
-			.json::<serde_json::Value>()
-			.await?;
-
-		match document.object_type() {
-			None => Err(UpubError::bad_request()),
-			Some(apb::ObjectType::Collection(_)) => Err(UpubError::unprocessable()),
-			Some(apb::ObjectType::Tombstone) => Err(UpubError::not_found()),
-			Some(apb::ObjectType::Activity(_)) => Ok(PullResult::Activity(document)),
-			Some(apb::ObjectType::Actor(_)) => Ok(PullResult::Actor(document)),
-			_ => Ok(PullResult::Object(document)),
-		}
-	}
-
-
 	async fn webfinger(&self, user: &str, host: &str) -> crate::Result<String> {
 		let subject = format!("acct:{user}@{host}");
 		let webfinger_uri = format!("https://{host}/.well-known/webfinger?resource={subject}");
@@ -224,39 +165,15 @@ impl Fetcher for Context {
 		Ok(instance_model)
 	}
 
-	async fn resolve_user(&self, mut document: serde_json::Value) -> crate::Result<model::actor::Model> {
+	async fn fetch_user(&self, id: &str) -> crate::Result<model::actor::Model> {
-		let id = document.id().ok_or_else(|| UpubError::field("id"))?.to_string();
+		if let Some(x) = model::actor::Entity::find_by_ap_id(id).one(self.db()).await? {
-
+			return Ok(x); // already in db, easy
-		// TODO try fetching these numbers from audience/generator fields to avoid making 2 more GETs every time
-		if let Some(followers_url) = &document.followers().id() {
-			let req = Self::request(
-				Method::GET, followers_url, None,
-				&format!("https://{}", self.domain()), self.pkey(), self.domain(),
-			).await;
-			if let Ok(res) = req {
-				if let Ok(user_followers) = res.json::<serde_json::Value>().await {
-					if let Some(total) = user_followers.total_items() {
-						document = document.set_followers_count(Some(total));
-					}
-				}
-			}
 		}
 
-		if let Some(following_url) = &document.following().id() {
+		let _domain = self.fetch_domain(&Context::server(id)).await?;
-			let req =  Self::request(
-				Method::GET, following_url, None,
-				&format!("https://{}", self.domain()), self.pkey(), self.domain(),
-			).await;
-			if let Ok(res) = req {
-				if let Ok(user_following) = res.json::<serde_json::Value>().await {
-					if let Some(total) = user_following.total_items() {
-						document = document.set_following_count(Some(total));
-					}
-				}
-			}
-		}
 
-		let user_model = model::actor::ActiveModel::new(&document)?;
+		let user_document = self.pull_user(id).await?;
+		let user_model = model::actor::ActiveModel::new(&user_document)?;
 
 		// TODO this may fail: while fetching, remote server may fetch our service actor.
 		//      if it does so with http signature, we will fetch that actor in background
@@ -265,21 +182,48 @@ impl Fetcher for Context {
 		
 		// TODO fetch it back to get the internal id
 		Ok(
-			model::actor::Entity::find_by_ap_id(&id)
+			model::actor::Entity::find_by_ap_id(id)
 				.one(self.db())
 				.await?
 				.ok_or_else(UpubError::internal_server_error)?
 		)
 	}
 
-	async fn fetch_user(&self, id: &str) -> crate::Result<model::actor::Model> {
+	async fn pull_user(&self, id: &str) -> crate::Result<serde_json::Value> {
-		if let Some(x) = model::actor::Entity::find_by_ap_id(id).one(self.db()).await? {
+		let mut user = Self::request(
-			return Ok(x); // already in db, easy
+			Method::GET, id, None, &format!("https://{}", self.domain()), self.pkey(), self.domain(),
+		).await?.json::<serde_json::Value>().await?;
+
+		// TODO try fetching these numbers from audience/generator fields to avoid making 2 more GETs
+		if let Some(followers_url) = &user.followers().id() {
+			let req = Self::request(
+				Method::GET, followers_url, None,
+				&format!("https://{}", self.domain()), self.pkey(), self.domain(),
+			).await;
+			if let Ok(res) = req {
+				if let Ok(user_followers) = res.json::<serde_json::Value>().await {
+					if let Some(total) = user_followers.total_items() {
+						user = user.set_followers_count(Some(total));
+					}
+				}
+			}
 		}
 
-		let document = self.pull(id).await?.actor()?;
+		if let Some(following_url) = &user.following().id() {
+			let req =  Self::request(
+				Method::GET, following_url, None,
+				&format!("https://{}", self.domain()), self.pkey(), self.domain(),
+			).await;
+			if let Ok(res) = req {
+				if let Ok(user_following) = res.json::<serde_json::Value>().await {
+					if let Some(total) = user_following.total_items() {
+						user = user.set_following_count(Some(total));
+					}
+				}
+			}
+		}
 
-		self.resolve_user(document).await
+		Ok(user)
 	}
 
 	async fn fetch_activity(&self, id: &str) -> crate::Result<model::activity::Model> {
@@ -287,13 +231,22 @@ impl Fetcher for Context {
 			return Ok(x); // already in db, easy
 		}
 
-		let activity = self.pull(id).await?.activity()?;
+		let _domain = self.fetch_domain(&Context::server(id)).await?;
 
-		self.resolve_activity(activity).await
+		let activity_document = self.pull_activity(id).await?;
+		let activity_model = self.insert_activity(activity_document, Some(Context::server(id))).await?;
+
+		let addressed = activity_model.addressed();
+		let expanded_addresses = self.expand_addressing(addressed).await?;
+		self.address_to(Some(activity_model.internal), None, &expanded_addresses).await?;
+
+		Ok(activity_model)
 	}
 
-	async fn resolve_activity(&self, activity: serde_json::Value) -> crate::Result<model::activity::Model> {
+	async fn pull_activity(&self, id: &str) -> crate::Result<serde_json::Value> {
-		let id = activity.id().ok_or_else(|| UpubError::field("id"))?.to_string();
+		let activity = Self::request(
+			Method::GET, id, None, &format!("https://{}", self.domain()), self.pkey(), self.domain(),
+		).await?.json::<serde_json::Value>().await?;
 
 		if let Some(activity_actor) = activity.actor().id() {
 			if let Err(e) = self.fetch_user(&activity_actor).await {
@@ -307,13 +260,7 @@ impl Fetcher for Context {
 			}
 		}
 
-		let activity_model = self.insert_activity(activity, Some(Context::server(&id))).await?;
+		Ok(activity)
-
-		let addressed = activity_model.addressed();
-		let expanded_addresses = self.expand_addressing(addressed).await?;
-		self.address_to(Some(activity_model.internal), None, &expanded_addresses).await?;
-
-		Ok(activity_model)
 	}
 
 	async fn fetch_thread(&self, _id: &str) -> crate::Result<()> {
@@ -321,52 +268,64 @@ impl Fetcher for Context {
 		todo!()
 	}
 
-	async fn fetch_object_r(&self, id: &str, depth: i32) -> crate::Result<model::object::Model> {
+	async fn fetch_object(&self, id: &str) -> crate::Result<model::object::Model> {
-		if let Some(x) = model::object::Entity::find_by_ap_id(id).one(self.db()).await? {
+		fetch_object_inner(self, id, 0).await
-			return Ok(x); // already in db, easy
-		}
-
-		let object = self.pull(id).await?.object()?;
-
-		self.resolve_object_r(object, depth).await
 	}
 
-	async fn resolve_object_r(&self, object: serde_json::Value, depth: i32) -> crate::Result<model::object::Model> {
+	async fn pull_object(&self, id: &str) -> crate::Result<serde_json::Value> {
-		let id = object.id().ok_or_else(|| UpubError::field("id"))?.to_string();
+		Ok(
-
+			Context::request(
-		if let Some(oid) = object.id() {
+				Method::GET, id, None, &format!("https://{}", self.domain()), self.pkey(), self.domain(),
-			if oid != id {
+			)
-				if let Some(x) = model::object::Entity::find_by_ap_id(oid).one(self.db()).await? {
+				.await?
-					return Ok(x); // already in db, but with id different that given url
+				.json::<serde_json::Value>()
-				}
+				.await?
-			}
+		)
-		}
-
-		if let Some(attributed_to) = object.attributed_to().id() {
-			if let Err(e) = self.fetch_user(&attributed_to).await {
-				tracing::warn!("could not get actor of fetched object: {e}");
-			}
-		}
-
-		let addressed = object.addressed();
-
-		if let Some(reply) = object.in_reply_to().id() {
-			if depth <= 16 {
-				self.fetch_object_r(&reply, depth + 1).await?;
-			} else {
-				tracing::warn!("thread deeper than 16, giving up fetching more replies");
-			}
-		}
-
-		let object_model = self.insert_object(object, None).await?;
-
-		let expanded_addresses = self.expand_addressing(addressed).await?;
-		self.address_to(None, Some(object_model.internal), &expanded_addresses).await?;
-
-		Ok(object_model)
 	}
 }
 
+#[async_recursion::async_recursion]
+async fn fetch_object_inner(ctx: &Context, id: &str, depth: usize) -> crate::Result<model::object::Model> {
+	if let Some(x) = model::object::Entity::find_by_ap_id(id).one(ctx.db()).await? {
+		return Ok(x); // already in db, easy
+	}
+
+	let _domain = ctx.fetch_domain(&Context::server(id)).await?;
+
+	let object = ctx.pull_object(id).await?;
+
+	if let Some(oid) = object.id() {
+		if oid != id {
+			if let Some(x) = model::object::Entity::find_by_ap_id(oid).one(ctx.db()).await? {
+				return Ok(x); // already in db, but with id different that given url
+			}
+		}
+	}
+
+	if let Some(attributed_to) = object.attributed_to().id() {
+		if let Err(e) = ctx.fetch_user(&attributed_to).await {
+			tracing::warn!("could not get actor of fetched object: {e}");
+		}
+	}
+
+	let addressed = object.addressed();
+
+	if let Some(reply) = object.in_reply_to().id() {
+		if depth <= 16 {
+			fetch_object_inner(ctx, &reply, depth + 1).await?;
+		} else {
+			tracing::warn!("thread deeper than 16, giving up fetching more replies");
+		}
+	}
+
+	let object_model = ctx.insert_object(object, None).await?;
+
+	let expanded_addresses = ctx.expand_addressing(addressed).await?;
+	ctx.address_to(None, Some(object_model.internal), &expanded_addresses).await?;
+
+	Ok(object_model)
+}
+
 #[axum::async_trait]
 pub trait Fetchable : Sync + Send {
 	async fn fetch(&mut self, ctx: &crate::server::Context) -> crate::Result<&mut Self>;

src/server/inbox.rs

@@ -1,5 +1,6 @@
 use apb::{target::Addressed, Activity, Base, Object};
-use sea_orm::{sea_query::Expr, ActiveValue::{Set, NotSet}, ColumnTrait, EntityTrait, QueryFilter, QuerySelect, SelectColumns};
+use reqwest::StatusCode;
+use sea_orm::{sea_query::Expr, ActiveValue::{Set, NotSet}, ColumnTrait, Condition, EntityTrait, QueryFilter, QuerySelect, SelectColumns};
 
 use crate::{errors::{LoggableError, UpubError}, model, server::{addresser::Addresser, builders::AnyQuery, normalizer::Normalizer}};
 
@@ -266,7 +267,7 @@ impl apb::server::Inbox for Context {
 		let announced = self.fetch_object(&announced_id).await?;
 		// relays send us activities as Announce, but we don't really want to count those towards the
 		// total shares count of an object, so just fetch the object and be done with it
-		if matches!(actor.actor_type, apb::ActorType::Person) {
+		if self.is_relay(&activity_model.actor) {
 			tracing::info!("relay {} broadcasted {}", activity_model.actor, announced_id);
 			return Ok(())
 		}

src/server/normalizer.rs

@@ -2,6 +2,8 @@ use apb::{Node, Base, Object, Document};
 use sea_orm::{sea_query::Expr, ActiveValue::{NotSet, Set}, ColumnTrait, EntityTrait, IntoActiveModel, QueryFilter};
 use crate::{errors::UpubError, model, server::Context};
 
+use super::fetcher::Fetcher;
+
 #[axum::async_trait]
 pub trait Normalizer {
 	async fn insert_object(&self, obj: impl apb::Object, server: Option<String>) -> crate::Result<model::object::Model>;

web/src/page/object.rs

@@ -12,13 +12,8 @@ pub fn ObjectPage(tl: Timeline) -> impl IntoView {
 	let auth = use_context::<Auth>().expect("missing auth context");
 	let id = params.get().get("id").cloned().unwrap_or_default();
 	let uid =  uriproxy::uri(URL_BASE, uriproxy::UriClass::Object, &id);
-	let object = create_local_resource(
+	let object = create_local_resource(move || params.get().get("id").cloned().unwrap_or_default(), move |oid| {
-		move || params.get().get("id").cloned().unwrap_or_default(),
+		async move {
-		move |oid| async move {
-			let tl_url = format!("{}/page", Uri::api(U::Context, &oid, false));
-			if !tl.next.get().starts_with(&tl_url) {
-				tl.reset(tl_url);
-			}
 			match CACHE.get(&Uri::full(U::Object, &oid)) {
 				Some(x) => Some(x.clone()),
 				None => {
@@ -36,7 +31,7 @@ pub fn ObjectPage(tl: Timeline) -> impl IntoView {
 				}
 			}
 		}
-	);
+	});
 	view! {
 		<div>
 			<Breadcrumb back=true >
@@ -60,6 +55,10 @@ pub fn ObjectPage(tl: Timeline) -> impl IntoView {
 					},
 					Some(Some(o)) => {
 						let object = o.clone();
+						let tl_url = format!("{}/page", Uri::api(U::Context, &o.context().id().unwrap_or_default(), false));
+						if !tl.next.get().starts_with(&tl_url) {
+							tl.reset(tl_url);
+						}
 						view!{
 							<Object object=object />
 							<div class="ml-1 mr-1 mt-2">

web/src/page/user.rs

@@ -32,10 +32,6 @@ pub fn UserPage(tl: Timeline) -> impl IntoView {
 		move || params.get().get("id").cloned().unwrap_or_default(),
 		move |id| {
 			async move {
-				let tl_url = format!("{}/outbox/page", Uri::api(U::Actor, &id, false));
-				if !tl.next.get().starts_with(&tl_url) {
-					tl.reset(tl_url);
-				}
 				match CACHE.get(&Uri::full(U::Actor, &id)) {
 					Some(x) => Some(x.clone()),
 					None => {
@@ -86,6 +82,10 @@ pub fn UserPage(tl: Timeline) -> impl IntoView {
 							let following = object.following_count().unwrap_or(0);
 							let followers = object.followers_count().unwrap_or(0);
 							let statuses = object.statuses_count().unwrap_or(0);
+							let tl_url = format!("{}/outbox/page", Uri::api(U::Actor, &id.clone(), false));
+							if !tl.next.get().starts_with(&tl_url) {
+								tl.reset(tl_url);
+							}
 							let following_me = object.following_me().unwrap_or(false);
 							let followed_by_me = object.followed_by_me().unwrap_or(false);
 							let _uid = uid.clone();