Something went wrong. Try again.
atproto Thingiverse but good
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379//! `space.polymodel.library.getPartFile` — reassemble a part's chunked geometry//! file and stream it to the client.//!//! The handler returns `Result<Response, AppError>`, not `XrpcResponse<_>`, so//! it can set `Content-Type` and `Content-Disposition` on the success body.//! `XrpcResponse` hardcodes `Content-Type: */*` and has no disposition hook.//! The error axis still goes through `AppError` → `GenericXrpcErrorResponse`//! like every other handler.//!//! **Streaming model.** Each chunk's blob is fetched from the actor's PDS via//! `com.atproto.sync.getBlob` and piped straight into the response body in//! offset order — the server never buffers the whole file (or a whole chunk).//! Chunk 0 is prefetched before the `Response` is returned so a first-fetch//! PDS failure maps to a clean 500. From chunk 1 onward a fetch failure or//! truncation surfaces as a broken body (headers are already committed); the//! client catches it via `file.size` + `file.digest` (see//! `polymodel_renderer_protocol::verify_part_file`).
use axum::body::Body;use axum::extract::State;use axum::http::{StatusCode, header};use axum::response::Response;use futures::stream::{self, StreamExt};use jacquard::identity::resolver::IdentityResolver;use jacquard_axum::ExtractXrpc;use jacquard_common::types::string::{Cid, Did};use jacquard_common::xrpc::XrpcExt;use polymodel_api::com_atproto::sync::get_blob::GetBlob;use polymodel_api::space_polymodel::library::File;use polymodel_api::space_polymodel::library::get_part_file::GetPartFileRequest;
use super::error::{AppResult, db, internal, invalid_request, not_found, verify_chunk_cid};use super::state::AppState;
/// Reassemble and stream a part's chunked geometry file.////// Accepts an at-URI pointing at a `space.polymodel.library.part` record. The/// part's `File` manifest is loaded from the local SQLite projection, validated,/// then each chunk's blob is fetched from the actor's PDS and streamed in offset/// order. Digest and total-length verification are client-side.pub(super) async fn get_part_file( State(state): State<AppState>, ExtractXrpc(req): ExtractXrpc<GetPartFileRequest>,) -> AppResult<Response> { let path = req .uri .path() .ok_or_else(|| invalid_request("missing at-uri path"))?; match path.collection.as_str() { "space.polymodel.library.part" => {} _ => return Err(invalid_request("at-uri is not a part record")), } let authority = req.uri.authority(); let rkey = path .rkey .ok_or_else(|| invalid_request("missing part rkey"))?;
// Resolve authority to DID before the DB query — the local projection is // keyed by DID, so a handle-authority at-uri would miss otherwise. let did = super::actor::resolve_actor(&state, &authority.convert()).await?;
#[derive(sqlx::FromRow)] struct PartDownloadRow { name: String, file_json: String, }
let row = db( sqlx::query_as!( PartDownloadRow, r#"SELECT name AS "name!", file_json AS "file_json!" FROM parts WHERE did = ? AND rkey = ?"#, did.as_ref(), rkey.as_ref(), ) .fetch_optional(&state.pool) .await, )? .ok_or_else(not_found)?;
let file: File = serde_json::from_str(&row.file_json) .map_err(|e| internal(format!("part file manifest parse failed: {e}")))?; if file.chunks.is_empty() { return Err(not_found()); } validate_file_manifest(&file)?;
stream_projected_file(&state, did, row.name, file).await}
async fn collect_verified_blob( response: jacquard_common::StreamingResponse, expected_size: i64, cid: &Cid, pds: impl AsRef<str>,) -> AppResult<Vec<u8>> { let (_, body) = response.into_parts(); let pds = pds.as_ref(); let mut stream = body.into_inner(); let mut bytes = Vec::with_capacity(usize::try_from(expected_size.max(0)).unwrap_or_default()); while let Some(chunk) = stream.next().await { let chunk = chunk.map_err(|error| internal(format!("blob fetch from {pds} failed: {error}")))?; bytes.extend_from_slice(&chunk); if bytes.len() as i64 > expected_size { return Err(invalid_request("blob exceeded projected chunk size")); } } if bytes.len() as i64 != expected_size { return Err(invalid_request("blob did not match projected chunk size")); } verify_chunk_cid(cid.as_str(), &bytes) .map_err(|error| invalid_request(format!("chunk integrity check failed: {error}")))?; Ok(bytes)}
/// Stream a projected file through its owner's PDS.////// This is shared by the public Part download and authenticated compound/// resource routes. The first blob is fetched before returning the response so/// an initial PDS failure is reported as an HTTP error; later failures surface/// as a body-stream error after headers are committed.pub(super) async fn stream_projected_file( state: &AppState, did: Did, name: String, file: File,) -> AppResult<Response> { let did_doc_response = state .resolver .resolve_did_doc(&did) .await .map_err(|e| internal(format!("DID document resolution failed: {e}")))?; let did_doc = did_doc_response .parse() .map_err(|e| internal(format!("DID document parse failed: {e}")))?; let pds_endpoint = did_doc .pds_endpoint() .ok_or_else(|| internal("DID document does not advertise a PDS endpoint"))? .to_owned();
let ordered_cids: Vec<Cid> = ordered_chunks(&file) .iter() .map(|chunk| { Cid::new_owned(chunk.blob.blob().r#ref.as_str().as_bytes()) .map_err(|e| internal(format!("part blob CID is invalid: {e}"))) }) .collect::<AppResult<_>>()?;
// Prefetch chunk 0 so a first-fetch PDS failure returns a clean 500 before // the 200 + body is committed. The response body chains this already-open // stream with lazy fetches for the remaining chunks. let first_cid = ordered_cids[0].clone(); let first_response = state .resolver .xrpc(pds_endpoint.borrow()) .download(&GetBlob { did: did.clone(), cid: first_cid.clone(), }) .await .map_err(|e| internal(format!("blob fetch from {pds_endpoint} failed: {e}")))?; if !first_response.status().is_success() { return Err(internal(format!( "blob fetch from {pds_endpoint} failed with HTTP {}", first_response.status() ))); } let first_bytes = collect_verified_blob( first_response, ordered_chunks(&file)[0].size, &first_cid, pds_endpoint.as_str(), ) .await?;
let remaining: Vec<(i64, Cid)> = ordered_chunks(&file)[1..] .iter() .zip(ordered_cids[1..].iter().cloned()) .map(|(chunk, cid)| (chunk.size, cid)) .collect(); let state_clone = state.clone(); let pds_clone = pds_endpoint.clone(); let did_clone = did.clone();
type BoxError = Box<dyn std::error::Error + Send + Sync>;
let first_stream = stream::once(async move { Ok::<_, BoxError>(bytes::Bytes::from(first_bytes)) }); let rest_stream = stream::iter(remaining).then(move |(expected_size, cid)| { let state = state_clone.clone(); let pds = pds_clone.clone(); let did = did_clone.clone(); let pds_for_error = pds.clone(); async move { let response = state .resolver .xrpc(pds.borrow()) .download(&GetBlob { did: did.clone(), cid: cid.clone(), }) .await .map_err(|e| -> BoxError { Box::new(internal(format!( "blob fetch from {pds_for_error} failed: {e}" ))) })?; if !response.status().is_success() { return Err(Box::new(internal(format!( "blob fetch from {pds} failed with HTTP {}", response.status() ))) as BoxError); } let bytes = collect_verified_blob(response, expected_size, &cid, &pds) .await .map_err(|error| -> BoxError { Box::new(error) })?; Ok::<_, BoxError>(bytes::Bytes::from(bytes)) } });
let body_stream = first_stream.chain(rest_stream);
let filename = download_filename(&name); let content_type = file.mime_type.to_string(); Response::builder() .status(StatusCode::OK) .header(header::CONTENT_TYPE, content_type) .header( header::CONTENT_DISPOSITION, format!("attachment; filename=\"{filename}\""), ) .header(header::CONTENT_LENGTH, file.size) .body(Body::from_stream(body_stream)) .map_err(|e| internal(format!("download response build failed: {e}")))}
/// Return a file's chunks sorted by offset. The streaming response yields them/// in this order so the client reassembles the original file.pub(super) fn ordered_chunks(file: &File) -> Vec<&polymodel_api::space_polymodel::library::Chunk> { let mut chunks: Vec<_> = file.chunks.iter().collect(); chunks.sort_by_key(|chunk| chunk.offset); chunks}
/// Cheap manifest preflight: offset contiguity, chunk count, per-chunk sanity./// Does **not** cap total file size — the PDS bounds each blob, and a total cap/// breaks the large-model case chunking exists for.pub(super) fn validate_file_manifest(file: &File) -> AppResult<()> { if file.size < 0 { return Err(invalid_request("part file size is negative")); } if file.chunks.len() > 4096 { return Err(invalid_request("part file has too many chunks")); } let mut chunks = file.chunks.iter().collect::<Vec<_>>(); chunks.sort_by_key(|chunk| chunk.offset); let mut expected_offset = 0_i64; for chunk in chunks { if chunk.offset != expected_offset || chunk.size < 0 { return Err(invalid_request("part file chunks are not contiguous")); } expected_offset = expected_offset .checked_add(chunk.size) .ok_or_else(|| invalid_request("part file chunk sizes overflow"))?; } if expected_offset != file.size { return Err(invalid_request("part file chunks do not match file size")); } Ok(())}
fn download_filename(name: &str) -> String { let cleaned = name .chars() .map(|ch| match ch { 'a'..='z' | 'A'..='Z' | '0'..='9' | '.' | '-' | '_' => ch, _ => '_', }) .collect::<String>(); if cleaned.trim_matches('_').is_empty() { "polymodel-part.bin".to_string() } else { cleaned }}
#[cfg(test)]mod tests { use super::{download_filename, ordered_chunks, validate_file_manifest}; use crate::appview::error::{ChunkIntegrityError, verify_chunk_cid}; use jacquard_common::deps::smol_str::SmolStr; use jacquard_common::types::blob::{Blob, BlobRef, MimeType}; use jacquard_common::types::cid::CidLink; use polymodel_api::space_polymodel::library::{Chunk, File};
fn blob_ref() -> BlobRef { BlobRef::from(Blob { r#ref: CidLink::new_owned( b"bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku", ) .unwrap(), mime_type: MimeType::new_owned("model/stl"), size: 4, }) }
fn file(chunks: Vec<Chunk>, size: i64) -> File { File { chunks, digest: None, mime_type: SmolStr::from("model/stl"), size, extra_data: None, } }
#[test] fn chunk_integrity_reports_distinct_failure_classes() { assert!(matches!( verify_chunk_cid("not-a-cid", b"chunk"), Err(ChunkIntegrityError::InvalidCid(_)) )); let cid = cid::Cid::new_v1( 0x55, cid::multihash::Multihash::wrap(0x55, b"chunk").unwrap(), ) .to_string(); assert_eq!( verify_chunk_cid(&cid, b"chunk"), Err(ChunkIntegrityError::UnsupportedHash(0x55)) ); let cid = cid::Cid::new_v1( 0x55, cid::multihash::Multihash::wrap(0x12, &[0_u8; 32]).unwrap(), ) .to_string(); assert_eq!( verify_chunk_cid(&cid, b"chunk"), Err(ChunkIntegrityError::DigestMismatch) ); }
#[test] fn download_filename_removes_header_unsafe_characters() { assert_eq!(download_filename("gear bracket.stl"), "gear_bracket.stl"); assert_eq!(download_filename("../../x\r\n"), ".._.._x__"); assert_eq!(download_filename("!!!"), "polymodel-part.bin"); }
#[test] fn validate_file_manifest_requires_contiguous_bounded_chunks() { let chunk = Chunk::builder().blob(blob_ref()).offset(0).size(4).build(); assert!(validate_file_manifest(&file(vec![chunk.clone()], 4)).is_ok());
let gapped = Chunk::builder().blob(blob_ref()).offset(2).size(4).build(); assert!(validate_file_manifest(&file(vec![gapped], 6)).is_err()); }
#[test] fn validate_file_manifest_rejects_too_many_chunks() { let chunk = Chunk::builder().blob(blob_ref()).offset(0).size(4).build(); let mut chunks = vec![chunk; 4097]; for (i, chunk) in chunks.iter_mut().enumerate() { chunk.offset = i as i64; } assert!(validate_file_manifest(&file(chunks, 4097 * 4)).is_err()); }
#[test] fn ordered_chunks_sorts_by_offset() { let chunk_b = Chunk::builder().blob(blob_ref()).offset(4).size(4).build(); let chunk_a = Chunk::builder().blob(blob_ref()).offset(0).size(4).build(); let file = file(vec![chunk_b, chunk_a], 8); let ordered = ordered_chunks(&file); assert_eq!(ordered[0].offset, 0); assert_eq!(ordered[1].offset, 4); }}