//! Tests for the 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 dioxus::prelude::dioxus_fullstack::FullstackContext;
use jacquard::oauth::authstore::ClientAuthStore;
use jacquard::oauth::client::OAuthSession;
use jacquard::oauth::keyset::Keyset;
use jacquard::oauth::scopes::Scopes;
use jacquard::oauth::session::{ClientSessionData, DpopClientData};
use jacquard::oauth::types::{OAuthTokenType, TokenSet};
use jacquard_common::deps::fluent_uri::Uri;
use jacquard_common::deps::smol_str::SmolStr;
use jacquard_common::session::SessionKey;
use jacquard_common::types::datetime::Datetime;
use jacquard_common::types::string::Did;
use polymodel_api::space_polymodel::actor::ProfileViewUnion;
use polymodel_api::space_polymodel::actor::get_session::{GetSessionOutput, SessionView};
use serde_json::json;
use sqlx::SqlitePool;
use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
use std::collections::HashSet;
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()
}

fn client_session(account_did: &str, session_id: &str) -> ClientSessionData {
    let dpop_key = Keyset::generate_es256("test")
        .unwrap()
        .public_jwks()
        .keys
        .into_iter()
        .next()
        .unwrap()
        .key;
    ClientSessionData {
        account_did: did(account_did),
        session_id: SmolStr::new(session_id),
        host_url: Uri::parse("https://pds.example.com").unwrap().to_owned(),
        authserver_url: SmolStr::new("https://issuer.example.com"),
        authserver_token_endpoint: SmolStr::new("https://issuer.example.com/token"),
        authserver_revocation_endpoint: None,
        scopes: Scopes::empty(),
        dpop_data: DpopClientData {
            dpop_key,
            dpop_authserver_nonce: SmolStr::default(),
            dpop_host_nonce: SmolStr::default(),
        },
        token_set: TokenSet {
            iss: SmolStr::new("https://issuer.example.com"),
            sub: did(account_did),
            aud: SmolStr::new("https://pds.example.com"),
            scope: None,
            refresh_token: None,
            access_token: SmolStr::new("token"),
            token_type: OAuthTokenType::DPoP,
            expires_at: Some(Datetime::raw_str("2099-01-01T00:00:00.000Z")),
        },
        resolved_scopes: None,
    }
}

async fn seed_oauth_session(state: &AppState, account_did: &str, session_id: &str) -> SessionKey {
    let data = client_session(account_did, session_id);
    let key = SessionKey::new(data.account_did.clone(), data.session_id.clone());
    state
        .oauth
        .registry
        .store
        .upsert_session(data)
        .await
        .unwrap();
    key
}

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 set_thing_created_at(pool: &SqlitePool, uri: &str, created_at: i64) {
    sqlx::query("UPDATE things SET created_at = ? WHERE uri = ?")
        .bind(created_at)
        .bind(uri)
        .execute(pool)
        .await
        .unwrap();
}

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();
}

async fn seed_list(pool: &SqlitePool, did: &str, rkey: &str, name: &str) -> String {
    let uri = format!("at://{did}/space.polymodel.graph.list/{rkey}");
    let now = chrono::Utc::now()
        .timestamp_nanos_opt()
        .expect("current timestamp must fit in i64 nanoseconds");
    sqlx::query(
        r#"INSERT INTO lists (did, rkey, uri, cid, name, description, purpose, created_at, indexed_at, record_json)
           VALUES (?, ?, ?, ?, ?, 'Print farm staples', 'curation', ?, ?, ?)"#,
    )
    .bind(did)
    .bind(rkey)
    .bind(&uri)
    .bind("bafyreilistcid")
    .bind(name)
    .bind(now)
    .bind(now)
    .bind(json!({
        "name": name,
        "description": "Print farm staples",
        "purpose": "curation",
        "createdAt": "2024-01-01T00:00:00.000Z"
    }).to_string())
    .execute(pool)
    .await
    .unwrap();
    uri
}

