//! Tests for the PM-26 appview read layer.
//!
//! The read helpers are exercised against an in-memory SQLite database with the
//! migrations applied and representative rows seeded directly (the projection is
//! tested separately in `indexing::projection`). The bsky profile fetch is never
//! hit here: the polymodel union variant comes from seeded SQLite, and the bsky
//! variant's union discrimination is covered by a serde round-trip.

#![cfg(test)]

use axum::body::Body;
use axum::http::{Request, StatusCode};
use jacquard_common::types::string::Did;
use polymodel_api::space_polymodel::actor::ProfileViewUnion;
use serde_json::json;
use sqlx::SqlitePool;
use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
use std::str::FromStr;
use tower::ServiceExt;

use super::error::AppError;
use super::state::AppState;
use super::views;

const DID_A: &str = "did:plc:aaaaaaaaaaaaaaaaaaaaaaaa";
const DID_B: &str = "did:plc:bbbbbbbbbbbbbbbbbbbbbbbb";

fn did(s: &str) -> Did {
    Did::new_owned(s).unwrap()
}

async fn pool() -> SqlitePool {
    let options = SqliteConnectOptions::from_str("sqlite::memory:").unwrap();
    let pool = SqlitePoolOptions::new()
        .max_connections(1)
        .connect_with(options)
        .await
        .unwrap();
    sqlx::migrate!("./migrations").run(&pool).await.unwrap();
    pool
}

async fn state() -> AppState {
    let pool = pool().await;
    let bootstrap = crate::oauth::bootstrap_oauth(pool.clone(), Some("http://localhost"))
        .expect("ephemeral OAuth bootstrap for tests");
    AppState::new(pool, bootstrap)
}

// Seeding uses runtime queries so the committed .sqlx cache stays focused on production SQL.

fn blob(cid: &str) -> serde_json::Value {
    json!({
        "$type": "blob",
        "ref": { "$link": cid },
        "mimeType": "image/jpeg",
        "size": 1234
    })
}

fn image(cid: &str, alt: &str) -> serde_json::Value {
    json!({
        "alt": alt,
        "aspectRatio": { "width": 4, "height": 3 },
        "image": blob(cid)
    })
}

fn strong_ref(uri: &str) -> serde_json::Value {
    json!({ "uri": uri, "cid": "bafyreicid" })
}

async fn seed_thing(
    pool: &SqlitePool,
    did: &str,
    rkey: &str,
    name: &str,
    models: &[&str],
    like_count: i64,
) -> String {
    let uri = format!("at://{did}/space.polymodel.library.thing/{rkey}");
    let now = chrono::Utc::now()
        .timestamp_nanos_opt()
        .expect("current timestamp must fit in i64 nanoseconds");
    let created = now - 86_400_000_000_000; // 1 day ago by default
    sqlx::query(
        r#"INSERT INTO things (did, rkey, uri, cid, name, summary, license, tags_json,
           tags_text, instructions_text, cover_json, derived_from_uri, created_at, indexed_at, record_json)
           VALUES (?, ?, ?, ?, ?, NULL, 'CC-BY-4.0', NULL, NULL, NULL, NULL, NULL, ?, ?, ?)"#,
    )
    .bind(did)
    .bind(rkey)
    .bind(&uri)
    .bind("cid")
    .bind(name)
    .bind(created)
    .bind(now)
    .bind(json!({
        "name": name,
        "license": "CC-BY-4.0",
        "cover": [image("bafkreiathingcover", "Thing cover")],
        "previews": [image("bafkreiathingpreview", "Thing preview")],
        "models": models.iter().map(|m| strong_ref(m)).collect::<Vec<_>>(),
        "createdAt": "2024-01-01T00:00:00.000Z"
    }).to_string())
    .execute(pool)
    .await
    .unwrap();
    for (pos, m) in models.iter().enumerate() {
        sqlx::query(
            "INSERT OR REPLACE INTO thing_models (thing_uri, model_uri, position) VALUES (?, ?, ?)",
        )
        .bind(&uri)
        .bind(*m)
        .bind(pos as i64)
        .execute(pool)
        .await
        .unwrap();
    }
    sqlx::query(
        "INSERT OR IGNORE INTO content_stats (uri, like_count, save_count, tag_count) VALUES (?, ?, 0, 0)",
    )
    .bind(&uri)
    .bind(like_count)
    .execute(pool)
    .await
    .unwrap();
    uri
}

