maarcadetweet/crates/appview/src/ingest.rs

//! `POST /internal/ingest-commit` — used by the PDS to push local commits
//! into the AppView so the user's own actions show up without waiting for
//! the Jetstream round-trip.
//!
//! Wire shape:
//! ```json
//! {
//!   "did":            "did:plc:abc",
//!   "collection":     "app.twi.post",
//!   "action":         "create",
//!   "rkey":           "3k2...",
//!   "cid":            "bafy...",   // optional
//!   "record":         { ... },     // optional; required for follow delete
//!   "subject_did":    "did:plc:..." // required for app.bsky.graph.follow
//! }
//! ```
//!
//! In production this endpoint would be protected with mTLS and a token
//! minted by the PDS; for now it's open inside the cluster.

use crate::indexer;
use crate::state::AppState;
use axum::extract::State;
use axum::http::{HeaderMap, StatusCode};
use axum::Json;
use serde::Deserialize;
use serde_json::Value;
use tracing::{info, warn};

#[derive(Debug, Deserialize)]
pub struct IngestCommitReq {
    pub did: String,
    /// The poster's current handle, as known by the PDS `users` table.
    /// Optional in the wire payload — the AppView falls back to an
    /// empty string, and the upsert COALESCE guard prevents the empty
    /// value from clobbering a backfilled handle from the Jetstream
    /// `identity` event path.
    #[serde(default)]
    pub handle: Option<String>,
    pub collection: String,
    pub action: String,
    pub rkey: String,
    #[serde(default)]
    pub cid: Option<String>,
    #[serde(default)]
    pub record: Option<Value>,
    /// Required for `app.bsky.graph.follow` because the record value isn't
    /// always preserved on delete events.
    #[serde(default)]
    pub subject_did: Option<String>,
}

/// Authenticate internal ingest requests.
/// - If `APPVIEW_INGEST_SECRET` env var is unset: dev mode, accept anything.
/// - If set: require `X-Ingest-Secret: <value>` header to match.
pub fn check_ingest_secret(
    headers: &HeaderMap,
    configured: Option<&str>,
) -> Result<(), (StatusCode, Json<Value>)> {
    let Some(expected) = configured else {
        return Ok(()); // dev mode
    };
    let provided = headers
        .get("x-ingest-secret")
        .and_then(|v| v.to_str().ok())
        .unwrap_or("");
    if constant_time_eq(provided.as_bytes(), expected.as_bytes()) {
        Ok(())
    } else {
        Err((
            StatusCode::UNAUTHORIZED,
            Json(serde_json::json!({
                "error": "AuthenticationRequired",
                "message": "missing or invalid X-Ingest-Secret",
            })),
        ))
    }
}

fn constant_time_eq(a: &[u8], b: &[u8]) -> bool {
    if a.len() != b.len() {
        return false;
    }
    let mut diff = 0u8;
    for (x, y) in a.iter().zip(b.iter()) {
        diff |= x ^ y;
    }
    diff == 0
}

pub async fn ingest_commit(
    State(state): State<AppState>,
    headers: HeaderMap,
    Json(req): Json<IngestCommitReq>,
) -> Result<Json<Value>, (StatusCode, Json<Value>)> {
    check_ingest_secret(&headers, state.cfg.appview_ingest_secret.as_deref())?;

    let result = apply(&state, &req).await;
    if let Err((status, body)) = &result {
        warn!(
            status = status.as_u16(),
            body = %body.0,
            did = %req.did,
            collection = %req.collection,
            action = %req.action,
            "ingest commit failed"
        );
    } else {
        info!(did = %req.did, collection = %req.collection,
              action = %req.action, rkey = %req.rkey, "ingested commit");
    }
    result.map(|applied| {
        Json(serde_json::json!({
            "ok": true,
            "applied": applied,
        }))
    })
}

