eric / rustwarden

//! `POST /inbox` handler — receives ActivityPub lifecycle activities
//! and dispatches them to [`crate::services::fediverse_signal_service::FediverseSignalService`]
//! (RFC 0012).
//!
//! The route reads the `Signature` HTTP header, delegates verification
//! to the configured `SignatureVerifier`, then routes by activity
//! `type`. Only `Update`, `Delete`, and `Revoke` are supported in this
//! slice; the rest get `501 Not Implemented`.
//!

use std::sync::Arc;

use axum::{Extension, body::Bytes, http::HeaderMap, http::StatusCode};
use sea_orm::DatabaseConnection;
use serde::Deserialize;
use uuid::Uuid;

use super::error::FediverseError;
use crate::observer::{FediverseEventObserver, NoopFediverseObserver};
use crate::services::fediverse_signal_service::{
    DeleteActorHooks, FediverseSignalService, SignatureVerifier, UpdatedActor, VerifyCtx,
};

/// Per-router state injected into the inbox handler.
///
/// Cheaply clonable (the verifier is `Arc`-ed). The application
/// instantiates this once via the federation router builder.
#[derive(Clone)]
pub struct InboxState {
    /// Pluggable signature verifier. Defaults to `AlwaysAcceptVerifier`
    /// when the consumer does not call
    /// `RustwardenBuilder::with_signature_verifier`.
    pub verifier: Arc<dyn SignatureVerifier>,
}

/// Minimal projection of the JSON-LD payload.
///
/// We only deserialise the fields the dispatcher needs; the full body
/// is forwarded as `serde_json::Value` to the service for audit /
/// re-processing.
#[derive(Debug, Deserialize)]
struct InboxActivity {
    #[serde(rename = "type")]
    kind: String,
    actor: Option<String>,
    object: Option<serde_json::Value>,
}

/// `POST /inbox` axum handler.
pub async fn handle(
    Extension(state): Extension<InboxState>,
    Extension(db): Extension<DatabaseConnection>,
    observer: Option<Extension<Arc<dyn FediverseEventObserver>>>,
    headers: HeaderMap,
    body: Bytes,
) -> Result<StatusCode, FediverseError> {
    let observer: Arc<dyn FediverseEventObserver> = match observer {
        Some(Extension(obs)) => obs,
        None => Arc::new(NoopFediverseObserver),
    };

    // 1. Signature verification (delegated).
    let signature_header = headers
        .get("signature")
        .and_then(|v| v.to_str().ok())
        .ok_or_else(|| FediverseError::Unauthorized("missing Signature header".to_string()))?;

    let activity: InboxActivity = match serde_json::from_slice(&body) {
        Ok(a) => a,
        Err(e) => {
            // RFC 0028 — emit a structural-abuse event so detectors can
            // count repeated malformed bodies from the same source. The
            // domain extraction is best-effort: a malformed body usually
            // means we can't even read the actor URL, so we fall back to
            // the empty string which the observer-side normaliser
            // suppresses.
            observer.on_inbox_abuse("", None, "malformed_json").await;
            return Err(FediverseError::BadRequest(format!("invalid JSON: {e}")));
        }
    };

    let actor_ap_id = activity.actor.as_deref().unwrap_or("");
    // Touch `signature_header` to keep the early validation in place
    // (its absence already errors above) — the value is also reachable
    // via `headers["signature"]` for impls.
    let _ = signature_header;

    // F8 / RFC 0027 — trust policy gate. Reject before any verifier work
    // or downstream effect when the source domain isn't allowed.
    let source_domain = url::Url::parse(actor_ap_id)
        .ok()
        .and_then(|u| u.host_str().map(|s| s.to_ascii_lowercase()))
        .ok_or_else(|| FediverseError::BadRequest("actor URL has no host".into()))?;
    crate::services::fediverse_trust_service::FediverseTrustService::new(db.clone())
        .assert_domain_allowed(&source_domain)
        .await
        .map_err(|e| FediverseError::Forbidden(e.to_string()))?;

    state
        .verifier
        .verify(&VerifyCtx {
            actor_ap_id,
            method: "post",
            path: "/inbox",
            headers: &headers,
            body: &body,
            public_key_pem: "",
        })
        .await
        .map_err(|e| FediverseError::Unauthorized(e.to_string()))?;

    // 2. Dispatch by activity kind.
    match activity.kind.as_str() {
        "Update" => dispatch_update(&db, &activity).await?,
        "Delete" => dispatch_delete(&db, &activity, observer.as_ref()).await?,
        "Revoke" => dispatch_revoke(&db, &activity).await?,
        other => {
            observer
                .on_inbox_abuse(&source_domain, Some(actor_ap_id), "unknown_activity_type")
                .await;
            return Err(FediverseError::Unsupported(other.to_string()));
        }
    }

    Ok(StatusCode::ACCEPTED)
}

