eric / rustwarden

//! RFC 0020 — Fediverse interoperability tests.
//!
//! Spawns a `wiremock` HTTP server playing the role of a remote
//! ActivityPub inbox (Mastodon-style) and exercises the signing /
//! verification chain end-to-end:
//!
//! 1. `signed_request_round_trip_through_wiremock` — build + sign +
//!    POST a request, capture it on the receiver side, re-verify the
//!    signature against the same public PEM via `parse_and_verify_cavage`.
//! 2. `outbound_delivers_when_inbox_returns_202` — `drain_one` with a
//!    delivery impl that returns `Ok` on 2xx flips the row to
//!    `Delivered`.
//! 3. `outbound_defers_when_inbox_returns_503` — the same harness with
//!    a 503 leaves the row in `Pending` with `next_retry_at` set per
//!    RFC 0010.

use std::sync::Arc;

use async_trait::async_trait;
use base64::Engine;
use chrono::Utc;
use reqwest::Client;
use sea_orm::{DatabaseBackend, DatabaseConnection, MockDatabase, MockExecResult};
use uuid::Uuid;
use wiremock::matchers::{header_exists, method, path};
use wiremock::{Mock, MockServer, Request, ResponseTemplate};

use rustwarden_core::error::AppError;
use rustwarden_fediverse::http_signatures::{
    build_signature_input, parse_and_verify_cavage, rsa_sha256_sign,
};
use rustwarden_fediverse::models::fediverse_outbound_queue;
use rustwarden_fediverse::services::actor_key_service::ActorKeyService;
use rustwarden_fediverse::services::fediverse_outbound_service::{
    FediverseOutboundService, OutboundDelivery, OutboundStatus,
};

/// Build the canonical `Signature` header for a draft-cavage signed
/// POST. Mirrors `HttpSignedDelivery::deliver` but lives in the test
/// module so the round-trip can be exercised without DB plumbing.
fn sign_post(private_pem: &str, key_id: &str, host: &str, path: &str, date: &str) -> String {
    let signed_headers = vec![
        "(request-target)".to_string(),
        "host".to_string(),
        "date".to_string(),
    ];
    let request_target = format!("post {path}");
    let input = build_signature_input(&signed_headers, &request_target, |h| match h {
        "host" => Some(host.to_string()),
        "date" => Some(date.to_string()),
        _ => None,
    })
    .expect("build_signature_input");

    let signature_bytes = rsa_sha256_sign(private_pem, input.as_bytes()).expect("sign");
    let signature_b64 = base64::engine::general_purpose::STANDARD.encode(&signature_bytes);

    format!(
        r#"keyId="{key_id}",algorithm="rsa-sha256",headers="(request-target) host date",signature="{signature_b64}""#,
    )
}

#[tokio::test]
async fn signed_request_round_trip_through_wiremock() {
    let server = MockServer::start().await;
    let inbox_path = "/inbox";

    Mock::given(method("POST"))
        .and(path(inbox_path))
        .and(header_exists("signature"))
        .respond_with(ResponseTemplate::new(202))
        .expect(1)
        .mount(&server)
        .await;

    let material = ActorKeyService::generate().expect("RSA gen");
    let url = format!("{}{inbox_path}", server.uri());
    let parsed = reqwest::Url::parse(&url).unwrap();
    let host = parsed.host_str().unwrap().to_string();
    let host_with_port = match parsed.port() {
        Some(p) => format!("{host}:{p}"),
        None => host.clone(),
    };
    let date = Utc::now().format("%a, %d %b %Y %H:%M:%S GMT").to_string();
    let key_id = "https://a.tld/users/alice#main-key";

    let signature_header = sign_post(
        &material.private_pem,
        key_id,
        &host_with_port,
        inbox_path,
        &date,
    );

    let body = br#"{"@context":"https://www.w3.org/ns/activitystreams","type":"Update"}"#;

    let response = Client::new()
        .post(&url)
        .header("Date", &date)
        .header("Host", &host_with_port)
        .header("Signature", &signature_header)
        .header("Content-Type", "application/activity+json")
        .body(body.to_vec())
        .send()
        .await
        .expect("post ok");
    assert_eq!(response.status().as_u16(), 202);

    // Capture the request that wiremock received and re-verify the
    // signature against the published public PEM. This proves the
    // signed input string is what reqwest actually puts on the wire.
    let received: Vec<Request> = server.received_requests().await.unwrap();
    assert_eq!(received.len(), 1);
    let req = &received[0];

    let header_lookup = |name: &str| -> Option<String> {
        // wiremock lowercases header names already.
        req.headers
            .get(name)
            .map(|v| v.to_str().unwrap().to_string())
    };

    let sig = header_lookup("signature").expect("signature header captured");
    parse_and_verify_cavage(
        &sig,
        &format!("post {inbox_path}"),
        &req.body,
        &material.public_pem,
        header_lookup,
    )
    .expect("signature must round-trip through wiremock");
}

/// Minimal `OutboundDelivery` that signs + posts via reqwest using a
/// pre-supplied keypair. Mirrors `HttpSignedDelivery::deliver` without
/// the DB lookup so we can exercise `drain_one` against wiremock with
/// only a `MockDatabase` for the queue.
struct InMemorySignedDelivery {
    private_pem: String,
    key_id: String,
}

