1
0
Fork 0
forked from alemi/upub
upub/src/server/dispatcher.rs

135 lines
4 KiB
Rust
Raw Normal View History

use reqwest::Method;
use sea_orm::{ColumnTrait, DatabaseConnection, EntityTrait, Order, QueryFilter, QueryOrder};
use tokio::{sync::broadcast, task::JoinHandle};
use apb::{ActivityMut, Node};
use crate::{model, routes::activitypub::jsonld::LD, server::{fetcher::Fetcher, Context}};
pub struct Dispatcher {
waker: broadcast::Sender<()>,
}
impl Default for Dispatcher {
fn default() -> Self {
let (waker, _) = broadcast::channel(1);
Dispatcher { waker }
}
}
impl Dispatcher {
pub fn spawn(&self, db: DatabaseConnection, domain: String, poll_interval: u64) -> JoinHandle<()> {
let mut 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;
2024-03-26 03:11:59 +01:00
}
})
}
pub fn wakeup(&self) {
match self.waker.send(()) {
Err(_) => tracing::error!("no worker to wakeup"),
Ok(n) => tracing::debug!("woken {n} workers"),
}
}
2024-03-26 03:11:59 +01:00
}
async fn worker(db: &DatabaseConnection, domain: &str, poll_interval: u64, waker: &mut broadcast::Receiver<()>) -> crate::Result<()> {
2024-03-26 03:11:59 +01:00
loop {
let Some(delivery) = model::delivery::Entity::find()
.filter(model::delivery::Column::NotBefore.lte(chrono::Utc::now()))
2024-03-26 03:11:59 +01:00
.order_by(model::delivery::Column::NotBefore, Order::Asc)
.one(db)
2024-03-26 03:11:59 +01:00
.await?
2024-03-26 23:53:44 +01:00
else {
tokio::select! {
biased;
_ = waker.recv() => {},
_ = tokio::time::sleep(std::time::Duration::from_secs(poll_interval)) => {},
}
2024-03-26 23:53:44 +01:00
continue
};
2024-03-26 03:11:59 +01:00
let del_row = model::delivery::ActiveModel {
internal: sea_orm::ActiveValue::Set(delivery.internal),
2024-03-26 03:11:59 +01:00
..Default::default()
};
let del = model::delivery::Entity::delete(del_row)
.exec(db)
2024-03-26 03:11:59 +01:00
.await?;
2024-03-26 03:11:59 +01:00
if del.rows_affected == 0 {
// another worker claimed this delivery
continue; // go back to the top
}
if delivery.expired() {
// try polling for another one
continue; // go back to top
}
2024-03-26 03:11:59 +01:00
tracing::info!("delivering {} to {}", delivery.activity, delivery.target);
let payload = match model::activity::Entity::find_by_ap_id(&delivery.activity)
2024-03-26 03:11:59 +01:00
.find_also_related(model::object::Entity)
.one(db)
2024-03-26 03:11:59 +01:00
.await? // TODO probably should not fail here and at least re-insert the delivery
{
2024-05-02 15:21:45 +02:00
Some((activity, None)) => activity.ap().ld_context(),
Some((activity, Some(object))) => {
2024-04-24 05:04:30 +02:00
let always_embed = matches!(
activity.activity_type,
apb::ActivityType::Create
| apb::ActivityType::Undo
| apb::ActivityType::Update
| apb::ActivityType::Accept(_)
| apb::ActivityType::Reject(_)
);
if always_embed {
2024-05-02 15:12:36 +02:00
activity.ap().set_object(Node::object(object.ap())).ld_context()
} else {
2024-05-02 15:12:36 +02:00
activity.ap().ld_context()
}
},
2024-03-26 03:11:59 +01:00
None => {
tracing::warn!("skipping dispatch for deleted object {}", delivery.activity);
continue;
},
};
let Some(actor) = model::actor::Entity::find_by_ap_id(&delivery.actor)
.one(db)
.await?
else {
tracing::error!("abandoning delivery of {} from non existant actor: {}", delivery.activity, delivery.actor);
continue;
2024-03-26 03:11:59 +01:00
};
let Some(key) = actor.private_key
else {
tracing::error!("abandoning delivery of {} from actor without private key: {}", delivery.activity, delivery.actor);
continue;
};
if let Err(e) = Context::request(
Method::POST, &delivery.target,
Some(&serde_json::to_string(&payload).unwrap()),
&delivery.actor, &key, domain
).await {
2024-03-26 03:11:59 +01:00
tracing::warn!("failed delivery of {} to {} : {e}", delivery.activity, delivery.target);
let new_delivery = model::delivery::ActiveModel {
internal: sea_orm::ActiveValue::NotSet,
2024-03-26 03:11:59 +01:00
not_before: sea_orm::ActiveValue::Set(delivery.next_delivery()),
actor: sea_orm::ActiveValue::Set(delivery.actor),
target: sea_orm::ActiveValue::Set(delivery.target),
activity: sea_orm::ActiveValue::Set(delivery.activity),
created: sea_orm::ActiveValue::Set(delivery.created),
attempt: sea_orm::ActiveValue::Set(delivery.attempt + 1),
};
model::delivery::Entity::insert(new_delivery).exec(db).await?;
2024-03-26 03:11:59 +01:00
}
}
}