From 2c0833ea4c2d44b8da35cf70b8ac12e2fe094915 Mon Sep 17 00:00:00 2001 From: Trezy Date: Fri, 13 Feb 2026 12:13:46 -0600 Subject: [PATCH] refactor!: split admin, repo, and xrpc into directory modules --- src/admin.rs | 597 --------------------------------------- src/admin/admins.rs | 89 ++++++ src/admin/auth.rs | 59 ++++ src/admin/backfill.rs | 95 +++++++ src/admin/bootstrap.rs | 25 ++ src/admin/hash.rs | 38 +++ src/admin/lexicons.rs | 185 +++++++++++++ src/admin/mod.rs | 26 ++ src/admin/stats.rs | 34 +++ src/admin/types.rs | 86 ++++++ src/repo.rs | 600 ---------------------------------------- src/repo/at_uri.rs | 43 +++ src/repo/dpop.rs | 222 +++++++++++++++ src/repo/media.rs | 110 ++++++++ src/repo/mod.rs | 12 + src/repo/pds.rs | 152 ++++++++++ src/repo/session.rs | 45 +++ src/repo/upload_blob.rs | 27 ++ src/xrpc.rs | 351 ----------------------- src/xrpc/mod.rs | 55 ++++ src/xrpc/procedure.rs | 178 ++++++++++++ src/xrpc/query.rs | 122 ++++++++ 22 files changed, 1603 insertions(+), 1548 deletions(-) delete mode 100644 src/admin.rs create mode 100644 src/admin/admins.rs create mode 100644 src/admin/auth.rs create mode 100644 src/admin/backfill.rs create mode 100644 src/admin/bootstrap.rs create mode 100644 src/admin/hash.rs create mode 100644 src/admin/lexicons.rs create mode 100644 src/admin/mod.rs create mode 100644 src/admin/stats.rs create mode 100644 src/admin/types.rs delete mode 100644 src/repo.rs create mode 100644 src/repo/at_uri.rs create mode 100644 src/repo/dpop.rs create mode 100644 src/repo/media.rs create mode 100644 src/repo/mod.rs create mode 100644 src/repo/pds.rs create mode 100644 src/repo/session.rs create mode 100644 src/repo/upload_blob.rs delete mode 100644 src/xrpc.rs create mode 100644 src/xrpc/mod.rs create mode 100644 src/xrpc/procedure.rs create mode 100644 src/xrpc/query.rs diff --git a/src/admin.rs b/src/admin.rs deleted file mode 100644 index a9582da..0000000 --- a/src/admin.rs +++ /dev/null @@ -1,597 +0,0 @@ -use axum::extract::{Path, State}; -use axum::http::StatusCode; -use axum::routing::{delete, get, post}; -use axum::{Json, Router}; -use serde::{Deserialize, Serialize}; -use serde_json::Value; -use sha2::{Digest, Sha256}; - -use crate::AppState; -use crate::error::AppError; -use crate::lexicon::{LexiconType, ParsedLexicon}; - -// --------------------------------------------------------------------------- -// Helpers -// --------------------------------------------------------------------------- - -/// SHA-256 hash a plaintext API key for storage/comparison. -pub(crate) fn hash_api_key(key: &str) -> String { - let hash = Sha256::digest(key.as_bytes()); - hex::encode(hash) -} - -// --------------------------------------------------------------------------- -// Admin auth middleware -// --------------------------------------------------------------------------- - -pub fn admin_routes(_state: AppState) -> Router { - Router::new() - .route("/lexicons", post(upload_lexicon).get(list_lexicons)) - .route("/lexicons/{id}", get(get_lexicon).delete(delete_lexicon)) - .route("/stats", get(stats)) - .route("/backfill", post(create_backfill)) - .route("/backfill/status", get(backfill_status)) - .route("/admins", post(create_admin).get(list_admins)) - .route("/admins/{id}", delete(delete_admin)) -} - -/// Axum extractor for admin auth. Checks the Bearer token against: -/// 1. The `admins` table (hashed key lookup) -/// 2. Falls back to `ADMIN_SECRET` env var for bootstrap -pub struct AdminAuth; - -impl axum::extract::FromRequestParts for AdminAuth { - type Rejection = AppError; - - async fn from_request_parts( - parts: &mut axum::http::request::Parts, - state: &AppState, - ) -> Result { - let header = parts - .headers - .get("authorization") - .and_then(|v| v.to_str().ok()) - .ok_or_else(|| AppError::Auth("missing Authorization header".into()))?; - - let token = header - .strip_prefix("Bearer ") - .ok_or_else(|| AppError::Auth("invalid Authorization scheme".into()))?; - - // 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}")))?; - - 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; - }); - return Ok(AdminAuth); - } - - // Fall back to ADMIN_SECRET env var - if let Some(ref secret) = state.config.admin_secret - && token == secret - { - return Ok(AdminAuth); - } - - Err(AppError::Auth("invalid admin credentials".into())) - } -} - -/// Bootstrap: if no admins exist and ADMIN_SECRET is set, create a bootstrap admin. -pub async fn bootstrap(db: &sqlx::PgPool, admin_secret: &Option) { - let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM admins") - .fetch_one(db) - .await - .unwrap_or((0,)); - - if count.0 > 0 { - return; - } - - if let Some(secret) = admin_secret { - let key_hash = hash_api_key(secret); - let _ = sqlx::query( - "INSERT INTO admins (name, api_key_hash) VALUES ($1, $2) ON CONFLICT DO NOTHING", - ) - .bind("bootstrap") - .bind(&key_hash) - .execute(db) - .await; - tracing::info!("created bootstrap admin from ADMIN_SECRET"); - } -} - -// --------------------------------------------------------------------------- -// Jetstream notification -// --------------------------------------------------------------------------- - -/// Send the current record collection list to the Jetstream task so it -/// reconnects with the updated filter. -async fn notify_jetstream(state: &AppState) { - let collections = state.lexicons.get_record_collections().await; - let _ = state.collections_tx.send(collections); -} - -// --------------------------------------------------------------------------- -// Request / response types -// --------------------------------------------------------------------------- - -#[derive(Serialize)] -struct LexiconSummary { - id: String, - revision: i32, - lexicon_type: String, - backfill: bool, - created_at: chrono::DateTime, - updated_at: chrono::DateTime, -} - -#[derive(Deserialize)] -struct UploadLexiconBody { - lexicon_json: Value, - #[serde(default = "default_backfill")] - backfill: bool, - target_collection: Option, -} - -fn default_backfill() -> bool { - true -} - -#[derive(Serialize)] -struct StatsResponse { - total_records: i64, - collections: Vec, -} - -#[derive(Serialize)] -struct CollectionStat { - collection: String, - count: i64, -} - -// --------------------------------------------------------------------------- -// Handlers -// --------------------------------------------------------------------------- - -/// POST /admin/lexicons — upload (upsert) a lexicon. -async fn upload_lexicon( - State(state): State, - _admin: AdminAuth, - Json(body): Json, -) -> Result<(StatusCode, Json), AppError> { - // Validate basic structure - let lexicon_version = body - .lexicon_json - .get("lexicon") - .and_then(|v| v.as_i64()) - .ok_or_else(|| { - AppError::BadRequest("lexicon JSON must have a numeric 'lexicon' field".into()) - })?; - - if lexicon_version != 1 { - return Err(AppError::BadRequest(format!( - "unsupported lexicon version: {lexicon_version}" - ))); - } - - let id = body - .lexicon_json - .get("id") - .and_then(|v| v.as_str()) - .ok_or_else(|| AppError::BadRequest("lexicon JSON must have a string 'id' field".into()))? - .to_string(); - - // Validate it parses correctly - ParsedLexicon::parse(body.lexicon_json.clone(), 1, body.target_collection.clone()) - .map_err(|e| AppError::BadRequest(format!("failed to parse lexicon: {e}")))?; - - // Upsert into database - let row: (i32,) = sqlx::query_as( - r#" - INSERT INTO lexicons (id, lexicon_json, backfill, target_collection) - VALUES ($1, $2, $3, $4) - ON CONFLICT (id) DO UPDATE SET - lexicon_json = EXCLUDED.lexicon_json, - backfill = EXCLUDED.backfill, - target_collection = EXCLUDED.target_collection, - revision = lexicons.revision + 1, - updated_at = NOW() - RETURNING revision - "#, - ) - .bind(&id) - .bind(&body.lexicon_json) - .bind(body.backfill) - .bind(&body.target_collection) - .fetch_one(&state.db) - .await - .map_err(|e| AppError::Internal(format!("failed to upsert lexicon: {e}")))?; - - let revision = row.0; - - // Update in-memory registry with correct revision - let parsed = ParsedLexicon::parse(body.lexicon_json, revision, body.target_collection) - .map_err(|e| AppError::Internal(format!("failed to re-parse lexicon: {e}")))?; - let is_record = parsed.lexicon_type == LexiconType::Record; - state.lexicons.upsert(parsed).await; - - if is_record { - notify_jetstream(&state).await; - } - - let status = if revision == 1 { - StatusCode::CREATED - } else { - StatusCode::OK - }; - - Ok(( - status, - Json(serde_json::json!({ - "id": id, - "revision": revision, - })), - )) -} - -/// GET /admin/lexicons — list all lexicons. -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", - ) - .fetch_all(&state.db) - .await - .map_err(|e| AppError::Internal(format!("failed to list lexicons: {e}")))?; - - let summaries: Vec = rows - .into_iter() - .map(|(id, revision, json, backfill, created_at, updated_at)| { - let lexicon_type = ParsedLexicon::parse(json, revision, None) - .map(|p| format!("{:?}", p.lexicon_type).to_lowercase()) - .unwrap_or_else(|_| "unknown".into()); - - LexiconSummary { - id, - revision, - lexicon_type, - backfill, - created_at, - updated_at, - } - }) - .collect(); - - Ok(Json(summaries)) -} - -/// GET /admin/lexicons/:id — get a single lexicon. -async fn get_lexicon( - State(state): State, - _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", - ) - .bind(&id) - .fetch_optional(&state.db) - .await - .map_err(|e| AppError::Internal(format!("failed to get lexicon: {e}")))?; - - let (id, revision, lexicon_json, backfill, created_at, updated_at) = - row.ok_or_else(|| AppError::NotFound(format!("lexicon '{id}' not found")))?; - - Ok(Json(serde_json::json!({ - "id": id, - "revision": revision, - "lexicon_json": lexicon_json, - "backfill": backfill, - "created_at": created_at, - "updated_at": updated_at, - }))) -} - -/// DELETE /admin/lexicons/:id — remove a lexicon. -async fn delete_lexicon( - State(state): State, - _admin: AdminAuth, - Path(id): Path, -) -> Result { - let result = sqlx::query("DELETE FROM lexicons WHERE id = $1") - .bind(&id) - .execute(&state.db) - .await - .map_err(|e| AppError::Internal(format!("failed to delete lexicon: {e}")))?; - - if result.rows_affected() == 0 { - return Err(AppError::NotFound(format!("lexicon '{id}' not found"))); - } - - state.lexicons.remove(&id).await; - notify_jetstream(&state).await; - - Ok(StatusCode::NO_CONTENT) -} - -/// GET /admin/stats — system statistics. -async fn stats( - State(state): State, - _admin: AdminAuth, -) -> Result, AppError> { - let total: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM records") - .fetch_one(&state.db) - .await - .map_err(|e| AppError::Internal(format!("failed to count records: {e}")))?; - - let collections: Vec<(String, i64)> = sqlx::query_as( - "SELECT collection, COUNT(*) FROM records GROUP BY collection ORDER BY collection", - ) - .fetch_all(&state.db) - .await - .map_err(|e| AppError::Internal(format!("failed to count by collection: {e}")))?; - - Ok(Json(StatsResponse { - total_records: total.0, - collections: collections - .into_iter() - .map(|(collection, count)| CollectionStat { collection, count }) - .collect(), - })) -} - -// --------------------------------------------------------------------------- -// Backfill endpoints -// --------------------------------------------------------------------------- - -#[derive(Deserialize)] -struct CreateBackfillBody { - collection: Option, - did: Option, -} - -#[derive(Serialize)] -struct BackfillJob { - id: String, - collection: Option, - did: Option, - status: String, - total_repos: Option, - processed_repos: Option, - total_records: Option, - error: Option, - started_at: Option>, - completed_at: Option>, - created_at: chrono::DateTime, -} - -/// POST /admin/backfill — create a new backfill job. -async fn create_backfill( - State(state): State, - _admin: AdminAuth, - Json(body): Json, -) -> Result<(StatusCode, Json), AppError> { - let row: (String,) = sqlx::query_as( - "INSERT INTO backfill_jobs (collection, did) VALUES ($1, $2) RETURNING id::text", - ) - .bind(&body.collection) - .bind(&body.did) - .fetch_one(&state.db) - .await - .map_err(|e| AppError::Internal(format!("failed to create backfill job: {e}")))?; - - Ok(( - StatusCode::CREATED, - Json(serde_json::json!({ - "id": row.0, - "status": "pending", - })), - )) -} - -/// GET /admin/backfill/status — list all backfill jobs. -async fn backfill_status( - State(state): State, - _admin: AdminAuth, -) -> Result>, AppError> { - #[allow(clippy::type_complexity)] - let rows: Vec<( - String, - Option, - Option, - String, - Option, - Option, - Option, - Option, - Option>, - Option>, - chrono::DateTime, - )> = sqlx::query_as( - "SELECT id::text, collection, did, status, total_repos, processed_repos, total_records, error, started_at, completed_at, created_at FROM backfill_jobs ORDER BY created_at DESC", - ) - .fetch_all(&state.db) - .await - .map_err(|e| AppError::Internal(format!("failed to list backfill jobs: {e}")))?; - - let jobs: Vec = rows - .into_iter() - .map( - |( - id, - collection, - did, - status, - total_repos, - processed_repos, - total_records, - error, - started_at, - completed_at, - created_at, - )| { - BackfillJob { - id, - collection, - did, - status, - total_repos, - processed_repos, - total_records, - error, - started_at, - completed_at, - created_at, - } - }, - ) - .collect(); - - Ok(Json(jobs)) -} - -// --------------------------------------------------------------------------- -// Admin management endpoints -// --------------------------------------------------------------------------- - -#[derive(Deserialize)] -struct CreateAdminBody { - name: String, -} - -#[derive(Serialize)] -struct AdminSummary { - id: String, - name: String, - created_at: chrono::DateTime, - last_used_at: Option>, -} - -/// POST /admin/admins — create a new admin. Returns the API key once. -async fn create_admin( - State(state): State, - _admin: AdminAuth, - Json(body): Json, -) -> Result<(StatusCode, Json), AppError> { - let api_key = uuid::Uuid::new_v4().to_string(); - let key_hash = hash_api_key(&api_key); - - let row: (String,) = sqlx::query_as( - "INSERT INTO admins (name, api_key_hash) VALUES ($1, $2) RETURNING id::text", - ) - .bind(&body.name) - .bind(&key_hash) - .fetch_one(&state.db) - .await - .map_err(|e| AppError::Internal(format!("failed to create admin: {e}")))?; - - Ok(( - StatusCode::CREATED, - Json(serde_json::json!({ - "id": row.0, - "name": body.name, - "api_key": api_key, - })), - )) -} - -/// GET /admin/admins — list all admins (without keys). -async fn list_admins( - State(state): State, - _admin: AdminAuth, -) -> Result>, AppError> { - #[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() - .map(|(id, name, created_at, last_used_at)| AdminSummary { - id, - name, - created_at, - last_used_at, - }) - .collect(); - - Ok(Json(admins)) -} - -/// DELETE /admin/admins/:id — remove an admin. -async fn delete_admin( - State(state): State, - _admin: AdminAuth, - Path(id): Path, -) -> Result { - let result = sqlx::query("DELETE FROM admins WHERE id::text = $1") - .bind(&id) - .execute(&state.db) - .await - .map_err(|e| AppError::Internal(format!("failed to delete admin: {e}")))?; - - if result.rows_affected() == 0 { - return Err(AppError::NotFound(format!("admin '{id}' not found"))); - } - - Ok(StatusCode::NO_CONTENT) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn hash_api_key_produces_deterministic_sha256_hex() { - let h1 = hash_api_key("test-key"); - let h2 = hash_api_key("test-key"); - assert_eq!(h1, h2); - assert_eq!(h1.len(), 64); - assert!(h1.chars().all(|c| c.is_ascii_hexdigit())); - } - - #[test] - fn hash_api_key_different_inputs_differ() { - let h1 = hash_api_key("key-a"); - let h2 = hash_api_key("key-b"); - assert_ne!(h1, h2); - } - - #[test] - fn hash_api_key_known_value() { - // SHA-256 of "hello" is well-known - let hash = hash_api_key("hello"); - assert_eq!( - hash, - "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824" - ); - } -} diff --git a/src/admin/admins.rs b/src/admin/admins.rs new file mode 100644 index 0000000..5632a3c --- /dev/null +++ b/src/admin/admins.rs @@ -0,0 +1,89 @@ +use axum::extract::{Path, State}; +use axum::http::StatusCode; +use axum::Json; +use serde_json::Value; + +use crate::AppState; +use crate::error::AppError; + +use super::auth::AdminAuth; +use super::hash::hash_api_key; +use super::types::{AdminSummary, CreateAdminBody}; + +/// POST /admin/admins — create a new admin. Returns the API key once. +pub(super) async fn create_admin( + State(state): State, + _admin: AdminAuth, + Json(body): Json, +) -> Result<(StatusCode, Json), AppError> { + let api_key = uuid::Uuid::new_v4().to_string(); + let key_hash = hash_api_key(&api_key); + + let row: (String,) = sqlx::query_as( + "INSERT INTO admins (name, api_key_hash) VALUES ($1, $2) RETURNING id::text", + ) + .bind(&body.name) + .bind(&key_hash) + .fetch_one(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to create admin: {e}")))?; + + Ok(( + StatusCode::CREATED, + Json(serde_json::json!({ + "id": row.0, + "name": body.name, + "api_key": api_key, + })), + )) +} + +/// GET /admin/admins — list all admins (without keys). +pub(super) async fn list_admins( + State(state): State, + _admin: AdminAuth, +) -> Result>, AppError> { + #[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() + .map(|(id, name, created_at, last_used_at)| AdminSummary { + id, + name, + created_at, + last_used_at, + }) + .collect(); + + Ok(Json(admins)) +} + +/// DELETE /admin/admins/:id — remove an admin. +pub(super) async fn delete_admin( + State(state): State, + _admin: AdminAuth, + Path(id): Path, +) -> Result { + let result = sqlx::query("DELETE FROM admins WHERE id::text = $1") + .bind(&id) + .execute(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to delete admin: {e}")))?; + + if result.rows_affected() == 0 { + return Err(AppError::NotFound(format!("admin '{id}' not found"))); + } + + Ok(StatusCode::NO_CONTENT) +} diff --git a/src/admin/auth.rs b/src/admin/auth.rs new file mode 100644 index 0000000..b175b4c --- /dev/null +++ b/src/admin/auth.rs @@ -0,0 +1,59 @@ +use crate::AppState; +use crate::error::AppError; + +use super::hash::hash_api_key; + +/// Axum extractor for admin auth. Checks the Bearer token against: +/// 1. The `admins` table (hashed key lookup) +/// 2. Falls back to `ADMIN_SECRET` env var for bootstrap +pub struct AdminAuth; + +impl axum::extract::FromRequestParts for AdminAuth { + type Rejection = AppError; + + async fn from_request_parts( + parts: &mut axum::http::request::Parts, + state: &AppState, + ) -> Result { + let header = parts + .headers + .get("authorization") + .and_then(|v| v.to_str().ok()) + .ok_or_else(|| AppError::Auth("missing Authorization header".into()))?; + + let token = header + .strip_prefix("Bearer ") + .ok_or_else(|| AppError::Auth("invalid Authorization scheme".into()))?; + + // 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}")))?; + + 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; + }); + return Ok(AdminAuth); + } + + // Fall back to ADMIN_SECRET env var + if let Some(ref secret) = state.config.admin_secret + && token == secret + { + return Ok(AdminAuth); + } + + Err(AppError::Auth("invalid admin credentials".into())) + } +} diff --git a/src/admin/backfill.rs b/src/admin/backfill.rs new file mode 100644 index 0000000..20cf235 --- /dev/null +++ b/src/admin/backfill.rs @@ -0,0 +1,95 @@ +use axum::extract::State; +use axum::http::StatusCode; +use axum::Json; +use serde_json::Value; + +use crate::AppState; +use crate::error::AppError; + +use super::auth::AdminAuth; +use super::types::{BackfillJob, CreateBackfillBody}; + +/// POST /admin/backfill — create a new backfill job. +pub(super) async fn create_backfill( + State(state): State, + _admin: AdminAuth, + Json(body): Json, +) -> Result<(StatusCode, Json), AppError> { + let row: (String,) = sqlx::query_as( + "INSERT INTO backfill_jobs (collection, did) VALUES ($1, $2) RETURNING id::text", + ) + .bind(&body.collection) + .bind(&body.did) + .fetch_one(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to create backfill job: {e}")))?; + + Ok(( + StatusCode::CREATED, + Json(serde_json::json!({ + "id": row.0, + "status": "pending", + })), + )) +} + +/// GET /admin/backfill/status — list all backfill jobs. +pub(super) async fn backfill_status( + State(state): State, + _admin: AdminAuth, +) -> Result>, AppError> { + #[allow(clippy::type_complexity)] + let rows: Vec<( + String, + Option, + Option, + String, + Option, + Option, + Option, + Option, + Option>, + Option>, + chrono::DateTime, + )> = sqlx::query_as( + "SELECT id::text, collection, did, status, total_repos, processed_repos, total_records, error, started_at, completed_at, created_at FROM backfill_jobs ORDER BY created_at DESC", + ) + .fetch_all(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to list backfill jobs: {e}")))?; + + let jobs: Vec = rows + .into_iter() + .map( + |( + id, + collection, + did, + status, + total_repos, + processed_repos, + total_records, + error, + started_at, + completed_at, + created_at, + )| { + BackfillJob { + id, + collection, + did, + status, + total_repos, + processed_repos, + total_records, + error, + started_at, + completed_at, + created_at, + } + }, + ) + .collect(); + + Ok(Json(jobs)) +} diff --git a/src/admin/bootstrap.rs b/src/admin/bootstrap.rs new file mode 100644 index 0000000..66f633f --- /dev/null +++ b/src/admin/bootstrap.rs @@ -0,0 +1,25 @@ +use super::hash::hash_api_key; + +/// Bootstrap: if no admins exist and ADMIN_SECRET is set, create a bootstrap admin. +pub async fn bootstrap(db: &sqlx::PgPool, admin_secret: &Option) { + let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM admins") + .fetch_one(db) + .await + .unwrap_or((0,)); + + if count.0 > 0 { + return; + } + + if let Some(secret) = admin_secret { + let key_hash = hash_api_key(secret); + let _ = sqlx::query( + "INSERT INTO admins (name, api_key_hash) VALUES ($1, $2) ON CONFLICT DO NOTHING", + ) + .bind("bootstrap") + .bind(&key_hash) + .execute(db) + .await; + tracing::info!("created bootstrap admin from ADMIN_SECRET"); + } +} diff --git a/src/admin/hash.rs b/src/admin/hash.rs new file mode 100644 index 0000000..6c00b3e --- /dev/null +++ b/src/admin/hash.rs @@ -0,0 +1,38 @@ +use sha2::{Digest, Sha256}; + +/// SHA-256 hash a plaintext API key for storage/comparison. +pub(crate) fn hash_api_key(key: &str) -> String { + let hash = Sha256::digest(key.as_bytes()); + hex::encode(hash) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn hash_api_key_produces_deterministic_sha256_hex() { + let h1 = hash_api_key("test-key"); + let h2 = hash_api_key("test-key"); + assert_eq!(h1, h2); + assert_eq!(h1.len(), 64); + assert!(h1.chars().all(|c| c.is_ascii_hexdigit())); + } + + #[test] + fn hash_api_key_different_inputs_differ() { + let h1 = hash_api_key("key-a"); + let h2 = hash_api_key("key-b"); + assert_ne!(h1, h2); + } + + #[test] + fn hash_api_key_known_value() { + // SHA-256 of "hello" is well-known + let hash = hash_api_key("hello"); + assert_eq!( + hash, + "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824" + ); + } +} diff --git a/src/admin/lexicons.rs b/src/admin/lexicons.rs new file mode 100644 index 0000000..41416c8 --- /dev/null +++ b/src/admin/lexicons.rs @@ -0,0 +1,185 @@ +use axum::extract::{Path, State}; +use axum::http::StatusCode; +use axum::Json; +use serde_json::Value; + +use crate::AppState; +use crate::error::AppError; +use crate::lexicon::{LexiconType, ParsedLexicon}; + +use super::auth::AdminAuth; +use super::types::{LexiconSummary, UploadLexiconBody}; + +/// Send the current record collection list to the Jetstream task so it +/// reconnects with the updated filter. +async fn notify_jetstream(state: &AppState) { + let collections = state.lexicons.get_record_collections().await; + let _ = state.collections_tx.send(collections); +} + +/// POST /admin/lexicons — upload (upsert) a lexicon. +pub(super) async fn upload_lexicon( + State(state): State, + _admin: AdminAuth, + Json(body): Json, +) -> Result<(StatusCode, Json), AppError> { + // Validate basic structure + let lexicon_version = body + .lexicon_json + .get("lexicon") + .and_then(|v| v.as_i64()) + .ok_or_else(|| { + AppError::BadRequest("lexicon JSON must have a numeric 'lexicon' field".into()) + })?; + + if lexicon_version != 1 { + return Err(AppError::BadRequest(format!( + "unsupported lexicon version: {lexicon_version}" + ))); + } + + let id = body + .lexicon_json + .get("id") + .and_then(|v| v.as_str()) + .ok_or_else(|| AppError::BadRequest("lexicon JSON must have a string 'id' field".into()))? + .to_string(); + + // Validate it parses correctly + ParsedLexicon::parse(body.lexicon_json.clone(), 1, body.target_collection.clone()) + .map_err(|e| AppError::BadRequest(format!("failed to parse lexicon: {e}")))?; + + // Upsert into database + let row: (i32,) = sqlx::query_as( + r#" + INSERT INTO lexicons (id, lexicon_json, backfill, target_collection) + VALUES ($1, $2, $3, $4) + ON CONFLICT (id) DO UPDATE SET + lexicon_json = EXCLUDED.lexicon_json, + backfill = EXCLUDED.backfill, + target_collection = EXCLUDED.target_collection, + revision = lexicons.revision + 1, + updated_at = NOW() + RETURNING revision + "#, + ) + .bind(&id) + .bind(&body.lexicon_json) + .bind(body.backfill) + .bind(&body.target_collection) + .fetch_one(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to upsert lexicon: {e}")))?; + + let revision = row.0; + + // Update in-memory registry with correct revision + let parsed = ParsedLexicon::parse(body.lexicon_json, revision, body.target_collection) + .map_err(|e| AppError::Internal(format!("failed to re-parse lexicon: {e}")))?; + let is_record = parsed.lexicon_type == LexiconType::Record; + state.lexicons.upsert(parsed).await; + + if is_record { + notify_jetstream(&state).await; + } + + let status = if revision == 1 { + StatusCode::CREATED + } else { + StatusCode::OK + }; + + Ok(( + status, + Json(serde_json::json!({ + "id": id, + "revision": revision, + })), + )) +} + +/// GET /admin/lexicons — list all lexicons. +pub(super) 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", + ) + .fetch_all(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to list lexicons: {e}")))?; + + let summaries: Vec = rows + .into_iter() + .map(|(id, revision, json, backfill, created_at, updated_at)| { + let lexicon_type = ParsedLexicon::parse(json, revision, None) + .map(|p| format!("{:?}", p.lexicon_type).to_lowercase()) + .unwrap_or_else(|_| "unknown".into()); + + LexiconSummary { + id, + revision, + lexicon_type, + backfill, + created_at, + updated_at, + } + }) + .collect(); + + Ok(Json(summaries)) +} + +/// GET /admin/lexicons/:id — get a single lexicon. +pub(super) async fn get_lexicon( + State(state): State, + _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", + ) + .bind(&id) + .fetch_optional(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to get lexicon: {e}")))?; + + let (id, revision, lexicon_json, backfill, created_at, updated_at) = + row.ok_or_else(|| AppError::NotFound(format!("lexicon '{id}' not found")))?; + + Ok(Json(serde_json::json!({ + "id": id, + "revision": revision, + "lexicon_json": lexicon_json, + "backfill": backfill, + "created_at": created_at, + "updated_at": updated_at, + }))) +} + +/// DELETE /admin/lexicons/:id — remove a lexicon. +pub(super) async fn delete_lexicon( + State(state): State, + _admin: AdminAuth, + Path(id): Path, +) -> Result { + let result = sqlx::query("DELETE FROM lexicons WHERE id = $1") + .bind(&id) + .execute(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to delete lexicon: {e}")))?; + + if result.rows_affected() == 0 { + return Err(AppError::NotFound(format!("lexicon '{id}' not found"))); + } + + state.lexicons.remove(&id).await; + notify_jetstream(&state).await; + + Ok(StatusCode::NO_CONTENT) +} diff --git a/src/admin/mod.rs b/src/admin/mod.rs new file mode 100644 index 0000000..5b79f7b --- /dev/null +++ b/src/admin/mod.rs @@ -0,0 +1,26 @@ +mod admins; +pub(crate) mod auth; +pub(crate) mod bootstrap; +mod backfill; +pub(crate) mod hash; +mod lexicons; +mod stats; +mod types; + +use axum::routing::{delete, get, post}; +use axum::Router; + +use crate::AppState; + +pub use bootstrap::bootstrap; + +pub fn admin_routes(_state: AppState) -> Router { + Router::new() + .route("/lexicons", post(lexicons::upload_lexicon).get(lexicons::list_lexicons)) + .route("/lexicons/{id}", get(lexicons::get_lexicon).delete(lexicons::delete_lexicon)) + .route("/stats", get(stats::stats)) + .route("/backfill", post(backfill::create_backfill)) + .route("/backfill/status", get(backfill::backfill_status)) + .route("/admins", post(admins::create_admin).get(admins::list_admins)) + .route("/admins/{id}", delete(admins::delete_admin)) +} diff --git a/src/admin/stats.rs b/src/admin/stats.rs new file mode 100644 index 0000000..ebcff64 --- /dev/null +++ b/src/admin/stats.rs @@ -0,0 +1,34 @@ +use axum::extract::State; +use axum::Json; + +use crate::AppState; +use crate::error::AppError; + +use super::auth::AdminAuth; +use super::types::{CollectionStat, StatsResponse}; + +/// GET /admin/stats — system statistics. +pub(super) async fn stats( + State(state): State, + _admin: AdminAuth, +) -> Result, AppError> { + let total: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM records") + .fetch_one(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to count records: {e}")))?; + + let collections: Vec<(String, i64)> = sqlx::query_as( + "SELECT collection, COUNT(*) FROM records GROUP BY collection ORDER BY collection", + ) + .fetch_all(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to count by collection: {e}")))?; + + Ok(Json(StatsResponse { + total_records: total.0, + collections: collections + .into_iter() + .map(|(collection, count)| CollectionStat { collection, count }) + .collect(), + })) +} diff --git a/src/admin/types.rs b/src/admin/types.rs new file mode 100644 index 0000000..60fb1c4 --- /dev/null +++ b/src/admin/types.rs @@ -0,0 +1,86 @@ +use serde::{Deserialize, Serialize}; +use serde_json::Value; + +// --------------------------------------------------------------------------- +// Lexicon types +// --------------------------------------------------------------------------- + +#[derive(Serialize)] +pub(super) struct LexiconSummary { + pub(super) id: String, + pub(super) revision: i32, + pub(super) lexicon_type: String, + pub(super) backfill: bool, + pub(super) created_at: chrono::DateTime, + pub(super) updated_at: chrono::DateTime, +} + +#[derive(Deserialize)] +pub(super) struct UploadLexiconBody { + pub(super) lexicon_json: Value, + #[serde(default = "default_backfill")] + pub(super) backfill: bool, + pub(super) target_collection: Option, +} + +fn default_backfill() -> bool { + true +} + +// --------------------------------------------------------------------------- +// Stats types +// --------------------------------------------------------------------------- + +#[derive(Serialize)] +pub(super) struct StatsResponse { + pub(super) total_records: i64, + pub(super) collections: Vec, +} + +#[derive(Serialize)] +pub(super) struct CollectionStat { + pub(super) collection: String, + pub(super) count: i64, +} + +// --------------------------------------------------------------------------- +// Backfill types +// --------------------------------------------------------------------------- + +#[derive(Deserialize)] +pub(super) struct CreateBackfillBody { + pub(super) collection: Option, + pub(super) did: Option, +} + +#[derive(Serialize)] +pub(super) struct BackfillJob { + pub(super) id: String, + pub(super) collection: Option, + pub(super) did: Option, + pub(super) status: String, + pub(super) total_repos: Option, + pub(super) processed_repos: Option, + pub(super) total_records: Option, + pub(super) error: Option, + pub(super) started_at: Option>, + pub(super) completed_at: Option>, + pub(super) created_at: chrono::DateTime, +} + +// --------------------------------------------------------------------------- +// Admin management types +// --------------------------------------------------------------------------- + +#[derive(Deserialize)] +pub(super) struct CreateAdminBody { + pub(super) name: String, +} + +#[derive(Serialize)] +pub(super) struct AdminSummary { + pub(super) id: String, + pub(super) name: String, + pub(super) created_at: chrono::DateTime, + pub(super) last_used_at: Option>, +} diff --git a/src/repo.rs b/src/repo.rs deleted file mode 100644 index 0ccff2e..0000000 --- a/src/repo.rs +++ /dev/null @@ -1,600 +0,0 @@ -use axum::body::Bytes; -use axum::extract::State; -use axum::http::{HeaderMap, StatusCode}; -use axum::response::{IntoResponse, Response}; -use base64::Engine; -use p256::pkcs8::EncodePrivateKey; -use serde::Deserialize; -use serde_json::{Value, json}; -use sha2::{Digest, Sha256}; - -use crate::AppState; -use crate::auth::Claims; -use crate::error::AppError; - -// --------------------------------------------------------------------------- -// AT URI parsing -// --------------------------------------------------------------------------- - -/// Extract the DID from an AT URI (at://did/collection/rkey). -pub(crate) fn parse_did_from_at_uri(uri: &str) -> Result { - let stripped = uri - .strip_prefix("at://") - .ok_or_else(|| AppError::Internal("AT URI must start with at://".into()))?; - - stripped - .split('/') - .next() - .map(|s| s.to_string()) - .ok_or_else(|| AppError::Internal("invalid AT URI".into())) -} - -// --------------------------------------------------------------------------- -// AIP session types -// --------------------------------------------------------------------------- - -#[derive(Deserialize)] -pub(crate) struct AtpSession { - pub(crate) access_token: String, - pub(crate) pds_endpoint: String, - pub(crate) dpop_jwk: DpopJwk, -} - -#[derive(Deserialize)] -pub(crate) struct DpopJwk { - pub(crate) x: String, - pub(crate) y: String, - pub(crate) d: String, -} - -// --------------------------------------------------------------------------- -// DPoP proof generation -// --------------------------------------------------------------------------- - -#[derive(serde::Serialize)] -struct DpopClaims { - jti: String, - htm: String, - htu: String, - iat: i64, - exp: i64, - ath: String, - #[serde(skip_serializing_if = "Option::is_none")] - nonce: Option, -} - -/// Fetch the user's AT Protocol session (PDS credentials) from AIP. -pub(crate) async fn get_atp_session(state: &AppState, token: &str) -> Result { - let url = format!( - "{}/api/atprotocol/session", - state.config.aip_url.trim_end_matches('/') - ); - - let resp = state - .http - .get(&url) - .header("authorization", format!("Bearer {token}")) - .send() - .await - .map_err(|e| AppError::Internal(format!("AIP session request failed: {e}")))?; - - if !resp.status().is_success() { - return Err(AppError::Auth(format!( - "AIP session returned {}", - resp.status() - ))); - } - - resp.json() - .await - .map_err(|e| AppError::Internal(format!("invalid AIP session response: {e}"))) -} - -/// Generate a DPoP proof JWT for a PDS request. -pub(crate) fn generate_dpop_proof( - method: &str, - url: &str, - dpop_jwk: &DpopJwk, - access_token: &str, - nonce: Option<&str>, -) -> Result { - // Decode the private key from base64url - let d_bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD - .decode(&dpop_jwk.d) - .map_err(|e| AppError::Internal(format!("invalid DPoP key d: {e}")))?; - - let secret_key = p256::SecretKey::from_slice(&d_bytes) - .map_err(|e| AppError::Internal(format!("invalid P-256 key: {e}")))?; - - let pkcs8_der = secret_key - .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()); - - // Public JWK for the header (no private component) - let public_jwk = jsonwebtoken::jwk::Jwk { - common: jsonwebtoken::jwk::CommonParameters { - public_key_use: None, - key_operations: None, - key_algorithm: None, - key_id: None, - x509_url: None, - x509_chain: None, - x509_sha1_fingerprint: None, - x509_sha256_fingerprint: None, - }, - algorithm: jsonwebtoken::jwk::AlgorithmParameters::EllipticCurve( - jsonwebtoken::jwk::EllipticCurveKeyParameters { - key_type: jsonwebtoken::jwk::EllipticCurveKeyType::EC, - curve: jsonwebtoken::jwk::EllipticCurve::P256, - x: dpop_jwk.x.clone(), - y: dpop_jwk.y.clone(), - }, - ), - }; - - let mut header = jsonwebtoken::Header::new(jsonwebtoken::Algorithm::ES256); - header.typ = Some("dpop+jwt".to_string()); - header.jwk = Some(public_jwk); - - // Access token hash (ath) - let ath_hash = Sha256::digest(access_token.as_bytes()); - let ath = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(ath_hash); - - let now = chrono::Utc::now().timestamp(); - let claims = DpopClaims { - jti: uuid::Uuid::new_v4().to_string(), - htm: method.to_uppercase(), - htu: url.to_string(), - iat: now, - exp: now + 300, - ath, - nonce: nonce.map(|n| n.to_string()), - }; - - jsonwebtoken::encode(&header, &claims, &encoding_key) - .map_err(|e| AppError::Internal(format!("DPoP proof signing failed: {e}"))) -} - -// --------------------------------------------------------------------------- -// PDS request helpers with DPoP + nonce retry -// --------------------------------------------------------------------------- - -/// Forward a PDS response back to the client, preserving status and body. -pub(crate) async fn forward_pds_response(resp: reqwest::Response) -> Result { - let status = resp.status(); - let body = resp - .bytes() - .await - .map_err(|e| AppError::Internal(format!("failed to read PDS response: {e}")))?; - - let axum_status = StatusCode::from_u16(status.as_u16()).unwrap(); - - if status.is_success() { - Ok(( - axum_status, - [(axum::http::header::CONTENT_TYPE, "application/json")], - body, - ) - .into_response()) - } else { - let body_str = String::from_utf8_lossy(&body); - tracing::warn!(status = %axum_status, body = %body_str, "PDS returned error"); - Err(AppError::PdsError(axum_status, body)) - } -} - -/// POST JSON to a PDS XRPC endpoint with DPoP auth and nonce retry. -/// Returns the raw reqwest::Response so callers can inspect the body. -pub(crate) async fn pds_post_json_raw( - state: &AppState, - session: &AtpSession, - xrpc_method: &str, - body: &Value, -) -> Result { - let url = format!( - "{}/xrpc/{xrpc_method}", - session.pds_endpoint.trim_end_matches('/') - ); - - let dpop = generate_dpop_proof("POST", &url, &session.dpop_jwk, &session.access_token, None)?; - - 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 failed: {e}")))?; - - // Retry with nonce if PDS requires it - 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); - } - - Ok(resp) -} - -/// POST a binary blob to the PDS with DPoP auth and nonce retry. -async fn pds_post_blob( - state: &AppState, - session: &AtpSession, - content_type: &str, - blob: Bytes, -) -> Result { - let url = format!( - "{}/xrpc/com.atproto.repo.uploadBlob", - session.pds_endpoint.trim_end_matches('/') - ); - - let dpop = generate_dpop_proof("POST", &url, &session.dpop_jwk, &session.access_token, None)?; - - let resp = state - .http - .post(&url) - .header("authorization", format!("DPoP {}", session.access_token)) - .header("dpop", &dpop) - .header("content-type", content_type) - .body(blob.clone()) - .send() - .await - .map_err(|e| AppError::Internal(format!("PDS uploadBlob failed: {e}")))?; - - 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; - } - - forward_pds_response(resp).await -} - -// --------------------------------------------------------------------------- -// Public handlers -// --------------------------------------------------------------------------- - -pub async fn upload_blob( - State(state): State, - claims: Claims, - headers: HeaderMap, - body: Bytes, -) -> Result { - let session = get_atp_session(&state, claims.token()).await?; - - let content_type = headers - .get("content-type") - .and_then(|v| v.to_str().ok()) - .unwrap_or("application/octet-stream"); - - pds_post_blob(&state, &session, content_type, body).await -} - -/// Walk `media[]` and add a `url` field to each blob so the frontend can -/// display images directly. -pub(crate) fn enrich_media_blobs(record: &mut Value, pds: &str, did: &str) { - let media = match record.get_mut("media").and_then(|m| m.as_array_mut()) { - Some(arr) => arr, - None => return, - }; - - let pds_base = pds.trim_end_matches('/'); - - for item in media.iter_mut() { - let cid = item - .get("blob") - .and_then(|b| b.get("ref")) - .and_then(|r| r.get("$link")) - .and_then(|l| l.as_str()) - .map(|s| s.to_string()); - - 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}" - )), - ); - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - - // ----------------------------------------------------------------------- - // parse_did_from_at_uri - // ----------------------------------------------------------------------- - - #[test] - fn parse_did_from_valid_at_uri() { - let did = parse_did_from_at_uri("at://did:plc:abc123/app.bsky.feed.post/3k2bqxyz").unwrap(); - assert_eq!(did, "did:plc:abc123"); - } - - #[test] - fn parse_did_from_uri_with_no_rkey() { - let did = parse_did_from_at_uri("at://did:plc:abc123/collection").unwrap(); - assert_eq!(did, "did:plc:abc123"); - } - - #[test] - fn parse_did_from_did_web_uri() { - let did = parse_did_from_at_uri("at://did:web:example.com/collection/rkey").unwrap(); - assert_eq!(did, "did:web:example.com"); - } - - #[test] - fn parse_did_from_uri_missing_prefix() { - let result = parse_did_from_at_uri("did:plc:abc123/collection/rkey"); - assert!(result.is_err()); - } - - // ----------------------------------------------------------------------- - // enrich_media_blobs - // ----------------------------------------------------------------------- - - #[test] - fn enrich_media_adds_url() { - let mut record = json!({ - "media": [{ - "blob": { - "ref": { "$link": "bafyreiabc" }, - "mimeType": "image/jpeg", - "size": 1024 - } - }] - }); - - enrich_media_blobs(&mut record, "https://pds.example.com", "did:plc:test"); - - let url = record["media"][0]["blob"]["url"].as_str().unwrap(); - assert_eq!( - url, - "https://pds.example.com/xrpc/com.atproto.sync.getBlob?did=did:plc:test&cid=bafyreiabc" - ); - } - - #[test] - fn enrich_media_noop_without_media() { - let mut record = json!({"title": "test"}); - enrich_media_blobs(&mut record, "https://pds.example.com", "did:plc:test"); - assert!(record.get("media").is_none()); - } - - #[test] - fn enrich_media_skips_items_without_ref() { - let mut record = json!({ - "media": [{ - "blob": { "mimeType": "image/png" } - }] - }); - - enrich_media_blobs(&mut record, "https://pds.example.com", "did:plc:test"); - assert!(record["media"][0]["blob"].get("url").is_none()); - } - - #[test] - fn enrich_media_handles_multiple_items() { - let mut record = json!({ - "media": [ - { "blob": { "ref": { "$link": "cid1" } } }, - { "blob": { "ref": { "$link": "cid2" } } } - ] - }); - - enrich_media_blobs(&mut record, "https://pds.example.com/", "did:plc:x"); - - let url1 = record["media"][0]["blob"]["url"].as_str().unwrap(); - let url2 = record["media"][1]["blob"]["url"].as_str().unwrap(); - assert!(url1.contains("cid1")); - assert!(url2.contains("cid2")); - } - - #[test] - fn enrich_media_trims_trailing_slash() { - let mut record = json!({ - "media": [{ - "blob": { "ref": { "$link": "bafytest" } } - }] - }); - - enrich_media_blobs(&mut record, "https://pds.example.com/", "did:plc:test"); - - let url = record["media"][0]["blob"]["url"].as_str().unwrap(); - assert!(url.starts_with("https://pds.example.com/xrpc/")); - assert!(!url.contains("//xrpc")); - } - - // ----------------------------------------------------------------------- - // generate_dpop_proof - // ----------------------------------------------------------------------- - - fn test_dpop_jwk() -> DpopJwk { - use p256::elliptic_curve::rand_core::OsRng; - use p256::elliptic_curve::sec1::ToEncodedPoint; - // Generate a valid P-256 key for testing - let secret = p256::SecretKey::random(&mut OsRng); - let public = secret.public_key(); - 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()), - } - } - - #[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 parts: Vec<&str> = token.split('.').collect(); - assert_eq!(parts.len(), 3, "JWT should have 3 parts"); - } - - #[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 header_b64 = token.split('.').next().unwrap(); - let header_bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD - .decode(header_b64) - .unwrap(); - let header: serde_json::Value = serde_json::from_slice(&header_bytes).unwrap(); - - assert_eq!(header["typ"], "dpop+jwt"); - assert_eq!(header["alg"], "ES256"); - assert!(header.get("jwk").is_some()); - } - - #[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 payload_b64 = token.split('.').nth(1).unwrap(); - let payload_bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD - .decode(payload_b64) - .unwrap(); - let claims: serde_json::Value = serde_json::from_slice(&payload_bytes).unwrap(); - - assert_eq!(claims["htm"], "GET"); - assert_eq!(claims["htu"], "https://pds.example.com/xrpc/test"); - assert!(claims.get("jti").is_some()); - assert!(claims.get("iat").is_some()); - assert!(claims.get("exp").is_some()); - assert!(claims.get("ath").is_some()); - assert!(claims.get("nonce").is_none()); - } - - #[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 payload_b64 = token.split('.').nth(1).unwrap(); - let payload_bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD - .decode(payload_b64) - .unwrap(); - let claims: serde_json::Value = serde_json::from_slice(&payload_bytes).unwrap(); - - assert_eq!(claims["nonce"], "abc123"); - } - - #[test] - 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 payload_b64 = token.split('.').nth(1).unwrap(); - let payload_bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD - .decode(payload_b64) - .unwrap(); - let claims: serde_json::Value = serde_json::from_slice(&payload_bytes).unwrap(); - - let expected_hash = Sha256::digest(access_token.as_bytes()); - let expected_ath = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(expected_hash); - assert_eq!(claims["ath"], expected_ath); - } - - #[test] - fn dpop_proof_invalid_key_returns_error() { - let jwk = DpopJwk { - x: "invalid".into(), - y: "invalid".into(), - d: "invalid".into(), - }; - let result = generate_dpop_proof("POST", "https://example.com", &jwk, "tok", None); - assert!(result.is_err()); - } -} diff --git a/src/repo/at_uri.rs b/src/repo/at_uri.rs new file mode 100644 index 0000000..137dfec --- /dev/null +++ b/src/repo/at_uri.rs @@ -0,0 +1,43 @@ +use crate::error::AppError; + +/// Extract the DID from an AT URI (at://did/collection/rkey). +pub(crate) fn parse_did_from_at_uri(uri: &str) -> Result { + let stripped = uri + .strip_prefix("at://") + .ok_or_else(|| AppError::Internal("AT URI must start with at://".into()))?; + + stripped + .split('/') + .next() + .map(|s| s.to_string()) + .ok_or_else(|| AppError::Internal("invalid AT URI".into())) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_did_from_valid_at_uri() { + let did = parse_did_from_at_uri("at://did:plc:abc123/app.bsky.feed.post/3k2bqxyz").unwrap(); + assert_eq!(did, "did:plc:abc123"); + } + + #[test] + fn parse_did_from_uri_with_no_rkey() { + let did = parse_did_from_at_uri("at://did:plc:abc123/collection").unwrap(); + assert_eq!(did, "did:plc:abc123"); + } + + #[test] + fn parse_did_from_did_web_uri() { + let did = parse_did_from_at_uri("at://did:web:example.com/collection/rkey").unwrap(); + assert_eq!(did, "did:web:example.com"); + } + + #[test] + fn parse_did_from_uri_missing_prefix() { + let result = parse_did_from_at_uri("did:plc:abc123/collection/rkey"); + assert!(result.is_err()); + } +} diff --git a/src/repo/dpop.rs b/src/repo/dpop.rs new file mode 100644 index 0000000..defbef3 --- /dev/null +++ b/src/repo/dpop.rs @@ -0,0 +1,222 @@ +use base64::Engine; +use p256::pkcs8::EncodePrivateKey; +use sha2::{Digest, Sha256}; + +use crate::error::AppError; + +use super::session::DpopJwk; + +#[derive(serde::Serialize)] +struct DpopClaims { + jti: String, + htm: String, + htu: String, + iat: i64, + exp: i64, + ath: String, + #[serde(skip_serializing_if = "Option::is_none")] + nonce: Option, +} + +/// Generate a DPoP proof JWT for a PDS request. +pub(crate) fn generate_dpop_proof( + method: &str, + url: &str, + dpop_jwk: &DpopJwk, + access_token: &str, + nonce: Option<&str>, +) -> Result { + // Decode the private key from base64url + let d_bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(&dpop_jwk.d) + .map_err(|e| AppError::Internal(format!("invalid DPoP key d: {e}")))?; + + let secret_key = p256::SecretKey::from_slice(&d_bytes) + .map_err(|e| AppError::Internal(format!("invalid P-256 key: {e}")))?; + + let pkcs8_der = secret_key + .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()); + + // Public JWK for the header (no private component) + let public_jwk = jsonwebtoken::jwk::Jwk { + common: jsonwebtoken::jwk::CommonParameters { + public_key_use: None, + key_operations: None, + key_algorithm: None, + key_id: None, + x509_url: None, + x509_chain: None, + x509_sha1_fingerprint: None, + x509_sha256_fingerprint: None, + }, + algorithm: jsonwebtoken::jwk::AlgorithmParameters::EllipticCurve( + jsonwebtoken::jwk::EllipticCurveKeyParameters { + key_type: jsonwebtoken::jwk::EllipticCurveKeyType::EC, + curve: jsonwebtoken::jwk::EllipticCurve::P256, + x: dpop_jwk.x.clone(), + y: dpop_jwk.y.clone(), + }, + ), + }; + + let mut header = jsonwebtoken::Header::new(jsonwebtoken::Algorithm::ES256); + header.typ = Some("dpop+jwt".to_string()); + header.jwk = Some(public_jwk); + + // Access token hash (ath) + let ath_hash = Sha256::digest(access_token.as_bytes()); + let ath = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(ath_hash); + + let now = chrono::Utc::now().timestamp(); + let claims = DpopClaims { + jti: uuid::Uuid::new_v4().to_string(), + htm: method.to_uppercase(), + htu: url.to_string(), + iat: now, + exp: now + 300, + ath, + nonce: nonce.map(|n| n.to_string()), + }; + + jsonwebtoken::encode(&header, &claims, &encoding_key) + .map_err(|e| AppError::Internal(format!("DPoP proof signing failed: {e}"))) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn test_dpop_jwk() -> DpopJwk { + use p256::elliptic_curve::rand_core::OsRng; + use p256::elliptic_curve::sec1::ToEncodedPoint; + // Generate a valid P-256 key for testing + let secret = p256::SecretKey::random(&mut OsRng); + let public = secret.public_key(); + 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()), + } + } + + #[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 parts: Vec<&str> = token.split('.').collect(); + assert_eq!(parts.len(), 3, "JWT should have 3 parts"); + } + + #[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 header_b64 = token.split('.').next().unwrap(); + let header_bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(header_b64) + .unwrap(); + let header: serde_json::Value = serde_json::from_slice(&header_bytes).unwrap(); + + assert_eq!(header["typ"], "dpop+jwt"); + assert_eq!(header["alg"], "ES256"); + assert!(header.get("jwk").is_some()); + } + + #[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 payload_b64 = token.split('.').nth(1).unwrap(); + let payload_bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(payload_b64) + .unwrap(); + let claims: serde_json::Value = serde_json::from_slice(&payload_bytes).unwrap(); + + assert_eq!(claims["htm"], "GET"); + assert_eq!(claims["htu"], "https://pds.example.com/xrpc/test"); + assert!(claims.get("jti").is_some()); + assert!(claims.get("iat").is_some()); + assert!(claims.get("exp").is_some()); + assert!(claims.get("ath").is_some()); + assert!(claims.get("nonce").is_none()); + } + + #[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 payload_b64 = token.split('.').nth(1).unwrap(); + let payload_bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(payload_b64) + .unwrap(); + let claims: serde_json::Value = serde_json::from_slice(&payload_bytes).unwrap(); + + assert_eq!(claims["nonce"], "abc123"); + } + + #[test] + 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 payload_b64 = token.split('.').nth(1).unwrap(); + let payload_bytes = base64::engine::general_purpose::URL_SAFE_NO_PAD + .decode(payload_b64) + .unwrap(); + let claims: serde_json::Value = serde_json::from_slice(&payload_bytes).unwrap(); + + let expected_hash = Sha256::digest(access_token.as_bytes()); + let expected_ath = base64::engine::general_purpose::URL_SAFE_NO_PAD.encode(expected_hash); + assert_eq!(claims["ath"], expected_ath); + } + + #[test] + fn dpop_proof_invalid_key_returns_error() { + let jwk = DpopJwk { + x: "invalid".into(), + y: "invalid".into(), + d: "invalid".into(), + }; + let result = generate_dpop_proof("POST", "https://example.com", &jwk, "tok", None); + assert!(result.is_err()); + } +} diff --git a/src/repo/media.rs b/src/repo/media.rs new file mode 100644 index 0000000..b648f9e --- /dev/null +++ b/src/repo/media.rs @@ -0,0 +1,110 @@ +use serde_json::{Value, json}; + +/// Walk `media[]` and add a `url` field to each blob so the frontend can +/// display images directly. +pub(crate) fn enrich_media_blobs(record: &mut Value, pds: &str, did: &str) { + let media = match record.get_mut("media").and_then(|m| m.as_array_mut()) { + Some(arr) => arr, + None => return, + }; + + let pds_base = pds.trim_end_matches('/'); + + for item in media.iter_mut() { + let cid = item + .get("blob") + .and_then(|b| b.get("ref")) + .and_then(|r| r.get("$link")) + .and_then(|l| l.as_str()) + .map(|s| s.to_string()); + + 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}" + )), + ); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn enrich_media_adds_url() { + let mut record = json!({ + "media": [{ + "blob": { + "ref": { "$link": "bafyreiabc" }, + "mimeType": "image/jpeg", + "size": 1024 + } + }] + }); + + enrich_media_blobs(&mut record, "https://pds.example.com", "did:plc:test"); + + let url = record["media"][0]["blob"]["url"].as_str().unwrap(); + assert_eq!( + url, + "https://pds.example.com/xrpc/com.atproto.sync.getBlob?did=did:plc:test&cid=bafyreiabc" + ); + } + + #[test] + fn enrich_media_noop_without_media() { + let mut record = json!({"title": "test"}); + enrich_media_blobs(&mut record, "https://pds.example.com", "did:plc:test"); + assert!(record.get("media").is_none()); + } + + #[test] + fn enrich_media_skips_items_without_ref() { + let mut record = json!({ + "media": [{ + "blob": { "mimeType": "image/png" } + }] + }); + + enrich_media_blobs(&mut record, "https://pds.example.com", "did:plc:test"); + assert!(record["media"][0]["blob"].get("url").is_none()); + } + + #[test] + fn enrich_media_handles_multiple_items() { + let mut record = json!({ + "media": [ + { "blob": { "ref": { "$link": "cid1" } } }, + { "blob": { "ref": { "$link": "cid2" } } } + ] + }); + + enrich_media_blobs(&mut record, "https://pds.example.com/", "did:plc:x"); + + let url1 = record["media"][0]["blob"]["url"].as_str().unwrap(); + let url2 = record["media"][1]["blob"]["url"].as_str().unwrap(); + assert!(url1.contains("cid1")); + assert!(url2.contains("cid2")); + } + + #[test] + fn enrich_media_trims_trailing_slash() { + let mut record = json!({ + "media": [{ + "blob": { "ref": { "$link": "bafytest" } } + }] + }); + + enrich_media_blobs(&mut record, "https://pds.example.com/", "did:plc:test"); + + let url = record["media"][0]["blob"]["url"].as_str().unwrap(); + assert!(url.starts_with("https://pds.example.com/xrpc/")); + assert!(!url.contains("//xrpc")); + } +} diff --git a/src/repo/mod.rs b/src/repo/mod.rs new file mode 100644 index 0000000..e7282f1 --- /dev/null +++ b/src/repo/mod.rs @@ -0,0 +1,12 @@ +mod at_uri; +mod dpop; +mod media; +mod pds; +pub(crate) mod session; +mod upload_blob; + +pub use upload_blob::upload_blob; +pub(crate) use at_uri::parse_did_from_at_uri; +pub(crate) use media::enrich_media_blobs; +pub(crate) use pds::{forward_pds_response, pds_post_json_raw}; +pub(crate) use session::{AtpSession, get_atp_session}; diff --git a/src/repo/pds.rs b/src/repo/pds.rs new file mode 100644 index 0000000..b430316 --- /dev/null +++ b/src/repo/pds.rs @@ -0,0 +1,152 @@ +use axum::body::Bytes; +use axum::http::StatusCode; +use axum::response::{IntoResponse, Response}; +use serde_json::Value; + +use crate::AppState; +use crate::error::AppError; + +use super::dpop::generate_dpop_proof; +use super::session::AtpSession; + +/// Forward a PDS response back to the client, preserving status and body. +pub(crate) async fn forward_pds_response(resp: reqwest::Response) -> Result { + let status = resp.status(); + let body = resp + .bytes() + .await + .map_err(|e| AppError::Internal(format!("failed to read PDS response: {e}")))?; + + let axum_status = StatusCode::from_u16(status.as_u16()).unwrap(); + + if status.is_success() { + Ok(( + axum_status, + [(axum::http::header::CONTENT_TYPE, "application/json")], + body, + ) + .into_response()) + } else { + let body_str = String::from_utf8_lossy(&body); + tracing::warn!(status = %axum_status, body = %body_str, "PDS returned error"); + Err(AppError::PdsError(axum_status, body)) + } +} + +/// POST JSON to a PDS XRPC endpoint with DPoP auth and nonce retry. +/// Returns the raw reqwest::Response so callers can inspect the body. +pub(crate) async fn pds_post_json_raw( + state: &AppState, + session: &AtpSession, + xrpc_method: &str, + body: &Value, +) -> Result { + let url = format!( + "{}/xrpc/{xrpc_method}", + session.pds_endpoint.trim_end_matches('/') + ); + + let dpop = generate_dpop_proof("POST", &url, &session.dpop_jwk, &session.access_token, None)?; + + 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 failed: {e}")))?; + + // Retry with nonce if PDS requires it + 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); + } + + Ok(resp) +} + +/// POST a binary blob to the PDS with DPoP auth and nonce retry. +pub(super) async fn pds_post_blob( + state: &AppState, + session: &AtpSession, + content_type: &str, + blob: Bytes, +) -> Result { + let url = format!( + "{}/xrpc/com.atproto.repo.uploadBlob", + session.pds_endpoint.trim_end_matches('/') + ); + + let dpop = generate_dpop_proof("POST", &url, &session.dpop_jwk, &session.access_token, None)?; + + let resp = state + .http + .post(&url) + .header("authorization", format!("DPoP {}", session.access_token)) + .header("dpop", &dpop) + .header("content-type", content_type) + .body(blob.clone()) + .send() + .await + .map_err(|e| AppError::Internal(format!("PDS uploadBlob failed: {e}")))?; + + 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; + } + + forward_pds_response(resp).await +} diff --git a/src/repo/session.rs b/src/repo/session.rs new file mode 100644 index 0000000..67d6d78 --- /dev/null +++ b/src/repo/session.rs @@ -0,0 +1,45 @@ +use serde::Deserialize; + +use crate::AppState; +use crate::error::AppError; + +#[derive(Deserialize)] +pub(crate) struct AtpSession { + pub(crate) access_token: String, + pub(crate) pds_endpoint: String, + pub(crate) dpop_jwk: DpopJwk, +} + +#[derive(Deserialize)] +pub(crate) struct DpopJwk { + pub(crate) x: String, + pub(crate) y: String, + pub(crate) d: String, +} + +/// Fetch the user's AT Protocol session (PDS credentials) from AIP. +pub(crate) async fn get_atp_session(state: &AppState, token: &str) -> Result { + let url = format!( + "{}/api/atprotocol/session", + state.config.aip_url.trim_end_matches('/') + ); + + let resp = state + .http + .get(&url) + .header("authorization", format!("Bearer {token}")) + .send() + .await + .map_err(|e| AppError::Internal(format!("AIP session request failed: {e}")))?; + + if !resp.status().is_success() { + return Err(AppError::Auth(format!( + "AIP session returned {}", + resp.status() + ))); + } + + resp.json() + .await + .map_err(|e| AppError::Internal(format!("invalid AIP session response: {e}"))) +} diff --git a/src/repo/upload_blob.rs b/src/repo/upload_blob.rs new file mode 100644 index 0000000..ce2a92f --- /dev/null +++ b/src/repo/upload_blob.rs @@ -0,0 +1,27 @@ +use axum::body::Bytes; +use axum::extract::State; +use axum::http::HeaderMap; +use axum::response::Response; + +use crate::AppState; +use crate::auth::Claims; +use crate::error::AppError; + +use super::pds::pds_post_blob; +use super::session::get_atp_session; + +pub async fn upload_blob( + State(state): State, + claims: Claims, + headers: HeaderMap, + body: Bytes, +) -> Result { + let session = get_atp_session(&state, claims.token()).await?; + + let content_type = headers + .get("content-type") + .and_then(|v| v.to_str().ok()) + .unwrap_or("application/octet-stream"); + + pds_post_blob(&state, &session, content_type, body).await +} diff --git a/src/xrpc.rs b/src/xrpc.rs deleted file mode 100644 index cb6ca58..0000000 --- a/src/xrpc.rs +++ /dev/null @@ -1,351 +0,0 @@ -use axum::Json; -use axum::extract::{Path, Query, State}; -use axum::http::StatusCode; -use axum::response::{IntoResponse, Response}; -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; - -// --------------------------------------------------------------------------- -// Catch-all handler -// --------------------------------------------------------------------------- - -/// Catch-all GET handler for XRPC queries. -pub async fn xrpc_get( - State(state): State, - Path(method): Path, - Query(params): Query>, -) -> Result { - let lexicon = state - .lexicons - .get(&method) - .await - .ok_or_else(|| AppError::BadRequest(format!("method not found: {method}")))?; - - if lexicon.lexicon_type != LexiconType::Query { - return Err(AppError::BadRequest(format!( - "{method} is not a query endpoint" - ))); - } - - handle_query(&state, &method, ¶ms, &lexicon).await -} - -/// Catch-all POST handler for XRPC procedures. -pub async fn xrpc_post( - State(state): State, - Path(method): Path, - claims: Claims, - Json(body): Json, -) -> Result { - let lexicon = state - .lexicons - .get(&method) - .await - .ok_or_else(|| AppError::BadRequest(format!("method not found: {method}")))?; - - if lexicon.lexicon_type != LexiconType::Procedure { - return Err(AppError::BadRequest(format!( - "{method} is not a procedure endpoint" - ))); - } - - handle_procedure(&state, &method, &claims, &body, &lexicon).await -} - -// --------------------------------------------------------------------------- -// Generic query handler -// --------------------------------------------------------------------------- - -async fn handle_query( - state: &AppState, - method: &str, - params: &HashMap, - lexicon: &crate::lexicon::ParsedLexicon, -) -> Result { - // Single-record query: has a `uri` parameter - if let Some(uri) = params.get("uri") { - return handle_get_record(state, uri).await; - } - - // 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 limit: i64 = params - .get("limit") - .and_then(|l| l.parse().ok()) - .unwrap_or(20) - .min(100); - - let offset: i64 = params - .get("cursor") - .and_then(|c| c.parse().ok()) - .unwrap_or(0); - - let did = params.get("did"); - - let rows: Vec<(String, String, Value)> = if let Some(did) = did { - sqlx::query_as( - "SELECT uri, did, record FROM records WHERE collection = $1 AND did = $2 ORDER BY indexed_at DESC LIMIT $3 OFFSET $4", - ) - .bind(collection) - .bind(did) - .bind(limit) - .bind(offset) - .fetch_all(&state.db) - .await - .map_err(|e| AppError::Internal(format!("DB query failed: {e}")))? - } else { - sqlx::query_as( - "SELECT uri, did, record FROM records WHERE collection = $1 ORDER BY indexed_at DESC LIMIT $2 OFFSET $3", - ) - .bind(collection) - .bind(limit) - .bind(offset) - .fetch_all(&state.db) - .await - .map_err(|e| AppError::Internal(format!("DB query failed: {e}")))? - }; - - let has_next_page = rows.len() as i64 == limit; - - // Resolve PDS endpoints for blob URL enrichment. - 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 - { - pds_map.insert(did.to_string(), pds); - } - } - - let records: Vec = rows - .into_iter() - .map(|(uri, did, mut record)| { - if let Some(pds) = pds_map.get(&did) { - repo::enrich_media_blobs(&mut record, pds, &did); - } - record - .as_object_mut() - .map(|obj| obj.insert("uri".to_string(), json!(uri))); - record - }) - .collect(); - - let mut result = json!({ "records": records }); - if has_next_page { - let next_cursor = (offset + limit).to_string(); - result - .as_object_mut() - .unwrap() - .insert("cursor".to_string(), json!(next_cursor)); - } - - Ok(Json(result).into_response()) -} - -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 (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); - - record - .as_object_mut() - .map(|obj| obj.insert("uri".to_string(), json!(uri))); - - Ok(Json(json!({ "record": record })).into_response()) -} - -// --------------------------------------------------------------------------- -// Generic procedure handler -// --------------------------------------------------------------------------- - -async fn handle_procedure( - state: &AppState, - method: &str, - claims: &Claims, - 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 session = repo::get_atp_session(state, claims.token()).await?; - - // Determine create vs put based on whether input has a `uri` field. - let has_uri = input.get("uri").and_then(|v| v.as_str()).is_some(); - - if has_uri { - handle_put_record(state, claims, input, collection, &session).await - } else { - handle_create_record(state, claims, input, collection, &session).await - } -} - -async fn handle_create_record( - state: &AppState, - claims: &Claims, - input: &Value, - collection: &str, - session: &repo::AtpSession, -) -> Result { - // Build record from input, adding $type - let mut record = input.clone(); - if let Some(obj) = record.as_object_mut() { - obj.insert("$type".to_string(), json!(collection)); - // Remove fields that are procedure params, not record fields - obj.remove("shouldPublish"); - } - - let pds_body = json!({ - "repo": claims.did(), - "collection": collection, - "record": record, - }); - - let resp = - repo::pds_post_json_raw(state, session, "com.atproto.repo.createRecord", &pds_body).await?; - - if resp.status().is_success() { - let bytes = resp - .bytes() - .await - .map_err(|e| AppError::Internal(format!("failed to read PDS response: {e}")))?; - - let pds_result: Value = serde_json::from_slice(&bytes) - .map_err(|e| AppError::Internal(format!("invalid PDS JSON: {e}")))?; - - if let (Some(uri), Some(cid)) = ( - pds_result.get("uri").and_then(|v| v.as_str()), - pds_result.get("cid").and_then(|v| v.as_str()), - ) { - let rkey = uri.split('/').next_back().unwrap_or_default(); - let _ = sqlx::query( - r#" - INSERT INTO records (uri, did, collection, rkey, record, cid) - VALUES ($1, $2, $3, $4, $5, $6) - ON CONFLICT (uri) DO UPDATE - SET record = EXCLUDED.record, - cid = EXCLUDED.cid - "#, - ) - .bind(uri) - .bind(claims.did()) - .bind(collection) - .bind(rkey) - .bind(&record) - .bind(cid) - .execute(&state.db) - .await; - } - - Ok(( - StatusCode::OK, - [(axum::http::header::CONTENT_TYPE, "application/json")], - bytes, - ) - .into_response()) - } else { - repo::forward_pds_response(resp).await - } -} - -async fn handle_put_record( - state: &AppState, - claims: &Claims, - input: &Value, - collection: &str, - session: &repo::AtpSession, -) -> Result { - let uri = input - .get("uri") - .and_then(|v| v.as_str()) - .ok_or_else(|| AppError::BadRequest("missing uri field".into()))?; - - let rkey = uri - .split('/') - .next_back() - .ok_or_else(|| AppError::Internal("invalid AT URI".into()))?; - - // Build record from input, adding $type - let mut record = input.clone(); - if let Some(obj) = record.as_object_mut() { - obj.insert("$type".to_string(), json!(collection)); - obj.remove("uri"); - obj.remove("shouldPublish"); - } - - let pds_body = json!({ - "repo": claims.did(), - "collection": collection, - "rkey": rkey, - "record": record, - }); - - let resp = - repo::pds_post_json_raw(state, session, "com.atproto.repo.putRecord", &pds_body).await?; - - if resp.status().is_success() { - let bytes = resp - .bytes() - .await - .map_err(|e| AppError::Internal(format!("failed to read PDS response: {e}")))?; - - let pds_result: Value = serde_json::from_slice(&bytes) - .map_err(|e| AppError::Internal(format!("invalid PDS JSON: {e}")))?; - - let cid = pds_result - .get("cid") - .and_then(|v| v.as_str()) - .unwrap_or_default(); - - let _ = sqlx::query( - r#" - INSERT INTO records (uri, did, collection, rkey, record, cid) - VALUES ($1, $2, $3, $4, $5, $6) - ON CONFLICT (uri) DO UPDATE - SET record = EXCLUDED.record, - cid = EXCLUDED.cid, - indexed_at = NOW() - "#, - ) - .bind(uri) - .bind(claims.did()) - .bind(collection) - .bind(rkey) - .bind(&record) - .bind(cid) - .execute(&state.db) - .await; - - Ok(( - StatusCode::OK, - [(axum::http::header::CONTENT_TYPE, "application/json")], - bytes, - ) - .into_response()) - } else { - repo::forward_pds_response(resp).await - } -} diff --git a/src/xrpc/mod.rs b/src/xrpc/mod.rs new file mode 100644 index 0000000..09298d9 --- /dev/null +++ b/src/xrpc/mod.rs @@ -0,0 +1,55 @@ +mod procedure; +mod query; + +use axum::Json; +use axum::extract::{Path, Query, State}; +use axum::response::Response; +use std::collections::HashMap; + +use crate::AppState; +use crate::auth::Claims; +use crate::error::AppError; +use crate::lexicon::LexiconType; + +/// Catch-all GET handler for XRPC queries. +pub async fn xrpc_get( + State(state): State, + Path(method): Path, + Query(params): Query>, +) -> Result { + let lexicon = state + .lexicons + .get(&method) + .await + .ok_or_else(|| AppError::BadRequest(format!("method not found: {method}")))?; + + if lexicon.lexicon_type != LexiconType::Query { + return Err(AppError::BadRequest(format!( + "{method} is not a query endpoint" + ))); + } + + query::handle_query(&state, &method, ¶ms, &lexicon).await +} + +/// Catch-all POST handler for XRPC procedures. +pub async fn xrpc_post( + State(state): State, + Path(method): Path, + claims: Claims, + Json(body): Json, +) -> Result { + let lexicon = state + .lexicons + .get(&method) + .await + .ok_or_else(|| AppError::BadRequest(format!("method not found: {method}")))?; + + if lexicon.lexicon_type != LexiconType::Procedure { + return Err(AppError::BadRequest(format!( + "{method} is not a procedure endpoint" + ))); + } + + procedure::handle_procedure(&state, &method, &claims, &body, &lexicon).await +} diff --git a/src/xrpc/procedure.rs b/src/xrpc/procedure.rs new file mode 100644 index 0000000..fb401da --- /dev/null +++ b/src/xrpc/procedure.rs @@ -0,0 +1,178 @@ +use axum::http::StatusCode; +use axum::response::{IntoResponse, Response}; +use serde_json::{Value, json}; + +use crate::AppState; +use crate::auth::Claims; +use crate::error::AppError; +use crate::repo; + +pub(super) async fn handle_procedure( + state: &AppState, + method: &str, + claims: &Claims, + 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 session = repo::get_atp_session(state, claims.token()).await?; + + // Determine create vs put based on whether input has a `uri` field. + let has_uri = input.get("uri").and_then(|v| v.as_str()).is_some(); + + if has_uri { + handle_put_record(state, claims, input, collection, &session).await + } else { + handle_create_record(state, claims, input, collection, &session).await + } +} + +async fn handle_create_record( + state: &AppState, + claims: &Claims, + input: &Value, + collection: &str, + session: &repo::AtpSession, +) -> Result { + // Build record from input, adding $type + let mut record = input.clone(); + if let Some(obj) = record.as_object_mut() { + obj.insert("$type".to_string(), json!(collection)); + // Remove fields that are procedure params, not record fields + obj.remove("shouldPublish"); + } + + let pds_body = json!({ + "repo": claims.did(), + "collection": collection, + "record": record, + }); + + let resp = + repo::pds_post_json_raw(state, session, "com.atproto.repo.createRecord", &pds_body).await?; + + if resp.status().is_success() { + let bytes = resp + .bytes() + .await + .map_err(|e| AppError::Internal(format!("failed to read PDS response: {e}")))?; + + let pds_result: Value = serde_json::from_slice(&bytes) + .map_err(|e| AppError::Internal(format!("invalid PDS JSON: {e}")))?; + + if let (Some(uri), Some(cid)) = ( + pds_result.get("uri").and_then(|v| v.as_str()), + pds_result.get("cid").and_then(|v| v.as_str()), + ) { + let rkey = uri.split('/').next_back().unwrap_or_default(); + let _ = sqlx::query( + r#" + INSERT INTO records (uri, did, collection, rkey, record, cid) + VALUES ($1, $2, $3, $4, $5, $6) + ON CONFLICT (uri) DO UPDATE + SET record = EXCLUDED.record, + cid = EXCLUDED.cid + "#, + ) + .bind(uri) + .bind(claims.did()) + .bind(collection) + .bind(rkey) + .bind(&record) + .bind(cid) + .execute(&state.db) + .await; + } + + Ok(( + StatusCode::OK, + [(axum::http::header::CONTENT_TYPE, "application/json")], + bytes, + ) + .into_response()) + } else { + repo::forward_pds_response(resp).await + } +} + +async fn handle_put_record( + state: &AppState, + claims: &Claims, + input: &Value, + collection: &str, + session: &repo::AtpSession, +) -> Result { + let uri = input + .get("uri") + .and_then(|v| v.as_str()) + .ok_or_else(|| AppError::BadRequest("missing uri field".into()))?; + + let rkey = uri + .split('/') + .next_back() + .ok_or_else(|| AppError::Internal("invalid AT URI".into()))?; + + // Build record from input, adding $type + let mut record = input.clone(); + if let Some(obj) = record.as_object_mut() { + obj.insert("$type".to_string(), json!(collection)); + obj.remove("uri"); + obj.remove("shouldPublish"); + } + + let pds_body = json!({ + "repo": claims.did(), + "collection": collection, + "rkey": rkey, + "record": record, + }); + + let resp = + repo::pds_post_json_raw(state, session, "com.atproto.repo.putRecord", &pds_body).await?; + + if resp.status().is_success() { + let bytes = resp + .bytes() + .await + .map_err(|e| AppError::Internal(format!("failed to read PDS response: {e}")))?; + + let pds_result: Value = serde_json::from_slice(&bytes) + .map_err(|e| AppError::Internal(format!("invalid PDS JSON: {e}")))?; + + let cid = pds_result + .get("cid") + .and_then(|v| v.as_str()) + .unwrap_or_default(); + + let _ = sqlx::query( + r#" + INSERT INTO records (uri, did, collection, rkey, record, cid) + VALUES ($1, $2, $3, $4, $5, $6) + ON CONFLICT (uri) DO UPDATE + SET record = EXCLUDED.record, + cid = EXCLUDED.cid, + indexed_at = NOW() + "#, + ) + .bind(uri) + .bind(claims.did()) + .bind(collection) + .bind(rkey) + .bind(&record) + .bind(cid) + .execute(&state.db) + .await; + + Ok(( + StatusCode::OK, + [(axum::http::header::CONTENT_TYPE, "application/json")], + bytes, + ) + .into_response()) + } else { + repo::forward_pds_response(resp).await + } +} diff --git a/src/xrpc/query.rs b/src/xrpc/query.rs new file mode 100644 index 0000000..7e9fbba --- /dev/null +++ b/src/xrpc/query.rs @@ -0,0 +1,122 @@ +use axum::Json; +use axum::response::{IntoResponse, Response}; +use serde_json::{Value, json}; +use std::collections::{HashMap, HashSet}; + +use crate::AppState; +use crate::error::AppError; +use crate::profile; +use crate::repo; + +pub(super) async fn handle_query( + state: &AppState, + method: &str, + params: &HashMap, + lexicon: &crate::lexicon::ParsedLexicon, +) -> Result { + // Single-record query: has a `uri` parameter + if let Some(uri) = params.get("uri") { + return handle_get_record(state, uri).await; + } + + // 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 limit: i64 = params + .get("limit") + .and_then(|l| l.parse().ok()) + .unwrap_or(20) + .min(100); + + let offset: i64 = params + .get("cursor") + .and_then(|c| c.parse().ok()) + .unwrap_or(0); + + let did = params.get("did"); + + let rows: Vec<(String, String, Value)> = if let Some(did) = did { + sqlx::query_as( + "SELECT uri, did, record FROM records WHERE collection = $1 AND did = $2 ORDER BY indexed_at DESC LIMIT $3 OFFSET $4", + ) + .bind(collection) + .bind(did) + .bind(limit) + .bind(offset) + .fetch_all(&state.db) + .await + .map_err(|e| AppError::Internal(format!("DB query failed: {e}")))? + } else { + sqlx::query_as( + "SELECT uri, did, record FROM records WHERE collection = $1 ORDER BY indexed_at DESC LIMIT $2 OFFSET $3", + ) + .bind(collection) + .bind(limit) + .bind(offset) + .fetch_all(&state.db) + .await + .map_err(|e| AppError::Internal(format!("DB query failed: {e}")))? + }; + + let has_next_page = rows.len() as i64 == limit; + + // Resolve PDS endpoints for blob URL enrichment. + 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 + { + pds_map.insert(did.to_string(), pds); + } + } + + let records: Vec = rows + .into_iter() + .map(|(uri, did, mut record)| { + if let Some(pds) = pds_map.get(&did) { + repo::enrich_media_blobs(&mut record, pds, &did); + } + record + .as_object_mut() + .map(|obj| obj.insert("uri".to_string(), json!(uri))); + record + }) + .collect(); + + let mut result = json!({ "records": records }); + if has_next_page { + let next_cursor = (offset + limit).to_string(); + result + .as_object_mut() + .unwrap() + .insert("cursor".to_string(), json!(next_cursor)); + } + + Ok(Json(result).into_response()) +} + +pub(super) 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 (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); + + record + .as_object_mut() + .map(|obj| obj.insert("uri".to_string(), json!(uri))); + + Ok(Json(json!({ "record": record })).into_response()) +} -- 2.54.0