#[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 feed_hot_cursor_pins_ranking_time_and_keyset_boundary() {
    let state = state().await;
    seed_identity(&state.pool, DID_A, "alice.com").await;
    let now_a = 1_700_000_000_000_000_000;
    let now_b = now_a + 90 * 86_400_000_000_000;
    let rows = [
        ("3aaaaaaaaaaaa", "fresh-low", 2, 1),
        ("3bbbbbbbbbbbb", "medium-high", 7, 12),
        ("3cccccccccccc", "older-high", 48, 100),
        ("3dddddddddddd", "newer-medium", 3, 4),
        ("3eeeeeeeeeeee", "old-low", 120, 2),
    ];
    for (rkey, name, age_hours, likes) in rows {
        let uri = seed_thing(&state.pool, DID_A, rkey, name, &[], likes).await;
        set_thing_created_at(&state.pool, &uri, now_a - age_hours * 3_600_000_000_000).await;
    }

    let expected = views::feed_hot_at(&state, 10, None, None, now_a)
        .await
        .unwrap()
        .items
        .into_iter()
        .map(|item| item.thing.uri)
        .collect::<Vec<_>>();
    assert_eq!(expected.len(), rows.len());

    let page1 = views::feed_hot_at(&state, 2, None, None, now_a)
        .await
        .unwrap();
    let cursor = page1
        .cursor
        .as_ref()
        .expect("first hot page should have a cursor");
    assert!(
        cursor.as_str().starts_with("hot:v1:"),
        "hot cursor must be versioned and opaque"
    );

    let page2 = views::feed_hot_at(&state, 10, Some(cursor.as_str()), None, now_b)
        .await
        .unwrap();
    let combined = page1
        .items
        .into_iter()
        .chain(page2.items)
        .map(|item| item.thing.uri)
        .collect::<Vec<_>>();
    let unique = combined.iter().collect::<HashSet<_>>();
    assert_eq!(
        unique.len(),
        combined.len(),
        "hot pagination duplicated an item"
    );
    assert_eq!(
        combined, expected,
        "hot pagination must keep the cursor-pinned order"
    );
}

#[tokio::test]
async fn feed_hot_cursor_boundary_handles_same_rkey_across_repos() {
    let state = state().await;
    seed_identity(&state.pool, DID_A, "alice.com").await;
    seed_identity(&state.pool, DID_B, "bob.com").await;
    let now = 1_700_000_000_000_000_000;
    let shared_rkey = "3sameeeeeeeee";
    let uri_a = seed_thing(&state.pool, DID_A, shared_rkey, "alice", &[], 5).await;
    let uri_b = seed_thing(&state.pool, DID_B, shared_rkey, "bob", &[], 5).await;
    set_thing_created_at(&state.pool, &uri_a, now - 86_400_000_000_000).await;
    set_thing_created_at(&state.pool, &uri_b, now - 86_400_000_000_000).await;

    let page1 = views::feed_hot_at(&state, 1, None, None, now)
        .await
        .unwrap();
    let cursor = page1.cursor.as_deref().expect("first row has another page");
    let page2 = views::feed_hot_at(&state, 1, Some(cursor), None, now)
        .await
        .unwrap();

    let uris = page1
        .items
        .into_iter()
        .chain(page2.items)
        .map(|item| item.thing.uri)
        .collect::<Vec<_>>();
    let unique = uris.iter().collect::<HashSet<_>>();
    assert_eq!(unique.len(), 2, "same-rkey tied rows must not duplicate");
    assert!(uris.iter().any(|uri| uri.as_ref() == uri_a));
    assert!(uris.iter().any(|uri| uri.as_ref() == uri_b));
}

