Compare commits
No commits in common. "b0626081340ad09f5d0bbbd6f7be7fb3395f208b" and "a60b5774a158ecd0a36d7b59627d1c13f24441eb" have entirely different histories.
b062608134
...
a60b5774a1
3 changed files with 84 additions and 65 deletions
|
@ -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?;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
@ -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(),
|
||||
}
|
||||
}
|
||||
|
|
|
@ -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(),
|
||||
}
|
||||
}
|
||||
|
|
Loading…
Reference in a new issue