async fn dispatch_update(
    db: &DatabaseConnection,
    activity: &InboxActivity,
) -> Result<(), FediverseError> {
    let object = activity
        .object
        .as_ref()
        .ok_or_else(|| FediverseError::BadRequest("Update missing `object`".into()))?;

    let ap_id = object
        .get("id")
        .and_then(|v| v.as_str())
        .ok_or_else(|| FediverseError::BadRequest("Update.object missing `id`".into()))?
        .to_string();

    let display_name = object
        .get("name")
        .and_then(|v| v.as_str())
        .map(str::to_string);
    let summary_html = object
        .get("summary")
        .and_then(|v| v.as_str())
        .map(str::to_string);
    let icon_url = object
        .get("icon")
        .and_then(|v| v.get("url"))
        .and_then(|v| v.as_str())
        .map(str::to_string);
    let public_key_pem = object
        .get("publicKey")
        .and_then(|v| v.get("publicKeyPem"))
        .and_then(|v| v.as_str())
        .map(str::to_string);

    FediverseSignalService::apply_update_actor(
        db,
        UpdatedActor {
            ap_id,
            display_name,
            summary_html,
            icon_url,
            public_key_pem,
            raw_json: Some(object.clone()),
        },
    )
    .await?;
    Ok(())
}

async fn dispatch_delete(
    db: &DatabaseConnection,
    activity: &InboxActivity,
    observer: &dyn FediverseEventObserver,
) -> Result<(), FediverseError> {
    // AP `Delete{Actor}` carries the actor URL either as a string in
    // `object` or as `{ "id": "...", "type": "Person" }`.
    let ap_id = activity
        .object
        .as_ref()
        .and_then(|o| {
            o.as_str()
                .map(str::to_string)
                .or_else(|| o.get("id").and_then(|v| v.as_str()).map(str::to_string))
        })
        .ok_or_else(|| FediverseError::BadRequest("Delete missing actor reference".into()))?;

    FediverseSignalService::apply_delete_actor_with_hooks(
        db,
        &ap_id,
        DeleteActorHooks {
            observer: Some(observer),
            ..Default::default()
        },
    )
    .await?;
    Ok(())
}

