//! PDS-side client that pushes local commits into the AppView's
//! `/internal/ingest-commit` endpoint.
//!
//! Why
//!
//! The AppView normally learns about a record via the Jetstream
//! round-trip. That's a few seconds of latency and a second moving
//! part to debug when it's down. Pushing directly from the PDS makes
//! the user's own writes visible in their own timeline the instant
//! they hit `POST /xrpc/com.atproto.repo.createRecord`.
//!
//! Failure model
//!
//! The push is best-effort. We never block a record write on the
//! AppView being reachable — if the AppView is down, the record is
//! already committed in the PDS's repo + blockstore, and the next
//! Jetstream replay will eventually pick it up. The push is logged
//! so an operator can detect persistent AppView outages.
//!
//! The PDS and AppView share a `X-Ingest-Secret` token (configured via
//! `APPVIEW_INGEST_SECRET` on both sides). When unset on the AppView
//! side the endpoint accepts anonymous requests (dev mode), so the
//! client doesn't bother sending the header in that case either.

use anyhow::{Context, Result};
use reqwest::header::HeaderMap;
use reqwest::Client;
use serde::Serialize;
use serde_json::Value;
use std::time::Duration;

#[derive(Debug, Serialize)]
struct IngestCommitBody<'a> {
    did: &'a str,
    collection: &'a str,
    action: &'a str,
    rkey: &'a str,
    cid: Option<&'a str>,
    record: Option<&'a Value>,
    subject_did: Option<&'a str>,
}

#[derive(Clone)]
pub struct AppViewPushClient {
    base_url: String,
    secret: Option<String>,
    client: Client,
}

impl AppViewPushClient {
    pub fn new(base_url: impl Into<String>, secret: Option<String>) -> Self {
        Self {
            base_url: base_url.into(),
            secret,
            client: Client::builder()
                .timeout(Duration::from_secs(5))
                .build()
                .unwrap(),
        }
    }

    /// Push a `create` event to the AppView. `record` should be the full
    /// AT-Protocol record value as JSON — the AppView's indexer reads
    /// `embed` / `reply` off it, which is why we can't just send the CID.
    ///
    /// Returns `Ok(true)` if the AppView applied the commit, `Ok(false)`
    /// if it returned a non-2xx status (logged as warn), and `Err(_)` if
    /// the request itself failed. The caller should treat any non-Ok as
    /// "the AppView will learn about this via Jetstream eventually".
    pub async fn push_create(
        &self,
        did: &str,
        collection: &str,
        rkey: &str,
        cid: &str,
        record: &Value,
    ) -> Result<bool> {
        self.push(
            did,
            collection,
            "create",
            rkey,
            Some(cid),
            Some(record),
            None,
        )
        .await
    }

    pub async fn push_delete(
        &self,
        did: &str,
        collection: &str,
        rkey: &str,
    ) -> Result<bool> {
        self.push(did, collection, "delete", rkey, None, None, None)
            .await
    }

    pub async fn push_follow_create(
        &self,
        did: &str,
        rkey: &str,
        subject_did: &str,
        record: &Value,
    ) -> Result<bool> {
        self.push(
            did,
            "app.bsky.graph.follow",
            "create",
            rkey,
            None,
            Some(record),
            Some(subject_did),
        )
        .await
    }

    pub async fn push_follow_delete(
        &self,
        did: &str,
        rkey: &str,
        subject_did: &str,
    ) -> Result<bool> {
        self.push(
            did,
            "app.bsky.graph.follow",
            "delete",
            rkey,
            None,
            None,
            Some(subject_did),
        )
        .await
    }

    async fn push(
        &self,
        did: &str,
        collection: &str,
        action: &str,
        rkey: &str,
        cid: Option<&str>,
        record: Option<&Value>,
        subject_did: Option<&str>,
    ) -> Result<bool> {
        let url = format!("{}/internal/ingest-commit", self.base_url);
        let body = IngestCommitBody {
            did,
            collection,
            action,
            rkey,
            cid,
            record,
            subject_did,
        };
        let mut req = self.client.post(&url).json(&body);
        if let Some(secret) = self.secret.as_deref() {
            let mut headers = HeaderMap::new();
            headers.insert(
                "x-ingest-secret",
                secret.parse().context("invalid ingest secret header value")?,
            );
            req = req.headers(headers);
        }
        let resp = req
            .send()
            .await
            .context("appview: ingest-commit send failed")?;
        if !resp.status().is_success() {
            let status = resp.status();
            let body = resp.text().await.unwrap_or_default();
            tracing::warn!(
                status = status.as_u16(),
                body,
                did,
                collection,
                action,
                rkey,
                "appview: ingest-commit returned non-success"
            );
            return Ok(false);
        }
        Ok(true)
    }
}