Something went wrong. Try again.
atproto Thingiverse but good
Something went wrong. Try again.
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455//! 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);}