Something went wrong. Try again.
atproto Thingiverse but good
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432//! XRPC passthrough proxy to the user's PDS (PM-47).//!//! A fixed, structural allowlist of non-`space.polymodel.*` XRPC methods is//! served from this origin so the browser stays single-origin (PM-9 §4.5).//! Each allowlisted method dispatches to its generated typed request struct and//! is forwarded to the user's PDS (authenticated methods, via the OAuth session//! agent's *signing* send path) or the public AppView (public methods). Upstream//! XRPC errors are forwarded with their real HTTP status + body; only genuine//! transport/internal failures become 500.//!//! Allowlist (the direct repo record operations only)://! - `com.atproto.repo.{createRecord, putRecord, deleteRecord, getRecord,//! listRecords, describeRepo, uploadBlob}` — authenticated, to the user's PDS.//! - `com.atproto.identity.resolveHandle` and `app.bsky.actor.getProfile` —//! public, to the AppView.//!//! Batch/admin repo methods (`applyWrites`, `importRepo`, `listMissingBlobs`)//! and every other NSID have no route and 404.//!//! Read-your-writes: `repo.getRecord` of a `space.polymodel.*` record reflects a//! not-yet-firehose-confirmed local write/delete via the `pending_writes` ///! `pending_deletes` tables populated by the shared eager-projection path (see//! `indexing::projection`). Proxied writes to `space.polymodel.*` are eagerly//! projected through that same path for read-your-own-writes.
use axum::body::Body;use axum::extract::State;use axum::http::{HeaderMap, HeaderValue, StatusCode, header};use axum::response::{IntoResponse, Response};use jacquard::client::{Agent, AgentSession};use jacquard::identity::PublicResolver;use jacquard::oauth::client::OAuthSession;use jacquard_axum::ExtractXrpc;use jacquard_axum::oauth::ExtractOAuthSession;use jacquard_common::error::{ClientError, ClientErrorKind, HttpError};use jacquard_common::types::string::Did;use jacquard_common::types::value::to_data;use jacquard_common::xrpc::{Response as XrpcResponse, XrpcClient, XrpcResp};use polymodel_api::com_atproto::identity::resolve_handle::ResolveHandleRequest;use polymodel_api::com_atproto::repo::{ create_record::CreateRecordRequest, delete_record::DeleteRecordRequest, describe_repo::DescribeRepoRequest, get_record::GetRecordRequest, list_records::ListRecordsRequest, put_record::PutRecordRequest, upload_blob::{UploadBlob, UploadBlobRequest},};use serde_json::Value;use sqlx::SqlitePool;
use super::error::{AppError, AppResult, db, invalid_request, unauthorized};use super::state::AppState;use crate::indexing::projection::{ProjectionEvent, project_eager_record};use crate::oauth::SqliteAuthStore;
type UserAgent = Agent<OAuthSession<PublicResolver, SqliteAuthStore>>;
const POLYMODEL_PREFIX: &str = "space.polymodel.";
// ===========================================================================// PDS reads (authenticated; forwarded to the user's PDS)// ===========================================================================
pub(super) async fn repo_get_record( State(state): State<AppState>, ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>, ExtractXrpc(req): ExtractXrpc<GetRecordRequest>,) -> AppResult<Response> { let agent = Agent::from(session); let collection = req.collection.as_ref().to_string(); let polymodel = collection.starts_with(POLYMODEL_PREFIX); // Pending-ops key: canonicalize the request repo to a DID (handles resolve) // so read-your-writes matches the DID-keyed pending rows for both DID- and // handle-authority reads. A resolved *other* repo can't match this account's // pending rows (the actor only writes its own repo), so there's no cross-repo // false positive. If the repo can't be canonicalized, skip the override and // forward the PDS result. let pending_uri = if polymodel { match super::actor::resolve_actor(&state, &req.repo).await { Ok(did) => { let rkey = req.rkey.as_ref().to_string(); Some(format!("at://{}/{}/{}", did.as_ref(), collection, rkey)) } Err(e) => { tracing::warn!(error = ?e, "pending-ops key: repo did not canonicalize; skipping override"); None } } } else { None }; let result = agent.send(req).await; if let Some(uri) = pending_uri { // Read-your-writes: a polymodel record may carry a pending local // write/delete the PDS/AppView hasn't reflected yet. Keyed on the // request — the PDS body is only consulted for cid comparison — so a // pending write still wins when a lagging source reports not-found. let pds_cid = match &result { Ok(resp) if resp.status().is_success() => { serde_json::from_slice::<Value>(resp.buffer()) .ok() .and_then(|v| v.get("cid").and_then(|c| c.as_str()).map(str::to_owned)) } _ => None, }; match try_get_record_override(&state.pool, &uri, pds_cid.as_deref()).await { Ok(Some(overridden)) => return Ok(overridden), Ok(None) => {} Err(e) => { tracing::warn!(error = ?e, "pending-ops override lookup failed; forwarding PDS result") } } } Ok(forward_xrpc(result))}
pub(super) async fn repo_list_records( ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>, ExtractXrpc(req): ExtractXrpc<ListRecordsRequest>,) -> AppResult<Response> { let agent = Agent::from(session); Ok(forward_xrpc(agent.send(req).await))}
pub(super) async fn repo_describe_repo( ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>, ExtractXrpc(req): ExtractXrpc<DescribeRepoRequest>,) -> AppResult<Response> { let agent = Agent::from(session); Ok(forward_xrpc(agent.send(req).await))}
// ===========================================================================// PDS writes (authenticated; forwarded to the user's PDS, eagerly projected)// ===========================================================================
pub(super) async fn repo_create_record( State(state): State<AppState>, ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>, ExtractXrpc(req): ExtractXrpc<CreateRecordRequest>,) -> AppResult<Response> { let agent = Agent::from(session); let did = actor_did(&agent).await?; let collection = req.collection.as_ref().to_string(); let record = serde_json::to_value(&req.record).ok(); let result = agent.send(req).await; if let Ok(resp) = &result && resp.status().is_success() && collection.starts_with(POLYMODEL_PREFIX) && let Some(record) = record.as_ref() { eager_project(&state, &did, &collection, "create", record, resp.buffer()).await; } Ok(forward_xrpc(result))}
pub(super) async fn repo_put_record( State(state): State<AppState>, ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>, ExtractXrpc(req): ExtractXrpc<PutRecordRequest>,) -> AppResult<Response> { let agent = Agent::from(session); let did = actor_did(&agent).await?; let collection = req.collection.as_ref().to_string(); let record = serde_json::to_value(&req.record).ok(); let result = agent.send(req).await; if let Ok(resp) = &result && resp.status().is_success() && collection.starts_with(POLYMODEL_PREFIX) && let Some(record) = record.as_ref() { eager_project(&state, &did, &collection, "update", record, resp.buffer()).await; } Ok(forward_xrpc(result))}
pub(super) async fn repo_delete_record( State(state): State<AppState>, ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>, ExtractXrpc(req): ExtractXrpc<DeleteRecordRequest>,) -> AppResult<Response> { let agent = Agent::from(session); let did = actor_did(&agent).await?; let collection = req.collection.as_ref().to_string(); let rkey = req.rkey.as_ref().to_string(); let result = agent.send(req).await; if let Ok(resp) = &result && resp.status().is_success() && collection.starts_with(POLYMODEL_PREFIX) { let event = ProjectionEvent { seq: 0, did: did.as_ref().to_string(), collection: collection.clone(), rkey, action: "delete".to_string(), record: None, cid: None, }; let _guard = state.write_lock.lock().await; if let Err(e) = project_eager_record(&state.pool, &event).await { tracing::error!(error = %e, %collection, "eager projection of proxied delete failed"); } } Ok(forward_xrpc(result))}
pub(super) async fn repo_upload_blob( ExtractOAuthSession(session): ExtractOAuthSession<PublicResolver, SqliteAuthStore>, headers: HeaderMap, ExtractXrpc(req): ExtractXrpc<UploadBlobRequest>,) -> AppResult<Response> { let agent = Agent::from(session); let UploadBlob { body } = req; // The generated `UploadBlob` carries raw `*/*` bytes and no mime; thread the // real inbound Content-Type through the call options (the same mechanism the // `AgentSessionExt::upload_blob` wrapper uses) so the PDS receives it. Using // `send_with_opts` directly preserves the raw response for faithful // status/body forwarding, which the typed wrapper would swallow. let mime = content_type(&headers); let mut opts = agent.opts().await; opts.extra_headers.push(( header::CONTENT_TYPE, HeaderValue::from_str(&mime) .map_err(|e| invalid_request(format!("invalid content-type header: {e}")))?, )); Ok(forward_xrpc( agent.send_with_opts(UploadBlob { body }, opts).await, ))}
// ===========================================================================// Public AppView reads (unauthenticated; forwarded to public.api.bsky.app)// ===========================================================================
pub(super) async fn identity_resolve_handle( State(state): State<AppState>, ExtractXrpc(req): ExtractXrpc<ResolveHandleRequest>,) -> AppResult<Response> { Ok(forward_xrpc(state.bsky.send(req).await))}
pub(super) async fn actor_get_profile( State(state): State<AppState>, ExtractXrpc(req): ExtractXrpc<polymodel_api::app_bsky::actor::get_profile::GetProfileRequest>,) -> AppResult<Response> { Ok(forward_xrpc(state.bsky.send(req).await))}
// ===========================================================================// Faithful upstream forwarding + helpers// ===========================================================================
/// Forward an upstream `agent.send` result as a raw Axum response, preserving/// the real HTTP status and body rather than collapsing it to a single code:/// - `Ok` (success **or** a 400/401 XRPC-error body) → that status + the buffer./// - `Err(Http)` (other non-2xx, e.g. 409/404/429/5xx) → that status + body./// - `Err(Auth)` → 401./// - anything else (transport/decode/…) → 500 (genuine internal failure).////// The typed `XrpcError` enum carries no status; the HTTP status comes from the/// response/error itself, so atproto error codes (`RecordNotFound` on HTTP 400,/// `InvalidSwap` on HTTP 409, …) are forwarded verbatim.pub(super) fn forward_xrpc<Resp: XrpcResp>( result: Result<XrpcResponse<Resp>, ClientError>,) -> Response { match result { Ok(resp) => raw_response(resp.status(), resp.buffer().to_vec()), Err(err) => match err.kind() { ClientErrorKind::Http { status } => { let body = err .source_err() .and_then(|s| s.downcast_ref::<HttpError>()) .and_then(|h| h.body.clone()) .map(|b| b.to_vec()) .unwrap_or_default(); raw_response(*status, body) } ClientErrorKind::Auth(_) => { AppError::Unauthorized("upstream authentication required".to_string()) .into_response() } other => { tracing::error!(error = ?other, "passthrough upstream transport/internal failure"); AppError::Internal("passthrough upstream failure".to_string()).into_response() } }, }}
fn raw_response(status: StatusCode, body: Vec<u8>) -> Response { Response::builder() .status(status) .header(header::CONTENT_TYPE, "application/json") .body(Body::from(body)) .expect("static response parts are always valid")}
async fn actor_did(agent: &UserAgent) -> AppResult<Did> { agent .session_info() .await .map(|(did, _)| did) .ok_or_else(|| unauthorized("OAuth session is required"))}
/// Inbound Content-Type, trimmed of parameters, defaulting to binary.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()}
/// Eagerly project a successful proxied create/update into the local SQLite/// projection (rkey/cid read from the PDS output) for read-your-writes.async fn eager_project( state: &AppState, did: &Did, collection: &str, action: &str, record: &Value, output_buffer: &[u8],) { let Some((rkey, cid)) = parse_record_ref(output_buffer) else { return; }; let record_data = match to_data(record) { Ok(d) => d, Err(e) => { tracing::warn!(error = %e, %collection, "proxied record to_data failed"); return; } }; let event = ProjectionEvent { seq: 0, did: did.as_ref().to_string(), collection: collection.to_string(), rkey, action: action.to_string(), record: Some(record_data), cid: Some(cid), }; // Serialize against PM-28's decision-making writes (which read local state // then mutate): the proxied write itself makes no decision, but its local // projection must not interleave with a locked read-then-write section. The // lock guards only the local projection, never the upstream PDS call. let _guard = state.write_lock.lock().await; if let Err(e) = project_eager_record(&state.pool, &event).await { tracing::error!(error = %e, %collection, %action, "eager projection of proxied write failed"); }}
/// Parse `(rkey, cid)` from a createRecord/putRecord output's raw JSON.fn parse_record_ref(output: &[u8]) -> Option<(String, String)> { let out: Value = serde_json::from_slice(output).ok()?; let uri = out.get("uri")?.as_str()?; let cid = out.get("cid")?.as_str()?.to_string(); let rkey = uri.rsplit('/').next()?.to_string(); Some((rkey, cid))}
/// Read-your-writes override for a single `space.polymodel.*` getRecord.////// Returns a replacement response when a not-yet-firehose-confirmed local op/// should win over the PDS/AppView.////// - A pending delete → `RecordNotFound` (HTTP 400)./// - A pending write the PDS has not caught up to → the local value/cid (200)./// "Not caught up" means `pds_cid` is `None` (the PDS reported not-found or/// errored) or differs from the pending cid.////// Returns `None` otherwise (no pending state, or a pending write whose cid/// already matches the PDS) so the caller forwards the PDS result.pub(super) async fn try_get_record_override( pool: &SqlitePool, uri: &str, pds_cid: Option<&str>,) -> AppResult<Option<Response>> { let pending_delete = db(sqlx::query!( "SELECT 1 AS hit FROM pending_deletes WHERE uri = ?", uri ) .fetch_optional(pool) .await)?; if pending_delete.is_some() { return Ok(Some(record_not_found_response())); }
let pending_write = db(sqlx::query_as!( PendingWrite, "SELECT cid, value FROM pending_writes WHERE uri = ?", uri ) .fetch_optional(pool) .await)?; if let Some(pw) = pending_write && pds_cid != Some(pw.cid.as_str()) { let value: Value = serde_json::from_str(&pw.value).unwrap_or(Value::Null); let body = serde_json::to_vec(&serde_json::json!({ "uri": uri, "cid": pw.cid, "value": value, })) .unwrap_or_default(); return Ok(Some(raw_response(StatusCode::OK, body))); } Ok(None)}
#[derive(sqlx::FromRow)]struct PendingWrite { cid: String, value: String,}
fn record_not_found_response() -> Response { // atproto repo.getRecord RecordNotFound is HTTP 400 with this body shape. let body = serde_json::to_vec(&serde_json::json!({ "error": "RecordNotFound", "message": "Record not found", })) .unwrap_or_default(); raw_response(StatusCode::BAD_REQUEST, body)}