8ae997da4a23609f42421f9fbc031e904a9570d8 / frogue / src / activitypub / delivery.rs
Raw
//! Delivery worker: polls pending deliveries and sends them to follower inboxes.
//! Also provides the shared `enqueue_delivery` helper for inserting activities
//! and creating delivery rows.
use std::sync::Arc;
use std::time::Duration;
use chrono::{DateTime, FixedOffset, Utc};
use sea_orm::{
ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter, QueryOrder,
QuerySelect, Set,
};
use tokio::sync::RwLock;
use tracing::{error, info, warn};
use crate::AppState;
use crate::entities::ap_deliveries;
use crate::entities::prelude::*;
/// Spawns the delivery worker background task.
///
/// Polls every 2 seconds for pending deliveries and sends them with HTTP signatures.
pub fn spawn(state: Arc<RwLock<AppState>>) -> tokio::task::JoinHandle<()> {
tokio::spawn(async move {
loop {
tokio::time::sleep(Duration::from_secs(2)).await;
let state_guard = state.read().await;
let db = state_guard.db.clone();
let public_url = state_guard.public_url().to_string();
let http_client = state_guard.http_client().clone();
drop(state_guard);
if let Err(e) = process_pending_deliveries(&db, &public_url, &http_client).await {
error!(error = %e, "delivery worker error");
}
}
})
}
/// Insert an activity and create a pending delivery row for it.
///
/// Shared helper used by inbox (Accept/Follow), following module (Follow, Undo),
/// and notes (Create). Returns the activity ID.
#[tracing::instrument(skip(db, activity_json), fields(object_url = %object_url, inbox_url = %inbox_url))]
pub async fn enqueue_delivery(
db: &DatabaseConnection,
local_actor_id: i64,
activity_uuid: uuid::Uuid,
activity_json: String,
object_url: String,
inbox_url: String,
) -> Result<i64, Box<dyn std::error::Error + Send + Sync>> {
let activity_model = crate::entities::ap_activities::ActiveModel {
uuid: Set(activity_uuid),
actor_id: Set(local_actor_id),
activity_json: Set(activity_json),
object_url: Set(object_url),
..Default::default()
};
let activity = activity_model.insert(db).await?;
let delivery_model = ap_deliveries::ActiveModel {
activity_id: Set(activity.id),
signer_actor_id: Set(local_actor_id),
inbox_url: Set(inbox_url),
status: Set("pending".to_string()),
attempts: Set(0),
next_attempt_at: Set(Utc::now().into()),
last_error: Set(None),
..Default::default()
};
delivery_model.insert(db).await?;
info!(activity_id = activity.id, "enqueued delivery");
Ok(activity.id)
}
async fn process_pending_deliveries(
db: &DatabaseConnection,
public_url: &str,
http_client: &reqwest_middleware::ClientWithMiddleware,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
// Find pending deliveries where next_attempt_at <= now
let now = Utc::now();
let deliveries = ApDeliveries::find()
.filter(ap_deliveries::Column::Status.eq("pending"))
.filter(ap_deliveries::Column::NextAttemptAt.lte(now))
.order_by_asc(ap_deliveries::Column::NextAttemptAt)
.limit(10)
.all(db)
.await?;
for delivery in deliveries {
if let Err(e) = process_delivery(db, &delivery, public_url, http_client).await {
error!(delivery_id = delivery.id, error = %e, "failed to process delivery");
}
}
Ok(())
}
async fn process_delivery(
db: &DatabaseConnection,
delivery: &ap_deliveries::Model,
public_url: &str,
http_client: &reqwest_middleware::ClientWithMiddleware,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let activity = ApActivities::find_by_id(delivery.activity_id)
.one(db)
.await?
.ok_or("activity not found")?;
let actor = ApActors::find_by_id(delivery.signer_actor_id)
.one(db)
.await?
.ok_or("actor not found")?;
// Get the actor URL (works for both user and repository actors)
let actor_url = crate::activitypub::actor::actor_url_for(db, public_url, &actor)
.await?
.ok_or("actor URL not found")?;
let key_id = format!("{}#main-key", actor_url);
// Parse the activity JSON
let activity_json: serde_json::Value = serde_json::from_str(&activity.activity_json)?;
let body = serde_json::to_string(&activity_json)?;
// Build the request
let inbox_url = &delivery.inbox_url;
let url = url::Url::parse(inbox_url)?;
let host = url.host_str().ok_or("no host in inbox_url")?;
let path = url.path();
let date = Utc::now().format("%a, %d %b %Y %H:%M:%S GMT").to_string();
// Sign the request using the signature helper
let (signature, digest) = crate::activitypub::signatures::sign_request(
&actor.private_key_pem,
&key_id,
"post",
path,
host,
&date,
body.as_bytes(),
)
.map_err(|e| format!("signing failed: {}", e))?;
// Send the request
let result = http_client
.post(inbox_url)
.header("Content-Type", "application/activity+json")
.header("Date", &date)
.header("Digest", &digest)
.header("Signature", &signature)
.body(body)
.send()
.await;
// Update delivery status
let mut updated_delivery: ap_deliveries::ActiveModel = delivery.clone().into();
match result {
Ok(response) if response.status().is_success() => {
info!(delivery_id = delivery.id, inbox_url = %inbox_url, "delivery succeeded");
updated_delivery.status = Set("done".to_string());
}
Ok(response) => {
let status = response.status();
let error_msg = format!("HTTP {}", status);
warn!(delivery_id = delivery.id, inbox_url = %inbox_url, status = %status, "delivery failed");
let attempts = delivery.attempts + 1;
updated_delivery.attempts = Set(attempts);
updated_delivery.last_error = Set(Some(error_msg.clone()));
if attempts >= 8 {
updated_delivery.status = Set("failed".to_string());
} else {
// Exponential backoff: 5 * 2^attempts seconds, max 1 hour
let backoff_secs = (5 * 2i64.pow(attempts as u32)).min(3600);
let next_attempt: DateTime<FixedOffset> =
(Utc::now() + chrono::Duration::seconds(backoff_secs)).into();
updated_delivery.next_attempt_at = Set(next_attempt);
}
}
Err(e) => {
let error_msg = format!("request failed: {}", e);
warn!(delivery_id = delivery.id, inbox_url = %inbox_url, error = %e, "delivery failed");
let attempts = delivery.attempts + 1;
updated_delivery.attempts = Set(attempts);
updated_delivery.last_error = Set(Some(error_msg.clone()));
if attempts >= 8 {
updated_delivery.status = Set("failed".to_string());
} else {
let backoff_secs = (5 * 2i64.pow(attempts as u32)).min(3600);
let next_attempt: DateTime<FixedOffset> =
(Utc::now() + chrono::Duration::seconds(backoff_secs)).into();
updated_delivery.next_attempt_at = Set(next_attempt);
}
}
}
updated_delivery.update(db).await?;
Ok(())
}