+23

.sqlx/query-5f02d646eb60f99f5cc1ae7b8b41e62d053a6b9f8e9452d5cef3526b8aef8288.json

··· 1 1 + { 2 2 + "db_name": "PostgreSQL", 3 3 + "query": "SELECT 1 as val FROM records WHERE repo_id = $1 AND repo_rev <= $2 LIMIT 1", 4 4 + "describe": { 5 5 + "columns": [ 6 6 + { 7 7 + "ordinal": 0, 8 8 + "name": "val", 9 9 + "type_info": "Int4" 10 10 + } 11 11 + ], 12 12 + "parameters": { 13 13 + "Left": [ 14 14 + "Uuid", 15 15 + "Text" 16 16 + ] 17 17 + }, 18 18 + "nullable": [ 19 19 + null 20 20 + ] 21 21 + }, 22 22 + "hash": "5f02d646eb60f99f5cc1ae7b8b41e62d053a6b9f8e9452d5cef3526b8aef8288" 23 23 + }

+2 -2

.sqlx/query-7bb1388dec372fe749462cd9b604e5802b770aeb110462208988141d31c86c92.json .sqlx/query-36001fc127d7a3ea4e53e43a559cd86107e74d02ddcc499afd81049ce3c6789b.json

··· 1 1 { 2 2 "db_name": "PostgreSQL", 3 3 - "query": "SELECT k.key_bytes, k.encryption_version FROM user_keys k JOIN users u ON k.user_id = u.id WHERE u.did = $1", 3 3 + "query": "SELECT key_bytes, encryption_version FROM user_keys k JOIN users u ON k.user_id = u.id WHERE u.did = $1", 4 4 "describe": { 5 5 "columns": [ 6 6 { ··· 24 24 true 25 25 ] 26 26 }, 27 27 - "hash": "7bb1388dec372fe749462cd9b604e5802b770aeb110462208988141d31c86c92" 27 27 + "hash": "36001fc127d7a3ea4e53e43a559cd86107e74d02ddcc499afd81049ce3c6789b" 28 28 }

+18

.sqlx/query-8a9e71f04ec779d5c10d79582cc398529e01be01a83898df3524bb35e3d2ed14.json

··· 1 1 + { 2 2 + "db_name": "PostgreSQL", 3 3 + "query": "INSERT INTO records (repo_id, collection, rkey, record_cid, repo_rev) VALUES ($1, $2, $3, $4, $5)\n ON CONFLICT (repo_id, collection, rkey) DO UPDATE SET record_cid = $4, repo_rev = $5, created_at = NOW()", 4 4 + "describe": { 5 5 + "columns": [], 6 6 + "parameters": { 7 7 + "Left": [ 8 8 + "Uuid", 9 9 + "Text", 10 10 + "Text", 11 11 + "Text", 12 12 + "Text" 13 13 + ] 14 14 + }, 15 15 + "nullable": [] 16 16 + }, 17 17 + "hash": "8a9e71f04ec779d5c10d79582cc398529e01be01a83898df3524bb35e3d2ed14" 18 18 + }

+2 -2

.sqlx/query-bf60faafb5c79a149ba237a984f78d068b5d691f6762641412a5aa1517605c04.json .sqlx/query-a3e7b0c0861eaf62dda8b3a2ea5573bbb64eef74473f2b73cb38e2948cb3d7cc.json

··· 1 1 { 2 2 "db_name": "PostgreSQL", 3 3 - "query": "SELECT record_cid FROM records WHERE repo_id = $1 AND collection = 'app.bsky.graph.follow'", 3 3 + "query": "SELECT record_cid FROM records WHERE repo_id = $1 AND collection = 'app.bsky.graph.follow' LIMIT 5000", 4 4 "describe": { 5 5 "columns": [ 6 6 { ··· 18 18 false 19 19 ] 20 20 }, 21 21 - "hash": "bf60faafb5c79a149ba237a984f78d068b5d691f6762641412a5aa1517605c04" 21 21 + "hash": "a3e7b0c0861eaf62dda8b3a2ea5573bbb64eef74473f2b73cb38e2948cb3d7cc" 22 22 }

-17

.sqlx/query-c61fc3b2fbdf6891269908ef21f13dcabdc3b032e9f767becae34ca176df18b6.json

··· 1 1 - { 2 2 - "db_name": "PostgreSQL", 3 3 - "query": "INSERT INTO records (repo_id, collection, rkey, record_cid) VALUES ($1, $2, $3, $4)\n ON CONFLICT (repo_id, collection, rkey) DO UPDATE SET record_cid = $4, created_at = NOW()", 4 4 - "describe": { 5 5 - "columns": [], 6 6 - "parameters": { 7 7 - "Left": [ 8 8 - "Uuid", 9 9 - "Text", 10 10 - "Text", 11 11 - "Text" 12 12 - ] 13 13 - }, 14 14 - "nullable": [] 15 15 - }, 16 16 - "hash": "c61fc3b2fbdf6891269908ef21f13dcabdc3b032e9f767becae34ca176df18b6" 17 17 - }

+47

.sqlx/query-f3f1634b4f03a4c365afa02c4504de758bc420f49a19092d5cd1c526c7c7461e.json

··· 1 1 + { 2 2 + "db_name": "PostgreSQL", 3 3 + "query": "\n SELECT record_cid, collection, rkey, created_at, repo_rev\n FROM records\n WHERE repo_id = $1 AND repo_rev > $2\n ORDER BY repo_rev ASC\n LIMIT 10\n ", 4 4 + "describe": { 5 5 + "columns": [ 6 6 + { 7 7 + "ordinal": 0, 8 8 + "name": "record_cid", 9 9 + "type_info": "Text" 10 10 + }, 11 11 + { 12 12 + "ordinal": 1, 13 13 + "name": "collection", 14 14 + "type_info": "Text" 15 15 + }, 16 16 + { 17 17 + "ordinal": 2, 18 18 + "name": "rkey", 19 19 + "type_info": "Text" 20 20 + }, 21 21 + { 22 22 + "ordinal": 3, 23 23 + "name": "created_at", 24 24 + "type_info": "Timestamptz" 25 25 + }, 26 26 + { 27 27 + "ordinal": 4, 28 28 + "name": "repo_rev", 29 29 + "type_info": "Text" 30 30 + } 31 31 + ], 32 32 + "parameters": { 33 33 + "Left": [ 34 34 + "Uuid", 35 35 + "Text" 36 36 + ] 37 37 + }, 38 38 + "nullable": [ 39 39 + false, 40 40 + false, 41 41 + false, 42 42 + false, 43 43 + true 44 44 + ] 45 45 + }, 46 46 + "hash": "f3f1634b4f03a4c365afa02c4504de758bc420f49a19092d5cd1c526c7c7461e" 47 47 + }

+7 -7

TODO.md

··· 168 168 - [x] Implement `app.bsky.actor.getProfiles` (PDS-level with proxy fallback). 169 169 170 170 ### Feed (`app.bsky.feed`) 171 171 - These are implemented at PDS level to enable local-first reads: 172 172 - - [ ] Implement `app.bsky.feed.getTimeline` (PDS-level with proxy). 173 173 - - [ ] Implement `app.bsky.feed.getAuthorFeed` (PDS-level with proxy). 174 174 - - [ ] Implement `app.bsky.feed.getActorLikes` (PDS-level with proxy). 175 175 - - [ ] Implement `app.bsky.feed.getPostThread` (PDS-level with proxy). 176 176 - - [ ] Implement `app.bsky.feed.getFeed` (PDS-level with proxy). 171 171 + These are implemented at PDS level to enable local-first reads (read-after-write pattern): 172 172 + - [x] Implement `app.bsky.feed.getTimeline` (PDS-level with proxy + RAW). 173 173 + - [x] Implement `app.bsky.feed.getAuthorFeed` (PDS-level with proxy + RAW). 174 174 + - [x] Implement `app.bsky.feed.getActorLikes` (PDS-level with proxy + RAW). 175 175 + - [x] Implement `app.bsky.feed.getPostThread` (PDS-level with proxy + RAW + NotFound handling). 176 176 + - [x] Implement `app.bsky.feed.getFeed` (proxy to feed generator). 177 177 178 178 ### Notification (`app.bsky.notification`) 179 179 - - [ ] Implement `app.bsky.notification.registerPush` (push notification registration). 179 179 + - [x] Implement `app.bsky.notification.registerPush` (push notification registration, proxied). 180 180 181 181 ## Deprecated Sync Endpoints (for compatibility) 182 182 - [ ] Implement `com.atproto.sync.getCheckout` (deprecated, still needed for compatibility).

+2

migrations/202512211600_add_repo_rev.sql

··· 1 1 + ALTER TABLE records ADD COLUMN repo_rev TEXT; 2 2 + CREATE INDEX idx_records_repo_rev ON records(repo_rev);

+24 -8

src/api/actor/profile.rs

··· 125 125 ) -> Response { 126 126 let auth_header = headers.get("Authorization").and_then(|h| h.to_str().ok()); 127 127 128 128 - let auth_did = auth_header.and_then(|h| { 129 129 - let token = crate::auth::extract_bearer_token_from_header(Some(h))?; 130 130 - crate::auth::get_did_from_token(&token).ok() 131 131 - }); 128 128 + let auth_did = if let Some(h) = auth_header { 129 129 + if let Some(token) = crate::auth::extract_bearer_token_from_header(Some(h)) { 130 130 + match crate::auth::validate_bearer_token(&state.db, &token).await { 131 131 + Ok(user) => Some(user.did), 132 132 + Err(_) => None, 133 133 + } 134 134 + } else { 135 135 + None 136 136 + } 137 137 + } else { 138 138 + None 139 139 + }; 132 140 133 141 let mut query_params = HashMap::new(); 134 142 query_params.insert("actor".to_string(), params.actor.clone()); ··· 167 175 ) -> Response { 168 176 let auth_header = headers.get("Authorization").and_then(|h| h.to_str().ok()); 169 177 170 170 - let auth_did = auth_header.and_then(|h| { 171 171 - let token = crate::auth::extract_bearer_token_from_header(Some(h))?; 172 172 - crate::auth::get_did_from_token(&token).ok() 173 173 - }); 178 178 + let auth_did = if let Some(h) = auth_header { 179 179 + if let Some(token) = crate::auth::extract_bearer_token_from_header(Some(h)) { 180 180 + match crate::auth::validate_bearer_token(&state.db, &token).await { 181 181 + Ok(user) => Some(user.did), 182 182 + Err(_) => None, 183 183 + } 184 184 + } else { 185 185 + None 186 186 + } 187 187 + } else { 188 188 + None 189 189 + }; 174 190 175 191 let mut query_params = HashMap::new(); 176 192 query_params.insert("actors".to_string(), params.actors.clone());

+33 -4

src/api/error.rs

··· 44 44 InvitesDisabled, 45 45 DatabaseError, 46 46 UpstreamFailure, 47 47 + UpstreamTimeout, 48 48 + UpstreamUnavailable(String), 49 49 + UpstreamError { status: u16, error: Option<String>, message: Option<String> }, 47 50 } 48 51 49 52 impl ApiError { 50 53 fn status_code(&self) -> StatusCode { 51 54 match self { 52 52 - Self::InternalError | Self::DatabaseError | Self::UpstreamFailure => { 53 53 - StatusCode::INTERNAL_SERVER_ERROR 55 55 + Self::InternalError | Self::DatabaseError => StatusCode::INTERNAL_SERVER_ERROR, 56 56 + Self::UpstreamFailure | Self::UpstreamUnavailable(_) => StatusCode::BAD_GATEWAY, 57 57 + Self::UpstreamTimeout => StatusCode::GATEWAY_TIMEOUT, 58 58 + Self::UpstreamError { status, .. } => { 59 59 + StatusCode::from_u16(*status).unwrap_or(StatusCode::BAD_GATEWAY) 54 60 } 55 61 Self::AuthenticationRequired 56 62 | Self::AuthenticationFailed ··· 83 89 84 90 fn error_name(&self) -> &'static str { 85 91 match self { 86 86 - Self::InternalError | Self::DatabaseError | Self::UpstreamFailure => "InternalError", 92 92 + Self::InternalError | Self::DatabaseError => "InternalError", 93 93 + Self::UpstreamFailure | Self::UpstreamUnavailable(_) => "UpstreamFailure", 94 94 + Self::UpstreamTimeout => "UpstreamTimeout", 95 95 + Self::UpstreamError { error, .. } => { 96 96 + if let Some(e) = error { 97 97 + return Box::leak(e.clone().into_boxed_str()); 98 98 + } 99 99 + "UpstreamError" 100 100 + } 87 101 Self::AuthenticationRequired => "AuthenticationRequired", 88 102 Self::AuthenticationFailed | Self::AuthenticationFailedMsg(_) => "AuthenticationFailed", 89 103 Self::InvalidToken => "InvalidToken", ··· 116 130 Self::AuthenticationFailedMsg(msg) 117 131 | Self::ExpiredTokenMsg(msg) 118 132 | Self::InvalidRequest(msg) 119 119 - | Self::RepoNotFoundMsg(msg) => Some(msg.clone()), 133 133 + | Self::RepoNotFoundMsg(msg) 134 134 + | Self::UpstreamUnavailable(msg) => Some(msg.clone()), 135 135 + Self::UpstreamError { message, .. } => message.clone(), 136 136 + Self::UpstreamTimeout => Some("Upstream service timed out".to_string()), 120 137 _ => None, 121 138 } 139 139 + } 140 140 + 141 141 + pub fn from_upstream_response( 142 142 + status: u16, 143 143 + body: &[u8], 144 144 + ) -> Self { 145 145 + if let Ok(parsed) = serde_json::from_slice::<serde_json::Value>(body) { 146 146 + let error = parsed.get("error").and_then(|v| v.as_str()).map(String::from); 147 147 + let message = parsed.get("message").and_then(|v| v.as_str()).map(String::from); 148 148 + return Self::UpstreamError { status, error, message }; 149 149 + } 150 150 + Self::UpstreamError { status, error: None, message: None } 122 151 } 123 152 } 124 153

+158

src/api/feed/actor_likes.rs

··· 1 1 + use crate::api::read_after_write::{ 2 2 + extract_repo_rev, format_munged_response, get_local_lag, get_records_since_rev, 3 3 + proxy_to_appview, FeedOutput, FeedViewPost, LikeRecord, PostView, RecordDescript, 4 4 + }; 5 5 + use crate::state::AppState; 6 6 + use axum::{ 7 7 + extract::{Query, State}, 8 8 + http::StatusCode, 9 9 + response::{IntoResponse, Response}, 10 10 + Json, 11 11 + }; 12 12 + use serde::Deserialize; 13 13 + use serde_json::Value; 14 14 + use std::collections::HashMap; 15 15 + use tracing::warn; 16 16 + 17 17 + #[derive(Deserialize)] 18 18 + pub struct GetActorLikesParams { 19 19 + pub actor: String, 20 20 + pub limit: Option<u32>, 21 21 + pub cursor: Option<String>, 22 22 + } 23 23 + 24 24 + fn insert_likes_into_feed(feed: &mut Vec<FeedViewPost>, likes: &[RecordDescript<LikeRecord>]) { 25 25 + for like in likes { 26 26 + let like_time = &like.indexed_at.to_rfc3339(); 27 27 + let idx = feed 28 28 + .iter() 29 29 + .position(|fi| &fi.post.indexed_at < like_time) 30 30 + .unwrap_or(feed.len()); 31 31 + 32 32 + let placeholder_post = PostView { 33 33 + uri: like.record.subject.uri.clone(), 34 34 + cid: like.record.subject.cid.clone(), 35 35 + author: crate::api::read_after_write::AuthorView { 36 36 + did: String::new(), 37 37 + handle: String::new(), 38 38 + display_name: None, 39 39 + avatar: None, 40 40 + extra: HashMap::new(), 41 41 + }, 42 42 + record: Value::Null, 43 43 + indexed_at: like.indexed_at.to_rfc3339(), 44 44 + embed: None, 45 45 + reply_count: 0, 46 46 + repost_count: 0, 47 47 + like_count: 0, 48 48 + quote_count: 0, 49 49 + extra: HashMap::new(), 50 50 + }; 51 51 + 52 52 + feed.insert( 53 53 + idx, 54 54 + FeedViewPost { 55 55 + post: placeholder_post, 56 56 + reply: None, 57 57 + reason: None, 58 58 + feed_context: None, 59 59 + extra: HashMap::new(), 60 60 + }, 61 61 + ); 62 62 + } 63 63 + } 64 64 + 65 65 + pub async fn get_actor_likes( 66 66 + State(state): State<AppState>, 67 67 + headers: axum::http::HeaderMap, 68 68 + Query(params): Query<GetActorLikesParams>, 69 69 + ) -> Response { 70 70 + let auth_header = headers.get("Authorization").and_then(|h| h.to_str().ok()); 71 71 + 72 72 + let auth_did = if let Some(h) = auth_header { 73 73 + if let Some(token) = crate::auth::extract_bearer_token_from_header(Some(h)) { 74 74 + match crate::auth::validate_bearer_token(&state.db, &token).await { 75 75 + Ok(user) => Some(user.did), 76 76 + Err(_) => None, 77 77 + } 78 78 + } else { 79 79 + None 80 80 + } 81 81 + } else { 82 82 + None 83 83 + }; 84 84 + 85 85 + let mut query_params = HashMap::new(); 86 86 + query_params.insert("actor".to_string(), params.actor.clone()); 87 87 + if let Some(limit) = params.limit { 88 88 + query_params.insert("limit".to_string(), limit.to_string()); 89 89 + } 90 90 + if let Some(cursor) = &params.cursor { 91 91 + query_params.insert("cursor".to_string(), cursor.clone()); 92 92 + } 93 93 + 94 94 + let proxy_result = 95 95 + match proxy_to_appview("app.bsky.feed.getActorLikes", &query_params, auth_header).await { 96 96 + Ok(r) => r, 97 97 + Err(e) => return e, 98 98 + }; 99 99 + 100 100 + if !proxy_result.status.is_success() { 101 101 + return (proxy_result.status, proxy_result.body).into_response(); 102 102 + } 103 103 + 104 104 + let rev = match extract_repo_rev(&proxy_result.headers) { 105 105 + Some(r) => r, 106 106 + None => return (proxy_result.status, proxy_result.body).into_response(), 107 107 + }; 108 108 + 109 109 + let mut feed_output: FeedOutput = match serde_json::from_slice(&proxy_result.body) { 110 110 + Ok(f) => f, 111 111 + Err(e) => { 112 112 + warn!("Failed to parse actor likes response: {:?}", e); 113 113 + return (proxy_result.status, proxy_result.body).into_response(); 114 114 + } 115 115 + }; 116 116 + 117 117 + let requester_did = match auth_did { 118 118 + Some(d) => d, 119 119 + None => return (StatusCode::OK, Json(feed_output)).into_response(), 120 120 + }; 121 121 + 122 122 + let actor_did = if params.actor.starts_with("did:") { 123 123 + params.actor.clone() 124 124 + } else { 125 125 + match sqlx::query_scalar!("SELECT did FROM users WHERE handle = $1", params.actor) 126 126 + .fetch_optional(&state.db) 127 127 + .await 128 128 + { 129 129 + Ok(Some(did)) => did, 130 130 + Ok(None) => return (StatusCode::OK, Json(feed_output)).into_response(), 131 131 + Err(e) => { 132 132 + warn!("Database error resolving actor handle: {:?}", e); 133 133 + return (proxy_result.status, proxy_result.body).into_response(); 134 134 + } 135 135 + } 136 136 + }; 137 137 + 138 138 + if actor_did != requester_did { 139 139 + return (StatusCode::OK, Json(feed_output)).into_response(); 140 140 + } 141 141 + 142 142 + let local_records = match get_records_since_rev(&state, &requester_did, &rev).await { 143 143 + Ok(r) => r, 144 144 + Err(e) => { 145 145 + warn!("Failed to get local records: {}", e); 146 146 + return (proxy_result.status, proxy_result.body).into_response(); 147 147 + } 148 148 + }; 149 149 + 150 150 + if local_records.likes.is_empty() { 151 151 + return (StatusCode::OK, Json(feed_output)).into_response(); 152 152 + } 153 153 + 154 154 + insert_likes_into_feed(&mut feed_output.feed, &local_records.likes); 155 155 + 156 156 + let lag = get_local_lag(&local_records); 157 157 + format_munged_response(feed_output, lag) 158 158 + }

+169

src/api/feed/author_feed.rs

··· 1 1 + use crate::api::read_after_write::{ 2 2 + extract_repo_rev, format_local_post, format_munged_response, get_local_lag, 3 3 + get_records_since_rev, insert_posts_into_feed, proxy_to_appview, FeedOutput, FeedViewPost, 4 4 + ProfileRecord, RecordDescript, 5 5 + }; 6 6 + use crate::state::AppState; 7 7 + use axum::{ 8 8 + extract::{Query, State}, 9 9 + http::StatusCode, 10 10 + response::{IntoResponse, Response}, 11 11 + Json, 12 12 + }; 13 13 + use serde::Deserialize; 14 14 + use std::collections::HashMap; 15 15 + use tracing::warn; 16 16 + 17 17 + #[derive(Deserialize)] 18 18 + pub struct GetAuthorFeedParams { 19 19 + pub actor: String, 20 20 + pub limit: Option<u32>, 21 21 + pub cursor: Option<String>, 22 22 + pub filter: Option<String>, 23 23 + #[serde(rename = "includePins")] 24 24 + pub include_pins: Option<bool>, 25 25 + } 26 26 + 27 27 + fn update_author_profile_in_feed( 28 28 + feed: &mut [FeedViewPost], 29 29 + author_did: &str, 30 30 + local_profile: &RecordDescript<ProfileRecord>, 31 31 + ) { 32 32 + for item in feed.iter_mut() { 33 33 + if item.post.author.did == author_did { 34 34 + if let Some(ref display_name) = local_profile.record.display_name { 35 35 + item.post.author.display_name = Some(display_name.clone()); 36 36 + } 37 37 + } 38 38 + } 39 39 + } 40 40 + 41 41 + pub async fn get_author_feed( 42 42 + State(state): State<AppState>, 43 43 + headers: axum::http::HeaderMap, 44 44 + Query(params): Query<GetAuthorFeedParams>, 45 45 + ) -> Response { 46 46 + let auth_header = headers.get("Authorization").and_then(|h| h.to_str().ok()); 47 47 + 48 48 + let auth_did = if let Some(h) = auth_header { 49 49 + if let Some(token) = crate::auth::extract_bearer_token_from_header(Some(h)) { 50 50 + match crate::auth::validate_bearer_token(&state.db, &token).await { 51 51 + Ok(user) => Some(user.did), 52 52 + Err(_) => None, 53 53 + } 54 54 + } else { 55 55 + None 56 56 + } 57 57 + } else { 58 58 + None 59 59 + }; 60 60 + 61 61 + let mut query_params = HashMap::new(); 62 62 + query_params.insert("actor".to_string(), params.actor.clone()); 63 63 + if let Some(limit) = params.limit { 64 64 + query_params.insert("limit".to_string(), limit.to_string()); 65 65 + } 66 66 + if let Some(cursor) = &params.cursor { 67 67 + query_params.insert("cursor".to_string(), cursor.clone()); 68 68 + } 69 69 + if let Some(filter) = &params.filter { 70 70 + query_params.insert("filter".to_string(), filter.clone()); 71 71 + } 72 72 + if let Some(include_pins) = params.include_pins { 73 73 + query_params.insert("includePins".to_string(), include_pins.to_string()); 74 74 + } 75 75 + 76 76 + let proxy_result = 77 77 + match proxy_to_appview("app.bsky.feed.getAuthorFeed", &query_params, auth_header).await { 78 78 + Ok(r) => r, 79 79 + Err(e) => return e, 80 80 + }; 81 81 + 82 82 + if !proxy_result.status.is_success() { 83 83 + return (proxy_result.status, proxy_result.body).into_response(); 84 84 + } 85 85 + 86 86 + let rev = match extract_repo_rev(&proxy_result.headers) { 87 87 + Some(r) => r, 88 88 + None => return (proxy_result.status, proxy_result.body).into_response(), 89 89 + }; 90 90 + 91 91 + let mut feed_output: FeedOutput = match serde_json::from_slice(&proxy_result.body) { 92 92 + Ok(f) => f, 93 93 + Err(e) => { 94 94 + warn!("Failed to parse author feed response: {:?}", e); 95 95 + return (proxy_result.status, proxy_result.body).into_response(); 96 96 + } 97 97 + }; 98 98 + 99 99 + let requester_did = match auth_did { 100 100 + Some(d) => d, 101 101 + None => return (StatusCode::OK, Json(feed_output)).into_response(), 102 102 + }; 103 103 + 104 104 + let actor_did = if params.actor.starts_with("did:") { 105 105 + params.actor.clone() 106 106 + } else { 107 107 + match sqlx::query_scalar!("SELECT did FROM users WHERE handle = $1", params.actor) 108 108 + .fetch_optional(&state.db) 109 109 + .await 110 110 + { 111 111 + Ok(Some(did)) => did, 112 112 + Ok(None) => return (StatusCode::OK, Json(feed_output)).into_response(), 113 113 + Err(e) => { 114 114 + warn!("Database error resolving actor handle: {:?}", e); 115 115 + return (proxy_result.status, proxy_result.body).into_response(); 116 116 + } 117 117 + } 118 118 + }; 119 119 + 120 120 + if actor_did != requester_did { 121 121 + return (StatusCode::OK, Json(feed_output)).into_response(); 122 122 + } 123 123 + 124 124 + let local_records = match get_records_since_rev(&state, &requester_did, &rev).await { 125 125 + Ok(r) => r, 126 126 + Err(e) => { 127 127 + warn!("Failed to get local records: {}", e); 128 128 + return (proxy_result.status, proxy_result.body).into_response(); 129 129 + } 130 130 + }; 131 131 + 132 132 + if local_records.count == 0 { 133 133 + return (StatusCode::OK, Json(feed_output)).into_response(); 134 134 + } 135 135 + 136 136 + let handle = match sqlx::query_scalar!("SELECT handle FROM users WHERE did = $1", requester_did) 137 137 + .fetch_optional(&state.db) 138 138 + .await 139 139 + { 140 140 + Ok(Some(h)) => h, 141 141 + Ok(None) => requester_did.clone(), 142 142 + Err(e) => { 143 143 + warn!("Database error fetching handle: {:?}", e); 144 144 + requester_did.clone() 145 145 + } 146 146 + }; 147 147 + 148 148 + if let Some(ref local_profile) = local_records.profile { 149 149 + update_author_profile_in_feed(&mut feed_output.feed, &requester_did, local_profile); 150 150 + } 151 151 + 152 152 + let local_posts: Vec<_> = local_records 153 153 + .posts 154 154 + .iter() 155 155 + .map(|p| { 156 156 + format_local_post( 157 157 + p, 158 158 + &requester_did, 159 159 + &handle, 160 160 + local_records.profile.as_ref(), 161 161 + ) 162 162 + }) 163 163 + .collect(); 164 164 + 165 165 + insert_posts_into_feed(&mut feed_output.feed, local_posts); 166 166 + 167 167 + let lag = get_local_lag(&local_records); 168 168 + format_munged_response(feed_output, lag) 169 169 + }

+131

src/api/feed/custom_feed.rs

··· 1 1 + use crate::api::proxy_client::{ 2 2 + is_ssrf_safe, proxy_client, validate_at_uri, validate_limit, MAX_RESPONSE_SIZE, 3 3 + }; 4 4 + use crate::api::ApiError; 5 5 + use crate::state::AppState; 6 6 + use axum::{ 7 7 + extract::{Query, State}, 8 8 + http::StatusCode, 9 9 + response::{IntoResponse, Response}, 10 10 + }; 11 11 + use serde::Deserialize; 12 12 + use std::collections::HashMap; 13 13 + use tracing::{error, info}; 14 14 + 15 15 + #[derive(Deserialize)] 16 16 + pub struct GetFeedParams { 17 17 + pub feed: String, 18 18 + pub limit: Option<u32>, 19 19 + pub cursor: Option<String>, 20 20 + } 21 21 + 22 22 + pub async fn get_feed( 23 23 + State(state): State<AppState>, 24 24 + headers: axum::http::HeaderMap, 25 25 + Query(params): Query<GetFeedParams>, 26 26 + ) -> Response { 27 27 + let token = match crate::auth::extract_bearer_token_from_header( 28 28 + headers.get("Authorization").and_then(|h| h.to_str().ok()), 29 29 + ) { 30 30 + Some(t) => t, 31 31 + None => return ApiError::AuthenticationRequired.into_response(), 32 32 + }; 33 33 + 34 34 + if let Err(e) = crate::auth::validate_bearer_token(&state.db, &token).await { 35 35 + return ApiError::from(e).into_response(); 36 36 + }; 37 37 + 38 38 + if let Err(e) = validate_at_uri(&params.feed) { 39 39 + return ApiError::InvalidRequest(format!("Invalid feed URI: {}", e)).into_response(); 40 40 + } 41 41 + 42 42 + let auth_header = headers.get("Authorization").and_then(|h| h.to_str().ok()); 43 43 + 44 44 + let appview_url = match std::env::var("APPVIEW_URL") { 45 45 + Ok(url) => url, 46 46 + Err(_) => { 47 47 + return ApiError::UpstreamUnavailable("No upstream AppView configured".to_string()) 48 48 + .into_response(); 49 49 + } 50 50 + }; 51 51 + 52 52 + if let Err(e) = is_ssrf_safe(&appview_url) { 53 53 + error!("SSRF check failed for appview URL: {}", e); 54 54 + return ApiError::UpstreamUnavailable(format!("Invalid upstream URL: {}", e)) 55 55 + .into_response(); 56 56 + } 57 57 + 58 58 + let limit = validate_limit(params.limit, 50, 100); 59 59 + let mut query_params = HashMap::new(); 60 60 + query_params.insert("feed".to_string(), params.feed.clone()); 61 61 + query_params.insert("limit".to_string(), limit.to_string()); 62 62 + if let Some(cursor) = &params.cursor { 63 63 + query_params.insert("cursor".to_string(), cursor.clone()); 64 64 + } 65 65 + 66 66 + let target_url = format!("{}/xrpc/app.bsky.feed.getFeed", appview_url); 67 67 + info!(target = %target_url, feed = %params.feed, "Proxying getFeed request"); 68 68 + 69 69 + let client = proxy_client(); 70 70 + let mut request_builder = client.get(&target_url).query(&query_params); 71 71 + 72 72 + if let Some(auth) = auth_header { 73 73 + request_builder = request_builder.header("Authorization", auth); 74 74 + } 75 75 + 76 76 + match request_builder.send().await { 77 77 + Ok(resp) => { 78 78 + let status = 79 79 + StatusCode::from_u16(resp.status().as_u16()).unwrap_or(StatusCode::BAD_GATEWAY); 80 80 + 81 81 + let content_length = resp.content_length().unwrap_or(0); 82 82 + if content_length > MAX_RESPONSE_SIZE { 83 83 + error!( 84 84 + content_length, 85 85 + max = MAX_RESPONSE_SIZE, 86 86 + "getFeed response too large" 87 87 + ); 88 88 + return ApiError::UpstreamFailure.into_response(); 89 89 + } 90 90 + 91 91 + let resp_headers = resp.headers().clone(); 92 92 + let body = match resp.bytes().await { 93 93 + Ok(b) => { 94 94 + if b.len() as u64 > MAX_RESPONSE_SIZE { 95 95 + error!(len = b.len(), "getFeed response body exceeded limit"); 96 96 + return ApiError::UpstreamFailure.into_response(); 97 97 + } 98 98 + b 99 99 + } 100 100 + Err(e) => { 101 101 + error!(error = ?e, "Error reading getFeed response"); 102 102 + return ApiError::UpstreamFailure.into_response(); 103 103 + } 104 104 + }; 105 105 + 106 106 + let mut response_builder = axum::response::Response::builder().status(status); 107 107 + if let Some(ct) = resp_headers.get("content-type") { 108 108 + response_builder = response_builder.header("content-type", ct); 109 109 + } 110 110 + 111 111 + match response_builder.body(axum::body::Body::from(body)) { 112 112 + Ok(r) => r, 113 113 + Err(e) => { 114 114 + error!(error = ?e, "Error building getFeed response"); 115 115 + ApiError::UpstreamFailure.into_response() 116 116 + } 117 117 + } 118 118 + } 119 119 + Err(e) => { 120 120 + error!(error = ?e, "Error proxying getFeed"); 121 121 + if e.is_timeout() { 122 122 + ApiError::UpstreamTimeout.into_response() 123 123 + } else if e.is_connect() { 124 124 + ApiError::UpstreamUnavailable("Failed to connect to upstream".to_string()) 125 125 + .into_response() 126 126 + } else { 127 127 + ApiError::UpstreamFailure.into_response() 128 128 + } 129 129 + } 130 130 + } 131 131 + }

+8

src/api/feed/mod.rs

··· 1 1 + mod actor_likes; 2 2 + mod author_feed; 3 3 + mod custom_feed; 4 4 + mod post_thread; 1 5 mod timeline; 2 6 7 7 + pub use actor_likes::get_actor_likes; 8 8 + pub use author_feed::get_author_feed; 9 9 + pub use custom_feed::get_feed; 10 10 + pub use post_thread::get_post_thread; 3 11 pub use timeline::get_timeline;

+322

src/api/feed/post_thread.rs

··· 1 1 + use crate::api::read_after_write::{ 2 2 + extract_repo_rev, format_local_post, format_munged_response, get_local_lag, 3 3 + get_records_since_rev, proxy_to_appview, PostRecord, PostView, RecordDescript, 4 4 + }; 5 5 + use crate::state::AppState; 6 6 + use axum::{ 7 7 + extract::{Query, State}, 8 8 + http::StatusCode, 9 9 + response::{IntoResponse, Response}, 10 10 + Json, 11 11 + }; 12 12 + use serde::{Deserialize, Serialize}; 13 13 + use serde_json::{json, Value}; 14 14 + use std::collections::HashMap; 15 15 + use tracing::warn; 16 16 + 17 17 + #[derive(Deserialize)] 18 18 + pub struct GetPostThreadParams { 19 19 + pub uri: String, 20 20 + pub depth: Option<u32>, 21 21 + #[serde(rename = "parentHeight")] 22 22 + pub parent_height: Option<u32>, 23 23 + } 24 24 + 25 25 + #[derive(Debug, Clone, Serialize, Deserialize)] 26 26 + #[serde(rename_all = "camelCase")] 27 27 + pub struct ThreadViewPost { 28 28 + #[serde(rename = "$type")] 29 29 + pub thread_type: Option<String>, 30 30 + pub post: PostView, 31 31 + #[serde(skip_serializing_if = "Option::is_none")] 32 32 + pub parent: Option<Box<ThreadNode>>, 33 33 + #[serde(skip_serializing_if = "Option::is_none")] 34 34 + pub replies: Option<Vec<ThreadNode>>, 35 35 + #[serde(flatten)] 36 36 + pub extra: HashMap<String, Value>, 37 37 + } 38 38 + 39 39 + #[derive(Debug, Clone, Serialize, Deserialize)] 40 40 + #[serde(untagged)] 41 41 + pub enum ThreadNode { 42 42 + Post(ThreadViewPost), 43 43 + NotFound(ThreadNotFound), 44 44 + Blocked(ThreadBlocked), 45 45 + } 46 46 + 47 47 + #[derive(Debug, Clone, Serialize, Deserialize)] 48 48 + #[serde(rename_all = "camelCase")] 49 49 + pub struct ThreadNotFound { 50 50 + #[serde(rename = "$type")] 51 51 + pub thread_type: String, 52 52 + pub uri: String, 53 53 + pub not_found: bool, 54 54 + } 55 55 + 56 56 + #[derive(Debug, Clone, Serialize, Deserialize)] 57 57 + #[serde(rename_all = "camelCase")] 58 58 + pub struct ThreadBlocked { 59 59 + #[serde(rename = "$type")] 60 60 + pub thread_type: String, 61 61 + pub uri: String, 62 62 + pub blocked: bool, 63 63 + pub author: Value, 64 64 + } 65 65 + 66 66 + #[derive(Debug, Clone, Serialize, Deserialize)] 67 67 + pub struct PostThreadOutput { 68 68 + pub thread: ThreadNode, 69 69 + #[serde(skip_serializing_if = "Option::is_none")] 70 70 + pub threadgate: Option<Value>, 71 71 + } 72 72 + 73 73 + const MAX_THREAD_DEPTH: usize = 10; 74 74 + 75 75 + fn add_replies_to_thread( 76 76 + thread: &mut ThreadViewPost, 77 77 + local_posts: &[RecordDescript<PostRecord>], 78 78 + author_did: &str, 79 79 + author_handle: &str, 80 80 + depth: usize, 81 81 + ) { 82 82 + if depth >= MAX_THREAD_DEPTH { 83 83 + return; 84 84 + } 85 85 + 86 86 + let thread_uri = &thread.post.uri; 87 87 + 88 88 + let replies: Vec<_> = local_posts 89 89 + .iter() 90 90 + .filter(|p| { 91 91 + p.record 92 92 + .reply 93 93 + .as_ref() 94 94 + .and_then(|r| r.get("parent")) 95 95 + .and_then(|parent| parent.get("uri")) 96 96 + .and_then(|u| u.as_str()) 97 97 + == Some(thread_uri) 98 98 + }) 99 99 + .map(|p| { 100 100 + let post_view = format_local_post(p, author_did, author_handle, None); 101 101 + ThreadNode::Post(ThreadViewPost { 102 102 + thread_type: Some("app.bsky.feed.defs#threadViewPost".to_string()), 103 103 + post: post_view, 104 104 + parent: None, 105 105 + replies: None, 106 106 + extra: HashMap::new(), 107 107 + }) 108 108 + }) 109 109 + .collect(); 110 110 + 111 111 + if !replies.is_empty() { 112 112 + match &mut thread.replies { 113 113 + Some(existing) => existing.extend(replies), 114 114 + None => thread.replies = Some(replies), 115 115 + } 116 116 + } 117 117 + 118 118 + if let Some(ref mut existing_replies) = thread.replies { 119 119 + for reply in existing_replies.iter_mut() { 120 120 + if let ThreadNode::Post(reply_thread) = reply { 121 121 + add_replies_to_thread(reply_thread, local_posts, author_did, author_handle, depth + 1); 122 122 + } 123 123 + } 124 124 + } 125 125 + } 126 126 + 127 127 + pub async fn get_post_thread( 128 128 + State(state): State<AppState>, 129 129 + headers: axum::http::HeaderMap, 130 130 + Query(params): Query<GetPostThreadParams>, 131 131 + ) -> Response { 132 132 + let auth_header = headers.get("Authorization").and_then(|h| h.to_str().ok()); 133 133 + 134 134 + let auth_did = if let Some(h) = auth_header { 135 135 + if let Some(token) = crate::auth::extract_bearer_token_from_header(Some(h)) { 136 136 + match crate::auth::validate_bearer_token(&state.db, &token).await { 137 137 + Ok(user) => Some(user.did), 138 138 + Err(_) => None, 139 139 + } 140 140 + } else { 141 141 + None 142 142 + } 143 143 + } else { 144 144 + None 145 145 + }; 146 146 + 147 147 + let mut query_params = HashMap::new(); 148 148 + query_params.insert("uri".to_string(), params.uri.clone()); 149 149 + if let Some(depth) = params.depth { 150 150 + query_params.insert("depth".to_string(), depth.to_string()); 151 151 + } 152 152 + if let Some(parent_height) = params.parent_height { 153 153 + query_params.insert("parentHeight".to_string(), parent_height.to_string()); 154 154 + } 155 155 + 156 156 + let proxy_result = 157 157 + match proxy_to_appview("app.bsky.feed.getPostThread", &query_params, auth_header).await { 158 158 + Ok(r) => r, 159 159 + Err(e) => return e, 160 160 + }; 161 161 + 162 162 + if proxy_result.status == StatusCode::NOT_FOUND { 163 163 + return handle_not_found(&state, &params.uri, auth_did, &proxy_result.headers).await; 164 164 + } 165 165 + 166 166 + if !proxy_result.status.is_success() { 167 167 + return (proxy_result.status, proxy_result.body).into_response(); 168 168 + } 169 169 + 170 170 + let rev = match extract_repo_rev(&proxy_result.headers) { 171 171 + Some(r) => r, 172 172 + None => return (proxy_result.status, proxy_result.body).into_response(), 173 173 + }; 174 174 + 175 175 + let mut thread_output: PostThreadOutput = match serde_json::from_slice(&proxy_result.body) { 176 176 + Ok(t) => t, 177 177 + Err(e) => { 178 178 + warn!("Failed to parse post thread response: {:?}", e); 179 179 + return (proxy_result.status, proxy_result.body).into_response(); 180 180 + } 181 181 + }; 182 182 + 183 183 + let requester_did = match auth_did { 184 184 + Some(d) => d, 185 185 + None => return (StatusCode::OK, Json(thread_output)).into_response(), 186 186 + }; 187 187 + 188 188 + let local_records = match get_records_since_rev(&state, &requester_did, &rev).await { 189 189 + Ok(r) => r, 190 190 + Err(e) => { 191 191 + warn!("Failed to get local records: {}", e); 192 192 + return (proxy_result.status, proxy_result.body).into_response(); 193 193 + } 194 194 + }; 195 195 + 196 196 + if local_records.posts.is_empty() { 197 197 + return (StatusCode::OK, Json(thread_output)).into_response(); 198 198 + } 199 199 + 200 200 + let handle = match sqlx::query_scalar!("SELECT handle FROM users WHERE did = $1", requester_did) 201 201 + .fetch_optional(&state.db) 202 202 + .await 203 203 + { 204 204 + Ok(Some(h)) => h, 205 205 + Ok(None) => requester_did.clone(), 206 206 + Err(e) => { 207 207 + warn!("Database error fetching handle: {:?}", e); 208 208 + requester_did.clone() 209 209 + } 210 210 + }; 211 211 + 212 212 + if let ThreadNode::Post(ref mut thread_post) = thread_output.thread { 213 213 + add_replies_to_thread(thread_post, &local_records.posts, &requester_did, &handle, 0); 214 214 + } 215 215 + 216 216 + let lag = get_local_lag(&local_records); 217 217 + format_munged_response(thread_output, lag) 218 218 + } 219 219 + 220 220 + async fn handle_not_found( 221 221 + state: &AppState, 222 222 + uri: &str, 223 223 + auth_did: Option<String>, 224 224 + headers: &axum::http::HeaderMap, 225 225 + ) -> Response { 226 226 + let rev = match extract_repo_rev(headers) { 227 227 + Some(r) => r, 228 228 + None => { 229 229 + return ( 230 230 + StatusCode::NOT_FOUND, 231 231 + Json(json!({"error": "NotFound", "message": "Post not found"})), 232 232 + ) 233 233 + .into_response() 234 234 + } 235 235 + }; 236 236 + 237 237 + let requester_did = match auth_did { 238 238 + Some(d) => d, 239 239 + None => { 240 240 + return ( 241 241 + StatusCode::NOT_FOUND, 242 242 + Json(json!({"error": "NotFound", "message": "Post not found"})), 243 243 + ) 244 244 + .into_response() 245 245 + } 246 246 + }; 247 247 + 248 248 + let uri_parts: Vec<&str> = uri.trim_start_matches("at://").split('/').collect(); 249 249 + if uri_parts.len() != 3 { 250 250 + return ( 251 251 + StatusCode::NOT_FOUND, 252 252 + Json(json!({"error": "NotFound", "message": "Post not found"})), 253 253 + ) 254 254 + .into_response(); 255 255 + } 256 256 + 257 257 + let post_did = uri_parts[0]; 258 258 + if post_did != requester_did { 259 259 + return ( 260 260 + StatusCode::NOT_FOUND, 261 261 + Json(json!({"error": "NotFound", "message": "Post not found"})), 262 262 + ) 263 263 + .into_response(); 264 264 + } 265 265 + 266 266 + let local_records = match get_records_since_rev(state, &requester_did, &rev).await { 267 267 + Ok(r) => r, 268 268 + Err(_) => { 269 269 + return ( 270 270 + StatusCode::NOT_FOUND, 271 271 + Json(json!({"error": "NotFound", "message": "Post not found"})), 272 272 + ) 273 273 + .into_response() 274 274 + } 275 275 + }; 276 276 + 277 277 + let local_post = local_records.posts.iter().find(|p| p.uri == uri); 278 278 + 279 279 + let local_post = match local_post { 280 280 + Some(p) => p, 281 281 + None => { 282 282 + return ( 283 283 + StatusCode::NOT_FOUND, 284 284 + Json(json!({"error": "NotFound", "message": "Post not found"})), 285 285 + ) 286 286 + .into_response() 287 287 + } 288 288 + }; 289 289 + 290 290 + let handle = match sqlx::query_scalar!("SELECT handle FROM users WHERE did = $1", requester_did) 291 291 + .fetch_optional(&state.db) 292 292 + .await 293 293 + { 294 294 + Ok(Some(h)) => h, 295 295 + Ok(None) => requester_did.clone(), 296 296 + Err(e) => { 297 297 + warn!("Database error fetching handle: {:?}", e); 298 298 + requester_did.clone() 299 299 + } 300 300 + }; 301 301 + 302 302 + let post_view = format_local_post( 303 303 + local_post, 304 304 + &requester_did, 305 305 + &handle, 306 306 + local_records.profile.as_ref(), 307 307 + ); 308 308 + 309 309 + let thread = PostThreadOutput { 310 310 + thread: ThreadNode::Post(ThreadViewPost { 311 311 + thread_type: Some("app.bsky.feed.defs#threadViewPost".to_string()), 312 312 + post: post_view, 313 313 + parent: None, 314 314 + replies: None, 315 315 + extra: HashMap::new(), 316 316 + }), 317 317 + threadgate: None, 318 318 + }; 319 319 + 320 320 + let lag = get_local_lag(&local_records); 321 321 + format_munged_response(thread, lag) 322 322 + }

+143 -48

src/api/feed/timeline.rs

··· 1 1 - // Yes, I know, this endpoint is an appview one, not for PDS. Who cares!! 2 2 - // Yes, this only gets posts that our DB/instance knows about. Who cares!!! 3 3 - 1 1 + use crate::api::read_after_write::{ 2 2 + extract_repo_rev, format_local_post, format_munged_response, get_local_lag, 3 3 + get_records_since_rev, insert_posts_into_feed, proxy_to_appview, FeedOutput, FeedViewPost, 4 4 + PostView, 5 5 + }; 4 6 use crate::state::AppState; 5 7 use axum::{ 6 6 - Json, 7 7 - extract::State, 8 8 + extract::{Query, State}, 8 9 http::StatusCode, 9 10 response::{IntoResponse, Response}, 11 11 + Json, 10 12 }; 11 13 use jacquard_repo::storage::BlockStore; 12 12 - use serde::Serialize; 13 13 - use serde_json::{Value, json}; 14 14 - use tracing::error; 14 14 + use serde::Deserialize; 15 15 + use serde_json::{json, Value}; 16 16 + use std::collections::HashMap; 17 17 + use tracing::warn; 15 18 16 16 - #[derive(Serialize)] 17 17 - pub struct TimelineOutput { 18 18 - pub feed: Vec<FeedViewPost>, 19 19 + #[derive(Deserialize)] 20 20 + pub struct GetTimelineParams { 21 21 + pub algorithm: Option<String>, 22 22 + pub limit: Option<u32>, 19 23 pub cursor: Option<String>, 20 24 } 21 25 22 22 - #[derive(Serialize)] 23 23 - pub struct FeedViewPost { 24 24 - pub post: PostView, 25 25 - } 26 26 - 27 27 - #[derive(Serialize)] 28 28 - #[serde(rename_all = "camelCase")] 29 29 - pub struct PostView { 30 30 - pub uri: String, 31 31 - pub cid: String, 32 32 - pub author: AuthorView, 33 33 - pub record: Value, 34 34 - pub indexed_at: String, 35 35 - } 36 36 - 37 37 - #[derive(Serialize)] 38 38 - pub struct AuthorView { 39 39 - pub did: String, 40 40 - pub handle: String, 41 41 - } 42 42 - 43 26 pub async fn get_timeline( 44 27 State(state): State<AppState>, 45 28 headers: axum::http::HeaderMap, 29 29 + Query(params): Query<GetTimelineParams>, 46 30 ) -> Response { 47 31 let token = match crate::auth::extract_bearer_token_from_header( 48 48 - headers.get("Authorization").and_then(|h| h.to_str().ok()) 32 32 + headers.get("Authorization").and_then(|h| h.to_str().ok()), 49 33 ) { 50 34 Some(t) => t, 51 35 None => { ··· 68 52 } 69 53 }; 70 54 71 71 - let user_query = sqlx::query!("SELECT id FROM users WHERE did = $1", auth_user.did) 55 55 + match std::env::var("APPVIEW_URL") { 56 56 + Ok(url) if !url.starts_with("http://127.0.0.1") => { 57 57 + return get_timeline_with_appview(&state, &headers, &params, &auth_user.did).await; 58 58 + } 59 59 + _ => {} 60 60 + } 61 61 + 62 62 + get_timeline_local_only(&state, &auth_user.did).await 63 63 + } 64 64 + 65 65 + async fn get_timeline_with_appview( 66 66 + state: &AppState, 67 67 + headers: &axum::http::HeaderMap, 68 68 + params: &GetTimelineParams, 69 69 + auth_did: &str, 70 70 + ) -> Response { 71 71 + let auth_header = headers.get("Authorization").and_then(|h| h.to_str().ok()); 72 72 + 73 73 + let mut query_params = HashMap::new(); 74 74 + if let Some(algo) = &params.algorithm { 75 75 + query_params.insert("algorithm".to_string(), algo.clone()); 76 76 + } 77 77 + if let Some(limit) = params.limit { 78 78 + query_params.insert("limit".to_string(), limit.to_string()); 79 79 + } 80 80 + if let Some(cursor) = &params.cursor { 81 81 + query_params.insert("cursor".to_string(), cursor.clone()); 82 82 + } 83 83 + 84 84 + let proxy_result = 85 85 + match proxy_to_appview("app.bsky.feed.getTimeline", &query_params, auth_header).await { 86 86 + Ok(r) => r, 87 87 + Err(e) => return e, 88 88 + }; 89 89 + 90 90 + if !proxy_result.status.is_success() { 91 91 + return (proxy_result.status, proxy_result.body).into_response(); 92 92 + } 93 93 + 94 94 + let rev = extract_repo_rev(&proxy_result.headers); 95 95 + if rev.is_none() { 96 96 + return (proxy_result.status, proxy_result.body).into_response(); 97 97 + } 98 98 + let rev = rev.unwrap(); 99 99 + 100 100 + let mut feed_output: FeedOutput = match serde_json::from_slice(&proxy_result.body) { 101 101 + Ok(f) => f, 102 102 + Err(e) => { 103 103 + warn!("Failed to parse timeline response: {:?}", e); 104 104 + return (proxy_result.status, proxy_result.body).into_response(); 105 105 + } 106 106 + }; 107 107 + 108 108 + let local_records = match get_records_since_rev(state, auth_did, &rev).await { 109 109 + Ok(r) => r, 110 110 + Err(e) => { 111 111 + warn!("Failed to get local records: {}", e); 112 112 + return (proxy_result.status, proxy_result.body).into_response(); 113 113 + } 114 114 + }; 115 115 + 116 116 + if local_records.count == 0 { 117 117 + return (proxy_result.status, proxy_result.body).into_response(); 118 118 + } 119 119 + 120 120 + let handle = match sqlx::query_scalar!("SELECT handle FROM users WHERE did = $1", auth_did) 72 121 .fetch_optional(&state.db) 73 73 - .await; 122 122 + .await 123 123 + { 124 124 + Ok(Some(h)) => h, 125 125 + Ok(None) => auth_did.to_string(), 126 126 + Err(e) => { 127 127 + warn!("Database error fetching handle: {:?}", e); 128 128 + auth_did.to_string() 129 129 + } 130 130 + }; 74 131 75 75 - let user_id = match user_query { 76 76 - Ok(Some(row)) => row.id, 77 77 - _ => { 132 132 + let local_posts: Vec<_> = local_records 133 133 + .posts 134 134 + .iter() 135 135 + .map(|p| format_local_post(p, auth_did, &handle, local_records.profile.as_ref())) 136 136 + .collect(); 137 137 + 138 138 + insert_posts_into_feed(&mut feed_output.feed, local_posts); 139 139 + 140 140 + let lag = get_local_lag(&local_records); 141 141 + format_munged_response(feed_output, lag) 142 142 + } 143 143 + 144 144 + async fn get_timeline_local_only(state: &AppState, auth_did: &str) -> Response { 145 145 + let user_id: uuid::Uuid = match sqlx::query_scalar!( 146 146 + "SELECT id FROM users WHERE did = $1", 147 147 + auth_did 148 148 + ) 149 149 + .fetch_optional(&state.db) 150 150 + .await 151 151 + { 152 152 + Ok(Some(id)) => id, 153 153 + Ok(None) => { 78 154 return ( 79 155 StatusCode::INTERNAL_SERVER_ERROR, 80 156 Json(json!({"error": "InternalError", "message": "User not found"})), 81 157 ) 82 158 .into_response(); 83 159 } 160 160 + Err(e) => { 161 161 + warn!("Database error fetching user: {:?}", e); 162 162 + return ( 163 163 + StatusCode::INTERNAL_SERVER_ERROR, 164 164 + Json(json!({"error": "InternalError", "message": "Database error"})), 165 165 + ) 166 166 + .into_response(); 167 167 + } 84 168 }; 85 169 86 170 let follows_query = sqlx::query!( 87 87 - "SELECT record_cid FROM records WHERE repo_id = $1 AND collection = 'app.bsky.graph.follow'", 171 171 + "SELECT record_cid FROM records WHERE repo_id = $1 AND collection = 'app.bsky.graph.follow' LIMIT 5000", 88 172 user_id 89 173 ) 90 90 - .fetch_all(&state.db) 91 91 - .await; 174 174 + .fetch_all(&state.db) 175 175 + .await; 92 176 93 177 let follow_cids: Vec<String> = match follows_query { 94 178 Ok(rows) => rows.iter().map(|r| r.record_cid.clone()).collect(), 95 95 - Err(e) => { 96 96 - error!("Failed to get follows: {:?}", e); 179 179 + Err(_) => { 97 180 return ( 98 181 StatusCode::INTERNAL_SERVER_ERROR, 99 182 Json(json!({"error": "InternalError"})), ··· 127 210 if followed_dids.is_empty() { 128 211 return ( 129 212 StatusCode::OK, 130 130 - Json(TimelineOutput { 213 213 + Json(FeedOutput { 131 214 feed: vec![], 132 215 cursor: None, 133 216 }), ··· 150 233 151 234 let posts = match posts_result { 152 235 Ok(rows) => rows, 153 153 - Err(e) => { 154 154 - error!("Failed to get posts: {:?}", e); 236 236 + Err(_) => { 155 237 return ( 156 238 StatusCode::INTERNAL_SERVER_ERROR, 157 239 Json(json!({"error": "InternalError"})), ··· 190 272 post: PostView { 191 273 uri, 192 274 cid: record_cid, 193 193 - author: AuthorView { 275 275 + author: crate::api::read_after_write::AuthorView { 194 276 did: author_did, 195 277 handle: author_handle, 278 278 + display_name: None, 279 279 + avatar: None, 280 280 + extra: HashMap::new(), 196 281 }, 197 282 record, 198 283 indexed_at: created_at.to_rfc3339(), 284 284 + embed: None, 285 285 + reply_count: 0, 286 286 + repost_count: 0, 287 287 + like_count: 0, 288 288 + quote_count: 0, 289 289 + extra: HashMap::new(), 199 290 }, 291 291 + reply: None, 292 292 + reason: None, 293 293 + feed_context: None, 294 294 + extra: HashMap::new(), 200 295 }); 201 296 } 202 297 203 203 - (StatusCode::OK, Json(TimelineOutput { feed, cursor: None })).into_response() 298 298 + (StatusCode::OK, Json(FeedOutput { feed, cursor: None })).into_response() 204 299 }

+4

src/api/mod.rs

··· 4 4 pub mod feed; 5 5 pub mod identity; 6 6 pub mod moderation; 7 7 + pub mod notification; 7 8 pub mod proxy; 9 9 + pub mod proxy_client; 10 10 + pub mod read_after_write; 8 11 pub mod repo; 9 12 pub mod server; 10 13 pub mod validation; 11 14 12 15 pub use error::ApiError; 16 16 + pub use proxy_client::{proxy_client, validate_at_uri, validate_did, validate_limit, AtUriParts};

+3

src/api/notification/mod.rs

··· 1 1 + mod register_push; 2 2 + 3 3 + pub use register_push::register_push;

+166

src/api/notification/register_push.rs

··· 1 1 + use crate::api::proxy_client::{is_ssrf_safe, proxy_client, validate_did}; 2 2 + use crate::api::ApiError; 3 3 + use crate::state::AppState; 4 4 + use axum::{ 5 5 + extract::State, 6 6 + http::{HeaderMap, StatusCode}, 7 7 + response::{IntoResponse, Response}, 8 8 + Json, 9 9 + }; 10 10 + use serde::Deserialize; 11 11 + use serde_json::json; 12 12 + use tracing::{error, info}; 13 13 + 14 14 + #[derive(Deserialize)] 15 15 + #[serde(rename_all = "camelCase")] 16 16 + pub struct RegisterPushInput { 17 17 + pub service_did: String, 18 18 + pub token: String, 19 19 + pub platform: String, 20 20 + pub app_id: String, 21 21 + } 22 22 + 23 23 + const VALID_PLATFORMS: &[&str] = &["ios", "android", "web"]; 24 24 + 25 25 + pub async fn register_push( 26 26 + State(state): State<AppState>, 27 27 + headers: HeaderMap, 28 28 + Json(input): Json<RegisterPushInput>, 29 29 + ) -> Response { 30 30 + let token = match crate::auth::extract_bearer_token_from_header( 31 31 + headers.get("Authorization").and_then(|h| h.to_str().ok()), 32 32 + ) { 33 33 + Some(t) => t, 34 34 + None => return ApiError::AuthenticationRequired.into_response(), 35 35 + }; 36 36 + 37 37 + let auth_user = match crate::auth::validate_bearer_token(&state.db, &token).await { 38 38 + Ok(user) => user, 39 39 + Err(e) => return ApiError::from(e).into_response(), 40 40 + }; 41 41 + 42 42 + if let Err(e) = validate_did(&input.service_did) { 43 43 + return ApiError::InvalidRequest(format!("Invalid serviceDid: {}", e)).into_response(); 44 44 + } 45 45 + 46 46 + if input.token.is_empty() || input.token.len() > 4096 { 47 47 + return ApiError::InvalidRequest("Invalid push token".to_string()).into_response(); 48 48 + } 49 49 + 50 50 + if !VALID_PLATFORMS.contains(&input.platform.as_str()) { 51 51 + return ApiError::InvalidRequest(format!( 52 52 + "Invalid platform. Must be one of: {}", 53 53 + VALID_PLATFORMS.join(", ") 54 54 + )) 55 55 + .into_response(); 56 56 + } 57 57 + 58 58 + if input.app_id.is_empty() || input.app_id.len() > 256 { 59 59 + return ApiError::InvalidRequest("Invalid appId".to_string()).into_response(); 60 60 + } 61 61 + 62 62 + let appview_url = match std::env::var("APPVIEW_URL") { 63 63 + Ok(url) => url, 64 64 + Err(_) => { 65 65 + return ApiError::UpstreamUnavailable("No upstream AppView configured".to_string()) 66 66 + .into_response(); 67 67 + } 68 68 + }; 69 69 + 70 70 + if let Err(e) = is_ssrf_safe(&appview_url) { 71 71 + error!("SSRF check failed for appview URL: {}", e); 72 72 + return ApiError::UpstreamUnavailable(format!("Invalid upstream URL: {}", e)) 73 73 + .into_response(); 74 74 + } 75 75 + 76 76 + let key_row = match sqlx::query!( 77 77 + "SELECT key_bytes, encryption_version FROM user_keys k JOIN users u ON k.user_id = u.id WHERE u.did = $1", 78 78 + auth_user.did 79 79 + ) 80 80 + .fetch_optional(&state.db) 81 81 + .await 82 82 + { 83 83 + Ok(Some(row)) => row, 84 84 + Ok(None) => { 85 85 + error!(did = %auth_user.did, "No signing key found for user"); 86 86 + return ApiError::InternalError.into_response(); 87 87 + } 88 88 + Err(e) => { 89 89 + error!(error = ?e, "Database error fetching signing key"); 90 90 + return ApiError::DatabaseError.into_response(); 91 91 + } 92 92 + }; 93 93 + 94 94 + let decrypted_key = 95 95 + match crate::config::decrypt_key(&key_row.key_bytes, key_row.encryption_version) { 96 96 + Ok(k) => k, 97 97 + Err(e) => { 98 98 + error!(error = ?e, "Failed to decrypt signing key"); 99 99 + return ApiError::InternalError.into_response(); 100 100 + } 101 101 + }; 102 102 + 103 103 + let service_token = match crate::auth::create_service_token( 104 104 + &auth_user.did, 105 105 + &input.service_did, 106 106 + "app.bsky.notification.registerPush", 107 107 + &decrypted_key, 108 108 + ) { 109 109 + Ok(t) => t, 110 110 + Err(e) => { 111 111 + error!(error = ?e, "Failed to create service token"); 112 112 + return ApiError::InternalError.into_response(); 113 113 + } 114 114 + }; 115 115 + 116 116 + let target_url = format!("{}/xrpc/app.bsky.notification.registerPush", appview_url); 117 117 + info!( 118 118 + target = %target_url, 119 119 + service_did = %input.service_did, 120 120 + platform = %input.platform, 121 121 + "Proxying registerPush request" 122 122 + ); 123 123 + 124 124 + let client = proxy_client(); 125 125 + let request_body = json!({ 126 126 + "serviceDid": input.service_did, 127 127 + "token": input.token, 128 128 + "platform": input.platform, 129 129 + "appId": input.app_id 130 130 + }); 131 131 + 132 132 + match client 133 133 + .post(&target_url) 134 134 + .header("Authorization", format!("Bearer {}", service_token)) 135 135 + .header("Content-Type", "application/json") 136 136 + .json(&request_body) 137 137 + .send() 138 138 + .await 139 139 + { 140 140 + Ok(resp) => { 141 141 + let status = 142 142 + StatusCode::from_u16(resp.status().as_u16()).unwrap_or(StatusCode::BAD_GATEWAY); 143 143 + if status.is_success() { 144 144 + StatusCode::OK.into_response() 145 145 + } else { 146 146 + let body = resp.bytes().await.unwrap_or_default(); 147 147 + error!( 148 148 + status = %status, 149 149 + "registerPush upstream error" 150 150 + ); 151 151 + ApiError::from_upstream_response(status.as_u16(), &body).into_response() 152 152 + } 153 153 + } 154 154 + Err(e) => { 155 155 + error!(error = ?e, "Error proxying registerPush"); 156 156 + if e.is_timeout() { 157 157 + ApiError::UpstreamTimeout.into_response() 158 158 + } else if e.is_connect() { 159 159 + ApiError::UpstreamUnavailable("Failed to connect to upstream".to_string()) 160 160 + .into_response() 161 161 + } else { 162 162 + ApiError::UpstreamFailure.into_response() 163 163 + } 164 164 + } 165 165 + } 166 166 + }

+8 -14

src/api/proxy.rs

··· 46 46 if let Some(token) = crate::auth::extract_bearer_token_from_header( 47 47 headers.get("Authorization").and_then(|h| h.to_str().ok()) 48 48 ) { 49 49 - if let Ok(did) = crate::auth::get_did_from_token(&token) { 50 50 - let key_row = sqlx::query!("SELECT k.key_bytes, k.encryption_version FROM user_keys k JOIN users u ON k.user_id = u.id WHERE u.did = $1", did) 51 51 - .fetch_optional(&state.db) 52 52 - .await; 53 53 - 54 54 - if let Ok(Some(row)) = key_row { 55 55 - if let Ok(decrypted_key) = crate::config::decrypt_key(&row.key_bytes, row.encryption_version) { 56 56 - if let Ok(new_token) = 57 57 - crate::auth::create_service_token(&did, aud, &method, &decrypted_key) 49 49 + if let Ok(auth_user) = crate::auth::validate_bearer_token(&state.db, &token).await { 50 50 + if let Some(key_bytes) = auth_user.key_bytes { 51 51 + if let Ok(new_token) = 52 52 + crate::auth::create_service_token(&auth_user.did, aud, &method, &key_bytes) 53 53 + { 54 54 + if let Ok(val) = 55 55 + axum::http::HeaderValue::from_str(&format!("Bearer {}", new_token)) 58 56 { 59 59 - if let Ok(val) = 60 60 - axum::http::HeaderValue::from_str(&format!("Bearer {}", new_token)) 61 61 - { 62 62 - auth_header_val = Some(val); 63 63 - } 57 57 + auth_header_val = Some(val); 64 58 } 65 59 } 66 60 }

+252

src/api/proxy_client.rs

··· 1 1 + use reqwest::{Client, ClientBuilder, Url}; 2 2 + use std::net::{IpAddr, SocketAddr, ToSocketAddrs}; 3 3 + use std::sync::OnceLock; 4 4 + use std::time::Duration; 5 5 + use tracing::warn; 6 6 + 7 7 + pub const DEFAULT_HEADERS_TIMEOUT: Duration = Duration::from_secs(10); 8 8 + pub const DEFAULT_BODY_TIMEOUT: Duration = Duration::from_secs(30); 9 9 + pub const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(5); 10 10 + pub const MAX_RESPONSE_SIZE: u64 = 10 * 1024 * 1024; 11 11 + 12 12 + static PROXY_CLIENT: OnceLock<Client> = OnceLock::new(); 13 13 + 14 14 + pub fn proxy_client() -> &'static Client { 15 15 + PROXY_CLIENT.get_or_init(|| { 16 16 + ClientBuilder::new() 17 17 + .timeout(DEFAULT_BODY_TIMEOUT) 18 18 + .connect_timeout(DEFAULT_CONNECT_TIMEOUT) 19 19 + .pool_max_idle_per_host(10) 20 20 + .pool_idle_timeout(Duration::from_secs(90)) 21 21 + .redirect(reqwest::redirect::Policy::none()) 22 22 + .build() 23 23 + .expect("Failed to build HTTP client - this indicates a TLS or system configuration issue") 24 24 + }) 25 25 + } 26 26 + 27 27 + pub fn is_ssrf_safe(url: &str) -> Result<(), SsrfError> { 28 28 + let parsed = Url::parse(url).map_err(|_| SsrfError::InvalidUrl)?; 29 29 + 30 30 + let scheme = parsed.scheme(); 31 31 + if scheme != "https" { 32 32 + let allow_http = std::env::var("ALLOW_HTTP_PROXY").is_ok() 33 33 + || url.starts_with("http://127.0.0.1") 34 34 + || url.starts_with("http://localhost"); 35 35 + 36 36 + if !allow_http { 37 37 + return Err(SsrfError::InsecureProtocol(scheme.to_string())); 38 38 + } 39 39 + } 40 40 + 41 41 + let host = parsed.host_str().ok_or(SsrfError::NoHost)?; 42 42 + 43 43 + if host == "localhost" { 44 44 + return Ok(()); 45 45 + } 46 46 + 47 47 + if let Ok(ip) = host.parse::<IpAddr>() { 48 48 + if ip.is_loopback() { 49 49 + return Ok(()); 50 50 + } 51 51 + if !is_unicast_ip(&ip) { 52 52 + return Err(SsrfError::NonUnicastIp(ip.to_string())); 53 53 + } 54 54 + return Ok(()); 55 55 + } 56 56 + 57 57 + let port = parsed.port().unwrap_or(if scheme == "https" { 443 } else { 80 }); 58 58 + let socket_addrs: Vec<SocketAddr> = match (host, port).to_socket_addrs() { 59 59 + Ok(addrs) => addrs.collect(), 60 60 + Err(_) => return Err(SsrfError::DnsResolutionFailed(host.to_string())), 61 61 + }; 62 62 + 63 63 + for addr in &socket_addrs { 64 64 + if !is_unicast_ip(&addr.ip()) { 65 65 + warn!( 66 66 + "DNS resolution for {} returned non-unicast IP: {}", 67 67 + host, 68 68 + addr.ip() 69 69 + ); 70 70 + return Err(SsrfError::NonUnicastIp(addr.ip().to_string())); 71 71 + } 72 72 + } 73 73 + 74 74 + Ok(()) 75 75 + } 76 76 + 77 77 + fn is_unicast_ip(ip: &IpAddr) -> bool { 78 78 + match ip { 79 79 + IpAddr::V4(v4) => { 80 80 + !v4.is_loopback() 81 81 + && !v4.is_broadcast() 82 82 + && !v4.is_multicast() 83 83 + && !v4.is_unspecified() 84 84 + && !v4.is_link_local() 85 85 + && !is_private_v4(v4) 86 86 + } 87 87 + IpAddr::V6(v6) => !v6.is_loopback() && !v6.is_multicast() && !v6.is_unspecified(), 88 88 + } 89 89 + } 90 90 + 91 91 + fn is_private_v4(ip: &std::net::Ipv4Addr) -> bool { 92 92 + let octets = ip.octets(); 93 93 + octets[0] == 10 94 94 + || (octets[0] == 172 && (16..=31).contains(&octets[1])) 95 95 + || (octets[0] == 192 && octets[1] == 168) 96 96 + || (octets[0] == 169 && octets[1] == 254) 97 97 + } 98 98 + 99 99 + #[derive(Debug, Clone)] 100 100 + pub enum SsrfError { 101 101 + InvalidUrl, 102 102 + InsecureProtocol(String), 103 103 + NoHost, 104 104 + NonUnicastIp(String), 105 105 + DnsResolutionFailed(String), 106 106 + } 107 107 + 108 108 + impl std::fmt::Display for SsrfError { 109 109 + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { 110 110 + match self { 111 111 + SsrfError::InvalidUrl => write!(f, "Invalid URL"), 112 112 + SsrfError::InsecureProtocol(p) => write!(f, "Insecure protocol: {}", p), 113 113 + SsrfError::NoHost => write!(f, "No host in URL"), 114 114 + SsrfError::NonUnicastIp(ip) => write!(f, "Non-unicast IP address: {}", ip), 115 115 + SsrfError::DnsResolutionFailed(host) => write!(f, "DNS resolution failed for: {}", host), 116 116 + } 117 117 + } 118 118 + } 119 119 + 120 120 + impl std::error::Error for SsrfError {} 121 121 + 122 122 + pub const HEADERS_TO_FORWARD: &[&str] = &[ 123 123 + "accept-language", 124 124 + "atproto-accept-labelers", 125 125 + "x-bsky-topics", 126 126 + ]; 127 127 + 128 128 + pub const RESPONSE_HEADERS_TO_FORWARD: &[&str] = &[ 129 129 + "atproto-repo-rev", 130 130 + "atproto-content-labelers", 131 131 + "retry-after", 132 132 + "content-type", 133 133 + ]; 134 134 + 135 135 + pub fn validate_at_uri(uri: &str) -> Result<AtUriParts, &'static str> { 136 136 + if !uri.starts_with("at://") { 137 137 + return Err("URI must start with at://"); 138 138 + } 139 139 + 140 140 + let path = uri.trim_start_matches("at://"); 141 141 + let parts: Vec<&str> = path.split('/').collect(); 142 142 + 143 143 + if parts.is_empty() { 144 144 + return Err("URI missing DID"); 145 145 + } 146 146 + 147 147 + let did = parts[0]; 148 148 + if !did.starts_with("did:") { 149 149 + return Err("Invalid DID in URI"); 150 150 + } 151 151 + 152 152 + if parts.len() > 1 { 153 153 + let collection = parts[1]; 154 154 + if collection.is_empty() || !collection.contains('.') { 155 155 + return Err("Invalid collection NSID"); 156 156 + } 157 157 + } 158 158 + 159 159 + Ok(AtUriParts { 160 160 + did: did.to_string(), 161 161 + collection: parts.get(1).map(|s| s.to_string()), 162 162 + rkey: parts.get(2).map(|s| s.to_string()), 163 163 + }) 164 164 + } 165 165 + 166 166 + #[derive(Debug, Clone)] 167 167 + pub struct AtUriParts { 168 168 + pub did: String, 169 169 + pub collection: Option<String>, 170 170 + pub rkey: Option<String>, 171 171 + } 172 172 + 173 173 + pub fn validate_limit(limit: Option<u32>, default: u32, max: u32) -> u32 { 174 174 + match limit { 175 175 + Some(l) if l == 0 => default, 176 176 + Some(l) if l > max => max, 177 177 + Some(l) => l, 178 178 + None => default, 179 179 + } 180 180 + } 181 181 + 182 182 + pub fn validate_did(did: &str) -> Result<(), &'static str> { 183 183 + if !did.starts_with("did:") { 184 184 + return Err("Invalid DID format"); 185 185 + } 186 186 + 187 187 + let parts: Vec<&str> = did.split(':').collect(); 188 188 + if parts.len() < 3 { 189 189 + return Err("DID must have at least method and identifier"); 190 190 + } 191 191 + 192 192 + let method = parts[1]; 193 193 + if method != "plc" && method != "web" { 194 194 + return Err("Unsupported DID method"); 195 195 + } 196 196 + 197 197 + Ok(()) 198 198 + } 199 199 + 200 200 + #[cfg(test)] 201 201 + mod tests { 202 202 + use super::*; 203 203 + 204 204 + #[test] 205 205 + fn test_ssrf_safe_https() { 206 206 + assert!(is_ssrf_safe("https://api.bsky.app/xrpc/test").is_ok()); 207 207 + } 208 208 + 209 209 + #[test] 210 210 + fn test_ssrf_blocks_http_by_default() { 211 211 + let result = is_ssrf_safe("http://external.example.com/xrpc/test"); 212 212 + assert!(matches!(result, Err(SsrfError::InsecureProtocol(_)) | Err(SsrfError::DnsResolutionFailed(_)))); 213 213 + } 214 214 + 215 215 + #[test] 216 216 + fn test_ssrf_allows_localhost_http() { 217 217 + assert!(is_ssrf_safe("http://127.0.0.1:8080/test").is_ok()); 218 218 + assert!(is_ssrf_safe("http://localhost:8080/test").is_ok()); 219 219 + } 220 220 + 221 221 + #[test] 222 222 + fn test_validate_at_uri() { 223 223 + let result = validate_at_uri("at://did:plc:test/app.bsky.feed.post/abc123"); 224 224 + assert!(result.is_ok()); 225 225 + let parts = result.unwrap(); 226 226 + assert_eq!(parts.did, "did:plc:test"); 227 227 + assert_eq!(parts.collection, Some("app.bsky.feed.post".to_string())); 228 228 + assert_eq!(parts.rkey, Some("abc123".to_string())); 229 229 + } 230 230 + 231 231 + #[test] 232 232 + fn test_validate_at_uri_invalid() { 233 233 + assert!(validate_at_uri("https://example.com").is_err()); 234 234 + assert!(validate_at_uri("at://notadid/collection/rkey").is_err()); 235 235 + } 236 236 + 237 237 + #[test] 238 238 + fn test_validate_limit() { 239 239 + assert_eq!(validate_limit(None, 50, 100), 50); 240 240 + assert_eq!(validate_limit(Some(0), 50, 100), 50); 241 241 + assert_eq!(validate_limit(Some(200), 50, 100), 100); 242 242 + assert_eq!(validate_limit(Some(75), 50, 100), 75); 243 243 + } 244 244 + 245 245 + #[test] 246 246 + fn test_validate_did() { 247 247 + assert!(validate_did("did:plc:abc123").is_ok()); 248 248 + assert!(validate_did("did:web:example.com").is_ok()); 249 249 + assert!(validate_did("notadid").is_err()); 250 250 + assert!(validate_did("did:unknown:test").is_err()); 251 251 + } 252 252 + }

+433

src/api/read_after_write.rs

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

+6 -4

src/api/repo/record/utils.rs

··· 25 25 current_root_cid: Option<Cid>, 26 26 new_mst_root: Cid, 27 27 ops: Vec<RecordOp>, 28 28 - blocks_cids: &Vec<String>, 28 28 + blocks_cids: &[String], 29 29 ) -> Result<CommitResult, String> { 30 30 let key_row = sqlx::query!( 31 31 "SELECT key_bytes, encryption_version FROM user_keys WHERE user_id = $1", ··· 63 63 .await 64 64 .map_err(|e| format!("DB Error (repos): {}", e))?; 65 65 66 66 + let rev_str = rev.to_string(); 66 67 for op in &ops { 67 68 match op { 68 69 RecordOp::Create { collection, rkey, cid } | RecordOp::Update { collection, rkey, cid } => { 69 70 sqlx::query!( 70 70 - "INSERT INTO records (repo_id, collection, rkey, record_cid) VALUES ($1, $2, $3, $4) 71 71 - ON CONFLICT (repo_id, collection, rkey) DO UPDATE SET record_cid = $4, created_at = NOW()", 71 71 + "INSERT INTO records (repo_id, collection, rkey, record_cid, repo_rev) VALUES ($1, $2, $3, $4, $5) 72 72 + ON CONFLICT (repo_id, collection, rkey) DO UPDATE SET record_cid = $4, repo_rev = $5, created_at = NOW()", 72 73 user_id, 73 74 collection, 74 75 rkey, 75 75 - cid.to_string() 76 76 + cid.to_string(), 77 77 + rev_str 76 78 ) 77 79 .execute(&mut *tx) 78 80 .await

+20 -1

src/lib.rs

··· 296 296 "/xrpc/app.bsky.actor.getProfiles", 297 297 get(api::actor::get_profiles), 298 298 ) 299 299 - // I know I know, I'm not supposed to implement appview endpoints. Leave me be 300 299 .route( 301 300 "/xrpc/app.bsky.feed.getTimeline", 302 301 get(api::feed::get_timeline), 302 302 + ) 303 303 + .route( 304 304 + "/xrpc/app.bsky.feed.getAuthorFeed", 305 305 + get(api::feed::get_author_feed), 306 306 + ) 307 307 + .route( 308 308 + "/xrpc/app.bsky.feed.getActorLikes", 309 309 + get(api::feed::get_actor_likes), 310 310 + ) 311 311 + .route( 312 312 + "/xrpc/app.bsky.feed.getPostThread", 313 313 + get(api::feed::get_post_thread), 314 314 + ) 315 315 + .route( 316 316 + "/xrpc/app.bsky.feed.getFeed", 317 317 + get(api::feed::get_feed), 318 318 + ) 319 319 + .route( 320 320 + "/xrpc/app.bsky.notification.registerPush", 321 321 + post(api::notification::register_push), 303 322 ) 304 323 .route("/.well-known/did.json", get(api::identity::well_known_did)) 305 324 .route("/u/{handle}/did.json", get(api::identity::user_did_doc))

+149

tests/appview_integration.rs

··· 1 1 + mod common; 2 2 + 3 3 + use common::{base_url, client, create_account_and_login}; 4 4 + use reqwest::StatusCode; 5 5 + use serde_json::{json, Value}; 6 6 + 7 7 + #[tokio::test] 8 8 + async fn test_get_author_feed_returns_appview_data() { 9 9 + let client = client(); 10 10 + let base = base_url().await; 11 11 + let (jwt, did) = create_account_and_login(&client).await; 12 12 + 13 13 + let res = client 14 14 + .get(format!( 15 15 + "{}/xrpc/app.bsky.feed.getAuthorFeed?actor={}", 16 16 + base, did 17 17 + )) 18 18 + .header("Authorization", format!("Bearer {}", jwt)) 19 19 + .send() 20 20 + .await 21 21 + .unwrap(); 22 22 + 23 23 + assert_eq!(res.status(), StatusCode::OK); 24 24 + 25 25 + let body: Value = res.json().await.unwrap(); 26 26 + assert!(body["feed"].is_array(), "Response should have feed array"); 27 27 + let feed = body["feed"].as_array().unwrap(); 28 28 + assert_eq!(feed.len(), 1, "Feed should have 1 post from appview"); 29 29 + assert_eq!( 30 30 + feed[0]["post"]["record"]["text"].as_str(), 31 31 + Some("Author feed post from appview"), 32 32 + "Post text should match appview response" 33 33 + ); 34 34 + } 35 35 + 36 36 + #[tokio::test] 37 37 + async fn test_get_actor_likes_returns_appview_data() { 38 38 + let client = client(); 39 39 + let base = base_url().await; 40 40 + let (jwt, did) = create_account_and_login(&client).await; 41 41 + 42 42 + let res = client 43 43 + .get(format!( 44 44 + "{}/xrpc/app.bsky.feed.getActorLikes?actor={}", 45 45 + base, did 46 46 + )) 47 47 + .header("Authorization", format!("Bearer {}", jwt)) 48 48 + .send() 49 49 + .await 50 50 + .unwrap(); 51 51 + 52 52 + assert_eq!(res.status(), StatusCode::OK); 53 53 + 54 54 + let body: Value = res.json().await.unwrap(); 55 55 + assert!(body["feed"].is_array(), "Response should have feed array"); 56 56 + let feed = body["feed"].as_array().unwrap(); 57 57 + assert_eq!(feed.len(), 1, "Feed should have 1 liked post from appview"); 58 58 + assert_eq!( 59 59 + feed[0]["post"]["record"]["text"].as_str(), 60 60 + Some("Liked post from appview"), 61 61 + "Post text should match appview response" 62 62 + ); 63 63 + } 64 64 + 65 65 + #[tokio::test] 66 66 + async fn test_get_post_thread_returns_appview_data() { 67 67 + let client = client(); 68 68 + let base = base_url().await; 69 69 + let (jwt, did) = create_account_and_login(&client).await; 70 70 + 71 71 + let res = client 72 72 + .get(format!( 73 73 + "{}/xrpc/app.bsky.feed.getPostThread?uri=at://{}/app.bsky.feed.post/test123", 74 74 + base, did 75 75 + )) 76 76 + .header("Authorization", format!("Bearer {}", jwt)) 77 77 + .send() 78 78 + .await 79 79 + .unwrap(); 80 80 + 81 81 + assert_eq!(res.status(), StatusCode::OK); 82 82 + 83 83 + let body: Value = res.json().await.unwrap(); 84 84 + assert!(body["thread"].is_object(), "Response should have thread object"); 85 85 + assert_eq!( 86 86 + body["thread"]["$type"].as_str(), 87 87 + Some("app.bsky.feed.defs#threadViewPost"), 88 88 + "Thread should be a threadViewPost" 89 89 + ); 90 90 + assert_eq!( 91 91 + body["thread"]["post"]["record"]["text"].as_str(), 92 92 + Some("Thread post from appview"), 93 93 + "Post text should match appview response" 94 94 + ); 95 95 + } 96 96 + 97 97 + #[tokio::test] 98 98 + async fn test_get_feed_returns_appview_data() { 99 99 + let client = client(); 100 100 + let base = base_url().await; 101 101 + let (jwt, _did) = create_account_and_login(&client).await; 102 102 + 103 103 + let res = client 104 104 + .get(format!( 105 105 + "{}/xrpc/app.bsky.feed.getFeed?feed=at://did:plc:test/app.bsky.feed.generator/test", 106 106 + base 107 107 + )) 108 108 + .header("Authorization", format!("Bearer {}", jwt)) 109 109 + .send() 110 110 + .await 111 111 + .unwrap(); 112 112 + 113 113 + assert_eq!(res.status(), StatusCode::OK); 114 114 + 115 115 + let body: Value = res.json().await.unwrap(); 116 116 + assert!(body["feed"].is_array(), "Response should have feed array"); 117 117 + let feed = body["feed"].as_array().unwrap(); 118 118 + assert_eq!(feed.len(), 1, "Feed should have 1 post from appview"); 119 119 + assert_eq!( 120 120 + feed[0]["post"]["record"]["text"].as_str(), 121 121 + Some("Custom feed post from appview"), 122 122 + "Post text should match appview response" 123 123 + ); 124 124 + } 125 125 + 126 126 + #[tokio::test] 127 127 + async fn test_register_push_proxies_to_appview() { 128 128 + let client = client(); 129 129 + let base = base_url().await; 130 130 + let (jwt, _did) = create_account_and_login(&client).await; 131 131 + 132 132 + let res = client 133 133 + .post(format!( 134 134 + "{}/xrpc/app.bsky.notification.registerPush", 135 135 + base 136 136 + )) 137 137 + .header("Authorization", format!("Bearer {}", jwt)) 138 138 + .json(&json!({ 139 139 + "serviceDid": "did:web:example.com", 140 140 + "token": "test-push-token", 141 141 + "platform": "ios", 142 142 + "appId": "xyz.bsky.app" 143 143 + })) 144 144 + .send() 145 145 + .await 146 146 + .unwrap(); 147 147 + 148 148 + assert_eq!(res.status(), StatusCode::OK); 149 149 + }

+116

tests/common/mod.rs

··· 233 233 }))) 234 234 .mount(mock_server) 235 235 .await; 236 236 + 237 237 + Mock::given(method("GET")) 238 238 + .and(path("/xrpc/app.bsky.feed.getTimeline")) 239 239 + .respond_with( 240 240 + ResponseTemplate::new(200) 241 241 + .insert_header("atproto-repo-rev", "0") 242 242 + .set_body_json(json!({ 243 243 + "feed": [], 244 244 + "cursor": null 245 245 + })) 246 246 + ) 247 247 + .mount(mock_server) 248 248 + .await; 249 249 + 250 250 + Mock::given(method("GET")) 251 251 + .and(path("/xrpc/app.bsky.feed.getAuthorFeed")) 252 252 + .respond_with( 253 253 + ResponseTemplate::new(200) 254 254 + .insert_header("atproto-repo-rev", "0") 255 255 + .set_body_json(json!({ 256 256 + "feed": [{ 257 257 + "post": { 258 258 + "uri": "at://did:plc:mock-author/app.bsky.feed.post/from-appview-author", 259 259 + "cid": "bafyappview123", 260 260 + "author": {"did": "did:plc:mock-author", "handle": "mock.author"}, 261 261 + "record": { 262 262 + "$type": "app.bsky.feed.post", 263 263 + "text": "Author feed post from appview", 264 264 + "createdAt": "2025-01-01T00:00:00Z" 265 265 + }, 266 266 + "indexedAt": "2025-01-01T00:00:00Z" 267 267 + } 268 268 + }], 269 269 + "cursor": "author-cursor" 270 270 + })), 271 271 + ) 272 272 + .mount(mock_server) 273 273 + .await; 274 274 + 275 275 + Mock::given(method("GET")) 276 276 + .and(path("/xrpc/app.bsky.feed.getActorLikes")) 277 277 + .respond_with( 278 278 + ResponseTemplate::new(200) 279 279 + .insert_header("atproto-repo-rev", "0") 280 280 + .set_body_json(json!({ 281 281 + "feed": [{ 282 282 + "post": { 283 283 + "uri": "at://did:plc:mock-likes/app.bsky.feed.post/liked-post", 284 284 + "cid": "bafyliked123", 285 285 + "author": {"did": "did:plc:mock-likes", "handle": "mock.likes"}, 286 286 + "record": { 287 287 + "$type": "app.bsky.feed.post", 288 288 + "text": "Liked post from appview", 289 289 + "createdAt": "2025-01-01T00:00:00Z" 290 290 + }, 291 291 + "indexedAt": "2025-01-01T00:00:00Z" 292 292 + } 293 293 + }], 294 294 + "cursor": null 295 295 + })), 296 296 + ) 297 297 + .mount(mock_server) 298 298 + .await; 299 299 + 300 300 + Mock::given(method("GET")) 301 301 + .and(path("/xrpc/app.bsky.feed.getPostThread")) 302 302 + .respond_with( 303 303 + ResponseTemplate::new(200) 304 304 + .insert_header("atproto-repo-rev", "0") 305 305 + .set_body_json(json!({ 306 306 + "thread": { 307 307 + "$type": "app.bsky.feed.defs#threadViewPost", 308 308 + "post": { 309 309 + "uri": "at://did:plc:mock/app.bsky.feed.post/thread-post", 310 310 + "cid": "bafythread123", 311 311 + "author": {"did": "did:plc:mock", "handle": "mock.handle"}, 312 312 + "record": { 313 313 + "$type": "app.bsky.feed.post", 314 314 + "text": "Thread post from appview", 315 315 + "createdAt": "2025-01-01T00:00:00Z" 316 316 + }, 317 317 + "indexedAt": "2025-01-01T00:00:00Z" 318 318 + }, 319 319 + "replies": [] 320 320 + } 321 321 + })), 322 322 + ) 323 323 + .mount(mock_server) 324 324 + .await; 325 325 + 326 326 + Mock::given(method("GET")) 327 327 + .and(path("/xrpc/app.bsky.feed.getFeed")) 328 328 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ 329 329 + "feed": [{ 330 330 + "post": { 331 331 + "uri": "at://did:plc:mock-feed/app.bsky.feed.post/custom-feed-post", 332 332 + "cid": "bafyfeed123", 333 333 + "author": {"did": "did:plc:mock-feed", "handle": "mock.feed"}, 334 334 + "record": { 335 335 + "$type": "app.bsky.feed.post", 336 336 + "text": "Custom feed post from appview", 337 337 + "createdAt": "2025-01-01T00:00:00Z" 338 338 + }, 339 339 + "indexedAt": "2025-01-01T00:00:00Z" 340 340 + } 341 341 + }], 342 342 + "cursor": null 343 343 + }))) 344 344 + .mount(mock_server) 345 345 + .await; 346 346 + 347 347 + Mock::given(method("POST")) 348 348 + .and(path("/xrpc/app.bsky.notification.registerPush")) 349 349 + .respond_with(ResponseTemplate::new(200)) 350 350 + .mount(mock_server) 351 351 + .await; 236 352 } 237 353 238 354 async fn spawn_app(database_url: String) -> String {

+122

tests/feed.rs

··· 1 1 + mod common; 2 2 + 3 3 + use common::{base_url, client, create_account_and_login}; 4 4 + use serde_json::json; 5 5 + 6 6 + #[tokio::test] 7 7 + async fn test_get_timeline_requires_auth() { 8 8 + let client = client(); 9 9 + let base = base_url().await; 10 10 + 11 11 + let res = client 12 12 + .get(format!("{}/xrpc/app.bsky.feed.getTimeline", base)) 13 13 + .send() 14 14 + .await 15 15 + .unwrap(); 16 16 + 17 17 + assert_eq!(res.status(), 401); 18 18 + } 19 19 + 20 20 + #[tokio::test] 21 21 + async fn test_get_author_feed_requires_actor() { 22 22 + let client = client(); 23 23 + let base = base_url().await; 24 24 + let (jwt, _did) = create_account_and_login(&client).await; 25 25 + 26 26 + let res = client 27 27 + .get(format!("{}/xrpc/app.bsky.feed.getAuthorFeed", base)) 28 28 + .header("Authorization", format!("Bearer {}", jwt)) 29 29 + .send() 30 30 + .await 31 31 + .unwrap(); 32 32 + 33 33 + assert_eq!(res.status(), 400); 34 34 + } 35 35 + 36 36 + #[tokio::test] 37 37 + async fn test_get_actor_likes_requires_actor() { 38 38 + let client = client(); 39 39 + let base = base_url().await; 40 40 + let (jwt, _did) = create_account_and_login(&client).await; 41 41 + 42 42 + let res = client 43 43 + .get(format!("{}/xrpc/app.bsky.feed.getActorLikes", base)) 44 44 + .header("Authorization", format!("Bearer {}", jwt)) 45 45 + .send() 46 46 + .await 47 47 + .unwrap(); 48 48 + 49 49 + assert_eq!(res.status(), 400); 50 50 + } 51 51 + 52 52 + #[tokio::test] 53 53 + async fn test_get_post_thread_requires_uri() { 54 54 + let client = client(); 55 55 + let base = base_url().await; 56 56 + let (jwt, _did) = create_account_and_login(&client).await; 57 57 + 58 58 + let res = client 59 59 + .get(format!("{}/xrpc/app.bsky.feed.getPostThread", base)) 60 60 + .header("Authorization", format!("Bearer {}", jwt)) 61 61 + .send() 62 62 + .await 63 63 + .unwrap(); 64 64 + 65 65 + assert_eq!(res.status(), 400); 66 66 + } 67 67 + 68 68 + #[tokio::test] 69 69 + async fn test_get_feed_requires_auth() { 70 70 + let client = client(); 71 71 + let base = base_url().await; 72 72 + 73 73 + let res = client 74 74 + .get(format!( 75 75 + "{}/xrpc/app.bsky.feed.getFeed?feed=at://did:plc:test/app.bsky.feed.generator/test", 76 76 + base 77 77 + )) 78 78 + .send() 79 79 + .await 80 80 + .unwrap(); 81 81 + 82 82 + assert_eq!(res.status(), 401); 83 83 + } 84 84 + 85 85 + #[tokio::test] 86 86 + async fn test_get_feed_requires_feed_param() { 87 87 + let client = client(); 88 88 + let base = base_url().await; 89 89 + let (jwt, _did) = create_account_and_login(&client).await; 90 90 + 91 91 + let res = client 92 92 + .get(format!("{}/xrpc/app.bsky.feed.getFeed", base)) 93 93 + .header("Authorization", format!("Bearer {}", jwt)) 94 94 + .send() 95 95 + .await 96 96 + .unwrap(); 97 97 + 98 98 + assert_eq!(res.status(), 400); 99 99 + } 100 100 + 101 101 + #[tokio::test] 102 102 + async fn test_register_push_requires_auth() { 103 103 + let client = client(); 104 104 + let base = base_url().await; 105 105 + 106 106 + let res = client 107 107 + .post(format!( 108 108 + "{}/xrpc/app.bsky.notification.registerPush", 109 109 + base 110 110 + )) 111 111 + .json(&json!({ 112 112 + "serviceDid": "did:web:example.com", 113 113 + "token": "test-token", 114 114 + "platform": "ios", 115 115 + "appId": "xyz.bsky.app" 116 116 + })) 117 117 + .send() 118 118 + .await 119 119 + .unwrap(); 120 120 + 121 121 + assert_eq!(res.status(), 401); 122 122 + }

+1 -1

tests/import_verification.rs

··· 23 23 async fn test_import_rejects_car_for_different_user() { 24 24 let client = client(); 25 25 26 26 - let (token_a, did_a) = create_account_and_login(&client).await; 26 26 + let (token_a, _did_a) = create_account_and_login(&client).await; 27 27 let (_token_b, did_b) = create_account_and_login(&client).await; 28 28 29 29 let export_res = client