src/server/dispatcher.rs

@@ -3,7 +3,7 @@ use sea_orm::{ColumnTrait, DatabaseConnection, EntityTrait, Order, QueryFilter,
 use tokio::{sync::broadcast, task::JoinHandle};
 
 use apb::{ActivityMut, Node};
-use crate::{model, routes::activitypub::jsonld::LD, server::{fetcher::Fetcher, Context}};
+use crate::{errors::UpubError, model, routes::activitypub::jsonld::LD, server::{fetcher::Fetcher, Context}};
 
 pub struct Dispatcher {
 	waker: broadcast::Sender<()>,
@@ -18,13 +18,10 @@ impl Default for Dispatcher {
 
 impl Dispatcher {
 	pub fn spawn(&self, db: DatabaseConnection, domain: String, poll_interval: u64) -> JoinHandle<()> {
-		let mut waker = self.waker.subscribe();
+		let waker = self.waker.subscribe();
 		tokio::spawn(async move {
-			loop {
-				if let Err(e) = worker(&db, &domain, poll_interval, &mut waker).await {
-					tracing::error!("delivery worker exited with error: {e}");
-				}
-				tokio::time::sleep(std::time::Duration::from_secs(poll_interval * 10)).await;
+			if let Err(e) = worker(db, domain, poll_interval, waker).await {
+				tracing::error!("delivery worker exited with error: {e}");
 			}
 		})
 	}
@@ -37,12 +34,12 @@ impl Dispatcher {
 	}
 }
 
-async fn worker(db: &DatabaseConnection, domain: &str, poll_interval: u64, waker: &mut broadcast::Receiver<()>) -> crate::Result<()> {
+async fn worker(db: DatabaseConnection, domain: String, poll_interval: u64, mut waker: broadcast::Receiver<()>) -> Result<(), UpubError> {
 	loop {
 		let Some(delivery) = model::delivery::Entity::find()
 			.filter(model::delivery::Column::NotBefore.lte(chrono::Utc::now()))
 			.order_by(model::delivery::Column::NotBefore, Order::Asc)
-			.one(db)
+			.one(&db)
 			.await?
 		else {
 			tokio::select! {
@@ -58,7 +55,7 @@ async fn worker(db: &DatabaseConnection, domain: &str, poll_interval: u64, waker
 			..Default::default()
 		};
 		let del = model::delivery::Entity::delete(del_row)
-			.exec(db)
+			.exec(&db)
 			.await?;
 
 		if del.rows_affected == 0 {
@@ -74,7 +71,7 @@ async fn worker(db: &DatabaseConnection, domain: &str, poll_interval: u64, waker
 
 		let payload = match model::activity::Entity::find_by_id(&delivery.activity)
 			.find_also_related(model::object::Entity)
-			.one(db)
+			.one(&db)
 			.await? // TODO probably should not fail here and at least re-insert the delivery
 		{
 			Some((activity, None)) => activity.ap().ld_context(),
@@ -101,7 +98,7 @@ async fn worker(db: &DatabaseConnection, domain: &str, poll_interval: u64, waker
 
 		let key = if delivery.actor == format!("https://{domain}") {
 			let Some(model::application::Model { private_key: key, .. }) = model::application::Entity::find()
-				.one(db).await?
+				.one(&db).await?
 			else {
 				tracing::error!("no private key configured for application");
 				continue;
@@ -109,7 +106,7 @@ async fn worker(db: &DatabaseConnection, domain: &str, poll_interval: u64, waker
 			key
 		} else {
 			let Some(model::user::Model{ private_key: Some(key), .. }) = model::user::Entity::find_by_id(&delivery.actor)
-				.one(db).await?
+				.one(&db).await?
 			else { 
 				tracing::error!("can not dispatch activity for user without private key: {}", delivery.actor);
 				continue;
@@ -121,7 +118,7 @@ async fn worker(db: &DatabaseConnection, domain: &str, poll_interval: u64, waker
 		if let Err(e) = Context::request(
 			Method::POST, &delivery.target,
 			Some(&serde_json::to_string(&payload).unwrap()),
-			&delivery.actor, &key, domain
+			&delivery.actor, &key, &domain
 		).await {
 			tracing::warn!("failed delivery of {} to {} : {e}", delivery.activity, delivery.target);
 			let new_delivery = model::delivery::ActiveModel {
@@ -133,7 +130,7 @@ async fn worker(db: &DatabaseConnection, domain: &str, poll_interval: u64, waker
 				created: sea_orm::ActiveValue::Set(delivery.created),
 				attempt: sea_orm::ActiveValue::Set(delivery.attempt + 1),
 			};
-			model::delivery::Entity::insert(new_delivery).exec(db).await?;
+			model::delivery::Entity::insert(new_delivery).exec(&db).await?;
 		}
 	}
 }

web/src/components/activity.rs

@@ -1,7 +1,7 @@
 use leptos::*;
 use crate::prelude::*;
 
-use apb::{target::Addressed, Base, Activity, Object};
+use apb::{target::Addressed, Activity, Object};
 
 
 #[component]
@@ -31,39 +31,3 @@ pub fn ActivityLine(activity: crate::Object) -> impl IntoView {
 		</div>
 	}
 }
-
-#[component]
-pub fn Item(item: crate::Object) -> impl IntoView {
-	let id = item.id().unwrap_or_default().to_string();
-	match item.object_type() {
-		// special case for placeholder activities
-		Some(apb::ObjectType::Note) | Some(apb::ObjectType::Document(_)) =>
-			view! { <Object object=item /> }.into_view(),
-		// everything else
-		Some(apb::ObjectType::Activity(t)) => {
-			let object_id = item.object().id().unwrap_or_default();
-			let object = match t {
-				apb::ActivityType::Create | apb::ActivityType::Announce => 
-					CACHE.get(&object_id).map(|obj| {
-						view! { <Object object=obj /> }
-					}.into_view()),
-				apb::ActivityType::Follow =>
-					CACHE.get(&object_id).map(|obj| {
-						view! {
-							<div class="ml-1">
-								<ActorBanner object=obj />
-								<FollowRequestButtons activity_id=id actor_id=object_id />
-							</div>
-						}
-					}.into_view()),
-				_ => None,
-			};
-			view! {
-				<ActivityLine activity=item />
-				{object}
-			}.into_view()
-		},
-		// should never happen
-		_ => view! { <p><code>type not implemented</code></p> }.into_view(),
-	}
-}

web/src/components/timeline.rs

@@ -92,15 +92,44 @@ pub fn TimelineRepliesRecursive(tl: Timeline, root: String) -> impl IntoView {
 		<For
 			each=root_values
 			key=|k| k.id().unwrap_or_default().to_string()
-			children=move |obj: crate::Object| {
-				let oid = obj.id().unwrap_or_default().to_string();
-				view! {
-					<div class="context depth-r">
-						<Item item=obj />
-						<div class="depth-r">
-							<TimelineRepliesRecursive tl=tl root=oid />
+			children=move |object: crate::Object| {
+				match object.object_type() {
+					Some(apb::ObjectType::Activity(apb::ActivityType::Create)) => {
+						let oid = object.object().id().unwrap_or_default().to_string();
+						if let Some(note) = CACHE.get(&oid) {
+							view! {
+								<div class="context depth-r">
+									<ActivityLine activity=object />
+									<Object object=note />
+									<div class="depth-r">
+										<TimelineRepliesRecursive tl=tl root=oid />
+									</div>
+								</div>
+							}
+						} else {
+							view! {
+								<div class="context depth-r">
+									<ActivityLine activity=object />
+								</div>
+							}
+						}
+					},
+					Some(apb::ObjectType::Activity(_)) => view! {
+						<div class="context depth-r">
+							<ActivityLine activity=object />
 						</div>
-					</div>
+					},
+					_ => {
+						let oid = object.id().unwrap_or_default().to_string();
+						view! {
+							<div class="context depth-r">
+								<Object object=object />
+								<div class="depth-r">
+									<TimelineRepliesRecursive tl=tl root=oid />
+								</div>
+							</div>
+						}
+					},
 				}
 			}
 		/ >
@@ -139,13 +168,42 @@ pub fn TimelineFeed(tl: Timeline) -> impl IntoView {
 			key=|k| k.to_string()
 			children=move |id: String| {
 				match CACHE.get(&id) {
-					Some(i) => view! {
-						<Item item=i />
-						<hr />
-					}.into_view(),
+					Some(item) => match item.object_type() {
+						// special case for placeholder activities
+						Some(apb::ObjectType::Note) => view! {
+								<Object object=item.clone() />
+								<hr/ >
+							}.into_view(),
+						// everything else
+						Some(apb::ObjectType::Activity(t)) => {
+							let object_id = item.object().id().unwrap_or_default();
+							let object = match t {
+								apb::ActivityType::Create | apb::ActivityType::Announce => 
+									CACHE.get(&object_id).map(|obj| {
+										view! { <Object object=obj /> }
+									}.into_view()),
+								apb::ActivityType::Follow =>
+									CACHE.get(&object_id).map(|obj| {
+										view! {
+											<div class="ml-1">
+												<ActorBanner object=obj />
+												<FollowRequestButtons activity_id=id actor_id=object_id />
+											</div>
+										}
+									}.into_view()),
+								_ => None,
+							};
+							view! {
+								<ActivityLine activity=item />
+								{object}
+								<hr/ >
+							}.into_view()
+						},
+						// should never happen
+						_ => view! { <p><code>type not implemented</code></p><hr /> }.into_view(),
+					},
 					None => view! {
 						<p><code>{id}</code>" "[<a href={uri}>go</a>]</p>
-						<hr />
 					}.into_view(),
 				}
 			}