async fn seed_model(
    pool: &SqlitePool,
    did: &str,
    rkey: &str,
    name: &str,
    parts: &[&str],
) -> String {
    let uri = format!("at://{did}/space.polymodel.library.model/{rkey}");
    let now = chrono::Utc::now()
        .timestamp_nanos_opt()
        .expect("current timestamp must fit in i64 nanoseconds");
    sqlx::query(
        r#"INSERT INTO models (did, rkey, uri, cid, name, summary, created_at, indexed_at, record_json)
           VALUES (?, ?, ?, ?, ?, NULL, ?, ?, ?)"#,
    )
    .bind(did)
    .bind(rkey)
    .bind(&uri)
    .bind("cid")
    .bind(name)
    .bind(now)
    .bind(now)
    .bind(json!({
        "name": name,
        "cover": [image("bafkreiamodelcover", "Model cover")],
        "previews": [image("bafkreiamodelpreview", "Model preview")],
        "parts": parts.iter().map(|p| strong_ref(p)).collect::<Vec<_>>(),
        "createdAt": "2024-01-01T00:00:00.000Z"
    }).to_string())
    .execute(pool)
    .await
    .unwrap();
    for (pos, p) in parts.iter().enumerate() {
        sqlx::query(
            "INSERT OR REPLACE INTO model_parts (model_uri, part_uri, position) VALUES (?, ?, ?)",
        )
        .bind(&uri)
        .bind(*p)
        .bind(pos as i64)
        .execute(pool)
        .await
        .unwrap();
    }
    uri
}

async fn seed_part(pool: &SqlitePool, did: &str, rkey: &str, name: &str) -> String {
    let uri = format!("at://{did}/space.polymodel.library.part/{rkey}");
    let now = chrono::Utc::now()
        .timestamp_nanos_opt()
        .expect("current timestamp must fit in i64 nanoseconds");
    sqlx::query(
        r#"INSERT INTO parts (did, rkey, uri, cid, name, format, file_json, created_at, indexed_at, record_json)
           VALUES (?, ?, ?, ?, ?, NULL, ?, ?, ?, ?)"#,
    )
    .bind(did)
    .bind(rkey)
    .bind(&uri)
    .bind("cid")
    .bind(name)
    .bind(json!({"mimeType": "model/stl", "size": 1, "chunks": []}).to_string())
    .bind(now)
    .bind(now)
    .bind(json!({
        "name": name,
        "file": {"mimeType": "model/stl", "size": 1, "chunks": []},
        "previews": [image("bafkreiapartpreview", "Part preview")],
        "createdAt": "2024-01-01T00:00:00.000Z"
    }).to_string())
    .execute(pool)
    .await
    .unwrap();
    uri
}

async fn seed_identity(pool: &SqlitePool, did: &str, handle: &str) {
    sqlx::query(
        "INSERT INTO identities (did, handle, updated_at) VALUES (?, ?, ?) \
         ON CONFLICT(did) DO UPDATE SET handle = excluded.handle",
    )
    .bind(did)
    .bind(handle)
    .bind(
        chrono::Utc::now()
            .timestamp_nanos_opt()
            .expect("current timestamp must fit in i64 nanoseconds"),
    )
    .execute(pool)
    .await
    .unwrap();
}

async fn seed_profile(pool: &SqlitePool, did: &str, display: &str) {
    let avatar = blob("bafkreiavatar").to_string();
    sqlx::query(
        r#"INSERT INTO profiles (did, display_name, description, avatar_json, default_license,
           pronouns, printers_json, links_json, record_json, indexed_at)
           VALUES (?, ?, NULL, ?, NULL, NULL, NULL, NULL, ?, ?)
           ON CONFLICT(did) DO UPDATE SET display_name = excluded.display_name,
           avatar_json = excluded.avatar_json, record_json = excluded.record_json"#,
    )
    .bind(did)
    .bind(display)
    .bind(avatar)
    .bind(json!({"displayName": display, "avatar": blob("bafkreiavatar")}).to_string())
    .bind(
        chrono::Utc::now()
            .timestamp_nanos_opt()
            .expect("current timestamp must fit in i64 nanoseconds"),
    )
    .execute(pool)
    .await
    .unwrap();
}