#[tokio::test]
async fn feed_hot_zero_engagement_fallback_paginates_after_zero_boundary() {
    let state = state().await;
    seed_identity(&state.pool, DID_A, "alice.com").await;
    let now = 1_700_000_000_000_000_000;
    for rkey in [
        "3dddddddddddd",
        "3cccccccccccc",
        "3bbbbbbbbbbbb",
        "3aaaaaaaaaaaa",
    ] {
        let uri = seed_thing(&state.pool, DID_A, rkey, rkey, &[], 0).await;
        set_thing_created_at(&state.pool, &uri, now - 86_400_000_000_000).await;
    }

    let page1 = views::feed_hot_at(&state, 2, None, None, now)
        .await
        .unwrap();
    assert_eq!(page1.items.len(), 2);
    assert_eq!(page1.items[0].thing.name.as_str(), "3dddddddddddd");
    assert_eq!(page1.items[1].thing.name.as_str(), "3cccccccccccc");
    let cursor = page1
        .cursor
        .as_deref()
        .expect("zero-score first page has cursor");

    let page2 = views::feed_hot_at(&state, 2, Some(cursor), None, now)
        .await
        .unwrap();
    assert_eq!(page2.items.len(), 2);
    assert_eq!(page2.items[0].thing.name.as_str(), "3bbbbbbbbbbbb");
    assert_eq!(page2.items[1].thing.name.as_str(), "3aaaaaaaaaaaa");
    assert!(page2.cursor.is_none());
}

#[tokio::test]
async fn feed_hot_ranks_older_high_engagement_beyond_previous_candidate_cap() {
    let state = state().await;
    seed_identity(&state.pool, DID_A, "alice.com").await;
    let now = 1_700_000_000_000_000_000;
    let older = seed_thing(
        &state.pool,
        DID_A,
        "3aaaaaaaaaaaa",
        "older-popular",
        &[],
        1_000_000,
    )
    .await;
    set_thing_created_at(&state.pool, &older, now - 30 * 86_400_000_000_000).await;

    for i in 0..501 {
        let rkey = format!("3new{i:010}");
        let uri = seed_thing(&state.pool, DID_A, &rkey, &format!("new-{i}"), &[], 0).await;
        set_thing_created_at(&state.pool, &uri, now - 3_600_000_000_000 + i).await;
    }

    let feed = views::feed_hot_at(&state, 5, None, None, now)
        .await
        .unwrap();
    assert_eq!(feed.items[0].thing.uri.as_ref(), older);
    assert_eq!(feed.items[0].thing.name.as_str(), "older-popular");
}

#[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 search_things_offset_cursor_paginates() {
    let state = state().await;
    seed_identity(&state.pool, DID_A, "alice.com").await;
    seed_thing(&state.pool, DID_A, "r1", "Modular enclosure alpha", &[], 0).await;
    seed_thing(&state.pool, DID_A, "r2", "Modular enclosure beta", &[], 0).await;
    seed_thing(&state.pool, DID_A, "r3", "Modular enclosure gamma", &[], 0).await;

    let page1 = views::search_things(&state, "enclosure", 2, None, None)
        .await
        .unwrap();
    assert_eq!(page1.items.len(), 2);
    let cursor = page1.cursor.expect("first page should expose a cursor");

    let page2 = views::search_things(&state, "enclosure", 2, Some(cursor.as_ref()), None)
        .await
        .unwrap();
    assert_eq!(page2.items.len(), 1);
    assert!(page2.cursor.is_none());

    let combined = page1
        .items
        .into_iter()
        .chain(page2.items)
        .map(|item| item.thing.uri)
        .collect::<Vec<_>>();
    let unique = combined.iter().collect::<HashSet<_>>();
    assert_eq!(unique.len(), 3, "search pagination duplicated an item");
}

