diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 8a1f7bf..542ad66 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -13,6 +13,21 @@ jobs: test: runs-on: ubuntu-latest + services: + postgres: + image: postgres:16-alpine + env: + POSTGRES_USER: happyview + POSTGRES_PASSWORD: happyview + POSTGRES_DB: happyview_test + ports: + - 5432:5432 + options: >- + --health-cmd pg_isready + --health-interval 10s + --health-timeout 5s + --health-retries 5 + steps: - name: Checkout repository uses: actions/checkout@v4 @@ -31,8 +46,13 @@ jobs: - name: Build run: cargo build --all-targets - - name: Run tests - run: cargo test + - name: Run unit tests + run: cargo test --lib + + - name: Run e2e tests + env: + TEST_DATABASE_URL: postgres://happyview:happyview@localhost:5432/happyview_test + run: cargo test --tests lint: runs-on: ubuntu-latest @@ -56,6 +76,4 @@ jobs: run: cargo fmt -- --check - name: Clippy - env: - SQLX_OFFLINE: true run: cargo clippy --all-targets -- -D warnings diff --git a/src/admin.rs b/src/admin.rs index a5b0568..a9582da 100644 --- a/src/admin.rs +++ b/src/admin.rs @@ -6,9 +6,9 @@ use serde::{Deserialize, Serialize}; use serde_json::Value; use sha2::{Digest, Sha256}; +use crate::AppState; use crate::error::AppError; use crate::lexicon::{LexiconType, ParsedLexicon}; -use crate::AppState; // --------------------------------------------------------------------------- // Helpers @@ -59,34 +59,31 @@ impl axum::extract::FromRequestParts for AdminAuth { // Check admins table first let key_hash = hash_api_key(token); - let found: Option<(String,)> = sqlx::query_as( - "SELECT id::text FROM admins WHERE api_key_hash = $1", - ) - .bind(&key_hash) - .fetch_optional(&state.db) - .await - .map_err(|e| AppError::Internal(format!("admin auth query failed: {e}")))?; + let found: Option<(String,)> = + sqlx::query_as("SELECT id::text FROM admins WHERE api_key_hash = $1") + .bind(&key_hash) + .fetch_optional(&state.db) + .await + .map_err(|e| AppError::Internal(format!("admin auth query failed: {e}")))?; if let Some((admin_id,)) = found { // Update last_used_at in the background let db = state.db.clone(); let admin_id = admin_id.clone(); tokio::spawn(async move { - let _ = sqlx::query( - "UPDATE admins SET last_used_at = NOW() WHERE id::text = $1", - ) - .bind(&admin_id) - .execute(&db) - .await; + let _ = sqlx::query("UPDATE admins SET last_used_at = NOW() WHERE id::text = $1") + .bind(&admin_id) + .execute(&db) + .await; }); return Ok(AdminAuth); } // Fall back to ADMIN_SECRET env var - if let Some(ref secret) = state.config.admin_secret { - if token == secret { - return Ok(AdminAuth); - } + if let Some(ref secret) = state.config.admin_secret + && token == secret + { + return Ok(AdminAuth); } Err(AppError::Auth("invalid admin credentials".into())) @@ -181,7 +178,9 @@ async fn upload_lexicon( .lexicon_json .get("lexicon") .and_then(|v| v.as_i64()) - .ok_or_else(|| AppError::BadRequest("lexicon JSON must have a numeric 'lexicon' field".into()))?; + .ok_or_else(|| { + AppError::BadRequest("lexicon JSON must have a numeric 'lexicon' field".into()) + })?; if lexicon_version != 1 { return Err(AppError::BadRequest(format!( @@ -254,6 +253,7 @@ async fn list_lexicons( State(state): State, _admin: AdminAuth, ) -> Result>, AppError> { + #[allow(clippy::type_complexity)] let rows: Vec<(String, i32, Value, bool, chrono::DateTime, chrono::DateTime)> = sqlx::query_as( "SELECT id, revision, lexicon_json, backfill, created_at, updated_at FROM lexicons ORDER BY id", @@ -289,6 +289,7 @@ async fn get_lexicon( _admin: AdminAuth, Path(id): Path, ) -> Result, AppError> { + #[allow(clippy::type_complexity)] let row: Option<(String, i32, Value, bool, chrono::DateTime, chrono::DateTime)> = sqlx::query_as( "SELECT id, revision, lexicon_json, backfill, created_at, updated_at FROM lexicons WHERE id = $1", @@ -413,6 +414,7 @@ async fn backfill_status( State(state): State, _admin: AdminAuth, ) -> Result>, AppError> { + #[allow(clippy::type_complexity)] let rows: Vec<( String, Option, @@ -435,7 +437,19 @@ async fn backfill_status( let jobs: Vec = rows .into_iter() .map( - |(id, collection, did, status, total_repos, processed_repos, total_records, error, started_at, completed_at, created_at)| { + |( + id, + collection, + did, + status, + total_repos, + processed_repos, + total_records, + error, + started_at, + completed_at, + created_at, + )| { BackfillJob { id, collection, @@ -506,13 +520,18 @@ async fn list_admins( State(state): State, _admin: AdminAuth, ) -> Result>, AppError> { - let rows: Vec<(String, String, chrono::DateTime, Option>)> = - sqlx::query_as( - "SELECT id::text, name, created_at, last_used_at FROM admins ORDER BY created_at", - ) - .fetch_all(&state.db) - .await - .map_err(|e| AppError::Internal(format!("failed to list admins: {e}")))?; + #[allow(clippy::type_complexity)] + let rows: Vec<( + String, + String, + chrono::DateTime, + Option>, + )> = sqlx::query_as( + "SELECT id::text, name, created_at, last_used_at FROM admins ORDER BY created_at", + ) + .fetch_all(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to list admins: {e}")))?; let admins: Vec = rows .into_iter() diff --git a/src/auth/middleware.rs b/src/auth/middleware.rs index 002de17..472574c 100644 --- a/src/auth/middleware.rs +++ b/src/auth/middleware.rs @@ -2,8 +2,8 @@ use axum::extract::FromRequestParts; use axum::http::request::Parts; use serde::Deserialize; -use crate::error::AppError; use crate::AppState; +use crate::error::AppError; /// Authenticated user identity extracted from an AIP-issued access token. #[derive(Debug, Clone)] diff --git a/src/backfill.rs b/src/backfill.rs index 0b1bf45..db2b79e 100644 --- a/src/backfill.rs +++ b/src/backfill.rs @@ -128,7 +128,12 @@ async fn fetch_records( let page_count = body.records.len(); for entry in body.records { - let rkey = entry.uri.split('/').last().unwrap_or_default().to_string(); + let rkey = entry + .uri + .split('/') + .next_back() + .unwrap_or_default() + .to_string(); records.push((entry.uri, rkey, entry.cid, entry.value)); } @@ -154,13 +159,12 @@ async fn run_job( job_id: &str, ) -> Result<(), String> { // Fetch the job - let job: (Option, Option) = sqlx::query_as( - "SELECT collection, did FROM backfill_jobs WHERE id::text = $1", - ) - .bind(job_id) - .fetch_one(db) - .await - .map_err(|e| format!("failed to fetch job: {e}"))?; + let job: (Option, Option) = + sqlx::query_as("SELECT collection, did FROM backfill_jobs WHERE id::text = $1") + .bind(job_id) + .fetch_one(db) + .await + .map_err(|e| format!("failed to fetch job: {e}"))?; let (job_collection, job_did) = job; @@ -279,7 +283,10 @@ async fn run_job( .execute(db) .await; - info!(job = job_id, processed_repos, total_records, "backfill completed"); + info!( + job = job_id, + processed_repos, total_records, "backfill completed" + ); Ok(()) } diff --git a/src/config.rs b/src/config.rs index 583edd3..7bd8dfe 100644 --- a/src/config.rs +++ b/src/config.rs @@ -21,17 +21,13 @@ impl Config { .ok() .and_then(|p| p.parse().ok()) .unwrap_or(3000), - database_url: env::var("DATABASE_URL") - .expect("DATABASE_URL must be set"), - aip_url: env::var("AIP_URL") - .expect("AIP_URL must be set"), + database_url: env::var("DATABASE_URL").expect("DATABASE_URL must be set"), + aip_url: env::var("AIP_URL").expect("AIP_URL must be set"), jetstream_url: env::var("JETSTREAM_URL") .unwrap_or_else(|_| "wss://jetstream2.us-west.bsky.network/subscribe".into()), admin_secret: env::var("ADMIN_SECRET").ok(), - relay_url: env::var("RELAY_URL") - .unwrap_or_else(|_| "https://bsky.network".into()), - plc_url: env::var("PLC_URL") - .unwrap_or_else(|_| "https://plc.directory".into()), + relay_url: env::var("RELAY_URL").unwrap_or_else(|_| "https://bsky.network".into()), + plc_url: env::var("PLC_URL").unwrap_or_else(|_| "https://plc.directory".into()), } } @@ -49,10 +45,18 @@ mod tests { unsafe fn clear_env() { for key in [ - "HOST", "PORT", "DATABASE_URL", "AIP_URL", "JETSTREAM_URL", - "ADMIN_SECRET", "RELAY_URL", "PLC_URL", + "HOST", + "PORT", + "DATABASE_URL", + "AIP_URL", + "JETSTREAM_URL", + "ADMIN_SECRET", + "RELAY_URL", + "PLC_URL", ] { - unsafe { env::remove_var(key); } + unsafe { + env::remove_var(key); + } } } diff --git a/src/error.rs b/src/error.rs index 5b26c6e..319689e 100644 --- a/src/error.rs +++ b/src/error.rs @@ -107,10 +107,7 @@ mod tests { #[test] fn display_formats() { - assert_eq!( - AppError::Auth("x".into()).to_string(), - "auth error: x" - ); + assert_eq!(AppError::Auth("x".into()).to_string(), "auth error: x"); assert_eq!( AppError::BadRequest("y".into()).to_string(), "bad request: y" @@ -119,10 +116,7 @@ mod tests { AppError::Internal("z".into()).to_string(), "internal error: z" ); - assert_eq!( - AppError::NotFound("w".into()).to_string(), - "not found: w" - ); + assert_eq!(AppError::NotFound("w".into()).to_string(), "not found: w"); assert_eq!( AppError::PdsError(StatusCode::BAD_GATEWAY, Bytes::new()).to_string(), "PDS error: 502 Bad Gateway" diff --git a/src/jetstream.rs b/src/jetstream.rs index 131ed91..6d16393 100644 --- a/src/jetstream.rs +++ b/src/jetstream.rs @@ -2,8 +2,8 @@ use futures_util::StreamExt; use serde::Deserialize; use serde_json::Value; use sqlx::PgPool; -use std::sync::atomic::{AtomicI64, Ordering}; use std::sync::Arc; +use std::sync::atomic::{AtomicI64, Ordering}; use tokio::sync::watch; use tokio_tungstenite::tungstenite::Message; @@ -57,7 +57,15 @@ pub fn spawn(db: PgPool, jetstream_url: String, mut collections_rx: watch::Recei // Connect and process events. If the collection list changes // mid-stream, `run` returns so we can reconnect with new filters. - match run(&db, &jetstream_url, &cursor, &collections, &mut collections_rx).await { + match run( + &db, + &jetstream_url, + &cursor, + &collections, + &mut collections_rx, + ) + .await + { Ok(()) => { tracing::info!("jetstream reconnecting due to collection change"); } diff --git a/src/lexicon.rs b/src/lexicon.rs index db81214..971f8df 100644 --- a/src/lexicon.rs +++ b/src/lexicon.rs @@ -43,7 +43,11 @@ pub struct ParsedLexicon { impl ParsedLexicon { /// Parse a lexicon JSON document into a `ParsedLexicon`. - pub fn parse(raw: Value, revision: i32, target_collection: Option) -> Result { + pub fn parse( + raw: Value, + revision: i32, + target_collection: Option, + ) -> Result { let id = raw .get("id") .and_then(|v| v.as_str()) @@ -94,6 +98,12 @@ pub struct LexiconRegistry { inner: Arc>>, } +impl Default for LexiconRegistry { + fn default() -> Self { + Self::new() + } +} + impl LexiconRegistry { pub fn new() -> Self { Self { @@ -277,11 +287,19 @@ mod tests { #[test] fn parse_query_lexicon() { - let parsed = ParsedLexicon::parse(query_lexicon_json(), 2, Some("games.gamesgamesgamesgames.game".into())).unwrap(); + let parsed = ParsedLexicon::parse( + query_lexicon_json(), + 2, + Some("games.gamesgamesgamesgames.game".into()), + ) + .unwrap(); assert_eq!(parsed.lexicon_type, LexiconType::Query); assert!(parsed.parameters.is_some()); assert!(parsed.output.is_some()); - assert_eq!(parsed.target_collection, Some("games.gamesgamesgamesgames.game".into())); + assert_eq!( + parsed.target_collection, + Some("games.gamesgamesgamesgames.game".into()) + ); assert_eq!(parsed.revision, 2); } @@ -316,12 +334,9 @@ mod tests { #[test] fn parse_target_collection_passthrough() { - let parsed = ParsedLexicon::parse( - query_lexicon_json(), - 1, - Some("custom.collection".into()), - ) - .unwrap(); + let parsed = + ParsedLexicon::parse(query_lexicon_json(), 1, Some("custom.collection".into())) + .unwrap(); assert_eq!(parsed.target_collection, Some("custom.collection".into())); } @@ -356,7 +371,13 @@ mod tests { reg.upsert(v2).await; assert_eq!(reg.count().await, 1); - assert_eq!(reg.get("games.gamesgamesgamesgames.game").await.unwrap().revision, 5); + assert_eq!( + reg.get("games.gamesgamesgamesgames.game") + .await + .unwrap() + .revision, + 5 + ); } #[tokio::test] diff --git a/src/main.rs b/src/main.rs index ed5768c..c7c11ab 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,6 +1,6 @@ use happyview::config::Config; use happyview::lexicon::LexiconRegistry; -use happyview::{admin, backfill, jetstream, server, AppState}; +use happyview::{AppState, admin, backfill, jetstream, server}; use tokio::sync::watch; use tracing::info; @@ -48,8 +48,17 @@ async fn main() { collections_tx, }; - jetstream::spawn(state.db.clone(), config.jetstream_url.clone(), collections_rx); - backfill::spawn_worker(state.db.clone(), state.http.clone(), config.relay_url.clone(), config.plc_url.clone()); + jetstream::spawn( + state.db.clone(), + config.jetstream_url.clone(), + collections_rx, + ); + backfill::spawn_worker( + state.db.clone(), + state.http.clone(), + config.relay_url.clone(), + config.plc_url.clone(), + ); let app = server::router(state); let addr = config.listen_addr(); diff --git a/src/profile.rs b/src/profile.rs index 722c28e..b513a74 100644 --- a/src/profile.rs +++ b/src/profile.rs @@ -36,7 +36,11 @@ struct GetRecordResponse { } /// Resolve a full profile for the given DID: DID document -> handle + PDS -> profile record. -pub async fn resolve_profile(http: &reqwest::Client, plc_url: &str, did: &str) -> Result { +pub async fn resolve_profile( + http: &reqwest::Client, + plc_url: &str, + did: &str, +) -> Result { let did_doc = resolve_did_document(http, plc_url, did).await?; let handle = did_doc @@ -52,10 +56,9 @@ pub async fn resolve_profile(http: &reqwest::Client, plc_url: &str, did: &str) - .map(|s| s.service_endpoint.clone()) .ok_or_else(|| AppError::NotFound("no PDS endpoint in DID document".into()))?; - let (display_name, description, avatar_url) = - fetch_profile_from_pds(http, &pds_endpoint, did) - .await - .unwrap_or((None, None, None)); + let (display_name, description, avatar_url) = fetch_profile_from_pds(http, &pds_endpoint, did) + .await + .unwrap_or((None, None, None)); Ok(Profile { did: did.to_string(), @@ -67,7 +70,11 @@ pub async fn resolve_profile(http: &reqwest::Client, plc_url: &str, did: &str) - } /// Resolve the PDS endpoint for a DID by fetching its DID document. -pub async fn resolve_pds_endpoint(http: &reqwest::Client, plc_url: &str, did: &str) -> Result { +pub async fn resolve_pds_endpoint( + http: &reqwest::Client, + plc_url: &str, + did: &str, +) -> Result { let did_doc = resolve_did_document(http, plc_url, did).await?; did_doc @@ -80,7 +87,11 @@ pub async fn resolve_pds_endpoint(http: &reqwest::Client, plc_url: &str, did: &s /// Fetch a DID document from the PLC directory. // TODO: handle did:web:* resolution (fetch https://{domain}/.well-known/did.json) -async fn resolve_did_document(http: &reqwest::Client, plc_url: &str, did: &str) -> Result { +async fn resolve_did_document( + http: &reqwest::Client, + plc_url: &str, + did: &str, +) -> Result { let url = format!("{}/{did}", plc_url.trim_end_matches('/')); let resp = http diff --git a/src/repo.rs b/src/repo.rs index 6952134..0ccff2e 100644 --- a/src/repo.rs +++ b/src/repo.rs @@ -5,12 +5,12 @@ use axum::response::{IntoResponse, Response}; use base64::Engine; use p256::pkcs8::EncodePrivateKey; use serde::Deserialize; -use serde_json::{json, Value}; +use serde_json::{Value, json}; use sha2::{Digest, Sha256}; +use crate::AppState; use crate::auth::Claims; use crate::error::AppError; -use crate::AppState; // --------------------------------------------------------------------------- // AT URI parsing @@ -110,8 +110,7 @@ pub(crate) fn generate_dpop_proof( .to_pkcs8_der() .map_err(|e| AppError::Internal(format!("PKCS#8 conversion failed: {e}")))?; - let encoding_key = - jsonwebtoken::EncodingKey::from_ec_der(pkcs8_der.as_bytes()); + let encoding_key = jsonwebtoken::EncodingKey::from_ec_der(pkcs8_der.as_bytes()); // Public JWK for the header (no private component) let public_jwk = jsonwebtoken::jwk::Jwk { @@ -212,35 +211,34 @@ pub(crate) async fn pds_post_json_raw( .map_err(|e| AppError::Internal(format!("PDS request failed: {e}")))?; // Retry with nonce if PDS requires it - if resp.status() == reqwest::StatusCode::UNAUTHORIZED { - if let Some(nonce) = resp + if resp.status() == reqwest::StatusCode::UNAUTHORIZED + && let Some(nonce) = resp .headers() .get("dpop-nonce") .and_then(|v| v.to_str().ok()) - { - let nonce = nonce.to_string(); - tracing::debug!("retrying with DPoP nonce"); - - let dpop = generate_dpop_proof( - "POST", - &url, - &session.dpop_jwk, - &session.access_token, - Some(&nonce), - )?; - - let resp = state - .http - .post(&url) - .header("authorization", format!("DPoP {}", session.access_token)) - .header("dpop", &dpop) - .json(body) - .send() - .await - .map_err(|e| AppError::Internal(format!("PDS request retry failed: {e}")))?; - - return Ok(resp); - } + { + let nonce = nonce.to_string(); + tracing::debug!("retrying with DPoP nonce"); + + let dpop = generate_dpop_proof( + "POST", + &url, + &session.dpop_jwk, + &session.access_token, + Some(&nonce), + )?; + + let resp = state + .http + .post(&url) + .header("authorization", format!("DPoP {}", session.access_token)) + .header("dpop", &dpop) + .json(body) + .send() + .await + .map_err(|e| AppError::Internal(format!("PDS request retry failed: {e}")))?; + + return Ok(resp); } Ok(resp) @@ -271,36 +269,35 @@ async fn pds_post_blob( .await .map_err(|e| AppError::Internal(format!("PDS uploadBlob failed: {e}")))?; - if resp.status() == reqwest::StatusCode::UNAUTHORIZED { - if let Some(nonce) = resp + if resp.status() == reqwest::StatusCode::UNAUTHORIZED + && let Some(nonce) = resp .headers() .get("dpop-nonce") .and_then(|v| v.to_str().ok()) - { - let nonce = nonce.to_string(); - tracing::debug!("retrying uploadBlob with DPoP nonce"); - - let dpop = generate_dpop_proof( - "POST", - &url, - &session.dpop_jwk, - &session.access_token, - Some(&nonce), - )?; - - let resp = state - .http - .post(&url) - .header("authorization", format!("DPoP {}", session.access_token)) - .header("dpop", &dpop) - .header("content-type", content_type) - .body(blob) - .send() - .await - .map_err(|e| AppError::Internal(format!("PDS uploadBlob retry failed: {e}")))?; - - return forward_pds_response(resp).await; - } + { + let nonce = nonce.to_string(); + tracing::debug!("retrying uploadBlob with DPoP nonce"); + + let dpop = generate_dpop_proof( + "POST", + &url, + &session.dpop_jwk, + &session.access_token, + Some(&nonce), + )?; + + let resp = state + .http + .post(&url) + .header("authorization", format!("DPoP {}", session.access_token)) + .header("dpop", &dpop) + .header("content-type", content_type) + .body(blob) + .send() + .await + .map_err(|e| AppError::Internal(format!("PDS uploadBlob retry failed: {e}")))?; + + return forward_pds_response(resp).await; } forward_pds_response(resp).await @@ -344,17 +341,16 @@ pub(crate) fn enrich_media_blobs(record: &mut Value, pds: &str, did: &str) { .and_then(|l| l.as_str()) .map(|s| s.to_string()); - if let Some(cid) = cid { - if let Some(blob) = item.get_mut("blob") { - if let Some(obj) = blob.as_object_mut() { - obj.insert( - "url".to_string(), - json!(format!( - "{pds_base}/xrpc/com.atproto.sync.getBlob?did={did}&cid={cid}" - )), - ); - } - } + if let Some(cid) = cid + && let Some(blob) = item.get_mut("blob") + && let Some(obj) = blob.as_object_mut() + { + obj.insert( + "url".to_string(), + json!(format!( + "{pds_base}/xrpc/com.atproto.sync.getBlob?did={did}&cid={cid}" + )), + ); } } } @@ -480,19 +476,23 @@ mod tests { let point = public.to_encoded_point(false); DpopJwk { - x: base64::engine::general_purpose::URL_SAFE_NO_PAD - .encode(point.x().unwrap()), - y: base64::engine::general_purpose::URL_SAFE_NO_PAD - .encode(point.y().unwrap()), - d: base64::engine::general_purpose::URL_SAFE_NO_PAD - .encode(secret.to_bytes()), + x: base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(point.x().unwrap()), + y: base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(point.y().unwrap()), + d: base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(secret.to_bytes()), } } #[test] fn dpop_proof_produces_valid_jwt_structure() { let jwk = test_dpop_jwk(); - let token = generate_dpop_proof("POST", "https://pds.example.com/xrpc/test", &jwk, "access-tok", None).unwrap(); + let token = generate_dpop_proof( + "POST", + "https://pds.example.com/xrpc/test", + &jwk, + "access-tok", + None, + ) + .unwrap(); let parts: Vec<&str> = token.split('.').collect(); assert_eq!(parts.len(), 3, "JWT should have 3 parts"); @@ -501,7 +501,14 @@ mod tests { #[test] fn dpop_proof_header_has_correct_fields() { let jwk = test_dpop_jwk(); - let token = generate_dpop_proof("POST", "https://pds.example.com/xrpc/test", &jwk, "access-tok", None).unwrap(); + let token = generate_dpop_proof( + "POST", + "https://pds.example.com/xrpc/test", + &jwk, + "access-tok", + None, + ) + .unwrap(); let header_b64 = token.split('.').next().unwrap(); let header_bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD @@ -517,7 +524,14 @@ mod tests { #[test] fn dpop_proof_claims_have_correct_fields() { let jwk = test_dpop_jwk(); - let token = generate_dpop_proof("GET", "https://pds.example.com/xrpc/test", &jwk, "my-access-token", None).unwrap(); + let token = generate_dpop_proof( + "GET", + "https://pds.example.com/xrpc/test", + &jwk, + "my-access-token", + None, + ) + .unwrap(); let payload_b64 = token.split('.').nth(1).unwrap(); let payload_bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD @@ -537,7 +551,14 @@ mod tests { #[test] fn dpop_proof_includes_nonce_when_provided() { let jwk = test_dpop_jwk(); - let token = generate_dpop_proof("POST", "https://pds.example.com/xrpc/test", &jwk, "tok", Some("abc123")).unwrap(); + let token = generate_dpop_proof( + "POST", + "https://pds.example.com/xrpc/test", + &jwk, + "tok", + Some("abc123"), + ) + .unwrap(); let payload_b64 = token.split('.').nth(1).unwrap(); let payload_bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD @@ -552,7 +573,8 @@ mod tests { fn dpop_proof_ath_is_sha256_of_access_token() { let jwk = test_dpop_jwk(); let access_token = "test-access-token"; - let token = generate_dpop_proof("POST", "https://example.com", &jwk, access_token, None).unwrap(); + let token = + generate_dpop_proof("POST", "https://example.com", &jwk, access_token, None).unwrap(); let payload_b64 = token.split('.').nth(1).unwrap(); let payload_bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD diff --git a/src/server.rs b/src/server.rs index 70339b6..ec2cf5a 100644 --- a/src/server.rs +++ b/src/server.rs @@ -4,13 +4,13 @@ use axum::{Json, Router}; use tower_http::cors::CorsLayer; use tower_http::trace::TraceLayer; +use crate::AppState; use crate::admin; use crate::auth::Claims; use crate::error::AppError; use crate::profile; use crate::repo; use crate::xrpc; -use crate::AppState; pub fn router(state: AppState) -> Router { Router::new() @@ -36,6 +36,7 @@ async fn get_profile( State(state): State, claims: Claims, ) -> Result, AppError> { - let profile = profile::resolve_profile(&state.http, &state.config.plc_url, claims.did()).await?; + let profile = + profile::resolve_profile(&state.http, &state.config.plc_url, claims.did()).await?; Ok(Json(profile)) } diff --git a/src/xrpc.rs b/src/xrpc.rs index 2cb6f41..cb6ca58 100644 --- a/src/xrpc.rs +++ b/src/xrpc.rs @@ -1,16 +1,16 @@ +use axum::Json; use axum::extract::{Path, Query, State}; use axum::http::StatusCode; use axum::response::{IntoResponse, Response}; -use axum::Json; -use serde_json::{json, Value}; +use serde_json::{Value, json}; use std::collections::{HashMap, HashSet}; +use crate::AppState; use crate::auth::Claims; use crate::error::AppError; use crate::lexicon::LexiconType; use crate::profile; use crate::repo; -use crate::AppState; // --------------------------------------------------------------------------- // Catch-all handler @@ -75,14 +75,11 @@ async fn handle_query( } // List query: needs a target collection to know what to query - let collection = lexicon - .target_collection - .as_deref() - .ok_or_else(|| { - AppError::BadRequest(format!( - "{method} has no target_collection configured for list queries" - )) - })?; + let collection = lexicon.target_collection.as_deref().ok_or_else(|| { + AppError::BadRequest(format!( + "{method} has no target_collection configured for list queries" + )) + })?; let limit: i64 = params .get("limit") @@ -126,7 +123,9 @@ async fn handle_query( let unique_dids: HashSet<&str> = rows.iter().map(|(_, did, _)| did.as_str()).collect(); let mut pds_map: HashMap = HashMap::new(); for did in unique_dids { - if let Ok(pds) = profile::resolve_pds_endpoint(&state.http, &state.config.plc_url, did).await { + if let Ok(pds) = + profile::resolve_pds_endpoint(&state.http, &state.config.plc_url, did).await + { pds_map.insert(did.to_string(), pds); } } @@ -159,15 +158,13 @@ async fn handle_query( async fn handle_get_record(state: &AppState, uri: &str) -> Result { let did = repo::parse_did_from_at_uri(uri)?; - let row: Option<(Value,)> = - sqlx::query_as("SELECT record FROM records WHERE uri = $1") - .bind(uri) - .fetch_optional(&state.db) - .await - .map_err(|e| AppError::Internal(format!("DB query failed: {e}")))?; + let row: Option<(Value,)> = sqlx::query_as("SELECT record FROM records WHERE uri = $1") + .bind(uri) + .fetch_optional(&state.db) + .await + .map_err(|e| AppError::Internal(format!("DB query failed: {e}")))?; - let (mut record,) = - row.ok_or_else(|| AppError::NotFound("record not found".into()))?; + let (mut record,) = row.ok_or_else(|| AppError::NotFound("record not found".into()))?; let pds = profile::resolve_pds_endpoint(&state.http, &state.config.plc_url, &did).await?; repo::enrich_media_blobs(&mut record, &pds, &did); @@ -190,14 +187,9 @@ async fn handle_procedure( input: &Value, lexicon: &crate::lexicon::ParsedLexicon, ) -> Result { - let collection = lexicon - .target_collection - .as_deref() - .ok_or_else(|| { - AppError::BadRequest(format!( - "{method} has no target_collection configured" - )) - })?; + let collection = lexicon.target_collection.as_deref().ok_or_else(|| { + AppError::BadRequest(format!("{method} has no target_collection configured")) + })?; let session = repo::get_atp_session(state, claims.token()).await?; @@ -232,7 +224,8 @@ async fn handle_create_record( "record": record, }); - let resp = repo::pds_post_json_raw(state, session, "com.atproto.repo.createRecord", &pds_body).await?; + let resp = + repo::pds_post_json_raw(state, session, "com.atproto.repo.createRecord", &pds_body).await?; if resp.status().is_success() { let bytes = resp @@ -247,7 +240,7 @@ async fn handle_create_record( pds_result.get("uri").and_then(|v| v.as_str()), pds_result.get("cid").and_then(|v| v.as_str()), ) { - let rkey = uri.split('/').last().unwrap_or_default(); + let rkey = uri.split('/').next_back().unwrap_or_default(); let _ = sqlx::query( r#" INSERT INTO records (uri, did, collection, rkey, record, cid) @@ -292,7 +285,7 @@ async fn handle_put_record( let rkey = uri .split('/') - .last() + .next_back() .ok_or_else(|| AppError::Internal("invalid AT URI".into()))?; // Build record from input, adding $type @@ -310,7 +303,8 @@ async fn handle_put_record( "record": record, }); - let resp = repo::pds_post_json_raw(state, session, "com.atproto.repo.putRecord", &pds_body).await?; + let resp = + repo::pds_post_json_raw(state, session, "com.atproto.repo.putRecord", &pds_body).await?; if resp.status().is_success() { let bytes = resp diff --git a/tests/common/app.rs b/tests/common/app.rs index ba7fff3..1bba7b0 100644 --- a/tests/common/app.rs +++ b/tests/common/app.rs @@ -1,7 +1,7 @@ use axum::Router; use happyview::config::Config; use happyview::lexicon::LexiconRegistry; -use happyview::{admin, server, AppState}; +use happyview::{AppState, admin, server}; use tokio::sync::watch; use wiremock::MockServer; diff --git a/tests/common/auth.rs b/tests/common/auth.rs index 3c9a799..eb9f918 100644 --- a/tests/common/auth.rs +++ b/tests/common/auth.rs @@ -17,9 +17,7 @@ pub fn admin_auth_header(token: &str) -> (HeaderName, HeaderValue) { pub async fn mock_aip_userinfo(mock_server: &MockServer, did: &str) { Mock::given(method("GET")) .and(path("/oauth/userinfo")) - .respond_with( - ResponseTemplate::new(200).set_body_json(fixtures::userinfo_response(did)), - ) + .respond_with(ResponseTemplate::new(200).set_body_json(fixtures::userinfo_response(did))) .mount(mock_server) .await; } diff --git a/tests/common/db.rs b/tests/common/db.rs index 92e582e..b5db95f 100644 --- a/tests/common/db.rs +++ b/tests/common/db.rs @@ -2,8 +2,8 @@ use sqlx::PgPool; /// Connect to the test database using `TEST_DATABASE_URL`. pub async fn test_pool() -> PgPool { - let url = std::env::var("TEST_DATABASE_URL") - .expect("TEST_DATABASE_URL must be set for e2e tests"); + let url = + std::env::var("TEST_DATABASE_URL").expect("TEST_DATABASE_URL must be set for e2e tests"); let pool = PgPool::connect(&url) .await diff --git a/tests/common/fixtures.rs b/tests/common/fixtures.rs index e23c420..c7321e6 100644 --- a/tests/common/fixtures.rs +++ b/tests/common/fixtures.rs @@ -1,4 +1,4 @@ -use serde_json::{json, Value}; +use serde_json::{Value, json}; /// A minimal record-type lexicon JSON for testing. pub fn game_record_lexicon() -> Value { diff --git a/tests/e2e_admin.rs b/tests/e2e_admin.rs index f13486a..0c5eee5 100644 --- a/tests/e2e_admin.rs +++ b/tests/e2e_admin.rs @@ -3,7 +3,7 @@ mod common; use axum::body::Body; use axum::http::{Request, StatusCode}; use http_body_util::BodyExt; -use serde_json::{json, Value}; +use serde_json::{Value, json}; use serial_test::serial; use tower::ServiceExt; @@ -391,11 +391,7 @@ async fn backfill_list_jobs() { // Create a job first app.router .clone() - .oneshot(admin_post( - "/admin/backfill", - &app.admin_secret, - &json!({}), - )) + .oneshot(admin_post("/admin/backfill", &app.admin_secret, &json!({}))) .await .unwrap(); diff --git a/tests/e2e_xrpc.rs b/tests/e2e_xrpc.rs index f92e6b7..66f9df4 100644 --- a/tests/e2e_xrpc.rs +++ b/tests/e2e_xrpc.rs @@ -3,7 +3,7 @@ mod common; use axum::body::Body; use axum::http::{Request, StatusCode}; use http_body_util::BodyExt; -use serde_json::{json, Value}; +use serde_json::{Value, json}; use serial_test::serial; use tower::ServiceExt; use wiremock::matchers::{method, path, path_regex}; @@ -88,7 +88,7 @@ async fn seed_lexicons(app: &TestApp) { /// Seed a record directly into the database. async fn seed_record(app: &TestApp, uri: &str, did: &str, collection: &str, record: &Value) { - let rkey = uri.split('/').last().unwrap_or("1"); + let rkey = uri.split('/').next_back().unwrap_or("1"); sqlx::query( "INSERT INTO records (uri, did, collection, rkey, record, cid) VALUES ($1, $2, $3, $4, $5, $6)", ) @@ -138,27 +138,23 @@ async fn profile_with_mocked_services_returns_200() { // Mock PLC directory Mock::given(method("GET")) .and(path(format!("/{did}"))) - .respond_with(ResponseTemplate::new(200).set_body_json( - fixtures::did_document(did, &app.mock_server.uri()), - )) + .respond_with( + ResponseTemplate::new(200) + .set_body_json(fixtures::did_document(did, &app.mock_server.uri())), + ) .mount(&app.mock_server) .await; // Mock PDS getRecord for profile Mock::given(method("GET")) .and(path("/xrpc/com.atproto.repo.getRecord")) - .respond_with( - ResponseTemplate::new(200).set_body_json(fixtures::profile_record()), - ) + .respond_with(ResponseTemplate::new(200).set_body_json(fixtures::profile_record())) .mount(&app.mock_server) .await; let resp = app .router - .oneshot(authed_get( - "/xrpc/app.bsky.actor.getProfile", - "valid-token", - )) + .oneshot(authed_get("/xrpc/app.bsky.actor.getProfile", "valid-token")) .await .unwrap(); @@ -226,9 +222,10 @@ async fn xrpc_get_single_record_by_uri() { // Mock PLC for PDS resolution Mock::given(method("GET")) .and(path(format!("/{did}"))) - .respond_with(ResponseTemplate::new(200).set_body_json( - fixtures::did_document(did, "https://pds.example.com"), - )) + .respond_with( + ResponseTemplate::new(200) + .set_body_json(fixtures::did_document(did, "https://pds.example.com")), + ) .mount(&app.mock_server) .await; @@ -236,7 +233,7 @@ async fn xrpc_get_single_record_by_uri() { .router .oneshot( Request::builder() - .uri(&format!( + .uri(format!( "/xrpc/games.gamesgamesgamesgames.listGames?uri={}", urlencoding::encode(uri) )) @@ -283,9 +280,10 @@ async fn xrpc_get_list_with_pagination() { // Mock PLC for enrichment Mock::given(method("GET")) .and(path_regex("/did:plc:.*")) - .respond_with(ResponseTemplate::new(200).set_body_json( - fixtures::did_document(did, "https://pds.example.com"), - )) + .respond_with( + ResponseTemplate::new(200) + .set_body_json(fixtures::did_document(did, "https://pds.example.com")), + ) .mount(&app.mock_server) .await; @@ -326,7 +324,7 @@ async fn xrpc_get_list_with_pagination() { .router .oneshot( Request::builder() - .uri(&format!( + .uri(format!( "/xrpc/games.gamesgamesgamesgames.listGames?limit=2&cursor={cursor}" )) .body(Body::empty()) @@ -350,9 +348,12 @@ async fn xrpc_get_list_filtered_by_did() { // Mock PLC Mock::given(method("GET")) .and(path_regex("/did:plc:.*")) - .respond_with(ResponseTemplate::new(200).set_body_json( - fixtures::did_document("did:plc:a", "https://pds.example.com"), - )) + .respond_with( + ResponseTemplate::new(200).set_body_json(fixtures::did_document( + "did:plc:a", + "https://pds.example.com", + )), + ) .mount(&app.mock_server) .await;