#[tokio::test]
async fn get_thing_hydrates_thing_with_ordered_models_and_record() {
    let state = state().await;
    let m1 = seed_model(&state.pool, DID_A, "m1", "Model One", &[]).await;
    let m2 = seed_model(&state.pool, DID_A, "m2", "Model Two", &[]).await;
    let thing = seed_thing(&state.pool, DID_A, "t1", "Cube", &[&m2, &m1], 3).await;
    seed_identity(&state.pool, DID_A, "alice.com").await;
    seed_profile(&state.pool, DID_A, "Alice").await;

    let (view, models) = views::get_thing(&state, &thing, None).await.unwrap();
    assert_eq!(view.uri.as_ref(), thing);
    assert_eq!(view.name.as_str(), "Cube");
    assert_eq!(view.like_count, 3);
    assert_eq!(view.model_count, 2);
    assert_eq!(
        view.cover.as_ref().unwrap()[0].thumb.as_ref(),
        "https://cdn.bsky.app/img/feed_thumbnail/plain/did:plc:aaaaaaaaaaaaaaaaaaaaaaaa/bafkreiathingcover@jpeg"
    );
    assert_eq!(
        view.previews.as_ref().unwrap()[0].fullsize.as_ref(),
        "https://cdn.bsky.app/img/feed_thumbnail/plain/did:plc:aaaaaaaaaaaaaaaaaaaaaaaa/bafkreiathingpreview@jpeg"
    );
    assert_eq!(
        view.author.avatar.as_ref().unwrap().as_ref(),
        "https://cdn.bsky.app/img/avatar/plain/did:plc:aaaaaaaaaaaaaaaaaaaaaaaa/bafkreiavatar@jpeg"
    );
    // record must be populated from record_json (never null).
    assert!(
        serde_json::to_string(&view.record)
            .unwrap()
            .contains("Cube")
    );
    // Junction order is preserved (m2 before m1 by position).
    assert_eq!(models.len(), 2);
    assert_eq!(models[0].uri.as_ref(), m2);
    assert_eq!(models[1].uri.as_ref(), m1);
    // Author handle comes from the identity projection.
    assert_eq!(view.author.handle.as_ref(), "alice.com");
}

#[tokio::test]
async fn get_thing_missing_is_not_found() {
    let state = state().await;
    let err = views::get_thing(
        &state,
        "at://did:plc:x/space.polymodel.library.thing/none",
        None,
    )
    .await
    .unwrap_err();
    assert!(matches!(err, AppError::NotFound));
}

#[tokio::test]
async fn get_model_hydrates_parts_and_parent_thing() {
    let state = state().await;
    seed_identity(&state.pool, DID_A, "alice.com").await;
    let p1 = seed_part(&state.pool, DID_A, "p1", "Part One").await;
    let p2 = seed_part(&state.pool, DID_A, "p2", "Part Two").await;
    let model = seed_model(&state.pool, DID_A, "m1", "Model", &[&p2, &p1]).await;
    let thing = seed_thing(&state.pool, DID_A, "t1", "Thing", &[&model], 0).await;

    let detail = views::get_model(&state, &model, None).await.unwrap();
    assert_eq!(detail.model.uri.as_ref(), model);
    assert_eq!(detail.model.part_count, 2);
    assert_eq!(
        detail.model.cover.as_ref().unwrap()[0].thumb.as_ref(),
        "https://cdn.bsky.app/img/feed_thumbnail/plain/did:plc:aaaaaaaaaaaaaaaaaaaaaaaa/bafkreiamodelcover@jpeg"
    );
    assert_eq!(detail.parts.len(), 2);
    assert_eq!(detail.parts[0].uri.as_ref(), p2); // ordered by position
    assert_eq!(
        detail.parts[0].previews.as_ref().unwrap()[0].thumb.as_ref(),
        "https://cdn.bsky.app/img/feed_thumbnail/plain/did:plc:aaaaaaaaaaaaaaaaaaaaaaaa/bafkreiapartpreview@jpeg"
    );
    assert_eq!(detail.thing.as_ref().unwrap().uri.as_ref(), thing);
    assert!(detail.thing.as_ref().unwrap().previews.is_some());
}