#[tokio::test]
async fn malformed_ranked_cursor_is_rejected() {
    let state = state().await;
    let search_err = views::search_things(&state, "x", 10, Some("not-a-number"), None)
        .await
        .unwrap_err();
    assert!(matches!(search_err, AppError::InvalidRequest(_)));
    let hot_err = views::feed_hot(&state, 10, Some("not-a-hot-cursor"), None)
        .await
        .unwrap_err();
    assert!(matches!(hot_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, cid, list_uri, subject_uri, created_at) VALUES (?, ?, ?, ?, ?, ?)")
        .bind(DID_A).bind("i1").bind("bafyreilistitemone").bind(list).bind(&thing).bind(1)
        .execute(&state.pool).await.unwrap();
    sqlx::query("INSERT INTO listitems (did, rkey, cid, list_uri, subject_uri, created_at) VALUES (?, ?, ?, ?, ?, ?)")
        .bind(DID_A).bind("i2").bind("bafyreilistitemtwo").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 list_pagination_filters_non_things_before_windowing() {
    let state = state().await;
    seed_identity(&state.pool, DID_A, "alice.com").await;
    let t1 = seed_thing(&state.pool, DID_A, "t1", "Thing one", &[], 0).await;
    let t2 = seed_thing(&state.pool, DID_A, "t2", "Thing two", &[], 0).await;
    let list = "at://did:plc:l/space.polymodel.graph.list/mixed";
    let non_thing = "at://did:plc:x/space.polymodel.graph.like/k1";
    for (rkey, subject, created_at) in [
        ("i1", non_thing, 1_i64),
        ("i2", t1.as_str(), 2_i64),
        ("i3", non_thing, 3_i64),
        ("i4", t2.as_str(), 4_i64),
    ] {
        sqlx::query("INSERT INTO listitems (did, rkey, cid, list_uri, subject_uri, created_at) VALUES (?, ?, ?, ?, ?, ?)")
            .bind(DID_A)
            .bind(rkey)
            .bind(format!("cid-{rkey}"))
            .bind(list)
            .bind(subject)
            .bind(created_at)
            .execute(&state.pool)
            .await
            .unwrap();
    }

    let (page1, cursor) = views::list_things(&state, list, 1, None, None)
        .await
        .unwrap();
    assert_eq!(page1.len(), 1);
    assert_eq!(page1[0].uri.as_ref(), t1);
    let cursor = cursor.expect("second thing should remain");

    let (page2, cursor) = views::list_things(&state, list, 1, Some(cursor.as_ref()), None)
        .await
        .unwrap();
    assert_eq!(page2.len(), 1);
    assert_eq!(page2[0].uri.as_ref(), t2);
    assert!(cursor.is_none());
}

#[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_variant_carries_lists() {
    let state = state().await;
    seed_identity(&state.pool, DID_A, "alice.com").await;
    seed_profile(&state.pool, DID_A, "Alice").await;
    let list = seed_list(&state.pool, DID_A, "fixtures", "Fixture jigs").await;
    let thing = seed_thing(&state.pool, DID_A, "t1", "Thing One", &[], 0).await;
    sqlx::query("INSERT INTO listitems (did, rkey, cid, list_uri, subject_uri, created_at) VALUES (?, ?, ?, ?, ?, ?)")
        .bind(DID_A)
        .bind("li1")
        .bind("bafyreilistitem")
        .bind(&list)
        .bind(&thing)
        .bind(1_i64)
        .execute(&state.pool)
        .await
        .unwrap();

    let view = views::polymodel_profile_view(&state, &did(DID_A))
        .await
        .unwrap()
        .expect("polymodel profile exists");
    let lists = view.lists.expect("profile lists should be populated");
    assert_eq!(lists.len(), 1);
    assert_eq!(lists[0].uri.as_ref(), list);
    assert_eq!(lists[0].name.as_str(), "Fixture jigs");
    assert_eq!(
        lists[0].description.as_ref().map(|s| s.as_str()),
        Some("Print farm staples")
    );
    assert_eq!(
        lists[0].purpose.as_ref().map(|s| s.as_str()),
        Some("curation")
    );
    assert_eq!(lists[0].item_count, 1);
}

#[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_session_in_process_helper_returns_authenticated_viewer() {
    let state = state().await;
    seed_identity(&state.pool, DID_A, "alice.com").await;
    seed_profile(&state.pool, DID_A, "Alice").await;
    let session_data = client_session(DID_A, "in-process-session");
    let session = OAuthSession::new(
        state.oauth.registry.clone(),
        state.oauth.client.clone(),
        session_data,
    );

    let output = super::actor::get_session_output(&state, Some(&session))
        .await
        .unwrap();

    match output.value {
        SessionView::SessionAuthenticated(auth) => {
            assert!(auth.authenticated);
            assert_eq!(auth.did, did(DID_A));
            assert_eq!(auth.profile.display_name.as_deref(), Some("Alice"));
        }
        other => panic!("expected authenticated session, got {other:?}"),
    }
}

