use super::test_support::*;
use super::views;
use crate::indexing::projection::{ProjectionEvent, ProjectionInput, project_event};
use http_body_util::BodyExt;
use jacquard_common::types::value::{Data, to_data};
use sha2::{Digest, Sha256};
use std::collections::HashMap;

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

#[tokio::test]
async fn ordinary_projection_feeds_compound_resolver_and_route() {
    use crate::ldraw::resolve::SqliteResolver;
    use crate::ldraw::{CanonicalPath, LdrawResourceResolver};
    use base64::Engine as _;

    let bytes = b"0 FILE main.ldr\\n0 NOFILE\\n";
    let (cid, _) = test_blob(bytes);
    let digest = Sha256::digest(bytes);
    let project = format!("at://{DID_A}/space.polymodel.library.thing/ordinary");
    let model = format!("at://{DID_A}/space.polymodel.library.model/main");
    let part = format!("at://{DID_A}/space.polymodel.library.part/main");
    let pool = pool().await;
    let event = |seq: u64, collection: &str, rkey: &str, record: Option<Data>| ProjectionEvent {
        seq,
        did: DID_A.to_owned(),
        collection: collection.to_owned(),
        rkey: rkey.to_owned(),
        action: "create".to_owned(),
        record,
        cid: Some(format!("record-{seq}")),
    };

    project_event(
        &pool,
        &ProjectionInput::Record(event(
            1,
            "space.polymodel.library.thing",
            "ordinary",
            Some(
                to_data(&json!({
                    "name": "Ordinary compound",
                    "license": "MIT",
                    "createdAt": "2024-01-01T00:00:00.000Z",
                    "models": [{ "uri": model, "cid": "bafyreimodel" }]
                }))
                .unwrap(),
            ),
        )),
    )
    .await
    .unwrap();
    project_event(
        &pool,
        &ProjectionInput::Record(event(
            2,
            "space.polymodel.library.model",
            "main",
            Some(
                to_data(&json!({
                    "name": "Main",
                    "createdAt": "2024-01-01T00:00:00.000Z",
                    "parts": [{ "uri": part, "cid": "bafyreipart" }]
                }))
                .unwrap(),
            ),
        )),
    )
    .await
    .unwrap();
    project_event(
        &pool,
        &ProjectionInput::Record(event(
            3,
            "space.polymodel.library.part",
            "main",
            Some(to_data(&json!({
                "name": "main.ldr",
                "format": "LDraw",
                "createdAt": "2024-01-01T00:00:00.000Z",
                "file": {
                    "mimeType": "application/x-ldraw",
                    "size": bytes.len(),
                    "digest": { "$bytes": base64::engine::general_purpose::STANDARD.encode(digest) },
                    "chunks": [{
                        "blob": { "$type": "blob", "ref": { "$link": cid }, "mimeType": "application/x-ldraw", "size": bytes.len() },
                        "offset": 0,
                        "size": bytes.len()
                    }]
                }
            })).unwrap()),
        )),
    )
    .await
    .unwrap();

    let project_uri = jacquard_common::types::string::AtUri::new_owned(&project).unwrap();
    let resource = SqliteResolver::new(pool.clone())
        .resolve(
            &project_uri,
            &CanonicalPath::parse("models/main.ldr").unwrap(),
        )
        .await
        .unwrap();
    assert_eq!(resource.source.resource_uri.as_ref(), part);
    assert_eq!(resource.target.blob_cids, vec![cid.clone()]);

    let mut blobs = HashMap::new();
    blobs.insert(cid, bytes.to_vec());
    let (pds, server) = loopback_pds_with_blobs(blobs).await;
    let bootstrap = crate::oauth::bootstrap_oauth(pool.clone(), Some("http://localhost"))
        .expect("ephemeral OAuth bootstrap for tests");
    let projection = pool.clone();
    let state = AppState::new_with_resolver(pool, bootstrap, loopback_resolver(&pds));
    let app = crate::appview::router().with_state(state);
    let response = app
        .clone()
        .oneshot(
            Request::builder()
                .method(Method::GET)
                .uri(format!(
                    "/xrpc/space.polymodel.library.getModelLoadBundle?uri={}",
                    urlencoding::encode(&part)
                ))
                .body(Body::empty())
                .unwrap(),
        )
        .await
        .unwrap();
    assert_eq!(response.status(), StatusCode::OK);
    let bundle_bytes = axum::body::to_bytes(response.into_body(), 1 << 20)
        .await
        .unwrap();
    let bundle = polymodel_renderer_protocol::ModelLoadBundle::decode(&bundle_bytes).unwrap();
    assert_eq!(bundle.primary, "models/main.ldr");
    assert_eq!(bundle.resources.len(), 1);
    let bundled = &bundle.resources[0];
    assert_eq!(bundled.key, "models/main.ldr");
    assert_eq!(bundled.target_cids[0], resource.target.blob_cids[0]);
    assert_eq!(bundled.bytes.as_ref(), bytes.as_slice());

    let route = crate::appview::ldraw::encode_binding(
        &crate::appview::ldraw::binding_for(&project_uri, &resource).unwrap(),
    )
    .unwrap();
    let fetched = app
        .clone()
        .oneshot(Request::builder().uri(&route).body(Body::empty()).unwrap())
        .await
        .unwrap();
    assert_eq!(fetched.status(), StatusCode::OK);
    assert_eq!(
        fetched
            .into_body()
            .collect()
            .await
            .unwrap()
            .to_bytes()
            .as_ref(),
        bytes.as_slice()
    );

    let route_variant = |mutate: &dyn Fn(&mut serde_json::Value)| {
        let encoded = route.strip_prefix("/app/ldraw/resources/").unwrap();
        let (payload, _) = encoded.rsplit_once('.').unwrap();
        let bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD
            .decode(payload)
            .unwrap();
        let mut payload: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        mutate(&mut payload);
        let bytes = serde_json::to_vec(&payload).unwrap();
        let digest = Sha256::digest(&bytes);
        format!(
            "/app/ldraw/resources/{}.{}",
            base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(bytes),
            digest
                .iter()
                .map(|byte| format!("{byte:02x}"))
                .collect::<String>()
        )
    };
    let fetch_status = |route: String| {
        let app = app.clone();
        async move {
            app.oneshot(Request::builder().uri(route).body(Body::empty()).unwrap())
                .await
                .unwrap()
                .status()
        }
    };

    // A route whose MAC is changed must fail before any source lookup.
    let mut tampered = route.clone().into_bytes();
    let last = tampered.len() - 1;
    tampered[last] = if tampered[last] == b'0' { b'1' } else { b'0' };
    assert_eq!(
        fetch_status(String::from_utf8(tampered).unwrap()).await,
        StatusCode::NOT_FOUND
    );

    // Re-signing a descriptor does not grant it another project's membership.
    let project_b = format!("at://{DID_B}/space.polymodel.library.thing/other");
    assert_eq!(
        fetch_status(route_variant(
            &|payload| payload["project"] = json!(project_b)
        ))
        .await,
        StatusCode::NOT_FOUND
    );
    assert_eq!(
        fetch_status(route_variant(&|payload| payload["source"] = json!(
            "at://did:plc:aaaaaaaaaaaaaaaaaaaaaaaa/space.polymodel.library.part/other"
        )))
        .await,
        StatusCode::NOT_FOUND
    );
    assert_eq!(
        fetch_status(route_variant(
            &|payload| payload["path"] = json!("models/other.ldr")
        ))
        .await,
        StatusCode::NOT_FOUND
    );
    assert_eq!(
        fetch_status(route_variant(
            &|payload| payload["len"] = json!(bytes.len() + 1)
        ))
        .await,
        StatusCode::NOT_FOUND
    );
    assert_eq!(
        fetch_status(route_variant(
            &|payload| payload["cids"] = json!(["bafkreiinvalid"])
        ))
        .await,
        StatusCode::NOT_FOUND
    );
    assert_eq!(
        fetch_status(route_variant(
            &|payload| payload["sha256"] = json!(vec![7_u8; 32])
        ))
        .await,
        StatusCode::NOT_FOUND
    );

    // A source metadata update invalidates an otherwise valid old binding.
    sqlx::query("UPDATE ldraw_resources SET sha256 = ? WHERE resource_uri = ?")
        .bind(vec![9_u8; 32])
        .bind(&part)
        .execute(&projection)
        .await
        .unwrap();
    assert_eq!(fetch_status(route).await, StatusCode::NOT_FOUND);
    server.abort();
}

