//! 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>; /// 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; pub(super) async fn bootstrap_profile( State(state): State, ExtractOAuthSession(session): ExtractOAuthSession, ) -> AppResult> { 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, ExtractOAuthSession(session): ExtractOAuthSession, headers: HeaderMap, body: Body, ) -> AppResult> { 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, ExtractOAuthSession(session): ExtractOAuthSession, ExtractXrpc(req): ExtractXrpc, ) -> AppResult> { 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, ExtractOAuthSession(session): ExtractOAuthSession, ExtractXrpc(req): ExtractXrpc, ) -> AppResult> { 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, ExtractOAuthSession(session): ExtractOAuthSession, ExtractXrpc(req): ExtractXrpc, ) -> AppResult> { 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::(&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, ExtractOAuthSession(session): ExtractOAuthSession, ExtractXrpc(req): ExtractXrpc, ) -> AppResult> { 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, ExtractOAuthSession(session): ExtractOAuthSession, ExtractXrpc(req): ExtractXrpc, ) -> AppResult> { 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, ExtractOAuthSession(session): ExtractOAuthSession, ExtractXrpc(req): ExtractXrpc, ) -> AppResult> { 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::(&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, ExtractOAuthSession(session): ExtractOAuthSession, ExtractXrpc(req): ExtractXrpc, ) -> AppResult> { 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::(&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, ExtractOAuthSession(session): ExtractOAuthSession, ExtractXrpc(req): ExtractXrpc, ) -> AppResult> { 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, ExtractOAuthSession(session): ExtractOAuthSession, ExtractXrpc(req): ExtractXrpc, ) -> AppResult> { 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::(&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, ExtractOAuthSession(session): ExtractOAuthSession, ExtractXrpc(req): ExtractXrpc, ) -> AppResult> { 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, ExtractOAuthSession(session): ExtractOAuthSession, ExtractXrpc(req): ExtractXrpc, ) -> AppResult> { let agent = Agent::from(session); let actor = authenticated_did(&agent).await?; delete_social::(&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, cid: Option, } pub(super) async fn authenticated_did(agent: &UserAgent) -> AppResult { 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 { // 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::(&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::(&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::(&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 { 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 { 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::(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::(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 { 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::(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 { 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( 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::(agent, rkey.clone()).await?; eager_delete(state, actor, collection, rkey.as_ref()).await } struct ExistingComposition { thing_uri: AtUri, thing_rkey: RecordKey, models: Vec, parts: Vec, } struct CompositionChild { rkey: RecordKey, 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 { 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> { let mut deleted_uris = Vec::new(); for model in &composition.models { if model_reference_count(state, &model.uri).await? > 1 { continue; } agent_delete_record::(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::(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 { 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 { 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::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::any(SmolStr::from(value)) .map_err(|e| invalid_request(format!("invalid rkey {value:?}: {e}"))) } fn rkey_from_uri(uri: &AtUri) -> AppResult> { 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::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, 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( state: &AppState, actor: &Did, collection: &str, rkey: &str, cid: Option, 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, digest: Vec, size: usize, } async fn upload_streamed_file( agent: &UserAgent, mut body: Body, mime_type: &str, ) -> AppResult { 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, offset: usize, bytes: Vec, ) -> 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, mime_type: &str, ) -> AppResult { agent .upload_blob(bytes, MimeType::new(mime_type)) .await .map_err(|e| agent_error("upload blob", e)) } async fn agent_delete_record(agent: &UserAgent, rkey: RecordKey) -> AppResult<()> where R: Collection + Serialize, { agent .delete_record::(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 { #[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 { 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 { 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> { #[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> { #[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 { 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 { 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(err: &XrpcError) -> 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::().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, 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 = 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 = XrpcError::Generic(generic); assert!(into_output_is_not_found(&err)); } #[test] fn into_output_not_found_via_typed_record_error() { let err: XrpcError = 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 = 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)); } }