//! XRPC passthrough proxy to the user's PDS (PM-47).
//!
//! A fixed, structural allowlist of non-`space.polymodel.*` XRPC methods is
//! served from this origin so the browser stays single-origin (PM-9 §4.5).
//! Each allowlisted method dispatches to its generated typed request struct and
//! is forwarded to the user's PDS (authenticated methods, via the OAuth session
//! agent's *signing* send path) or the public AppView (public methods). Upstream
//! XRPC errors are forwarded with their real HTTP status + body; only genuine
//! transport/internal failures become 500.
//!
//! Allowlist (the direct repo record operations only):
//! - `com.atproto.repo.{createRecord, putRecord, deleteRecord, getRecord,
//!   listRecords, describeRepo, uploadBlob}` — authenticated, to the user's PDS.
//! - `com.atproto.identity.resolveHandle` and `app.bsky.actor.getProfile` —
//!   public, to the AppView.
//!
//! Batch/admin repo methods (`applyWrites`, `importRepo`, `listMissingBlobs`)
//! and every other NSID have no route and 404.
//!
//! Read-your-writes: `repo.getRecord` of a `space.polymodel.*` record reflects a
//! not-yet-firehose-confirmed local write/delete via the `pending_writes` /
//! `pending_deletes` tables populated by the shared eager-projection path (see
//! `indexing::projection`). Proxied writes to `space.polymodel.*` are eagerly
//! projected through that same path for read-your-own-writes.

use axum::body::Body;
use axum::extract::State;
use axum::http::{HeaderMap, HeaderValue, StatusCode, header};
use axum::response::{IntoResponse, Response};
use jacquard::client::{Agent, AgentSession};
use jacquard::identity::PublicResolver;
use jacquard::oauth::client::OAuthSession;
use jacquard_axum::ExtractXrpc;
use jacquard_axum::oauth::ExtractOAuthSession;
use jacquard_common::error::{ClientError, ClientErrorKind, HttpError};
use jacquard_common::types::ident::AtIdentifier;
use jacquard_common::types::nsid::Nsid;
use jacquard_common::types::recordkey::Rkey;
use jacquard_common::types::string::{AtUri, Cid, Did};
use jacquard_common::types::value::to_data;
use jacquard_common::xrpc::{Response as XrpcResponse, XrpcClient, XrpcResp};
use polymodel_api::com_atproto::identity::resolve_handle::ResolveHandleRequest;
use polymodel_api::com_atproto::repo::{
    create_record::CreateRecordRequest,
    delete_record::DeleteRecordRequest,
    describe_repo::DescribeRepoRequest,
    get_record::GetRecordRequest,
    list_records::ListRecordsRequest,
    put_record::PutRecordRequest,
    upload_blob::{UploadBlob, UploadBlobRequest},
};
use serde_json::Value;
use sqlx::SqlitePool;

use super::content_type::content_type;
use super::error::{AppError, AppResult, db, invalid_request, unauthorized};
use super::state::AppState;
use crate::indexing::projection::{ProjectionEvent, project_eager_record};
use crate::oauth::SqliteAuthStore;

type UserAgent = Agent<OAuthSession<PublicResolver, SqliteAuthStore>>;

const POLYMODEL_PREFIX: &str = "space.polymodel.";

// ===========================================================================
// PDS reads (authenticated; forwarded to the user's PDS)
// ===========================================================================

pub(super) async fn repo_get_record(
    State(state): State<AppState>,
    ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>,
    ExtractXrpc(req): ExtractXrpc<GetRecordRequest>,
) -> AppResult<Response> {
    let agent = Agent::from(session);
    let collection = req.collection.as_ref().to_string();
    let polymodel = collection.starts_with(POLYMODEL_PREFIX);
    // Pending-ops key: canonicalize the request repo to a DID (handles resolve)
    // so read-your-writes matches the DID-keyed pending rows for both DID- and
    // handle-authority reads. A resolved *other* repo can't match this account's
    // pending rows (the actor only writes its own repo), so there's no cross-repo
    // false positive. If the repo can't be canonicalized, skip the override and
    // forward the PDS result.
    let pending_uri = if polymodel {
        match super::actor::resolve_actor(&state, &req.repo).await {
            Ok(did) => {
                let rkey = req.rkey.as_ref().to_string();
                Some(format!("at://{}/{}/{}", did.as_ref(), collection, rkey))
            }
            Err(e) => {
                tracing::warn!(error = ?e, "pending-ops key: repo did not canonicalize; skipping override");
                None
            }
        }
    } else {
        None
    };
    let result = agent.send(req).await;
    if let Some(uri) = pending_uri {
        // Read-your-writes: a polymodel record may carry a pending local
        // write/delete the PDS/AppView hasn't reflected yet. Keyed on the
        // request — the PDS body is only consulted for cid comparison — so a
        // pending write still wins when a lagging source reports not-found.
        let pds_cid = match &result {
            Ok(resp) if resp.status().is_success() => {
                parse_record_cid(resp.buffer()).map(|cid| cid.as_str().to_owned())
            }
            _ => None,
        };
        match try_get_record_override(&state.pool, &uri, pds_cid.as_deref()).await {
            Ok(Some(overridden)) => return Ok(overridden),
            Ok(None) => {}
            Err(e) => {
                tracing::warn!(error = ?e, "pending-ops override lookup failed; forwarding PDS result")
            }
        }
    }
    Ok(forward_xrpc(result))
}