async fn dispatch_revoke(
    db: &DatabaseConnection,
    activity: &InboxActivity,
) -> Result<(), FediverseError> {
    let object = activity
        .object
        .as_ref()
        .ok_or_else(|| FediverseError::BadRequest("Revoke missing `object`".into()))?;

    let team_id = object
        .get("team")
        .and_then(|v| v.as_str())
        .and_then(|s| s.rsplit('/').next())
        .and_then(|s| Uuid::parse_str(s).ok())
        .ok_or_else(|| FediverseError::BadRequest("Revoke.object.team missing/malformed".into()))?;

    let remote_identity_id = object
        .get("subject")
        .and_then(|v| v.as_str())
        .and_then(|s| Uuid::parse_str(s).ok())
        .ok_or_else(|| {
            FediverseError::BadRequest("Revoke.object.subject missing/malformed".into())
        })?;

    FediverseSignalService::apply_revoke_membership(db, team_id, remote_identity_id).await?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use async_trait::async_trait;
    use axum::{Router, body::Body, http::Request, routing::post};
    use rustwarden_core::error::AppError;
    use sea_orm::{DatabaseBackend, MockDatabase, MockExecResult};
    use tower::ServiceExt;

    fn make_router(verifier: Arc<dyn SignatureVerifier>, db: DatabaseConnection) -> Router {
        Router::new()
            .route("/inbox", post(handle))
            .layer(Extension(InboxState { verifier }))
            .layer(Extension(db))
    }

    fn signed_request(body: &str) -> Request<Body> {
        Request::builder()
            .uri("/inbox")
            .method("POST")
            .header("signature", "keyId=\"test\",signature=\"...\"")
            .header("content-type", "application/activity+json")
            .body(Body::from(body.to_string()))
            .unwrap()
    }

    struct AlwaysAccept;

    #[async_trait]
    impl SignatureVerifier for AlwaysAccept {
        async fn verify(&self, _ctx: &VerifyCtx<'_>) -> Result<(), AppError> {
            Ok(())
        }
    }

    struct AlwaysReject;

    #[async_trait]
    impl SignatureVerifier for AlwaysReject {
        async fn verify(&self, _ctx: &VerifyCtx<'_>) -> Result<(), AppError> {
            Err(AppError::Internal("nope".into()))
        }
    }

    /// F8 enforcement helper: seed three mock results so that
    /// `FediverseTrustService::assert_domain_allowed` answers Ok with
    /// no real DB. Mode is `open_with_blocklist` + empty blocklist, the
    /// cheapest permissive shape for unit tests.
    fn permissive_trust_mock(mock: MockDatabase) -> MockDatabase {
        use crate::models::fediverse_blocked_domain;
        use rustwarden_core::models::app_settings;
        let now = chrono::Utc::now();
        mock.append_query_results([vec![app_settings::Model {
            id: Uuid::new_v4(),
            key: "fediverse_enabled".to_string(),
            value: "true".to_string(),
            description: None,
            updated_by: None,
            updated_at: now,
        }]])
        .append_query_results([vec![app_settings::Model {
            id: Uuid::new_v4(),
            key: "fediverse_trust_mode".to_string(),
            value: "open_with_blocklist".to_string(),
            description: None,
            updated_by: None,
            updated_at: now,
        }]])
        .append_query_results([Vec::<fediverse_blocked_domain::Model>::new()])
    }

    fn db_for_update() -> DatabaseConnection {
        let now = chrono::Utc::now();
        let row = crate::models::remote_identity::Model {
            id: Uuid::new_v4(),
            ap_id: "https://a.tld/users/alice".into(),
            username: "alice".into(),
            domain: "a.tld".into(),
            display_name: None,
            summary_html: None,
            inbox_url: "https://a.tld/users/alice/inbox".into(),
            shared_inbox_url: None,
            public_key_pem: String::new(),
            icon_url: None,
            actor_type: "Person".into(),
            raw_json: None,
            last_fetched_at: now,
            disabled_at: None,
            created_at: now,
            updated_at: now,
        };
        let mut after = row.clone();
        after.display_name = Some("Alice".into());
        permissive_trust_mock(MockDatabase::new(DatabaseBackend::Postgres))
            .append_query_results([vec![row]])
            .append_query_results([vec![after]])
            .append_exec_results([MockExecResult {
                last_insert_id: 0,
                rows_affected: 1,
            }])
            .into_connection()
    }

    fn db_for_delete_noop() -> DatabaseConnection {
        // SELECT returns empty → DeleteOutcome::default(), no UPDATE.
        permissive_trust_mock(MockDatabase::new(DatabaseBackend::Postgres))
            .append_query_results([Vec::<crate::models::remote_identity::Model>::new()])
            .into_connection()
    }

    fn db_for_revoke_noop() -> DatabaseConnection {
        permissive_trust_mock(MockDatabase::new(DatabaseBackend::Postgres))
            .append_exec_results([MockExecResult {
                last_insert_id: 0,
                rows_affected: 0,
            }])
            .into_connection()
    }

    #[tokio::test]
    async fn inbox_returns_202_for_valid_update_actor() {
        let body = serde_json::json!({
            "type": "Update",
            "actor": "https://a.tld/users/alice",
            "object": {
                "id": "https://a.tld/users/alice",
                "type": "Person",
                "name": "Alice",
            }
        })
        .to_string();
        let router = make_router(Arc::new(AlwaysAccept), db_for_update());
        let resp = router.oneshot(signed_request(&body)).await.unwrap();
        assert_eq!(resp.status(), StatusCode::ACCEPTED);
    }

    #[tokio::test]
    async fn inbox_returns_202_for_valid_delete_actor() {
        let body = serde_json::json!({
            "type": "Delete",
            "actor": "https://a.tld/users/alice",
            "object": "https://a.tld/users/alice"
        })
        .to_string();
        let router = make_router(Arc::new(AlwaysAccept), db_for_delete_noop());
        let resp = router.oneshot(signed_request(&body)).await.unwrap();
        assert_eq!(resp.status(), StatusCode::ACCEPTED);
    }

    #[tokio::test]
    async fn inbox_returns_202_for_valid_revoke_membership() {
        let team_id = Uuid::new_v4();
        let remote_id = Uuid::new_v4();
        let body = serde_json::json!({
            "type": "Revoke",
            "actor": "https://a.tld/users/alice",
            "object": {
                "type": "TeamMembership",
                "team": format!("https://b.tld/teams/{team_id}"),
                "subject": remote_id.to_string(),
            }
        })
        .to_string();
        let router = make_router(Arc::new(AlwaysAccept), db_for_revoke_noop());
        let resp = router.oneshot(signed_request(&body)).await.unwrap();
        assert_eq!(resp.status(), StatusCode::ACCEPTED);
    }

    #[tokio::test]
    async fn inbox_returns_401_when_signature_header_missing() {
        let body = "{\"type\":\"Update\"}";
        let req = Request::builder()
            .uri("/inbox")
            .method("POST")
            .body(Body::from(body))
            .unwrap();
        let router = make_router(
            Arc::new(AlwaysAccept),
            MockDatabase::new(DatabaseBackend::Postgres).into_connection(),
        );
        let resp = router.oneshot(req).await.unwrap();
        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
    }

    #[tokio::test]
    async fn inbox_returns_401_when_verifier_rejects() {
        let body = serde_json::json!({
            "type": "Update", "actor": "https://a.tld/users/alice"
        })
        .to_string();
        let router = make_router(
            Arc::new(AlwaysReject),
            permissive_trust_mock(MockDatabase::new(DatabaseBackend::Postgres)).into_connection(),
        );
        let resp = router.oneshot(signed_request(&body)).await.unwrap();
        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
    }

    #[tokio::test]
    async fn inbox_returns_400_for_invalid_json() {
        let router = make_router(
            Arc::new(AlwaysAccept),
            MockDatabase::new(DatabaseBackend::Postgres).into_connection(),
        );
        let resp = router.oneshot(signed_request("not json")).await.unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
    }

    #[tokio::test]
    async fn inbox_returns_501_for_unknown_activity_type() {
        let body = serde_json::json!({
            "type": "Like", "actor": "https://a.tld/users/alice"
        })
        .to_string();
        let router = make_router(
            Arc::new(AlwaysAccept),
            permissive_trust_mock(MockDatabase::new(DatabaseBackend::Postgres)).into_connection(),
        );
        let resp = router.oneshot(signed_request(&body)).await.unwrap();
        assert_eq!(resp.status(), StatusCode::NOT_IMPLEMENTED);
    }

    #[tokio::test]
    async fn inbox_smoke_test_with_always_accept_verifier() {
        use crate::services::fediverse_signal_service::AlwaysAcceptVerifier;
        let body = serde_json::json!({
            "type": "Delete",
            "actor": "https://a.tld/users/ghost",
            "object": "https://a.tld/users/ghost"
        })
        .to_string();
        let router = make_router(Arc::new(AlwaysAcceptVerifier::new()), db_for_delete_noop());
        let resp = router.oneshot(signed_request(&body)).await.unwrap();
        assert_eq!(resp.status(), StatusCode::ACCEPTED);
    }

    // ─── RFC 0028 / Lot 7b — observer emission points ──────────────

    use std::sync::atomic::{AtomicUsize, Ordering};

    #[derive(Default)]
    struct CountingObserver {
        reasons: std::sync::Mutex<Vec<String>>,
        calls: AtomicUsize,
        delete_calls: std::sync::Mutex<Vec<(String, String, Uuid)>>,
    }

    #[async_trait]
    impl FediverseEventObserver for CountingObserver {
        async fn on_inbox_abuse(&self, _domain: &str, _actor: Option<&str>, reason: &str) {
            self.calls.fetch_add(1, Ordering::SeqCst);
            self.reasons.lock().unwrap().push(reason.to_string());
        }

        async fn on_delete_actor(&self, domain: &str, ap_id: &str, remote_identity_id: Uuid) {
            self.delete_calls.lock().unwrap().push((
                domain.to_string(),
                ap_id.to_string(),
                remote_identity_id,
            ));
        }
    }

    fn make_router_with_observer(
        verifier: Arc<dyn SignatureVerifier>,
        db: DatabaseConnection,
        observer: Arc<dyn FediverseEventObserver>,
    ) -> Router {
        Router::new()
            .route("/inbox", post(handle))
            .layer(Extension(InboxState { verifier }))
            .layer(Extension(db))
            .layer(Extension(observer))
    }

    #[tokio::test]
    async fn inbox_emits_malformed_json_observer_event() {
        let observer = Arc::new(CountingObserver::default());
        let router = make_router_with_observer(
            Arc::new(AlwaysAccept),
            MockDatabase::new(DatabaseBackend::Postgres).into_connection(),
            observer.clone(),
        );
        let resp = router.oneshot(signed_request("not json")).await.unwrap();
        assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
        let reasons = observer.reasons.lock().unwrap().clone();
        assert_eq!(reasons, vec!["malformed_json".to_string()]);
    }

    fn db_for_delete_hit(id: Uuid) -> DatabaseConnection {
        let now = chrono::Utc::now();
        let row = crate::models::remote_identity::Model {
            id,
            ap_id: "https://a.tld/users/alice".into(),
            username: "alice".into(),
            domain: "a.tld".into(),
            display_name: None,
            summary_html: None,
            inbox_url: "https://a.tld/users/alice/inbox".into(),
            shared_inbox_url: None,
            public_key_pem: String::new(),
            icon_url: None,
            actor_type: "Person".into(),
            raw_json: None,
            last_fetched_at: now,
            disabled_at: None,
            created_at: now,
            updated_at: now,
        };
        let mut after = row.clone();
        after.disabled_at = Some(now);
        permissive_trust_mock(MockDatabase::new(DatabaseBackend::Postgres))
            .append_query_results([vec![row]])
            .append_query_results([vec![after]])
            .append_exec_results([MockExecResult {
                last_insert_id: 0,
                rows_affected: 1,
            }])
            .into_connection()
    }

    #[tokio::test]
    async fn inbox_emits_delete_actor_observer_event_when_row_disabled() {
        let id = Uuid::new_v4();
        let body = serde_json::json!({
            "type": "Delete",
            "actor": "https://a.tld/users/alice",
            "object": "https://a.tld/users/alice"
        })
        .to_string();
        let observer = Arc::new(CountingObserver::default());
        let router = make_router_with_observer(
            Arc::new(AlwaysAccept),
            db_for_delete_hit(id),
            observer.clone(),
        );
        let resp = router.oneshot(signed_request(&body)).await.unwrap();
        assert_eq!(resp.status(), StatusCode::ACCEPTED);
        let calls = observer.delete_calls.lock().unwrap().clone();
        assert_eq!(calls.len(), 1);
        assert_eq!(calls[0].0, "a.tld");
        assert_eq!(calls[0].1, "https://a.tld/users/alice");
        assert_eq!(calls[0].2, id);
    }

    #[tokio::test]
    async fn inbox_does_not_emit_delete_actor_observer_event_when_row_missing() {
        let body = serde_json::json!({
            "type": "Delete",
            "actor": "https://a.tld/users/ghost",
            "object": "https://a.tld/users/ghost"
        })
        .to_string();
        let observer = Arc::new(CountingObserver::default());
        let router = make_router_with_observer(
            Arc::new(AlwaysAccept),
            db_for_delete_noop(),
            observer.clone(),
        );
        let resp = router.oneshot(signed_request(&body)).await.unwrap();
        assert_eq!(resp.status(), StatusCode::ACCEPTED);
        assert!(observer.delete_calls.lock().unwrap().is_empty());
    }

    #[tokio::test]
    async fn inbox_emits_unknown_activity_type_observer_event() {
        let body = serde_json::json!({
            "type": "Like",
            "actor": "https://a.tld/users/alice"
        })
        .to_string();
        let observer = Arc::new(CountingObserver::default());
        let router = make_router_with_observer(
            Arc::new(AlwaysAccept),
            permissive_trust_mock(MockDatabase::new(DatabaseBackend::Postgres)).into_connection(),
            observer.clone(),
        );
        let resp = router.oneshot(signed_request(&body)).await.unwrap();
        assert_eq!(resp.status(), StatusCode::NOT_IMPLEMENTED);
        let reasons = observer.reasons.lock().unwrap().clone();
        assert_eq!(reasons, vec!["unknown_activity_type".to_string()]);
    }
}