diff --git a/src/main.rs b/src/main.rs index 4e05330..38104d7 100644 --- a/src/main.rs +++ b/src/main.rs @@ -10,7 +10,7 @@ use hydrant::control::Hydrant; use hydrant::deps::futures::StreamExt; use jacquard_common::IntoStatic; use jacquard_common::types::ident::AtIdentifier; -use serde::{Deserialize, Serialize}; +use serde::Deserialize; use tracing; use tracing_subscriber::EnvFilter; @@ -23,6 +23,16 @@ struct AppState { mutes: fjall::Keyspace, notifications: fjall::Keyspace, seen: fjall::Keyspace, + cdn_url: String, +} + +impl AppState { + fn cdn(&self, category: &str, did: &str, link: &str) -> String { + format!( + "{}/img/{}/plain/{}/{}@jpeg", + self.cdn_url, category, did, link + ) + } } #[tokio::main] @@ -80,6 +90,8 @@ async fn main() -> miette::Result<()> { "app.bsky.labeler.service".to_string(), ]); + let cdn_url = std::env::var("CDN_URL").unwrap_or_else(|_| "https://cdn.bsky.app".to_string()); + let hydrant = Hydrant::new(cfg).await?; let hydrant_clone = hydrant.clone(); let app_state = AppState { @@ -90,6 +102,7 @@ async fn main() -> miette::Result<()> { mutes, notifications, seen, + cdn_url, }; if let Ok(seed_account) = std::env::var("SEED_ACCOUNT") { @@ -435,7 +448,7 @@ async fn get_post_view( } } - Ok(serde_json::json!({ + let mut post_view = serde_json::json!({ "uri": uri_str, "cid": record.cid.to_string(), "author": author_profile, @@ -445,7 +458,68 @@ async fn get_post_view( "likeCount": app_state.hydrant.backlinks.count(uri_str.to_string()).source("app.bsky.feed.like").run().await.unwrap_or(0), "indexedAt": chrono::Utc::now().to_rfc3339(), "viewer": viewer_state, - })) + "labels": [], + }); + + let val_json = serde_json::to_value(&record.value).unwrap_or(serde_json::json!({})); + if let Some(record_embed) = val_json.get("embed") { + let t = record_embed + .get("$type") + .and_then(|v| v.as_str()) + .unwrap_or(""); + if t == "app.bsky.embed.images" { + if let Some(images) = record_embed.get("images").and_then(|v| v.as_array()) { + let mut view_images = Vec::new(); + for img in images { + let link = img + .get("image") + .and_then(|v| v.get("ref")) + .and_then(|v| v.get("$link")) + .and_then(|v| v.as_str()) + .unwrap_or(""); + let thumb = app_state.cdn("feed_thumbnail", repo.did.as_str(), link); + let fullsize = app_state.cdn("feed_fullsize", repo.did.as_str(), link); + view_images.push(serde_json::json!({ + "thumb": thumb, + "fullsize": fullsize, + "alt": img.get("alt").and_then(|v| v.as_str()).unwrap_or(""), + "aspectRatio": img.get("aspectRatio"), + })); + } + post_view["embed"] = serde_json::json!({ + "$type": "app.bsky.embed.images#view", + "images": view_images, + }); + } + } else if t == "app.bsky.embed.record" { + if let Some(quoted_uri) = record_embed + .get("record") + .and_then(|v| v.get("uri")) + .and_then(|v| v.as_str()) + { + // For now, let's just do a shallow hydration of the quoted post to avoid infinite recursion + // We use None for viewer_did here to simplify and avoid auth-loops + if let Ok(quoted_post) = Box::pin(get_post_view(app_state, quoted_uri, None)).await + { + post_view["embed"] = serde_json::json!({ + "$type": "app.bsky.embed.record#view", + "record": { + "$type": "app.bsky.embed.record#viewRecord", + "uri": quoted_post["uri"], + "cid": quoted_post["cid"], + "author": quoted_post["author"], + "value": quoted_post["record"], + "labels": quoted_post["labels"], + "indexedAt": quoted_post["indexedAt"], + "embeds": quoted_post.get("embed").map(|e| vec![e]).unwrap_or_default(), + } + }); + } + } + } + } + + Ok(post_view) } async fn get_thread_view_post( @@ -550,9 +624,10 @@ async fn get_profile( StatusCode::BAD_REQUEST })?; + let viewer_did = get_auth_did(&req); match app_state.hydrant.repos.resolve(&ident).await { Ok(repo) => { - match get_profile_internal(&app_state, repo.did.as_str(), None).await { + match get_profile_internal(&app_state, repo.did.as_str(), viewer_did.as_deref()).await { Ok(profile) => return Ok(Json(profile).into_response()), Err(e) => { tracing::warn!("failed to get profile for {}: {e}", repo.did); @@ -698,6 +773,10 @@ async fn get_profile_internal( "postsCount": posts_count, "indexedAt": chrono::Utc::now().to_rfc3339(), "viewer": viewer_state, + "labels": [], + "associated": { + "chat": { "allowIncoming": "all" } + } }); if let Some(rec) = profile_record { @@ -711,10 +790,26 @@ async fn get_profile_internal( obj.insert("description".to_string(), description.clone()); } if let Some(avatar) = value.get("avatar") { - obj.insert("avatar".to_string(), avatar.clone()); + let link = avatar + .get("ref") + .and_then(|v| v.get("$link")) + .and_then(|v| v.as_str()) + .unwrap_or(""); + obj.insert( + "avatar".to_string(), + serde_json::json!(app_state.cdn("avatar", did_str, link)), + ); } if let Some(banner) = value.get("banner") { - obj.insert("banner".to_string(), banner.clone()); + let link = banner + .get("ref") + .and_then(|v| v.get("$link")) + .and_then(|v| v.as_str()) + .unwrap_or(""); + obj.insert( + "banner".to_string(), + serde_json::json!(app_state.cdn("banner", did_str, link)), + ); } } } @@ -758,49 +853,145 @@ async fn get_author_feed( let limit = params.limit.unwrap_or(50).min(100); - let record_list = match repo + // Get posts + let posts_list = match repo .list_records("app.bsky.feed.post", limit, true, params.cursor.as_deref()) .await { Ok(rl) => rl, Err(e) => { - tracing::error!("failed to list records for {}: {e}", did); + tracing::error!("failed to list posts for {}: {e}", did); return proxy_request(req).await; } }; - let author_profile = match get_profile_internal(&app_state, did.as_str(), None).await { - Ok(p) => p, + // Get reposts + let reposts_list = match repo + .list_records( + "app.bsky.feed.repost", + limit, + true, + params.cursor.as_deref(), + ) + .await + { + Ok(rl) => rl, Err(e) => { - tracing::error!("failed to get profile for {}: {e}", did); + tracing::error!("failed to list reposts for {}: {e}", did); return proxy_request(req).await; } }; + let viewer_did = get_auth_did(&req); + let author_profile = + match get_profile_internal(&app_state, did.as_str(), viewer_did.as_deref()).await { + Ok(p) => p, + Err(e) => { + tracing::error!("failed to get profile for {}: {e}", did); + return proxy_request(req).await; + } + }; + + let mut all_items = Vec::new(); + for rec in posts_list.records { + all_items.push((rec.rkey.as_str().to_string(), "post", rec)); + } + for rec in reposts_list.records { + all_items.push((rec.rkey.as_str().to_string(), "repost", rec)); + } + + // Sort by rkey (roughly chronological) descending + all_items.sort_by(|a, b| b.0.cmp(&a.0)); + all_items.truncate(limit); + let mut feed = Vec::new(); - for rec in record_list.records { - let uri = format!( - "at://{}/app.bsky.feed.post/{}", - did.as_str(), - rec.rkey.as_str() - ); - feed.push(serde_json::json!({ - "post": { - "uri": uri, - "cid": rec.cid.to_string(), - "author": author_profile.clone(), - "record": rec.value, - "replyCount": app_state.hydrant.backlinks.count(uri.clone()).source("app.bsky.feed.post").run().await.unwrap_or(0), - "repostCount": app_state.hydrant.backlinks.count(uri.clone()).source("app.bsky.feed.repost").run().await.unwrap_or(0), - "likeCount": app_state.hydrant.backlinks.count(uri.clone()).source("app.bsky.feed.like").run().await.unwrap_or(0), - "indexedAt": chrono::Utc::now().to_rfc3339(), + for (rkey, kind, rec) in &all_items { + if *kind == "post" { + let uri = format!("at://{}/app.bsky.feed.post/{}", did.as_str(), rkey); + let post = match get_post_view(&app_state, &uri, viewer_did.as_deref()).await { + Ok(mut p) => { + p["author"] = author_profile.clone(); + p + } + Err(_) => { + serde_json::json!({ + "uri": uri, + "cid": rec.cid.to_string(), + "author": author_profile.clone(), + "record": rec.value, + "replyCount": app_state.hydrant.backlinks.count(uri.clone()).source("app.bsky.feed.post").run().await.unwrap_or(0), + "repostCount": app_state.hydrant.backlinks.count(uri.clone()).source("app.bsky.feed.repost").run().await.unwrap_or(0), + "likeCount": app_state.hydrant.backlinks.count(uri.clone()).source("app.bsky.feed.like").run().await.unwrap_or(0), + "indexedAt": chrono::Utc::now().to_rfc3339(), + "labels": [], + }) + } + }; + + let val_json = serde_json::to_value(&rec.value).unwrap_or(serde_json::json!({})); + let mut feed_item = serde_json::json!({ "post": post }); + + // Add reply context if it's a reply + if let Some(reply) = val_json.get("reply") { + if let Some(parent_uri) = reply + .get("parent") + .and_then(|p| p.get("uri")) + .and_then(|u| u.as_str()) + { + if let Ok(parent_post) = + get_post_view(&app_state, parent_uri, viewer_did.as_deref()).await + { + let root_uri = reply + .get("root") + .and_then(|p| p.get("uri")) + .and_then(|u| u.as_str()) + .unwrap_or(parent_uri); + if let Ok(root_post) = + get_post_view(&app_state, root_uri, viewer_did.as_deref()).await + { + feed_item["reply"] = serde_json::json!({ + "root": root_post, + "parent": parent_post, + }); + } + } + } } - })); + + feed.push(feed_item); + } else if *kind == "repost" { + let val_json = serde_json::to_value(&rec.value).unwrap_or(serde_json::json!({})); + if let Some(subject_uri) = val_json + .get("subject") + .and_then(|s| s.get("uri")) + .and_then(|u| u.as_str()) + { + if let Ok(post_view) = + get_post_view(&app_state, subject_uri, viewer_did.as_deref()).await + { + feed.push(serde_json::json!({ + "post": post_view, + "reason": { + "$type": "app.bsky.feed.defs#reasonRepost", + "by": author_profile.clone(), + "indexedAt": val_json.get("createdAt").and_then(|c| c.as_str()).map(|s| s.to_string()).unwrap_or_else(|| chrono::Utc::now().to_rfc3339()), + } + })); + } + } + } } + let next_cursor = if posts_list.cursor.is_some() || reposts_list.cursor.is_some() { + // This is a naive cursor but works for now as long as we sort by rkey + all_items.last().map(|i| i.0.clone()) + } else { + None + }; + Ok(Json(serde_json::json!({ "feed": feed, - "cursor": record_list.cursor.map(|c| c.to_string()) + "cursor": next_cursor })) .into_response()) } @@ -853,6 +1044,7 @@ async fn get_likes( }; let mut likes = Vec::new(); + let viewer_did = get_auth_did(&req); for bl in backlinks_page.backlinks { let uri = match jacquard_common::types::string::AtUri::new(bl.uri.as_str()) { Ok(uri) => uri, @@ -863,13 +1055,16 @@ async fn get_likes( }; let author_ident = uri.authority(); - let profile = match get_profile_internal(&app_state, author_ident.as_str(), None).await { - Ok(p) => p, - Err(e) => { - tracing::warn!("failed to get profile for {author_ident}: {e}"); - continue; - } - }; + let profile = + match get_profile_internal(&app_state, author_ident.as_str(), viewer_did.as_deref()) + .await + { + Ok(p) => p, + Err(e) => { + tracing::warn!("failed to get profile for {author_ident}: {e}"); + continue; + } + }; let repo = match app_state.hydrant.repos.resolve(author_ident).await { Ok(repo) => repo, Err(e) => { @@ -956,6 +1151,7 @@ async fn get_reposted_by( }; let mut reposted_by = Vec::new(); + let viewer_did = get_auth_did(&req); for bl in backlinks_page.backlinks { let uri = match jacquard_common::types::string::AtUri::new(bl.uri.as_str()) { Ok(uri) => uri, @@ -965,13 +1161,16 @@ async fn get_reposted_by( } }; let author_ident = uri.authority(); - let profile = match get_profile_internal(&app_state, author_ident.as_str(), None).await { - Ok(p) => p, - Err(e) => { - tracing::warn!("failed to get profile for {author_ident}: {e}"); - continue; - } - }; + let profile = + match get_profile_internal(&app_state, author_ident.as_str(), viewer_did.as_deref()) + .await + { + Ok(p) => p, + Err(e) => { + tracing::warn!("failed to get profile for {author_ident}: {e}"); + continue; + } + }; reposted_by.push(profile); } @@ -992,12 +1191,9 @@ fn extract_query_array(query: &str, key: &str) -> Vec { let prefix1 = format!("{}=", key); let prefix2 = format!("{}[]=", key); for part in query.split('&') { - let mut val = None; - if part.starts_with(&prefix1) { - val = Some(&part[prefix1.len()..]); - } else if part.starts_with(&prefix2) { - val = Some(&part[prefix2.len()..]); - } + let val = part + .strip_prefix(&prefix1) + .or_else(|| part.strip_prefix(&prefix2)); if let Some(v) = val { if let Ok(decoded) = urlencoding::decode(v) { res.push(decoded.into_owned()); @@ -1018,6 +1214,7 @@ async fn get_profiles( return proxy_request(req).await; } + let viewer_did = get_auth_did(&req); let mut profiles = Vec::new(); for actor in actors { let ident = match AtIdentifier::new(&actor) { @@ -1029,7 +1226,9 @@ async fn get_profiles( }; match app_state.hydrant.repos.resolve(&ident).await { Ok(repo) => { - match get_profile_internal(&app_state, repo.did.as_str(), None).await { + match get_profile_internal(&app_state, repo.did.as_str(), viewer_did.as_deref()) + .await + { Ok(profile) => profiles.push(profile), Err(e) => tracing::warn!("failed to get profile for {}: {e}", repo.did), } @@ -1052,9 +1251,10 @@ async fn get_posts( return proxy_request(req).await; } + let viewer_did = get_auth_did(&req); let mut posts = Vec::new(); for uri in uris { - match get_post_view(&app_state, &uri, None).await { + match get_post_view(&app_state, &uri, viewer_did.as_deref()).await { Ok(post) => posts.push(post), Err(e) => tracing::warn!("failed to get post view for {uri}: {e}"), } @@ -1113,23 +1313,25 @@ async fn get_follows( }; let mut follows = Vec::new(); + let viewer_did = get_auth_did(&req); for rec in record_list.records { let value = serde_json::to_value(&rec.value).unwrap_or(serde_json::json!({})); if let Some(subject_did) = value.get("subject").and_then(|s| s.as_str()) { - match get_profile_internal(&app_state, subject_did, None).await { + match get_profile_internal(&app_state, subject_did, viewer_did.as_deref()).await { Ok(profile) => follows.push(profile), Err(e) => tracing::warn!("failed to get profile for {subject_did}: {e}"), } } } - let subject_profile = match get_profile_internal(&app_state, repo.did.as_str(), None).await { - Ok(p) => p, - Err(e) => { - tracing::error!("failed to get profile for {}: {e}", repo.did); - return proxy_request(req).await; - } - }; + let subject_profile = + match get_profile_internal(&app_state, repo.did.as_str(), viewer_did.as_deref()).await { + Ok(p) => p, + Err(e) => { + tracing::error!("failed to get profile for {}: {e}", repo.did); + return proxy_request(req).await; + } + }; Ok(Json(serde_json::json!({ "subject": subject_profile, @@ -1185,11 +1387,14 @@ async fn get_followers( }; let mut followers = Vec::new(); + let viewer_did = get_auth_did(&req); for bl in backlinks_page.backlinks { match jacquard_common::types::string::AtUri::new(bl.uri.as_str()) { Ok(uri) => { let author_ident = uri.authority(); - match get_profile_internal(&app_state, author_ident.as_str(), None).await { + match get_profile_internal(&app_state, author_ident.as_str(), viewer_did.as_deref()) + .await + { Ok(profile) => followers.push(profile), Err(e) => tracing::warn!("failed to get profile for {author_ident}: {e}"), } @@ -1198,13 +1403,14 @@ async fn get_followers( } } - let subject_profile = match get_profile_internal(&app_state, repo.did.as_str(), None).await { - Ok(p) => p, - Err(e) => { - tracing::error!("failed to get profile for {}: {e}", repo.did); - return proxy_request(req).await; - } - }; + let subject_profile = + match get_profile_internal(&app_state, repo.did.as_str(), viewer_did.as_deref()).await { + Ok(p) => p, + Err(e) => { + tracing::error!("failed to get profile for {}: {e}", repo.did); + return proxy_request(req).await; + } + }; let cursor = backlinks_page .next_cursor @@ -1227,10 +1433,7 @@ async fn get_age_assurance_state() -> Result { fn get_auth_did(req: &Request) -> Option { let auth = req.headers().get("authorization")?.to_str().ok()?; - if !auth.starts_with("Bearer ") { - return None; - } - let token = &auth[7..]; + let token = auth.strip_prefix("Bearer ")?; let parts: Vec<&str> = token.split('.').collect(); if parts.len() != 3 { return None; @@ -1301,13 +1504,10 @@ async fn delete_bookmark( })?; let key = format!("bookmark:{}:{}", did, payload.uri); - app_state - .bookmarks - .remove(key.as_bytes()) - .map_err(|e| { - tracing::error!("failed to remove bookmark: {e}"); - StatusCode::INTERNAL_SERVER_ERROR - })?; + app_state.bookmarks.remove(key.as_bytes()).map_err(|e| { + tracing::error!("failed to remove bookmark: {e}"); + StatusCode::INTERNAL_SERVER_ERROR + })?; Ok(Json(serde_json::json!({})).into_response()) } @@ -1371,7 +1571,7 @@ async fn get_bookmarks( let mut bookmarks = Vec::new(); for (key, cid_bytes) in fetched_items { let key_str = String::from_utf8_lossy(&key); - let uri_str = &key_str[prefix.len()..]; + let uri_str = key_str.strip_prefix(&prefix).unwrap_or(&key_str); let cid_str = String::from_utf8_lossy(&cid_bytes); match jacquard_common::types::string::AtUri::new(uri_str) { @@ -1381,13 +1581,11 @@ async fn get_bookmarks( let rkey = uri.rkey().unwrap().0.as_str(); match app_state.hydrant.repos.resolve(author_ident).await { - Ok(repo) => { - match repo.get_record(collection, rkey).await { - Ok(Some(record)) => { - match get_profile_internal(&app_state, repo.did.as_str(), None).await - { - Ok(author_profile) => { - bookmarks.push(serde_json::json!({ + Ok(repo) => match repo.get_record(collection, rkey).await { + Ok(Some(record)) => { + match get_profile_internal(&app_state, repo.did.as_str(), None).await { + Ok(author_profile) => { + bookmarks.push(serde_json::json!({ "uri": uri_str, "cid": cid_str.to_string(), "author": author_profile, @@ -1397,16 +1595,15 @@ async fn get_bookmarks( "likeCount": app_state.hydrant.backlinks.count(uri_str.to_string()).source("app.bsky.feed.like").run().await.unwrap_or(0), "indexedAt": chrono::Utc::now().to_rfc3339(), })); - } - Err(e) => { - tracing::warn!("failed to get profile for {}: {e}", repo.did) - } + } + Err(e) => { + tracing::warn!("failed to get profile for {}: {e}", repo.did) } } - Ok(None) => tracing::warn!("bookmark record not found: {uri_str}"), - Err(e) => tracing::warn!("failed to get bookmark record {uri_str}: {e}"), } - } + Ok(None) => tracing::warn!("bookmark record not found: {uri_str}"), + Err(e) => tracing::warn!("failed to get bookmark record {uri_str}: {e}"), + }, Err(e) => tracing::warn!("failed to resolve actor {author_ident}: {e}"), } } @@ -1504,13 +1701,10 @@ async fn update_draft( let tid = payload.draft.id; let key = format!("draft:{}:{}", did, tid); - let existing = app_state - .drafts - .get(key.as_bytes()) - .map_err(|e| { - tracing::error!("failed to get draft from db: {e}"); - StatusCode::INTERNAL_SERVER_ERROR - })?; + let existing = app_state.drafts.get(key.as_bytes()).map_err(|e| { + tracing::error!("failed to get draft from db: {e}"); + StatusCode::INTERNAL_SERVER_ERROR + })?; if let Some(existing_bytes) = existing.as_deref() { match serde_json::from_slice::(&existing_bytes) { Ok(mut existing_obj) => { @@ -1782,6 +1976,7 @@ async fn get_timeline( let mut feed = Vec::new(); let mut next_cursor = None; + let viewer_did = get_auth_did(&req); for (did, col, rkey, value) in all_items { next_cursor = Some(rkey.clone()); let uri = format!("at://{}/{}/{}", did.as_str(), col, rkey); @@ -1789,7 +1984,7 @@ async fn get_timeline( let val_json = serde_json::to_value(&value).unwrap_or(serde_json::json!({})); if col == "app.bsky.feed.post" { - match get_post_view(&app_state, &uri, None).await { + match get_post_view(&app_state, &uri, viewer_did.as_deref()).await { Ok(post_view) => feed.push(serde_json::json!({ "post": post_view })), Err(e) => tracing::warn!("failed to get post view for {uri}: {e}"), } @@ -1799,9 +1994,11 @@ async fn get_timeline( .and_then(|s| s.get("uri")) .and_then(|u| u.as_str()) { - match get_post_view(&app_state, subject_uri, None).await { + match get_post_view(&app_state, subject_uri, viewer_did.as_deref()).await { Ok(post_view) => { - match get_profile_internal(&app_state, did.as_str(), None).await { + match get_profile_internal(&app_state, did.as_str(), viewer_did.as_deref()) + .await + { Ok(reposter_profile) => { feed.push(serde_json::json!({ "post": post_view, @@ -1818,7 +2015,9 @@ async fn get_timeline( } } Err(e) => { - tracing::warn!("failed to get post view for reposted post {subject_uri}: {e}"); + tracing::warn!( + "failed to get post view for reposted post {subject_uri}: {e}" + ); } } } @@ -1910,7 +2109,9 @@ async fn get_quotes( Ok(Some(rec)) => rec, Ok(None) => continue, Err(e) => { - tracing::error!("failed to get record {collection}/{rkey} from repo {author_ident}: {e}"); + tracing::error!( + "failed to get record {collection}/{rkey} from repo {author_ident}: {e}" + ); continue; } }; @@ -1937,7 +2138,8 @@ async fn get_quotes( }); if is_quote { - match get_post_view(&app_state, bl.uri.as_str(), None).await { + let viewer_did = get_auth_did(&req); + match get_post_view(&app_state, bl.uri.as_str(), viewer_did.as_deref()).await { Ok(post_view) => { posts.push(post_view); found += 1; @@ -1982,7 +2184,10 @@ async fn get_actor_likes( })?; let ident = AtIdentifier::new(¶ms.actor).map_err(|e| { - tracing::error!("failed to create AtIdentifier for actor {}: {e}", params.actor); + tracing::error!( + "failed to create AtIdentifier for actor {}: {e}", + params.actor + ); StatusCode::BAD_REQUEST })?; let repo = match app_state.hydrant.repos.resolve(&ident).await { @@ -2007,6 +2212,7 @@ async fn get_actor_likes( }; let mut feed = Vec::new(); + let viewer_did = get_auth_did(&req); for rec in record_list.records { let value = serde_json::to_value(&rec.value).unwrap_or(serde_json::json!({})); if let Some(subject_uri) = value @@ -2014,9 +2220,11 @@ async fn get_actor_likes( .and_then(|s| s.get("uri")) .and_then(|u| u.as_str()) { - match get_post_view(&app_state, subject_uri, None).await { + match get_post_view(&app_state, subject_uri, viewer_did.as_deref()).await { Ok(post_view) => feed.push(serde_json::json!({ "post": post_view })), - Err(e) => tracing::warn!("failed to get post view for liked post {subject_uri}: {e}"), + Err(e) => { + tracing::warn!("failed to get post view for liked post {subject_uri}: {e}") + } } } } @@ -2050,7 +2258,10 @@ async fn get_relationships( } let ident = AtIdentifier::new(¶ms.actor).map_err(|e| { - tracing::error!("failed to create AtIdentifier for actor {}: {e}", params.actor); + tracing::error!( + "failed to create AtIdentifier for actor {}: {e}", + params.actor + ); StatusCode::BAD_REQUEST })?; let repo = match app_state.hydrant.repos.resolve(&ident).await { @@ -2204,14 +2415,20 @@ async fn get_known_followers( cursor = rl.cursor.map(|c| c.to_string()); } Err(e) => { - tracing::error!("failed to list follows for auth user {}: {e}", auth_repo.did); + tracing::error!( + "failed to list follows for auth user {}: {e}", + auth_repo.did + ); break; } } } let ident = AtIdentifier::new(¶ms.actor).map_err(|e| { - tracing::error!("failed to create AtIdentifier for actor {}: {e}", params.actor); + tracing::error!( + "failed to create AtIdentifier for actor {}: {e}", + params.actor + ); StatusCode::BAD_REQUEST })?; let repo = match app_state.hydrant.repos.resolve(&ident).await { @@ -2346,7 +2563,9 @@ async fn get_blocks( if let Some(subject_did) = value.get("subject").and_then(|s| s.as_str()) { match get_profile_internal(&app_state, subject_did, None).await { Ok(profile) => blocks.push(profile), - Err(e) => tracing::warn!("failed to get profile for blocked actor {subject_did}: {e}"), + Err(e) => { + tracing::warn!("failed to get profile for blocked actor {subject_did}: {e}") + } } } } @@ -2409,13 +2628,15 @@ async fn get_list(State(app_state): State, req: Request) -> Result p, - Err(e) => { - tracing::error!("failed to get author profile for {}: {e}", repo.did); - return proxy_request(req).await; - } - }; + let viewer_did = get_auth_did(&req); + let author_profile = + match get_profile_internal(&app_state, repo.did.as_str(), viewer_did.as_deref()).await { + Ok(p) => p, + Err(e) => { + tracing::error!("failed to get author profile for {}: {e}", repo.did); + return proxy_request(req).await; + } + }; let val_json = serde_json::to_value(&record.value).unwrap_or(serde_json::json!({})); @@ -2524,7 +2745,10 @@ async fn get_lists( })?; let ident = AtIdentifier::new(¶ms.actor).map_err(|e| { - tracing::error!("failed to create AtIdentifier for actor {}: {e}", params.actor); + tracing::error!( + "failed to create AtIdentifier for actor {}: {e}", + params.actor + ); StatusCode::BAD_REQUEST })?; let repo = match app_state.hydrant.repos.resolve(&ident).await { @@ -2535,13 +2759,15 @@ async fn get_lists( } }; - let author_profile = match get_profile_internal(&app_state, repo.did.as_str(), None).await { - Ok(p) => p, - Err(e) => { - tracing::error!("failed to get author profile for {}: {e}", repo.did); - return proxy_request(req).await; - } - }; + let viewer_did = get_auth_did(&req); + let author_profile = + match get_profile_internal(&app_state, repo.did.as_str(), viewer_did.as_deref()).await { + Ok(p) => p, + Err(e) => { + tracing::error!("failed to get author profile for {}: {e}", repo.did); + return proxy_request(req).await; + } + }; let limit = params.limit.unwrap_or(50).min(100); let record_list = match repo @@ -2589,7 +2815,10 @@ async fn get_lists_with_membership( })?; let ident = AtIdentifier::new(¶ms.actor).map_err(|e| { - tracing::error!("failed to create AtIdentifier for actor {}: {e}", params.actor); + tracing::error!( + "failed to create AtIdentifier for actor {}: {e}", + params.actor + ); StatusCode::BAD_REQUEST })?; let repo = match app_state.hydrant.repos.resolve(&ident).await { @@ -2612,9 +2841,9 @@ async fn get_lists_with_membership( } }; - let mut lists: Vec = Vec::new(); + let lists: Vec = Vec::new(); for rec in record_list.records { - let uri = format!( + let _uri = format!( "at://{}/app.bsky.graph.list/{}", repo.did.as_str(), rec.rkey.as_str() @@ -2642,7 +2871,10 @@ async fn get_actor_feeds( })?; let ident = AtIdentifier::new(¶ms.actor).map_err(|e| { - tracing::error!("failed to create AtIdentifier for actor {}: {e}", params.actor); + tracing::error!( + "failed to create AtIdentifier for actor {}: {e}", + params.actor + ); StatusCode::BAD_REQUEST })?; let repo = match app_state.hydrant.repos.resolve(&ident).await { @@ -2653,13 +2885,15 @@ async fn get_actor_feeds( } }; - let author_profile = match get_profile_internal(&app_state, repo.did.as_str(), None).await { - Ok(p) => p, - Err(e) => { - tracing::error!("failed to get author profile for {}: {e}", repo.did); - return proxy_request(req).await; - } - }; + let viewer_did = get_auth_did(&req); + let author_profile = + match get_profile_internal(&app_state, repo.did.as_str(), viewer_did.as_deref()).await { + Ok(p) => p, + Err(e) => { + tracing::error!("failed to get author profile for {}: {e}", repo.did); + return proxy_request(req).await; + } + }; let limit = params.limit.unwrap_or(50).min(100); let record_list = match repo @@ -2756,13 +2990,15 @@ async fn get_feed_generator( } }; - let author_profile = match get_profile_internal(&app_state, repo.did.as_str(), None).await { - Ok(p) => p, - Err(e) => { - tracing::error!("failed to get author profile for {}: {e}", repo.did); - return proxy_request(req).await; - } - }; + let viewer_did = get_auth_did(&req); + let author_profile = + match get_profile_internal(&app_state, repo.did.as_str(), viewer_did.as_deref()).await { + Ok(p) => p, + Err(e) => { + tracing::error!("failed to get author profile for {}: {e}", repo.did); + return proxy_request(req).await; + } + }; let val_json = serde_json::to_value(&record.value).unwrap_or(serde_json::json!({})); @@ -2798,6 +3034,7 @@ async fn get_feed_generators( } let mut views = Vec::new(); + let viewer_did = get_auth_did(&req); for feed_uri in feeds { let uri = match jacquard_common::types::string::AtUri::new(&feed_uri) { Ok(u) => u, @@ -2810,14 +3047,15 @@ async fn get_feed_generators( let rkey = uri.rkey().unwrap().0.as_str().to_string(); match app_state.hydrant.repos.resolve(author_ident).await { - Ok(repo) => { - match repo.get_record("app.bsky.feed.generator", &rkey).await { - Ok(Some(record)) => { - match get_profile_internal(&app_state, repo.did.as_str(), None).await { - Ok(author_profile) => { - let val_json = serde_json::to_value(&record.value) - .unwrap_or(serde_json::json!({})); - views.push(serde_json::json!({ + Ok(repo) => match repo.get_record("app.bsky.feed.generator", &rkey).await { + Ok(Some(record)) => { + match get_profile_internal(&app_state, repo.did.as_str(), viewer_did.as_deref()) + .await + { + Ok(author_profile) => { + let val_json = serde_json::to_value(&record.value) + .unwrap_or(serde_json::json!({})); + views.push(serde_json::json!({ "uri": feed_uri, "cid": record.cid.to_string(), "did": val_json.get("did").and_then(|v| v.as_str()).unwrap_or(""), @@ -2828,18 +3066,19 @@ async fn get_feed_generators( "likeCount": app_state.hydrant.backlinks.count(feed_uri.clone()).source("app.bsky.feed.like").run().await.unwrap_or(0), "indexedAt": chrono::Utc::now().to_rfc3339(), })); - } - Err(e) => { - tracing::warn!("failed to get author profile for {}: {e}", repo.did) - } + } + Err(e) => { + tracing::warn!("failed to get author profile for {}: {e}", repo.did) } } - Ok(None) => tracing::warn!("feed generator record not found: {author_ident}/{rkey}"), - Err(e) => tracing::warn!( - "failed to get feed generator record {author_ident}/{rkey}: {e}" - ), } - } + Ok(None) => { + tracing::warn!("feed generator record not found: {author_ident}/{rkey}") + } + Err(e) => { + tracing::warn!("failed to get feed generator record {author_ident}/{rkey}: {e}") + } + }, Err(e) => tracing::warn!("failed to resolve repo for {author_ident}: {e}"), } } @@ -3032,7 +3271,7 @@ async fn get_mutes( let mut next_cursor = None; for key in fetched_keys { let key_str = String::from_utf8_lossy(&key); - let muted_did = &key_str[prefix.len()..]; + let muted_did = key_str.strip_prefix(&prefix).unwrap_or(&key_str); match get_profile_internal(&app_state, muted_did, None).await { Ok(profile) => mutes.push(profile), Err(e) => tracing::warn!("failed to get profile for muted actor {muted_did}: {e}"), @@ -3079,7 +3318,9 @@ async fn notification_indexer(app_state: AppState) { Some(subj_uri.to_string()), )); } - Err(e) => tracing::warn!("failed to parse notification like uri {subj_uri}: {e}"), + Err(e) => { + tracing::warn!("failed to parse notification like uri {subj_uri}: {e}") + } } } } @@ -3097,7 +3338,9 @@ async fn notification_indexer(app_state: AppState) { Some(subj_uri.to_string()), )); } - Err(e) => tracing::warn!("failed to parse notification repost uri {subj_uri}: {e}"), + Err(e) => tracing::warn!( + "failed to parse notification repost uri {subj_uri}: {e}" + ), } } } @@ -3121,7 +3364,9 @@ async fn notification_indexer(app_state: AppState) { Some(parent_uri.to_string()), )); } - Err(e) => tracing::warn!("failed to parse notification reply uri {parent_uri}: {e}"), + Err(e) => tracing::warn!( + "failed to parse notification reply uri {parent_uri}: {e}" + ), } } @@ -3172,7 +3417,9 @@ async fn notification_indexer(app_state: AppState) { Some(qu.to_string()), )); } - Err(e) => tracing::warn!("failed to parse notification quote uri {qu}: {e}"), + Err(e) => tracing::warn!( + "failed to parse notification quote uri {qu}: {e}" + ), } } } @@ -3307,7 +3554,9 @@ async fn list_notifications( notif["author"] = author_profile; } Err(e) => { - tracing::warn!("failed to get author profile for notification from {author_did}: {e}"); + tracing::warn!( + "failed to get author profile for notification from {author_did}: {e}" + ); continue; } } @@ -3374,7 +3623,8 @@ async fn get_unread_count( #[derive(Deserialize)] struct UpdateSeenReq { - seenAt: String, + #[serde(rename = "seenAt")] + seen_at: String, } async fn update_seen( @@ -3396,7 +3646,7 @@ async fn update_seen( let seen_key = format!("seen:{}", did); app_state .seen - .insert(seen_key.as_bytes(), payload.seenAt.as_bytes()) + .insert(seen_key.as_bytes(), payload.seen_at.as_bytes()) .map_err(|e| { tracing::error!("failed to insert seen: {e}"); StatusCode::INTERNAL_SERVER_ERROR @@ -3463,10 +3713,11 @@ async fn get_feed(State(app_state): State, req: Request) -> Result