pub(super) async fn repo_list_records(
    ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>,
    ExtractXrpc(req): ExtractXrpc<ListRecordsRequest>,
) -> AppResult<Response> {
    let agent = Agent::from(session);
    Ok(forward_xrpc(agent.send(req).await))
}

pub(super) async fn repo_describe_repo(
    ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>,
    ExtractXrpc(req): ExtractXrpc<DescribeRepoRequest>,
) -> AppResult<Response> {
    let agent = Agent::from(session);
    Ok(forward_xrpc(agent.send(req).await))
}

// ===========================================================================
// PDS writes (authenticated; forwarded to the user's PDS, eagerly projected)
// ===========================================================================

pub(super) async fn repo_create_record(
    State(state): State<AppState>,
    ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>,
    ExtractXrpc(req): ExtractXrpc<CreateRecordRequest>,
) -> AppResult<Response> {
    let agent = Agent::from(session);
    let did = actor_did(&agent).await?;
    let collection = req.collection.clone();
    let record = serde_json::to_value(&req.record).ok();
    let result = agent.send(req).await;
    if let Ok(resp) = &result
        && resp.status().is_success()
        && collection.as_str().starts_with(POLYMODEL_PREFIX)
        && let Some(record) = record.as_ref()
    {
        eager_project(&state, &did, &collection, "create", record, resp.buffer()).await;
    }
    Ok(forward_xrpc(result))
}

pub(super) async fn repo_put_record(
    State(state): State<AppState>,
    ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>,
    ExtractXrpc(req): ExtractXrpc<PutRecordRequest>,
) -> AppResult<Response> {
    let agent = Agent::from(session);
    let did = actor_did(&agent).await?;
    let collection = req.collection.clone();
    let record = serde_json::to_value(&req.record).ok();
    let result = agent.send(req).await;
    if let Ok(resp) = &result
        && resp.status().is_success()
        && collection.as_str().starts_with(POLYMODEL_PREFIX)
        && let Some(record) = record.as_ref()
    {
        eager_project(&state, &did, &collection, "update", record, resp.buffer()).await;
    }
    Ok(forward_xrpc(result))
}

pub(super) async fn repo_delete_record(
    State(state): State<AppState>,
    ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>,
    ExtractXrpc(req): ExtractXrpc<DeleteRecordRequest>,
) -> AppResult<Response> {
    let agent = Agent::from(session);
    let did = actor_did(&agent).await?;
    let collection = req.collection.as_ref().to_string();
    let rkey = req.rkey.as_ref().to_string();
    let result = agent.send(req).await;
    if let Ok(resp) = &result
        && resp.status().is_success()
        && collection.starts_with(POLYMODEL_PREFIX)
    {
        let event = ProjectionEvent {
            seq: 0,
            did: did.as_ref().to_string(),
            collection: collection.clone(),
            rkey,
            action: "delete".to_string(),
            record: None,
            cid: None,
        };
        let _guard = state.write_lock.lock().await;
        if let Err(e) = project_eager_record(&state.pool, &event).await {
            tracing::error!(error = %e, %collection, "eager projection of proxied delete failed");
        }
    }
    Ok(forward_xrpc(result))
}

