Something went wrong. Try again.
atproto Thingiverse but good
Something went wrong. Try again.
Rust
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583158415851586158715881589159015911592159315941595159615971598159916001601160216031604160516061607160816091610161116121613161416151616161716181619162016211622162316241625162616271628162916301631163216331634163516361637163816391640164116421643164416451646164716481649165016511652165316541655165616571658165916601661166216631664166516661667166816691670167116721673167416751676167716781679168016811682168316841685168616871688168916901691169216931694169516961697169816991700170117021703170417051706170717081709171017111712171317141715171617171718171917201721172217231724172517261727172817291730173117321733173417351736173717381739174017411742174317441745174617471748174917501751175217531754175517561757175817591760176117621763176417651766176717681769177017711772177317741775177617771778177917801781178217831784178517861787178817891790179117921793179417951796179717981799180018011802180318041805180618071808180918101811181218131814181518161817181818191820182118221823182418251826182718281829183018311832183318341835183618371838183918401841184218431844184518461847184818491850185118521853185418551856185718581859186018611862186318641865186618671868186918701871187218731874187518761877187818791880188118821883188418851886188718881889189018911892189318941895189618971898189919001901190219031904190519061907190819091910191119121913191419151916191719181919192019211922192319241925192619271928192919301931193219331934193519361937193819391940194119421943194419451946194719481949195019511952195319541955195619571958195919601961196219631964196519661967196819691970197119721973197419751976197719781979198019811982198319841985198619871988198919901991199219931994199519961997199819992000200120022003200420052006200720082009201020112012201320142015201620172018201920202021202220232024202520262027202820292030203120322033203420352036203720382039204020412042204320442045204620472048204920502051205220532054205520562057205820592060206120622063206420652066206720682069207020712072207320742075207620772078207920802081208220832084208520862087208820892090209120922093209420952096209720982099210021012102210321042105210621072108210921102111211221132114211521162117211821192120212121222123212421252126212721282129213021312132213321342135213621372138213921402141214221432144214521462147214821492150215121522153215421552156215721582159216021612162216321642165216621672168216921702171217221732174217521762177217821792180218121822183218421852186218721882189219021912192219321942195219621972198219922002201220222032204220522062207220822092210221122122213221422152216221722182219222022212222222322242225222622272228222922302231223222332234223522362237223822392240//! 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, header};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::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(SmolStr::from(staged.digest.clone())), mime_type: SmolStr::from(mime_type.clone()), size: staged.size as i64, extra_data: None, }; let upload_id = format!("{}-{}", &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: String, 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() .iter() .map(|b| format!("{b:02x}")) .collect(); 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 str, 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(())}
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()}
pub(super) fn content_type(headers: &HeaderMap) -> String { headers .get(header::CONTENT_TYPE) .and_then(|v| v.to_str().ok()) .and_then(|v| v.split(';').next()) .map(str::trim) .filter(|v| !v.is_empty()) .unwrap_or("application/octet-stream") .to_string()}
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 hex_sha256(bytes: &[u8]) -> String { let digest = Sha256::digest(bytes); digest.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("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 jacquard_common::types::string::Did; use serde_json::json; use sqlx::SqlitePool; use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
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 upload_staging_round_trips_owner_scoped_file_metadata() { let state = state().await; let actor = did(DID_A); let bytes = b"solid cube"; let digest = hex_sha256(bytes); let file = File { chunks: Vec::new(), digest: Some(SmolStr::from(digest.clone())), 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_deref(), Some(digest.as_str())); 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)); }}