A lexicon-driven AppView for ATProto.
Something went wrong. Try again.
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404use axum::Json;use axum::extract::State;use axum::http::StatusCode;use serde::Deserialize;use serde_json::Value;use uuid::Uuid;
use crate::AppState;use crate::db::{adapt_sql, now_rfc3339};use crate::error::AppError;use crate::event_log::{EventLog, Severity, log_event};use crate::tap;
use super::auth::UserAuth;use super::permissions::Permission;use super::types::{BackfillJob, CreateBackfillBody};
// ---------------------------------------------------------------------------// Relay discovery (reused from old backfill module)// ---------------------------------------------------------------------------
#[derive(Deserialize)]struct ListReposResponse { repos: Vec<RepoEntry>, cursor: Option<String>,}
#[derive(Deserialize)]struct RepoEntry { did: String,}
/// Discover all DIDs that have records in `collection` via the relay's/// `com.atproto.sync.listReposByCollection` endpoint. Paginates until done.async fn list_repos_by_collection( http: &reqwest::Client, relay_url: &str, collection: &str,) -> Result<Vec<String>, String> { let base = relay_url.trim_end_matches('/'); let mut dids = Vec::new(); let mut cursor: Option<String> = None;
loop { let mut url = format!( "{base}/xrpc/com.atproto.sync.listReposByCollection?collection={collection}&limit=1000" ); if let Some(ref c) = cursor { url.push_str(&format!("&cursor={c}")); }
let resp = http .get(&url) .send() .await .map_err(|e| format!("relay request failed: {e}"))?;
if !resp.status().is_success() { return Err(format!("relay returned {}", resp.status())); }
let body: ListReposResponse = resp .json() .await .map_err(|e| format!("invalid relay response: {e}"))?;
let page_count = body.repos.len(); for repo in body.repos { dids.push(repo.did); }
match body.cursor { Some(c) if page_count > 0 => cursor = Some(c), _ => break, } }
Ok(dids)}
// ---------------------------------------------------------------------------// Admin handlers// ---------------------------------------------------------------------------
/// POST /admin/backfill — create a backfill job, discover repos, and add them to Tap.pub(super) async fn create_backfill( State(state): State<AppState>, admin: UserAuth, Json(body): Json<CreateBackfillBody>,) -> Result<(StatusCode, Json<Value>), AppError> { admin.require(Permission::BackfillCreate).await?; let backend = state.db_backend;
let now = now_rfc3339(); let job_id = Uuid::new_v4().to_string(); let sql = adapt_sql( "INSERT INTO backfill_jobs (id, collection, did, created_at) VALUES (?, ?, ?, ?) RETURNING id", backend, ); let row: (String,) = sqlx::query_as(&sql) .bind(&job_id) .bind(&body.collection) .bind(&body.did) .bind(&now) .fetch_one(&state.db) .await .map_err(|e| AppError::Internal(format!("failed to create backfill job: {e}")))?;
let job_id = row.0.clone();
let now = now_rfc3339(); let sql = adapt_sql( "UPDATE backfill_jobs SET status = 'running', started_at = ? WHERE id = ?", backend, ); let _ = sqlx::query(&sql) .bind(&now) .bind(&job_id) .execute(&state.db) .await;
log_event( &state.db, EventLog { event_type: "backfill.started".to_string(), severity: Severity::Info, actor_did: Some(admin.did.clone()), subject: body.collection.clone(), detail: serde_json::json!({ "job_id": job_id.clone(), }), }, backend, ) .await;
let collections: Vec<String> = if let Some(ref col) = body.collection { let lexicon_exists: bool = state .lexicons .get(col) .await .is_some_and(|lex| lex.lexicon_type == crate::lexicon::LexiconType::Record); if !lexicon_exists { let error = format!("no record-type lexicon registered for collection '{col}'"); let now = now_rfc3339(); let sql = adapt_sql( "UPDATE backfill_jobs SET status = 'failed', completed_at = ?, error = ? WHERE id = ?", backend, ); let _ = sqlx::query(&sql) .bind(&now) .bind(&error) .bind(&job_id) .execute(&state.db) .await;
return Ok(( StatusCode::CREATED, Json(serde_json::json!({ "id": job_id, "status": "failed", "error": error, })), )); } vec![col.clone()] } else { let sql = adapt_sql( "SELECT id FROM lexicons WHERE backfill = 1 AND json_extract(lexicon_json, '$.defs.main.type') = 'record'", backend, ); let rows: Vec<(String,)> = sqlx::query_as(&sql) .fetch_all(&state.db) .await .map_err(|e| { AppError::Internal(format!("failed to query backfill-eligible lexicons: {e}")) })?; rows.into_iter().map(|(id,)| id).collect() };
if collections.is_empty() { let now = now_rfc3339(); let sql = adapt_sql( "UPDATE backfill_jobs SET status = 'completed', completed_at = ?, error = 'no backfill-eligible collections' WHERE id = ?", backend, ); let _ = sqlx::query(&sql) .bind(&now) .bind(&job_id) .execute(&state.db) .await;
return Ok(( StatusCode::CREATED, Json(serde_json::json!({ "id": job_id, "status": "completed", "error": "no backfill-eligible collections", })), )); }
let mut all_dids = Vec::new();
for collection in &collections { let dids = if let Some(ref did) = body.did { vec![did.clone()] } else { match list_repos_by_collection(&state.http, &state.config.relay_url, collection).await { Ok(dids) => dids, Err(e) => { tracing::warn!(collection, error = %e, "failed to discover repos, skipping"); continue; } } };
all_dids.extend(dids); }
all_dids.sort(); all_dids.dedup();
let total_repos = all_dids.len() as i32;
let sql = adapt_sql( "UPDATE backfill_jobs SET total_repos = ? WHERE id = ?", backend, ); let _ = sqlx::query(&sql) .bind(total_repos) .bind(&job_id) .execute(&state.db) .await;
for chunk in all_dids.chunks(1000) { if let Err(e) = tap::remove_repos( &state.http, &state.config.tap_url, state.config.tap_admin_password.as_deref(), chunk, ) .await { tracing::warn!(error = %e, "failed to remove repos from tap, continuing"); } }
if !all_dids.is_empty() { for chunk in all_dids.chunks(1000) { if let Err(e) = tap::add_repos( &state.http, &state.config.tap_url, state.config.tap_admin_password.as_deref(), chunk, ) .await { tracing::warn!(error = %e, "failed to add repos to tap"); let now = now_rfc3339(); let sql = adapt_sql( "UPDATE backfill_jobs SET status = 'failed', completed_at = ?, error = ? WHERE id = ?", backend, ); let _ = sqlx::query(&sql) .bind(&now) .bind(&e) .bind(&job_id) .execute(&state.db) .await;
log_event( &state.db, EventLog { event_type: "backfill.failed".to_string(), severity: Severity::Error, actor_did: None, subject: body.collection.clone(), detail: serde_json::json!({ "job_id": job_id.clone(), "error": e, }), }, backend, ) .await;
return Ok(( StatusCode::CREATED, Json(serde_json::json!({ "id": job_id, "status": "failed", "error": e, })), )); } } }
let now = now_rfc3339(); let sql = adapt_sql( "UPDATE backfill_jobs SET status = 'completed', completed_at = ?, processed_repos = ? WHERE id = ?", backend, ); let _ = sqlx::query(&sql) .bind(&now) .bind(total_repos) .bind(&job_id) .execute(&state.db) .await;
log_event( &state.db, EventLog { event_type: "backfill.completed".to_string(), severity: Severity::Info, actor_did: None, subject: body.collection.clone(), detail: serde_json::json!({ "job_id": job_id.clone(), "total_repos": total_repos, }), }, backend, ) .await;
Ok(( StatusCode::CREATED, Json(serde_json::json!({ "id": job_id, "status": "completed", "total_repos": total_repos, })), ))}
/// GET /admin/backfill/status — list all backfill jobs.pub(super) async fn backfill_status( State(state): State<AppState>, auth: UserAuth,) -> Result<Json<Vec<BackfillJob>>, AppError> { auth.require(Permission::BackfillRead).await?; let backend = state.db_backend;
let sql = adapt_sql( "SELECT id, collection, did, status, total_repos, processed_repos, total_records, error, started_at, completed_at, created_at FROM backfill_jobs ORDER BY created_at DESC", backend, ); #[allow(clippy::type_complexity)] let rows: Vec<( String, Option<String>, Option<String>, String, Option<i32>, Option<i32>, Option<i32>, 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 backfill jobs: {e}")))?;
let jobs: Vec<BackfillJob> = 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))}