Something went wrong. Try again.
A minimal atproto Personal Data Server, on SQLite. github.com/eth0net/manapds
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238//! Blobs: taking one in, handing it back, and listing what records name.
use std::sync::Arc;
use axum::{ Json, body::Bytes, extract::{State, rejection::BytesRejection}, http::{HeaderMap, header}, response::{IntoResponse, Response},};use serde::{Deserialize, Serialize};
use crate::account::Manager;use crate::server::parse;use crate::store;use crate::xrpc::{self, Params, auth::Access};
/// What an upload is taken as when the caller says nothing.const UNSAID: &str = "application/octet-stream";
/// What a page holds when the caller does not ask for a size.const PAGE: u32 = 500;
/// The most any page holds, however many are asked for.const LONGEST_PAGE: u32 = 1000;
/// Which blob to serve, and whose.#[derive(Debug, Deserialize)]pub(crate) struct Wanted { did: String, cid: String,}
/// Whose blobs to page through, and from where.#[derive(Debug, Deserialize)]pub(crate) struct Listing { did: String, since: Option<String>, limit: Option<u32>, cursor: Option<String>,}
/// One page of the blobs an account's records name.#[derive(Debug, Serialize)]pub(crate) struct Page { #[serde(skip_serializing_if = "Option::is_none")] cursor: Option<String>, cids: Vec<String>,}
/// Which page of the blobs nothing uploaded to ask for.#[derive(Debug, Deserialize)]pub(crate) struct Missing { limit: Option<u32>, cursor: Option<String>,}
/// One page of the blobs a record names that nothing uploaded.#[derive(Debug, Serialize)]pub(crate) struct MissingPage { #[serde(skip_serializing_if = "Option::is_none")] cursor: Option<String>, blobs: Vec<RecordBlob>,}
/// One of them, beside a record naming it.#[derive(Debug, Serialize)]#[serde(rename_all = "camelCase")]pub(crate) struct RecordBlob { cid: String, record_uri: String,}
/// A blob as a record refers to one.#[derive(Debug, Serialize)]#[serde(rename_all = "camelCase")]pub(crate) struct Reference { #[serde(rename = "$type")] kind: &'static str, #[serde(rename = "ref")] link: Link, mime_type: String, size: u64,}
/// The CID inside one, in the shape JSON writes a link.#[derive(Debug, Serialize)]pub(crate) struct Link { #[serde(rename = "$link")] link: String,}
/// What an upload answers with.#[derive(Debug, Serialize)]pub(crate) struct Uploaded { blob: Reference,}
/// `com.atproto.repo.uploadBlob`////// # Errors////// If the account holds no repository, or storage will not answer.pub(crate) async fn upload( State(accounts): State<Arc<Manager>>, access: Access, headers: HeaderMap, body: Result<Bytes, BytesRejection>,) -> xrpc::Result<Json<Uploaded>> { // The only route whose body is read whole, so the only one whose refusal // does not already come back in the XRPC shape. let body = body.map_err(|rejection| { xrpc::Error::new(xrpc::Status::PayloadTooLarge).saying(rejection.body_text()) })?; let mime = headers .get(header::CONTENT_TYPE) .and_then(|mime| mime.to_str().ok()) // A charset says nothing about what the bytes are, and what a record // holds has to match the type an appview's lexicon will accept. .map(|mime| mime.split(';').next().unwrap_or(mime).trim()) .filter(|mime| !mime.is_empty()) .unwrap_or(UNSAID) .to_owned(); let blob = accounts .upload_blob(&access.did, mime, body.to_vec()) .await?; Ok(Json(Uploaded { blob: shown(&blob) }))}
/// `com.atproto.sync.getBlob`////// # Errors////// If nothing answers to the identifier, it uploaded no such blob, or storage/// will not answer.pub(crate) async fn get( State(accounts): State<Arc<Manager>>, Params(query): Params<Wanted>,) -> xrpc::Result<Response> { // The lexicon asks for a DID here rather than a handle, so nothing is // resolved: a blob is served by whoever holds it or by nobody. let did = parse::did(&query.did)?; let cid = parse::cid(&query.cid)?; let (blob, bytes) = accounts .blob(&did, &cid) .await? .ok_or_else(|| xrpc::Error::invalid_request("Blob not found").named("BlobNotFound"))?;
Ok(( [ (header::CONTENT_TYPE, blob.mime), // A blob is addressed by its bytes, so nothing it is served under // can ever be a different blob. ( header::CACHE_CONTROL, "public, max-age=31536000, immutable".to_owned(), ), // Anyone's bytes under this server's origin, so a browser is told // three times over not to run them. (header::X_CONTENT_TYPE_OPTIONS, "nosniff".to_owned()), (header::CONTENT_DISPOSITION, "attachment".to_owned()), ( header::CONTENT_SECURITY_POLICY, "default-src 'none'; sandbox".to_owned(), ), ], bytes, ) .into_response())}
/// `com.atproto.sync.listBlobs`////// # Errors////// If the identifier is not a DID or `since` is not a revision, nothing/// answers to the identifier, or storage will not answer.pub(crate) async fn list( State(accounts): State<Arc<Manager>>, Params(query): Params<Listing>,) -> xrpc::Result<Json<Page>> { let did = parse::did(&query.did)?; let since = query.since.as_deref().map(parse::since).transpose()?; let limit = query.limit.unwrap_or(PAGE).clamp(1, LONGEST_PAGE); let cids: Vec<String> = accounts .blobs(&did, since, limit, query.cursor) .await? .iter() .map(ToString::to_string) .collect();
// A short page carries no cursor, which is what stops a client asking. let cursor = (u32::try_from(cids.len()).unwrap_or(u32::MAX) == limit) .then(|| cids.last().cloned()) .flatten(); Ok(Json(Page { cursor, cids }))}
/// `com.atproto.repo.listMissingBlobs`////// # Errors////// If the caller holds no repository, or storage will not answer.pub(crate) async fn list_missing( State(accounts): State<Arc<Manager>>, access: Access, Params(query): Params<Missing>,) -> xrpc::Result<Json<MissingPage>> { let limit = query.limit.unwrap_or(PAGE).clamp(1, LONGEST_PAGE); let blobs: Vec<RecordBlob> = accounts .missing_blobs(&access.did, limit, query.cursor) .await? .into_iter() .map(|missing| RecordBlob { cid: missing.cid.to_string(), record_uri: missing.uri, }) .collect();
let cursor = (u32::try_from(blobs.len()).unwrap_or(u32::MAX) == limit) .then(|| blobs.last().map(|last| last.cid.clone())) .flatten(); Ok(Json(MissingPage { cursor, blobs }))}
/// One stored blob, in the shape a record would name it.fn shown(blob: &store::Blob) -> Reference { Reference { kind: "blob", link: Link { link: blob.cid.to_string(), }, mime_type: blob.mime.clone(), size: blob.size, }}