ap delivery worker

cruciverb-server/src/activitypub/deliver.rs

@@ -0,0 +1,125 @@
+use std::sync::Arc;
+use std::time::Duration;
+
+use crate::state::AppState;
+use super::signatures;
+
+pub fn start_delivery_worker(state: Arc<AppState>) {
+    tokio::spawn(async move {
+        loop {
+            if state.ap.is_some() {
+                // #16: skip processing when federation is frozen
+                if !super::is_frozen() {
+                    process_queue(&state).await;
+                }
+            }
+            tokio::time::sleep(Duration::from_secs(30)).await;
+        }
+    });
+}
+
+async fn process_queue(state: &Arc<AppState>) {
+    let ap = match state.ap.as_ref() {
+        Some(ap) => ap,
+        None => return,
+    };
+
+    let now = chrono::Utc::now().to_rfc3339();
+
+    let deliveries: Vec<(i64, String, String, i64)> = {
+        let dbi = state.db.lock().await;
+        let mut stmt = dbi.db.prepare(
+            "SELECT id, inbox_uri, payload, attempts FROM ap_delivery_queue
+             WHERE next_retry <= ?1 ORDER BY next_retry ASC LIMIT 20"
+        ).unwrap();
+        stmt.query_map([&now], |row| {
+            Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?))
+        }).unwrap().filter_map(|r| r.ok()).collect()
+    };
+
+    if deliveries.is_empty() { return; }
+
+    let client = reqwest::Client::builder()
+        .timeout(Duration::from_secs(10))
+        .build()
+        .unwrap();
+
+    for (id, inbox_uri, payload, attempts) in deliveries {
+        // #8: DNS rebinding protection on delivery targets
+        if crate::security::is_internal_url(&inbox_uri) {
+            tracing::warn!("refusing delivery to internal URL: {}", inbox_uri);
+            let dbi = state.db.lock().await;
+            dbi.db.execute("DELETE FROM ap_delivery_queue WHERE id = ?1", [id]).ok();
+            continue;
+        }
+
+        let host = inbox_uri.split("//").nth(1)
+            .and_then(|s| s.split('/').next())
+            .unwrap_or("");
+        let path = inbox_uri.split("//").nth(1)
+            .and_then(|s| s.find('/').map(|i| &s[i..]))
+            .unwrap_or("/inbox");
+
+        let key_id = format!("{}#main-key", ap.actor_uri);
+        let sig_headers = signatures::sign_request(
+            &ap.keys.private_key,
+            &key_id,
+            "POST",
+            path,
+            host,
+            Some(payload.as_bytes()),
+        );
+
+        let mut req = client.post(&inbox_uri)
+            .header("Content-Type", "application/activity+json");
+
+        for (name, value) in &sig_headers {
+            req = req.header(name.as_str(), value.as_str());
+        }
+
+        match req.body(payload.clone()).send().await {
+            Ok(resp) if resp.status().is_success() || resp.status().as_u16() == 202 => {
+                let dbi = state.db.lock().await;
+                dbi.db.execute("DELETE FROM ap_delivery_queue WHERE id = ?1", [id]).ok();
+                crate::db::track_federation_activity(&dbi.db, host, "outbound");
+                tracing::info!("delivered to {}", inbox_uri);
+            }
+            Ok(resp) => {
+                tracing::warn!("delivery to {} failed: {}", inbox_uri, resp.status());
+                handle_failure(state, id, attempts).await;
+            }
+            Err(e) => {
+                tracing::warn!("delivery to {} error: {}", inbox_uri, e);
+                handle_failure(state, id, attempts).await;
+            }
+        }
+    }
+}
+
+async fn handle_failure(state: &Arc<AppState>, id: i64, attempts: i64) {
+    let dbi = state.db.lock().await;
+    let new_attempts = attempts + 1;
+
+    if new_attempts >= 10 {
+        // give up after 10 attempts
+        dbi.db.execute("DELETE FROM ap_delivery_queue WHERE id = ?1", [id]).ok();
+        tracing::info!("dropping delivery {} after {} attempts", id, new_attempts);
+        return;
+    }
+
+    // exponential backoff: 1m, 5m, 30m, 2h, 12h, 24h cap
+    let delay_secs = match new_attempts {
+        1 => 60,
+        2 => 300,
+        3 => 1800,
+        4 => 7200,
+        5 => 43200,
+        _ => 86400,
+    };
+
+    let next_retry = (chrono::Utc::now() + chrono::Duration::seconds(delay_secs)).to_rfc3339();
+    dbi.db.execute(
+        "UPDATE ap_delivery_queue SET attempts = ?1, next_retry = ?2 WHERE id = ?3",
+        rusqlite::params![new_attempts, next_retry, id],
+    ).ok();
+}