#[tokio::test]
async fn get_session_route_accepts_x_jacquard_session_header() {
    let state = state().await;
    seed_identity(&state.pool, DID_A, "alice.com").await;
    seed_profile(&state.pool, DID_A, "Alice").await;
    let key = seed_oauth_session(&state, DID_A, "header-session").await;
    let encoded = jacquard_axum::oauth::encode_session_key(&key).unwrap();
    let header_name = state.oauth_config.session_header.clone();
    let app = super::router().with_state(state);

    let resp = app
        .oneshot(
            Request::builder()
                .uri("/xrpc/space.polymodel.actor.getSession")
                .header(header_name, encoded)
                .body(Body::empty())
                .unwrap(),
        )
        .await
        .unwrap();

    assert_eq!(resp.status(), StatusCode::OK);
    let body = body_text(resp).await;
    let output: GetSessionOutput = serde_json::from_str(&body).unwrap();
    match output.value {
        SessionView::SessionAuthenticated(auth) => {
            assert!(auth.authenticated);
            assert_eq!(auth.did, did(DID_A));
        }
        other => panic!("expected authenticated session from header, got {other:?}"),
    }
}

#[tokio::test]
async fn fullstack_context_bridge_returns_authenticated_session_from_header() {
    let state = state().await;
    seed_identity(&state.pool, DID_A, "alice.com").await;
    seed_profile(&state.pool, DID_A, "Alice").await;
    let key = seed_oauth_session(&state, DID_A, "fullstack-header-session").await;
    let encoded = jacquard_axum::oauth::encode_session_key(&key).unwrap();
    let header_name = state.oauth_config.session_header.clone();

    let mut parts = Request::builder()
        .uri("/")
        .header(header_name, encoded)
        .body(())
        .unwrap()
        .into_parts()
        .0;
    parts.extensions.insert(state);

    let output = FullstackContext::new(parts)
        .scope(super::ssr::get_session_output_from_fullstack_context())
        .await
        .unwrap();

    match output.value {
        SessionView::SessionAuthenticated(auth) => {
            assert!(auth.authenticated);
            assert_eq!(auth.did, did(DID_A));
        }
        other => panic!("expected authenticated session from fullstack bridge, got {other:?}"),
    }
}

#[tokio::test]
async fn fullstack_context_feed_helper_supports_following() {
    let state = state().await;
    seed_identity(&state.pool, DID_A, "alice.com").await;
    seed_identity(&state.pool, DID_B, "bob.com").await;
    seed_thing(
        &state.pool,
        DID_B,
        "3bbbbbbbbbbbb",
        "followed project",
        &[],
        0,
    )
    .await;
    sqlx::query(
        "INSERT INTO follows (follower_did, followed_did, rkey, cid, created_at) VALUES (?, ?, ?, ?, ?)",
    )
    .bind(DID_A)
    .bind(DID_B)
    .bind("followb")
    .bind("bafyreifollowb")
    .bind(1_i64)
    .execute(&state.pool)
    .await
    .unwrap();
    let key = seed_oauth_session(&state, DID_A, "fullstack-following-session").await;
    let encoded = jacquard_axum::oauth::encode_session_key(&key).unwrap();
    let header_name = state.oauth_config.session_header.clone();

    let mut parts = Request::builder()
        .uri("/")
        .header(header_name, encoded)
        .body(())
        .unwrap()
        .into_parts()
        .0;
    parts.extensions.insert(state);

    let feed = FullstackContext::new(parts)
        .scope(super::ssr::get_feed_from_fullstack_context("following", 24))
        .await
        .unwrap();

    assert_eq!(feed.items.len(), 1);
    assert_eq!(feed.items[0].thing.name.as_str(), "followed project");
}