#[tokio::test]
async fn author_things_keyset_paginates_by_rkey() {
    let state = state().await;
    seed_identity(&state.pool, DID_A, "alice.com").await;
    // rkeys are time-ordered TIDs; ordered rkeys make rkey DESC deterministic.
    seed_thing(&state.pool, DID_A, "3zzzzzzzzzzzz", "t3", &[], 0).await;
    seed_thing(&state.pool, DID_A, "3yyyyyyyyyyyy", "t2", &[], 0).await;
    seed_thing(&state.pool, DID_A, "3xxxxxxxxxxxx", "t1", &[], 0).await;

    let page1 = views::author_things(&state, DID_A, 2, None, None)
        .await
        .unwrap();
    assert_eq!(page1.items.len(), 2);
    assert_eq!(page1.items[0].thing.name.as_str(), "t3");
    let cursor = page1.cursor.expect("cursor present when more remain");

    let page2 = views::author_things(&state, DID_A, 2, Some(cursor.as_ref()), None)
        .await
        .unwrap();
    assert_eq!(page2.items.len(), 1);
    assert_eq!(page2.items[0].thing.name.as_str(), "t1");
    assert!(page2.cursor.is_none());
}

#[tokio::test]
async fn feed_hot_is_deterministic_with_tied_scores() {
    let state = state().await;
    seed_identity(&state.pool, DID_A, "alice.com").await;
    // Two things with identical engagement and near-identical age -> tied score;
    // ordering must still be deterministic via the rkey tie-break.
    seed_thing(&state.pool, DID_A, "3aaaaaaaaaaaa", "a", &[], 5).await;
    seed_thing(&state.pool, DID_A, "3bbbbbbbbbbbb", "b", &[], 5).await;

    let feed = views::feed_hot(&state, 10, None, None).await.unwrap();
    assert_eq!(feed.items.len(), 2);
    let feed_again = views::feed_hot(&state, 10, None, None).await.unwrap();
    assert_eq!(
        feed.items[0].thing.uri.as_ref(),
        feed_again.items[0].thing.uri.as_ref()
    );
    // Tied scores break by rkey DESC.
    assert_eq!(feed.items[0].thing.name.as_str(), "b");
}

#[tokio::test]
async fn search_things_uses_fts() {
    let state = state().await;
    seed_identity(&state.pool, DID_A, "alice.com").await;
    seed_thing(&state.pool, DID_A, "r1", "Calibration Cube", &[], 0).await;
    seed_thing(&state.pool, DID_A, "r2", "Friendly Sphere", &[], 0).await;

    let feed = views::search_things(&state, "cube", 10, None, None)
        .await
        .unwrap();
    assert_eq!(feed.items.len(), 1);
    assert_eq!(feed.items[0].thing.name.as_str(), "Calibration Cube");
}

#[tokio::test]
async fn malformed_ranked_cursor_is_rejected() {
    let state = state().await;
    let err = views::search_things(&state, "x", 10, Some("not-a-number"), None)
        .await
        .unwrap_err();
    assert!(matches!(err, AppError::InvalidRequest(_)));
}

#[tokio::test]
async fn list_filters_non_thing_subjects() {
    let state = state().await;
    seed_identity(&state.pool, DID_A, "alice.com").await;
    let thing = seed_thing(&state.pool, DID_A, "t1", "Thing", &[], 0).await;
    let list = "at://did:plc:l/space.polymodel.graph.list/self";
    let non_thing = "at://did:plc:x/space.polymodel.graph.like/k1";
    sqlx::query("INSERT INTO listitems (did, rkey, list_uri, subject_uri, created_at) VALUES (?, ?, ?, ?, ?)")
        .bind(DID_A).bind("i1").bind(list).bind(&thing).bind(1)
        .execute(&state.pool).await.unwrap();
    sqlx::query("INSERT INTO listitems (did, rkey, list_uri, subject_uri, created_at) VALUES (?, ?, ?, ?, ?)")
        .bind(DID_A).bind("i2").bind(list).bind(non_thing).bind(2)
        .execute(&state.pool).await.unwrap();

    let (items, _cursor) = views::list_things(&state, list, 10, None, None)
        .await
        .unwrap();
    assert_eq!(items.len(), 1, "non-thing list item must be filtered out");
    assert_eq!(items[0].uri.as_ref(), thing);
}

#[tokio::test]
async fn polymodel_profile_variant_carries_stats() {
    let state = state().await;
    seed_identity(&state.pool, DID_A, "alice.com").await;
    seed_profile(&state.pool, DID_A, "Alice").await;
    seed_thing(&state.pool, DID_A, "t1", "Thing One", &[], 4).await;
    seed_thing(&state.pool, DID_A, "t2", "Thing Two", &[], 6).await;

    let view = views::polymodel_profile_view(&state, &did(DID_A))
        .await
        .unwrap()
        .expect("polymodel profile exists");
    assert_eq!(view.did.as_ref(), DID_A);
    assert_eq!(view.handle.as_str(), "alice.com");
    assert_eq!(
        view.display_name.as_ref().map(|s| s.as_str()),
        Some("Alice")
    );
    assert_eq!(view.thing_count, Some(2));
    assert_eq!(view.total_likes, Some(10));
    assert_eq!(
        view.avatar.as_ref().unwrap().as_ref(),
        "https://cdn.bsky.app/img/avatar/plain/did:plc:aaaaaaaaaaaaaaaaaaaaaaaa/bafkreiavatar@jpeg"
    );
}