async fn apply(
    state: &AppState,
    req: &IngestCommitReq,
) -> Result<bool, (StatusCode, Json<Value>)> {
    match (req.collection.as_str(), req.action.as_str()) {
        ("app.twi.post", "create") | ("app.bsky.feed.post", "create") => {
            let record = req
                .record
                .clone()
                .unwrap_or_else(|| serde_json::json!({"text": "", "createdAt": chrono::Utc::now().to_rfc3339()}));
            let cid = req.cid.clone().unwrap_or_default();
            let mut row = indexer::PostRow::from_record(
                &req.did,
                &req.rkey,
                &req.collection,
                &cid,
                &record,
                req.handle.as_deref(),
            );
            indexer::upsert_post(&state.db, &mut row).await.map_err(db_err)?;
            Ok(true)
        }
        ("app.twi.post", "delete") | ("app.bsky.feed.post", "delete") => {
            let uri = format!("at://{}/{}/{}", req.did, req.collection, req.rkey);
            indexer::delete_post(&state.db, &uri).await.map_err(db_err)?;
            Ok(true)
        }
        ("app.bsky.feed.like", "create") => {
            indexer::upsert_like(
                &state.db,
                &req.did,
                &req.rkey,
                req.cid.as_deref(),
                req.record.as_ref(),
            )
            .await
            .map_err(db_err)?;
            Ok(true)
        }
        ("app.bsky.feed.like", "delete") => {
            indexer::delete_like(&state.db, &req.did, &req.rkey)
                .await
                .map_err(db_err)?;
            Ok(true)
        }
        ("app.bsky.feed.repost", "create") => {
            indexer::upsert_repost(
                &state.db,
                &req.did,
                &req.rkey,
                req.cid.as_deref(),
                req.record.as_ref(),
            )
            .await
            .map_err(db_err)?;
            Ok(true)
        }
        ("app.bsky.feed.repost", "delete") => {
            indexer::delete_repost(&state.db, &req.did, &req.rkey)
                .await
                .map_err(db_err)?;
            Ok(true)
        }
        ("app.bsky.graph.follow", "create") => {
            let subject = req
                .subject_did
                .clone()
                .or_else(|| {
                    req.record
                        .as_ref()
                        .and_then(|r| r.get("subject"))
                        .and_then(|s| s.as_str())
                        .map(str::to_string)
                })
                .ok_or_else(|| bad_request("follow create requires subject_did or record.subject"))?;
            indexer::upsert_follow(
                &state.db,
                &req.did,
                &subject,
                req.record.as_ref(),
            )
            .await
            .map_err(db_err)?;
            Ok(true)
        }
        ("app.bsky.graph.follow", "delete") => {
            let subject = req
                .subject_did
                .clone()
                .or_else(|| {
                    req.record
                        .as_ref()
                        .and_then(|r| r.get("subject"))
                        .and_then(|s| s.as_str())
                        .map(str::to_string)
                })
                .ok_or_else(|| bad_request("follow delete requires subject_did"))?;
            indexer::delete_follow(&state.db, &req.did, &subject)
                .await
                .map_err(db_err)?;
            Ok(true)
        }
        ("app.bsky.actor.profile", "create") if req.rkey == "self" => {
            // Profile record push from the PDS — populate the
            // `profiles` cache so the ProfileView-Page and PostCard
            // avatar get the new display name / bio / avatar / banner
            // without waiting for the next handle-sync pass.
            let record = match &req.record {
                Some(r) if !r.is_null() => r.clone(),
                _ => return Ok(false),
            };
            // Use the handle the PDS provided when present. We
            // deliberately do NOT fall back to a DB lookup here:
            // the AppView has no `users` table — the PDS owns that
            // state. If the PDS omits the handle, we write an empty
            // string and the `handle_sync` worker (or a subsequent
            // Jetstream `identity` event) will fill it in.
            let handle = req
                .handle
                .clone()
                .filter(|h| !h.is_empty())
                .unwrap_or_default();
            indexer::upsert_profile(&state.db, &req.did, &handle, &record)
                .await
                .map_err(db_err)?;
            Ok(true)
        }
        ("app.bsky.actor.profile", "delete") if req.rkey == "self" => {
            sqlx::query("DELETE FROM profiles WHERE did = $1")
                .bind(&req.did)
                .execute(&state.db)
                .await
                .map_err(db_err)?;
            Ok(true)
        }
        (coll, action) => {
            // Unrecognised collection/action — return ok=false so the PDS
            // doesn't retry. Future collections should be added above.
            tracing::debug!(collection = %coll, action = %action, "ingest: unhandled");
            Ok(false)
        }
    }
}

fn db_err(e: impl std::fmt::Display) -> (StatusCode, Json<Value>) {
    (
        StatusCode::INTERNAL_SERVER_ERROR,
        Json(serde_json::json!({
            "error": "InternalServerError",
            "message": e.to_string(),
        })),
    )
}

fn bad_request(msg: &str) -> (StatusCode, Json<Value>) {
    (
        StatusCode::BAD_REQUEST,
        Json(serde_json::json!({
            "error": "InvalidRequest",
            "message": msg,
        })),
    )
}

#[cfg(test)]
mod tests {
    use super::*;
    use axum::http::HeaderValue;

    #[test]
    fn no_secret_configured_allows_anonymous() {
        let h = HeaderMap::new();
        assert!(check_ingest_secret(&h, None).is_ok());
    }

    #[test]
    fn secret_required_when_configured() {
        let h = HeaderMap::new();
        assert!(check_ingest_secret(&h, Some("hunter2")).is_err());
    }

    #[test]
    fn secret_matches() {
        let mut h = HeaderMap::new();
        h.insert("x-ingest-secret", HeaderValue::from_static("hunter2"));
        assert!(check_ingest_secret(&h, Some("hunter2")).is_ok());
    }

    #[test]
    fn secret_mismatched() {
        let mut h = HeaderMap::new();
        h.insert("x-ingest-secret", HeaderValue::from_static("hunter3"));
        assert!(check_ingest_secret(&h, Some("hunter2")).is_err());
    }
}