Something went wrong. Try again.
atproto Thingiverse but good
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449use super::test_support::*;use crate::indexing::projection::{ ProjectionEvent, ProjectionInput, project_eager_record, project_event,};use jacquard_common::deps::bytes::Bytes;use jacquard_common::error::{AuthError, ClientError};use jacquard_common::types::value::{Data, to_data};use jacquard_common::xrpc::Response as XrpcResponse;use polymodel_api::com_atproto::repo::get_record::GetRecordResponse;
#[test]fn parse_record_ref_accepts_valid_create_response() { let did = did(DID_A); let collection = jacquard_common::types::nsid::Nsid::new_owned("space.polymodel.library.thing").unwrap(); let output = format!( r#"{{"uri":"at://{DID_A}/space.polymodel.library.thing/t1","cid":"bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"}}"# ); let (rkey, cid) = super::proxy::parse_record_ref(output.as_bytes(), &did, &collection) .expect("valid typed response"); assert_eq!(rkey.as_str(), "t1"); assert!(cid.is_valid());}
#[test]fn parse_record_ref_rejects_malformed_json() { let did = did(DID_A); let collection = jacquard_common::types::nsid::Nsid::new_owned("space.polymodel.library.thing").unwrap(); assert!(super::proxy::parse_record_ref(b"not json", &did, &collection).is_none());}
#[test]fn parse_record_ref_rejects_missing_fields() { let did = did(DID_A); let collection = jacquard_common::types::nsid::Nsid::new_owned("space.polymodel.library.thing").unwrap(); assert!( super::proxy::parse_record_ref( br#"{"uri":"at://did:plc:aaaaaaaaaaaaaaaaaaaaaaaa/space.polymodel.library.thing/t1"}"#, &did, &collection, ) .is_none() );}
#[test]fn parse_record_ref_rejects_trailing_slash() { let did = did(DID_A); let collection = jacquard_common::types::nsid::Nsid::new_owned("space.polymodel.library.thing").unwrap(); assert!(super::proxy::parse_record_ref( br#"{"uri":"at://did:plc:aaaaaaaaaaaaaaaaaaaaaaaa/space.polymodel.library.thing/","cid":"bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"}"#, &did, &collection, ).is_none());}
#[test]fn parse_record_ref_rejects_mismatched_authority() { let did = did(DID_A); let collection = jacquard_common::types::nsid::Nsid::new_owned("space.polymodel.library.thing").unwrap(); assert!(super::proxy::parse_record_ref( br#"{"uri":"at://did:plc:bbbbbbbbbbbbbbbbbbbbbbbb/space.polymodel.library.thing/t1","cid":"bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"}"#, &did, &collection, ).is_none());}
#[test]fn parse_record_ref_rejects_mismatched_collection() { let did = did(DID_A); let collection = jacquard_common::types::nsid::Nsid::new_owned("space.polymodel.library.thing").unwrap(); assert!(super::proxy::parse_record_ref( br#"{"uri":"at://did:plc:aaaaaaaaaaaaaaaaaaaaaaaa/space.polymodel.library.model/t1","cid":"bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"}"#, &did, &collection, ).is_none());}
#[test]fn parse_record_ref_rejects_empty_rkey() { let did = did(DID_A); let collection = jacquard_common::types::nsid::Nsid::new_owned("space.polymodel.library.thing").unwrap(); assert!(super::proxy::parse_record_ref( br#"{"uri":"at://did:plc:aaaaaaaaaaaaaaaaaaaaaaaa/space.polymodel.library.thing","cid":"bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"}"#, &did, &collection, ).is_none());}
#[test]fn parse_record_ref_rejects_invalid_cid() { let did = did(DID_A); let collection = jacquard_common::types::nsid::Nsid::new_owned("space.polymodel.library.thing").unwrap(); assert!(super::proxy::parse_record_ref( br#"{"uri":"at://did:plc:aaaaaaaaaaaaaaaaaaaaaaaa/space.polymodel.library.thing/t1","cid":"not-a-cid"}"#, &did, &collection, ).is_none());}
#[test]fn parse_record_cid_accepts_valid_get_record_response() { let cid = super::proxy::parse_record_cid( br#"{"cid":"bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku"}"#, ) .expect("valid cid"); assert!(cid.is_valid());}
#[test]fn parse_record_cid_rejects_malformed_json() { assert!(super::proxy::parse_record_cid(b"not json").is_none());}
#[test]fn parse_record_cid_rejects_missing_cid() { assert!(super::proxy::parse_record_cid(br#"{"uri":"at://did:plc:a"}"#).is_none());}
#[test]fn parse_record_cid_rejects_invalid_cid() { assert!(super::proxy::parse_record_cid(br#"{"cid":"not-a-cid"}"#).is_none());}
#[tokio::test]async fn forward_xrpc_preserves_success_body_and_json_content_type() { let body = b"{\"uri\":\"at://x\"}"; let resp = super::proxy::forward_xrpc::<GetRecordResponse>(Ok(XrpcResponse::new( Bytes::from_static(body), StatusCode::OK, ))); assert_eq!(resp.status(), StatusCode::OK); assert_eq!(resp.headers()["content-type"], "application/json"); assert_eq!(body_text(resp).await, std::str::from_utf8(body).unwrap());}
#[tokio::test]async fn forward_xrpc_preserves_error_status_body_and_json_content_type() { let body = b"{\"error\":\"InvalidSwap\"}"; let resp = super::proxy::forward_xrpc::<GetRecordResponse>(Err(ClientError::http( StatusCode::CONFLICT, Some(Bytes::from_static(body)), ))); assert_eq!(resp.status(), StatusCode::CONFLICT); assert_eq!(resp.headers()["content-type"], "application/json"); assert_eq!(body_text(resp).await, std::str::from_utf8(body).unwrap());}
// ---------------------------------------------------------------------------// PM-47 passthrough proxy: faithful forwarding, read-your-writes override,// eager projection, structural allowlist, and session gating.// ---------------------------------------------------------------------------
#[tokio::test]async fn forward_xrpc_passes_through_success_body() { let resp = crate::appview::proxy::forward_xrpc::<GetRecordResponse>(Ok(XrpcResponse::new( Bytes::from_static(b"{\"uri\":\"at://x\"}"), StatusCode::OK, ))); assert_eq!(resp.status(), StatusCode::OK); assert_eq!(body_text(resp).await, "{\"uri\":\"at://x\"}");}
#[tokio::test]async fn forward_xrpc_forwards_400_xrpc_error_verbatim() { // atproto getRecord RecordNotFound is HTTP 400 + an error body (route c: an // Ok response with a non-success status). The proxy must NOT turn it into 500. let resp = crate::appview::proxy::forward_xrpc::<GetRecordResponse>(Ok(XrpcResponse::new( Bytes::from_static(b"{\"error\":\"RecordNotFound\",\"message\":\"Record not found\"}"), StatusCode::BAD_REQUEST, ))); assert_eq!(resp.status(), StatusCode::BAD_REQUEST); assert!(body_text(resp).await.contains("RecordNotFound"));}
#[tokio::test]async fn forward_xrpc_forwards_http_error_status_and_body() { // route b: Err(Http) for a non-2xx that isn't 400/401 (e.g. 409 InvalidSwap). let err = ClientError::http( StatusCode::CONFLICT, Some(Bytes::from_static(b"{\"error\":\"InvalidSwap\"}")), ); let resp = crate::appview::proxy::forward_xrpc::<GetRecordResponse>(Err(err)); assert_eq!(resp.status(), StatusCode::CONFLICT); assert!(body_text(resp).await.contains("InvalidSwap"));}
#[tokio::test]async fn forward_xrpc_forwards_429_and_500_verbatim() { for status in [ StatusCode::TOO_MANY_REQUESTS, StatusCode::INTERNAL_SERVER_ERROR, ] { let err = ClientError::http(status, Some(Bytes::from_static(b"{}"))); let resp = crate::appview::proxy::forward_xrpc::<GetRecordResponse>(Err(err)); assert_eq!(resp.status(), status); }}
#[tokio::test]async fn forward_xrpc_auth_error_is_401() { let resp = crate::appview::proxy::forward_xrpc::<GetRecordResponse>(Err(ClientError::auth( AuthError::NotAuthenticated, ))); assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);}
#[tokio::test]async fn forward_xrpc_transport_error_is_500() { // Genuine transport/internal failures are the only thing that becomes 500. let err = ClientError::transport(std::io::Error::other("boom")); let resp = crate::appview::proxy::forward_xrpc::<GetRecordResponse>(Err(err)); assert_eq!(resp.status(), StatusCode::INTERNAL_SERVER_ERROR);}
#[tokio::test]async fn override_returns_local_for_pending_write_with_divergent_cid() { let pool = pool().await; let uri = format!("at://{DID_A}/space.polymodel.library.thing/t1"); sqlx::query("INSERT INTO pending_writes (uri, cid, value, written_at) VALUES (?, ?, ?, ?)") .bind(&uri) .bind("LOCALCID") .bind(json!({"name": "fresh"}).to_string()) .bind(1i64) .execute(&pool) .await .unwrap(); let resp = crate::appview::proxy::try_get_record_override(&pool, &uri, Some("PDSCID")) .await .unwrap() .expect("divergent pending write overrides the PDS result"); assert_eq!(resp.status(), StatusCode::OK); let body = body_text(resp).await; assert!(body.contains("LOCALCID"), "{body}"); assert!(body.contains("fresh"), "{body}"); assert!(!body.contains("stale"), "{body}");}
#[tokio::test]async fn override_returns_not_found_for_pending_delete() { let pool = pool().await; let uri = format!("at://{DID_A}/space.polymodel.library.thing/t1"); sqlx::query("INSERT INTO pending_deletes (uri, deleted_at) VALUES (?, ?)") .bind(&uri) .bind(1i64) .execute(&pool) .await .unwrap(); let resp = crate::appview::proxy::try_get_record_override(&pool, &uri, Some("PDSCID")) .await .unwrap() .expect("pending delete overrides the PDS result"); assert_eq!(resp.status(), StatusCode::BAD_REQUEST); assert!(body_text(resp).await.contains("RecordNotFound"));}
#[tokio::test]async fn override_is_none_when_no_pending_state() { let pool = pool().await; let uri = format!("at://{DID_A}/space.polymodel.library.thing/t1"); assert!( crate::appview::proxy::try_get_record_override(&pool, &uri, Some("PDSCID")) .await .unwrap() .is_none() );}
#[tokio::test]async fn override_is_none_when_pending_write_cid_matches_pds() { let pool = pool().await; let uri = format!("at://{DID_A}/space.polymodel.library.thing/t1"); sqlx::query("INSERT INTO pending_writes (uri, cid, value, written_at) VALUES (?, ?, ?, ?)") .bind(&uri) .bind("SAMECID") .bind("{}") .bind(1i64) .execute(&pool) .await .unwrap(); // Cids match → the firehose has effectively confirmed → forward the PDS. assert!( crate::appview::proxy::try_get_record_override(&pool, &uri, Some("SAMECID")) .await .unwrap() .is_none() );}
#[tokio::test]async fn override_returns_local_when_pds_reports_not_found() { // A pending write must win even when a lagging PDS reports not-found // (pds_cid is None) rather than a divergent record — the core // read-your-writes case. let pool = pool().await; let uri = format!("at://{DID_A}/space.polymodel.library.thing/t1"); sqlx::query("INSERT INTO pending_writes (uri, cid, value, written_at) VALUES (?, ?, ?, ?)") .bind(&uri) .bind("LOCALCID") .bind(json!({"name": "fresh"}).to_string()) .bind(1i64) .execute(&pool) .await .unwrap(); let resp = crate::appview::proxy::try_get_record_override(&pool, &uri, None) .await .unwrap() .expect("pending write wins over a lagging not-found"); assert_eq!(resp.status(), StatusCode::OK); let body = body_text(resp).await; assert!(body.contains("LOCALCID"), "{body}"); assert!(body.contains("fresh"), "{body}");}
fn poly_event(action: &str, record: Option<Data>) -> ProjectionEvent { ProjectionEvent { seq: 1, did: DID_A.to_string(), collection: "space.polymodel.library.thing".to_string(), rkey: "t1".to_string(), action: action.to_string(), record, cid: Some("C1".to_string()), }}
#[tokio::test]async fn eager_write_marks_pending_and_firehose_clears_it() { let pool = pool().await; let uri = format!("at://{DID_A}/space.polymodel.library.thing/t1"); let record = to_data( &json!({"name": "x", "license": "CC-BY-4.0", "createdAt": "2024-01-01T00:00:00.000Z"}), ) .unwrap(); project_eager_record(&pool, &poly_event("create", Some(record.clone()))) .await .unwrap(); let pending_cid: String = sqlx::query_scalar("SELECT cid FROM pending_writes WHERE uri = ?") .bind(&uri) .fetch_one(&pool) .await .unwrap(); assert_eq!(pending_cid, "C1");
// The Hydrant #commit re-delivering the same create clears the pending row. project_event( &pool, &ProjectionInput::Record(poly_event("create", Some(record))), ) .await .unwrap(); let remaining: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM pending_writes WHERE uri = ?") .bind(&uri) .fetch_one(&pool) .await .unwrap(); assert_eq!(remaining, 0);}
#[tokio::test]async fn eager_delete_marks_pending_and_firehose_clears_it() { let pool = pool().await; let uri = format!("at://{DID_A}/space.polymodel.library.thing/t1"); project_eager_record(&pool, &poly_event("delete", None)) .await .unwrap(); let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM pending_deletes WHERE uri = ?") .bind(&uri) .fetch_one(&pool) .await .unwrap(); assert_eq!(n, 1);
project_event(&pool, &ProjectionInput::Record(poly_event("delete", None))) .await .unwrap(); let n: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM pending_deletes WHERE uri = ?") .bind(&uri) .fetch_one(&pool) .await .unwrap(); assert_eq!(n, 0);}
#[tokio::test]async fn proxy_structural_allowlist_404s_non_allowlisted_methods() { let state = state().await; let app = crate::appview::router().with_state(state); for path in [ "/xrpc/com.atproto.repo.applyWrites", "/xrpc/com.atproto.repo.importRepo", "/xrpc/com.atproto.repo.listMissingBlobs", "/xrpc/app.bsky.feed.getTimeline", "/xrpc/com.atproto.server.getSession", ] { let resp = app .clone() .oneshot(Request::builder().uri(path).body(Body::empty()).unwrap()) .await .unwrap(); assert_eq!( resp.status(), StatusCode::NOT_FOUND, "{path} must not be proxied" ); }}
#[tokio::test]async fn proxy_pds_routes_require_an_oauth_session() { let state = state().await; let app = crate::appview::router().with_state(state); // createRecord (procedure) with no session → 401. let resp = app .clone() .oneshot( Request::builder() .method("POST") .uri("/xrpc/com.atproto.repo.createRecord") .header("content-type", "application/json") .body(Body::from("{}")) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); // getRecord (query) with no session → 401. let resp = app .oneshot( Request::builder() .uri( "/xrpc/com.atproto.repo.getRecord?repo=did:plc:x\ &collection=space.polymodel.library.thing&rkey=t1", ) .body(Body::empty()) .unwrap(), ) .await .unwrap(); assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);}