Something went wrong. Try again.
A lexicon-driven AppView for ATProto.
Something went wrong. Try again.
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266use axum::http::StatusCode;use axum::response::{IntoResponse, Response};use serde_json::{Value, json};
use crate::AppState;use crate::auth::Claims;use crate::db::{adapt_sql, now_rfc3339};use crate::error::AppError;use crate::lexicon::ProcedureAction;use crate::record_refs::sync_refs;use crate::repo;
pub(super) async fn handle_procedure( state: &AppState, method: &str, claims: &Claims, input: &Value, params: &std::collections::HashMap<String, Value>, lexicon: &crate::lexicon::ParsedLexicon,) -> Result<Response, AppError> { if let Some(ref script) = lexicon.script { return crate::lua::execute_procedure_script( state, method, claims, input, params, lexicon, script, ) .await; }
let collection = lexicon.target_collection.as_deref().ok_or_else(|| { AppError::BadRequest(format!("{method} has no target_collection configured")) })?;
let session = repo::get_oauth_session(state, claims.did()).await?;
match &lexicon.action { ProcedureAction::Create => { handle_create_record(state, claims, input, collection, &session).await } ProcedureAction::Update => { handle_put_record(state, claims, input, collection, &session).await } ProcedureAction::Delete => { handle_delete_record(state, claims, input, collection, &session).await } ProcedureAction::Upsert => { // Backwards-compatible: sniff for `uri` field to decide create vs put. 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: &crate::HappyViewOAuthSession,) -> Result<Response, AppError> { // 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 backend = state.db_backend; let rkey = uri.split('/').next_back().unwrap_or_default(); let record_str = serde_json::to_string(&record).unwrap_or_default(); let sql = adapt_sql( r#" INSERT INTO records (uri, did, collection, rkey, record, cid, created_at) VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT (uri) DO UPDATE SET record = EXCLUDED.record, cid = EXCLUDED.cid "#, backend, ); let now = now_rfc3339(); let _ = sqlx::query(&sql) .bind(uri) .bind(claims.did()) .bind(collection) .bind(rkey) .bind(&record_str) .bind(cid) .bind(&now) .execute(&state.db) .await;
let _ = sync_refs(&state.db, uri, collection, &record, backend).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: &crate::HappyViewOAuthSession,) -> Result<Response, AppError> { 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 backend = state.db_backend; let record_str = serde_json::to_string(&record).unwrap_or_default(); let now = now_rfc3339(); let sql = adapt_sql( r#" INSERT INTO records (uri, did, collection, rkey, record, cid, created_at) VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT (uri) DO UPDATE SET record = EXCLUDED.record, cid = EXCLUDED.cid, indexed_at = ? "#, backend, ); let _ = sqlx::query(&sql) .bind(uri) .bind(claims.did()) .bind(collection) .bind(rkey) .bind(&record_str) .bind(cid) .bind(&now) .bind(&now) .execute(&state.db) .await;
let _ = sync_refs(&state.db, uri, collection, &record, backend).await;
Ok(( StatusCode::OK, [(axum::http::header::CONTENT_TYPE, "application/json")], bytes, ) .into_response()) } else { repo::forward_pds_response(resp).await }}
async fn handle_delete_record( state: &AppState, claims: &Claims, input: &Value, collection: &str, session: &crate::HappyViewOAuthSession,) -> Result<Response, AppError> { 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()))?;
let pds_body = json!({ "repo": claims.did(), "collection": collection, "rkey": rkey, });
let resp = repo::pds_post_json_raw(state, session, "com.atproto.repo.deleteRecord", &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 backend = state.db_backend; let sql = adapt_sql("DELETE FROM records WHERE uri = ?", backend); let _ = sqlx::query(&sql).bind(uri).execute(&state.db).await;
Ok(( StatusCode::OK, [(axum::http::header::CONTENT_TYPE, "application/json")], bytes, ) .into_response()) } else { repo::forward_pds_response(resp).await }}