#[tokio::test]
async fn anonymous_compound_route_traverses_mpd_cycles_and_manifest_only_companions() {
    let project = format!("at://{DID_A}/space.polymodel.library.thing/compound");
    let root_uri = format!("at://{DID_A}/space.polymodel.library.part/root");
    let child_uri = format!("at://{DID_A}/space.polymodel.library.part/child");
    let companion_uri = format!("at://{DID_A}/space.polymodel.library.part/companion");
    let root_bytes = b"0 FILE root.ldr\n1 16 0 0 0 1 0 0 0 1 0 0 0 0 child.dat\n0 FILE child.dat\n1 16 0 0 0 1 0 0 0 1 0 0 0 0 root.ldr\n0 NOFILE\n";
    let child_bytes = b"1 16 0 0 0 1 0 0 0 1 0 0 0 0 models/root.ldr\n";
    let companion_bytes = b"0 FILE companion.dat\n0 NOFILE\n";
    let (root_cid, _) = test_blob(root_bytes);
    let (child_cid, _) = test_blob(child_bytes);
    let (companion_cid, _) = test_blob(companion_bytes);
    let mut blobs = HashMap::new();
    blobs.insert(root_cid.clone(), root_bytes.to_vec());
    blobs.insert(child_cid.clone(), child_bytes.to_vec());
    blobs.insert(companion_cid.clone(), companion_bytes.to_vec());
    let (pds, server) = loopback_pds_with_blobs(blobs).await;
    let pool = pool().await;
    seed_compound_resource(
        &pool,
        &project,
        &root_uri,
        "models",
        "models/root.ldr",
        root_bytes,
        &root_cid,
    )
    .await;
    seed_compound_resource(
        &pool,
        &project,
        &child_uri,
        "mpd",
        "mpd/child.dat",
        child_bytes,
        &child_cid,
    )
    .await;
    seed_compound_resource(
        &pool,
        &project,
        &companion_uri,
        "mpd",
        "mpd/companion.dat",
        companion_bytes,
        &companion_cid,
    )
    .await;
    seed_manifest_edge(
        &pool,
        ManifestEdge {
            source_uri: &root_uri,
            root: "mpd",
            path: "mpd/child.dat",
            target_uri: &child_uri,
            target_cid: &child_cid,
            bytes: child_bytes,
            ordinal: 0,
        },
    )
    .await;
    seed_manifest_edge(
        &pool,
        ManifestEdge {
            source_uri: &root_uri,
            root: "mpd",
            path: "mpd/companion.dat",
            target_uri: &companion_uri,
            target_cid: &companion_cid,
            bytes: companion_bytes,
            ordinal: 1,
        },
    )
    .await;
    seed_manifest_edge(
        &pool,
        ManifestEdge {
            source_uri: &child_uri,
            root: "models",
            path: "models/root.ldr",
            target_uri: &root_uri,
            target_cid: &root_cid,
            bytes: root_bytes,
            ordinal: 0,
        },
    )
    .await;
    let bootstrap = crate::oauth::bootstrap_oauth(pool.clone(), Some("http://localhost"))
        .expect("ephemeral OAuth bootstrap for tests");
    let state = AppState::new_with_resolver(pool, bootstrap, loopback_resolver(&pds));
    let app = crate::appview::router().with_state(state);
    let request = || {
        Request::builder()
            .method(Method::POST)
            .uri("/app/model-load-bundle")
            .header("content-type", "application/json")
            .body(Body::from(json!({"uri": root_uri}).to_string()))
            .unwrap()
    };
    let first = app.clone().oneshot(request()).await.unwrap();
    assert_eq!(first.status(), StatusCode::OK);
    let first = polymodel_renderer_protocol::ModelLoadBundle::decode(
        &axum::body::to_bytes(first.into_body(), 1 << 20)
            .await
            .unwrap(),
    )
    .unwrap();
    let second = app.oneshot(request()).await.unwrap();
    assert_eq!(second.status(), StatusCode::OK);
    let second = polymodel_renderer_protocol::ModelLoadBundle::decode(
        &axum::body::to_bytes(second.into_body(), 1 << 20)
            .await
            .unwrap(),
    )
    .unwrap();
    assert_eq!(first, second);
    let keys = first
        .resources
        .iter()
        .map(|resource| resource.key.as_str())
        .collect::<Vec<_>>();
    assert_eq!(
        keys,
        ["models/root.ldr", "mpd/child.dat", "mpd/companion.dat"]
    );
    server.abort();
}