goat.navy
forked from
tranquil.farm/tranquil-pds
Our Personal Data Server from scratch!
4.4 kB
136 lines
1use axum::{
2 Json,
3 body::Body,
4 extract::State,
5 http::StatusCode,
6 http::header,
7 response::{IntoResponse, Response},
8};
9use serde::{Deserialize, Serialize};
10use tracing::error;
11use tranquil_pds::api::error::ApiError;
12use tranquil_pds::api::query::XrpcQuery;
13use tranquil_pds::state::AppState;
14use tranquil_pds::sync::util::{RepoAccessLevel, assert_repo_availability};
15use tranquil_types::{CidLink, Did, Tid};
16
17#[derive(Deserialize)]
18pub struct GetBlobParams {
19 pub did: Did,
20 pub cid: CidLink,
21}
22
23pub async fn get_blob(
24 State(state): State<AppState>,
25 XrpcQuery(params): XrpcQuery<GetBlobParams>,
26) -> Response {
27 let did = params.did;
28 let cid = params.cid;
29
30 let _account =
31 match assert_repo_availability(state.repos.repo.as_ref(), &did, RepoAccessLevel::Public)
32 .await
33 {
34 Ok(a) => a,
35 Err(e) => return e.into_response(),
36 };
37
38 let blob_result = state.repos.blob.get_blob_metadata(&cid).await;
39 match blob_result {
40 Ok(Some(metadata)) => match state.blob_store.get(&metadata.storage_key).await {
41 Ok(data) => Response::builder()
42 .status(StatusCode::OK)
43 .header(header::CONTENT_TYPE, &metadata.mime_type)
44 .header(header::CONTENT_LENGTH, metadata.size_bytes.to_string())
45 .header("x-content-type-options", "nosniff")
46 .header("content-security-policy", "default-src 'none'; sandbox")
47 .body(Body::from(data))
48 .unwrap_or_else(|_| ApiError::InternalError(None).into_response()),
49 Err(e) => {
50 error!("Failed to fetch blob from storage: {:?}", e);
51 ApiError::BlobNotFound(Some("Blob not found in storage".into())).into_response()
52 }
53 },
54 Ok(None) => ApiError::BlobNotFound(Some("Blob not found".into())).into_response(),
55 Err(e) => {
56 error!("DB error in get_blob: {:?}", e);
57 ApiError::InternalError(Some(format!("Database error: {}", e))).into_response()
58 }
59 }
60}
61
62#[derive(Deserialize)]
63pub struct ListBlobsParams {
64 pub did: Did,
65 pub since: Option<Tid>,
66 pub limit: Option<i64>,
67 pub cursor: Option<String>,
68}
69
70#[derive(Serialize)]
71pub struct ListBlobsOutput {
72 #[serde(skip_serializing_if = "Option::is_none")]
73 pub cursor: Option<CidLink>,
74 pub cids: Vec<CidLink>,
75}
76
77pub async fn list_blobs(
78 State(state): State<AppState>,
79 XrpcQuery(params): XrpcQuery<ListBlobsParams>,
80) -> Response {
81 let did = params.did;
82
83 let account =
84 match assert_repo_availability(state.repos.repo.as_ref(), &did, RepoAccessLevel::Public)
85 .await
86 {
87 Ok(a) => a,
88 Err(e) => return e.into_response(),
89 };
90
91 let limit = params.limit.unwrap_or(500).clamp(1, 1000);
92 let cursor_cid = params.cursor.as_deref().unwrap_or("");
93 let user_id = account.user_id;
94
95 let cids_result: Result<Vec<CidLink>, _> = if let Some(since) = ¶ms.since {
96 state
97 .repos
98 .blob
99 .list_blobs_since_rev(&did, since)
100 .await
101 .map(|cids| {
102 let mut cids = cids;
103 cids.sort();
104 cids.into_iter()
105 .filter(|c| c.as_str() > cursor_cid)
106 .take(usize::try_from(limit + 1).unwrap_or(0))
107 .collect()
108 })
109 } else {
110 state
111 .repos
112 .blob
113 .list_blobs_by_user(user_id, Some(cursor_cid), limit + 1)
114 .await
115 };
116 match cids_result {
117 Ok(cids) => {
118 let limit_usize = usize::try_from(limit).unwrap_or(0);
119 let has_more = cids.len() > limit_usize;
120 let cids: Vec<CidLink> = cids.into_iter().take(limit_usize).collect();
121 let next_cursor = if has_more { cids.last().cloned() } else { None };
122 (
123 StatusCode::OK,
124 Json(ListBlobsOutput {
125 cursor: next_cursor,
126 cids,
127 }),
128 )
129 .into_response()
130 }
131 Err(e) => {
132 error!("DB error in list_blobs: {:?}", e);
133 ApiError::InternalError(Some(format!("Database error: {}", e))).into_response()
134 }
135 }
136}