diff --git a/migrations/20260215000000_add_target_collection.sql b/migrations/20260215000000_add_target_collection.sql new file mode 100644 index 0000000..55fd8e7 --- /dev/null +++ b/migrations/20260215000000_add_target_collection.sql @@ -0,0 +1 @@ +ALTER TABLE lexicons ADD COLUMN target_collection TEXT; diff --git a/src/admin.rs b/src/admin.rs index 7296154..a980c12 100644 --- a/src/admin.rs +++ b/src/admin.rs @@ -88,6 +88,7 @@ struct UploadLexiconBody { lexicon_json: Value, #[serde(default = "default_backfill")] backfill: bool, + target_collection: Option, } fn default_backfill() -> bool { @@ -137,17 +138,18 @@ async fn upload_lexicon( .to_string(); // Validate it parses correctly - ParsedLexicon::parse(body.lexicon_json.clone(), 1) + 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) - VALUES ($1, $2, $3) + 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 @@ -156,6 +158,7 @@ async fn upload_lexicon( .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}")))?; @@ -163,7 +166,7 @@ async fn upload_lexicon( let revision = row.0; // Update in-memory registry with correct revision - let parsed = ParsedLexicon::parse(body.lexicon_json, 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; @@ -203,7 +206,7 @@ async fn list_lexicons( let summaries: Vec = rows .into_iter() .map(|(id, revision, json, backfill, created_at, updated_at)| { - let lexicon_type = ParsedLexicon::parse(json, revision) + let lexicon_type = ParsedLexicon::parse(json, revision, None) .map(|p| format!("{:?}", p.lexicon_type).to_lowercase()) .unwrap_or_else(|_| "unknown".into()); diff --git a/src/lexicon.rs b/src/lexicon.rs index 5ab9d18..e5bf7ee 100644 --- a/src/lexicon.rs +++ b/src/lexicon.rs @@ -37,11 +37,13 @@ pub struct ParsedLexicon { pub raw: Value, /// Database revision number. pub revision: i32, + /// For queries/procedures: the backing record collection NSID. + pub target_collection: Option, } impl ParsedLexicon { /// Parse a lexicon JSON document into a `ParsedLexicon`. - pub fn parse(raw: Value, revision: i32) -> Result { + pub fn parse(raw: Value, revision: i32, target_collection: Option) -> Result { let id = raw .get("id") .and_then(|v| v.as_str()) @@ -81,6 +83,7 @@ impl ParsedLexicon { record_schema, raw, revision, + target_collection, }) } } @@ -100,8 +103,8 @@ impl LexiconRegistry { /// Load all lexicons from the database, replacing any existing entries. pub async fn load_from_db(&self, db: &sqlx::PgPool) -> Result<(), String> { - let rows: Vec<(String, Value, i32)> = - sqlx::query_as("SELECT id, lexicon_json, revision FROM lexicons") + let rows: Vec<(String, Value, i32, Option)> = + sqlx::query_as("SELECT id, lexicon_json, revision, target_collection FROM lexicons") .fetch_all(db) .await .map_err(|e| format!("failed to load lexicons: {e}"))?; @@ -110,8 +113,8 @@ impl LexiconRegistry { inner.clear(); let mut loaded = 0u32; - for (id, json, revision) in rows { - match ParsedLexicon::parse(json, revision) { + for (id, json, revision, target_collection) in rows { + match ParsedLexicon::parse(json, revision, target_collection) { Ok(parsed) => { inner.insert(id, parsed); loaded += 1; diff --git a/src/main.rs b/src/main.rs index 4217a30..53e2483 100644 --- a/src/main.rs +++ b/src/main.rs @@ -7,6 +7,7 @@ mod lexicon; mod profile; mod repo; mod server; +mod xrpc; use config::Config; use lexicon::LexiconRegistry; diff --git a/src/repo.rs b/src/repo.rs index 0b96af6..7aa9c33 100644 --- a/src/repo.rs +++ b/src/repo.rs @@ -1,8 +1,7 @@ use axum::body::Bytes; -use axum::extract::{Query, State}; +use axum::extract::State; use axum::http::{HeaderMap, StatusCode}; use axum::response::{IntoResponse, Response}; -use axum::Json; use base64::Engine; use p256::pkcs8::EncodePrivateKey; use serde::Deserialize; @@ -11,17 +10,14 @@ use sha2::{Digest, Sha256}; use crate::auth::Claims; use crate::error::AppError; -use crate::profile; use crate::AppState; -const COLLECTION: &str = "games.gamesgamesgamesgames.game"; - // --------------------------------------------------------------------------- // AT URI parsing // --------------------------------------------------------------------------- /// Extract the DID from an AT URI (at://did/collection/rkey). -fn parse_did_from_at_uri(uri: &str) -> Result { +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()))?; @@ -38,17 +34,17 @@ fn parse_did_from_at_uri(uri: &str) -> Result { // --------------------------------------------------------------------------- #[derive(Deserialize)] -struct AtpSession { - access_token: String, - pds_endpoint: String, - dpop_jwk: DpopJwk, +pub(crate) struct AtpSession { + pub(crate) access_token: String, + pub(crate) pds_endpoint: String, + pub(crate) dpop_jwk: DpopJwk, } #[derive(Deserialize)] -struct DpopJwk { - x: String, - y: String, - d: String, +pub(crate) struct DpopJwk { + pub(crate) x: String, + pub(crate) y: String, + pub(crate) d: String, } // --------------------------------------------------------------------------- @@ -68,7 +64,7 @@ struct DpopClaims { } /// Fetch the user's AT Protocol session (PDS credentials) from AIP. -async fn get_atp_session(state: &AppState, token: &str) -> Result { +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('/') @@ -95,7 +91,7 @@ async fn get_atp_session(state: &AppState, token: &str) -> Result Result { +pub(crate) async fn forward_pds_response(resp: reqwest::Response) -> Result { let status = resp.status(); let body = resp .bytes() @@ -192,7 +188,7 @@ async fn forward_pds_response(resp: reqwest::Response) -> Result, - claims: Claims, - Json(input): Json, -) -> Result { - let session = get_atp_session(&state, claims.token()).await?; - - let now = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true); - - let mut record = json!({ - "$type": COLLECTION, - "name": input["name"], - "createdAt": now, - }); - - let rec = record.as_object_mut().unwrap(); - - for key in &[ - "summary", - "applicationType", - "genres", - "modes", - "themes", - "playerPerspectives", - "releases", - "media", - "parent", - ] { - if let Some(val) = input.get(*key) { - if !val.is_null() { - rec.insert((*key).to_string(), val.clone()); - } - } - } - - if input.get("shouldPublish").and_then(|v| v.as_bool()) == Some(true) { - rec.insert("publishedAt".to_string(), json!(now)); - } - - let pds_body = json!({ - "repo": claims.did(), - "collection": COLLECTION, - "record": record, - }); - - let resp = 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('/').last().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 { - forward_pds_response(resp).await - } -} - -pub async fn put_game( - State(state): State, - claims: Claims, - Json(input): Json, -) -> Result { - let uri = input - .get("uri") - .and_then(|v| v.as_str()) - .ok_or_else(|| AppError::Auth("missing uri field".into()))?; - - // Extract rkey from AT URI: at://did/collection/rkey - let rkey = uri - .split('/') - .last() - .ok_or_else(|| AppError::Internal("invalid AT URI".into()))?; - - let session = get_atp_session(&state, claims.token()).await?; - - let now = chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true); - - let created_at = input - .get("createdAt") - .and_then(|v| v.as_str()) - .unwrap_or(&now); - - let mut record = json!({ - "$type": COLLECTION, - "name": input["name"], - "createdAt": created_at, - }); - - let rec = record.as_object_mut().unwrap(); - - for key in &[ - "summary", - "applicationType", - "genres", - "modes", - "themes", - "playerPerspectives", - "releases", - "media", - "parent", - ] { - if let Some(val) = input.get(*key) { - if !val.is_null() { - rec.insert((*key).to_string(), val.clone()); - } - } - } - - if input.get("shouldPublish").and_then(|v| v.as_bool()) == Some(true) { - rec.insert("publishedAt".to_string(), json!(now)); - } - - let pds_body = json!({ - "repo": claims.did(), - "collection": COLLECTION, - "rkey": rkey, - "record": record, - }); - - let resp = 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 { - forward_pds_response(resp).await - } -} - pub async fn upload_blob( State(state): State, claims: Claims, @@ -529,140 +326,9 @@ pub async fn upload_blob( pds_post_blob(&state, &session, content_type, body).await } -// --------------------------------------------------------------------------- -// getGame (public, unauthenticated) -// --------------------------------------------------------------------------- - -#[derive(Deserialize)] -pub struct GetGameParams { - uri: String, -} - -pub async fn get_game( - State(state): State, - Query(params): Query, -) -> Result, AppError> { - let did = parse_did_from_at_uri(¶ms.uri)?; - - let row: Option<(Value,)> = sqlx::query_as( - "SELECT record FROM records WHERE uri = $1", - ) - .bind(¶ms.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("game record not found".into()))?; - - let pds = profile::resolve_pds_endpoint(&state.http, &did).await?; - enrich_media_blobs(&mut record, &pds, &did); - - record - .as_object_mut() - .unwrap() - .insert("uri".to_string(), json!(params.uri)); - - Ok(Json(json!({ "game": record }))) -} - -// --------------------------------------------------------------------------- -// listGames (public, unauthenticated) -// --------------------------------------------------------------------------- - -#[derive(Deserialize)] -pub struct ListGamesParams { - cursor: Option, - did: Option, - limit: Option, -} - -pub async fn list_games( - State(state): State, - Query(params): Query, -) -> Result, AppError> { - let limit = params.limit.unwrap_or(20).min(100); - let offset: i64 = params - .cursor - .as_deref() - .and_then(|c| c.parse().ok()) - .unwrap_or(0); - - let rows: Vec<(String, String, Value)> = if let Some(ref did) = params.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; - - // Collect unique DIDs and resolve their PDS endpoints for blob URL enrichment. - let unique_dids: std::collections::HashSet<&str> = rows.iter().map(|(_, did, _)| did.as_str()).collect(); - let mut pds_map: std::collections::HashMap = std::collections::HashMap::new(); - for did in unique_dids { - if let Ok(pds) = profile::resolve_pds_endpoint(&state.http, did).await { - pds_map.insert(did.to_string(), pds); - } - } - - let games: Vec = rows - .into_iter() - .filter_map(|(uri, did, record)| { - let mut record = record; - if let Some(pds) = pds_map.get(&did) { - enrich_media_blobs(&mut record, pds, &did); - } - - let name = record.get("name")?.as_str()?.to_string(); - let summary = record.get("summary").and_then(|s| s.as_str()).map(|s| s.to_string()); - let media = record.get("media").cloned(); - - let mut game = json!({ - "uri": uri, - "name": name, - }); - - let obj = game.as_object_mut().unwrap(); - if let Some(summary) = summary { - obj.insert("summary".to_string(), json!(summary)); - } - if let Some(media) = media { - obj.insert("media".to_string(), media); - } - - Some(game) - }) - .collect(); - - let mut result = json!({ "games": games }); - 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)) -} - /// Walk `media[]` and add a `url` field to each blob so the frontend can /// display images directly. -fn enrich_media_blobs(record: &mut Value, pds: &str, did: &str) { +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, diff --git a/src/server.rs b/src/server.rs index aa4605b..57ea5d7 100644 --- a/src/server.rs +++ b/src/server.rs @@ -9,6 +9,7 @@ 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 { @@ -16,26 +17,12 @@ pub fn router(state: AppState) -> Router { .route("/health", get(health)) .nest("/admin", admin::admin_routes(state.clone())) .route("/xrpc/app.bsky.actor.getProfile", get(get_profile)) - .route( - "/xrpc/games.gamesgamesgamesgames.createGame", - post(repo::create_game), - ) - .route( - "/xrpc/games.gamesgamesgamesgames.getGame", - get(repo::get_game), - ) - .route( - "/xrpc/games.gamesgamesgamesgames.listGames", - get(repo::list_games), - ) - .route( - "/xrpc/games.gamesgamesgamesgames.putGame", - post(repo::put_game), - ) .route( "/xrpc/com.atproto.repo.uploadBlob", post(repo::upload_blob).layer(DefaultBodyLimit::max(50 * 1024 * 1024)), ) + // Catch-all for dynamically registered lexicons + .route("/xrpc/{method}", get(xrpc::xrpc_get).post(xrpc::xrpc_post)) .layer(TraceLayer::new_for_http()) .layer(CorsLayer::permissive()) .with_state(state) diff --git a/src/xrpc.rs b/src/xrpc.rs new file mode 100644 index 0000000..da73ee6 --- /dev/null +++ b/src/xrpc.rs @@ -0,0 +1,357 @@ +use axum::extract::{Path, Query, State}; +use axum::http::StatusCode; +use axum::response::{IntoResponse, Response}; +use axum::Json; +use serde_json::{json, Value}; +use std::collections::{HashMap, HashSet}; + +use crate::auth::Claims; +use crate::error::AppError; +use crate::lexicon::LexiconType; +use crate::profile; +use crate::repo; +use crate::AppState; + +// --------------------------------------------------------------------------- +// 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, 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, &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('/').last().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('/') + .last() + .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 + } +}