pub(super) async fn repo_upload_blob(
    ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>,
    headers: HeaderMap,
    ExtractXrpc(req): ExtractXrpc<UploadBlobRequest>,
) -> AppResult<Response> {
    let agent = Agent::from(session);
    let UploadBlob { body } = req;
    // The generated `UploadBlob` carries raw `*/*` bytes and no mime; thread the
    // real inbound Content-Type through the call options (the same mechanism the
    // `AgentSessionExt::upload_blob` wrapper uses) so the PDS receives it. Using
    // `send_with_opts` directly preserves the raw response for faithful
    // status/body forwarding, which the typed wrapper would swallow.
    let mime = content_type(&headers);
    let mut opts = agent.opts().await;
    opts.extra_headers.push((
        header::CONTENT_TYPE,
        HeaderValue::from_str(&mime)
            .map_err(|e| invalid_request(format!("invalid content-type header: {e}")))?,
    ));
    Ok(forward_xrpc(
        agent.send_with_opts(UploadBlob { body }, opts).await,
    ))
}

// ===========================================================================
// Public AppView reads (unauthenticated; forwarded to public.api.bsky.app)
// ===========================================================================

pub(super) async fn identity_resolve_handle(
    State(state): State<AppState>,
    ExtractXrpc(req): ExtractXrpc<ResolveHandleRequest>,
) -> AppResult<Response> {
    Ok(forward_xrpc(state.bsky.send(req).await))
}

pub(super) async fn actor_get_profile(
    State(state): State<AppState>,
    ExtractXrpc(req): ExtractXrpc<polymodel_api::app_bsky::actor::get_profile::GetProfileRequest>,
) -> AppResult<Response> {
    Ok(forward_xrpc(state.bsky.send(req).await))
}

// ===========================================================================
// Faithful upstream forwarding + helpers
// ===========================================================================

/// Forward an upstream `agent.send` result as a raw Axum response, preserving
/// the real HTTP status and body rather than collapsing it to a single code:
/// - `Ok` (success **or** a 400/401 XRPC-error body) → that status + the buffer.
/// - `Err(Http)` (other non-2xx, e.g. 409/404/429/5xx) → that status + body.
/// - `Err(Auth)` → 401.
/// - anything else (transport/decode/…) → 500 (genuine internal failure).
///
/// The typed `XrpcError` enum carries no status; the HTTP status comes from the
/// response/error itself, so atproto error codes (`RecordNotFound` on HTTP 400,
/// `InvalidSwap` on HTTP 409, …) are forwarded verbatim.
pub(super) fn forward_xrpc<Resp: XrpcResp>(
    result: Result<XrpcResponse<Resp>, ClientError>,
) -> Response {
    match result {
        Ok(resp) => raw_response(resp.status(), resp.buffer().to_vec()),
        Err(err) => match err.kind() {
            ClientErrorKind::Http { status } => {
                let body = err
                    .source_err()
                    .and_then(|s| s.downcast_ref::<HttpError>())
                    .and_then(|h| h.body.clone())
                    .map(|b| b.to_vec())
                    .unwrap_or_default();
                raw_response(*status, body)
            }
            ClientErrorKind::Auth(_) => {
                AppError::Unauthorized("upstream authentication required".to_string())
                    .into_response()
            }
            other => {
                tracing::error!(error = ?other, "passthrough upstream transport/internal failure");
                AppError::Internal("passthrough upstream failure".to_string()).into_response()
            }
        },
    }
}

fn raw_response(status: StatusCode, body: Vec<u8>) -> Response {
    Response::builder()
        .status(status)
        .header(header::CONTENT_TYPE, "application/json")
        .body(Body::from(body))
        .expect("static response parts are always valid")
}

async fn actor_did(agent: &UserAgent) -> AppResult<Did> {
    agent
        .session_info()
        .await
        .map(|(did, _)| did)
        .ok_or_else(|| unauthorized("OAuth session is required"))
}

/// Eagerly project a successful proxied create/update into the local SQLite
/// projection (rkey/cid read from the PDS output) for read-your-writes.
async fn eager_project(
    state: &AppState,
    did: &Did,
    collection: &Nsid,
    action: &str,
    record: &Value,
    output_buffer: &[u8],
) {
    let Some((rkey, cid)) = parse_record_ref(output_buffer, did, collection) else {
        tracing::warn!(%collection, "invalid proxied record reference; skipping eager projection");
        return;
    };
    let record_data = match to_data(record) {
        Ok(d) => d,
        Err(e) => {
            tracing::warn!(error = %e, %collection, "proxied record to_data failed");
            return;
        }
    };
    let event = ProjectionEvent {
        seq: 0,
        did: did.as_ref().to_string(),
        collection: collection.as_str().to_string(),
        rkey: rkey.as_str().to_string(),
        action: action.to_string(),
        record: Some(record_data),
        cid: Some(cid.as_str().to_owned()),
    };
    // Serialize against PM-28's decision-making writes (which read local state
    // then mutate): the proxied write itself makes no decision, but its local
    // projection must not interleave with a locked read-then-write section. The
    // lock guards only the local projection, never the upstream PDS call.
    let _guard = state.write_lock.lock().await;
    if let Err(e) = project_eager_record(&state.pool, &event).await {
        tracing::error!(error = %e, %collection, %action, "eager projection of proxied write failed");
    }
}

