From 626e6362c1df09cff44582061cbb4ea4b04488ad Mon Sep 17 00:00:00 2001 From: dawn <90008@gaze.systems> Date: Sat, 25 Apr 2026 13:58:56 +0300 Subject: [PATCH] log errors better --- rust-toolchain.toml | 3 + src/main.rs | 1620 ++++++++++++++++++++++++++++--------------- 2 files changed, 1081 insertions(+), 542 deletions(-) create mode 100644 rust-toolchain.toml diff --git a/rust-toolchain.toml b/rust-toolchain.toml new file mode 100644 index 0000000..0440a5d --- /dev/null +++ b/rust-toolchain.toml @@ -0,0 +1,3 @@ +[toolchain] +channel = "stable" +components = ["rust-analyzer", "rust-src", "rustfmt"] diff --git a/src/main.rs b/src/main.rs index 142695a..4e05330 100644 --- a/src/main.rs +++ b/src/main.rs @@ -11,6 +11,7 @@ use hydrant::deps::futures::StreamExt; use jacquard_common::IntoStatic; use jacquard_common::types::ident::AtIdentifier; use serde::{Deserialize, Serialize}; +use tracing; use tracing_subscriber::EnvFilter; #[derive(Clone)] @@ -346,9 +347,10 @@ async fn get_post_thread( ) -> Result { // let hydrant = &app_state.hydrant; let query_str = req.uri().query().unwrap_or(""); - let Ok(params) = serde_urlencoded::from_str::(query_str) else { - return Err(StatusCode::BAD_REQUEST); - }; + let params = serde_urlencoded::from_str::(query_str).map_err(|e| { + tracing::error!("failed to parse query params: {e}"); + StatusCode::BAD_REQUEST + })?; let viewer_did = get_auth_did(&req); match get_thread_view_post( @@ -466,7 +468,7 @@ async fn get_thread_view_post( .and_then(|p| p.get("uri")) .and_then(|u| u.as_str()) { - if let Ok(parent_thread) = Box::pin(get_thread_view_post( + match Box::pin(get_thread_view_post( app_state, parent_uri, 0, @@ -475,10 +477,15 @@ async fn get_thread_view_post( )) .await { - thread - .as_object_mut() - .unwrap() - .insert("parent".to_string(), parent_thread); + Ok(parent_thread) => { + thread + .as_object_mut() + .unwrap() + .insert("parent".to_string(), parent_thread); + } + Err(e) => { + tracing::warn!("failed to fetch parent thread {parent_uri}: {e}"); + } } } } @@ -496,7 +503,7 @@ async fn get_thread_view_post( .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; let mut replies = Vec::new(); for bl in replies_page.backlinks { - if let Ok(reply_thread) = Box::pin(get_thread_view_post( + match Box::pin(get_thread_view_post( app_state, bl.uri.as_str(), depth - 1, @@ -505,7 +512,12 @@ async fn get_thread_view_post( )) .await { - replies.push(reply_thread); + Ok(reply_thread) => { + replies.push(reply_thread); + } + Err(e) => { + tracing::warn!("failed to fetch reply thread {}: {e}", bl.uri); + } } } thread @@ -528,17 +540,27 @@ async fn get_profile( ) -> Result { // let hydrant = &app_state.hydrant; let query_str = req.uri().query().unwrap_or(""); - let Ok(params) = serde_urlencoded::from_str::(query_str) else { - return Err(StatusCode::BAD_REQUEST); - }; - - let Ok(ident) = AtIdentifier::new(¶ms.actor) else { - return Err(StatusCode::BAD_REQUEST); - }; - - if let Ok(repo) = app_state.hydrant.repos.resolve(&ident).await { - if let Ok(profile) = get_profile_internal(&app_state, repo.did.as_str(), None).await { - return Ok(Json(profile).into_response()); + let params = serde_urlencoded::from_str::(query_str).map_err(|e| { + tracing::error!("failed to parse query params: {e}"); + StatusCode::BAD_REQUEST + })?; + + let ident = AtIdentifier::new(¶ms.actor).map_err(|e| { + tracing::error!("failed to parse actor identifier: {e}"); + StatusCode::BAD_REQUEST + })?; + + match app_state.hydrant.repos.resolve(&ident).await { + Ok(repo) => { + match get_profile_internal(&app_state, repo.did.as_str(), None).await { + Ok(profile) => return Ok(Json(profile).into_response()), + Err(e) => { + tracing::warn!("failed to get profile for {}: {e}", repo.did); + } + } + } + Err(e) => { + tracing::error!("failed to resolve actor {}: {e}", params.actor); } } @@ -715,30 +737,44 @@ async fn get_author_feed( ) -> Result { // let hydrant = &app_state.hydrant; let query_str = req.uri().query().unwrap_or(""); - let Ok(params) = serde_urlencoded::from_str::(query_str) else { - return Err(StatusCode::BAD_REQUEST); - }; - - let Ok(ident) = AtIdentifier::new(¶ms.actor) else { - return Err(StatusCode::BAD_REQUEST); - }; - - let Ok(repo) = app_state.hydrant.repos.resolve(&ident).await else { - return proxy_request(req).await; + let params = serde_urlencoded::from_str::(query_str).map_err(|e| { + tracing::error!("failed to parse query params: {e}"); + StatusCode::BAD_REQUEST + })?; + + let ident = AtIdentifier::new(¶ms.actor).map_err(|e| { + tracing::error!("failed to parse actor identifier: {e}"); + StatusCode::BAD_REQUEST + })?; + + let repo = match app_state.hydrant.repos.resolve(&ident).await { + Ok(repo) => repo, + Err(e) => { + tracing::error!("failed to resolve actor {}: {e}", params.actor); + return proxy_request(req).await; + } }; let did = repo.did.clone(); let limit = params.limit.unwrap_or(50).min(100); - let Ok(record_list) = repo + let record_list = match repo .list_records("app.bsky.feed.post", limit, true, params.cursor.as_deref()) .await - else { - return proxy_request(req).await; + { + Ok(rl) => rl, + Err(e) => { + tracing::error!("failed to list records for {}: {e}", did); + return proxy_request(req).await; + } }; - let Ok(author_profile) = get_profile_internal(&app_state, did.as_str(), None).await else { - return proxy_request(req).await; + let author_profile = match get_profile_internal(&app_state, did.as_str(), None).await { + Ok(p) => p, + Err(e) => { + tracing::error!("failed to get profile for {}: {e}", did); + return proxy_request(req).await; + } }; let mut feed = Vec::new(); @@ -784,9 +820,10 @@ async fn get_likes( ) -> Result { // let hydrant = &app_state.hydrant; let query_str = req.uri().query().unwrap_or(""); - let Ok(params) = serde_urlencoded::from_str::(query_str) else { - return Err(StatusCode::BAD_REQUEST); - }; + let params = serde_urlencoded::from_str::(query_str).map_err(|e| { + tracing::error!("failed to parse query params: {e}"); + StatusCode::BAD_REQUEST + })?; let limit = params.limit.unwrap_or(50).min(100); @@ -797,34 +834,62 @@ async fn get_likes( .source("app.bsky.feed.like") .limit(limit); if let Some(cursor_str) = params.cursor { - if let Ok(c) = data_encoding::BASE64URL_NOPAD.decode(cursor_str.as_bytes()) { - fetch = fetch.cursor(c); + match data_encoding::BASE64URL_NOPAD.decode(cursor_str.as_bytes()) { + Ok(c) => { + fetch = fetch.cursor(c); + } + Err(e) => { + tracing::warn!("failed to decode cursor {cursor_str}: {e}"); + } } } - let Ok(backlinks_page) = fetch.run().await else { - return proxy_request(req).await; + let backlinks_page = match fetch.run().await { + Ok(bp) => bp, + Err(e) => { + tracing::error!("failed to fetch backlinks for {}: {e}", params.uri); + return proxy_request(req).await; + } }; let mut likes = Vec::new(); for bl in backlinks_page.backlinks { - let Ok(uri) = jacquard_common::types::string::AtUri::new(bl.uri.as_str()) else { - continue; + let uri = match jacquard_common::types::string::AtUri::new(bl.uri.as_str()) { + Ok(uri) => uri, + Err(e) => { + tracing::warn!("failed to parse uri {}: {e}", bl.uri); + continue; + } }; let author_ident = uri.authority(); - let Ok(profile) = get_profile_internal(&app_state, author_ident.as_str(), None).await - else { - continue; + 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 Ok(repo) = app_state.hydrant.repos.resolve(author_ident).await else { - continue; + let repo = match app_state.hydrant.repos.resolve(author_ident).await { + Ok(repo) => repo, + Err(e) => { + tracing::warn!("failed to resolve actor {author_ident}: {e}"); + continue; + } }; let collection = uri.collection().unwrap().as_str(); let rkey = uri.rkey().unwrap().0.as_str(); - let Ok(Some(record)) = repo.get_record(collection, rkey).await else { - continue; + let record = match repo.get_record(collection, rkey).await { + Ok(Some(record)) => record, + Ok(None) => { + tracing::warn!("record not found: {collection}/{rkey}"); + continue; + } + Err(e) => { + tracing::warn!("failed to get record {collection}/{rkey}: {e}"); + continue; + } }; let value = serde_json::to_value(record.value).unwrap_or(serde_json::json!({})); let created_at = value @@ -858,9 +923,10 @@ async fn get_reposted_by( ) -> Result { // let hydrant = &app_state.hydrant; let query_str = req.uri().query().unwrap_or(""); - let Ok(params) = serde_urlencoded::from_str::(query_str) else { - return Err(StatusCode::BAD_REQUEST); - }; + let params = serde_urlencoded::from_str::(query_str).map_err(|e| { + tracing::error!("failed to parse query params: {e}"); + StatusCode::BAD_REQUEST + })?; let limit = params.limit.unwrap_or(50).min(100); @@ -871,24 +937,40 @@ async fn get_reposted_by( .source("app.bsky.feed.repost") .limit(limit); if let Some(cursor_str) = params.cursor { - if let Ok(c) = data_encoding::BASE64URL_NOPAD.decode(cursor_str.as_bytes()) { - fetch = fetch.cursor(c); + match data_encoding::BASE64URL_NOPAD.decode(cursor_str.as_bytes()) { + Ok(c) => { + fetch = fetch.cursor(c); + } + Err(e) => { + tracing::warn!("failed to decode cursor {cursor_str}: {e}"); + } } } - let Ok(backlinks_page) = fetch.run().await else { - return proxy_request(req).await; + let backlinks_page = match fetch.run().await { + Ok(bp) => bp, + Err(e) => { + tracing::error!("failed to fetch backlinks for {}: {e}", params.uri); + return proxy_request(req).await; + } }; let mut reposted_by = Vec::new(); for bl in backlinks_page.backlinks { - let Ok(uri) = jacquard_common::types::string::AtUri::new(bl.uri.as_str()) else { - continue; + let uri = match jacquard_common::types::string::AtUri::new(bl.uri.as_str()) { + Ok(uri) => uri, + Err(e) => { + tracing::warn!("failed to parse uri {}: {e}", bl.uri); + continue; + } }; let author_ident = uri.authority(); - let Ok(profile) = get_profile_internal(&app_state, author_ident.as_str(), None).await - else { - continue; + 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; + } }; reposted_by.push(profile); } @@ -938,13 +1020,21 @@ async fn get_profiles( let mut profiles = Vec::new(); for actor in actors { - if let Ok(ident) = AtIdentifier::new(&actor) { - if let Ok(repo) = app_state.hydrant.repos.resolve(&ident).await { - if let Ok(profile) = get_profile_internal(&app_state, repo.did.as_str(), None).await - { - profiles.push(profile); + let ident = match AtIdentifier::new(&actor) { + Ok(ident) => ident, + Err(e) => { + tracing::warn!("failed to parse actor identifier {actor}: {e}"); + continue; + } + }; + match app_state.hydrant.repos.resolve(&ident).await { + Ok(repo) => { + match get_profile_internal(&app_state, repo.did.as_str(), None).await { + Ok(profile) => profiles.push(profile), + Err(e) => tracing::warn!("failed to get profile for {}: {e}", repo.did), } } + Err(e) => tracing::warn!("failed to resolve actor {actor}: {e}"), } } @@ -964,8 +1054,9 @@ async fn get_posts( let mut posts = Vec::new(); for uri in uris { - if let Ok(post) = get_post_view(&app_state, &uri, None).await { - posts.push(post); + match get_post_view(&app_state, &uri, None).await { + Ok(post) => posts.push(post), + Err(e) => tracing::warn!("failed to get post view for {uri}: {e}"), } } @@ -987,19 +1078,25 @@ async fn get_follows( ) -> Result { // let hydrant = &app_state.hydrant; let query_str = req.uri().query().unwrap_or(""); - let Ok(params) = serde_urlencoded::from_str::(query_str) else { - return Err(StatusCode::BAD_REQUEST); - }; - - let Ok(ident) = AtIdentifier::new(¶ms.actor) else { - return Err(StatusCode::BAD_REQUEST); - }; - let Ok(repo) = app_state.hydrant.repos.resolve(&ident).await else { - return proxy_request(req).await; + let params: GetFollowsParams = serde_urlencoded::from_str(query_str).map_err(|e| { + tracing::error!("failed to parse query params: {e}"); + StatusCode::BAD_REQUEST + })?; + + let ident = AtIdentifier::new(¶ms.actor).map_err(|e| { + tracing::error!("failed to parse actor identifier: {e}"); + StatusCode::BAD_REQUEST + })?; + let repo = match app_state.hydrant.repos.resolve(&ident).await { + Ok(repo) => repo, + Err(e) => { + tracing::error!("failed to resolve actor {}: {e}", params.actor); + return proxy_request(req).await; + } }; let limit = params.limit.unwrap_or(50).min(100); - let Ok(record_list) = repo + let record_list = match repo .list_records( "app.bsky.graph.follow", limit, @@ -1007,23 +1104,31 @@ async fn get_follows( params.cursor.as_deref(), ) .await - else { - return proxy_request(req).await; + { + Ok(rl) => rl, + Err(e) => { + tracing::error!("failed to list follows for {}: {e}", repo.did); + return proxy_request(req).await; + } }; let mut follows = Vec::new(); 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()) { - if let Ok(profile) = get_profile_internal(&app_state, subject_did, None).await { - follows.push(profile); + match get_profile_internal(&app_state, subject_did, None).await { + Ok(profile) => follows.push(profile), + Err(e) => tracing::warn!("failed to get profile for {subject_did}: {e}"), } } } - let Ok(subject_profile) = get_profile_internal(&app_state, repo.did.as_str(), None).await - else { - return proxy_request(req).await; + 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; + } }; Ok(Json(serde_json::json!({ @@ -1040,15 +1145,21 @@ async fn get_followers( ) -> Result { // let hydrant = &app_state.hydrant; let query_str = req.uri().query().unwrap_or(""); - let Ok(params) = serde_urlencoded::from_str::(query_str) else { - return Err(StatusCode::BAD_REQUEST); - }; - - let Ok(ident) = AtIdentifier::new(¶ms.actor) else { - return Err(StatusCode::BAD_REQUEST); - }; - let Ok(repo) = app_state.hydrant.repos.resolve(&ident).await else { - return proxy_request(req).await; + let params: GetFollowsParams = serde_urlencoded::from_str(query_str).map_err(|e| { + tracing::error!("failed to parse query params: {e}"); + StatusCode::BAD_REQUEST + })?; + + let ident = AtIdentifier::new(¶ms.actor).map_err(|e| { + tracing::error!("failed to parse actor identifier: {e}"); + StatusCode::BAD_REQUEST + })?; + let repo = match app_state.hydrant.repos.resolve(&ident).await { + Ok(repo) => repo, + Err(e) => { + tracing::error!("failed to resolve actor {}: {e}", params.actor); + return proxy_request(req).await; + } }; let limit = params.limit.unwrap_or(50).min(100); @@ -1059,29 +1170,40 @@ async fn get_followers( .source("app.bsky.graph.follow") .limit(limit); if let Some(cursor_str) = params.cursor { - if let Ok(c) = data_encoding::BASE64URL_NOPAD.decode(cursor_str.as_bytes()) { - fetch = fetch.cursor(c); + match data_encoding::BASE64URL_NOPAD.decode(cursor_str.as_bytes()) { + Ok(c) => fetch = fetch.cursor(c), + Err(e) => tracing::warn!("failed to decode cursor {cursor_str}: {e}"), } } - let Ok(backlinks_page) = fetch.run().await else { - return proxy_request(req).await; + let backlinks_page = match fetch.run().await { + Ok(bp) => bp, + Err(e) => { + tracing::error!("failed to fetch backlinks for {}: {e}", repo.did); + return proxy_request(req).await; + } }; let mut followers = Vec::new(); for bl in backlinks_page.backlinks { - if let Ok(uri) = jacquard_common::types::string::AtUri::new(bl.uri.as_str()) { - let author_ident = uri.authority(); - if let Ok(profile) = get_profile_internal(&app_state, author_ident.as_str(), None).await - { - followers.push(profile); + 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 { + Ok(profile) => followers.push(profile), + Err(e) => tracing::warn!("failed to get profile for {author_ident}: {e}"), + } } + Err(e) => tracing::warn!("failed to parse uri {}: {e}", bl.uri), } } - let Ok(subject_profile) = get_profile_internal(&app_state, repo.did.as_str(), None).await - else { - return proxy_request(req).await; + 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 cursor = backlinks_page @@ -1140,15 +1262,19 @@ async fn create_bookmark( let body_bytes = axum::body::to_bytes(req.into_body(), usize::MAX) .await .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; - let Ok(payload) = serde_json::from_slice::(&body_bytes) else { - return Err(StatusCode::BAD_REQUEST); - }; + let payload: CreateBookmarkReq = serde_json::from_slice(&body_bytes).map_err(|e| { + tracing::error!("failed to parse create bookmark request: {e}"); + StatusCode::BAD_REQUEST + })?; let key = format!("bookmark:{}:{}", did, payload.uri); app_state .bookmarks .insert(key.as_bytes(), payload.cid.as_bytes()) - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + .map_err(|e| { + tracing::error!("failed to insert bookmark: {e}"); + StatusCode::INTERNAL_SERVER_ERROR + })?; Ok(Json(serde_json::json!({})).into_response()) } @@ -1169,15 +1295,19 @@ async fn delete_bookmark( let body_bytes = axum::body::to_bytes(req.into_body(), usize::MAX) .await .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; - let Ok(payload) = serde_json::from_slice::(&body_bytes) else { - return Err(StatusCode::BAD_REQUEST); - }; + let payload: DeleteBookmarkReq = serde_json::from_slice(&body_bytes).map_err(|e| { + tracing::error!("failed to parse delete bookmark request: {e}"); + StatusCode::BAD_REQUEST + })?; let key = format!("bookmark:{}:{}", did, payload.uri); app_state .bookmarks .remove(key.as_bytes()) - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + .map_err(|e| { + tracing::error!("failed to remove bookmark: {e}"); + StatusCode::INTERNAL_SERVER_ERROR + })?; Ok(Json(serde_json::json!({})).into_response()) } @@ -1221,8 +1351,12 @@ async fn get_bookmarks( let mut fetched_items = Vec::new(); let mut next_cursor = None; for item in iter { - let Ok((key, cid_bytes)) = item.into_inner() else { - continue; + let (key, cid_bytes) = match item.into_inner() { + Ok(inner) => inner, + Err(e) => { + tracing::warn!("failed to get item from bookmarks iterator: {e}"); + continue; + } }; if !key.starts_with(prefix.as_bytes()) { break; @@ -1240,29 +1374,43 @@ async fn get_bookmarks( let uri_str = &key_str[prefix.len()..]; let cid_str = String::from_utf8_lossy(&cid_bytes); - if let Ok(uri) = jacquard_common::types::string::AtUri::new(uri_str) { - let author_ident = uri.authority(); - let collection = uri.collection().unwrap().as_str(); - let rkey = uri.rkey().unwrap().0.as_str(); - - if let Ok(repo) = app_state.hydrant.repos.resolve(author_ident).await { - if let Ok(Some(record)) = repo.get_record(collection, rkey).await { - if let Ok(author_profile) = - get_profile_internal(&app_state, repo.did.as_str(), None).await - { - bookmarks.push(serde_json::json!({ - "uri": uri_str, - "cid": cid_str.to_string(), - "author": author_profile, - "record": record.value, - "replyCount": app_state.hydrant.backlinks.count(uri_str.to_string()).source("app.bsky.feed.post").run().await.unwrap_or(0), - "repostCount": app_state.hydrant.backlinks.count(uri_str.to_string()).source("app.bsky.feed.repost").run().await.unwrap_or(0), - "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(), - })); + match jacquard_common::types::string::AtUri::new(uri_str) { + Ok(uri) => { + let author_ident = uri.authority(); + let collection = uri.collection().unwrap().as_str(); + 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!({ + "uri": uri_str, + "cid": cid_str.to_string(), + "author": author_profile, + "record": record.value, + "replyCount": app_state.hydrant.backlinks.count(uri_str.to_string()).source("app.bsky.feed.post").run().await.unwrap_or(0), + "repostCount": app_state.hydrant.backlinks.count(uri_str.to_string()).source("app.bsky.feed.repost").run().await.unwrap_or(0), + "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) + } + } + } + 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}"), } } + Err(e) => tracing::warn!("failed to parse bookmark uri {uri_str}: {e}"), } if bookmarks.len() >= limit { @@ -1293,10 +1441,14 @@ async fn create_draft( let body_bytes = axum::body::to_bytes(req.into_body(), usize::MAX) .await - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; - let Ok(payload) = serde_json::from_slice::(&body_bytes) else { - return Err(StatusCode::BAD_REQUEST); - }; + .map_err(|e| { + tracing::error!("failed to read request body: {e}"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + let payload = serde_json::from_slice::(&body_bytes).map_err(|e| { + tracing::error!("failed to deserialize create_draft request: {e}"); + StatusCode::BAD_REQUEST + })?; let tid = jacquard_common::types::tid::Tid::now(1.try_into().unwrap()).to_string(); let key = format!("draft:{}:{}", did, tid); @@ -1340,10 +1492,14 @@ async fn update_draft( let body_bytes = axum::body::to_bytes(req.into_body(), usize::MAX) .await - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; - let Ok(payload) = serde_json::from_slice::(&body_bytes) else { - return Err(StatusCode::BAD_REQUEST); - }; + .map_err(|e| { + tracing::error!("failed to read request body: {e}"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + let payload = serde_json::from_slice::(&body_bytes).map_err(|e| { + tracing::error!("failed to deserialize update_draft request: {e}"); + StatusCode::BAD_REQUEST + })?; let tid = payload.draft.id; let key = format!("draft:{}:{}", did, tid); @@ -1351,19 +1507,29 @@ async fn update_draft( let existing = app_state .drafts .get(key.as_bytes()) - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + .map_err(|e| { + tracing::error!("failed to get draft from db: {e}"); + StatusCode::INTERNAL_SERVER_ERROR + })?; if let Some(existing_bytes) = existing.as_deref() { - if let Ok(mut existing_obj) = serde_json::from_slice::(&existing_bytes) { - let now = chrono::Utc::now().to_rfc3339(); - existing_obj["draft"] = payload.draft.draft; - existing_obj["updatedAt"] = serde_json::json!(now); - - let val = - serde_json::to_vec(&existing_obj).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; - app_state - .drafts - .insert(key.as_bytes(), val) - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + match serde_json::from_slice::(&existing_bytes) { + Ok(mut existing_obj) => { + let now = chrono::Utc::now().to_rfc3339(); + existing_obj["draft"] = payload.draft.draft; + existing_obj["updatedAt"] = serde_json::json!(now); + + let val = serde_json::to_vec(&existing_obj).map_err(|e| { + tracing::error!("failed to serialize updated draft: {e}"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + app_state.drafts.insert(key.as_bytes(), val).map_err(|e| { + tracing::error!("failed to insert updated draft to db: {e}"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + } + Err(e) => { + tracing::error!("failed to parse existing draft from db: {e}"); + } } } @@ -1385,16 +1551,20 @@ async fn delete_draft( let body_bytes = axum::body::to_bytes(req.into_body(), usize::MAX) .await - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; - let Ok(payload) = serde_json::from_slice::(&body_bytes) else { - return Err(StatusCode::BAD_REQUEST); - }; + .map_err(|e| { + tracing::error!("failed to read request body: {e}"); + StatusCode::INTERNAL_SERVER_ERROR + })?; + let payload = serde_json::from_slice::(&body_bytes).map_err(|e| { + tracing::error!("failed to deserialize delete_draft request: {e}"); + StatusCode::BAD_REQUEST + })?; let key = format!("draft:{}:{}", did, payload.id); - app_state - .drafts - .remove(key.as_bytes()) - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + app_state.drafts.remove(key.as_bytes()).map_err(|e| { + tracing::error!("failed to remove draft from db: {e}"); + StatusCode::INTERNAL_SERVER_ERROR + })?; Ok(Json(serde_json::json!({})).into_response()) } @@ -1438,15 +1608,20 @@ async fn get_drafts( let mut next_cursor = None; for item in iter { - let Ok((key, val_bytes)) = item.into_inner() else { - continue; + let (key, val_bytes) = match item.into_inner() { + Ok(v) => v, + Err(e) => { + tracing::error!("failed to get item from drafts iterator: {e}"); + continue; + } }; if !key.starts_with(prefix.as_bytes()) { break; } - if let Ok(obj) = serde_json::from_slice::(&val_bytes) { - drafts.push(obj); + match serde_json::from_slice::(&val_bytes) { + Ok(obj) => drafts.push(obj), + Err(e) => tracing::error!("failed to parse draft from db: {e}"), } if drafts.len() >= limit { @@ -1478,19 +1653,25 @@ async fn get_timeline( ) -> Result { // let hydrant = &app_state.hydrant; let query_str = req.uri().query().unwrap_or(""); - let Ok(params) = serde_urlencoded::from_str::(query_str) else { - return Err(StatusCode::BAD_REQUEST); - }; + let params = serde_urlencoded::from_str::(query_str).map_err(|e| { + tracing::error!("failed to parse get_timeline query params: {e}"); + StatusCode::BAD_REQUEST + })?; let Some(did_str) = get_auth_did(&req) else { return Err(StatusCode::UNAUTHORIZED); }; - let Ok(ident) = AtIdentifier::new(&did_str) else { - return Err(StatusCode::BAD_REQUEST); - }; - let Ok(repo) = app_state.hydrant.repos.resolve(&ident).await else { - return proxy_request(req).await; + let ident = AtIdentifier::new(&did_str).map_err(|e| { + tracing::error!("failed to create AtIdentifier from did_str {did_str}: {e}"); + StatusCode::BAD_REQUEST + })?; + let repo = match app_state.hydrant.repos.resolve(&ident).await { + Ok(r) => r, + Err(e) => { + tracing::error!("failed to resolve repo for {ident}: {e}"); + return proxy_request(req).await; + } }; // Get all follows of auth user @@ -1499,17 +1680,26 @@ async fn get_timeline( let mut cursor = None; loop { - let Ok(record_list) = repo + let record_list = match repo .list_records("app.bsky.graph.follow", 100, true, cursor.as_deref()) .await - else { - break; + { + Ok(rl) => rl, + Err(e) => { + tracing::error!("failed to list follows for {}: {e}", repo.did); + break; + } }; 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()) { - if let Ok(did) = jacquard_common::types::string::Did::new(subject_did) { - follows.insert(did.into_static()); + match jacquard_common::types::string::Did::new(subject_did) { + Ok(did) => { + follows.insert(did.into_static()); + } + Err(e) => { + tracing::error!("failed to parse subject DID {subject_did}: {e}"); + } } } } @@ -1524,49 +1714,60 @@ async fn get_timeline( let mut all_items = Vec::new(); for did in follows { - if let Ok(followed_repo) = app_state.hydrant.repos.get(&did).info().await { - if let Some(info) = followed_repo { + match app_state.hydrant.repos.get(&did).info().await { + Ok(Some(info)) => { if !info.tracked { continue; } - } else { + } + Ok(None) => continue, + Err(e) => { + tracing::error!("failed to get info for followed repo {did}: {e}"); continue; } - } else { - continue; } let followed_repo = app_state.hydrant.repos.get(&did); // get posts - if let Ok(record_list) = followed_repo + match followed_repo .list_records("app.bsky.feed.post", limit, true, None) .await { - for rec in record_list.records { - let rkey = rec.rkey.as_str().to_string(); - all_items.push(( - did.clone(), - "app.bsky.feed.post".to_string(), - rkey, - rec.value, - )); + Ok(record_list) => { + for rec in record_list.records { + let rkey = rec.rkey.as_str().to_string(); + all_items.push(( + did.clone(), + "app.bsky.feed.post".to_string(), + rkey, + rec.value, + )); + } + } + Err(e) => { + tracing::warn!("failed to list posts for {did}: {e}"); } } // get reposts - if let Ok(record_list) = followed_repo + match followed_repo .list_records("app.bsky.feed.repost", limit, true, None) .await { - for rec in record_list.records { - let rkey = rec.rkey.as_str().to_string(); - all_items.push(( - did.clone(), - "app.bsky.feed.repost".to_string(), - rkey, - rec.value, - )); + Ok(record_list) => { + for rec in record_list.records { + let rkey = rec.rkey.as_str().to_string(); + all_items.push(( + did.clone(), + "app.bsky.feed.repost".to_string(), + rkey, + rec.value, + )); + } + } + Err(e) => { + tracing::warn!("failed to list reposts for {did}: {e}"); } } } @@ -1588,8 +1789,9 @@ async fn get_timeline( let val_json = serde_json::to_value(&value).unwrap_or(serde_json::json!({})); if col == "app.bsky.feed.post" { - if let Ok(post_view) = get_post_view(&app_state, &uri, None).await { - feed.push(serde_json::json!({ "post": post_view })); + match get_post_view(&app_state, &uri, None).await { + Ok(post_view) => feed.push(serde_json::json!({ "post": post_view })), + Err(e) => tracing::warn!("failed to get post view for {uri}: {e}"), } } else if col == "app.bsky.feed.repost" { if let Some(subject_uri) = val_json @@ -1597,18 +1799,26 @@ async fn get_timeline( .and_then(|s| s.get("uri")) .and_then(|u| u.as_str()) { - if let Ok(post_view) = get_post_view(&app_state, subject_uri, None).await { - if let Ok(reposter_profile) = - get_profile_internal(&app_state, did.as_str(), None).await - { - feed.push(serde_json::json!({ - "post": post_view, - "reason": { - "$type": "app.bsky.feed.defs#reasonRepost", - "by": reposter_profile, - "indexedAt": val_json.get("createdAt").and_then(|c| c.as_str()).map(|s| s.to_string()).unwrap_or_else(|| chrono::Utc::now().to_rfc3339()), + match get_post_view(&app_state, subject_uri, None).await { + Ok(post_view) => { + match get_profile_internal(&app_state, did.as_str(), None).await { + Ok(reposter_profile) => { + feed.push(serde_json::json!({ + "post": post_view, + "reason": { + "$type": "app.bsky.feed.defs#reasonRepost", + "by": reposter_profile, + "indexedAt": val_json.get("createdAt").and_then(|c| c.as_str()).map(|s| s.to_string()).unwrap_or_else(|| chrono::Utc::now().to_rfc3339()), + } + })); } - })); + Err(e) => { + tracing::warn!("failed to get profile for reposter {did}: {e}"); + } + } + } + Err(e) => { + tracing::warn!("failed to get post view for reposted post {subject_uri}: {e}"); } } } @@ -1639,9 +1849,10 @@ async fn get_quotes( ) -> Result { // let hydrant = &app_state.hydrant; let query_str = req.uri().query().unwrap_or(""); - let Ok(params) = serde_urlencoded::from_str::(query_str) else { - return Err(StatusCode::BAD_REQUEST); - }; + let params = serde_urlencoded::from_str::(query_str).map_err(|e| { + tracing::error!("failed to parse get_quotes query params: {e}"); + StatusCode::BAD_REQUEST + })?; let limit = params.limit.unwrap_or(50).min(100); @@ -1652,13 +1863,22 @@ async fn get_quotes( .source("app.bsky.feed.post") .limit(limit * 3); if let Some(cursor_str) = params.cursor { - if let Ok(c) = data_encoding::BASE64URL_NOPAD.decode(cursor_str.as_bytes()) { - fetch = fetch.cursor(c); + match data_encoding::BASE64URL_NOPAD.decode(cursor_str.as_bytes()) { + Ok(c) => { + fetch = fetch.cursor(c); + } + Err(e) => { + tracing::error!("failed to decode cursor {cursor_str}: {e}"); + } } } - let Ok(backlinks_page) = fetch.run().await else { - return proxy_request(req).await; + let backlinks_page = match fetch.run().await { + Ok(bp) => bp, + Err(e) => { + tracing::error!("failed to fetch backlinks for {}: {e}", params.uri); + return proxy_request(req).await; + } }; let mut posts = Vec::new(); @@ -1668,18 +1888,31 @@ async fn get_quotes( break; } - let Ok(uri) = jacquard_common::types::string::AtUri::new(bl.uri.as_str()) else { - continue; + let uri = match jacquard_common::types::string::AtUri::new(bl.uri.as_str()) { + Ok(u) => u, + Err(e) => { + tracing::error!("failed to parse backlink URI {}: {e}", bl.uri); + continue; + } }; let author_ident = uri.authority(); let collection = uri.collection().unwrap().as_str(); let rkey = uri.rkey().unwrap().0.as_str(); - let Ok(repo) = app_state.hydrant.repos.resolve(author_ident).await else { - continue; + let repo = match app_state.hydrant.repos.resolve(author_ident).await { + Ok(r) => r, + Err(e) => { + tracing::error!("failed to resolve repo for {author_ident}: {e}"); + continue; + } }; - let Ok(Some(record)) = repo.get_record(collection, rkey).await else { - continue; + let record = match repo.get_record(collection, rkey).await { + Ok(Some(rec)) => rec, + Ok(None) => continue, + Err(e) => { + tracing::error!("failed to get record {collection}/{rkey} from repo {author_ident}: {e}"); + continue; + } }; let value = serde_json::to_value(&record.value).unwrap_or(serde_json::json!({})); @@ -1704,9 +1937,14 @@ async fn get_quotes( }); if is_quote { - if let Ok(post_view) = get_post_view(&app_state, bl.uri.as_str(), None).await { - posts.push(post_view); - found += 1; + match get_post_view(&app_state, bl.uri.as_str(), None).await { + Ok(post_view) => { + posts.push(post_view); + found += 1; + } + Err(e) => { + tracing::warn!("failed to get post view for quoted post {}: {e}", bl.uri); + } } } } @@ -1738,24 +1976,34 @@ async fn get_actor_likes( ) -> Result { // let hydrant = &app_state.hydrant; let query_str = req.uri().query().unwrap_or(""); - let Ok(params) = serde_urlencoded::from_str::(query_str) else { - return Err(StatusCode::BAD_REQUEST); - }; - - let Ok(ident) = AtIdentifier::new(¶ms.actor) else { - return Err(StatusCode::BAD_REQUEST); - }; - let Ok(repo) = app_state.hydrant.repos.resolve(&ident).await else { - return proxy_request(req).await; + let params = serde_urlencoded::from_str::(query_str).map_err(|e| { + tracing::error!("failed to parse get_actor_likes query params: {e}"); + StatusCode::BAD_REQUEST + })?; + + let ident = AtIdentifier::new(¶ms.actor).map_err(|e| { + tracing::error!("failed to create AtIdentifier for actor {}: {e}", params.actor); + StatusCode::BAD_REQUEST + })?; + let repo = match app_state.hydrant.repos.resolve(&ident).await { + Ok(r) => r, + Err(e) => { + tracing::error!("failed to resolve repo for {ident}: {e}"); + return proxy_request(req).await; + } }; let limit = params.limit.unwrap_or(50).min(100); - let Ok(record_list) = repo + let record_list = match repo .list_records("app.bsky.feed.like", limit, true, params.cursor.as_deref()) .await - else { - return proxy_request(req).await; + { + Ok(rl) => rl, + Err(e) => { + tracing::error!("failed to list likes for {}: {e}", repo.did); + return proxy_request(req).await; + } }; let mut feed = Vec::new(); @@ -1766,8 +2014,9 @@ async fn get_actor_likes( .and_then(|s| s.get("uri")) .and_then(|u| u.as_str()) { - if let Ok(post_view) = get_post_view(&app_state, subject_uri, None).await { - feed.push(serde_json::json!({ "post": post_view })); + match get_post_view(&app_state, subject_uri, None).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}"), } } } @@ -1790,30 +2039,44 @@ async fn get_relationships( ) -> Result { // let hydrant = &app_state.hydrant; let query_str = req.uri().query().unwrap_or(""); - let Ok(params) = serde_urlencoded::from_str::(query_str) else { - return Err(StatusCode::BAD_REQUEST); - }; + let params = serde_urlencoded::from_str::(query_str).map_err(|e| { + tracing::error!("failed to parse get_relationships query params: {e}"); + StatusCode::BAD_REQUEST + })?; let others = extract_query_array(query_str, "others"); if others.is_empty() { return proxy_request(req).await; } - let Ok(ident) = AtIdentifier::new(¶ms.actor) else { - return Err(StatusCode::BAD_REQUEST); - }; - let Ok(repo) = app_state.hydrant.repos.resolve(&ident).await else { - return proxy_request(req).await; + let ident = AtIdentifier::new(¶ms.actor).map_err(|e| { + tracing::error!("failed to create AtIdentifier for actor {}: {e}", params.actor); + StatusCode::BAD_REQUEST + })?; + let repo = match app_state.hydrant.repos.resolve(&ident).await { + Ok(r) => r, + Err(e) => { + tracing::error!("failed to resolve repo for {ident}: {e}"); + return proxy_request(req).await; + } }; let actor_did = repo.did.as_str().to_string(); let mut relationships = Vec::new(); for other in others { - let Ok(other_ident) = AtIdentifier::new(&other) else { - continue; + let other_ident = match AtIdentifier::new(&other) { + Ok(i) => i, + Err(e) => { + tracing::error!("failed to create AtIdentifier for other actor {other}: {e}"); + continue; + } }; - let Ok(other_repo) = app_state.hydrant.repos.resolve(&other_ident).await else { - continue; + let other_repo = match app_state.hydrant.repos.resolve(&other_ident).await { + Ok(r) => r, + Err(e) => { + tracing::error!("failed to resolve repo for other actor {other_ident}: {e}"); + continue; + } }; let other_did = other_repo.did.as_str().to_string(); @@ -1822,53 +2085,61 @@ async fn get_relationships( let mut cursor = None; loop { - if let Ok(rl) = repo + match repo .list_records("app.bsky.graph.follow", 100, true, cursor.as_deref()) .await { - for rec in &rl.records { - let val = serde_json::to_value(&rec.value).unwrap_or(serde_json::json!({})); - if val.get("subject").and_then(|s| s.as_str()) == Some(&other_did) { - following = Some(format!( - "at://{}/app.bsky.graph.follow/{}", - actor_did, - rec.rkey.as_str() - )); + Ok(rl) => { + for rec in &rl.records { + let val = serde_json::to_value(&rec.value).unwrap_or(serde_json::json!({})); + if val.get("subject").and_then(|s| s.as_str()) == Some(&other_did) { + following = Some(format!( + "at://{}/app.bsky.graph.follow/{}", + actor_did, + rec.rkey.as_str() + )); + break; + } + } + if following.is_some() || rl.cursor.is_none() { break; } + cursor = rl.cursor.map(|c| c.to_string()); } - if following.is_some() || rl.cursor.is_none() { + Err(e) => { + tracing::error!("failed to list follows for {actor_did}: {e}"); break; } - cursor = rl.cursor.map(|c| c.to_string()); - } else { - break; } } let mut cursor = None; loop { - if let Ok(rl) = other_repo + match other_repo .list_records("app.bsky.graph.follow", 100, true, cursor.as_deref()) .await { - for rec in &rl.records { - let val = serde_json::to_value(&rec.value).unwrap_or(serde_json::json!({})); - if val.get("subject").and_then(|s| s.as_str()) == Some(&actor_did) { - followed_by = Some(format!( - "at://{}/app.bsky.graph.follow/{}", - other_did, - rec.rkey.as_str() - )); + Ok(rl) => { + for rec in &rl.records { + let val = serde_json::to_value(&rec.value).unwrap_or(serde_json::json!({})); + if val.get("subject").and_then(|s| s.as_str()) == Some(&actor_did) { + followed_by = Some(format!( + "at://{}/app.bsky.graph.follow/{}", + other_did, + rec.rkey.as_str() + )); + break; + } + } + if followed_by.is_some() || rl.cursor.is_none() { break; } + cursor = rl.cursor.map(|c| c.to_string()); } - if followed_by.is_some() || rl.cursor.is_none() { + Err(e) => { + tracing::error!("failed to list follows for {other_did}: {e}"); break; } - cursor = rl.cursor.map(|c| c.to_string()); - } else { - break; } } @@ -1893,47 +2164,62 @@ async fn get_known_followers( ) -> Result { // let hydrant = &app_state.hydrant; let query_str = req.uri().query().unwrap_or(""); - let Ok(params) = serde_urlencoded::from_str::(query_str) else { - return Err(StatusCode::BAD_REQUEST); - }; + let params = serde_urlencoded::from_str::(query_str).map_err(|e| { + tracing::error!("failed to parse get_known_followers query params: {e}"); + StatusCode::BAD_REQUEST + })?; let Some(did_str) = get_auth_did(&req) else { return Err(StatusCode::UNAUTHORIZED); }; - let Ok(auth_ident) = AtIdentifier::new(&did_str) else { - return Err(StatusCode::BAD_REQUEST); - }; - let Ok(auth_repo) = app_state.hydrant.repos.resolve(&auth_ident).await else { - return proxy_request(req).await; + let auth_ident = AtIdentifier::new(&did_str).map_err(|e| { + tracing::error!("failed to create AtIdentifier for auth user {did_str}: {e}"); + StatusCode::BAD_REQUEST + })?; + let auth_repo = match app_state.hydrant.repos.resolve(&auth_ident).await { + Ok(r) => r, + Err(e) => { + tracing::error!("failed to resolve repo for auth user {auth_ident}: {e}"); + return proxy_request(req).await; + } }; let mut auth_follows = std::collections::HashSet::new(); let mut cursor = None; loop { - if let Ok(rl) = auth_repo + match auth_repo .list_records("app.bsky.graph.follow", 100, true, cursor.as_deref()) .await { - for rec in &rl.records { - let val = serde_json::to_value(&rec.value).unwrap_or(serde_json::json!({})); - if let Some(subj) = val.get("subject").and_then(|s| s.as_str()) { - auth_follows.insert(subj.to_string()); + Ok(rl) => { + for rec in &rl.records { + let val = serde_json::to_value(&rec.value).unwrap_or(serde_json::json!({})); + if let Some(subj) = val.get("subject").and_then(|s| s.as_str()) { + auth_follows.insert(subj.to_string()); + } + } + if rl.cursor.is_none() { + break; } + cursor = rl.cursor.map(|c| c.to_string()); } - if rl.cursor.is_none() { + Err(e) => { + tracing::error!("failed to list follows for auth user {}: {e}", auth_repo.did); break; } - cursor = rl.cursor.map(|c| c.to_string()); - } else { - break; } } - let Ok(ident) = AtIdentifier::new(¶ms.actor) else { - return Err(StatusCode::BAD_REQUEST); - }; - let Ok(repo) = app_state.hydrant.repos.resolve(&ident).await else { - return proxy_request(req).await; + let ident = AtIdentifier::new(¶ms.actor).map_err(|e| { + tracing::error!("failed to create AtIdentifier for actor {}: {e}", params.actor); + StatusCode::BAD_REQUEST + })?; + let repo = match app_state.hydrant.repos.resolve(&ident).await { + Ok(r) => r, + Err(e) => { + tracing::error!("failed to resolve repo for {ident}: {e}"); + return proxy_request(req).await; + } }; let limit = params.limit.unwrap_or(50).min(100); @@ -1945,13 +2231,22 @@ async fn get_known_followers( .source("app.bsky.graph.follow") .limit(limit * 3); if let Some(cursor_str) = params.cursor { - if let Ok(c) = data_encoding::BASE64URL_NOPAD.decode(cursor_str.as_bytes()) { - fetch = fetch.cursor(c); + match data_encoding::BASE64URL_NOPAD.decode(cursor_str.as_bytes()) { + Ok(c) => { + fetch = fetch.cursor(c); + } + Err(e) => { + tracing::error!("failed to decode cursor {cursor_str}: {e}"); + } } } - let Ok(backlinks_page) = fetch.run().await else { - return proxy_request(req).await; + let backlinks_page = match fetch.run().await { + Ok(bp) => bp, + Err(e) => { + tracing::error!("failed to fetch backlinks for {}: {e}", repo.did); + return proxy_request(req).await; + } }; let mut followers = Vec::new(); @@ -1960,23 +2255,34 @@ async fn get_known_followers( if found >= limit { break; } - let Ok(uri) = jacquard_common::types::string::AtUri::new(bl.uri.as_str()) else { - continue; + let uri = match jacquard_common::types::string::AtUri::new(bl.uri.as_str()) { + Ok(u) => u, + Err(e) => { + tracing::error!("failed to parse backlink URI {}: {e}", bl.uri); + continue; + } }; let author_ident = uri.authority().as_str().to_string(); if auth_follows.contains(&author_ident) { - if let Ok(profile) = get_profile_internal(&app_state, author_ident.as_str(), None).await - { - followers.push(profile); - found += 1; + match get_profile_internal(&app_state, author_ident.as_str(), None).await { + Ok(profile) => { + followers.push(profile); + found += 1; + } + Err(e) => { + tracing::warn!("failed to get profile for follower {author_ident}: {e}"); + } } } } - let Ok(subject_profile) = get_profile_internal(&app_state, repo.did.as_str(), None).await - else { - return proxy_request(req).await; + 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 subject profile for {}: {e}", repo.did); + return proxy_request(req).await; + } }; let cursor = backlinks_page @@ -1997,22 +2303,28 @@ async fn get_blocks( ) -> Result { // let hydrant = &app_state.hydrant; let query_str = req.uri().query().unwrap_or(""); - let Ok(params) = serde_urlencoded::from_str::(query_str) else { - return Err(StatusCode::BAD_REQUEST); - }; + let params = serde_urlencoded::from_str::(query_str).map_err(|e| { + tracing::error!("failed to parse get_blocks query params: {e}"); + StatusCode::BAD_REQUEST + })?; let Some(did_str) = get_auth_did(&req) else { return Err(StatusCode::UNAUTHORIZED); }; - let Ok(ident) = AtIdentifier::new(&did_str) else { - return Err(StatusCode::BAD_REQUEST); - }; - let Ok(repo) = app_state.hydrant.repos.resolve(&ident).await else { - return proxy_request(req).await; + let ident = AtIdentifier::new(&did_str).map_err(|e| { + tracing::error!("failed to create AtIdentifier for did_str {did_str}: {e}"); + StatusCode::BAD_REQUEST + })?; + let repo = match app_state.hydrant.repos.resolve(&ident).await { + Ok(r) => r, + Err(e) => { + tracing::error!("failed to resolve repo for {ident}: {e}"); + return proxy_request(req).await; + } }; let limit = params.limit.unwrap_or(50).min(100); - let Ok(record_list) = repo + let record_list = match repo .list_records( "app.bsky.graph.block", limit, @@ -2020,16 +2332,21 @@ async fn get_blocks( params.cursor.as_deref(), ) .await - else { - return proxy_request(req).await; + { + Ok(rl) => rl, + Err(e) => { + tracing::error!("failed to list blocks for {}: {e}", repo.did); + return proxy_request(req).await; + } }; let mut blocks = Vec::new(); 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()) { - if let Ok(profile) = get_profile_internal(&app_state, subject_did, None).await { - blocks.push(profile); + 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}"), } } } @@ -2052,31 +2369,52 @@ async fn get_list(State(app_state): State, req: Request) -> Result, } - let Ok(params) = serde_urlencoded::from_str::(query_str) else { - return Err(StatusCode::BAD_REQUEST); - }; + let params = serde_urlencoded::from_str::(query_str).map_err(|e| { + tracing::error!("failed to parse get_list query params: {e}"); + StatusCode::BAD_REQUEST + })?; - let Ok(uri) = jacquard_common::types::string::AtUri::new(¶ms.list) else { - return Err(StatusCode::BAD_REQUEST); - }; + let uri = jacquard_common::types::string::AtUri::new(¶ms.list).map_err(|e| { + tracing::error!("failed to parse list URI {}: {e}", params.list); + StatusCode::BAD_REQUEST + })?; let author_ident = uri.authority(); let rkey = uri .rkey() - .ok_or(StatusCode::BAD_REQUEST)? + .ok_or_else(|| { + tracing::error!("missing rkey in list URI {}", params.list); + StatusCode::BAD_REQUEST + })? .0 .as_str() .to_string(); - let Ok(repo) = app_state.hydrant.repos.resolve(author_ident).await else { - return proxy_request(req).await; + let repo = match app_state.hydrant.repos.resolve(author_ident).await { + Ok(r) => r, + Err(e) => { + tracing::error!("failed to resolve repo for {author_ident}: {e}"); + return proxy_request(req).await; + } }; - let Ok(Some(record)) = repo.get_record("app.bsky.graph.list", &rkey).await else { - return proxy_request(req).await; + let record = match repo.get_record("app.bsky.graph.list", &rkey).await { + Ok(Some(rec)) => rec, + Ok(None) => { + tracing::error!("list record not found: {author_ident}/app.bsky.graph.list/{rkey}"); + return proxy_request(req).await; + } + Err(e) => { + tracing::error!("failed to get list record {author_ident}/{rkey}: {e}"); + return proxy_request(req).await; + } }; - let Ok(author_profile) = get_profile_internal(&app_state, repo.did.as_str(), None).await else { - return proxy_request(req).await; + 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 val_json = serde_json::to_value(&record.value).unwrap_or(serde_json::json!({})); @@ -2101,40 +2439,67 @@ async fn get_list(State(app_state): State, req: Request) -> Result { + fetch = fetch.cursor(c); + } + Err(e) => { + tracing::error!("failed to decode cursor {cursor_str}: {e}"); + } } } - let Ok(backlinks_page) = fetch.run().await else { - return proxy_request(req).await; + let backlinks_page = match fetch.run().await { + Ok(bp) => bp, + Err(e) => { + tracing::error!("failed to fetch backlinks for list {}: {e}", params.list); + return proxy_request(req).await; + } }; let mut items = Vec::new(); for bl in backlinks_page.backlinks { - let Ok(item_uri) = jacquard_common::types::string::AtUri::new(bl.uri.as_str()) else { - continue; + let item_uri = match jacquard_common::types::string::AtUri::new(bl.uri.as_str()) { + Ok(u) => u, + Err(e) => { + tracing::error!("failed to parse listitem URI {}: {e}", bl.uri); + continue; + } }; let item_author = item_uri.authority(); let item_rkey = item_uri.rkey().unwrap().0.as_str(); - let Ok(item_repo) = app_state.hydrant.repos.resolve(item_author).await else { - continue; + let item_repo = match app_state.hydrant.repos.resolve(item_author).await { + Ok(r) => r, + Err(e) => { + tracing::error!("failed to resolve repo for listitem author {item_author}: {e}"); + continue; + } }; - let Ok(Some(item_rec)) = item_repo + let item_rec = match item_repo .get_record("app.bsky.graph.listitem", item_rkey) .await - else { - continue; + { + Ok(Some(rec)) => rec, + Ok(None) => continue, + Err(e) => { + tracing::error!("failed to get listitem record {item_author}/{item_rkey}: {e}"); + continue; + } }; let item_val = serde_json::to_value(&item_rec.value).unwrap_or(serde_json::json!({})); if let Some(subject_did) = item_val.get("subject").and_then(|s| s.as_str()) { - if let Ok(subject_profile) = get_profile_internal(&app_state, subject_did, None).await { - items.push(serde_json::json!({ - "uri": bl.uri.as_str(), - "subject": subject_profile, - })); + match get_profile_internal(&app_state, subject_did, None).await { + Ok(subject_profile) => { + items.push(serde_json::json!({ + "uri": bl.uri.as_str(), + "subject": subject_profile, + })); + } + Err(e) => { + tracing::warn!("failed to get profile for listitem subject {subject_did}: {e}"); + } } } } @@ -2153,27 +2518,41 @@ async fn get_lists( ) -> Result { // let hydrant = &app_state.hydrant; let query_str = req.uri().query().unwrap_or(""); - let Ok(params) = serde_urlencoded::from_str::(query_str) else { - return Err(StatusCode::BAD_REQUEST); - }; - - let Ok(ident) = AtIdentifier::new(¶ms.actor) else { - return Err(StatusCode::BAD_REQUEST); - }; - let Ok(repo) = app_state.hydrant.repos.resolve(&ident).await else { - return proxy_request(req).await; + let params = serde_urlencoded::from_str::(query_str).map_err(|e| { + tracing::error!("failed to parse get_lists query params: {e}"); + StatusCode::BAD_REQUEST + })?; + + let ident = AtIdentifier::new(¶ms.actor).map_err(|e| { + tracing::error!("failed to create AtIdentifier for actor {}: {e}", params.actor); + StatusCode::BAD_REQUEST + })?; + let repo = match app_state.hydrant.repos.resolve(&ident).await { + Ok(r) => r, + Err(e) => { + tracing::error!("failed to resolve repo for {ident}: {e}"); + return proxy_request(req).await; + } }; - let Ok(author_profile) = get_profile_internal(&app_state, repo.did.as_str(), None).await else { - return proxy_request(req).await; + 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 limit = params.limit.unwrap_or(50).min(100); - let Ok(record_list) = repo + let record_list = match repo .list_records("app.bsky.graph.list", limit, true, params.cursor.as_deref()) .await - else { - return proxy_request(req).await; + { + Ok(rl) => rl, + Err(e) => { + tracing::error!("failed to list lists for {}: {e}", repo.did); + return proxy_request(req).await; + } }; let mut lists = Vec::new(); @@ -2204,23 +2583,33 @@ async fn get_lists_with_membership( ) -> Result { // let hydrant = &app_state.hydrant; let query_str = req.uri().query().unwrap_or(""); - let Ok(params) = serde_urlencoded::from_str::(query_str) else { - return Err(StatusCode::BAD_REQUEST); - }; - - let Ok(ident) = AtIdentifier::new(¶ms.actor) else { - return Err(StatusCode::BAD_REQUEST); - }; - let Ok(repo) = app_state.hydrant.repos.resolve(&ident).await else { - return proxy_request(req).await; + let params = serde_urlencoded::from_str::(query_str).map_err(|e| { + tracing::error!("failed to parse get_lists_with_membership query params: {e}"); + StatusCode::BAD_REQUEST + })?; + + let ident = AtIdentifier::new(¶ms.actor).map_err(|e| { + tracing::error!("failed to create AtIdentifier for actor {}: {e}", params.actor); + StatusCode::BAD_REQUEST + })?; + let repo = match app_state.hydrant.repos.resolve(&ident).await { + Ok(r) => r, + Err(e) => { + tracing::error!("failed to resolve repo for {ident}: {e}"); + return proxy_request(req).await; + } }; // All lists created by actor - let Ok(record_list) = repo + let record_list = match repo .list_records("app.bsky.graph.list", 100, true, None) .await - else { - return proxy_request(req).await; + { + Ok(rl) => rl, + Err(e) => { + tracing::error!("failed to list lists for {}: {e}", repo.did); + return proxy_request(req).await; + } }; let mut lists: Vec = Vec::new(); @@ -2247,23 +2636,33 @@ async fn get_actor_feeds( ) -> Result { // let hydrant = &app_state.hydrant; let query_str = req.uri().query().unwrap_or(""); - let Ok(params) = serde_urlencoded::from_str::(query_str) else { - return Err(StatusCode::BAD_REQUEST); - }; - - let Ok(ident) = AtIdentifier::new(¶ms.actor) else { - return Err(StatusCode::BAD_REQUEST); - }; - let Ok(repo) = app_state.hydrant.repos.resolve(&ident).await else { - return proxy_request(req).await; + let params: GetAuthorFeedParams = serde_urlencoded::from_str(query_str).map_err(|e| { + tracing::error!("failed to parse get_actor_feeds query params: {e}"); + StatusCode::BAD_REQUEST + })?; + + let ident = AtIdentifier::new(¶ms.actor).map_err(|e| { + tracing::error!("failed to create AtIdentifier for actor {}: {e}", params.actor); + StatusCode::BAD_REQUEST + })?; + let repo = match app_state.hydrant.repos.resolve(&ident).await { + Ok(r) => r, + Err(e) => { + tracing::error!("failed to resolve repo for {ident}: {e}"); + return proxy_request(req).await; + } }; - let Ok(author_profile) = get_profile_internal(&app_state, repo.did.as_str(), None).await else { - return proxy_request(req).await; + 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 limit = params.limit.unwrap_or(50).min(100); - let Ok(record_list) = repo + let record_list = match repo .list_records( "app.bsky.feed.generator", limit, @@ -2271,22 +2670,31 @@ async fn get_actor_feeds( params.cursor.as_deref(), ) .await - else { - return proxy_request(req).await; + { + Ok(rl) => rl, + Err(e) => { + tracing::error!("failed to list feed generators for {}: {e}", repo.did); + return proxy_request(req).await; + } }; let mut feeds = Vec::new(); for rec in record_list.records { let val_json = serde_json::to_value(&rec.value).unwrap_or(serde_json::json!({})); + let uri = format!( + "at://{}/app.bsky.feed.generator/{}", + repo.did.as_str(), + rec.rkey.as_str() + ); feeds.push(serde_json::json!({ - "uri": format!("at://{}/app.bsky.feed.generator/{}", repo.did.as_str(), rec.rkey.as_str()), + "uri": uri, "cid": rec.cid.to_string(), "did": val_json.get("did").and_then(|v| v.as_str()).unwrap_or(""), "creator": author_profile.clone(), "displayName": val_json.get("displayName").and_then(|v| v.as_str()).unwrap_or(""), "description": val_json.get("description").and_then(|v| v.as_str()), "avatar": val_json.get("avatar").and_then(|v| v.as_str()), - "likeCount": app_state.hydrant.backlinks.count(format!("at://{}/app.bsky.feed.generator/{}", repo.did.as_str(), rec.rkey.as_str())).source("app.bsky.feed.like").run().await.unwrap_or(0), + "likeCount": app_state.hydrant.backlinks.count(uri).source("app.bsky.feed.like").run().await.unwrap_or(0), "indexedAt": chrono::Utc::now().to_rfc3339(), })); } @@ -2308,31 +2716,52 @@ async fn get_feed_generator( struct GetFeedGeneratorParams { feed: String, } - let Ok(params) = serde_urlencoded::from_str::(query_str) else { - return Err(StatusCode::BAD_REQUEST); - }; + let params: GetFeedGeneratorParams = serde_urlencoded::from_str(query_str).map_err(|e| { + tracing::error!("failed to parse get_feed_generator query params: {e}"); + StatusCode::BAD_REQUEST + })?; - let Ok(uri) = jacquard_common::types::string::AtUri::new(¶ms.feed) else { - return Err(StatusCode::BAD_REQUEST); - }; + let uri = jacquard_common::types::string::AtUri::new(¶ms.feed).map_err(|e| { + tracing::error!("failed to parse feed uri {}: {e}", params.feed); + StatusCode::BAD_REQUEST + })?; let author_ident = uri.authority(); let rkey = uri .rkey() - .ok_or(StatusCode::BAD_REQUEST)? + .ok_or_else(|| { + tracing::error!("missing rkey in feed uri {}", params.feed); + StatusCode::BAD_REQUEST + })? .0 .as_str() .to_string(); - let Ok(repo) = app_state.hydrant.repos.resolve(author_ident).await else { - return proxy_request(req).await; + let repo = match app_state.hydrant.repos.resolve(author_ident).await { + Ok(r) => r, + Err(e) => { + tracing::error!("failed to resolve repo for {author_ident}: {e}"); + return proxy_request(req).await; + } }; - let Ok(Some(record)) = repo.get_record("app.bsky.feed.generator", &rkey).await else { - return proxy_request(req).await; + let record = match repo.get_record("app.bsky.feed.generator", &rkey).await { + Ok(Some(r)) => r, + Ok(None) => { + tracing::error!("feed generator record not found: {author_ident}/{rkey}"); + return proxy_request(req).await; + } + Err(e) => { + tracing::error!("failed to get feed generator record {author_ident}/{rkey}: {e}"); + return proxy_request(req).await; + } }; - let Ok(author_profile) = get_profile_internal(&app_state, repo.did.as_str(), None).await else { - return proxy_request(req).await; + 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 val_json = serde_json::to_value(&record.value).unwrap_or(serde_json::json!({})); @@ -2370,32 +2799,48 @@ async fn get_feed_generators( let mut views = Vec::new(); for feed_uri in feeds { - let Ok(uri) = jacquard_common::types::string::AtUri::new(&feed_uri) else { - continue; + let uri = match jacquard_common::types::string::AtUri::new(&feed_uri) { + Ok(u) => u, + Err(e) => { + tracing::warn!("failed to parse feed uri {feed_uri}: {e}"); + continue; + } }; let author_ident = uri.authority(); let rkey = uri.rkey().unwrap().0.as_str().to_string(); - if let Ok(repo) = app_state.hydrant.repos.resolve(author_ident).await { - if let Ok(Some(record)) = repo.get_record("app.bsky.feed.generator", &rkey).await { - if let Ok(author_profile) = - get_profile_internal(&app_state, repo.did.as_str(), None).await - { - 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(""), - "creator": author_profile, - "displayName": val_json.get("displayName").and_then(|v| v.as_str()).unwrap_or(""), - "description": val_json.get("description").and_then(|v| v.as_str()), - "avatar": val_json.get("avatar").and_then(|v| v.as_str()), - "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(), - })); + 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!({ + "uri": feed_uri, + "cid": record.cid.to_string(), + "did": val_json.get("did").and_then(|v| v.as_str()).unwrap_or(""), + "creator": author_profile, + "displayName": val_json.get("displayName").and_then(|v| v.as_str()).unwrap_or(""), + "description": val_json.get("description").and_then(|v| v.as_str()), + "avatar": val_json.get("avatar").and_then(|v| v.as_str()), + "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) + } + } + } + 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}"), } } @@ -2418,9 +2863,10 @@ async fn put_preferences( let body_bytes = axum::body::to_bytes(req.into_body(), usize::MAX) .await .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; - let Ok(payload) = serde_json::from_slice::(&body_bytes) else { - return Err(StatusCode::BAD_REQUEST); - }; + let payload: PutPreferencesReq = serde_json::from_slice(&body_bytes).map_err(|e| { + tracing::error!("failed to parse put_preferences request: {e}"); + StatusCode::BAD_REQUEST + })?; let key = format!("prefs:{}", did); let val = @@ -2473,15 +2919,21 @@ async fn mute_actor( let body_bytes = axum::body::to_bytes(req.into_body(), usize::MAX) .await .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; - let Ok(payload) = serde_json::from_slice::(&body_bytes) else { - return Err(StatusCode::BAD_REQUEST); - }; - - let Ok(ident) = AtIdentifier::new(&payload.actor) else { - return Err(StatusCode::BAD_REQUEST); - }; - let Ok(repo) = app_state.hydrant.repos.resolve(&ident).await else { - return Err(StatusCode::NOT_FOUND); + let payload: MuteActorReq = serde_json::from_slice(&body_bytes).map_err(|e| { + tracing::error!("failed to parse mute_actor/unmute_actor request: {e}"); + StatusCode::BAD_REQUEST + })?; + + let ident = AtIdentifier::new(&payload.actor).map_err(|e| { + tracing::error!("failed to parse actor identifier {}: {e}", payload.actor); + StatusCode::BAD_REQUEST + })?; + let repo = match app_state.hydrant.repos.resolve(&ident).await { + Ok(r) => r, + Err(e) => { + tracing::error!("failed to resolve repo for {ident}: {e}"); + return Err(StatusCode::NOT_FOUND); + } }; let key = format!("mute:{}:{}", did, repo.did.as_str()); @@ -2505,15 +2957,21 @@ async fn unmute_actor( let body_bytes = axum::body::to_bytes(req.into_body(), usize::MAX) .await .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; - let Ok(payload) = serde_json::from_slice::(&body_bytes) else { - return Err(StatusCode::BAD_REQUEST); - }; - - let Ok(ident) = AtIdentifier::new(&payload.actor) else { - return Err(StatusCode::BAD_REQUEST); - }; - let Ok(repo) = app_state.hydrant.repos.resolve(&ident).await else { - return Err(StatusCode::NOT_FOUND); + let payload: MuteActorReq = serde_json::from_slice(&body_bytes).map_err(|e| { + tracing::error!("failed to parse mute_actor/unmute_actor request: {e}"); + StatusCode::BAD_REQUEST + })?; + + let ident = AtIdentifier::new(&payload.actor).map_err(|e| { + tracing::error!("failed to parse actor identifier {}: {e}", payload.actor); + StatusCode::BAD_REQUEST + })?; + let repo = match app_state.hydrant.repos.resolve(&ident).await { + Ok(r) => r, + Err(e) => { + tracing::error!("failed to resolve repo for {ident}: {e}"); + return Err(StatusCode::NOT_FOUND); + } }; let key = format!("mute:{}:{}", did, repo.did.as_str()); @@ -2535,9 +2993,10 @@ async fn get_mutes( }; let query_str = req.uri().query().unwrap_or(""); - let Ok(params) = serde_urlencoded::from_str::(query_str) else { - return Err(StatusCode::BAD_REQUEST); - }; + let params: GetBookmarksParams = serde_urlencoded::from_str(query_str).map_err(|e| { + tracing::error!("failed to parse get_mutes query params: {e}"); + StatusCode::BAD_REQUEST + })?; let limit = params.limit.unwrap_or(50).min(100); let prefix = format!("mute:{}:", did); @@ -2553,8 +3012,12 @@ async fn get_mutes( let mut fetched_keys = Vec::new(); for item in iter { - let Ok((key, _)) = item.into_inner() else { - continue; + let (key, _) = match item.into_inner() { + Ok(inner) => inner, + Err(e) => { + tracing::warn!("failed to get item from mutes iterator: {e}"); + continue; + } }; if !key.starts_with(prefix.as_bytes()) { break; @@ -2570,8 +3033,9 @@ async fn get_mutes( for key in fetched_keys { let key_str = String::from_utf8_lossy(&key); let muted_did = &key_str[prefix.len()..]; - if let Ok(profile) = get_profile_internal(&app_state, muted_did, None).await { - mutes.push(profile); + 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}"), } next_cursor = Some(key_str.to_string()); } @@ -2607,12 +3071,15 @@ async fn notification_indexer(app_state: AppState) { .and_then(|s| s.get("uri")) .and_then(|u| u.as_str()) { - if let Ok(u) = jacquard_common::types::string::AtUri::new(subj_uri) { - targets.push(( - u.authority().as_str().to_string(), - "like", - Some(subj_uri.to_string()), - )); + match jacquard_common::types::string::AtUri::new(subj_uri) { + Ok(u) => { + targets.push(( + u.authority().as_str().to_string(), + "like", + Some(subj_uri.to_string()), + )); + } + Err(e) => tracing::warn!("failed to parse notification like uri {subj_uri}: {e}"), } } } @@ -2622,12 +3089,15 @@ async fn notification_indexer(app_state: AppState) { .and_then(|s| s.get("uri")) .and_then(|u| u.as_str()) { - if let Ok(u) = jacquard_common::types::string::AtUri::new(subj_uri) { - targets.push(( - u.authority().as_str().to_string(), - "repost", - Some(subj_uri.to_string()), - )); + match jacquard_common::types::string::AtUri::new(subj_uri) { + Ok(u) => { + targets.push(( + u.authority().as_str().to_string(), + "repost", + Some(subj_uri.to_string()), + )); + } + Err(e) => tracing::warn!("failed to parse notification repost uri {subj_uri}: {e}"), } } } @@ -2643,12 +3113,15 @@ async fn notification_indexer(app_state: AppState) { .and_then(|p| p.get("uri")) .and_then(|u| u.as_str()) { - if let Ok(u) = jacquard_common::types::string::AtUri::new(parent_uri) { - targets.push(( - u.authority().as_str().to_string(), - "reply", - Some(parent_uri.to_string()), - )); + match jacquard_common::types::string::AtUri::new(parent_uri) { + Ok(u) => { + targets.push(( + u.authority().as_str().to_string(), + "reply", + Some(parent_uri.to_string()), + )); + } + Err(e) => tracing::warn!("failed to parse notification reply uri {parent_uri}: {e}"), } } @@ -2691,12 +3164,15 @@ async fn notification_indexer(app_state: AppState) { }; if let Some(qu) = quote_uri { - if let Ok(u) = jacquard_common::types::string::AtUri::new(qu) { - targets.push(( - u.authority().as_str().to_string(), - "quote", - Some(qu.to_string()), - )); + match jacquard_common::types::string::AtUri::new(qu) { + Ok(u) => { + targets.push(( + u.authority().as_str().to_string(), + "quote", + Some(qu.to_string()), + )); + } + Err(e) => tracing::warn!("failed to parse notification quote uri {qu}: {e}"), } } } @@ -2779,8 +3255,13 @@ async fn list_notifications( // Collect all notification keys into a vector so we can iterate in reverse let mut all_keys = Vec::new(); for item in app_state.notifications.prefix(prefix.as_bytes()) { - if let Ok((k, v)) = item.into_inner() { - all_keys.push((k, v)); + match item.into_inner() { + Ok((k, v)) => { + all_keys.push((k, v)); + } + Err(e) => { + tracing::warn!("failed to get notification item: {e}"); + } } } @@ -2807,8 +3288,12 @@ async fn list_notifications( for (key, val_bytes) in all_keys.into_iter().take(limit) { next_cursor = Some(String::from_utf8_lossy(&key).to_string()); - let Ok(val) = serde_json::from_slice::(&val_bytes) else { - continue; + let val = match serde_json::from_slice::(&val_bytes) { + Ok(v) => v, + Err(e) => { + tracing::warn!("failed to parse notification json: {e}"); + continue; + } }; let author_did = val.get("author_did").and_then(|a| a.as_str()).unwrap_or(""); @@ -2817,10 +3302,14 @@ async fn list_notifications( let is_read = indexed_at <= last_seen.as_str(); let mut notif = val.clone(); - if let Ok(author_profile) = get_profile_internal(&app_state, author_did, None).await { - notif["author"] = author_profile; - } else { - continue; + match get_profile_internal(&app_state, author_did, None).await { + Ok(author_profile) => { + notif["author"] = author_profile; + } + Err(e) => { + tracing::warn!("failed to get author profile for notification from {author_did}: {e}"); + continue; + } } notif["isRead"] = serde_json::json!(is_read); @@ -2858,15 +3347,22 @@ async fn get_unread_count( let mut count = 0; for item in app_state.notifications.prefix(prefix.as_bytes()) { - let Ok((_key, val_bytes)) = item.into_inner() else { - continue; + let (_key, val_bytes) = match item.into_inner() { + Ok(inner) => inner, + Err(e) => { + tracing::warn!("failed to get notification item for count: {e}"); + continue; + } }; - if let Ok(val) = serde_json::from_slice::(&val_bytes) { - if let Some(indexed_at) = val.get("indexedAt").and_then(|a| a.as_str()) { - if indexed_at > last_seen.as_str() { - count += 1; + match serde_json::from_slice::(&val_bytes) { + Ok(val) => { + if let Some(indexed_at) = val.get("indexedAt").and_then(|a| a.as_str()) { + if indexed_at > last_seen.as_str() { + count += 1; + } } } + Err(e) => tracing::warn!("failed to parse notification json for count: {e}"), } } @@ -2892,15 +3388,19 @@ async fn update_seen( let body_bytes = axum::body::to_bytes(req.into_body(), usize::MAX) .await .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; - let Ok(payload) = serde_json::from_slice::(&body_bytes) else { - return Err(StatusCode::BAD_REQUEST); - }; + let payload: UpdateSeenReq = serde_json::from_slice(&body_bytes).map_err(|e| { + tracing::error!("failed to parse update_seen request: {e}"); + StatusCode::BAD_REQUEST + })?; let seen_key = format!("seen:{}", did); app_state .seen .insert(seen_key.as_bytes(), payload.seenAt.as_bytes()) - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?; + .map_err(|e| { + tracing::error!("failed to insert seen: {e}"); + StatusCode::INTERNAL_SERVER_ERROR + })?; Ok(Json(serde_json::json!({})).into_response()) } @@ -2917,45 +3417,68 @@ struct GetFeedParams { async fn get_feed(State(app_state): State, req: Request) -> Result { // let hydrant = &app_state.hydrant; let query_str = req.uri().query().unwrap_or(""); - let Ok(params) = serde_urlencoded::from_str::(query_str) else { - return Err(StatusCode::BAD_REQUEST); - }; - - let Ok(uri) = jacquard_common::types::string::AtUri::new(¶ms.feed) else { - return Err(StatusCode::BAD_REQUEST); - }; + let params: GetFeedParams = serde_urlencoded::from_str(query_str).map_err(|e| { + tracing::error!("failed to parse get_feed query params: {e}"); + StatusCode::BAD_REQUEST + })?; + + let uri = jacquard_common::types::string::AtUri::new(¶ms.feed).map_err(|e| { + tracing::error!("failed to parse feed uri {}: {e}", params.feed); + StatusCode::BAD_REQUEST + })?; let author_ident = uri.authority(); let rkey = uri .rkey() - .ok_or(StatusCode::BAD_REQUEST)? + .ok_or_else(|| { + tracing::error!("missing rkey in feed uri {}", params.feed); + StatusCode::BAD_REQUEST + })? .0 .as_str() .to_string(); - let Ok(repo) = app_state.hydrant.repos.resolve(author_ident).await else { - return proxy_request(req).await; + let repo = match app_state.hydrant.repos.resolve(author_ident).await { + Ok(r) => r, + Err(e) => { + tracing::error!("failed to resolve repo for {author_ident}: {e}"); + return proxy_request(req).await; + } }; - let Ok(Some(record)) = repo.get_record("app.bsky.feed.generator", &rkey).await else { - return proxy_request(req).await; + let record = match repo.get_record("app.bsky.feed.generator", &rkey).await { + Ok(Some(r)) => r, + Ok(None) => { + tracing::error!("feed generator record not found: {author_ident}/{rkey}"); + return proxy_request(req).await; + } + Err(e) => { + tracing::error!("failed to get feed generator record {author_ident}/{rkey}: {e}"); + return proxy_request(req).await; + } }; let val_json = serde_json::to_value(&record.value).unwrap_or(serde_json::json!({})); let Some(service_did) = val_json.get("did").and_then(|d| d.as_str()) else { + tracing::error!("missing did in feed generator record {author_ident}/{rkey}"); return Err(StatusCode::BAD_REQUEST); }; - let Ok(service_did_parsed) = jacquard_common::types::string::Did::new(service_did) else { - return Err(StatusCode::BAD_REQUEST); - }; + let service_did_parsed = jacquard_common::types::string::Did::new(service_did).map_err(|e| { + tracing::error!("failed to parse service did {service_did}: {e}"); + StatusCode::BAD_REQUEST + })?; - let Ok((doc_data, _)) = app_state + let (doc_data, _) = match app_state .hydrant .resolver() .resolve_raw_doc(&service_did_parsed) .await - else { - return proxy_request(req).await; + { + Ok(d) => d, + Err(e) => { + tracing::error!("failed to resolve service did {service_did}: {e}"); + return proxy_request(req).await; + } }; let doc_json = serde_json::to_value(&doc_data).unwrap_or(serde_json::json!({})); @@ -3013,29 +3536,42 @@ async fn get_feed(State(app_state): State, req: Request) -> Result r, + Err(e) => { + tracing::error!("failed to send request to feed generator {skeleton_url}: {e}"); + return proxy_request(req).await; + } }; - let Ok(skeleton) = res.json::().await else { - return proxy_request(req).await; + let skeleton = match res.json::().await { + Ok(s) => s, + Err(e) => { + tracing::error!("failed to parse feed skeleton json: {e}"); + return proxy_request(req).await; + } }; let mut hydrated_feed = Vec::new(); if let Some(feed_items) = skeleton.get("feed").and_then(|f| f.as_array()) { for item in feed_items { if let Some(post_uri) = item.get("post").and_then(|p| p.as_str()) { - if let Ok(post_view) = get_post_view(&app_state, post_uri, None).await { - let mut feed_item = serde_json::json!({ - "post": post_view - }); - if let Some(reason) = item.get("reason") { - feed_item - .as_object_mut() - .unwrap() - .insert("reason".to_string(), reason.clone()); + match get_post_view(&app_state, post_uri, None).await { + Ok(post_view) => { + let mut feed_item = serde_json::json!({ + "post": post_view + }); + if let Some(reason) = item.get("reason") { + feed_item + .as_object_mut() + .unwrap() + .insert("reason".to_string(), reason.clone()); + } + hydrated_feed.push(feed_item); + } + Err(e) => { + tracing::warn!("failed to get post view for {post_uri} in feed: {e}"); } - hydrated_feed.push(feed_item); } } } -- 2.51.2