Something went wrong. Try again.
A lexicon-driven AppView for ATProto.
Something went wrong. Try again.
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188use axum::Json;use axum::extract::{Path, State};use axum::http::StatusCode;use serde_json::Value;
use crate::AppState;use crate::db::{adapt_sql, now_rfc3339};use crate::error::AppError;use crate::lexicon::{LexiconType, ParsedLexicon, ProcedureAction};use crate::resolve::{fetch_lexicon_from_pds, resolve_nsid_authority};
use super::auth::UserAuth;use super::permissions::Permission;use super::types::{AddNetworkLexiconBody, NetworkLexiconSummary};
/// Broadcast the current record collection list so the Jetstream/// subscription syncs its filter.async fn notify_collections(state: &AppState) { let collections = state.lexicons.get_record_collections().await; let _ = state.collections_tx.send(collections);}
/// POST /admin/network-lexicons — add a network lexicon to watch.pub(super) async fn add( State(state): State<AppState>, auth: UserAuth, Json(body): Json<AddNetworkLexiconBody>,) -> Result<(StatusCode, Json<Value>), AppError> { auth.require(Permission::LexiconsCreate).await?; let nsid = &body.nsid;
// Resolve NSID authority via DNS TXT lookup. let (authority_did, pds_endpoint) = resolve_nsid_authority(&state.http, &state.config.plc_url, nsid).await?;
// Fetch the lexicon from the authority's PDS. let lexicon_json = fetch_lexicon_from_pds(&state.http, &pds_endpoint, &authority_did, nsid).await?;
// Parse to validate. let parsed = ParsedLexicon::parse( lexicon_json.clone(), 1, body.target_collection.clone(), ProcedureAction::Upsert, None, None, None, ) .map_err(|e| AppError::BadRequest(format!("failed to parse lexicon: {e}")))?;
let backend = state.db_backend; let now = now_rfc3339(); let lexicon_json_str = serde_json::to_string(&lexicon_json).unwrap_or_default();
// Upsert into lexicons table with network source. let sql = adapt_sql( r#" INSERT INTO lexicons (id, lexicon_json, backfill, target_collection, source, authority_did, last_fetched_at, created_at) VALUES (?, ?, 0, ?, 'network', ?, ?, ?) ON CONFLICT (id) DO UPDATE SET lexicon_json = EXCLUDED.lexicon_json, target_collection = EXCLUDED.target_collection, source = 'network', authority_did = EXCLUDED.authority_did, last_fetched_at = ?, revision = lexicons.revision + 1, updated_at = ? RETURNING revision "#, backend, ); let row: (i32,) = sqlx::query_as(&sql) .bind(nsid) .bind(&lexicon_json_str) .bind(&body.target_collection) .bind(&authority_did) .bind(&now) .bind(&now) .bind(&now) .bind(&now) .fetch_one(&state.db) .await .map_err(|e| AppError::Internal(format!("failed to upsert network lexicon: {e}")))?;
let revision = row.0;
// Update in-memory registry. let is_record = parsed.lexicon_type == LexiconType::Record; let parsed = ParsedLexicon::parse( lexicon_json, revision, body.target_collection, ProcedureAction::Upsert, None, None, None, ) .map_err(|e| AppError::Internal(format!("failed to re-parse lexicon: {e}")))?; state.lexicons.upsert(parsed).await;
if is_record { notify_collections(&state).await; }
Ok(( StatusCode::CREATED, Json(serde_json::json!({ "nsid": nsid, "authority_did": authority_did, "revision": revision, })), ))}
/// GET /admin/network-lexicons — list tracked network lexicons.pub(super) async fn list( State(state): State<AppState>, auth: UserAuth,) -> Result<Json<Vec<NetworkLexiconSummary>>, AppError> { auth.require(Permission::LexiconsRead).await?;
let backend = state.db_backend; let sql = adapt_sql( "SELECT id, authority_did, target_collection, last_fetched_at, created_at FROM lexicons WHERE source = 'network' ORDER BY id", backend, ); #[allow(clippy::type_complexity)] let rows: Vec<( String, Option<String>, Option<String>, Option<String>, String, )> = sqlx::query_as(&sql) .fetch_all(&state.db) .await .map_err(|e| AppError::Internal(format!("failed to list network lexicons: {e}")))?;
let summaries: Vec<NetworkLexiconSummary> = rows .into_iter() .map( |(nsid, authority_did, target_collection, last_fetched_at, created_at)| { NetworkLexiconSummary { nsid, authority_did: authority_did.unwrap_or_default(), target_collection, last_fetched_at, created_at, } }, ) .collect();
Ok(Json(summaries))}
/// DELETE /admin/network-lexicons/{nsid} — stop watching a network lexicon.pub(super) async fn remove( State(state): State<AppState>, auth: UserAuth, Path(nsid): Path<String>,) -> Result<StatusCode, AppError> { auth.require(Permission::LexiconsDelete).await?;
let backend = state.db_backend; let sql = adapt_sql( "DELETE FROM lexicons WHERE id = ? AND source = 'network'", backend, ); let result = sqlx::query(&sql) .bind(&nsid) .execute(&state.db) .await .map_err(|e| AppError::Internal(format!("failed to delete network lexicon: {e}")))?;
if result.rows_affected() == 0 { return Err(AppError::NotFound(format!( "network lexicon '{nsid}' not found" ))); }
state.lexicons.remove(&nsid).await; notify_collections(&state).await;
Ok(StatusCode::NO_CONTENT)}