Something went wrong. Try again.
atproto Thingiverse but good
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394239523962397//! PM-28 authenticated write endpoints.//!//! These XRPC routes deliberately use strict OAuth extraction, not the browser//! redirect extractor. The write order is PDS-first: only after the authenticated//! actor's PDS accepts a record/blob operation do we eagerly project into SQLite//! for read-your-own-writes. Hydrant/firehose will later deliver the same event;//! projection reuse keeps convergence idempotent.
use axum::Json;use axum::body::Body;use axum::extract::State;use axum::http::HeaderMap;use bytes::Bytes;use http_body_util::BodyExt;use jacquard::client::{Agent, AgentSession, AgentSessionExt};use jacquard::identity::PublicResolver;use jacquard::oauth::client::OAuthSession;use jacquard_axum::oauth::ExtractOAuthSession;use jacquard_axum::{ExtractXrpc, XrpcResponse};use jacquard_common::deps::smol_str::SmolStr;use jacquard_common::types::blob::{BlobRef, MimeType};use jacquard_common::types::collection::Collection;use jacquard_common::types::collection::RecordError;use jacquard_common::types::ident::AtIdentifier;use jacquard_common::types::recordkey::{RecordKey, Rkey};use jacquard_common::types::string::{AtUri, Cid, Datetime, Did};use jacquard_common::types::value::to_data;use jacquard_common::xrpc::XrpcError;use jacquard_common::xrpc::atproto::CreateRecordError;use polymodel_api::app_bsky::actor::profile::Profile as BskyProfile;use polymodel_api::com_atproto::repo::get_record::GetRecordError;use polymodel_api::com_atproto::repo::strong_ref::StrongRef;use polymodel_api::space_polymodel::actor::ProfileView;use polymodel_api::space_polymodel::actor::bootstrap_profile::{ BootstrapProfileOutput, BootstrapProfileOutputStatus, BootstrapProfileRequest,};use polymodel_api::space_polymodel::actor::profile::Profile as PolymodelProfile;use polymodel_api::space_polymodel::graph::create_follow::{ CreateFollowOutput, CreateFollowRequest,};use polymodel_api::space_polymodel::graph::create_like::{CreateLikeOutput, CreateLikeRequest};use polymodel_api::space_polymodel::graph::create_save::{CreateSaveOutput, CreateSaveRequest};use polymodel_api::space_polymodel::graph::create_tag::{CreateTagOutput, CreateTagRequest};use polymodel_api::space_polymodel::graph::delete_follow::{ DeleteFollowOutput, DeleteFollowRequest,};use polymodel_api::space_polymodel::graph::delete_like::{DeleteLikeOutput, DeleteLikeRequest};use polymodel_api::space_polymodel::graph::delete_save::{DeleteSaveOutput, DeleteSaveRequest};use polymodel_api::space_polymodel::graph::delete_tag::{DeleteTagOutput, DeleteTagRequest};use polymodel_api::space_polymodel::graph::follow::Follow;use polymodel_api::space_polymodel::graph::like::Like;use polymodel_api::space_polymodel::graph::save::Save;use polymodel_api::space_polymodel::graph::tag::Tag;use polymodel_api::space_polymodel::library::delete_thing::{ DeleteThingOutput, DeleteThingRequest,};use polymodel_api::space_polymodel::library::publish_thing::{ ModelInput, ModelPublishResult, PublishThingOutput, PublishThingRequest, RecordRef, ThingInput,};use polymodel_api::space_polymodel::library::stage_file::{StageFileOutput, StageFileOutputStatus};use polymodel_api::space_polymodel::library::update_thing::{ ModelInput as UpdateModelInput, PartInput as UpdatePartInput, ThingInput as UpdateThingInput, UpdateThingOutput, UpdateThingRequest,};use polymodel_api::space_polymodel::library::{Chunk, File};use polymodel_api::space_polymodel::library::{model::Model, part::Part, thing::Thing};use serde::Serialize;use sha2::{Digest, Sha256};use std::collections::BTreeSet;
use super::content_type::content_type;use super::error::{AppError, AppResult, db, internal, invalid_request, unauthorized};use super::state::AppState;use super::views;use crate::indexing::projection::{ProjectionEvent, project_eager_record};use crate::oauth::SqliteAuthStore;
const PROFILE_COLLECTION: &str = "space.polymodel.actor.profile";const THING_COLLECTION: &str = "space.polymodel.library.thing";const MODEL_COLLECTION: &str = "space.polymodel.library.model";const PART_COLLECTION: &str = "space.polymodel.library.part";const FOLLOW_COLLECTION: &str = "space.polymodel.graph.follow";const LIKE_COLLECTION: &str = "space.polymodel.graph.like";const SAVE_COLLECTION: &str = "space.polymodel.graph.save";const TAG_COLLECTION: &str = "space.polymodel.graph.tag";const CHUNK_SIZE: usize = 10 * 1024 * 1024;const MAX_STAGE_FILE_SIZE: usize = 100 * 1024 * 1024;
pub(super) type UserAgent = Agent<OAuthSession<PublicResolver, SqliteAuthStore>>;
/// Strict OAuth session extractor specialized for this app's resolver + store./// Shared with the PM-43 draft handlers so they authenticate identically.pub(super) type ExtractSession = ExtractOAuthSession<PublicResolver, SqliteAuthStore>;
pub(super) async fn bootstrap_profile( State(state): State<AppState>, ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>,) -> AppResult<XrpcResponse<BootstrapProfileRequest>> { let agent = Agent::from(session); let actor = authenticated_did(&agent).await?; let outcome = ensure_polymodel_profile(&state, &agent, &actor, true).await?; Ok(XrpcResponse(BootstrapProfileOutput { cid: outcome.cid, profile: outcome.profile, status: outcome.status, uri: outcome.uri, extra_data: None, }))}
pub(super) async fn stage_file( State(state): State<AppState>, ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>, headers: HeaderMap, body: Body,) -> AppResult<Json<StageFileOutput>> { let agent = Agent::from(session); let actor = authenticated_did(&agent).await?; let _ = ensure_polymodel_profile(&state, &agent, &actor, false).await?;
let mime_type = content_type(&headers); let filename = filename_header(&headers); let staged = upload_streamed_file(&agent, body, mime_type.as_str()).await?;
let file = File { chunks: staged.chunks, digest: Some(Bytes::copy_from_slice(&staged.digest)), mime_type: SmolStr::from(mime_type.clone()), size: staged.size as i64, extra_data: None, }; let upload_id = format!("{}-{}", hex_bytes(&staged.digest[..16]), staged.size); persist_upload( &state, &actor, UploadPersistence { upload_id: &upload_id, digest: &staged.digest, mime_type: &mime_type, size: staged.size, filename: filename.as_deref(), file: &file, }, ) .await?;
Ok(Json(StageFileOutput { file, filename: filename.map(SmolStr::from), status: Some(StageFileOutputStatus::Uploaded), upload_id: SmolStr::from(upload_id), extra_data: None, }))}
pub(super) async fn publish_thing( State(state): State<AppState>, ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>, ExtractXrpc(req): ExtractXrpc<PublishThingRequest>,) -> AppResult<XrpcResponse<PublishThingRequest>> { let agent = Agent::from(session); let actor = authenticated_did(&agent).await?; let _ = ensure_polymodel_profile(&state, &agent, &actor, false).await?; let _write_guard = state.write_lock.lock().await; let out = publish_composition(&state, &agent, &actor, req.thing).await?; Ok(XrpcResponse(out))}
pub(super) async fn update_thing( State(state): State<AppState>, ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>, ExtractXrpc(req): ExtractXrpc<UpdateThingRequest>,) -> AppResult<XrpcResponse<UpdateThingRequest>> { let agent = Agent::from(session); let actor = authenticated_did(&agent).await?; let _ = ensure_polymodel_profile(&state, &agent, &actor, false).await?; let _write_guard = state.write_lock.lock().await; validate_owned_uri(&actor, &req.uri, THING_COLLECTION)?; let previous = load_owned_composition(&state, &actor, &req.uri).await?; let out = update_composition(&state, &agent, &actor, req.uri, req.thing, &previous).await?; Ok(XrpcResponse(UpdateThingOutput { models: out.models, thing: out.thing, extra_data: None, }))}
pub(super) async fn delete_thing( State(state): State<AppState>, ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>, ExtractXrpc(req): ExtractXrpc<DeleteThingRequest>,) -> AppResult<XrpcResponse<DeleteThingRequest>> { let agent = Agent::from(session); let actor = authenticated_did(&agent).await?; let _write_guard = state.write_lock.lock().await; validate_owned_uri(&actor, &req.uri, THING_COLLECTION)?; let existing = load_owned_composition(&state, &actor, &req.uri).await?; let mut deleted_uris = delete_composition_children(&state, &agent, &actor, &existing).await?; agent_delete_record::<Thing>(&agent, existing.thing_rkey.clone()).await?; eager_delete( &state, &actor, THING_COLLECTION, existing.thing_rkey.as_ref(), ) .await?; deleted_uris.insert(0, existing.thing_uri); Ok(XrpcResponse(DeleteThingOutput { deleted: true, deleted_uris: Some(deleted_uris), extra_data: None, }))}
pub(super) async fn create_like( State(state): State<AppState>, ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>, ExtractXrpc(req): ExtractXrpc<CreateLikeRequest>,) -> AppResult<XrpcResponse<CreateLikeRequest>> { let agent = Agent::from(session); let actor = authenticated_did(&agent).await?; let _ = ensure_polymodel_profile(&state, &agent, &actor, false).await?; validate_subject(&state, &req.subject.uri).await?; if let Some(existing) = lookup_existing_social(&state, &actor, LIKE_COLLECTION, req.subject.uri.as_ref()).await? { return Ok(XrpcResponse(CreateLikeOutput { cid: existing.cid, created: Some(false), uri: existing.uri, extra_data: None, })); } let record = Like { created_at: Datetime::now(), subject: req.subject, extra_data: None, }; let out = agent .create_record(record.clone(), None) .await .map_err(|e| agent_error("create like", e))?; let rkey = rkey_from_uri(&out.uri)?; eager_project_record( &state, &actor, LIKE_COLLECTION, rkey.as_ref(), Some(out.cid.clone()), &record, "create", ) .await?; Ok(XrpcResponse(CreateLikeOutput { cid: out.cid, created: Some(true), uri: out.uri, extra_data: None, }))}
pub(super) async fn create_follow( State(state): State<AppState>, ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>, ExtractXrpc(req): ExtractXrpc<CreateFollowRequest>,) -> AppResult<XrpcResponse<CreateFollowRequest>> { let agent = Agent::from(session); let actor = authenticated_did(&agent).await?; let _ = ensure_polymodel_profile(&state, &agent, &actor, false).await?; if actor == req.subject { return Err(invalid_request("cannot follow yourself")); } ensure_follow_subject(&state, &req.subject).await?; if let Some(existing) = lookup_existing_follow(&state, &actor, &req.subject).await? { return Ok(XrpcResponse(CreateFollowOutput { cid: existing.cid, created: Some(false), uri: existing.uri, extra_data: None, })); } let record = Follow { created_at: Datetime::now(), subject: req.subject, extra_data: None, }; let out = agent .create_record(record.clone(), None) .await .map_err(|e| agent_error("create follow", e))?; let rkey = rkey_from_uri(&out.uri)?; eager_project_record( &state, &actor, FOLLOW_COLLECTION, rkey.as_ref(), Some(out.cid.clone()), &record, "create", ) .await?; Ok(XrpcResponse(CreateFollowOutput { cid: out.cid, created: Some(true), uri: out.uri, extra_data: None, }))}
pub(super) async fn delete_follow( State(state): State<AppState>, ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>, ExtractXrpc(req): ExtractXrpc<DeleteFollowRequest>,) -> AppResult<XrpcResponse<DeleteFollowRequest>> { let agent = Agent::from(session); let actor = authenticated_did(&agent).await?; let uri = match (req.uri, req.subject) { (Some(uri), _) => uri, (None, Some(subject)) => lookup_follow_uri(&state, &actor, &subject).await?, (None, None) => return Err(invalid_request("deleteFollow requires uri or subject")), }; delete_social::<Follow>(&state, &agent, &actor, FOLLOW_COLLECTION, &uri).await?; Ok(XrpcResponse(DeleteFollowOutput { deleted: true, uri: Some(uri), extra_data: None, }))}
pub(super) async fn delete_like( State(state): State<AppState>, ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>, ExtractXrpc(req): ExtractXrpc<DeleteLikeRequest>,) -> AppResult<XrpcResponse<DeleteLikeRequest>> { let agent = Agent::from(session); let actor = authenticated_did(&agent).await?; let uri = match (req.uri, req.subject) { (Some(uri), _) => uri, (None, Some(subject)) => { lookup_social_uri(&state, &actor, LIKE_COLLECTION, subject.uri.as_ref()).await? } (None, None) => return Err(invalid_request("deleteLike requires uri or subject")), }; delete_social::<Like>(&state, &agent, &actor, LIKE_COLLECTION, &uri).await?; Ok(XrpcResponse(DeleteLikeOutput { deleted: true, uri: Some(uri), extra_data: None, }))}
pub(super) async fn create_save( State(state): State<AppState>, ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>, ExtractXrpc(req): ExtractXrpc<CreateSaveRequest>,) -> AppResult<XrpcResponse<CreateSaveRequest>> { let agent = Agent::from(session); let actor = authenticated_did(&agent).await?; let _ = ensure_polymodel_profile(&state, &agent, &actor, false).await?; validate_subject(&state, &req.subject.uri).await?; if let Some(existing) = lookup_existing_social(&state, &actor, SAVE_COLLECTION, req.subject.uri.as_ref()).await? { return Ok(XrpcResponse(CreateSaveOutput { cid: existing.cid, created: Some(false), uri: existing.uri, extra_data: None, })); } let record = Save { created_at: Datetime::now(), note: req.note, subject: req.subject, extra_data: None, }; let out = agent .create_record(record.clone(), None) .await .map_err(|e| agent_error("create save", e))?; let rkey = rkey_from_uri(&out.uri)?; eager_project_record( &state, &actor, SAVE_COLLECTION, rkey.as_ref(), Some(out.cid.clone()), &record, "create", ) .await?; Ok(XrpcResponse(CreateSaveOutput { cid: out.cid, created: Some(true), uri: out.uri, extra_data: None, }))}
pub(super) async fn delete_save( State(state): State<AppState>, ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>, ExtractXrpc(req): ExtractXrpc<DeleteSaveRequest>,) -> AppResult<XrpcResponse<DeleteSaveRequest>> { let agent = Agent::from(session); let actor = authenticated_did(&agent).await?; let uri = match (req.uri, req.subject) { (Some(uri), _) => uri, (None, Some(subject)) => { lookup_social_uri(&state, &actor, SAVE_COLLECTION, subject.uri.as_ref()).await? } (None, None) => return Err(invalid_request("deleteSave requires uri or subject")), }; delete_social::<Save>(&state, &agent, &actor, SAVE_COLLECTION, &uri).await?; Ok(XrpcResponse(DeleteSaveOutput { deleted: true, uri: Some(uri), extra_data: None, }))}
pub(super) async fn create_tag( State(state): State<AppState>, ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>, ExtractXrpc(req): ExtractXrpc<CreateTagRequest>,) -> AppResult<XrpcResponse<CreateTagRequest>> { let agent = Agent::from(session); let actor = authenticated_did(&agent).await?; let _ = ensure_polymodel_profile(&state, &agent, &actor, false).await?; validate_subject(&state, &req.subject.uri).await?; let record = Tag { created_at: Datetime::now(), subject: req.subject, tag: req.tag, extra_data: None, }; let out = agent .create_record(record.clone(), None) .await .map_err(|e| agent_error("create tag", e))?; let rkey = rkey_from_uri(&out.uri)?; eager_project_record( &state, &actor, TAG_COLLECTION, rkey.as_ref(), Some(out.cid.clone()), &record, "create", ) .await?; Ok(XrpcResponse(CreateTagOutput { cid: out.cid, created: Some(true), uri: out.uri, extra_data: None, }))}
pub(super) async fn delete_tag( State(state): State<AppState>, ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>, ExtractXrpc(req): ExtractXrpc<DeleteTagRequest>,) -> AppResult<XrpcResponse<DeleteTagRequest>> { let agent = Agent::from(session); let actor = authenticated_did(&agent).await?; delete_social::<Tag>(&state, &agent, &actor, TAG_COLLECTION, &req.uri).await?; Ok(XrpcResponse(DeleteTagOutput { deleted: true, uri: Some(req.uri), extra_data: None, }))}
pub(super) struct BootstrapOutcome { profile: ProfileView, status: BootstrapProfileOutputStatus, uri: Option<AtUri>, cid: Option<Cid>,}
pub(super) async fn authenticated_did(agent: &UserAgent) -> AppResult<Did> { agent .session_info() .await .map(|(did, _)| did) .ok_or_else(|| unauthorized("OAuth session is required"))}
/// Ensure the actor's handle is present in the `identities` projection, mirroring/// what a firehose `#identity` event would write. No-op when already projected;/// otherwise resolves the handle (local cache, then DID document) and upserts it.async fn ensure_actor_identity(state: &AppState, actor: &Did) -> AppResult<()> { if views::handle_for_did(state, actor).await?.is_some() { return Ok(()); } let handle = views::resolve_handle(state, actor).await?; let now = chrono::Utc::now().timestamp_nanos_opt().unwrap_or_default(); let actor_ref = actor.as_ref(); db(sqlx::query!( "INSERT INTO identities (did, handle, updated_at) VALUES (?, ?, ?) \ ON CONFLICT(did) DO UPDATE SET handle = excluded.handle, updated_at = excluded.updated_at", actor_ref, handle, now, ) .execute(&state.pool) .await)?; Ok(())}
pub(super) async fn ensure_polymodel_profile( state: &AppState, agent: &UserAgent, actor: &Did, allow_create: bool,) -> AppResult<BootstrapOutcome> { // Read-your-own-writes: record the actor's handle in the identity projection // so author hydration on an immediate read doesn't 500 (the firehose is // otherwise the only writer of `identities`, and it lags a fresh sign-in). ensure_actor_identity(state, actor).await?;
if let Some(profile) = views::polymodel_profile_view(state, actor).await? { return Ok(BootstrapOutcome { profile, status: BootstrapProfileOutputStatus::ExistingLocal, uri: None, cid: None, }); }
let profile_uri = at_uri_owned(actor, PROFILE_COLLECTION, "self")?; // A missing record surfaces as an XRPC `RecordNotFound` decoded by // `into_output()`, not as a transport-level `Err` — so a not-found must be // caught in both arms or it becomes a spurious 500 "decode failed". let existing_remote = match agent.get_record::<PolymodelProfile, _>(&profile_uri).await { Ok(resp) => match resp.into_output() { Ok(output) => Some(output), Err(e) if into_output_is_not_found(&e) => None, Err(e) => return Err(internal(format!("profile record decode failed: {e}"))), }, Err(e) if client_error_is_not_found(&e) => None, Err(e) => return Err(client_error("fetch polymodel profile", e)), }; if let Some(output) = existing_remote { eager_project_record( state, actor, PROFILE_COLLECTION, "self", output.cid.clone(), &output.value, "create", ) .await?; let profile = views::polymodel_profile_view(state, actor) .await? .ok_or_else(|| internal("projected profile was not readable"))?; return Ok(BootstrapOutcome { profile, status: BootstrapProfileOutputStatus::ExistingRemote, uri: Some(output.uri), cid: output.cid, }); }
if !allow_create { return Err(invalid_request( "Polymodel profile has not been bootstrapped", )); }
let bsky_uri = at_uri_owned(actor, "app.bsky.actor.profile", "self")?; let bsky = match agent.get_record::<BskyProfile, _>(&bsky_uri).await { Ok(resp) => match resp.into_output() { Ok(output) => output.value, Err(e) if into_output_is_not_found(&e) => empty_bsky_profile(), Err(e) => return Err(internal(format!("bsky profile decode failed: {e}"))), }, Err(e) if client_error_is_not_found(&e) => empty_bsky_profile(), Err(e) => return Err(client_error("fetch bsky profile", e)), };
let profile = PolymodelProfile { avatar: bsky.avatar, default_license: None, description: bsky.description, display_name: bsky.display_name, links: None, printers: None, pronouns: bsky.pronouns, extra_data: None, }; let create = agent .create_record(profile.clone(), Some(record_key("self")?)) .await; let output = match create { Ok(out) => { eager_project_record( state, actor, PROFILE_COLLECTION, "self", Some(out.cid.clone()), &profile, "create", ) .await?; BootstrapOutcome { profile: views::polymodel_profile_view(state, actor) .await? .ok_or_else(|| internal("created profile was not readable"))?, status: BootstrapProfileOutputStatus::Created, uri: Some(out.uri), cid: Some(out.cid), } } Err(e) if agent_error_is_conflict(&e) || agent_error_is_not_found(&e) => { let resp = agent .get_record::<PolymodelProfile, _>(&profile_uri) .await .map_err(|e| client_error("refetch raced profile", e))?; let remote = resp .into_output() .map_err(|e| internal(format!("raced profile decode failed: {e}")))?; eager_project_record( state, actor, PROFILE_COLLECTION, "self", remote.cid.clone(), &remote.value, "create", ) .await?; BootstrapOutcome { profile: views::polymodel_profile_view(state, actor) .await? .ok_or_else(|| internal("raced profile was not readable"))?, status: BootstrapProfileOutputStatus::ExistingRemote, uri: Some(remote.uri), cid: remote.cid, } } Err(e) => return Err(agent_error("create polymodel profile", e)), }; Ok(output)}
pub(super) async fn publish_composition( state: &AppState, agent: &UserAgent, actor: &Did, thing: ThingInput,) -> AppResult<PublishThingOutput> { let mut model_results = Vec::new(); let mut model_refs = Vec::new();
for model_input in thing.models { let (model_ref, result) = publish_model(state, agent, actor, model_input).await?; model_refs.push(model_ref); model_results.push(result); }
let record = Thing { cover: thing.cover, created_at: Datetime::now(), derived_from: thing.derived_from, instructions: thing.instructions, intended_to_work_with: thing.intended_to_work_with, license: thing.license, models: Some(model_refs), name: thing.name, previews: thing.previews, summary: thing.summary, tags: thing.tags, extra_data: None, };
let out = agent .create_record(record.clone(), None) .await .map_err(|e| agent_error("create thing", e))?; let rkey = rkey_from_uri(&out.uri)?; eager_project_record( state, actor, THING_COLLECTION, rkey.as_ref(), Some(out.cid.clone()), &record, "create", ) .await?; let created = RecordRef { cid: out.cid, uri: out.uri, extra_data: None, };
Ok(PublishThingOutput { thing: created, models: model_results, extra_data: None, })}
async fn update_composition( state: &AppState, agent: &UserAgent, actor: &Did, thing_uri: AtUri, thing: UpdateThingInput, previous: &ExistingComposition,) -> AppResult<PublishThingOutput> { let mut model_results = Vec::new(); let mut model_refs = Vec::new();
for model_input in thing.models { let (model_ref, result) = update_model(state, agent, actor, model_input, previous).await?; model_refs.push(model_ref); model_results.push(result); }
let record = Thing { cover: thing.cover, created_at: Datetime::now(), derived_from: thing.derived_from, instructions: thing.instructions, intended_to_work_with: thing.intended_to_work_with, license: thing.license, models: Some(model_refs), name: thing.name, previews: thing.previews, summary: thing.summary, tags: thing.tags, extra_data: None, };
let rkey = rkey_from_uri(&thing_uri)?; let out = agent .put_record::<Thing>(rkey.clone(), record.clone()) .await .map_err(|e| agent_error("put thing", e))?; eager_project_record( state, actor, THING_COLLECTION, rkey.as_ref(), Some(out.cid.clone()), &record, "update", ) .await?;
Ok(PublishThingOutput { thing: RecordRef { cid: out.cid, uri: out.uri, extra_data: None, }, models: model_results, extra_data: None, })}
async fn update_model( state: &AppState, agent: &UserAgent, actor: &Did, model_input: UpdateModelInput, previous: &ExistingComposition,) -> AppResult<(StrongRef, ModelPublishResult)> { let mut part_refs = Vec::new(); let mut part_record_refs = Vec::new(); for part_input in model_input.parts { let part_ref = update_part(state, agent, actor, part_input, previous).await?; part_record_refs.push(RecordRef { uri: part_ref.uri.clone(), cid: part_ref.cid.clone(), extra_data: None, }); part_refs.push(part_ref); }
let model = Model { cover: model_input.cover, created_at: Datetime::now(), derived_from: model_input.derived_from, instructions: model_input.instructions, intended_to_work_with: model_input.intended_to_work_with, license: model_input.license, name: model_input.name, parts: part_refs, previews: model_input.previews, summary: model_input.summary, tags: model_input.tags, extra_data: None, };
let (uri, cid, action) = if let Some(uri) = model_input.uri { validate_owned_uri(actor, &uri, MODEL_COLLECTION)?; let rkey = rkey_from_uri(&uri)?; validate_existing_update_child( state, actor, &uri, &rkey, MODEL_COLLECTION, &previous.models, ) .await?; let out = agent .put_record::<Model>(rkey.clone(), model.clone()) .await .map_err(|e| agent_error("put model", e))?; eager_project_record( state, actor, MODEL_COLLECTION, rkey.as_ref(), Some(out.cid.clone()), &model, "update", ) .await?; (out.uri, out.cid, "update") } else { let out = agent .create_record(model.clone(), None) .await .map_err(|e| agent_error("create model", e))?; let rkey = rkey_from_uri(&out.uri)?; eager_project_record( state, actor, MODEL_COLLECTION, rkey.as_ref(), Some(out.cid.clone()), &model, "create", ) .await?; (out.uri, out.cid, "create") }; tracing::debug!(action, uri = %uri, "updated thing model composition member"); let model_ref = strong_ref(uri.clone(), cid.clone()); Ok(( model_ref, ModelPublishResult { model: RecordRef { uri, cid, extra_data: None, }, parts: part_record_refs, extra_data: None, }, ))}
async fn update_part( state: &AppState, agent: &UserAgent, actor: &Did, input: UpdatePartInput, previous: &ExistingComposition,) -> AppResult<StrongRef> { let file = match (input.file, input.upload_id) { (Some(file), _) => file, (None, Some(upload_id)) => load_upload_file(state, actor, upload_id.as_ref()).await?, (None, None) => return Err(invalid_request("part requires file or uploadId")), }; let part = Part { created_at: Datetime::now(), derived_from: input.derived_from, dimensions: input.dimensions, file, format: input.format, intended_to_work_with: input.intended_to_work_with, name: input.name, notes: input.notes, previews: input.previews, print_settings: input.print_settings, units: input.units, extra_data: None, };
if let Some(uri) = input.uri { validate_owned_uri(actor, &uri, PART_COLLECTION)?; let rkey = rkey_from_uri(&uri)?; validate_existing_update_child(state, actor, &uri, &rkey, PART_COLLECTION, &previous.parts) .await?; let out = agent .put_record::<Part>(rkey.clone(), part.clone()) .await .map_err(|e| agent_error("put part", e))?; eager_project_record( state, actor, PART_COLLECTION, rkey.as_ref(), Some(out.cid.clone()), &part, "update", ) .await?; return Ok(strong_ref(out.uri, out.cid)); }
let out = agent .create_record(part.clone(), None) .await .map_err(|e| agent_error("create part", e))?; let rkey = rkey_from_uri(&out.uri)?; eager_project_record( state, actor, PART_COLLECTION, rkey.as_ref(), Some(out.cid.clone()), &part, "create", ) .await?; Ok(strong_ref(out.uri, out.cid))}
async fn publish_model( state: &AppState, agent: &UserAgent, actor: &Did, model_input: ModelInput,) -> AppResult<(StrongRef, ModelPublishResult)> { let mut part_refs = Vec::new(); let mut part_record_refs = Vec::new(); for part_input in model_input.parts { let part_ref = publish_part(state, agent, actor, part_input).await?; part_record_refs.push(RecordRef { uri: part_ref.uri.clone(), cid: part_ref.cid.clone(), extra_data: None, }); part_refs.push(part_ref); } let model = Model { cover: model_input.cover, created_at: Datetime::now(), derived_from: model_input.derived_from, instructions: model_input.instructions, intended_to_work_with: model_input.intended_to_work_with, license: model_input.license, name: model_input.name, parts: part_refs, previews: model_input.previews, summary: model_input.summary, tags: model_input.tags, extra_data: None, }; let out = agent .create_record(model.clone(), None) .await .map_err(|e| agent_error("create model", e))?; let rkey = rkey_from_uri(&out.uri)?; eager_project_record( state, actor, MODEL_COLLECTION, rkey.as_ref(), Some(out.cid.clone()), &model, "create", ) .await?; let model_ref = strong_ref(out.uri.clone(), out.cid.clone()); let record_ref = RecordRef { uri: out.uri, cid: out.cid, extra_data: None, }; Ok(( model_ref, ModelPublishResult { model: record_ref, parts: part_record_refs, extra_data: None, }, ))}
async fn publish_part( state: &AppState, agent: &UserAgent, actor: &Did, input: polymodel_api::space_polymodel::library::publish_thing::PartInput,) -> AppResult<StrongRef> { let file = match (input.file, input.upload_id) { (Some(file), _) => file, (None, Some(upload_id)) => load_upload_file(state, actor, upload_id.as_ref()).await?, (None, None) => return Err(invalid_request("part requires file or uploadId")), }; let part = Part { created_at: Datetime::now(), derived_from: input.derived_from, dimensions: input.dimensions, file, format: input.format, intended_to_work_with: input.intended_to_work_with, name: input.name, notes: input.notes, previews: input.previews, print_settings: input.print_settings, units: input.units, extra_data: None, }; let out = agent .create_record(part.clone(), None) .await .map_err(|e| agent_error("create part", e))?; let rkey = rkey_from_uri(&out.uri)?; eager_project_record( state, actor, PART_COLLECTION, rkey.as_ref(), Some(out.cid.clone()), &part, "create", ) .await?; Ok(strong_ref(out.uri, out.cid))}
async fn delete_social<R>( state: &AppState, agent: &UserAgent, actor: &Did, collection: &str, uri: &AtUri,) -> AppResult<()>where R: Collection + Serialize,{ validate_owned_uri(actor, uri, collection)?; let rkey = rkey_from_uri(uri)?; agent_delete_record::<R>(agent, rkey.clone()).await?; eager_delete(state, actor, collection, rkey.as_ref()).await}
struct ExistingComposition { thing_uri: AtUri, thing_rkey: RecordKey<Rkey>, models: Vec<CompositionChild>, parts: Vec<CompositionChild>,}
struct CompositionChild { rkey: RecordKey<Rkey>, uri: AtUri,}
#[derive(sqlx::FromRow)]struct CompositionRecordRow { did: String, rkey: String, uri: String,}
async fn load_owned_composition( state: &AppState, actor: &Did, thing_uri: &AtUri,) -> AppResult<ExistingComposition> { let thing = db(sqlx::query_as!( CompositionRecordRow, "SELECT did, rkey, uri FROM things WHERE uri = ?", thing_uri.as_ref() ) .fetch_optional(&state.pool) .await)? .ok_or_else(|| invalid_request("thing is not indexed locally"))?; ensure_row_owned(actor, &thing, THING_COLLECTION)?;
let models = db(sqlx::query_as!( CompositionRecordRow, "SELECT m.did, m.rkey, m.uri FROM thing_models tm \ JOIN models m ON m.uri = tm.model_uri \ WHERE tm.thing_uri = ? ORDER BY tm.position", thing_uri.as_ref() ) .fetch_all(&state.pool) .await)?; let parts = db(sqlx::query_as!( CompositionRecordRow, "SELECT p.did, p.rkey, p.uri FROM thing_models tm \ JOIN model_parts mp ON mp.model_uri = tm.model_uri \ JOIN parts p ON p.uri = mp.part_uri \ WHERE tm.thing_uri = ? ORDER BY tm.position, mp.position", thing_uri.as_ref() ) .fetch_all(&state.pool) .await)?;
let thing_uri = at_uri_from_row(&thing)?; let mut model_children = Vec::new(); let mut seen_models = BTreeSet::new(); for row in models { ensure_row_owned(actor, &row, MODEL_COLLECTION)?; if seen_models.insert(row.uri.clone()) { model_children.push(CompositionChild { rkey: record_key(&row.rkey)?, uri: at_uri_from_row(&row)?, }); } } let mut part_children = Vec::new(); let mut seen_parts = BTreeSet::new(); for row in parts { ensure_row_owned(actor, &row, PART_COLLECTION)?; if seen_parts.insert(row.uri.clone()) { part_children.push(CompositionChild { rkey: record_key(&row.rkey)?, uri: at_uri_from_row(&row)?, }); } }
Ok(ExistingComposition { thing_uri, thing_rkey: record_key(&thing.rkey)?, models: model_children, parts: part_children, })}
async fn delete_composition_children( state: &AppState, agent: &UserAgent, actor: &Did, composition: &ExistingComposition,) -> AppResult<Vec<AtUri>> { let mut deleted_uris = Vec::new(); for model in &composition.models { if model_reference_count(state, &model.uri).await? > 1 { continue; } agent_delete_record::<Model>(agent, model.rkey.clone()).await?; eager_delete(state, actor, MODEL_COLLECTION, model.rkey.as_ref()).await?; deleted_uris.push(model.uri.clone()); } for part in &composition.parts { if part_reference_count(state, &part.uri).await? > 0 { continue; } agent_delete_record::<Part>(agent, part.rkey.clone()).await?; eager_delete(state, actor, PART_COLLECTION, part.rkey.as_ref()).await?; deleted_uris.push(part.uri.clone()); } Ok(deleted_uris)}
async fn model_reference_count(state: &AppState, uri: &AtUri) -> AppResult<i64> { db(sqlx::query_scalar!( "SELECT COUNT(*) AS count FROM thing_models WHERE model_uri = ?", uri.as_ref() ) .fetch_one(&state.pool) .await)}
async fn part_reference_count(state: &AppState, uri: &AtUri) -> AppResult<i64> { db(sqlx::query_scalar!( "SELECT COUNT(*) AS count FROM model_parts WHERE part_uri = ?", uri.as_ref() ) .fetch_one(&state.pool) .await)}
fn ensure_row_owned(actor: &Did, row: &impl OwnedRow, collection: &str) -> AppResult<()> { if row.did() != actor.as_ref() { return Err(invalid_request(format!( "{collection} composition record is not owned by the authenticated actor" ))); } Ok(())}
trait OwnedRow { fn did(&self) -> &str; fn uri(&self) -> &str;}
impl OwnedRow for CompositionRecordRow { fn did(&self) -> &str { &self.did } fn uri(&self) -> &str { &self.uri }}
fn at_uri_from_row(row: &impl OwnedRow) -> AppResult<AtUri> { AtUri::new_owned(row.uri()).map_err(|e| internal(format!("stored at-uri is invalid: {e}")))}
fn strong_ref(uri: AtUri, cid: Cid) -> StrongRef { StrongRef { uri, cid, extra_data: None, }}
fn record_key(value: &str) -> AppResult<RecordKey<Rkey>> { RecordKey::any(SmolStr::from(value)) .map_err(|e| invalid_request(format!("invalid rkey {value:?}: {e}")))}
fn rkey_from_uri(uri: &AtUri) -> AppResult<RecordKey<Rkey>> { let rkey = uri .rkey() .ok_or_else(|| invalid_request("at-uri is missing record key"))?; record_key(rkey.as_ref())}
fn at_uri_owned(actor: &Did, collection: &str, rkey: &str) -> AppResult<AtUri> { AtUri::new_owned(format!("at://{}/{}/{}", actor.as_ref(), collection, rkey)) .map_err(|e| invalid_request(format!("invalid at-uri: {e}")))}
fn validate_owned_uri(actor: &Did, uri: &AtUri, expected_collection: &str) -> AppResult<()> { match uri.authority() { AtIdentifier::Did(did) if did.as_ref() == actor.as_ref() => {} _ => { return Err(invalid_request( "record URI is not owned by the authenticated actor", )); } } let collection = uri .collection() .ok_or_else(|| invalid_request("record URI missing collection"))?; if collection.as_ref() != expected_collection { return Err(invalid_request(format!( "expected {expected_collection} URI" ))); } if uri.rkey().is_none() { return Err(invalid_request("record URI missing rkey")); } Ok(())}
async fn validate_existing_update_child( state: &AppState, actor: &Did, uri: &AtUri, rkey: &RecordKey<Rkey>, collection: &str, current_children: &[CompositionChild],) -> AppResult<()> { if current_children.iter().any(|child| &child.rkey == rkey) { return Ok(()); } let exists = match collection { MODEL_COLLECTION => db(sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM models WHERE did = ? AND uri = ?)", actor.as_ref(), uri.as_ref() ) .fetch_one(&state.pool) .await)?, PART_COLLECTION => db(sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM parts WHERE did = ? AND uri = ?)", actor.as_ref(), uri.as_ref() ) .fetch_one(&state.pool) .await)?, _ => return Err(invalid_request("unsupported composition child collection")), }; if exists == 0 { return Err(invalid_request( "existing child uri is not indexed as an owned composition record", )); } Ok(())}
async fn validate_subject(state: &AppState, uri: &AtUri) -> AppResult<()> { let collection = uri .collection() .ok_or_else(|| invalid_request("subject URI missing collection"))?; let table = match collection.as_ref() { THING_COLLECTION => "things", MODEL_COLLECTION => "models", PART_COLLECTION => "parts", _ => { return Err(invalid_request( "subject must be a Polymodel thing, model, or part", )); } }; let exists = match table { "things" => db(sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM things WHERE uri = ?)", uri.as_ref() ) .fetch_one(&state.pool) .await)?, "models" => db(sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM models WHERE uri = ?)", uri.as_ref() ) .fetch_one(&state.pool) .await)?, "parts" => db(sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM parts WHERE uri = ?)", uri.as_ref() ) .fetch_one(&state.pool) .await)?, _ => unreachable!(), }; if exists == 0 { return Err(invalid_request("subject is not indexed locally")); } Ok(())}
async fn eager_project_record<R: Serialize>( state: &AppState, actor: &Did, collection: &str, rkey: &str, cid: Option<Cid>, record: &R, action: &str,) -> AppResult<()> { let record = to_data(record).map_err(|e| internal(format!("record serialize: {e}")))?; let event = ProjectionEvent { seq: 0, did: actor.as_ref().to_string(), collection: collection.to_string(), rkey: rkey.to_string(), action: action.to_string(), record: Some(record), cid: cid.map(|c| c.to_string()), }; project_eager_record(&state.pool, &event) .await .map_err(|e| internal(format!("eager projection failed: {e}")))}
async fn eager_delete( state: &AppState, actor: &Did, collection: &str, rkey: &str,) -> AppResult<()> { let event = ProjectionEvent { seq: 0, did: actor.as_ref().to_string(), collection: collection.to_string(), rkey: rkey.to_string(), action: "delete".to_string(), record: None, cid: None, }; project_eager_record(&state.pool, &event) .await .map_err(|e| internal(format!("eager delete projection failed: {e}")))}
struct StreamedUpload { chunks: Vec<Chunk>, digest: Vec<u8>, size: usize,}
async fn upload_streamed_file( agent: &UserAgent, mut body: Body, mime_type: &str,) -> AppResult<StreamedUpload> { let mut hasher = Sha256::new(); let mut chunks = Vec::new(); let mut pending = Vec::with_capacity(CHUNK_SIZE); let mut size = 0usize;
while let Some(frame) = body .frame() .await .transpose() .map_err(|e| invalid_request(format!("stageFile body read failed: {e}")))? { let Ok(data) = frame.into_data() else { continue; }; size = size .checked_add(data.len()) .ok_or_else(|| invalid_request("stageFile body is too large"))?; if size > MAX_STAGE_FILE_SIZE { return Err(invalid_request(format!( "stageFile body exceeds the {} byte limit", MAX_STAGE_FILE_SIZE ))); } hasher.update(&data); let mut remaining = data.as_ref(); while !remaining.is_empty() { let take = (CHUNK_SIZE - pending.len()).min(remaining.len()); pending.extend_from_slice(&remaining[..take]); remaining = &remaining[take..]; if pending.len() == CHUNK_SIZE { let offset = size - remaining.len() - pending.len(); upload_one_chunk( agent, mime_type, &mut chunks, offset, std::mem::take(&mut pending), ) .await?; pending = Vec::with_capacity(CHUNK_SIZE); } } }
if size == 0 { return Err(invalid_request( "stageFile requires a non-empty raw byte body", )); } if !pending.is_empty() { let offset = size - pending.len(); upload_one_chunk(agent, mime_type, &mut chunks, offset, pending).await?; } let digest = hasher.finalize().to_vec(); Ok(StreamedUpload { chunks, digest, size, })}
async fn upload_one_chunk( agent: &UserAgent, mime_type: &str, chunks: &mut Vec<Chunk>, offset: usize, bytes: Vec<u8>,) -> AppResult<()> { let size = bytes.len(); let blob = agent_upload_blob(agent, bytes, mime_type).await?; chunks.push(Chunk { blob: BlobRef::from(blob), offset: offset as i64, size: size as i64, extra_data: None, }); Ok(())}
pub(super) async fn agent_upload_blob( agent: &UserAgent, bytes: Vec<u8>, mime_type: &str,) -> AppResult<jacquard_common::types::blob::Blob> { agent .upload_blob(bytes, MimeType::new(mime_type)) .await .map_err(|e| agent_error("upload blob", e))}
async fn agent_delete_record<R>(agent: &UserAgent, rkey: RecordKey<Rkey>) -> AppResult<()>where R: Collection + Serialize,{ agent .delete_record::<R>(rkey) .await .map(|_| ()) .map_err(|e| agent_error("delete record", e))}
struct UploadPersistence<'a> { upload_id: &'a str, digest: &'a [u8], mime_type: &'a str, size: usize, filename: Option<&'a str>, file: &'a File,}
async fn persist_upload( state: &AppState, actor: &Did, upload: UploadPersistence<'_>,) -> AppResult<()> { let file_json = serde_json::to_string(upload.file) .map_err(|e| internal(format!("file metadata serialize: {e}")))?; let chunks_json = serde_json::to_string(&upload.file.chunks) .map_err(|e| internal(format!("chunk metadata serialize: {e}")))?; let now = chrono::Utc::now().timestamp_nanos_opt().unwrap_or_default(); db(sqlx::query!( "INSERT INTO upload_staging (owner_did, upload_id, sha256, mime_type, size, filename, status, file_json, chunks_json, created_at, updated_at) \ VALUES (?, ?, ?, ?, ?, ?, 'uploaded', ?, ?, ?, ?) \ ON CONFLICT(owner_did, upload_id) DO UPDATE SET sha256 = excluded.sha256, mime_type = excluded.mime_type, size = excluded.size, filename = excluded.filename, status = excluded.status, file_json = excluded.file_json, chunks_json = excluded.chunks_json, updated_at = excluded.updated_at", actor.as_ref(), upload.upload_id, upload.digest, upload.mime_type, upload.size as i64, upload.filename, file_json, chunks_json, now, now, ).execute(&state.pool).await)?; Ok(())}
pub(super) async fn load_upload_file( state: &AppState, actor: &Did, upload_id: &str,) -> AppResult<File> { #[derive(sqlx::FromRow)] struct Row { file_json: String, } let row = db(sqlx::query_as!(Row, "SELECT file_json FROM upload_staging WHERE owner_did = ? AND upload_id = ? AND status = 'uploaded'", actor.as_ref(), upload_id).fetch_optional(&state.pool).await)? .ok_or_else(|| invalid_request("unknown or unavailable uploadId"))?; serde_json::from_str(&row.file_json) .map_err(|e| internal(format!("staged file metadata parse: {e}")))}
async fn lookup_social_uri( state: &AppState, actor: &Did, collection: &str, subject_uri: &str,) -> AppResult<AtUri> { lookup_existing_social(state, actor, collection, subject_uri) .await? .map(|record| record.uri) .ok_or_else(|| invalid_request("social record is not indexed locally"))}
async fn ensure_follow_subject(state: &AppState, subject: &Did) -> AppResult<()> { let subject_ref = subject.as_ref(); let exists = db(sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM identities WHERE did = ? UNION SELECT 1 FROM profiles WHERE did = ?) AS \"exists!: bool\"", subject_ref, subject_ref, ) .fetch_one(&state.pool) .await)?; if !exists { let _ = views::resolve_handle(state, subject).await?; } Ok(())}
async fn lookup_follow_uri(state: &AppState, actor: &Did, subject: &Did) -> AppResult<AtUri> { lookup_existing_follow(state, actor, subject) .await? .map(|record| record.uri) .ok_or_else(|| invalid_request("follow record is not indexed locally"))}
struct ExistingSocial { cid: Cid, uri: AtUri,}
async fn lookup_existing_social( state: &AppState, actor: &Did, collection: &str, subject_uri: &str,) -> AppResult<Option<ExistingSocial>> { #[derive(sqlx::FromRow)] struct Row { cid: String, rkey: String, } let row = match collection { LIKE_COLLECTION => db(sqlx::query_as!( Row, "SELECT cid, rkey FROM likes WHERE did = ? AND subject_uri = ?", actor.as_ref(), subject_uri ) .fetch_optional(&state.pool) .await)?, SAVE_COLLECTION => db(sqlx::query_as!( Row, "SELECT cid, rkey FROM saves WHERE did = ? AND subject_uri = ?", actor.as_ref(), subject_uri ) .fetch_optional(&state.pool) .await)?, _ => return Err(invalid_request("unsupported social lookup")), }; let existing = row .map(|row| { let uri = at_uri_owned(actor, collection, &row.rkey)?; let cid = Cid::new_owned(row.cid.as_bytes()) .map_err(|e| internal(format!("stored social cid is invalid: {e}")))?; Ok(ExistingSocial { cid, uri }) }) .transpose()?; Ok(existing)}
async fn lookup_existing_follow( state: &AppState, actor: &Did, subject: &Did,) -> AppResult<Option<ExistingSocial>> { #[derive(sqlx::FromRow)] struct Row { cid: String, rkey: String, } let row = db(sqlx::query_as!( Row, "SELECT cid, rkey FROM follows WHERE follower_did = ? AND followed_did = ?", actor.as_ref(), subject.as_ref(), ) .fetch_optional(&state.pool) .await)?; row.map(|row| { let uri = at_uri_owned(actor, FOLLOW_COLLECTION, &row.rkey)?; let cid = Cid::new_owned(row.cid.as_bytes()) .map_err(|e| internal(format!("stored follow cid is invalid: {e}")))?; Ok(ExistingSocial { cid, uri }) }) .transpose()}
fn filename_header(headers: &HeaderMap) -> Option<String> { headers .get("x-polymodel-filename") .and_then(|v| v.to_str().ok()) .map(str::trim) .filter(|v| !v.is_empty()) .map(ToOwned::to_owned)}
#[cfg(test)]fn sha256_bytes(bytes: &[u8]) -> Vec<u8> { Sha256::digest(bytes).to_vec()}
fn hex_bytes(bytes: &[u8]) -> String { bytes.iter().map(|b| format!("{b:02x}")).collect()}
fn agent_error(operation: &str, err: jacquard::client::AgentError) -> AppError { if agent_error_is_auth(&err) { tracing::warn!(operation, error = %err, "authenticated PDS operation failed authorization"); return unauthorized(format!( "{operation}: authenticated PDS operation was not authorized" )); } tracing::error!(operation, error = %err, "authenticated PDS operation failed"); internal("authenticated PDS operation failed")}
fn client_error(operation: &str, err: jacquard_common::error::ClientError) -> AppError { if err.is_auth() { tracing::warn!(operation, error = %err, "authenticated PDS read failed authorization"); return unauthorized("authenticated PDS read was not authorized"); } tracing::error!(operation, error = %err, "authenticated PDS read failed"); internal("authenticated PDS read failed")}
fn client_error_is_not_found(err: &jacquard_common::error::ClientError) -> bool { err.is_not_found()}
/// Trait for XRPC error enums that carry a `RecordNotFound` variant.trait XrpcNotFound: core::error::Error { fn is_record_not_found(&self) -> bool;}
impl XrpcNotFound for GetRecordError { fn is_record_not_found(&self) -> bool { matches!(self, GetRecordError::RecordNotFound(_)) }}
impl XrpcNotFound for RecordError { fn is_record_not_found(&self) -> bool { matches!(self, RecordError::RecordNotFound(_)) }}
/// A `getRecord` for a missing record can come back as a successful HTTP/// response whose body decodes (via `into_output()`) to an XRPC `RecordNotFound`/// error rather than a transport-level `Err`. Detect that via the typed/// `RecordNotFound` variant, with a `Generic` fallback for non-conforming/// envelopes, so the caller can treat the record as absent.fn into_output_is_not_found<E: XrpcNotFound>(err: &XrpcError<E>) -> bool { match err { XrpcError::Xrpc(e) => e.is_record_not_found(), XrpcError::Generic(g) => g.error.as_str() == "RecordNotFound", _ => false, }}
/// An empty `app.bsky.actor.profile` used when an actor has no Bluesky profile/// record to copy basics from during polymodel profile bootstrap.fn empty_bsky_profile() -> BskyProfile { BskyProfile { avatar: None, banner: None, created_at: None, description: None, display_name: None, joined_via_starter_pack: None, labels: None, pinned_post: None, pronouns: None, website: None, extra_data: None, }}
fn agent_error_is_auth(err: &jacquard::client::AgentError) -> bool { err.is_auth()}
fn agent_error_is_not_found(err: &jacquard::client::AgentError) -> bool { err.client_error().is_some_and(|c| c.is_not_found())}
fn agent_error_is_conflict(err: &jacquard::client::AgentError) -> bool { err.source_downcast::<CreateRecordError>().is_some_and(|e| { matches!( e, CreateRecordError::AlreadyExists(_) | CreateRecordError::InvalidSwap(_) ) }) || err.client_error().is_some_and(|c| c.is_conflict())}
#[cfg(test)]mod tests { use std::str::FromStr;
use axum::response::IntoResponse; use axum_extra::extract::PrivateCookieJar; use jacquard_common::types::string::Did; use serde_json::json; use sqlx::SqlitePool; use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions}; use tower::ServiceExt;
use crate::appview::test_support::{ seed_identity, seed_oauth_session_at, seed_profile, test_blob, };
use super::*;
const DID_A: &str = "did:plc:aaaaaaaaaaaaaaaaaaaaaaaa"; const DID_B: &str = "did:plc:bbbbbbbbbbbbbbbbbbbbbbbb";
async fn state() -> AppState { 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(); let bootstrap = crate::oauth::bootstrap_oauth(pool.clone(), Some("http://localhost")) .expect("ephemeral OAuth bootstrap for tests"); AppState::new(pool, bootstrap) }
fn did(s: &str) -> Did { Did::new_owned(s).unwrap() }
fn uri(did: &str, collection: &str, rkey: &str) -> String { format!("at://{did}/{collection}/{rkey}") }
async fn seed_content_row( pool: &SqlitePool, table: &str, did: &str, rkey: &str, name: &str, ) -> String { let collection = match table { "things" => THING_COLLECTION, "models" => MODEL_COLLECTION, "parts" => PART_COLLECTION, _ => unreachable!(), }; let uri = uri(did, collection, rkey); let now = chrono::Utc::now().timestamp_nanos_opt().unwrap(); match table { "things" => { sqlx::query( "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 (?, ?, ?, 'bafycid', ?, NULL, 'CC-BY-4.0', NULL, NULL, NULL, NULL, NULL, ?, ?, ?)", ) .bind(did) .bind(rkey) .bind(&uri) .bind(name) .bind(now) .bind(now) .bind(json!({"name": name, "license": "CC-BY-4.0", "models": []}).to_string()) .execute(pool) .await .unwrap(); } "models" => { sqlx::query( "INSERT INTO models (did, rkey, uri, cid, name, summary, created_at, indexed_at, record_json) VALUES (?, ?, ?, 'bafycid', ?, NULL, ?, ?, ?)", ) .bind(did) .bind(rkey) .bind(&uri) .bind(name) .bind(now) .bind(now) .bind(json!({"name": name, "parts": []}).to_string()) .execute(pool) .await .unwrap(); } "parts" => { let file_json = json!({"mimeType": "model/stl", "size": 1, "chunks": []}).to_string(); sqlx::query( "INSERT INTO parts (did, rkey, uri, cid, name, format, file_json, created_at, indexed_at, record_json) VALUES (?, ?, ?, 'bafycid', ?, 'stl', ?, ?, ?, ?)", ) .bind(did) .bind(rkey) .bind(&uri) .bind(name) .bind(&file_json) .bind(now) .bind(now) .bind(json!({"name": name, "file": {"mimeType": "model/stl", "size": 1, "chunks": []}}).to_string()) .execute(pool) .await .unwrap(); } _ => unreachable!(), } uri }
#[tokio::test] async fn stage_file_route_accepts_format_matrix_and_persists_owner_metadata() { let cases: &[(&str, &str, &[u8])] = &[ ("model/stl", "model.stl", b"solid matrix\nendsolid matrix\n"), ("model/obj", "model.obj", b"v 0 0 0\n"), ( "model/gltf+json", "model.gltf", br#"{"asset":{"version":"2.0"}}"#, ), ("model/gltf-binary", "model.glb", b"glTF\x02\0\0\0"), ("model/3mf", "model.3mf", b"PK\x03\x04matrix"), ( "application/x-ldraw", "model.dat", b"0 FILE model.dat\n1 16 0 0 0 part.dat\n", ), ( "application/octet-stream", "extensionless", b"solid bounded-evidence\nendsolid bounded-evidence\n", ), ];
for (case_index, &(mime, filename, bytes)) in cases.iter().enumerate() { let mut state = state().await; let (pds, server) = crate::appview::test_support::loopback_pds_with_upload(mime).await; state.resolver = crate::appview::test_support::loopback_resolver(&pds); seed_identity(&state.pool, DID_A, "alice.com").await; seed_profile(&state.pool, DID_A, "Alice").await; let key = seed_oauth_session_at( &state, DID_A, &format!("stage-format-{case_index}"), Some(&pds), ) .await; let cookie_response = jacquard_axum::oauth::set_session_cookie( PrivateCookieJar::new(state.cookie_key.clone()), &state.oauth_config, &key, ) .unwrap() .into_response(); let cookie = cookie_response .headers() .get(axum::http::header::SET_COOKIE) .unwrap() .to_str() .unwrap() .split(';') .next() .unwrap() .to_owned(); let app = crate::appview::router().with_state(state.clone()); let digest = sha256_bytes(bytes); let response = app .oneshot( axum::http::Request::builder() .method("POST") .uri("/xrpc/space.polymodel.library.stageFile") .header(axum::http::header::COOKIE, cookie) .header("content-type", mime) .header("x-polymodel-filename", filename) .body(axum::body::Body::from(bytes.to_vec())) .unwrap(), ) .await .unwrap(); assert_eq!(response.status(), axum::http::StatusCode::OK, "{mime}"); let body = axum::body::to_bytes(response.into_body(), 1 << 20) .await .unwrap(); let output: StageFileOutput = serde_json::from_slice(&body).unwrap(); assert_eq!( output.status, Some(StageFileOutputStatus::Uploaded), "{mime}" ); assert_eq!(output.filename.as_deref(), Some(filename), "{mime}"); assert_eq!(output.file.mime_type.as_str(), mime, "{mime}"); assert_eq!(output.file.size, bytes.len() as i64, "{mime}"); assert_eq!( output.file.digest.as_deref(), Some(digest.as_slice()), "{mime}" ); assert_eq!(output.file.chunks.len(), 1, "{mime}"); let output_chunk = &output.file.chunks[0]; // Independently derive the expected chunk blob CID from the // request bytes — not from the response — so a bad upload // response cannot echo a wrong CID. let expected_cid = test_blob(bytes).0; let expected_offset: i64 = 0; let expected_size: i64 = bytes.len() as i64; assert_eq!(output_chunk.offset, expected_offset, "{mime}"); assert_eq!(output_chunk.size, expected_size, "{mime}"); assert_eq!( output_chunk.blob.blob().r#ref.as_str(), expected_cid, "{mime}" ); assert_eq!(output_chunk.blob.blob().mime_type.as_str(), mime, "{mime}"); assert_eq!(output_chunk.blob.blob().size, bytes.len(), "{mime}"); let upload_id = output.upload_id.to_string(); let row = sqlx::query_as::<_, (String, String, i64, Vec<u8>, String)>( "SELECT owner_did, filename, size, sha256, file_json FROM upload_staging WHERE owner_did = ? AND upload_id = ?", ) .bind(DID_A) .bind(&upload_id) .fetch_one(&state.pool) .await .unwrap(); assert_eq!(row.0, DID_A, "{mime}"); assert_eq!(row.1, filename, "{mime}"); assert_eq!(row.2, expected_size, "{mime}"); assert_eq!(row.3, digest, "{mime}"); let stored: File = serde_json::from_str(&row.4).unwrap(); assert_eq!(stored.mime_type.as_str(), mime, "{mime}"); assert_eq!(stored.size, expected_size, "{mime}"); assert_eq!(stored.digest.as_deref(), Some(digest.as_slice()), "{mime}"); assert_eq!(stored.chunks.len(), 1, "{mime}"); let stored_chunk = &stored.chunks[0]; // Assert stored manifest CID equals the independently-derived CID. assert_eq!(stored_chunk.offset, expected_offset, "{mime}"); assert_eq!(stored_chunk.size, expected_size, "{mime}"); assert_eq!( stored_chunk.blob.blob().r#ref.as_str(), expected_cid, "{mime}" ); assert_eq!(stored_chunk.blob.blob().mime_type.as_str(), mime, "{mime}"); assert_eq!(stored_chunk.blob.blob().size, bytes.len(), "{mime}"); assert_eq!( load_upload_file(&state, &did(DID_A), &upload_id) .await .unwrap() .size, bytes.len() as i64, "{mime}" ); assert!( load_upload_file(&state, &did(DID_B), &upload_id) .await .is_err(), "{mime}" ); server.abort(); } }
#[tokio::test] async fn upload_staging_round_trips_owner_scoped_file_metadata() { let state = state().await; let actor = did(DID_A); let bytes = b"solid cube"; let digest = sha256_bytes(bytes); let file = File { chunks: Vec::new(), digest: Some(Bytes::copy_from_slice(&digest)), mime_type: SmolStr::from("model/stl"), size: bytes.len() as i64, extra_data: None, };
persist_upload( &state, &actor, UploadPersistence { upload_id: "upload-1", digest: &digest, mime_type: "model/stl", size: bytes.len(), filename: Some("cube.stl"), file: &file, }, ) .await .unwrap();
let loaded = load_upload_file(&state, &actor, "upload-1").await.unwrap(); assert_eq!( loaded.digest.as_ref().map(Bytes::as_ref), Some(digest.as_slice()) ); assert_eq!(loaded.mime_type.as_str(), "model/stl"); assert_eq!(loaded.size, bytes.len() as i64); assert!( load_upload_file(&state, &did(DID_B), "upload-1") .await .is_err() ); }
#[tokio::test] async fn existing_like_lookup_returns_uri_and_cid_for_idempotent_duplicate_create() { let state = state().await; let actor = did(DID_A); let subject = seed_content_row(&state.pool, "things", DID_A, "thing1", "Thing").await; let cid = "bafyreiaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; sqlx::query( "INSERT INTO likes (did, rkey, cid, subject_uri, created_at) VALUES (?, ?, ?, ?, 1)", ) .bind(DID_A) .bind("like1") .bind(cid) .bind(&subject) .execute(&state.pool) .await .unwrap();
let existing = lookup_existing_social(&state, &actor, LIKE_COLLECTION, &subject) .await .unwrap() .expect("like should be indexed"); assert_eq!(existing.uri.as_ref(), uri(DID_A, LIKE_COLLECTION, "like1")); assert_eq!(existing.cid.to_string(), cid); }
#[tokio::test] async fn subject_validation_accepts_indexed_content_and_rejects_unknowns() { let state = state().await; let thing = seed_content_row(&state.pool, "things", DID_A, "thing1", "Thing").await; let thing_uri = AtUri::new_owned(&thing).unwrap(); validate_subject(&state, &thing_uri).await.unwrap();
let missing = AtUri::new_owned(uri(DID_A, THING_COLLECTION, "missing")).unwrap(); assert!(validate_subject(&state, &missing).await.is_err()); let unsupported = AtUri::new_owned(uri(DID_A, LIKE_COLLECTION, "like1")).unwrap(); assert!(validate_subject(&state, &unsupported).await.is_err()); }
#[tokio::test] async fn composition_load_requires_all_existing_children_to_be_owned() { let state = state().await; let thing = seed_content_row(&state.pool, "things", DID_A, "thing1", "Thing").await; let model = seed_content_row(&state.pool, "models", DID_A, "model1", "Model").await; let part = seed_content_row(&state.pool, "parts", DID_A, "part1", "Part").await; sqlx::query("INSERT INTO thing_models (thing_uri, model_uri, position) VALUES (?, ?, 0)") .bind(&thing) .bind(&model) .execute(&state.pool) .await .unwrap(); sqlx::query("INSERT INTO model_parts (model_uri, part_uri, position) VALUES (?, ?, 0)") .bind(&model) .bind(&part) .execute(&state.pool) .await .unwrap();
let composition = load_owned_composition(&state, &did(DID_A), &AtUri::new_owned(&thing).unwrap()) .await .unwrap(); assert_eq!(composition.models.len(), 1); assert_eq!(composition.parts.len(), 1);
sqlx::query("UPDATE parts SET did = ? WHERE uri = ?") .bind(DID_B) .bind(&part) .execute(&state.pool) .await .unwrap(); assert!( load_owned_composition(&state, &did(DID_A), &AtUri::new_owned(&thing).unwrap()) .await .is_err() ); }
#[tokio::test] async fn update_child_validation_allows_owned_detached_records_to_be_reattached() { let state = state().await; let actor = did(DID_A); let thing = seed_content_row(&state.pool, "things", DID_A, "thing1", "Thing").await; let active_model = seed_content_row(&state.pool, "models", DID_A, "model1", "Model").await; let detached_model = seed_content_row(&state.pool, "models", DID_A, "model2", "Detached").await; let foreign_model = seed_content_row(&state.pool, "models", DID_B, "model3", "Foreign").await; sqlx::query("INSERT INTO thing_models (thing_uri, model_uri, position) VALUES (?, ?, 0)") .bind(&thing) .bind(&active_model) .execute(&state.pool) .await .unwrap();
let previous = load_owned_composition(&state, &actor, &AtUri::new_owned(&thing).unwrap()) .await .unwrap(); let detached_uri = AtUri::new_owned(&detached_model).unwrap(); let detached_rkey = rkey_from_uri(&detached_uri).unwrap(); validate_existing_update_child( &state, &actor, &detached_uri, &detached_rkey, MODEL_COLLECTION, &previous.models, ) .await .unwrap();
let foreign_uri = AtUri::new_owned(&foreign_model).unwrap(); let foreign_rkey = rkey_from_uri(&foreign_uri).unwrap(); assert!( validate_existing_update_child( &state, &actor, &foreign_uri, &foreign_rkey, MODEL_COLLECTION, &previous.models, ) .await .is_err() ); }
#[tokio::test] async fn child_reference_counts_identify_shared_delete_children() { let state = state().await; let thing_a = seed_content_row(&state.pool, "things", DID_A, "thing1", "Thing A").await; let thing_b = seed_content_row(&state.pool, "things", DID_A, "thing2", "Thing B").await; let model = seed_content_row(&state.pool, "models", DID_A, "model1", "Model").await; let part = seed_content_row(&state.pool, "parts", DID_A, "part1", "Part").await; sqlx::query("INSERT INTO thing_models (thing_uri, model_uri, position) VALUES (?, ?, 0)") .bind(&thing_a) .bind(&model) .execute(&state.pool) .await .unwrap(); sqlx::query("INSERT INTO thing_models (thing_uri, model_uri, position) VALUES (?, ?, 0)") .bind(&thing_b) .bind(&model) .execute(&state.pool) .await .unwrap(); sqlx::query("INSERT INTO model_parts (model_uri, part_uri, position) VALUES (?, ?, 0)") .bind(&model) .bind(&part) .execute(&state.pool) .await .unwrap();
assert_eq!( model_reference_count(&state, &AtUri::new_owned(&model).unwrap()) .await .unwrap(), 2 ); assert_eq!( part_reference_count(&state, &AtUri::new_owned(&part).unwrap()) .await .unwrap(), 1 ); }
#[tokio::test] async fn child_reference_counts_preserve_parts_reused_across_models() { let state = state().await; let thing = seed_content_row(&state.pool, "things", DID_A, "thing1", "Thing").await; let model_a = seed_content_row(&state.pool, "models", DID_A, "model1", "Model A").await; let model_b = seed_content_row(&state.pool, "models", DID_A, "model2", "Model B").await; let part = seed_content_row(&state.pool, "parts", DID_A, "part1", "Shared Part").await; for (position, model) in [(0, &model_a), (1, &model_b)] { sqlx::query( "INSERT INTO thing_models (thing_uri, model_uri, position) VALUES (?, ?, ?)", ) .bind(&thing) .bind(model) .bind(position) .execute(&state.pool) .await .unwrap(); sqlx::query("INSERT INTO model_parts (model_uri, part_uri, position) VALUES (?, ?, 0)") .bind(model) .bind(&part) .execute(&state.pool) .await .unwrap(); }
let composition = load_owned_composition(&state, &did(DID_A), &AtUri::new_owned(&thing).unwrap()) .await .unwrap(); assert_eq!(composition.models.len(), 2); assert_eq!( composition.parts.len(), 1, "reused part should be deduped once" ); assert_eq!( part_reference_count(&state, &AtUri::new_owned(&part).unwrap()) .await .unwrap(), 2 ); }
// --- PM-61 typed error classification helpers ---
#[test] fn client_error_auth_maps_to_unauthorized() { let err = jacquard_common::error::ClientError::auth( jacquard_common::error::AuthError::TokenExpired, ); let result = client_error("test op", err); assert!(matches!(result, AppError::Unauthorized(_))); }
#[test] fn client_error_http_401_maps_to_unauthorized() { let err = jacquard_common::error::ClientError::http(http::StatusCode::UNAUTHORIZED, None); let result = client_error("test op", err); assert!(matches!(result, AppError::Unauthorized(_))); }
#[test] fn client_error_other_maps_to_internal() { let err = jacquard_common::error::ClientError::http( http::StatusCode::INTERNAL_SERVER_ERROR, None, ); let result = client_error("test op", err); assert!(!matches!(result, AppError::Unauthorized(_))); }
#[test] fn client_error_is_not_found_typed() { let err = jacquard_common::error::ClientError::http(http::StatusCode::NOT_FOUND, None); assert!(client_error_is_not_found(&err));
let err = jacquard_common::error::ClientError::http(http::StatusCode::OK, None); assert!(!client_error_is_not_found(&err)); }
#[test] fn into_output_not_found_via_typed_get_record_error() { let err: XrpcError<GetRecordError> = XrpcError::Xrpc(GetRecordError::RecordNotFound(None)); assert!(into_output_is_not_found(&err)); }
#[test] fn into_output_not_found_via_generic_code() { let generic = jacquard_common::xrpc::GenericXrpcError { error: "RecordNotFound".into(), message: None, nsid: "com.atproto.repo.getRecord", method: "GET", http_status: http::StatusCode::BAD_REQUEST, }; let err: XrpcError<GetRecordError> = XrpcError::Generic(generic); assert!(into_output_is_not_found(&err)); }
#[test] fn into_output_not_found_via_typed_record_error() { let err: XrpcError<RecordError> = XrpcError::Xrpc(RecordError::RecordNotFound(None)); assert!(into_output_is_not_found(&err)); }
#[test] fn into_output_not_found_false_for_auth_error() { let err: XrpcError<GetRecordError> = XrpcError::Auth(jacquard_common::error::AuthError::TokenExpired); assert!(!into_output_is_not_found(&err)); }
#[test] fn agent_error_auth_typed() { let err = jacquard::client::AgentError::auth(jacquard_common::error::AuthError::TokenExpired); assert!(agent_error_is_auth(&err)); }
#[test] fn agent_error_auth_from_wrapped_client_401() { let err = jacquard::client::AgentError::from(jacquard_common::error::ClientError::auth( jacquard_common::error::AuthError::TokenExpired, )); assert!(agent_error_is_auth(&err)); }
#[test] fn agent_error_not_found_from_client_404() { let err = jacquard::client::AgentError::from(jacquard_common::error::ClientError::http( http::StatusCode::NOT_FOUND, None, )); assert!(agent_error_is_not_found(&err)); }
#[test] fn agent_error_conflict_from_already_exists() { let err = jacquard::client::AgentError::sub_operation( "create record", CreateRecordError::AlreadyExists(None), ); assert!(agent_error_is_conflict(&err)); }
#[test] fn agent_error_conflict_from_invalid_swap() { let err = jacquard::client::AgentError::sub_operation( "create record", CreateRecordError::InvalidSwap(None), ); assert!(agent_error_is_conflict(&err)); }
#[test] fn agent_error_conflict_from_client_409() { let err = jacquard::client::AgentError::from(jacquard_common::error::ClientError::http( http::StatusCode::CONFLICT, None, )); assert!(agent_error_is_conflict(&err)); }
#[test] fn agent_error_not_conflict_for_internal_error() { let err = jacquard::client::AgentError::from(jacquard_common::error::ClientError::http( http::StatusCode::INTERNAL_SERVER_ERROR, None, )); assert!(!agent_error_is_conflict(&err)); }}