use super::test_support::*;
use crate::indexing::projection::{
    ProjectionEvent, ProjectionInput, project_eager_record, project_event,
};
use jacquard_common::deps::bytes::Bytes;
use jacquard_common::error::{AuthError, ClientError};
use jacquard_common::types::value::{Data, to_data};
use jacquard_common::xrpc::Response as XrpcResponse;
use polymodel_api::com_atproto::repo::get_record::GetRecordResponse;

#[test]
fn parse_record_ref_accepts_valid_create_response() {
    let did = did(DID_A);
    let collection =
        jacquard_common::types::nsid::Nsid::new_owned("space.polymodel.library.thing").unwrap();
    let output = format!(
        r#"{{"uri":"at://{DID_A}/space.polymodel.library.thing/t1","cid":"bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"}}"#
    );
    let (rkey, cid) = super::proxy::parse_record_ref(output.as_bytes(), &did, &collection)
        .expect("valid typed response");
    assert_eq!(rkey.as_str(), "t1");
    assert!(cid.is_valid());
}

#[test]
fn parse_record_ref_rejects_malformed_json() {
    let did = did(DID_A);
    let collection =
        jacquard_common::types::nsid::Nsid::new_owned("space.polymodel.library.thing").unwrap();
    assert!(super::proxy::parse_record_ref(b"not json", &did, &collection).is_none());
}

#[test]
fn parse_record_ref_rejects_missing_fields() {
    let did = did(DID_A);
    let collection =
        jacquard_common::types::nsid::Nsid::new_owned("space.polymodel.library.thing").unwrap();
    assert!(
        super::proxy::parse_record_ref(
            br#"{"uri":"at://did:plc:aaaaaaaaaaaaaaaaaaaaaaaa/space.polymodel.library.thing/t1"}"#,
            &did,
            &collection,
        )
        .is_none()
    );
}

#[test]
fn parse_record_ref_rejects_trailing_slash() {
    let did = did(DID_A);
    let collection =
        jacquard_common::types::nsid::Nsid::new_owned("space.polymodel.library.thing").unwrap();
    assert!(super::proxy::parse_record_ref(
        br#"{"uri":"at://did:plc:aaaaaaaaaaaaaaaaaaaaaaaa/space.polymodel.library.thing/","cid":"bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"}"#,
        &did,
        &collection,
    ).is_none());
}

#[test]
fn parse_record_ref_rejects_mismatched_authority() {
    let did = did(DID_A);
    let collection =
        jacquard_common::types::nsid::Nsid::new_owned("space.polymodel.library.thing").unwrap();
    assert!(super::proxy::parse_record_ref(
        br#"{"uri":"at://did:plc:bbbbbbbbbbbbbbbbbbbbbbbb/space.polymodel.library.thing/t1","cid":"bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"}"#,
        &did,
        &collection,
    ).is_none());
}

#[test]
fn parse_record_ref_rejects_mismatched_collection() {
    let did = did(DID_A);
    let collection =
        jacquard_common::types::nsid::Nsid::new_owned("space.polymodel.library.thing").unwrap();
    assert!(super::proxy::parse_record_ref(
        br#"{"uri":"at://did:plc:aaaaaaaaaaaaaaaaaaaaaaaa/space.polymodel.library.model/t1","cid":"bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"}"#,
        &did,
        &collection,
    ).is_none());
}

#[test]
fn parse_record_ref_rejects_empty_rkey() {
    let did = did(DID_A);
    let collection =
        jacquard_common::types::nsid::Nsid::new_owned("space.polymodel.library.thing").unwrap();
    assert!(super::proxy::parse_record_ref(
        br#"{"uri":"at://did:plc:aaaaaaaaaaaaaaaaaaaaaaaa/space.polymodel.library.thing","cid":"bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"}"#,
        &did,
        &collection,
    ).is_none());
}

#[test]
fn parse_record_ref_rejects_invalid_cid() {
    let did = did(DID_A);
    let collection =
        jacquard_common::types::nsid::Nsid::new_owned("space.polymodel.library.thing").unwrap();
    assert!(super::proxy::parse_record_ref(
        br#"{"uri":"at://did:plc:aaaaaaaaaaaaaaaaaaaaaaaa/space.polymodel.library.thing/t1","cid":"not-a-cid"}"#,
        &did,
        &collection,
    ).is_none());
}