#[tokio::test]
async fn polymodel_profile_absent_returns_none() {
    let state = state().await;
    let view = views::polymodel_profile_view(&state, &did(DID_B))
        .await
        .unwrap();
    assert!(view.is_none());
}

#[tokio::test]
async fn union_serializes_type_tag() {
    // The polymodel variant must discriminate on $type without a network fetch.
    let state = state().await;
    seed_identity(&state.pool, DID_A, "alice.com").await;
    seed_profile(&state.pool, DID_A, "Alice").await;
    let view = views::polymodel_profile_view(&state, &did(DID_A))
        .await
        .unwrap()
        .unwrap();
    let union = ProfileViewUnion::ProfileView(Box::new(view));
    let json = serde_json::to_value(&union).unwrap();
    assert_eq!(
        json.get("$type").and_then(|v| v.as_str()),
        Some("space.polymodel.actor.defs#profileView")
    );
}

// --- router-level (ExtractXrpc + handler + response) ---

async fn body_text(resp: axum::response::Response) -> String {
    let bytes = axum::body::to_bytes(resp.into_body(), 1 << 20)
        .await
        .unwrap();
    String::from_utf8(bytes.to_vec()).unwrap()
}

#[tokio::test]
async fn get_session_route_returns_unauthenticated_marker() {
    let state = state().await;
    let app = super::router().with_state(state);
    let resp = app
        .oneshot(
            Request::builder()
                .uri("/xrpc/space.polymodel.actor.getSession")
                .body(Body::empty())
                .unwrap(),
        )
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::OK);
    let body = body_text(resp).await;
    assert!(body.contains("\"authenticated\":false"), "body was: {body}");
}

#[tokio::test]
async fn get_thing_route_decodes_query_and_404s_when_missing() {
    let state = state().await;
    let app = super::router().with_state(state);
    let resp = app
        .oneshot(
            Request::builder()
                .uri("/xrpc/space.polymodel.library.getThing?uri=at%3A%2F%2Fdid%3Aplc%3Ax%2Fspace.polymodel.library.thing%2Fnone")
                .body(Body::empty())
                .unwrap(),
        )
        .await
        .unwrap();
    assert_eq!(resp.status(), StatusCode::NOT_FOUND);
}

#[tokio::test]
async fn negative_offset_cursor_is_rejected() {
    let state = state().await;
    let errs = [
        views::search_things(&state, "x", 10, Some("-1"), None)
            .await
            .unwrap_err(),
        views::feed_hot(&state, 10, Some("-5"), None)
            .await
            .unwrap_err(),
        views::list_things(
            &state,
            "at://did:plc:l/space.polymodel.graph.list/self",
            10,
            Some("-1"),
            None,
        )
        .await
        .unwrap_err(),
    ];
    for err in errs {
        assert!(
            matches!(err, AppError::InvalidRequest(_)),
            "negative offset cursor must be InvalidRequest"
        );
    }
}

#[tokio::test]
async fn feed_recent_paginates_by_rkey() {
    let state = state().await;
    seed_identity(&state.pool, DID_A, "alice.com").await;
    seed_thing(&state.pool, DID_A, "3zzzzzzzzzzzz", "t3", &[], 0).await;
    seed_thing(&state.pool, DID_A, "3yyyyyyyyyyyy", "t2", &[], 0).await;
    seed_thing(&state.pool, DID_A, "3xxxxxxxxxxxx", "t1", &[], 0).await;

    let page1 = views::feed_recent(&state, 2, None, None).await.unwrap();
    assert_eq!(page1.items.len(), 2);
    assert_eq!(page1.items[0].thing.name.as_str(), "t3");
    let cursor = page1.cursor.expect("more remain");
    let page2 = views::feed_recent(&state, 2, Some(cursor.as_ref()), None)
        .await
        .unwrap();
    assert_eq!(page2.items.len(), 1);
    assert_eq!(page2.items[0].thing.name.as_str(), "t1");
    assert!(page2.cursor.is_none());
}