#[derive(serde::Deserialize)]
struct RecordRef<'a> {
    #[serde(borrow)]
    uri: AtUri<&'a str>,
    #[serde(borrow)]
    cid: Cid<&'a str>,
}

#[derive(serde::Deserialize)]
struct RecordCid<'a> {
    #[serde(borrow)]
    cid: Cid<&'a str>,
}

/// Parse and validate `(rkey, cid)` from a createRecord/putRecord response.
pub(super) fn parse_record_ref<'a>(
    output: &'a [u8],
    expected_did: &Did,
    expected_collection: &Nsid,
) -> Option<(Rkey<&'a str>, Cid<&'a str>)> {
    let out: RecordRef<'a> = serde_json::from_slice(output).ok()?;
    if !out.cid.is_valid() {
        return None;
    }
    match out.uri.authority() {
        AtIdentifier::Did(did) if did.as_ref() == expected_did.as_ref() => {}
        _ => return None,
    }
    if out.uri.collection()?.as_ref() != expected_collection.as_str() {
        return None;
    }
    let rkey = out.uri.rkey()?;
    if rkey.as_str().is_empty() {
        return None;
    }
    Some((rkey, out.cid))
}

/// Parse and semantically validate the CID from a getRecord response.
pub(super) fn parse_record_cid<'a>(output: &'a [u8]) -> Option<Cid<&'a str>> {
    let out: RecordCid<'a> = serde_json::from_slice(output).ok()?;
    out.cid.is_valid().then_some(out.cid)
}

/// Read-your-writes override for a single `space.polymodel.*` getRecord.
///
/// Returns a replacement response when a not-yet-firehose-confirmed local op
/// should win over the PDS/AppView.
///
/// - A pending delete → `RecordNotFound` (HTTP 400).
/// - A pending write the PDS has not caught up to → the local value/cid (200).
///   "Not caught up" means `pds_cid` is `None` (the PDS reported not-found or
///   errored) or differs from the pending cid.
///
/// Returns `None` otherwise (no pending state, or a pending write whose cid
/// already matches the PDS) so the caller forwards the PDS result.
pub(super) async fn try_get_record_override(
    pool: &SqlitePool,
    uri: &str,
    pds_cid: Option<&str>,
) -> AppResult<Option<Response>> {
    let pending_delete = db(sqlx::query!(
        "SELECT 1 AS hit FROM pending_deletes WHERE uri = ?",
        uri
    )
    .fetch_optional(pool)
    .await)?;
    if pending_delete.is_some() {
        return Ok(Some(record_not_found_response()));
    }

    let pending_write = db(sqlx::query_as!(
        PendingWrite,
        "SELECT cid, value FROM pending_writes WHERE uri = ?",
        uri
    )
    .fetch_optional(pool)
    .await)?;
    if let Some(pw) = pending_write
        && pds_cid != Some(pw.cid.as_str())
    {
        let value: Value = serde_json::from_str(&pw.value).unwrap_or(Value::Null);
        let body = serde_json::to_vec(&serde_json::json!({
            "uri": uri,
            "cid": pw.cid,
            "value": value,
        }))
        .unwrap_or_default();
        return Ok(Some(raw_response(StatusCode::OK, body)));
    }
    Ok(None)
}

#[derive(sqlx::FromRow)]
struct PendingWrite {
    cid: String,
    value: String,
}

fn record_not_found_response() -> Response {
    // atproto repo.getRecord RecordNotFound is HTTP 400 with this body shape.
    let body = serde_json::to_vec(&serde_json::json!({
        "error": "RecordNotFound",
        "message": "Record not found",
    }))
    .unwrap_or_default();
    raw_response(StatusCode::BAD_REQUEST, body)
}