#[async_trait]
impl OutboundDelivery for InMemorySignedDelivery {
    async fn deliver(&self, item: &fediverse_outbound_queue::Model) -> Result<(), AppError> {
        let url = reqwest::Url::parse(&item.target_inbox)
            .map_err(|e| AppError::Internal(format!("bad URL: {e}")))?;
        let host = url
            .host_str()
            .ok_or_else(|| AppError::Internal("no host".into()))?
            .to_string();
        let host_with_port = match url.port() {
            Some(p) => format!("{host}:{p}"),
            None => host.clone(),
        };
        let path = url.path().to_string();
        let date = Utc::now().format("%a, %d %b %Y %H:%M:%S GMT").to_string();

        let signature_header = sign_post(
            &self.private_pem,
            &self.key_id,
            &host_with_port,
            &path,
            &date,
        );

        let response = Client::new()
            .post(&item.target_inbox)
            .header("Date", &date)
            .header("Host", &host_with_port)
            .header("Signature", &signature_header)
            .header("Content-Type", "application/activity+json")
            .body(item.payload_json.clone())
            .send()
            .await
            .map_err(|e| AppError::Internal(format!("HTTP delivery failed: {e}")))?;

        if response.status().is_success() {
            Ok(())
        } else {
            Err(AppError::Internal(format!(
                "remote inbox responded {}",
                response.status(),
            )))
        }
    }
}

fn pending_row(target_inbox: &str, target_domain: &str) -> fediverse_outbound_queue::Model {
    fediverse_outbound_queue::Model {
        id: Uuid::new_v4(),
        activity_id: "https://a.tld/activities/1".to_string(),
        payload_json: r#"{"type":"Update"}"#.to_string(),
        target_inbox: target_inbox.to_string(),
        target_domain: target_domain.to_string(),
        status: "pending".to_string(),
        attempts: 0,
        max_attempts: 8,
        last_error: None,
        next_retry_at: None,
        created_at: Utc::now(),
        delivered_at: None,
        actor_user_id: Some(Uuid::new_v4()),
    }
}

#[tokio::test]
async fn outbound_delivers_when_inbox_returns_202() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/inbox"))
        .respond_with(ResponseTemplate::new(202))
        .expect(1)
        .mount(&server)
        .await;

    let target_inbox = format!("{}/inbox", server.uri());
    let row = pending_row(&target_inbox, "wiremock.local");

    // Two fixtures: the SELECT-candidate then the SELECT-after-update
    // (sea-orm mock returns subsequent fixtures for `.update()` reads).
    let mut delivered = row.clone();
    delivered.status = OutboundStatus::Delivered.as_str().to_string();
    delivered.attempts = 1;
    delivered.delivered_at = Some(Utc::now());

    let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres)
        .append_query_results([vec![row.clone()]])
        .append_query_results([vec![delivered.clone()]])
        .append_exec_results([MockExecResult {
            last_insert_id: 0,
            rows_affected: 1,
        }])
        .into_connection();

    let material = ActorKeyService::generate().expect("RSA gen");
    let delivery = InMemorySignedDelivery {
        private_pem: material.private_pem,
        key_id: "https://a.tld/users/alice#main-key".to_string(),
    };

    let result = FediverseOutboundService::drain_one(&db, &delivery)
        .await
        .expect("drain ok")
        .expect("row drained");

    assert_eq!(result.status, OutboundStatus::Delivered.as_str());
    assert_eq!(result.attempts, 1);
    assert!(result.delivered_at.is_some());
}

#[tokio::test]
async fn outbound_defers_when_inbox_returns_503() {
    let server = MockServer::start().await;
    Mock::given(method("POST"))
        .and(path("/inbox"))
        .respond_with(ResponseTemplate::new(503))
        .expect(1)
        .mount(&server)
        .await;

    let target_inbox = format!("{}/inbox", server.uri());
    let row = pending_row(&target_inbox, "wiremock.local");

    let mut deferred = row.clone();
    deferred.status = OutboundStatus::Pending.as_str().to_string();
    deferred.attempts = 1;
    deferred.last_error = Some("remote inbox responded 503".to_string());
    deferred.next_retry_at = Some(Utc::now() + chrono::Duration::minutes(5));

    let db: DatabaseConnection = MockDatabase::new(DatabaseBackend::Postgres)
        .append_query_results([vec![row.clone()]])
        .append_query_results([vec![deferred.clone()]])
        .append_exec_results([MockExecResult {
            last_insert_id: 0,
            rows_affected: 1,
        }])
        .into_connection();

    let material = ActorKeyService::generate().expect("RSA gen");
    let delivery = InMemorySignedDelivery {
        private_pem: material.private_pem,
        key_id: "https://a.tld/users/alice#main-key".to_string(),
    };

    let result = FediverseOutboundService::drain_one(&db, &delivery)
        .await
        .expect("drain ok")
        .expect("row drained");

    assert_eq!(result.status, OutboundStatus::Pending.as_str());
    assert_eq!(result.attempts, 1);
    assert!(result.last_error.is_some());
    assert!(result.next_retry_at.is_some());
}

// `Arc` is unused in this file but documents the intent of sharing the
// MockServer handle across spawned tasks should we extend the harness.
#[allow(dead_code)]
fn _share_server(server: MockServer) -> Arc<MockServer> {
    Arc::new(server)
}