this repo has no description
lewis.moe
1use crate::api::ApiError;
2use crate::api::proxy_client::{
3 MAX_RESPONSE_SIZE, RESPONSE_HEADERS_TO_FORWARD, is_ssrf_safe, proxy_client,
4};
5use crate::state::AppState;
6use axum::{
7 Json,
8 http::{HeaderMap, HeaderValue, StatusCode},
9 response::{IntoResponse, Response},
10};
11use bytes::Bytes;
12use chrono::{DateTime, Utc};
13use cid::Cid;
14use jacquard_repo::storage::BlockStore;
15use serde::{Deserialize, Serialize};
16use serde_json::Value;
17use std::collections::HashMap;
18use tracing::{error, info, warn};
19use uuid::Uuid;
20
21pub const REPO_REV_HEADER: &str = "atproto-repo-rev";
22pub const UPSTREAM_LAG_HEADER: &str = "atproto-upstream-lag";
23
24#[derive(Debug, Clone, Serialize, Deserialize)]
25#[serde(rename_all = "camelCase")]
26pub struct PostRecord {
27 #[serde(rename = "$type")]
28 pub record_type: Option<String>,
29 pub text: String,
30 pub created_at: String,
31 #[serde(skip_serializing_if = "Option::is_none")]
32 pub reply: Option<Value>,
33 #[serde(skip_serializing_if = "Option::is_none")]
34 pub embed: Option<Value>,
35 #[serde(skip_serializing_if = "Option::is_none")]
36 pub langs: Option<Vec<String>>,
37 #[serde(skip_serializing_if = "Option::is_none")]
38 pub labels: Option<Value>,
39 #[serde(skip_serializing_if = "Option::is_none")]
40 pub tags: Option<Vec<String>>,
41 #[serde(flatten)]
42 pub extra: HashMap<String, Value>,
43}
44
45#[derive(Debug, Clone, Serialize, Deserialize)]
46#[serde(rename_all = "camelCase")]
47pub struct ProfileRecord {
48 #[serde(rename = "$type")]
49 pub record_type: Option<String>,
50 #[serde(skip_serializing_if = "Option::is_none")]
51 pub display_name: Option<String>,
52 #[serde(skip_serializing_if = "Option::is_none")]
53 pub description: Option<String>,
54 #[serde(skip_serializing_if = "Option::is_none")]
55 pub avatar: Option<Value>,
56 #[serde(skip_serializing_if = "Option::is_none")]
57 pub banner: Option<Value>,
58 #[serde(flatten)]
59 pub extra: HashMap<String, Value>,
60}
61
62#[derive(Debug, Clone)]
63pub struct RecordDescript<T> {
64 pub uri: String,
65 pub cid: String,
66 pub indexed_at: DateTime<Utc>,
67 pub record: T,
68}
69
70#[derive(Debug, Clone, Serialize, Deserialize)]
71#[serde(rename_all = "camelCase")]
72pub struct LikeRecord {
73 #[serde(rename = "$type")]
74 pub record_type: Option<String>,
75 pub subject: LikeSubject,
76 pub created_at: String,
77 #[serde(flatten)]
78 pub extra: HashMap<String, Value>,
79}
80
81#[derive(Debug, Clone, Serialize, Deserialize)]
82#[serde(rename_all = "camelCase")]
83pub struct LikeSubject {
84 pub uri: String,
85 pub cid: String,
86}
87
88#[derive(Debug, Default)]
89pub struct LocalRecords {
90 pub count: usize,
91 pub profile: Option<RecordDescript<ProfileRecord>>,
92 pub posts: Vec<RecordDescript<PostRecord>>,
93 pub likes: Vec<RecordDescript<LikeRecord>>,
94}
95
96pub async fn get_records_since_rev(
97 state: &AppState,
98 did: &str,
99 rev: &str,
100) -> Result<LocalRecords, String> {
101 let mut result = LocalRecords::default();
102 let user_id: Uuid = sqlx::query_scalar!("SELECT id FROM users WHERE did = $1", did)
103 .fetch_optional(&state.db)
104 .await
105 .map_err(|e| format!("DB error: {}", e))?
106 .ok_or_else(|| "User not found".to_string())?;
107 let rows = sqlx::query!(
108 r#"
109 SELECT record_cid, collection, rkey, created_at, repo_rev
110 FROM records
111 WHERE repo_id = $1 AND repo_rev > $2
112 ORDER BY repo_rev ASC
113 LIMIT 10
114 "#,
115 user_id,
116 rev
117 )
118 .fetch_all(&state.db)
119 .await
120 .map_err(|e| format!("DB error fetching records: {}", e))?;
121 if rows.is_empty() {
122 return Ok(result);
123 }
124 let sanity_check = sqlx::query_scalar!(
125 "SELECT 1 as val FROM records WHERE repo_id = $1 AND repo_rev <= $2 LIMIT 1",
126 user_id,
127 rev
128 )
129 .fetch_optional(&state.db)
130 .await
131 .map_err(|e| format!("DB error sanity check: {}", e))?;
132 if sanity_check.is_none() {
133 warn!("Sanity check failed: no records found before rev {}", rev);
134 return Ok(result);
135 }
136 struct RowData {
137 cid_str: String,
138 collection: String,
139 rkey: String,
140 created_at: DateTime<Utc>,
141 }
142 let mut row_data: Vec<RowData> = Vec::with_capacity(rows.len());
143 let mut cids: Vec<Cid> = Vec::with_capacity(rows.len());
144 for row in &rows {
145 if let Ok(cid) = row.record_cid.parse::<Cid>() {
146 cids.push(cid);
147 row_data.push(RowData {
148 cid_str: row.record_cid.clone(),
149 collection: row.collection.clone(),
150 rkey: row.rkey.clone(),
151 created_at: row.created_at,
152 });
153 }
154 }
155 let blocks: Vec<Option<Bytes>> = state
156 .block_store
157 .get_many(&cids)
158 .await
159 .map_err(|e| format!("Error fetching blocks: {}", e))?;
160 for (data, block_opt) in row_data.into_iter().zip(blocks.into_iter()) {
161 let block_bytes = match block_opt {
162 Some(b) => b,
163 None => continue,
164 };
165 result.count += 1;
166 let uri = format!("at://{}/{}/{}", did, data.collection, data.rkey);
167 if data.collection == "app.bsky.actor.profile" && data.rkey == "self" {
168 if let Ok(record) = serde_ipld_dagcbor::from_slice::<ProfileRecord>(&block_bytes) {
169 result.profile = Some(RecordDescript {
170 uri,
171 cid: data.cid_str,
172 indexed_at: data.created_at,
173 record,
174 });
175 }
176 } else if data.collection == "app.bsky.feed.post" {
177 if let Ok(record) = serde_ipld_dagcbor::from_slice::<PostRecord>(&block_bytes) {
178 result.posts.push(RecordDescript {
179 uri,
180 cid: data.cid_str,
181 indexed_at: data.created_at,
182 record,
183 });
184 }
185 } else if data.collection == "app.bsky.feed.like"
186 && let Ok(record) = serde_ipld_dagcbor::from_slice::<LikeRecord>(&block_bytes) {
187 result.likes.push(RecordDescript {
188 uri,
189 cid: data.cid_str,
190 indexed_at: data.created_at,
191 record,
192 });
193 }
194 }
195 Ok(result)
196}
197
198pub fn get_local_lag(local: &LocalRecords) -> Option<i64> {
199 let mut oldest: Option<DateTime<Utc>> = local.profile.as_ref().map(|p| p.indexed_at);
200 for post in &local.posts {
201 match oldest {
202 None => oldest = Some(post.indexed_at),
203 Some(o) if post.indexed_at < o => oldest = Some(post.indexed_at),
204 _ => {}
205 }
206 }
207 for like in &local.likes {
208 match oldest {
209 None => oldest = Some(like.indexed_at),
210 Some(o) if like.indexed_at < o => oldest = Some(like.indexed_at),
211 _ => {}
212 }
213 }
214 oldest.map(|o| (Utc::now() - o).num_milliseconds())
215}
216
217pub fn extract_repo_rev(headers: &HeaderMap) -> Option<String> {
218 headers
219 .get(REPO_REV_HEADER)
220 .and_then(|h| h.to_str().ok())
221 .map(|s| s.to_string())
222}
223
224#[derive(Debug)]
225pub struct ProxyResponse {
226 pub status: StatusCode,
227 pub headers: HeaderMap,
228 pub body: bytes::Bytes,
229}
230
231impl ProxyResponse {
232 pub fn into_response(self) -> Response {
233 let mut response = Response::builder().status(self.status);
234 for (key, value) in self.headers.iter() {
235 response = response.header(key, value);
236 }
237 response.body(axum::body::Body::from(self.body)).unwrap()
238 }
239}
240
241pub async fn proxy_to_appview(
242 method: &str,
243 params: &HashMap<String, String>,
244 auth_did: &str,
245 auth_key_bytes: Option<&[u8]>,
246) -> Result<ProxyResponse, Response> {
247 let appview_url = std::env::var("APPVIEW_URL").map_err(|_| {
248 ApiError::UpstreamUnavailable("No upstream AppView configured".to_string()).into_response()
249 })?;
250 if let Err(e) = is_ssrf_safe(&appview_url) {
251 error!("SSRF check failed for appview URL: {}", e);
252 return Err(
253 ApiError::UpstreamUnavailable(format!("Invalid upstream URL: {}", e)).into_response(),
254 );
255 }
256 let target_url = format!("{}/xrpc/{}", appview_url, method);
257 info!(target = %target_url, "Proxying request to appview");
258 let client = proxy_client();
259 let mut request_builder = client.get(&target_url).query(params);
260 if let Some(key_bytes) = auth_key_bytes {
261 let appview_did =
262 std::env::var("APPVIEW_DID").unwrap_or_else(|_| "did:web:api.bsky.app".to_string());
263 match crate::auth::create_service_token(auth_did, &appview_did, method, key_bytes) {
264 Ok(service_token) => {
265 request_builder =
266 request_builder.header("Authorization", format!("Bearer {}", service_token));
267 }
268 Err(e) => {
269 error!(error = ?e, "Failed to create service token");
270 return Err(ApiError::InternalError.into_response());
271 }
272 }
273 }
274 match request_builder.send().await {
275 Ok(resp) => {
276 let status =
277 StatusCode::from_u16(resp.status().as_u16()).unwrap_or(StatusCode::BAD_GATEWAY);
278 let headers: HeaderMap = resp
279 .headers()
280 .iter()
281 .filter(|(k, _)| {
282 RESPONSE_HEADERS_TO_FORWARD
283 .iter()
284 .any(|h| k.as_str().eq_ignore_ascii_case(h))
285 })
286 .filter_map(|(k, v)| {
287 let name = axum::http::HeaderName::try_from(k.as_str()).ok()?;
288 let value = HeaderValue::from_bytes(v.as_bytes()).ok()?;
289 Some((name, value))
290 })
291 .collect();
292 let content_length = resp.content_length().unwrap_or(0);
293 if content_length > MAX_RESPONSE_SIZE {
294 error!(
295 content_length,
296 max = MAX_RESPONSE_SIZE,
297 "Upstream response too large"
298 );
299 return Err(ApiError::UpstreamFailure.into_response());
300 }
301 let body = resp.bytes().await.map_err(|e| {
302 error!(error = ?e, "Error reading proxy response body");
303 ApiError::UpstreamFailure.into_response()
304 })?;
305 if body.len() as u64 > MAX_RESPONSE_SIZE {
306 error!(
307 len = body.len(),
308 max = MAX_RESPONSE_SIZE,
309 "Upstream response body exceeded size limit"
310 );
311 return Err(ApiError::UpstreamFailure.into_response());
312 }
313 Ok(ProxyResponse {
314 status,
315 headers,
316 body,
317 })
318 }
319 Err(e) => {
320 error!(error = ?e, "Error sending proxy request");
321 if e.is_timeout() {
322 Err(ApiError::UpstreamTimeout.into_response())
323 } else if e.is_connect() {
324 Err(
325 ApiError::UpstreamUnavailable("Failed to connect to upstream".to_string())
326 .into_response(),
327 )
328 } else {
329 Err(ApiError::UpstreamFailure.into_response())
330 }
331 }
332 }
333}
334
335pub fn format_munged_response<T: Serialize>(data: T, lag: Option<i64>) -> Response {
336 let mut response = (StatusCode::OK, Json(data)).into_response();
337 if let Some(lag_ms) = lag
338 && let Ok(header_val) = HeaderValue::from_str(&lag_ms.to_string()) {
339 response
340 .headers_mut()
341 .insert(UPSTREAM_LAG_HEADER, header_val);
342 }
343 response
344}
345
346#[derive(Debug, Clone, Serialize, Deserialize)]
347#[serde(rename_all = "camelCase")]
348pub struct AuthorView {
349 pub did: String,
350 pub handle: String,
351 #[serde(skip_serializing_if = "Option::is_none")]
352 pub display_name: Option<String>,
353 #[serde(skip_serializing_if = "Option::is_none")]
354 pub avatar: Option<String>,
355 #[serde(flatten)]
356 pub extra: HashMap<String, Value>,
357}
358
359#[derive(Debug, Clone, Serialize, Deserialize)]
360#[serde(rename_all = "camelCase")]
361pub struct PostView {
362 pub uri: String,
363 pub cid: String,
364 pub author: AuthorView,
365 pub record: Value,
366 pub indexed_at: String,
367 #[serde(skip_serializing_if = "Option::is_none")]
368 pub embed: Option<Value>,
369 #[serde(default)]
370 pub reply_count: i64,
371 #[serde(default)]
372 pub repost_count: i64,
373 #[serde(default)]
374 pub like_count: i64,
375 #[serde(default)]
376 pub quote_count: i64,
377 #[serde(flatten)]
378 pub extra: HashMap<String, Value>,
379}
380
381#[derive(Debug, Clone, Serialize, Deserialize)]
382#[serde(rename_all = "camelCase")]
383pub struct FeedViewPost {
384 pub post: PostView,
385 #[serde(skip_serializing_if = "Option::is_none")]
386 pub reply: Option<Value>,
387 #[serde(skip_serializing_if = "Option::is_none")]
388 pub reason: Option<Value>,
389 #[serde(skip_serializing_if = "Option::is_none")]
390 pub feed_context: Option<String>,
391 #[serde(flatten)]
392 pub extra: HashMap<String, Value>,
393}
394
395#[derive(Debug, Clone, Serialize, Deserialize)]
396pub struct FeedOutput {
397 pub feed: Vec<FeedViewPost>,
398 #[serde(skip_serializing_if = "Option::is_none")]
399 pub cursor: Option<String>,
400}
401
402pub fn format_local_post(
403 descript: &RecordDescript<PostRecord>,
404 author_did: &str,
405 author_handle: &str,
406 profile: Option<&RecordDescript<ProfileRecord>>,
407) -> PostView {
408 let display_name = profile.and_then(|p| p.record.display_name.clone());
409 PostView {
410 uri: descript.uri.clone(),
411 cid: descript.cid.clone(),
412 author: AuthorView {
413 did: author_did.to_string(),
414 handle: author_handle.to_string(),
415 display_name,
416 avatar: None,
417 extra: HashMap::new(),
418 },
419 record: serde_json::to_value(&descript.record).unwrap_or(Value::Null),
420 indexed_at: descript.indexed_at.to_rfc3339(),
421 embed: descript.record.embed.clone(),
422 reply_count: 0,
423 repost_count: 0,
424 like_count: 0,
425 quote_count: 0,
426 extra: HashMap::new(),
427 }
428}
429
430pub fn insert_posts_into_feed(feed: &mut Vec<FeedViewPost>, posts: Vec<PostView>) {
431 if posts.is_empty() {
432 return;
433 }
434 let new_items: Vec<FeedViewPost> = posts
435 .into_iter()
436 .map(|post| FeedViewPost {
437 post,
438 reply: None,
439 reason: None,
440 feed_context: None,
441 extra: HashMap::new(),
442 })
443 .collect();
444 feed.extend(new_items);
445 feed.sort_by(|a, b| b.post.indexed_at.cmp(&a.post.indexed_at));
446}