#[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 unauthenticated_write_routes_return_xrpc_401_not_redirects() {
    let state = state().await;
    let app = super::router().with_state(state);
    let routes = [
        (
            "/xrpc/space.polymodel.actor.bootstrapProfile",
            "application/json",
            "{}",
        ),
        (
            "/xrpc/space.polymodel.library.stageFile",
            "model/stl",
            "solid cube",
        ),
        (
            "/xrpc/space.polymodel.graph.createLike",
            "application/json",
            r#"{"subject":{"uri":"at://did:plc:aaaaaaaaaaaaaaaaaaaaaaaa/space.polymodel.library.thing/t1","cid":"bafycid"}}"#,
        ),
    ];

    for (path, content_type, body) in routes {
        let resp = app
            .clone()
            .oneshot(
                Request::builder()
                    .method("POST")
                    .uri(path)
                    .header("content-type", content_type)
                    .body(Body::from(body))
                    .unwrap(),
            )
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::UNAUTHORIZED, "route {path}");
        assert!(
            resp.headers().get("location").is_none(),
            "strict XRPC auth must not redirect for {path}"
        );
        let body = body_text(resp).await;
        assert!(body.contains("AuthenticationRequired"), "body was: {body}");
    }
}

#[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::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());
}

#[tokio::test]
async fn feed_following_cursor_handles_same_rkey_across_followed_repos() {
    let state = state().await;
    seed_identity(&state.pool, DID_A, "alice.com").await;
    seed_identity(&state.pool, DID_B, "bob.com").await;
    seed_identity(&state.pool, "did:plc:viewer", "viewer.com").await;
    seed_thing(&state.pool, DID_A, "3sameeeeeeeee", "Alice thing", &[], 0).await;
    seed_thing(&state.pool, DID_B, "3sameeeeeeeee", "Bob thing", &[], 0).await;
    sqlx::query("INSERT INTO follows (follower_did, followed_did, rkey, cid, created_at) VALUES (?, ?, ?, ?, ?)")
        .bind("did:plc:viewer")
        .bind(DID_A)
        .bind("fa")
        .bind("cid-a")
        .bind(1_i64)
        .execute(&state.pool)
        .await
        .unwrap();
    sqlx::query("INSERT INTO follows (follower_did, followed_did, rkey, cid, created_at) VALUES (?, ?, ?, ?, ?)")
        .bind("did:plc:viewer")
        .bind(DID_B)
        .bind("fb")
        .bind("cid-b")
        .bind(2_i64)
        .execute(&state.pool)
        .await
        .unwrap();

    let viewer = did("did:plc:viewer");
    let page1 = views::feed_following(&state, viewer.borrow(), 1, None)
        .await
        .unwrap();
    let cursor = page1.cursor.as_deref().expect("first row has another page");
    assert!(cursor.starts_with("following:v1:"));
    let page2 = views::feed_following(&state, viewer.borrow(), 1, Some(cursor))
        .await
        .unwrap();

    let names = page1
        .items
        .into_iter()
        .chain(page2.items.into_iter())
        .map(|item| item.thing.name.to_string())
        .collect::<std::collections::HashSet<_>>();
    assert_eq!(names.len(), 2);
    assert!(names.contains("Alice thing"));
    assert!(names.contains("Bob thing"));
}

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

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;

use crate::indexing::projection::{
    ProjectionEvent, ProjectionInput, project_eager_record, project_event,
};

#[tokio::test]
async fn forward_xrpc_passes_through_success_body() {
    let resp = super::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 = super::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 = super::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 = super::proxy::forward_xrpc::<GetRecordResponse>(Err(err));
        assert_eq!(resp.status(), status);
    }
}

#[tokio::test]
async fn forward_xrpc_auth_error_is_401() {
    let resp = super::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 = super::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 = super::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 = super::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!(
        super::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!(
        super::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 = super::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 = super::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 = super::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);
}