#[test]
fn parse_record_cid_accepts_valid_get_record_response() {
    let cid = super::proxy::parse_record_cid(
        br#"{"cid":"bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"}"#,
    )
    .expect("valid cid");
    assert!(cid.is_valid());
}

#[test]
fn parse_record_cid_rejects_malformed_json() {
    assert!(super::proxy::parse_record_cid(b"not json").is_none());
}

#[test]
fn parse_record_cid_rejects_missing_cid() {
    assert!(super::proxy::parse_record_cid(br#"{"uri":"at://did:plc:a"}"#).is_none());
}

#[test]
fn parse_record_cid_rejects_invalid_cid() {
    assert!(super::proxy::parse_record_cid(br#"{"cid":"not-a-cid"}"#).is_none());
}

#[tokio::test]
async fn forward_xrpc_preserves_success_body_and_json_content_type() {
    let body = b"{\"uri\":\"at://x\"}";
    let resp = super::proxy::forward_xrpc::<GetRecordResponse>(Ok(XrpcResponse::new(
        Bytes::from_static(body),
        StatusCode::OK,
    )));
    assert_eq!(resp.status(), StatusCode::OK);
    assert_eq!(resp.headers()["content-type"], "application/json");
    assert_eq!(body_text(resp).await, std::str::from_utf8(body).unwrap());
}

#[tokio::test]
async fn forward_xrpc_preserves_error_status_body_and_json_content_type() {
    let body = b"{\"error\":\"InvalidSwap\"}";
    let resp = super::proxy::forward_xrpc::<GetRecordResponse>(Err(ClientError::http(
        StatusCode::CONFLICT,
        Some(Bytes::from_static(body)),
    )));
    assert_eq!(resp.status(), StatusCode::CONFLICT);
    assert_eq!(resp.headers()["content-type"], "application/json");
    assert_eq!(body_text(resp).await, std::str::from_utf8(body).unwrap());
}

// ---------------------------------------------------------------------------
// PM-47 passthrough proxy: faithful forwarding, read-your-writes override,
// eager projection, structural allowlist, and session gating.
// ---------------------------------------------------------------------------

#[tokio::test]
async fn forward_xrpc_passes_through_success_body() {
    let resp = crate::appview::proxy::forward_xrpc::<GetRecordResponse>(Ok(XrpcResponse::new(
        Bytes::from_static(b"{\"uri\":\"at://x\"}"),
        StatusCode::OK,
    )));
    assert_eq!(resp.status(), StatusCode::OK);
    assert_eq!(body_text(resp).await, "{\"uri\":\"at://x\"}");
}

#[tokio::test]
async fn forward_xrpc_forwards_400_xrpc_error_verbatim() {
    // atproto getRecord RecordNotFound is HTTP 400 + an error body (route c: an
    // Ok response with a non-success status). The proxy must NOT turn it into 500.
    let resp = crate::appview::proxy::forward_xrpc::<GetRecordResponse>(Ok(XrpcResponse::new(
        Bytes::from_static(b"{\"error\":\"RecordNotFound\",\"message\":\"Record not found\"}"),
        StatusCode::BAD_REQUEST,
    )));
    assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
    assert!(body_text(resp).await.contains("RecordNotFound"));
}

#[tokio::test]
async fn forward_xrpc_forwards_http_error_status_and_body() {
    // route b: Err(Http) for a non-2xx that isn't 400/401 (e.g. 409 InvalidSwap).
    let err = ClientError::http(
        StatusCode::CONFLICT,
        Some(Bytes::from_static(b"{\"error\":\"InvalidSwap\"}")),
    );
    let resp = crate::appview::proxy::forward_xrpc::<GetRecordResponse>(Err(err));
    assert_eq!(resp.status(), StatusCode::CONFLICT);
    assert!(body_text(resp).await.contains("InvalidSwap"));
}

#[tokio::test]
async fn forward_xrpc_forwards_429_and_500_verbatim() {
    for status in [
        StatusCode::TOO_MANY_REQUESTS,
        StatusCode::INTERNAL_SERVER_ERROR,
    ] {
        let err = ClientError::http(status, Some(Bytes::from_static(b"{}")));
        let resp = crate::appview::proxy::forward_xrpc::<GetRecordResponse>(Err(err));
        assert_eq!(resp.status(), status);
    }
}

#[tokio::test]
async fn forward_xrpc_auth_error_is_401() {
    let resp = crate::appview::proxy::forward_xrpc::<GetRecordResponse>(Err(ClientError::auth(
        AuthError::NotAuthenticated,
    )));
    assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
}

#[tokio::test]
async fn forward_xrpc_transport_error_is_500() {
    // Genuine transport/internal failures are the only thing that becomes 500.
    let err = ClientError::transport(std::io::Error::other("boom"));
    let resp = crate::appview::proxy::forward_xrpc::<GetRecordResponse>(Err(err));
    assert_eq!(resp.status(), StatusCode::INTERNAL_SERVER_ERROR);
}

#[tokio::test]
async fn override_returns_local_for_pending_write_with_divergent_cid() {
    let pool = pool().await;
    let uri = format!("at://{DID_A}/space.polymodel.library.thing/t1");
    sqlx::query("INSERT INTO pending_writes (uri, cid, value, written_at) VALUES (?, ?, ?, ?)")
        .bind(&uri)
        .bind("LOCALCID")
        .bind(json!({"name": "fresh"}).to_string())
        .bind(1i64)
        .execute(&pool)
        .await
        .unwrap();
    let resp = crate::appview::proxy::try_get_record_override(&pool, &uri, Some("PDSCID"))
        .await
        .unwrap()
        .expect("divergent pending write overrides the PDS result");
    assert_eq!(resp.status(), StatusCode::OK);
    let body = body_text(resp).await;
    assert!(body.contains("LOCALCID"), "{body}");
    assert!(body.contains("fresh"), "{body}");
    assert!(!body.contains("stale"), "{body}");
}

#[tokio::test]
async fn override_returns_not_found_for_pending_delete() {
    let pool = pool().await;
    let uri = format!("at://{DID_A}/space.polymodel.library.thing/t1");
    sqlx::query("INSERT INTO pending_deletes (uri, deleted_at) VALUES (?, ?)")
        .bind(&uri)
        .bind(1i64)
        .execute(&pool)
        .await
        .unwrap();
    let resp = crate::appview::proxy::try_get_record_override(&pool, &uri, Some("PDSCID"))
        .await
        .unwrap()
        .expect("pending delete overrides the PDS result");
    assert_eq!(resp.status(), StatusCode::BAD_REQUEST);
    assert!(body_text(resp).await.contains("RecordNotFound"));
}

#[tokio::test]
async fn override_is_none_when_no_pending_state() {
    let pool = pool().await;
    let uri = format!("at://{DID_A}/space.polymodel.library.thing/t1");
    assert!(
        crate::appview::proxy::try_get_record_override(&pool, &uri, Some("PDSCID"))
            .await
            .unwrap()
            .is_none()
    );
}

#[tokio::test]
async fn override_is_none_when_pending_write_cid_matches_pds() {
    let pool = pool().await;
    let uri = format!("at://{DID_A}/space.polymodel.library.thing/t1");
    sqlx::query("INSERT INTO pending_writes (uri, cid, value, written_at) VALUES (?, ?, ?, ?)")
        .bind(&uri)
        .bind("SAMECID")
        .bind("{}")
        .bind(1i64)
        .execute(&pool)
        .await
        .unwrap();
    // Cids match → the firehose has effectively confirmed → forward the PDS.
    assert!(
        crate::appview::proxy::try_get_record_override(&pool, &uri, Some("SAMECID"))
            .await
            .unwrap()
            .is_none()
    );
}

#[tokio::test]
async fn override_returns_local_when_pds_reports_not_found() {
    // A pending write must win even when a lagging PDS reports not-found
    // (pds_cid is None) rather than a divergent record — the core
    // read-your-writes case.
    let pool = pool().await;
    let uri = format!("at://{DID_A}/space.polymodel.library.thing/t1");
    sqlx::query("INSERT INTO pending_writes (uri, cid, value, written_at) VALUES (?, ?, ?, ?)")
        .bind(&uri)
        .bind("LOCALCID")
        .bind(json!({"name": "fresh"}).to_string())
        .bind(1i64)
        .execute(&pool)
        .await
        .unwrap();
    let resp = crate::appview::proxy::try_get_record_override(&pool, &uri, None)
        .await
        .unwrap()
        .expect("pending write wins over a lagging not-found");
    assert_eq!(resp.status(), StatusCode::OK);
    let body = body_text(resp).await;
    assert!(body.contains("LOCALCID"), "{body}");
    assert!(body.contains("fresh"), "{body}");
}

fn poly_event(action: &str, record: Option<Data>) -> ProjectionEvent {
    ProjectionEvent {
        seq: 1,
        did: DID_A.to_string(),
        collection: "space.polymodel.library.thing".to_string(),
        rkey: "t1".to_string(),
        action: action.to_string(),
        record,
        cid: Some("C1".to_string()),
    }
}

#[tokio::test]
async fn eager_write_marks_pending_and_firehose_clears_it() {
    let pool = pool().await;
    let uri = format!("at://{DID_A}/space.polymodel.library.thing/t1");
    let record = to_data(
        &json!({"name": "x", "license": "CC-BY-4.0", "createdAt": "2024-01-01T00:00:00.000Z"}),
    )
    .unwrap();
    project_eager_record(&pool, &poly_event("create", Some(record.clone())))
        .await
        .unwrap();
    let pending_cid: String = sqlx::query_scalar("SELECT cid FROM pending_writes WHERE uri = ?")
        .bind(&uri)
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(pending_cid, "C1");

    // The Hydrant #commit re-delivering the same create clears the pending row.
    project_event(
        &pool,
        &ProjectionInput::Record(poly_event("create", Some(record))),
    )
    .await
    .unwrap();
    let remaining: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM pending_writes WHERE uri = ?")
        .bind(&uri)
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(remaining, 0);
}

#[tokio::test]
async fn eager_delete_marks_pending_and_firehose_clears_it() {
    let pool = pool().await;
    let uri = format!("at://{DID_A}/space.polymodel.library.thing/t1");
    project_eager_record(&pool, &poly_event("delete", None))
        .await
        .unwrap();
    let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM pending_deletes WHERE uri = ?")
        .bind(&uri)
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(n, 1);

    project_event(&pool, &ProjectionInput::Record(poly_event("delete", None)))
        .await
        .unwrap();
    let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM pending_deletes WHERE uri = ?")
        .bind(&uri)
        .fetch_one(&pool)
        .await
        .unwrap();
    assert_eq!(n, 0);
}

#[tokio::test]
async fn proxy_structural_allowlist_404s_non_allowlisted_methods() {
    let state = state().await;
    let app = crate::appview::router().with_state(state);
    for path in [
        "/xrpc/com.atproto.repo.applyWrites",
        "/xrpc/com.atproto.repo.importRepo",
        "/xrpc/com.atproto.repo.listMissingBlobs",
        "/xrpc/app.bsky.feed.getTimeline",
        "/xrpc/com.atproto.server.getSession",
    ] {
        let resp = app
            .clone()
            .oneshot(Request::builder().uri(path).body(Body::empty()).unwrap())
            .await
            .unwrap();
        assert_eq!(
            resp.status(),
            StatusCode::NOT_FOUND,
            "{path} must not be proxied"
        );
    }
}

#[tokio::test]
async fn proxy_pds_routes_require_an_oauth_session() {
    let state = state().await;
    let app = crate::appview::router().with_state(state);
    // createRecord (procedure) with no session → 401.
    let resp = app
        .clone()
        .oneshot(
            Request::builder()
                .method("POST")
                .uri("/xrpc/com.atproto.repo.createRecord")
                .header("content-type", "application/json")
                .body(Body::from("{}"))
                .unwrap(),
        )
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
    // getRecord (query) with no session → 401.
    let resp = app
        .oneshot(
            Request::builder()
                .uri(
                    "/xrpc/com.atproto.repo.getRecord?repo=did:plc:x\
                     &collection=space.polymodel.library.thing&rkey=t1",
                )
                .body(Body::empty())
                .unwrap(),
        )
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
}