From 33b56e5b226c4903d3538366fbc4a6313c5c8cfe Mon Sep 17 00:00:00 2001 From: Trezy Date: Wed, 13 May 2026 23:05:02 -0500 Subject: [PATCH 01/15] fix: add better reporting and proper resume on backfills Signed-off-by: Trezy --- .../20260513000000_add_backfill_stage.sql | 4 + .../20260513000001_create_backfill_repos.sql | 7 + .../20260513000000_add_backfill_stage.sql | 4 + .../20260513000001_create_backfill_repos.sql | 7 + src/admin/backfill.rs | 714 +++++++++++------- src/admin/mod.rs | 2 +- src/admin/types.rs | 25 +- src/main.rs | 2 + web/src/app/dashboard/backfill/page.tsx | 68 +- web/src/types/backfill.ts | 1 + 10 files changed, 553 insertions(+), 281 deletions(-) create mode 100644 migrations/postgres/20260513000000_add_backfill_stage.sql create mode 100644 migrations/postgres/20260513000001_create_backfill_repos.sql create mode 100644 migrations/sqlite/20260513000000_add_backfill_stage.sql create mode 100644 migrations/sqlite/20260513000001_create_backfill_repos.sql diff --git a/migrations/postgres/20260513000000_add_backfill_stage.sql b/migrations/postgres/20260513000000_add_backfill_stage.sql new file mode 100644 index 0000000..dc47b0f --- /dev/null +++ b/migrations/postgres/20260513000000_add_backfill_stage.sql @@ -0,0 +1,4 @@ +ALTER TABLE backfill_jobs ADD COLUMN stage TEXT NOT NULL DEFAULT 'pending'; + +UPDATE backfill_jobs SET stage = status WHERE status IN ('completed', 'failed'); +UPDATE backfill_jobs SET stage = 'failed', status = 'failed', error = 'interrupted by restart' WHERE status = 'running'; diff --git a/migrations/postgres/20260513000001_create_backfill_repos.sql b/migrations/postgres/20260513000001_create_backfill_repos.sql new file mode 100644 index 0000000..9c0d25b --- /dev/null +++ b/migrations/postgres/20260513000001_create_backfill_repos.sql @@ -0,0 +1,7 @@ +CREATE TABLE IF NOT EXISTS backfill_repos ( + job_id UUID NOT NULL REFERENCES backfill_jobs(id) ON DELETE CASCADE, + did TEXT NOT NULL, + pds_endpoint TEXT, + status TEXT NOT NULL DEFAULT 'pending', + PRIMARY KEY (job_id, did) +); diff --git a/migrations/sqlite/20260513000000_add_backfill_stage.sql b/migrations/sqlite/20260513000000_add_backfill_stage.sql new file mode 100644 index 0000000..dc47b0f --- /dev/null +++ b/migrations/sqlite/20260513000000_add_backfill_stage.sql @@ -0,0 +1,4 @@ +ALTER TABLE backfill_jobs ADD COLUMN stage TEXT NOT NULL DEFAULT 'pending'; + +UPDATE backfill_jobs SET stage = status WHERE status IN ('completed', 'failed'); +UPDATE backfill_jobs SET stage = 'failed', status = 'failed', error = 'interrupted by restart' WHERE status = 'running'; diff --git a/migrations/sqlite/20260513000001_create_backfill_repos.sql b/migrations/sqlite/20260513000001_create_backfill_repos.sql new file mode 100644 index 0000000..0e613eb --- /dev/null +++ b/migrations/sqlite/20260513000001_create_backfill_repos.sql @@ -0,0 +1,7 @@ +CREATE TABLE IF NOT EXISTS backfill_repos ( + job_id TEXT NOT NULL REFERENCES backfill_jobs(id) ON DELETE CASCADE, + did TEXT NOT NULL, + pds_endpoint TEXT, + status TEXT NOT NULL DEFAULT 'pending', + PRIMARY KEY (job_id, did) +); diff --git a/src/admin/backfill.rs b/src/admin/backfill.rs index 6b8ce1e..b7c62a5 100644 --- a/src/admin/backfill.rs +++ b/src/admin/backfill.rs @@ -22,7 +22,7 @@ use super::permissions::Permission; use super::types::{BackfillJob, CreateBackfillBody}; // --------------------------------------------------------------------------- -// Relay discovery (reused from old backfill module) +// Response types // --------------------------------------------------------------------------- #[derive(Deserialize)] @@ -36,10 +36,6 @@ struct RepoEntry { did: String, } -// --------------------------------------------------------------------------- -// PDS record types -// --------------------------------------------------------------------------- - #[derive(Deserialize)] struct ListRecordsResponse { records: Vec, @@ -53,15 +49,133 @@ struct RecordEntry { value: serde_json::Value, } -/// 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, +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +async fn set_stage(state: &AppState, job_id: &str, stage: &str) { + let sql = adapt_sql( + "UPDATE backfill_jobs SET stage = ? WHERE id = ?", + state.db_backend, + ); + let _ = sqlx::query(&sql) + .bind(stage) + .bind(job_id) + .execute(&state.db) + .await; +} + +async fn update_job_counter(state: &AppState, job_id: &str, column: &str, value: i32) { + let sql = adapt_sql( + &format!("UPDATE backfill_jobs SET {column} = ? WHERE id = ?"), + state.db_backend, + ); + let _ = sqlx::query(&sql) + .bind(value) + .bind(job_id) + .execute(&state.db) + .await; +} + +async fn count_repos(state: &AppState, job_id: &str) -> i32 { + let sql = adapt_sql( + "SELECT COUNT(*) FROM backfill_repos WHERE job_id = ?", + state.db_backend, + ); + sqlx::query_as::<_, (i32,)>(&sql) + .bind(job_id) + .fetch_one(&state.db) + .await + .map(|(c,)| c) + .unwrap_or(0) +} + +async fn cleanup_repos(state: &AppState, job_id: &str) { + let sql = adapt_sql( + "DELETE FROM backfill_repos WHERE job_id = ?", + state.db_backend, + ); + let _ = sqlx::query(&sql).bind(job_id).execute(&state.db).await; +} + +async fn fail_job(state: &AppState, job_id: &str, error: &str) { + let now = now_rfc3339(); + let sql = adapt_sql( + "UPDATE backfill_jobs SET status = 'failed', stage = 'failed', completed_at = ?, error = ? WHERE id = ?", + state.db_backend, + ); + let _ = sqlx::query(&sql) + .bind(&now) + .bind(error) + .bind(job_id) + .execute(&state.db) + .await; + cleanup_repos(state, job_id).await; +} + +async fn complete_job( + state: &AppState, + job_id: &str, + processed_repos: i32, + total_records: i32, + error: Option<&str>, +) { + let now = now_rfc3339(); + let sql = adapt_sql( + "UPDATE backfill_jobs SET status = 'completed', stage = 'completed', completed_at = ?, processed_repos = ?, total_records = ?, error = ? WHERE id = ?", + state.db_backend, + ); + let _ = sqlx::query(&sql) + .bind(&now) + .bind(processed_repos) + .bind(total_records) + .bind(error) + .bind(job_id) + .execute(&state.db) + .await; + cleanup_repos(state, job_id).await; +} + +// --------------------------------------------------------------------------- +// Phase 1: Discover repos via relay +// --------------------------------------------------------------------------- + +async fn run_discovery_phase( + state: &AppState, + job_id: &str, + collections: &[String], + specific_did: Option<&str>, +) { + set_stage(state, job_id, "discovering_repos").await; + + if let Some(did) = specific_did { + let sql = adapt_sql( + "INSERT INTO backfill_repos (job_id, did) VALUES (?, ?) ON CONFLICT DO NOTHING", + state.db_backend, + ); + let _ = sqlx::query(&sql) + .bind(job_id) + .bind(did) + .execute(&state.db) + .await; + } else { + for collection in collections { + if let Err(e) = discover_repos_from_relay(state, job_id, collection).await { + tracing::warn!(collection, error = %e, "failed to discover repos, skipping"); + } + } + } + + let total = count_repos(state, job_id).await; + update_job_counter(state, job_id, "total_repos", total).await; +} + +async fn discover_repos_from_relay( + state: &AppState, + job_id: &str, collection: &str, -) -> Result, String> { - let base = relay_url.trim_end_matches('/'); - let mut dids = Vec::new(); +) -> Result<(), String> { + let base = state.config.relay_url.trim_end_matches('/'); let mut cursor: Option = None; loop { @@ -72,7 +186,8 @@ async fn list_repos_by_collection( url.push_str(&format!("&cursor={c}")); } - let resp = http + let resp = state + .http .get(&url) .send() .await @@ -88,23 +203,210 @@ async fn list_repos_by_collection( .map_err(|e| format!("invalid relay response: {e}"))?; let page_count = body.repos.len(); - for repo in body.repos { - dids.push(repo.did); + + for repo in &body.repos { + let sql = adapt_sql( + "INSERT INTO backfill_repos (job_id, did) VALUES (?, ?) ON CONFLICT DO NOTHING", + state.db_backend, + ); + let _ = sqlx::query(&sql) + .bind(job_id) + .bind(&repo.did) + .execute(&state.db) + .await; } + let total = count_repos(state, job_id).await; + update_job_counter(state, job_id, "total_repos", total).await; + match body.cursor { Some(c) if page_count > 0 => cursor = Some(c), _ => break, } } - Ok(dids) + Ok(()) +} + +// --------------------------------------------------------------------------- +// Phase 2: Resolve PDS endpoints +// --------------------------------------------------------------------------- + +async fn run_resolution_phase(state: &AppState, job_id: &str) { + set_stage(state, job_id, "resolving_pds").await; + + let sql = adapt_sql( + "SELECT did FROM backfill_repos WHERE job_id = ? AND pds_endpoint IS NULL", + state.db_backend, + ); + let unresolved: Vec<(String,)> = sqlx::query_as(&sql) + .bind(job_id) + .fetch_all(&state.db) + .await + .unwrap_or_default(); + + let sql = adapt_sql( + "SELECT COUNT(*) FROM backfill_repos WHERE job_id = ? AND pds_endpoint IS NOT NULL", + state.db_backend, + ); + let already_resolved: i32 = sqlx::query_as::<_, (i32,)>(&sql) + .bind(job_id) + .fetch_one(&state.db) + .await + .map(|(c,)| c) + .unwrap_or(0); + + let mut resolved_count = already_resolved; + + for (did,) in &unresolved { + match profile::resolve_pds_endpoint(&state.http, &state.config.plc_url, did).await { + Ok(pds) => { + let sql = adapt_sql( + "UPDATE backfill_repos SET pds_endpoint = ? WHERE job_id = ? AND did = ?", + state.db_backend, + ); + let _ = sqlx::query(&sql) + .bind(&pds) + .bind(job_id) + .bind(did) + .execute(&state.db) + .await; + } + Err(e) => { + tracing::warn!(did, error = %e, "failed to resolve PDS endpoint, skipping DID"); + } + } + resolved_count += 1; + if resolved_count % 100 == 0 { + update_job_counter(state, job_id, "processed_repos", resolved_count).await; + } + } + + update_job_counter(state, job_id, "processed_repos", resolved_count).await; } // --------------------------------------------------------------------------- -// PDS record fetching +// Phase 3: Fetch records from PDS instances // --------------------------------------------------------------------------- +async fn run_fetching_phase(state: &AppState, job_id: &str, collections: &[String]) { + set_stage(state, job_id, "fetching_records").await; + + // Load pending repos grouped by PDS + let sql = adapt_sql( + "SELECT did, pds_endpoint FROM backfill_repos WHERE job_id = ? AND status = 'pending' AND pds_endpoint IS NOT NULL", + state.db_backend, + ); + let rows: Vec<(String, String)> = sqlx::query_as(&sql) + .bind(job_id) + .fetch_all(&state.db) + .await + .unwrap_or_default(); + + let mut pds_to_dids: HashMap> = HashMap::new(); + for (did, pds) in rows { + pds_to_dids.entry(pds).or_default().push(did); + } + + // Count already-completed repos for accurate progress + let sql = adapt_sql( + "SELECT COUNT(*) FROM backfill_repos WHERE job_id = ? AND status = 'completed'", + state.db_backend, + ); + let already_completed: i32 = sqlx::query_as::<_, (i32,)>(&sql) + .bind(job_id) + .fetch_one(&state.db) + .await + .map(|(c,)| c) + .unwrap_or(0); + + let processed_repos = Arc::new(AtomicI32::new(already_completed)); + let total_records = Arc::new(AtomicI32::new(0)); + let state = Arc::new(state.clone()); + let collections = Arc::new(collections.to_vec()); + let job_id_arc = Arc::new(job_id.to_string()); + + let pds_entries: Vec<(String, Vec)> = pds_to_dids.into_iter().collect(); + + stream::iter(pds_entries) + .for_each_concurrent(10, |(pds_endpoint, dids)| { + let state = Arc::clone(&state); + let collections = Arc::clone(&collections); + let processed_repos = Arc::clone(&processed_repos); + let total_records = Arc::clone(&total_records); + let job_id = Arc::clone(&job_id_arc); + + async move { + stream::iter(dids) + .for_each_concurrent(3, |did| { + let state = Arc::clone(&state); + let collections = Arc::clone(&collections); + let processed_repos = Arc::clone(&processed_repos); + let total_records = Arc::clone(&total_records); + let pds_endpoint = pds_endpoint.clone(); + let job_id = Arc::clone(&job_id); + + async move { + for collection in collections.iter() { + match fetch_records_from_pds( + &state, + &pds_endpoint, + &did, + collection, + ) + .await + { + Ok(count) => { + total_records + .fetch_add(count as i32, Ordering::Relaxed); + } + Err(e) => { + tracing::warn!( + did, + collection, + pds = %pds_endpoint, + error = %e, + "failed to fetch records from PDS" + ); + } + } + } + + // Mark DID as completed + let sql = adapt_sql( + "UPDATE backfill_repos SET status = 'completed' WHERE job_id = ? AND did = ?", + state.db_backend, + ); + let _ = sqlx::query(&sql) + .bind(job_id.as_str()) + .bind(&did) + .execute(&state.db) + .await; + + let repos = processed_repos.fetch_add(1, Ordering::Relaxed) + 1; + + if repos % 100 == 0 { + let records = total_records.load(Ordering::Relaxed); + let backend = state.db_backend; + let sql = adapt_sql( + "UPDATE backfill_jobs SET processed_repos = ?, total_records = ? WHERE id = ?", + backend, + ); + let _ = sqlx::query(&sql) + .bind(repos) + .bind(records) + .bind(job_id.as_str()) + .execute(&state.db) + .await; + } + } + }) + .await; + } + }) + .await; +} + /// Fetch all records for a given DID and collection from a PDS via /// `com.atproto.repo.listRecords`, paginating and handling rate limits. async fn fetch_records_from_pds( @@ -147,7 +449,7 @@ async fn fetch_records_from_pds( "rate limited by PDS, sleeping" ); tokio::time::sleep(tokio::time::Duration::from_secs(retry_after)).await; - continue; // retry same page + continue; } if !resp.status().is_success() { @@ -187,77 +489,31 @@ async fn fetch_records_from_pds( } // --------------------------------------------------------------------------- -// Admin handlers +// Background backfill worker // --------------------------------------------------------------------------- -/// POST /admin/backfill — create a backfill job and spawn background work. -pub(super) async fn create_backfill( - State(state): State, - admin: UserAuth, - Json(body): Json, -) -> Result<(StatusCode, Json), AppError> { - admin.require(Permission::BackfillCreate).await?; +async fn run_backfill_job(state: AppState, job_id: String) { let backend = state.db_backend; - let now = now_rfc3339(); - let job_id = Uuid::new_v4().to_string(); + // Load job metadata let sql = adapt_sql( - "INSERT INTO backfill_jobs (id, collection, did, status, started_at, created_at) VALUES (?, ?, ?, 'running', ?, ?) RETURNING id", + "SELECT collection, did, stage FROM backfill_jobs WHERE id = ?", backend, ); - let row: (String,) = sqlx::query_as(&sql) + let job: Option<(Option, Option, String)> = sqlx::query_as(&sql) .bind(&job_id) - .bind(&body.collection) - .bind(&body.did) - .bind(&now) - .bind(&now) - .fetch_one(&state.db) + .fetch_optional(&state.db) .await - .map_err(|e| AppError::Internal(format!("failed to create backfill job: {e}")))?; - - let job_id = row.0.clone(); - - 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; + .ok() + .flatten(); - // Clone what we need and spawn the background job - let spawn_state = state.clone(); - let spawn_job_id = job_id.clone(); - let spawn_body = body.clone(); - tokio::spawn(async move { - run_backfill_job(spawn_state, spawn_job_id, spawn_body).await; - }); - - Ok(( - StatusCode::CREATED, - Json(serde_json::json!({ - "id": job_id, - "status": "running", - })), - )) -} - -// --------------------------------------------------------------------------- -// Background backfill worker -// --------------------------------------------------------------------------- - -async fn run_backfill_job(state: AppState, job_id: String, body: CreateBackfillBody) { - let backend = state.db_backend; + let Some((collection, did, stage)) = job else { + tracing::error!(job_id, "backfill job not found"); + return; + }; // Determine target collections - let collections: Vec = if let Some(ref col) = body.collection { + let collections: Vec = if let Some(ref col) = collection { let lexicon_exists: bool = state .lexicons .get(col) @@ -297,157 +553,64 @@ async fn run_backfill_job(state: AppState, job_id: String, body: CreateBackfillB return; } - // Discover DIDs - 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); + // Run phases, skipping those already completed + if matches!(stage.as_str(), "pending" | "discovering_repos") { + run_discovery_phase(&state, &job_id, &collections, did.as_deref()).await; + + let total = count_repos(&state, &job_id).await; + if total == 0 { + complete_job(&state, &job_id, 0, 0, None).await; + log_event( + &state.db, + EventLog { + event_type: "backfill.completed".to_string(), + severity: Severity::Info, + actor_did: None, + subject: collection.clone(), + detail: serde_json::json!({ + "job_id": job_id, + "total_repos": 0, + "total_records": 0, + }), + }, + backend, + ) + .await; + return; + } } - all_dids.sort(); - all_dids.dedup(); + if matches!( + stage.as_str(), + "pending" | "discovering_repos" | "resolving_pds" + ) { + run_resolution_phase(&state, &job_id).await; + } - let total_repos = all_dids.len() as i32; + run_fetching_phase(&state, &job_id, &collections).await; - // Update total_repos in DB + // Read final counters from backfill_repos before cleanup let sql = adapt_sql( - "UPDATE backfill_jobs SET total_repos = ? WHERE id = ?", - backend, + "SELECT COUNT(*) FROM backfill_repos WHERE job_id = ? AND status = 'completed'", + state.db_backend, ); - let _ = sqlx::query(&sql) - .bind(total_repos) + let final_processed: i32 = sqlx::query_as::<_, (i32,)>(&sql) .bind(&job_id) - .execute(&state.db) - .await; - - if all_dids.is_empty() { - complete_job(&state, &job_id, 0, 0, None).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, - "total_repos": 0, - "total_records": 0, - }), - }, - backend, - ) - .await; - return; - } - - // Resolve DIDs to PDS endpoints and group by PDS - let mut pds_to_dids: HashMap> = HashMap::new(); - - for did in &all_dids { - match profile::resolve_pds_endpoint(&state.http, &state.config.plc_url, did).await { - Ok(pds) => { - pds_to_dids.entry(pds).or_default().push(did.clone()); - } - Err(e) => { - tracing::warn!(did, error = %e, "failed to resolve PDS endpoint, skipping DID"); - } - } - } - - let processed_repos = Arc::new(AtomicI32::new(0)); - let total_records = Arc::new(AtomicI32::new(0)); - - let state = Arc::new(state); - let collections = Arc::new(collections); - let job_id_arc = Arc::new(job_id.clone()); - - // Process PDSes with nested concurrency - let pds_entries: Vec<(String, Vec)> = pds_to_dids.into_iter().collect(); - - stream::iter(pds_entries) - .for_each_concurrent(10, |(pds_endpoint, dids)| { - let state = Arc::clone(&state); - let collections = Arc::clone(&collections); - let processed_repos = Arc::clone(&processed_repos); - let total_records = Arc::clone(&total_records); - let job_id = Arc::clone(&job_id_arc); - - async move { - stream::iter(dids) - .for_each_concurrent(3, |did| { - let state = Arc::clone(&state); - let collections = Arc::clone(&collections); - let processed_repos = Arc::clone(&processed_repos); - let total_records = Arc::clone(&total_records); - let pds_endpoint = pds_endpoint.clone(); - let job_id = Arc::clone(&job_id); - - async move { - for collection in collections.iter() { - match fetch_records_from_pds( - &state, - &pds_endpoint, - &did, - collection, - ) - .await - { - Ok(count) => { - total_records - .fetch_add(count as i32, Ordering::Relaxed); - } - Err(e) => { - tracing::warn!( - did, - collection, - pds = %pds_endpoint, - error = %e, - "failed to fetch records from PDS" - ); - } - } - } - - let repos = processed_repos.fetch_add(1, Ordering::Relaxed) + 1; - - // Update DB progress every 100 repos - if repos % 100 == 0 { - let records = total_records.load(Ordering::Relaxed); - let backend = state.db_backend; - let sql = adapt_sql( - "UPDATE backfill_jobs SET processed_repos = ?, total_records = ? WHERE id = ?", - backend, - ); - let _ = sqlx::query(&sql) - .bind(repos) - .bind(records) - .bind(job_id.as_str()) - .execute(&state.db) - .await; - } - } - }) - .await; - } - }) - .await; + .fetch_one(&state.db) + .await + .map(|(c,)| c) + .unwrap_or(0); - let final_processed = processed_repos.load(Ordering::Relaxed); - let final_records = total_records.load(Ordering::Relaxed); + let sql = adapt_sql( + "SELECT total_records FROM backfill_jobs WHERE id = ?", + state.db_backend, + ); + let final_records: i32 = sqlx::query_as::<_, (i32,)>(&sql) + .bind(&job_id) + .fetch_one(&state.db) + .await + .map(|(c,)| c) + .unwrap_or(0); complete_job(&state, &job_id, final_processed, final_records, None).await; @@ -457,7 +620,7 @@ async fn run_backfill_job(state: AppState, job_id: String, body: CreateBackfillB event_type: "backfill.completed".to_string(), severity: Severity::Info, actor_did: None, - subject: body.collection.clone(), + subject: collection, detail: serde_json::json!({ "job_id": job_id, "total_repos": final_processed, @@ -470,46 +633,64 @@ async fn run_backfill_job(state: AppState, job_id: String, body: CreateBackfillB } // --------------------------------------------------------------------------- -// Helper functions +// Admin handlers // --------------------------------------------------------------------------- -async fn fail_job(state: &AppState, job_id: &str, error: &str) { - let now = now_rfc3339(); +/// POST /admin/backfill — create a backfill job and spawn background work. +pub(super) async fn create_backfill( + State(state): State, + admin: UserAuth, + Json(body): Json, +) -> Result<(StatusCode, Json), 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( - "UPDATE backfill_jobs SET status = 'failed', completed_at = ?, error = ? WHERE id = ?", + "INSERT INTO backfill_jobs (id, collection, did, status, stage, started_at, created_at) VALUES (?, ?, ?, 'running', 'pending', ?, ?) RETURNING id", backend, ); - let _ = sqlx::query(&sql) + let row: (String,) = sqlx::query_as(&sql) + .bind(&job_id) + .bind(&body.collection) + .bind(&body.did) .bind(&now) - .bind(error) - .bind(job_id) - .execute(&state.db) - .await; -} + .bind(&now) + .fetch_one(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to create backfill job: {e}")))?; -async fn complete_job( - state: &AppState, - job_id: &str, - processed_repos: i32, - total_records: i32, - error: Option<&str>, -) { - let now = now_rfc3339(); - let backend = state.db_backend; + let job_id = row.0.clone(); - let sql = adapt_sql( - "UPDATE backfill_jobs SET status = 'completed', completed_at = ?, processed_repos = ?, total_records = ?, error = ? WHERE id = ?", + 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, - ); - let _ = sqlx::query(&sql) - .bind(&now) - .bind(processed_repos) - .bind(total_records) - .bind(error) - .bind(job_id) - .execute(&state.db) - .await; + ) + .await; + + let spawn_state = state.clone(); + let spawn_job_id = job_id.clone(); + tokio::spawn(async move { + run_backfill_job(spawn_state, spawn_job_id).await; + }); + + Ok(( + StatusCode::CREATED, + Json(serde_json::json!({ + "id": job_id, + "status": "running", + })), + )) } /// GET /admin/backfill/status — list all backfill jobs. @@ -521,7 +702,7 @@ pub(super) async fn backfill_status( 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", + "SELECT id, collection, did, status, stage, 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)] @@ -530,6 +711,7 @@ pub(super) async fn backfill_status( Option, Option, String, + String, Option, Option, Option, @@ -550,6 +732,7 @@ pub(super) async fn backfill_status( collection, did, status, + stage, total_repos, processed_repos, total_records, @@ -563,6 +746,7 @@ pub(super) async fn backfill_status( collection, did, status, + stage, total_repos, processed_repos, total_records, @@ -577,3 +761,27 @@ pub(super) async fn backfill_status( Ok(Json(jobs)) } + +// --------------------------------------------------------------------------- +// Startup resumption +// --------------------------------------------------------------------------- + +/// Resume any backfill jobs that were running when the server last stopped. +pub async fn resume_backfill_jobs(state: &AppState) { + let sql = adapt_sql( + "SELECT id FROM backfill_jobs WHERE status = 'running'", + state.db_backend, + ); + let rows: Vec<(String,)> = sqlx::query_as(&sql) + .fetch_all(&state.db) + .await + .unwrap_or_default(); + + for (job_id,) in rows { + tracing::info!(job_id, "resuming interrupted backfill job"); + let spawn_state = state.clone(); + tokio::spawn(async move { + run_backfill_job(spawn_state, job_id).await; + }); + } +} diff --git a/src/admin/mod.rs b/src/admin/mod.rs index 6a65c66..fb308f4 100644 --- a/src/admin/mod.rs +++ b/src/admin/mod.rs @@ -1,7 +1,7 @@ mod api_clients; mod api_keys; pub(crate) mod auth; -mod backfill; +pub mod backfill; mod dead_letters; mod domains; mod events; diff --git a/src/admin/types.rs b/src/admin/types.rs index c75bcaf..4871b2e 100644 --- a/src/admin/types.rs +++ b/src/admin/types.rs @@ -69,18 +69,19 @@ pub(super) struct CreateBackfillBody { } #[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: String, +pub(crate) struct BackfillJob { + pub(crate) id: String, + pub(crate) collection: Option, + pub(crate) did: Option, + pub(crate) status: String, + pub(crate) stage: String, + pub(crate) total_repos: Option, + pub(crate) processed_repos: Option, + pub(crate) total_records: Option, + pub(crate) error: Option, + pub(crate) started_at: Option, + pub(crate) completed_at: Option, + pub(crate) created_at: String, } // --------------------------------------------------------------------------- diff --git a/src/main.rs b/src/main.rs index 33b998a..16f92fe 100644 --- a/src/main.rs +++ b/src/main.rs @@ -644,6 +644,8 @@ async fn main() { state.db_backend, )); + happyview::admin::backfill::resume_backfill_jobs(&state).await; + let app = server::router(state); let addr = config.listen_addr(); diff --git a/web/src/app/dashboard/backfill/page.tsx b/web/src/app/dashboard/backfill/page.tsx index 74585b5..89e92bc 100644 --- a/web/src/app/dashboard/backfill/page.tsx +++ b/web/src/app/dashboard/backfill/page.tsx @@ -77,8 +77,7 @@ export default function BackfillPage() { ID Collection DID - Progress - Records + Status Started @@ -86,7 +85,7 @@ export default function BackfillPage() { {jobs.length === 0 && ( No backfill jobs yet. @@ -105,12 +104,7 @@ export default function BackfillPage() { {job.did ?? "All"} - {job.processed_repos != null && job.total_repos != null - ? `${job.processed_repos} / ${job.total_repos}` - : "--"} - - - {job.total_records?.toLocaleString() ?? "--"} + {job.started_at @@ -127,11 +121,51 @@ export default function BackfillPage() { ); } -function CreateDialog({ - onSuccess, -}: { - onSuccess: () => void; -}) { +function StageDisplay({ job }: { job: BackfillJob }) { + const repos = job.total_repos?.toLocaleString() ?? "0"; + const processed = job.processed_repos?.toLocaleString() ?? "0"; + const records = job.total_records?.toLocaleString() ?? "0"; + + switch (job.stage) { + case "pending": + return Pending; + case "discovering_repos": + return ( + + Discovering repos… +
{repos} found + + ); + case "resolving_pds": + return ( + + Resolving PDS… ({processed} / {repos}) + + ); + case "fetching_records": + return ( + + Fetching records… ({processed} / {repos} repos, {records} records) + + ); + case "completed": + return ( + + Completed — {repos} repos, {records} records + + ); + case "failed": + return ( + + Failed{job.error ? ` — ${job.error}` : ""} + + ); + default: + return {job.stage}; + } +} + +function CreateDialog({ onSuccess }: { onSuccess: () => void }) { const [collection, setCollection] = useState(null); const [did, setDid] = useState(""); const [error, setError] = useState(null); @@ -177,7 +211,11 @@ function CreateDialog({ { const target = e.target as HTMLElement; - if (target.closest("[data-slot='combobox-item'], [data-slot='combobox-content']")) { + if ( + target.closest( + "[data-slot='combobox-item'], [data-slot='combobox-content']", + ) + ) { e.preventDefault(); } }} diff --git a/web/src/types/backfill.ts b/web/src/types/backfill.ts index f65ea7c..5fea8a7 100644 --- a/web/src/types/backfill.ts +++ b/web/src/types/backfill.ts @@ -3,6 +3,7 @@ export interface BackfillJob { collection: string | null did: string | null status: string + stage: string total_repos: number | null processed_repos: number | null total_records: number | null -- 2.51.2 From a75135214de9c066f0064b8c00040cbebdda4d96 Mon Sep 17 00:00:00 2001 From: Trezy Date: Wed, 13 May 2026 23:15:26 -0500 Subject: [PATCH 02/15] fix: add better retry logic for backfills Signed-off-by: Trezy --- src/admin/backfill.rs | 64 +++++++++++++++++++++++++++++-------------- src/profile.rs | 42 ++++++++++++++++++++++++---- 2 files changed, 81 insertions(+), 25 deletions(-) diff --git a/src/admin/backfill.rs b/src/admin/backfill.rs index b7c62a5..026da56 100644 --- a/src/admin/backfill.rs +++ b/src/admin/backfill.rs @@ -17,6 +17,30 @@ use crate::event_log::{EventLog, Severity, log_event}; use crate::profile; use crate::record_handler::{self, RecordEvent}; +/// Parse rate-limit sleep duration from response headers. +/// Checks `RateLimit-Reset` (Unix timestamp, used by XRPC servers) first, +/// then `retry-after` (seconds), defaulting to 5s. +fn parse_retry_after(headers: &reqwest::header::HeaderMap) -> u64 { + if let Some(reset) = headers + .get("ratelimit-reset") + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.parse::().ok()) + { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() as i64; + let wait = (reset - now).max(1) as u64; + return wait.min(120); + } + + headers + .get("retry-after") + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.parse::().ok()) + .unwrap_or(5) +} + use super::auth::UserAuth; use super::permissions::Permission; use super::types::{BackfillJob, CreateBackfillBody}; @@ -186,12 +210,23 @@ async fn discover_repos_from_relay( url.push_str(&format!("&cursor={c}")); } - let resp = state - .http - .get(&url) - .send() - .await - .map_err(|e| format!("relay request failed: {e}"))?; + let resp = loop { + let r = state + .http + .get(&url) + .send() + .await + .map_err(|e| format!("relay request failed: {e}"))?; + + if r.status() == reqwest::StatusCode::TOO_MANY_REQUESTS { + let wait = parse_retry_after(r.headers()); + tracing::warn!(collection, wait, "rate limited by relay, sleeping"); + tokio::time::sleep(tokio::time::Duration::from_secs(wait)).await; + continue; + } + + break r; + }; if !resp.status().is_success() { return Err(format!("relay returned {}", resp.status())); @@ -434,21 +469,10 @@ async fn fetch_records_from_pds( .await .map_err(|e| format!("PDS request failed: {e}"))?; - // Handle rate limiting if resp.status() == reqwest::StatusCode::TOO_MANY_REQUESTS { - let retry_after = resp - .headers() - .get("retry-after") - .and_then(|v| v.to_str().ok()) - .and_then(|v| v.parse::().ok()) - .unwrap_or(5); - tracing::warn!( - did, - collection, - retry_after, - "rate limited by PDS, sleeping" - ); - tokio::time::sleep(tokio::time::Duration::from_secs(retry_after)).await; + let wait = parse_retry_after(resp.headers()); + tracing::warn!(did, collection, wait, "rate limited by PDS, sleeping"); + tokio::time::sleep(tokio::time::Duration::from_secs(wait)).await; continue; } diff --git a/src/profile.rs b/src/profile.rs index a31bede..25bf03b 100644 --- a/src/profile.rs +++ b/src/profile.rs @@ -2,6 +2,27 @@ use serde::{Deserialize, Serialize}; use crate::error::AppError; +fn parse_retry_after(headers: &reqwest::header::HeaderMap) -> u64 { + if let Some(reset) = headers + .get("ratelimit-reset") + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.parse::().ok()) + { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() as i64; + let wait = (reset - now).max(1) as u64; + return wait.min(120); + } + + headers + .get("retry-after") + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.parse::().ok()) + .unwrap_or(5) +} + #[derive(Serialize)] pub struct Profile { pub did: String, @@ -140,11 +161,22 @@ pub async fn resolve_did_document( format!("{}/{did}", plc_url.trim_end_matches('/')) }; - let resp = http - .get(&url) - .send() - .await - .map_err(|e| AppError::Internal(format!("DID resolution failed: {e}")))?; + let resp = loop { + let r = http + .get(&url) + .send() + .await + .map_err(|e| AppError::Internal(format!("DID resolution failed: {e}")))?; + + if r.status() == reqwest::StatusCode::TOO_MANY_REQUESTS { + let wait = parse_retry_after(r.headers()); + tracing::warn!(did, wait, "rate limited during DID resolution, sleeping"); + tokio::time::sleep(tokio::time::Duration::from_secs(wait)).await; + continue; + } + + break r; + }; if !resp.status().is_success() { return Err(AppError::NotFound(format!( -- 2.51.2 From 47ee0b0583a0f66ea9a133d019c4411317618b22 Mon Sep 17 00:00:00 2001 From: Trezy Date: Thu, 14 May 2026 07:25:55 -0500 Subject: [PATCH 03/15] fix: better icons for stages Signed-off-by: Trezy --- web/src/app/dashboard/backfill/page.tsx | 258 +++++++++++++++++++----- 1 file changed, 213 insertions(+), 45 deletions(-) diff --git a/web/src/app/dashboard/backfill/page.tsx b/web/src/app/dashboard/backfill/page.tsx index 89e92bc..4f7d91d 100644 --- a/web/src/app/dashboard/backfill/page.tsx +++ b/web/src/app/dashboard/backfill/page.tsx @@ -5,7 +5,9 @@ import { useCallback, useEffect, useState } from "react"; import { useCurrentUser } from "@/hooks/use-current-user"; import { createBackfillJob, getBackfillJobs, getLexicons } from "@/lib/api"; import type { BackfillJob } from "@/types/backfill"; +import { CheckCircle2, Circle, Loader2 } from "lucide-react"; import { SiteHeader } from "@/components/site-header"; +import { Badge } from "@/components/ui/badge"; import { Button } from "@/components/ui/button"; import { Combobox, @@ -27,6 +29,12 @@ import { } from "@/components/ui/responsive-dialog"; import { Input } from "@/components/ui/input"; import { Label } from "@/components/ui/label"; +import { + Sheet, + SheetContent, + SheetHeader, + SheetTitle, +} from "@/components/ui/sheet"; import { Table, TableBody, @@ -36,10 +44,55 @@ import { TableRow, } from "@/components/ui/table"; +const STAGES = [ + "pending", + "discovering_repos", + "resolving_pds", + "fetching_records", + "completed", + "failed", +] as const; + +const STAGE_LABELS: Record = { + pending: "Pending", + discovering_repos: "Discovering repos", + resolving_pds: "Resolving PDS", + fetching_records: "Fetching records", + completed: "Completed", + failed: "Failed", +}; + +function stageBadge(stage: string) { + switch (stage) { + case "completed": + return ( + + completed + + ); + case "failed": + return failed; + case "pending": + return pending; + default: + return ( + + {STAGE_LABELS[stage] ?? stage} + + ); + } +} + +function stageIndex(stage: string): number { + const idx = STAGES.indexOf(stage as (typeof STAGES)[number]); + return idx === -1 ? 0 : idx; +} + export default function BackfillPage() { const { hasPermission } = useCurrentUser(); const [jobs, setJobs] = useState([]); const [error, setError] = useState(null); + const [selectedJobId, setSelectedJobId] = useState(null); const load = useCallback(() => { getBackfillJobs() @@ -51,12 +104,13 @@ export default function BackfillPage() { load(); }, [load]); - // Auto-refresh every 5 seconds useEffect(() => { const interval = setInterval(load, 5000); return () => clearInterval(interval); }, [load]); + const selectedJob = jobs.find((j) => j.id === selectedJobId) ?? null; + return ( <> @@ -93,7 +147,11 @@ export default function BackfillPage() { )} {jobs.map((job) => ( - + setSelectedJobId(job.id)} + > {job.id.slice(0, 8)} @@ -103,9 +161,7 @@ export default function BackfillPage() { {job.did ?? "All"} - - - + {stageBadge(job.stage)} {job.started_at ? new Date(job.started_at).toLocaleString() @@ -116,53 +172,165 @@ export default function BackfillPage() { + + { + if (!open) setSelectedJobId(null); + }} + > + + {selectedJob && } + + ); } -function StageDisplay({ job }: { job: BackfillJob }) { - const repos = job.total_repos?.toLocaleString() ?? "0"; - const processed = job.processed_repos?.toLocaleString() ?? "0"; - const records = job.total_records?.toLocaleString() ?? "0"; +function JobDetail({ job }: { job: BackfillJob }) { + const current = stageIndex(job.stage); - switch (job.stage) { - case "pending": - return Pending; - case "discovering_repos": - return ( - - Discovering repos… -
{repos} found - - ); - case "resolving_pds": - return ( - - Resolving PDS… ({processed} / {repos}) - - ); - case "fetching_records": - return ( - - Fetching records… ({processed} / {repos} repos, {records} records) - - ); - case "completed": - return ( - - Completed — {repos} repos, {records} records - - ); - case "failed": - return ( - - Failed{job.error ? ` — ${job.error}` : ""} + return ( + <> + + + Backfill Details + + +
+
+
+ Job ID +

{job.id}

+
+
+ Collection +

{job.collection ?? "All"}

+
+
+ DID +

{job.did ?? "All"}

+
+
+ Created +

+ {new Date(job.created_at).toLocaleString()} +

+
+
+ Started +

+ {job.started_at + ? new Date(job.started_at).toLocaleString() + : "--"} +

+
+ {job.completed_at && ( +
+ Completed +

+ {new Date(job.completed_at).toLocaleString()} +

+
+ )} +
+ + {job.error && ( +
+ Error +
+ {job.error} +
+
+ )} + +
+ Progress +
+ = stageIndex("discovering_repos")} + value={job.total_repos?.toLocaleString()} + suffix="repos found" + /> + = stageIndex("resolving_pds")} + value={ + current >= stageIndex("resolving_pds") + ? `${job.processed_repos?.toLocaleString() ?? "0"} / ${job.total_repos?.toLocaleString() ?? "0"}` + : undefined + } + suffix="resolved" + /> + = stageIndex("fetching_records")} + value={ + current >= stageIndex("fetching_records") + ? `${job.processed_repos?.toLocaleString() ?? "0"} / ${job.total_repos?.toLocaleString() ?? "0"} repos` + : undefined + } + suffix={ + current >= stageIndex("fetching_records") + ? `${job.total_records?.toLocaleString() ?? "0"} records` + : undefined + } + /> +
+
+
+ + ); +} + +function ProgressRow({ + label, + active, + reached, + value, + suffix, +}: { + label: string; + active: boolean; + reached: boolean; + value?: string; + suffix?: string; +}) { + const done = reached && !active; + + return ( +
+ + {active ? ( + + ) : done ? ( + + ) : ( + + )} + + {label} + {reached && value && ( + + {value} + {suffix ? ` · ${suffix}` : ""} - ); - default: - return {job.stage}; - } + )} +
+ ); } function CreateDialog({ onSuccess }: { onSuccess: () => void }) { -- 2.51.2 From 5a3a4b92b966cadf33efe21b3c8d62c1ba2b6c47 Mon Sep 17 00:00:00 2001 From: Trezy Date: Thu, 14 May 2026 08:25:25 -0500 Subject: [PATCH 04/15] feat: allow backfills to be cancelled Signed-off-by: Trezy --- src/admin/backfill.rs | 147 ++++++++++++++++++++++-- src/admin/mod.rs | 1 + web/src/app/dashboard/backfill/page.tsx | 68 ++++++++++- web/src/lib/api.ts | 7 ++ 4 files changed, 210 insertions(+), 13 deletions(-) diff --git a/src/admin/backfill.rs b/src/admin/backfill.rs index 026da56..633096a 100644 --- a/src/admin/backfill.rs +++ b/src/admin/backfill.rs @@ -1,9 +1,9 @@ use std::collections::HashMap; use std::sync::Arc; -use std::sync::atomic::{AtomicI32, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicI32, Ordering}; use axum::Json; -use axum::extract::State; +use axum::extract::{Path, State}; use axum::http::StatusCode; use futures_util::stream::{self, StreamExt}; use serde::Deserialize; @@ -137,6 +137,42 @@ async fn fail_job(state: &AppState, job_id: &str, error: &str) { cleanup_repos(state, job_id).await; } +async fn is_cancelled(state: &AppState, job_id: &str) -> bool { + let sql = adapt_sql( + "SELECT status FROM backfill_jobs WHERE id = ?", + state.db_backend, + ); + sqlx::query_as::<_, (String,)>(&sql) + .bind(job_id) + .fetch_optional(&state.db) + .await + .ok() + .flatten() + .is_some_and(|(status,)| status == "cancelling") +} + +async fn request_cancel(state: &AppState, job_id: &str) { + let sql = adapt_sql( + "UPDATE backfill_jobs SET status = 'cancelling' WHERE id = ? AND status = 'running'", + state.db_backend, + ); + let _ = sqlx::query(&sql).bind(job_id).execute(&state.db).await; +} + +async fn finalise_cancel(state: &AppState, job_id: &str) { + let now = now_rfc3339(); + let sql = adapt_sql( + "UPDATE backfill_jobs SET status = 'cancelled', stage = 'cancelled', completed_at = ?, error = 'cancelled by user' WHERE id = ?", + state.db_backend, + ); + let _ = sqlx::query(&sql) + .bind(&now) + .bind(job_id) + .execute(&state.db) + .await; + cleanup_repos(state, job_id).await; +} + async fn complete_job( state: &AppState, job_id: &str, @@ -184,6 +220,9 @@ async fn run_discovery_phase( .await; } else { for collection in collections { + if is_cancelled(state, job_id).await { + return; + } if let Err(e) = discover_repos_from_relay(state, job_id, collection).await { tracing::warn!(collection, error = %e, "failed to discover repos, skipping"); } @@ -254,6 +293,10 @@ async fn discover_repos_from_relay( let total = count_repos(state, job_id).await; update_job_counter(state, job_id, "total_repos", total).await; + if is_cancelled(state, job_id).await { + return Ok(()); + } + match body.cursor { Some(c) if page_count > 0 => cursor = Some(c), _ => break, @@ -314,6 +357,9 @@ async fn run_resolution_phase(state: &AppState, job_id: &str) { resolved_count += 1; if resolved_count % 100 == 0 { update_job_counter(state, job_id, "processed_repos", resolved_count).await; + if is_cancelled(state, job_id).await { + return; + } } } @@ -357,6 +403,7 @@ async fn run_fetching_phase(state: &AppState, job_id: &str, collections: &[Strin let processed_repos = Arc::new(AtomicI32::new(already_completed)); let total_records = Arc::new(AtomicI32::new(0)); + let cancelled = Arc::new(AtomicBool::new(false)); let state = Arc::new(state.clone()); let collections = Arc::new(collections.to_vec()); let job_id_arc = Arc::new(job_id.to_string()); @@ -369,6 +416,7 @@ async fn run_fetching_phase(state: &AppState, job_id: &str, collections: &[Strin let collections = Arc::clone(&collections); let processed_repos = Arc::clone(&processed_repos); let total_records = Arc::clone(&total_records); + let cancelled = Arc::clone(&cancelled); let job_id = Arc::clone(&job_id_arc); async move { @@ -378,10 +426,15 @@ async fn run_fetching_phase(state: &AppState, job_id: &str, collections: &[Strin let collections = Arc::clone(&collections); let processed_repos = Arc::clone(&processed_repos); let total_records = Arc::clone(&total_records); + let cancelled = Arc::clone(&cancelled); let pds_endpoint = pds_endpoint.clone(); let job_id = Arc::clone(&job_id); async move { + if cancelled.load(Ordering::Relaxed) { + return; + } + for collection in collections.iter() { match fetch_records_from_pds( &state, @@ -433,6 +486,10 @@ async fn run_fetching_phase(state: &AppState, job_id: &str, collections: &[Strin .bind(job_id.as_str()) .execute(&state.db) .await; + + if is_cancelled(&state, job_id.as_str()).await { + cancelled.store(true, Ordering::Relaxed); + } } } }) @@ -581,6 +638,12 @@ async fn run_backfill_job(state: AppState, job_id: String) { if matches!(stage.as_str(), "pending" | "discovering_repos") { run_discovery_phase(&state, &job_id, &collections, did.as_deref()).await; + if is_cancelled(&state, &job_id).await { + tracing::info!(job_id, "backfill job cancelled"); + finalise_cancel(&state, &job_id).await; + return; + } + let total = count_repos(&state, &job_id).await; if total == 0 { complete_job(&state, &job_id, 0, 0, None).await; @@ -609,10 +672,21 @@ async fn run_backfill_job(state: AppState, job_id: String) { "pending" | "discovering_repos" | "resolving_pds" ) { run_resolution_phase(&state, &job_id).await; + + if is_cancelled(&state, &job_id).await { + tracing::info!(job_id, "backfill job cancelled"); + finalise_cancel(&state, &job_id).await; + return; + } } run_fetching_phase(&state, &job_id, &collections).await; + if is_cancelled(&state, &job_id).await { + tracing::info!(job_id, "backfill job cancelled"); + return; + } + // Read final counters from backfill_repos before cleanup let sql = adapt_sql( "SELECT COUNT(*) FROM backfill_repos WHERE job_id = ? AND status = 'completed'", @@ -717,6 +791,50 @@ pub(super) async fn create_backfill( )) } +/// POST /admin/backfill/{id}/cancel — cancel a running backfill job. +pub(super) async fn cancel_backfill( + State(state): State, + admin: UserAuth, + Path(job_id): Path, +) -> Result, AppError> { + admin.require(Permission::BackfillCreate).await?; + + let sql = adapt_sql( + "SELECT status FROM backfill_jobs WHERE id = ?", + state.db_backend, + ); + let row: Option<(String,)> = sqlx::query_as(&sql) + .bind(&job_id) + .fetch_optional(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to query backfill job: {e}")))?; + + match row { + None => Err(AppError::NotFound("backfill job not found".into())), + Some((status,)) if status != "running" => Err(AppError::BadRequest(format!( + "job is not running (status: {status})" + ))), + Some(_) => { + request_cancel(&state, &job_id).await; + log_event( + &state.db, + EventLog { + event_type: "backfill.cancelling".to_string(), + severity: Severity::Info, + actor_did: Some(admin.did.clone()), + subject: None, + detail: serde_json::json!({ "job_id": job_id }), + }, + state.db_backend, + ) + .await; + Ok(Json( + serde_json::json!({ "id": job_id, "status": "cancelling" }), + )) + } + } +} + /// GET /admin/backfill/status — list all backfill jobs. pub(super) async fn backfill_status( State(state): State, @@ -791,21 +909,30 @@ pub(super) async fn backfill_status( // --------------------------------------------------------------------------- /// Resume any backfill jobs that were running when the server last stopped. +/// Jobs stuck in `cancelling` are finalised immediately. pub async fn resume_backfill_jobs(state: &AppState) { let sql = adapt_sql( - "SELECT id FROM backfill_jobs WHERE status = 'running'", + "SELECT id, status FROM backfill_jobs WHERE status IN ('running', 'cancelling')", state.db_backend, ); - let rows: Vec<(String,)> = sqlx::query_as(&sql) + let rows: Vec<(String, String)> = sqlx::query_as(&sql) .fetch_all(&state.db) .await .unwrap_or_default(); - for (job_id,) in rows { - tracing::info!(job_id, "resuming interrupted backfill job"); - let spawn_state = state.clone(); - tokio::spawn(async move { - run_backfill_job(spawn_state, job_id).await; - }); + for (job_id, status) in rows { + if status == "cancelling" { + tracing::info!( + job_id, + "finalising cancelled backfill job from previous run" + ); + finalise_cancel(state, &job_id).await; + } else { + tracing::info!(job_id, "resuming interrupted backfill job"); + let spawn_state = state.clone(); + tokio::spawn(async move { + run_backfill_job(spawn_state, job_id).await; + }); + } } } diff --git a/src/admin/mod.rs b/src/admin/mod.rs index fb308f4..6ad2260 100644 --- a/src/admin/mod.rs +++ b/src/admin/mod.rs @@ -37,6 +37,7 @@ pub fn admin_routes(_state: AppState) -> Router { .route("/stats", get(stats::stats)) .route("/backfill", post(backfill::create_backfill)) .route("/backfill/status", get(backfill::backfill_status)) + .route("/backfill/{id}/cancel", post(backfill::cancel_backfill)) .route("/events", get(events::list_events)) .route("/users", post(users::create_user).get(users::list_users)) .route("/users/transfer-super", post(users::transfer_super)) diff --git a/web/src/app/dashboard/backfill/page.tsx b/web/src/app/dashboard/backfill/page.tsx index 4f7d91d..df1d918 100644 --- a/web/src/app/dashboard/backfill/page.tsx +++ b/web/src/app/dashboard/backfill/page.tsx @@ -3,7 +3,12 @@ import { useCallback, useEffect, useState } from "react"; import { useCurrentUser } from "@/hooks/use-current-user"; -import { createBackfillJob, getBackfillJobs, getLexicons } from "@/lib/api"; +import { + cancelBackfillJob, + createBackfillJob, + getBackfillJobs, + getLexicons, +} from "@/lib/api"; import type { BackfillJob } from "@/types/backfill"; import { CheckCircle2, Circle, Loader2 } from "lucide-react"; import { SiteHeader } from "@/components/site-header"; @@ -32,6 +37,7 @@ import { Label } from "@/components/ui/label"; import { Sheet, SheetContent, + SheetFooter, SheetHeader, SheetTitle, } from "@/components/ui/sheet"; @@ -51,6 +57,8 @@ const STAGES = [ "fetching_records", "completed", "failed", + "cancelled", + "cancelling", ] as const; const STAGE_LABELS: Record = { @@ -60,6 +68,8 @@ const STAGE_LABELS: Record = { fetching_records: "Fetching records", completed: "Completed", failed: "Failed", + cancelled: "Cancelled", + cancelling: "Cancelling", }; function stageBadge(stage: string) { @@ -72,6 +82,18 @@ function stageBadge(stage: string) { ); case "failed": return failed; + case "cancelled": + return ( + + cancelled + + ); + case "cancelling": + return ( + + cancelling + + ); case "pending": return pending; default: @@ -180,7 +202,16 @@ export default function BackfillPage() { }} > - {selectedJob && } + {selectedJob && ( + { + await cancelBackfillJob(selectedJob.id); + load(); + }} + /> + )} @@ -188,8 +219,27 @@ export default function BackfillPage() { ); } -function JobDetail({ job }: { job: BackfillJob }) { +function JobDetail({ + job, + canCancel, + onCancel, +}: { + job: BackfillJob; + canCancel: boolean; + onCancel: () => Promise; +}) { + const [cancelling, setCancelling] = useState(false); const current = stageIndex(job.stage); + const isActive = job.status === "running" || job.status === "cancelling"; + + async function handleCancel() { + setCancelling(true); + try { + await onCancel(); + } finally { + setCancelling(false); + } + } return ( <> @@ -284,6 +334,18 @@ function JobDetail({ job }: { job: BackfillJob }) { + {canCancel && isActive && ( + + + + )} ); } diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts index f05a348..d923355 100644 --- a/web/src/lib/api.ts +++ b/web/src/lib/api.ts @@ -170,6 +170,13 @@ export function createBackfillJob(body: { collection?: string; did?: string }) { }); } +export function cancelBackfillJob(id: string) { + return apiFetch<{ id: string; status: string }>( + `/admin/backfill/${id}/cancel`, + { method: "POST" }, + ); +} + // Users export function getUsers() { return apiFetch("/admin/users"); -- 2.51.2 From bd76151975c370a5778b2924a4f8efe40030691e Mon Sep 17 00:00:00 2001 From: Trezy Date: Thu, 14 May 2026 08:46:40 -0500 Subject: [PATCH 05/15] docs: update backfill deets Signed-off-by: Trezy --- .../docs/api-reference/admin/backfill.md | 63 +++++++++++++++++-- .../content/docs/getting-started/dashboard.md | 2 +- packages/docs/content/docs/guides/backfill.md | 46 +++++++++++--- 3 files changed, 99 insertions(+), 12 deletions(-) diff --git a/packages/docs/content/docs/api-reference/admin/backfill.md b/packages/docs/content/docs/api-reference/admin/backfill.md index f667856..b4a5a2e 100644 --- a/packages/docs/content/docs/api-reference/admin/backfill.md +++ b/packages/docs/content/docs/api-reference/admin/backfill.md @@ -96,10 +96,61 @@ curl -X POST http://127.0.0.1:3000/admin/backfill \ ```json { "id": "550e8400-e29b-41d4-a716-446655440000", - "status": "pending" + "status": "running" } ``` +## Cancel a backfill job + +``` +POST /admin/backfill/{id}/cancel +``` + +Requests cancellation of a running backfill job. The job status transitions to `cancelling` immediately; the background worker will stop at its next checkpoint and set the final status to `cancelled`. See the [Backfill guide](../../guides/backfill.md#cancelling-a-job) for details on the two-phase process. + +```ts tab="TypeScript" tab-group="language" +const response = await fetch( + `http://127.0.0.1:3000/admin/backfill/${jobId}/cancel`, + { method: "POST", headers }, +); +const data = await response.json(); +``` +```js tab="JavaScript" tab-group="language" +const response = await fetch( + `http://127.0.0.1:3000/admin/backfill/${jobId}/cancel`, + { method: "POST", headers }, +); +const data = await response.json(); +``` +```rust tab="Rust" tab-group="language" +let response = client + .post(format!("http://127.0.0.1:3000/admin/backfill/{job_id}/cancel")) + .bearer_auth(token) + .send() + .await?; +let data: serde_json::Value = response.json().await?; +``` +```go tab="Go" tab-group="language" +url := fmt.Sprintf("http://127.0.0.1:3000/admin/backfill/%s/cancel", jobID) +req, _ := http.NewRequest("POST", url, nil) +req.Header.Set("Authorization", "Bearer "+token) +resp, err := http.DefaultClient.Do(req) +``` +```sh tab="cURL" tab-group="language" +curl -X POST "http://127.0.0.1:3000/admin/backfill/$JOB_ID/cancel" -H "$AUTH" +``` + +**Response**: `200 OK` + +```json +{ + "id": "550e8400-e29b-41d4-a716-446655440000", + "status": "cancelling" +} +``` + +Returns `400` if the job is not currently running, or `404` if the job ID is not found. + ## List backfill jobs ``` @@ -112,9 +163,10 @@ interface BackfillJob { collection: string | null; did: string | null; status: string; - total_repos: number; - processed_repos: number; - total_records: number; + stage: string; + total_repos: number | null; + processed_repos: number | null; + total_records: number | null; error: string | null; started_at: string | null; completed_at: string | null; @@ -158,6 +210,7 @@ curl http://127.0.0.1:3000/admin/backfill/status -H "$AUTH" "collection": "xyz.statusphere.status", "did": null, "status": "completed", + "stage": "completed", "total_repos": 42, "processed_repos": 42, "total_records": 1000, @@ -168,3 +221,5 @@ curl http://127.0.0.1:3000/admin/backfill/status -H "$AUTH" } ] ``` + +The `status` field tracks the overall job state (`running`, `cancelling`, `cancelled`, `completed`, `failed`). The `stage` field tracks the current processing phase (`pending`, `discovering_repos`, `resolving_pds`, `fetching_records`, `completed`, `failed`, `cancelled`). diff --git a/packages/docs/content/docs/getting-started/dashboard.md b/packages/docs/content/docs/getting-started/dashboard.md index 324a49b..1e55bb8 100644 --- a/packages/docs/content/docs/getting-started/dashboard.md +++ b/packages/docs/content/docs/getting-started/dashboard.md @@ -49,7 +49,7 @@ Navigate to **Records** to browse all indexed atproto records. Records are group ### Backfill -Navigate to **Backfill** to view and manage backfill jobs. You can start a new backfill for any record-type lexicon to import historical records from the network. The page shows job status, progress (repos processed / total), and record counts. See [Backfill](../guides/backfill.md) for how the process works. +Navigate to **Backfill** to view and manage backfill jobs. You can start a new backfill for any record-type lexicon to import historical records from the network. The table shows each job's collection, DID scope, current stage, and start time. Click a row to open a detail sheet with full metadata and a stage-by-stage progress log that updates in real time. Running jobs can be cancelled from the detail sheet — the job transitions to "cancelling" while the worker finishes its current batch, then to "cancelled". See [Backfill](../guides/backfill.md) for how the process works. ### Dead Letters diff --git a/packages/docs/content/docs/guides/backfill.md b/packages/docs/content/docs/guides/backfill.md index 2414484..da7cc42 100644 --- a/packages/docs/content/docs/guides/backfill.md +++ b/packages/docs/content/docs/guides/backfill.md @@ -13,17 +13,49 @@ See the [admin API](../api-reference/admin/backfill.md) for endpoint details. ## How it works -1. **Determine target collections**: uses the specified collection, or all record lexicons with `backfill: true` -2. **Discover DIDs**: HappyView calls the relay's `com.atproto.sync.listReposByCollection` to find repos that contain records for each target collection (paginated) -3. **Resolve each PDS**: for each discovered DID, HappyView resolves the DID document via PLC to find the user's PDS endpoint -4. **Fetch records**: HappyView calls `com.atproto.repo.listRecords` on each PDS for the target collection (paginated) and upserts each record into the local database -5. **Track progress**: counters for `processed_repos` and `total_records` are updated as the job runs +A backfill job runs through three sequential phases: + +1. **Discovering repos** — HappyView calls the relay's `com.atproto.sync.listReposByCollection` to find repos that contain records for each target collection. Discovered DIDs are stored in a tracking table so progress can be resumed. +2. **Resolving PDS** — For each discovered DID, HappyView resolves the DID document (via PLC directory or `did:web`) to find the user's PDS endpoint. +3. **Fetching records** — HappyView calls `com.atproto.repo.listRecords` on each PDS for the target collection(s), upserting each record into the local database. PDS endpoints are processed concurrently (up to 10 PDS hosts, 3 DIDs per host). + +Progress counters (`total_repos`, `processed_repos`, `total_records`) and the current `stage` are updated in real time. The dashboard's Backfill page shows live progress, and clicking a job opens a detail sheet with a stage-by-stage progress log. + +### Rate limiting + +All three phases handle HTTP 429 responses. HappyView reads the `RateLimit-Reset` header (a Unix timestamp, the AT Protocol convention) to determine how long to wait, falling back to the `retry-after` header, then defaulting to 5 seconds. ## Job lifecycle -A backfill job moves through `pending → running → completed` (or `failed`). Unlike earlier versions of HappyView that relied on Tap, the job is only marked `completed` once every discovered repo has been fully processed — there is no separate downstream queue. Progress is visible in real time on the dashboard's Backfill page. +A backfill job has both a `status` (overall state) and a `stage` (current phase): + +| Status | Description | +| ------------ | ---------------------------------------------------- | +| `running` | Job is actively processing | +| `cancelling` | Cancel requested, waiting for the worker to stop | +| `cancelled` | Worker has stopped and cleaned up | +| `completed` | All repos processed successfully | +| `failed` | An error occurred | + +The `stage` field tracks which phase the job is in: `pending`, `discovering_repos`, `resolving_pds`, `fetching_records`, `completed`, `failed`, or `cancelled`. + +## Cancelling a job + +Running jobs can be cancelled via `POST /admin/backfill/{id}/cancel` or the Cancel button in the dashboard. Cancellation is two-phase: + +1. The endpoint sets the job status to `cancelling`. +2. The worker checks for cancellation at natural checkpoints (between relay pages, every 100 DIDs during resolution, every 100 repos during fetching). When it detects the `cancelling` status, it stops work and sets the final status to `cancelled`. + +This means there may be a short delay between clicking Cancel and the job fully stopping, depending on what the worker is doing at that moment. + +## Resuming after restart + +Backfill jobs survive server restarts. On startup, HappyView checks for jobs that were running when the server last stopped: + +- **Running** jobs are re-spawned and resume from where they left off. Each phase is idempotent — discovery skips already-known DIDs, resolution skips already-resolved endpoints, and fetching skips already-completed repos. +- **Cancelling** jobs (where the cancel was requested but the worker hadn't stopped yet) are immediately finalised as `cancelled`. -If a job fails midway, the `error` field contains the failure reason. Re-running the backfill resumes from scratch but is idempotent (records are upserted by URI). +Per-DID progress is tracked in the database, so a job that was halfway through fetching records will pick up from the next unprocessed repo, not start over. ## Re-running backfills -- 2.51.2 From 54d8e28229c986745c8a09c07a5aedc69aff610f Mon Sep 17 00:00:00 2001 From: Trezy Date: Thu, 14 May 2026 09:16:37 -0500 Subject: [PATCH 06/15] fix: issues with backfill stage/status separation and progress tracking Signed-off-by: Trezy --- src/admin/backfill.rs | 52 ++++++++++--------- web/src/app/dashboard/backfill/page.tsx | 66 +++++++++++-------------- 2 files changed, 54 insertions(+), 64 deletions(-) diff --git a/src/admin/backfill.rs b/src/admin/backfill.rs index 633096a..eaa4e79 100644 --- a/src/admin/backfill.rs +++ b/src/admin/backfill.rs @@ -125,7 +125,7 @@ async fn cleanup_repos(state: &AppState, job_id: &str) { async fn fail_job(state: &AppState, job_id: &str, error: &str) { let now = now_rfc3339(); let sql = adapt_sql( - "UPDATE backfill_jobs SET status = 'failed', stage = 'failed', completed_at = ?, error = ? WHERE id = ?", + "UPDATE backfill_jobs SET status = 'failed', completed_at = ?, error = ? WHERE id = ?", state.db_backend, ); let _ = sqlx::query(&sql) @@ -162,7 +162,7 @@ async fn request_cancel(state: &AppState, job_id: &str) { async fn finalise_cancel(state: &AppState, job_id: &str) { let now = now_rfc3339(); let sql = adapt_sql( - "UPDATE backfill_jobs SET status = 'cancelled', stage = 'cancelled', completed_at = ?, error = 'cancelled by user' WHERE id = ?", + "UPDATE backfill_jobs SET status = 'cancelled', completed_at = ?, error = 'cancelled by user' WHERE id = ?", state.db_backend, ); let _ = sqlx::query(&sql) @@ -370,7 +370,7 @@ async fn run_resolution_phase(state: &AppState, job_id: &str) { // Phase 3: Fetch records from PDS instances // --------------------------------------------------------------------------- -async fn run_fetching_phase(state: &AppState, job_id: &str, collections: &[String]) { +async fn run_fetching_phase(state: &AppState, job_id: &str, collections: &[String]) -> (i32, i32) { set_stage(state, job_id, "fetching_records").await; // Load pending repos grouped by PDS @@ -401,6 +401,9 @@ async fn run_fetching_phase(state: &AppState, job_id: &str, collections: &[Strin .map(|(c,)| c) .unwrap_or(0); + // Reset processed_repos for the fetching phase + update_job_counter(state, job_id, "processed_repos", already_completed).await; + let processed_repos = Arc::new(AtomicI32::new(already_completed)); let total_records = Arc::new(AtomicI32::new(0)); let cancelled = Arc::new(AtomicBool::new(false)); @@ -497,6 +500,23 @@ async fn run_fetching_phase(state: &AppState, job_id: &str, collections: &[Strin } }) .await; + + let final_repos = processed_repos.load(Ordering::Relaxed); + let final_records = total_records.load(Ordering::Relaxed); + + // Persist final counts so they're accurate regardless of batch size + let sql = adapt_sql( + "UPDATE backfill_jobs SET processed_repos = ?, total_records = ? WHERE id = ?", + state.db_backend, + ); + let _ = sqlx::query(&sql) + .bind(final_repos) + .bind(final_records) + .bind(job_id) + .execute(&state.db) + .await; + + (final_repos, final_records) } /// Fetch all records for a given DID and collection from a PDS via @@ -680,36 +700,14 @@ async fn run_backfill_job(state: AppState, job_id: String) { } } - run_fetching_phase(&state, &job_id, &collections).await; + let (final_processed, final_records) = run_fetching_phase(&state, &job_id, &collections).await; if is_cancelled(&state, &job_id).await { tracing::info!(job_id, "backfill job cancelled"); + finalise_cancel(&state, &job_id).await; return; } - // Read final counters from backfill_repos before cleanup - let sql = adapt_sql( - "SELECT COUNT(*) FROM backfill_repos WHERE job_id = ? AND status = 'completed'", - state.db_backend, - ); - let final_processed: i32 = sqlx::query_as::<_, (i32,)>(&sql) - .bind(&job_id) - .fetch_one(&state.db) - .await - .map(|(c,)| c) - .unwrap_or(0); - - let sql = adapt_sql( - "SELECT total_records FROM backfill_jobs WHERE id = ?", - state.db_backend, - ); - let final_records: i32 = sqlx::query_as::<_, (i32,)>(&sql) - .bind(&job_id) - .fetch_one(&state.db) - .await - .map(|(c,)| c) - .unwrap_or(0); - complete_job(&state, &job_id, final_processed, final_records, None).await; log_event( diff --git a/web/src/app/dashboard/backfill/page.tsx b/web/src/app/dashboard/backfill/page.tsx index df1d918..e602e03 100644 --- a/web/src/app/dashboard/backfill/page.tsx +++ b/web/src/app/dashboard/backfill/page.tsx @@ -50,30 +50,14 @@ import { TableRow, } from "@/components/ui/table"; -const STAGES = [ - "pending", +const PROGRESS_PHASES = [ "discovering_repos", "resolving_pds", "fetching_records", - "completed", - "failed", - "cancelled", - "cancelling", ] as const; -const STAGE_LABELS: Record = { - pending: "Pending", - discovering_repos: "Discovering repos", - resolving_pds: "Resolving PDS", - fetching_records: "Fetching records", - completed: "Completed", - failed: "Failed", - cancelled: "Cancelled", - cancelling: "Cancelling", -}; - -function stageBadge(stage: string) { - switch (stage) { +function statusBadge(job: BackfillJob) { + switch (job.status) { case "completed": return ( @@ -94,20 +78,22 @@ function stageBadge(stage: string) { cancelling ); - case "pending": - return pending; - default: + case "running": return ( - {STAGE_LABELS[stage] ?? stage} + {job.stage === "pending" ? "starting" : job.stage.replace(/_/g, " ")} ); + default: + return {job.status}; } } -function stageIndex(stage: string): number { - const idx = STAGES.indexOf(stage as (typeof STAGES)[number]); - return idx === -1 ? 0 : idx; +function phaseIndex(stage: string): number { + const idx = PROGRESS_PHASES.indexOf( + stage as (typeof PROGRESS_PHASES)[number], + ); + return idx; } export default function BackfillPage() { @@ -183,7 +169,7 @@ export default function BackfillPage() { {job.did ?? "All"} - {stageBadge(job.stage)} + {statusBadge(job)} {job.started_at ? new Date(job.started_at).toLocaleString() @@ -229,9 +215,15 @@ function JobDetail({ onCancel: () => Promise; }) { const [cancelling, setCancelling] = useState(false); - const current = stageIndex(job.stage); + const current = phaseIndex(job.stage); + const allDone = job.status === "completed"; const isActive = job.status === "running" || job.status === "cancelling"; + function hasReached(phase: (typeof PROGRESS_PHASES)[number]): boolean { + if (allDone) return true; + return current >= phaseIndex(phase); + } + async function handleCancel() { setCancelling(true); try { @@ -300,17 +292,17 @@ function JobDetail({
= stageIndex("discovering_repos")} + active={isActive && job.stage === "discovering_repos"} + reached={hasReached("discovering_repos")} value={job.total_repos?.toLocaleString()} suffix="repos found" /> = stageIndex("resolving_pds")} + active={isActive && job.stage === "resolving_pds"} + reached={hasReached("resolving_pds")} value={ - current >= stageIndex("resolving_pds") + hasReached("resolving_pds") ? `${job.processed_repos?.toLocaleString() ?? "0"} / ${job.total_repos?.toLocaleString() ?? "0"}` : undefined } @@ -318,15 +310,15 @@ function JobDetail({ /> = stageIndex("fetching_records")} + active={isActive && job.stage === "fetching_records"} + reached={hasReached("fetching_records")} value={ - current >= stageIndex("fetching_records") + hasReached("fetching_records") ? `${job.processed_repos?.toLocaleString() ?? "0"} / ${job.total_repos?.toLocaleString() ?? "0"} repos` : undefined } suffix={ - current >= stageIndex("fetching_records") + hasReached("fetching_records") ? `${job.total_records?.toLocaleString() ?? "0"} records` : undefined } -- 2.51.2 From 81c4ea84c1f0cccc5027ea4141286f414af2b1a3 Mon Sep 17 00:00:00 2001 From: Trezy Date: Thu, 14 May 2026 09:36:27 -0500 Subject: [PATCH 07/15] fix: update column type in backfill migration Signed-off-by: Trezy --- migrations/postgres/20260513000001_create_backfill_repos.sql | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/migrations/postgres/20260513000001_create_backfill_repos.sql b/migrations/postgres/20260513000001_create_backfill_repos.sql index 9c0d25b..0e613eb 100644 --- a/migrations/postgres/20260513000001_create_backfill_repos.sql +++ b/migrations/postgres/20260513000001_create_backfill_repos.sql @@ -1,5 +1,5 @@ CREATE TABLE IF NOT EXISTS backfill_repos ( - job_id UUID NOT NULL REFERENCES backfill_jobs(id) ON DELETE CASCADE, + job_id TEXT NOT NULL REFERENCES backfill_jobs(id) ON DELETE CASCADE, did TEXT NOT NULL, pds_endpoint TEXT, status TEXT NOT NULL DEFAULT 'pending', -- 2.51.2 From 25f095520d4190fd93c8e2466e2de6ce1e639c7e Mon Sep 17 00:00:00 2001 From: Trezy Date: Thu, 14 May 2026 10:32:54 -0500 Subject: [PATCH 08/15] fix: harden backfill cancel idempotency, SQL injection guard, DID retry cap, and row a11y Signed-off-by: Trezy --- src/admin/backfill.rs | 20 ++++++++--- src/profile.rs | 44 +++++++++++++++++-------- web/src/app/dashboard/backfill/page.tsx | 10 +++++- 3 files changed, 55 insertions(+), 19 deletions(-) diff --git a/src/admin/backfill.rs b/src/admin/backfill.rs index eaa4e79..d0fc378 100644 --- a/src/admin/backfill.rs +++ b/src/admin/backfill.rs @@ -90,10 +90,19 @@ async fn set_stage(state: &AppState, job_id: &str, stage: &str) { } async fn update_job_counter(state: &AppState, job_id: &str, column: &str, value: i32) { - let sql = adapt_sql( - &format!("UPDATE backfill_jobs SET {column} = ? WHERE id = ?"), - state.db_backend, - ); + let query = match column { + "total_repos" => "UPDATE backfill_jobs SET total_repos = ? WHERE id = ?", + "processed_repos" => "UPDATE backfill_jobs SET processed_repos = ? WHERE id = ?", + "total_records" => "UPDATE backfill_jobs SET total_records = ? WHERE id = ?", + other => { + tracing::error!( + column = other, + "update_job_counter called with unknown column" + ); + return; + } + }; + let sql = adapt_sql(query, state.db_backend); let _ = sqlx::query(&sql) .bind(value) .bind(job_id) @@ -809,6 +818,9 @@ pub(super) async fn cancel_backfill( match row { None => Err(AppError::NotFound("backfill job not found".into())), + Some((ref status,)) if status == "cancelling" || status == "cancelled" => { + Ok(Json(serde_json::json!({ "id": job_id, "status": status }))) + } Some((status,)) if status != "running" => Err(AppError::BadRequest(format!( "job is not running (status: {status})" ))), diff --git a/src/profile.rs b/src/profile.rs index 25bf03b..298ea01 100644 --- a/src/profile.rs +++ b/src/profile.rs @@ -161,21 +161,37 @@ pub async fn resolve_did_document( format!("{}/{did}", plc_url.trim_end_matches('/')) }; - let resp = loop { - let r = http - .get(&url) - .send() - .await - .map_err(|e| AppError::Internal(format!("DID resolution failed: {e}")))?; - - if r.status() == reqwest::StatusCode::TOO_MANY_REQUESTS { - let wait = parse_retry_after(r.headers()); - tracing::warn!(did, wait, "rate limited during DID resolution, sleeping"); - tokio::time::sleep(tokio::time::Duration::from_secs(wait)).await; - continue; + let resp = { + let max_retries = 5; + let mut attempts = 0; + loop { + let r = http + .get(&url) + .send() + .await + .map_err(|e| AppError::Internal(format!("DID resolution failed: {e}")))?; + + if r.status() == reqwest::StatusCode::TOO_MANY_REQUESTS { + attempts += 1; + if attempts >= max_retries { + return Err(AppError::Internal(format!( + "DID resolution for {did} rate-limited after {max_retries} retries" + ))); + } + let wait = parse_retry_after(r.headers()); + tracing::warn!( + did, + wait, + attempts, + max_retries, + "rate limited during DID resolution, sleeping" + ); + tokio::time::sleep(tokio::time::Duration::from_secs(wait)).await; + continue; + } + + break r; } - - break r; }; if !resp.status().is_success() { diff --git a/web/src/app/dashboard/backfill/page.tsx b/web/src/app/dashboard/backfill/page.tsx index e602e03..f2ed0b6 100644 --- a/web/src/app/dashboard/backfill/page.tsx +++ b/web/src/app/dashboard/backfill/page.tsx @@ -157,8 +157,16 @@ export default function BackfillPage() { {jobs.map((job) => ( setSelectedJobId(job.id)} + onKeyDown={(e) => { + if (e.key === "Enter" || e.key === " ") { + e.preventDefault(); + setSelectedJobId(job.id); + } + }} > {job.id.slice(0, 8)} -- 2.51.2 From 39d9465d8fcca7eabf8146000fa98eaba3b05cbc Mon Sep 17 00:00:00 2001 From: Trezy Date: Tue, 12 May 2026 10:49:44 -0500 Subject: [PATCH 09/15] fix: add permissions to the internal API Signed-off-by: Trezy --- src/admin/mod.rs | 1 + src/admin/permissions.rs | 465 ++++++++++++++++++++++++++++++++++++++- src/admin/users.rs | 52 ++++- tests/e2e_permissions.rs | 258 ++++++++++++++++++++++ web/src/lib/api.ts | 23 ++ 5 files changed, 796 insertions(+), 3 deletions(-) create mode 100644 tests/e2e_permissions.rs diff --git a/src/admin/mod.rs b/src/admin/mod.rs index 6ad2260..d4077ac 100644 --- a/src/admin/mod.rs +++ b/src/admin/mod.rs @@ -126,4 +126,5 @@ pub fn admin_routes(_state: AppState) -> Router { .route("/dead-letters/{id}/dismiss", post(dead_letters::dismiss)) .route("/dead-letters/{id}/retry", post(dead_letters::retry)) .route("/dead-letters/{id}/reindex", post(dead_letters::reindex)) + .route("/permissions", get(users::list_permissions)) } diff --git a/src/admin/permissions.rs b/src/admin/permissions.rs index 3a8406f..1bc0977 100644 --- a/src/admin/permissions.rs +++ b/src/admin/permissions.rs @@ -2,7 +2,15 @@ use std::collections::HashSet; use serde::{Deserialize, Serialize}; -/// All 37 permissions in the system. +#[derive(Serialize)] +pub struct PermissionInfo { + pub key: &'static str, + pub name: &'static str, + pub description: &'static str, + pub category: &'static str, +} + +/// All permissions in the system. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] pub enum Permission { #[serde(rename = "lexicons:create")] @@ -150,6 +158,257 @@ impl Permission { } } + pub fn info(&self) -> PermissionInfo { + match self { + Self::LexiconsCreate => PermissionInfo { + key: "lexicons:create", + name: "Create Lexicons", + description: "Upload and register new lexicon schemas", + category: "Lexicons", + }, + Self::LexiconsRead => PermissionInfo { + key: "lexicons:read", + name: "View Lexicons", + description: "View registered lexicon schemas", + category: "Lexicons", + }, + Self::LexiconsDelete => PermissionInfo { + key: "lexicons:delete", + name: "Delete Lexicons", + description: "Remove lexicon schemas", + category: "Lexicons", + }, + Self::RecordsRead => PermissionInfo { + key: "records:read", + name: "View Records", + description: "Browse indexed AT Protocol records", + category: "Records", + }, + Self::RecordsDelete => PermissionInfo { + key: "records:delete", + name: "Delete Records", + description: "Delete individual records from the index", + category: "Records", + }, + Self::RecordsDeleteCollection => PermissionInfo { + key: "records:delete-collection", + name: "Delete Collections", + description: "Bulk-delete all records in a collection", + category: "Records", + }, + Self::ScriptVariablesCreate => PermissionInfo { + key: "script-variables:create", + name: "Create Script Variables", + description: "Add or update environment variables for Lua scripts", + category: "Script Variables", + }, + Self::ScriptVariablesRead => PermissionInfo { + key: "script-variables:read", + name: "View Script Variables", + description: "View script environment variable keys and values", + category: "Script Variables", + }, + Self::ScriptVariablesDelete => PermissionInfo { + key: "script-variables:delete", + name: "Delete Script Variables", + description: "Remove script environment variables", + category: "Script Variables", + }, + Self::UsersCreate => PermissionInfo { + key: "users:create", + name: "Create Users", + description: "Add new dashboard users", + category: "Users", + }, + Self::UsersRead => PermissionInfo { + key: "users:read", + name: "View Users", + description: "View the user list and their permissions", + category: "Users", + }, + Self::UsersUpdate => PermissionInfo { + key: "users:update", + name: "Update Users", + description: "Modify user permissions", + category: "Users", + }, + Self::UsersDelete => PermissionInfo { + key: "users:delete", + name: "Delete Users", + description: "Remove dashboard users", + category: "Users", + }, + Self::ApiKeysCreate => PermissionInfo { + key: "api-keys:create", + name: "Create API Keys", + description: "Generate new API keys for admin access", + category: "API Keys", + }, + Self::ApiKeysRead => PermissionInfo { + key: "api-keys:read", + name: "View API Keys", + description: "View existing API keys", + category: "API Keys", + }, + Self::ApiKeysDelete => PermissionInfo { + key: "api-keys:delete", + name: "Revoke API Keys", + description: "Revoke existing API keys", + category: "API Keys", + }, + Self::BackfillCreate => PermissionInfo { + key: "backfill:create", + name: "Start Backfill", + description: "Trigger historical record backfill jobs", + category: "Backfill", + }, + Self::BackfillRead => PermissionInfo { + key: "backfill:read", + name: "View Backfill", + description: "View backfill job status and progress", + category: "Backfill", + }, + Self::StatsRead => PermissionInfo { + key: "stats:read", + name: "View Stats", + description: "View collection statistics and record counts", + category: "System", + }, + Self::EventsRead => PermissionInfo { + key: "events:read", + name: "View Events", + description: "View the event log", + category: "System", + }, + Self::LabelersCreate => PermissionInfo { + key: "labelers:create", + name: "Add Labelers", + description: "Subscribe to external labeler services", + category: "Labelers", + }, + Self::LabelersRead => PermissionInfo { + key: "labelers:read", + name: "View Labelers", + description: "View subscribed labeler services", + category: "Labelers", + }, + Self::LabelersDelete => PermissionInfo { + key: "labelers:delete", + name: "Remove Labelers", + description: "Unsubscribe from labeler services", + category: "Labelers", + }, + Self::SettingsManage => PermissionInfo { + key: "settings:manage", + name: "Manage Settings", + description: "Modify instance settings, logo, and configuration", + category: "Settings", + }, + Self::PluginsRead => PermissionInfo { + key: "plugins:read", + name: "View Plugins", + description: "View installed plugins and their configuration", + category: "Plugins", + }, + Self::PluginsCreate => PermissionInfo { + key: "plugins:create", + name: "Install Plugins", + description: "Install and configure new plugins", + category: "Plugins", + }, + Self::PluginsDelete => PermissionInfo { + key: "plugins:delete", + name: "Remove Plugins", + description: "Uninstall plugins", + category: "Plugins", + }, + Self::ApiClientsView => PermissionInfo { + key: "api-clients:view", + name: "View API Clients", + description: "View registered OAuth API clients", + category: "API Clients", + }, + Self::ApiClientsCreate => PermissionInfo { + key: "api-clients:create", + name: "Create API Clients", + description: "Register new OAuth API clients", + category: "API Clients", + }, + Self::ApiClientsEdit => PermissionInfo { + key: "api-clients:edit", + name: "Edit API Clients", + description: "Modify API client settings and credentials", + category: "API Clients", + }, + Self::ApiClientsDelete => PermissionInfo { + key: "api-clients:delete", + name: "Delete API Clients", + description: "Remove registered API clients", + category: "API Clients", + }, + Self::DeadLettersRead => PermissionInfo { + key: "dead-letters:read", + name: "View Dead Letters", + description: "View failed hook executions", + category: "Dead Letters", + }, + Self::DeadLettersManage => PermissionInfo { + key: "dead-letters:manage", + name: "Manage Dead Letters", + description: "Retry, re-index, or dismiss dead letters", + category: "Dead Letters", + }, + Self::SpacesCreate => PermissionInfo { + key: "spaces:create", + name: "Create Spaces", + description: "Create new permissioned data spaces", + category: "Spaces", + }, + Self::SpacesRead => PermissionInfo { + key: "spaces:read", + name: "View Spaces", + description: "View space details and metadata", + category: "Spaces", + }, + Self::SpacesUpdate => PermissionInfo { + key: "spaces:update", + name: "Update Spaces", + description: "Modify space settings", + category: "Spaces", + }, + Self::SpacesDelete => PermissionInfo { + key: "spaces:delete", + name: "Delete Spaces", + description: "Remove spaces and their data", + category: "Spaces", + }, + Self::SpacesManageMembers => PermissionInfo { + key: "spaces:manage-members", + name: "Manage Members", + description: "Add or remove space members and roles", + category: "Spaces", + }, + Self::SpacesManageInvites => PermissionInfo { + key: "spaces:manage-invites", + name: "Manage Invites", + description: "Create and revoke space invitations", + category: "Spaces", + }, + Self::SpacesManageRecords => PermissionInfo { + key: "spaces:manage-records", + name: "Manage Records", + description: "Read and write records within spaces", + category: "Spaces", + }, + Self::SpacesManageCredentials => PermissionInfo { + key: "spaces:manage-credentials", + name: "Manage Credentials", + description: "Issue and revoke space access credentials", + category: "Spaces", + }, + } + } + /// All permissions. pub fn all() -> HashSet { HashSet::from([ @@ -198,8 +457,65 @@ impl Permission { } } +/// Ordered list of all permissions with metadata. +pub fn catalog() -> Vec { + use Permission::*; + [ + LexiconsCreate, + LexiconsRead, + LexiconsDelete, + RecordsRead, + RecordsDelete, + RecordsDeleteCollection, + ScriptVariablesCreate, + ScriptVariablesRead, + ScriptVariablesDelete, + UsersCreate, + UsersRead, + UsersUpdate, + UsersDelete, + ApiKeysCreate, + ApiKeysRead, + ApiKeysDelete, + BackfillCreate, + BackfillRead, + StatsRead, + EventsRead, + LabelersCreate, + LabelersRead, + LabelersDelete, + SettingsManage, + PluginsRead, + PluginsCreate, + PluginsDelete, + ApiClientsView, + ApiClientsCreate, + ApiClientsEdit, + ApiClientsDelete, + DeadLettersRead, + DeadLettersManage, + SpacesCreate, + SpacesRead, + SpacesUpdate, + SpacesDelete, + SpacesManageMembers, + SpacesManageInvites, + SpacesManageRecords, + SpacesManageCredentials, + ] + .iter() + .map(|p| p.info()) + .collect() +} + +/// Check whether a permission string is recognized. +#[allow(dead_code)] +pub fn is_valid(key: &str) -> bool { + serde_json::from_value::(serde_json::Value::String(key.to_string())).is_ok() +} + /// Predefined permission templates. -#[derive(Debug, Clone, Copy, Deserialize)] +#[derive(Debug, Clone, Copy, Deserialize, Serialize)] #[serde(rename_all = "snake_case")] pub enum Template { Viewer, @@ -208,7 +524,47 @@ pub enum Template { FullAccess, } +#[derive(Serialize)] +pub struct TemplateInfo { + pub key: String, + pub label: &'static str, + pub permissions: Vec<&'static str>, +} + impl Template { + pub const ALL: &[Template] = &[ + Template::Viewer, + Template::Operator, + Template::Manager, + Template::FullAccess, + ]; + + pub fn key(&self) -> &'static str { + match self { + Self::Viewer => "viewer", + Self::Operator => "operator", + Self::Manager => "manager", + Self::FullAccess => "full_access", + } + } + + pub fn label(&self) -> &'static str { + match self { + Self::Viewer => "Viewer", + Self::Operator => "Operator", + Self::Manager => "Manager", + Self::FullAccess => "Full Access", + } + } + + pub fn info(&self) -> TemplateInfo { + TemplateInfo { + key: self.key().to_string(), + label: self.label(), + permissions: self.permissions().iter().map(|p| p.as_str()).collect(), + } + } + pub fn permissions(&self) -> HashSet { match self { Self::Viewer => HashSet::from([ @@ -262,3 +618,108 @@ impl Template { } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn catalog_covers_all_permissions() { + let catalog_keys: Vec<&str> = catalog().iter().map(|p| p.key).collect(); + for perm in Permission::all() { + assert!( + catalog_keys.contains(&perm.as_str()), + "Permission {} missing from catalog()", + perm.as_str() + ); + } + } + + #[test] + fn catalog_has_no_duplicates() { + let entries = catalog(); + let mut seen = std::collections::HashSet::new(); + for entry in &entries { + assert!( + seen.insert(entry.key), + "Duplicate key in catalog: {}", + entry.key + ); + } + } + + #[test] + fn info_key_matches_as_str() { + for perm in Permission::all() { + assert_eq!(perm.info().key, perm.as_str()); + } + } + + #[test] + fn info_fields_are_nonempty() { + for perm in Permission::all() { + let info = perm.info(); + assert!(!info.name.is_empty(), "{} has empty name", info.key); + assert!( + !info.description.is_empty(), + "{} has empty description", + info.key + ); + assert!(!info.category.is_empty(), "{} has empty category", info.key); + } + } + + #[test] + fn is_valid_accepts_known_permissions() { + assert!(is_valid("lexicons:create")); + assert!(is_valid("spaces:manage-members")); + } + + #[test] + fn is_valid_rejects_unknown_permissions() { + assert!(!is_valid("fake:permission")); + assert!(!is_valid("")); + } + + #[test] + fn template_full_access_covers_all() { + assert_eq!(Template::FullAccess.permissions(), Permission::all()); + } + + #[test] + fn template_viewer_is_subset_of_operator() { + let viewer = Template::Viewer.permissions(); + let operator = Template::Operator.permissions(); + assert!(viewer.is_subset(&operator)); + } + + #[test] + fn template_operator_is_subset_of_manager() { + let operator = Template::Operator.permissions(); + let manager = Template::Manager.permissions(); + assert!(operator.is_subset(&manager)); + } + + #[test] + fn template_info_permissions_match_template_permissions() { + for t in Template::ALL { + let info = t.info(); + let expected: HashSet<&str> = t.permissions().iter().map(|p| p.as_str()).collect(); + let actual: HashSet<&str> = info.permissions.into_iter().collect(); + assert_eq!(expected, actual, "Template {:?} info mismatch", t); + } + } + + #[test] + fn spaces_permissions_are_in_spaces_category() { + for entry in catalog() { + if entry.key.starts_with("spaces:") { + assert_eq!( + entry.category, "Spaces", + "{} should be in Spaces category", + entry.key + ); + } + } + } +} diff --git a/src/admin/users.rs b/src/admin/users.rs index 6b8aaf7..375da13 100644 --- a/src/admin/users.rs +++ b/src/admin/users.rs @@ -10,7 +10,7 @@ use crate::error::AppError; use crate::event_log::{EventLog, Severity, log_event}; use super::auth::UserAuth; -use super::permissions::Permission; +use super::permissions::{self, Permission}; use super::types::{CreateUserBody, TransferSuperBody, UpdatePermissionsBody, UserSummary}; /// POST /admin/users — create a new user with template or explicit permissions. @@ -480,3 +480,53 @@ pub(super) async fn transfer_super( Ok(StatusCode::NO_CONTENT) } + +/// GET /admin/permissions — list all available permissions and templates. +pub(super) async fn list_permissions( + State(state): State, + auth: UserAuth, +) -> Result, AppError> { + auth.require(Permission::UsersRead).await?; + + let spaces_enabled = crate::feature_flags::is_enabled( + &state.db, + crate::feature_flags::FeatureFlag::SPACES_ENABLED, + state.db_backend, + ) + .await; + + let all_permissions: Vec = permissions::catalog() + .into_iter() + .filter(|p| spaces_enabled || p.category != "Spaces") + .map(|p| { + serde_json::json!({ + "key": p.key, + "name": p.name, + "description": p.description, + "category": p.category, + }) + }) + .collect(); + + let templates: Vec = permissions::Template::ALL + .iter() + .map(|t| { + let info = t.info(); + let perms: Vec<&str> = info + .permissions + .into_iter() + .filter(|p| spaces_enabled || !p.starts_with("spaces:")) + .collect(); + serde_json::json!({ + "key": info.key, + "label": info.label, + "permissions": perms, + }) + }) + .collect(); + + Ok(Json(serde_json::json!({ + "permissions": all_permissions, + "templates": templates, + }))) +} diff --git a/tests/e2e_permissions.rs b/tests/e2e_permissions.rs new file mode 100644 index 0000000..6827d23 --- /dev/null +++ b/tests/e2e_permissions.rs @@ -0,0 +1,258 @@ +mod common; + +use axum::body::Body; +use axum::http::{Method, Request, StatusCode}; +use http_body_util::BodyExt; +use serde_json::{Value, json}; +use serial_test::serial; +use tower::ServiceExt; + +use common::app::TestApp; + +async fn json_body(resp: axum::response::Response) -> Value { + let body = resp.into_body().collect().await.unwrap().to_bytes(); + serde_json::from_slice(&body).unwrap() +} + +fn admin_get( + uri: &str, + cookie: (axum::http::HeaderName, axum::http::HeaderValue), +) -> Request { + Request::builder() + .uri(uri) + .header(cookie.0, cookie.1) + .body(Body::empty()) + .unwrap() +} + +fn admin_put( + uri: &str, + cookie: (axum::http::HeaderName, axum::http::HeaderValue), + body: &Value, +) -> Request { + Request::builder() + .method(Method::PUT) + .uri(uri) + .header(cookie.0, cookie.1) + .header("content-type", "application/json") + .body(Body::from(serde_json::to_vec(body).unwrap())) + .unwrap() +} + +fn admin_delete( + uri: &str, + cookie: (axum::http::HeaderName, axum::http::HeaderValue), +) -> Request { + Request::builder() + .method(Method::DELETE) + .uri(uri) + .header(cookie.0, cookie.1) + .body(Body::empty()) + .unwrap() +} + +#[tokio::test] +#[serial] +async fn permissions_requires_auth() { + common::require_db!(); + let app = TestApp::new().await; + + let resp = app + .router + .clone() + .oneshot( + Request::builder() + .uri("/admin/permissions") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); +} + +#[tokio::test] +#[serial] +async fn permissions_returns_catalog() { + common::require_db!(); + let app = TestApp::new().await; + + let resp = app + .router + .clone() + .oneshot(admin_get("/admin/permissions", app.admin_cookie())) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + let body = json_body(resp).await; + + let permissions = body["permissions"].as_array().expect("permissions array"); + assert!(!permissions.is_empty()); + + let first = &permissions[0]; + assert!(first["key"].is_string()); + assert!(first["name"].is_string()); + assert!(first["description"].is_string()); + assert!(first["category"].is_string()); + + let templates = body["templates"].as_array().expect("templates array"); + assert!(!templates.is_empty()); + + let first_template = &templates[0]; + assert!(first_template["key"].is_string()); + assert!(first_template["label"].is_string()); + assert!(first_template["permissions"].is_array()); +} + +#[tokio::test] +#[serial] +async fn permissions_excludes_spaces_when_flag_disabled() { + common::require_db!(); + let app = TestApp::new().await; + + let resp = app + .router + .clone() + .oneshot(admin_get("/admin/permissions", app.admin_cookie())) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + let body = json_body(resp).await; + + let permissions = body["permissions"].as_array().unwrap(); + let has_spaces = permissions + .iter() + .any(|p| p["key"].as_str().unwrap_or("").starts_with("spaces:")); + assert!( + !has_spaces, + "spaces permissions should be excluded when flag is disabled" + ); + + let has_spaces_category = permissions + .iter() + .any(|p| p["category"].as_str().unwrap_or("") == "Spaces"); + assert!( + !has_spaces_category, + "Spaces category should not appear when flag is disabled" + ); + + let templates = body["templates"].as_array().unwrap(); + for template in templates { + let perms = template["permissions"].as_array().unwrap(); + let has_spaces_perm = perms + .iter() + .any(|p| p.as_str().unwrap_or("").starts_with("spaces:")); + assert!( + !has_spaces_perm, + "template {:?} should not contain spaces permissions when flag is disabled", + template["key"] + ); + } +} + +#[tokio::test] +#[serial] +async fn permissions_includes_spaces_when_flag_enabled() { + common::require_db!(); + let app = TestApp::new().await; + + // Enable the spaces feature flag + let resp = app + .router + .clone() + .oneshot(admin_put( + "/admin/settings/feature.spaces_enabled", + app.admin_cookie(), + &json!({ "value": "true" }), + )) + .await + .unwrap(); + assert!(resp.status().is_success()); + + let resp = app + .router + .clone() + .oneshot(admin_get("/admin/permissions", app.admin_cookie())) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + let body = json_body(resp).await; + + let permissions = body["permissions"].as_array().unwrap(); + let has_spaces = permissions + .iter() + .any(|p| p["key"].as_str().unwrap_or("").starts_with("spaces:")); + assert!( + has_spaces, + "spaces permissions should be included when flag is enabled" + ); + + let templates = body["templates"].as_array().unwrap(); + let manager = templates + .iter() + .find(|t| t["key"] == "manager") + .expect("manager template"); + let manager_perms = manager["permissions"].as_array().unwrap(); + let has_spaces_perm = manager_perms + .iter() + .any(|p| p.as_str().unwrap_or("").starts_with("spaces:")); + assert!( + has_spaces_perm, + "manager template should include spaces permissions when flag is enabled" + ); +} + +#[tokio::test] +#[serial] +async fn permissions_spaces_removed_after_disabling_flag() { + common::require_db!(); + let app = TestApp::new().await; + + // Enable + let resp = app + .router + .clone() + .oneshot(admin_put( + "/admin/settings/feature.spaces_enabled", + app.admin_cookie(), + &json!({ "value": "true" }), + )) + .await + .unwrap(); + assert!(resp.status().is_success()); + + // Disable + let resp = app + .router + .clone() + .oneshot(admin_delete( + "/admin/settings/feature.spaces_enabled", + app.admin_cookie(), + )) + .await + .unwrap(); + assert!(resp.status().is_success()); + + let resp = app + .router + .clone() + .oneshot(admin_get("/admin/permissions", app.admin_cookie())) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + let body = json_body(resp).await; + + let permissions = body["permissions"].as_array().unwrap(); + let has_spaces = permissions + .iter() + .any(|p| p["key"].as_str().unwrap_or("").starts_with("spaces:")); + assert!( + !has_spaces, + "spaces permissions should be gone after disabling flag" + ); +} diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts index d923355..f75875b 100644 --- a/web/src/lib/api.ts +++ b/web/src/lib/api.ts @@ -220,6 +220,29 @@ export function transferSuper(body: { target_user_id: string }) { }); } +// Permissions catalog +export type PermissionEntry = { + key: string; + name: string; + description: string; + category: string; +}; + +export type PermissionTemplate = { + key: string; + label: string; + permissions: string[]; +}; + +export type PermissionsCatalog = { + permissions: PermissionEntry[]; + templates: PermissionTemplate[]; +}; + +export function getPermissions() { + return apiFetch("/admin/permissions"); +} + // API Keys export function getApiKeys() { return apiFetch("/admin/api-keys"); -- 2.51.2 From dced56697a5aed830de2e9a6122b916f5c275f9c Mon Sep 17 00:00:00 2001 From: Trezy Date: Tue, 12 May 2026 10:50:18 -0500 Subject: [PATCH 10/15] fix: allow sheets to be much larger Signed-off-by: Trezy --- web/src/app/dashboard/dead-letters/page.tsx | 2 +- web/src/app/dashboard/records/page.tsx | 2 +- web/src/components/ui/sheet.tsx | 4 ++-- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/web/src/app/dashboard/dead-letters/page.tsx b/web/src/app/dashboard/dead-letters/page.tsx index 24ff16f..d871d5d 100644 --- a/web/src/app/dashboard/dead-letters/page.tsx +++ b/web/src/app/dashboard/dead-letters/page.tsx @@ -560,7 +560,7 @@ export default function DeadLettersPage() { if (!open) setViewDetail(null); }} > - + {viewDetail && ( <> diff --git a/web/src/app/dashboard/records/page.tsx b/web/src/app/dashboard/records/page.tsx index 01cbc40..4f0bb02 100644 --- a/web/src/app/dashboard/records/page.tsx +++ b/web/src/app/dashboard/records/page.tsx @@ -442,7 +442,7 @@ export default function RecordsPage() { if (!open) setViewRecord(null); }} > - + {viewRecord && ( <> diff --git a/web/src/components/ui/sheet.tsx b/web/src/components/ui/sheet.tsx index 5963090..cd427e3 100644 --- a/web/src/components/ui/sheet.tsx +++ b/web/src/components/ui/sheet.tsx @@ -62,9 +62,9 @@ function SheetContent({ className={cn( "bg-background data-[state=open]:animate-in data-[state=closed]:animate-out fixed z-50 flex flex-col gap-4 shadow-lg transition ease-in-out data-[state=closed]:duration-300 data-[state=open]:duration-500", side === "right" && - "data-[state=closed]:slide-out-to-right data-[state=open]:slide-in-from-right inset-y-0 right-0 h-full w-3/4 border-l sm:max-w-sm", + "data-[state=closed]:slide-out-to-right data-[state=open]:slide-in-from-right inset-y-0 right-0 h-full w-[90%] border-l lg:w-1/2", side === "left" && - "data-[state=closed]:slide-out-to-left data-[state=open]:slide-in-from-left inset-y-0 left-0 h-full w-3/4 border-r sm:max-w-sm", + "data-[state=closed]:slide-out-to-left data-[state=open]:slide-in-from-left inset-y-0 left-0 h-full w-[90%] border-r lg:w-1/2", side === "top" && "data-[state=closed]:slide-out-to-top data-[state=open]:slide-in-from-top inset-x-0 top-0 h-auto border-b", side === "bottom" && -- 2.51.2 From 7e6a1f111ead434de8f08d001c323ee555c9f313 Mon Sep 17 00:00:00 2001 From: Trezy Date: Tue, 12 May 2026 10:50:57 -0500 Subject: [PATCH 11/15] fix: move user permission management into sheets fixes #23 Signed-off-by: Trezy --- web/src/app/dashboard/settings/users/page.tsx | 438 +++++++++++------- 1 file changed, 266 insertions(+), 172 deletions(-) diff --git a/web/src/app/dashboard/settings/users/page.tsx b/web/src/app/dashboard/settings/users/page.tsx index 84c6d45..e41f8cd 100644 --- a/web/src/app/dashboard/settings/users/page.tsx +++ b/web/src/app/dashboard/settings/users/page.tsx @@ -1,7 +1,7 @@ "use client"; -import React, { useCallback, useEffect, useState } from "react"; -import { ChevronDown, ChevronRight, Shield, Trash2 } from "lucide-react"; +import React, { useCallback, useEffect, useMemo, useState } from "react"; +import { ChevronRight, Search, Shield, Trash2 } from "lucide-react"; import { useAuth } from "@/lib/auth-context"; import { @@ -10,13 +10,16 @@ import { deleteUser, updateUserPermissions, transferSuper, + getPermissions, } from "@/lib/api"; +import type { PermissionEntry, PermissionTemplate } from "@/lib/api"; import type { UserSummary } from "@/types/users"; import { SiteHeader } from "@/components/site-header"; import { Button } from "@/components/ui/button"; import { Input } from "@/components/ui/input"; import { Label } from "@/components/ui/label"; import { Badge } from "@/components/ui/badge"; +import { Card, CardContent } from "@/components/ui/card"; import { Switch } from "@/components/ui/switch"; import { Select, @@ -43,45 +46,61 @@ import { ResponsiveDialogTitle, ResponsiveDialogTrigger, } from "@/components/ui/responsive-dialog"; +import { + Sheet, + SheetContent, + SheetFooter, + SheetHeader, + SheetTitle, + SheetDescription, +} from "@/components/ui/sheet"; -const PERMISSION_CATEGORIES: Record = { - Lexicons: ["lexicons:create", "lexicons:read", "lexicons:delete"], - Records: ["records:read", "records:delete", "records:delete-collection"], - "Script Variables": [ - "script-variables:create", - "script-variables:read", - "script-variables:delete", - ], - Users: ["users:create", "users:read", "users:update", "users:delete"], - "API Keys": ["api-keys:create", "api-keys:read", "api-keys:delete"], - Backfill: ["backfill:create", "backfill:read"], - "API Clients": ["api-clients:view", "api-clients:create", "api-clients:edit", "api-clients:delete"], - Plugins: ["plugins:read", "plugins:create", "plugins:delete"], - System: ["stats:read", "events:read"], +type BskyProfile = { + avatar?: string; + displayName?: string; + description?: string; }; -const ALL_PERMISSIONS = Object.values(PERMISSION_CATEGORIES).flat(); - -const TEMPLATES = [ - { value: "viewer", label: "Viewer" }, - { value: "operator", label: "Operator" }, - { value: "manager", label: "Manager" }, - { value: "full_access", label: "Full Access" }, -] as const; - -const TEMPLATE_PERMISSIONS: Record = { - viewer: ["lexicons:read", "records:read", "script-variables:read", "users:read", "api-keys:read", "backfill:read", "stats:read", "events:read"], - operator: ["lexicons:read", "records:read", "records:delete", "script-variables:read", "script-variables:create", "users:read", "api-keys:read", "backfill:read", "backfill:create", "stats:read", "events:read"], - manager: ["lexicons:create", "lexicons:read", "lexicons:delete", "records:read", "records:delete", "records:delete-collection", "script-variables:create", "script-variables:read", "script-variables:delete", "users:read", "api-keys:read", "backfill:create", "backfill:read", "stats:read", "events:read", "plugins:read", "plugins:create", "plugins:delete"], - full_access: ALL_PERMISSIONS, -}; +function buildCategories(permissions: PermissionEntry[]): Record { + const cats: Record = {}; + for (const p of permissions) { + if (!cats[p.category]) cats[p.category] = []; + cats[p.category].push(p); + } + return cats; +} export default function UsersPage() { const { did: currentDid } = useAuth(); const [users, setUsers] = useState([]); const [handles, setHandles] = useState>({}); const [error, setError] = useState(null); - const [expandedUserId, setExpandedUserId] = useState(null); + const [selectedUserId, setSelectedUserId] = useState(null); + const [permSearch, setPermSearch] = useState(""); + const [permissionEntries, setPermissionEntries] = useState([]); + const [profiles, setProfiles] = useState>({}); + const [templates, setTemplates] = useState([]); + + const permissionCategories = React.useMemo(() => buildCategories(permissionEntries), [permissionEntries]); + const filteredCategories = useMemo(() => { + if (!permSearch.trim()) return permissionCategories; + const terms = permSearch.toLowerCase().split(/\s+/); + const result: Record = {}; + for (const [category, permissions] of Object.entries(permissionCategories)) { + const matched = permissions.filter((p) => { + const haystack = `${p.name} ${p.description} ${p.category} ${p.key}`.toLowerCase(); + return terms.every((term) => haystack.includes(term)); + }); + if (matched.length > 0) result[category] = matched; + } + return result; + }, [permissionCategories, permSearch]); + const allPermissionKeys = React.useMemo(() => permissionEntries.map((p) => p.key), [permissionEntries]); + const templatePermissions = React.useMemo(() => { + const map: Record = {}; + for (const t of templates) map[t.key] = t.permissions; + return map; + }, [templates]); const currentUser = users.find((u) => u.did === currentDid); const isCurrentUserSuper = currentUser?.is_super ?? false; @@ -94,6 +113,12 @@ export default function UsersPage() { useEffect(() => { load(); + getPermissions() + .then((catalog) => { + setPermissionEntries(catalog.permissions); + setTemplates(catalog.templates); + }) + .catch((e) => setError(e instanceof Error ? e.message : String(e))); }, [load]); // Resolve DIDs to handles via PLC directory @@ -116,6 +141,27 @@ export default function UsersPage() { } }, [users, handles]); + // Fetch Bluesky profile when a user is selected + useEffect(() => { + if (!selectedUserId) return; + const user = users.find((u) => u.id === selectedUserId); + if (!user || user.did in profiles) return; + fetch(`https://public.api.bsky.app/xrpc/app.bsky.actor.getProfile?actor=${encodeURIComponent(user.did)}`) + .then((res) => (res.ok ? res.json() : null)) + .then((data) => { + if (!data) return; + setProfiles((prev) => ({ + ...prev, + [user.did]: { + avatar: data.avatar, + displayName: data.displayName, + description: data.description, + }, + })); + }) + .catch(() => {}); + }, [selectedUserId, users, profiles]); + async function handleDelete(id: string) { try { await deleteUser(id); @@ -193,7 +239,7 @@ export default function UsersPage() {

Users

{(isCurrentUserSuper || currentUser?.permissions.includes("users:create")) && ( - + )}
@@ -201,19 +247,18 @@ export default function UsersPage() { - User Permissions Created Last Used - + {users.length === 0 && ( No users yet. @@ -221,125 +266,169 @@ export default function UsersPage() { )} {users.map((user) => ( - - - -
+
+ + {(() => { + const selectedUser = users.find((u) => u.id === selectedUserId); + return ( + { if (!open) { setSelectedUserId(null); setPermSearch(""); } }}> + + {selectedUser && ( + <> + + User + + + + + {profiles[selectedUser.did]?.avatar && ( + + )} +
+

+ {profiles[selectedUser.did]?.displayName || handles[selectedUser.did] ? ( + <> + {profiles[selectedUser.did]?.displayName && ( + {profiles[selectedUser.did].displayName} + )} + {handles[selectedUser.did] && ( + + @{handles[selectedUser.did]} + + )} + ) : ( - - )} - - - - setExpandedUserId( - expandedUserId === user.id ? null : user.id - ) - } - > -

-
- {handles[user.did] && ( - @{handles[user.did]} + {selectedUser.did} )} - {user.did} -
- {user.is_super && ( - - Owner - - )} -
- - - setExpandedUserId( - expandedUserId === user.id ? null : user.id - ) - } - > - {user.is_super - ? `${ALL_PERMISSIONS.length}/${ALL_PERMISSIONS.length}` - : `${user.permissions.length}/${ALL_PERMISSIONS.length}`} - - - setExpandedUserId( - expandedUserId === user.id ? null : user.id - ) - } - > - {new Date(user.created_at).toLocaleString()} - - - setExpandedUserId( - expandedUserId === user.id ? null : user.id - ) - } - > - {user.last_used_at - ? new Date(user.last_used_at).toLocaleString() - : "Never"} - - -
- {isCurrentUserSuper && ( - handleTransferSuper(user.id)} - /> +

+

{selectedUser.did}

+ {profiles[selectedUser.did]?.description && ( +

{profiles[selectedUser.did].description}

)} -
- - - {expandedUserId === user.id && ( - - - + + +
+
+ Role +

+ {selectedUser.is_super ? ( + Owner + ) : "Member"} +

+
+
+ Permissions +

+ {selectedUser.is_super + ? `${allPermissionKeys.length}/${allPermissionKeys.length}` + : `${selectedUser.permissions.filter((p) => allPermissionKeys.includes(p)).length}/${allPermissionKeys.length}`} +

+
+
+ Created +

{new Date(selectedUser.created_at).toLocaleString()}

+
+
+ Last Active +

+ {selectedUser.last_used_at + ? new Date(selectedUser.last_used_at).toLocaleString() + : "Never"} +

+
+
+ +
+ +
+ + setPermSearch(e.target.value)} + className="pl-9" + /> +
+ +
+ +
+ + +
+ {isCurrentUserSuper && ( + handleTransferSuper(selectedUser.id)} /> - - - )} - - ))} - - -
+ )} + +
+ + + )} + + + ); + })()} ); @@ -350,47 +439,50 @@ function PermissionsPanel({ isSelf, currentUserPermissions, isCurrentUserSuper, + filteredCategories, onToggle, }: { user: UserSummary; isSelf: boolean; currentUserPermissions: string[]; isCurrentUserSuper: boolean; + filteredCategories: Record; onToggle: (user: UserSummary, permission: string, enabled: boolean) => void; }) { const canUpdate = isCurrentUserSuper || currentUserPermissions.includes("users:update"); return ( -
- {Object.entries(PERMISSION_CATEGORIES).map(([category, permissions]) => ( -
+
+ {Object.entries(filteredCategories).map(([category, permissions]) => ( +

{category}

-
+
{permissions.map((perm) => { - const enabled = user.is_super || user.permissions.includes(perm); + const enabled = user.is_super || user.permissions.includes(perm.key); return ( -
+
- onToggle(user, perm, checked) + onToggle(user, perm.key, checked) } - className="scale-75" + className="mt-0.5 scale-75" />
); @@ -422,14 +514,12 @@ function TransferOwnershipDialog({ @@ -457,8 +547,12 @@ function TransferOwnershipDialog({ function AddUserDialog({ onSuccess, + templates, + templatePermissions, }: { onSuccess: () => void; + templates: PermissionTemplate[]; + templatePermissions: Record; }) { const [did, setDid] = useState(""); const [template, setTemplate] = useState(""); @@ -511,8 +605,8 @@ function AddUserDialog({ - {TEMPLATES.map((t) => ( - + {templates.map((t) => ( + {t.label} ))} @@ -521,7 +615,7 @@ function AddUserDialog({
{template && (

- Grants {TEMPLATE_PERMISSIONS[template]?.length ?? 0} permissions. + Grants {templatePermissions[template]?.length ?? 0} permissions.

)}
-- 2.51.2 From b258408121613bc9d6f29587a23011c4872cc4be Mon Sep 17 00:00:00 2001 From: Trezy Date: Tue, 12 May 2026 11:12:49 -0500 Subject: [PATCH 12/15] fix: use correct permission IDs in dashboard fixes #24 Signed-off-by: Trezy --- web/src/app/dashboard/settings/users/page.tsx | 172 +++++++++++++----- web/src/components/ui/sonner.tsx | 4 + 2 files changed, 132 insertions(+), 44 deletions(-) diff --git a/web/src/app/dashboard/settings/users/page.tsx b/web/src/app/dashboard/settings/users/page.tsx index e41f8cd..6bb9f43 100644 --- a/web/src/app/dashboard/settings/users/page.tsx +++ b/web/src/app/dashboard/settings/users/page.tsx @@ -2,6 +2,7 @@ import React, { useCallback, useEffect, useMemo, useState } from "react"; import { ChevronRight, Search, Shield, Trash2 } from "lucide-react"; +import { toast } from "sonner"; import { useAuth } from "@/lib/auth-context"; import { @@ -76,6 +77,8 @@ export default function UsersPage() { const [handles, setHandles] = useState>({}); const [error, setError] = useState(null); const [selectedUserId, setSelectedUserId] = useState(null); + const [pendingPermissions, setPendingPermissions] = useState([]); + const [saving, setSaving] = useState(false); const [permSearch, setPermSearch] = useState(""); const [permissionEntries, setPermissionEntries] = useState([]); const [profiles, setProfiles] = useState>({}); @@ -141,6 +144,13 @@ export default function UsersPage() { } }, [users, handles]); + // Initialize pending permissions when a user is selected + useEffect(() => { + if (!selectedUserId) return; + const user = users.find((u) => u.id === selectedUserId); + if (user) setPendingPermissions([...user.permissions]); + }, [selectedUserId, users]); + // Fetch Bluesky profile when a user is selected useEffect(() => { if (!selectedUserId) return; @@ -171,53 +181,59 @@ export default function UsersPage() { } } - async function handleTogglePermission( - user: UserSummary, + function handleTogglePermission( + _user: UserSummary, permission: string, enabled: boolean ) { - const grant: string[] = []; - const revoke: string[] = []; - - const [ns, action] = permission.split(":"); - - if (enabled) { - grant.push(permission); - // Adding a write permission also enables its read counterpart - if (action === "create" || action === "update" || action === "delete") { - const readPerm = `${ns}:read`; - if (!user.permissions.includes(readPerm)) { - grant.push(readPerm); - } - } - // Adding records:delete-collection also enables records:delete - if (permission === "records:delete-collection" && !user.permissions.includes("records:delete")) { - grant.push("records:delete"); - } - } else { - revoke.push(permission); - // Removing read also removes all write permissions in the same namespace - if (action === "read") { - for (const p of user.permissions) { - if (p.startsWith(`${ns}:`) && p !== permission) { - revoke.push(p); + setPendingPermissions((prev) => { + const perms = new Set(prev); + const [ns, action] = permission.split(":"); + + const nsReadPerm = allPermissionKeys.find( + (k) => k.startsWith(`${ns}:`) && (k.endsWith(":read") || k.endsWith(":view")) + ); + const isReadAction = action === "read" || action === "view"; + + if (enabled) { + perms.add(permission); + if (!isReadAction && nsReadPerm) perms.add(nsReadPerm); + if (permission === "records:delete-collection") perms.add("records:delete"); + } else { + perms.delete(permission); + if (isReadAction) { + for (const p of prev) { + if (p.startsWith(`${ns}:`) && p !== permission) perms.delete(p); } } + if (permission === "records:delete") perms.delete("records:delete-collection"); } - // Removing records:delete also removes records:delete-collection - if (permission === "records:delete" && user.permissions.includes("records:delete-collection")) { - revoke.push("records:delete-collection"); - } - } + return [...perms]; + }); + } + + async function handleSavePermissions(userId: string, originalPermissions: string[]) { + const originalSet = new Set(originalPermissions); + const pendingSet = new Set(pendingPermissions); + + const grant = pendingPermissions.filter((p) => !originalSet.has(p)); + const revoke = originalPermissions.filter((p) => !pendingSet.has(p)); + + if (grant.length === 0 && revoke.length === 0) return; + + setSaving(true); try { const body: { grant?: string[]; revoke?: string[] } = {}; if (grant.length > 0) body.grant = grant; if (revoke.length > 0) body.revoke = revoke; - await updateUserPermissions(user.id, body); + await updateUserPermissions(userId, body); + toast.success("Permissions updated"); load(); } catch (e: unknown) { setError(e instanceof Error ? e.message : String(e)); + } finally { + setSaving(false); } } @@ -307,8 +323,27 @@ export default function UsersPage() { {(() => { const selectedUser = users.find((u) => u.id === selectedUserId); return ( - { if (!open) { setSelectedUserId(null); setPermSearch(""); } }}> - + { + if (!open) { + const user = users.find((u) => u.id === selectedUserId); + if (user) { + const origSet = new Set(user.permissions); + const pendSet = new Set(pendingPermissions); + const unsaved = pendingPermissions.some((p) => !origSet.has(p)) || user.permissions.some((p) => !pendSet.has(p)); + if (unsaved) { + toast.warning("You have unsaved changes. Save or cancel before closing."); + return; + } + } + setSelectedUserId(null); + setPermSearch(""); + } + }}> + { + if (e.target instanceof HTMLElement && e.target.closest("[data-sonner-toaster]")) { + e.preventDefault(); + } + }}> {selectedUser && ( <> @@ -349,6 +384,14 @@ export default function UsersPage() { + {(() => { + const originalSet = new Set(selectedUser.permissions); + const pendingSet = new Set(pendingPermissions); + const added = pendingPermissions.filter((p) => !originalSet.has(p)).length; + const removed = selectedUser.permissions.filter((p) => !pendingSet.has(p)).length; + const hasChanges = added > 0 || removed > 0; + return ( + <>
Role @@ -363,7 +406,14 @@ export default function UsersPage() {

{selectedUser.is_super ? `${allPermissionKeys.length}/${allPermissionKeys.length}` - : `${selectedUser.permissions.filter((p) => allPermissionKeys.includes(p)).length}/${allPermissionKeys.length}`} + : `${pendingPermissions.filter((p) => allPermissionKeys.includes(p)).length}/${allPermissionKeys.length}`} + {hasChanges && ( + + {added > 0 && +{added}} + {added > 0 && removed > 0 && " "} + {removed > 0 && -{removed}} + + )}

@@ -399,12 +449,23 @@ export default function UsersPage() { currentUserPermissions={currentUser?.permissions ?? []} isCurrentUserSuper={isCurrentUserSuper} filteredCategories={filteredCategories} + pendingPermissions={pendingPermissions} + originalPermissions={selectedUser.permissions} onToggle={handleTogglePermission} />
-
+
+ {isCurrentUserSuper && ( handleTransferSuper(selectedUser.id)} /> )} +
+
+
+ + ); + })()} )} @@ -440,6 +512,8 @@ function PermissionsPanel({ currentUserPermissions, isCurrentUserSuper, filteredCategories, + pendingPermissions, + originalPermissions, onToggle, }: { user: UserSummary; @@ -447,9 +521,12 @@ function PermissionsPanel({ currentUserPermissions: string[]; isCurrentUserSuper: boolean; filteredCategories: Record; + pendingPermissions: string[]; + originalPermissions: string[]; onToggle: (user: UserSummary, permission: string, enabled: boolean) => void; }) { const canUpdate = isCurrentUserSuper || currentUserPermissions.includes("users:update"); + const originalSet = new Set(originalPermissions); return (
@@ -460,7 +537,10 @@ function PermissionsPanel({

{permissions.map((perm) => { - const enabled = user.is_super || user.permissions.includes(perm.key); + const enabled = user.is_super || pendingPermissions.includes(perm.key); + const wasEnabled = user.is_super || originalSet.has(perm.key); + const isAdded = enabled && !wasEnabled; + const isRemoved = !enabled && wasEnabled; return (
- {perm.name} + + {perm.name} + {isAdded && } + {isRemoved && } + {perm.description}
diff --git a/web/src/components/ui/sonner.tsx b/web/src/components/ui/sonner.tsx index 9b20afe..7a41a46 100644 --- a/web/src/components/ui/sonner.tsx +++ b/web/src/components/ui/sonner.tsx @@ -17,6 +17,7 @@ const Toaster = ({ ...props }: ToasterProps) => { , info: , @@ -30,6 +31,9 @@ const Toaster = ({ ...props }: ToasterProps) => { "--normal-text": "var(--popover-foreground)", "--normal-border": "var(--border)", "--border-radius": "var(--radius)", + "--toast-close-button-start": "unset", + "--toast-close-button-end": "0", + "--toast-close-button-transform": "translate(35%, -35%)", } as React.CSSProperties } {...props} -- 2.51.2 From 274ab01d24a71f5fddb8da1c65c02623e4603104 Mon Sep 17 00:00:00 2001 From: Trezy Date: Thu, 14 May 2026 11:09:32 -0500 Subject: [PATCH 13/15] fix: bulk insert in discovery, seed records on resume, dedupe retry helper, add cancel tests Signed-off-by: Trezy --- src/admin/backfill.rs | 78 +++++++++++++------------ src/http_retry.rs | 23 ++++++++ src/lib.rs | 1 + src/profile.rs | 22 +------ tests/e2e_admin.rs | 131 ++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 198 insertions(+), 57 deletions(-) create mode 100644 src/http_retry.rs diff --git a/src/admin/backfill.rs b/src/admin/backfill.rs index d0fc378..7534db5 100644 --- a/src/admin/backfill.rs +++ b/src/admin/backfill.rs @@ -14,33 +14,10 @@ use crate::AppState; use crate::db::{adapt_sql, now_rfc3339}; use crate::error::AppError; use crate::event_log::{EventLog, Severity, log_event}; +use crate::http_retry::parse_retry_after; use crate::profile; use crate::record_handler::{self, RecordEvent}; -/// Parse rate-limit sleep duration from response headers. -/// Checks `RateLimit-Reset` (Unix timestamp, used by XRPC servers) first, -/// then `retry-after` (seconds), defaulting to 5s. -fn parse_retry_after(headers: &reqwest::header::HeaderMap) -> u64 { - if let Some(reset) = headers - .get("ratelimit-reset") - .and_then(|v| v.to_str().ok()) - .and_then(|v| v.parse::().ok()) - { - let now = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_secs() as i64; - let wait = (reset - now).max(1) as u64; - return wait.min(120); - } - - headers - .get("retry-after") - .and_then(|v| v.to_str().ok()) - .and_then(|v| v.parse::().ok()) - .unwrap_or(5) -} - use super::auth::UserAuth; use super::permissions::Permission; use super::types::{BackfillJob, CreateBackfillBody}; @@ -249,6 +226,7 @@ async fn discover_repos_from_relay( ) -> Result<(), String> { let base = state.config.relay_url.trim_end_matches('/'); let mut cursor: Option = None; + let mut running_total: i32 = count_repos(state, job_id).await; loop { let mut url = format!( @@ -287,20 +265,34 @@ async fn discover_repos_from_relay( let page_count = body.repos.len(); - for repo in &body.repos { - let sql = adapt_sql( - "INSERT INTO backfill_repos (job_id, did) VALUES (?, ?) ON CONFLICT DO NOTHING", - state.db_backend, + if !body.repos.is_empty() { + let base_sql = "INSERT INTO backfill_repos (job_id, did) VALUES "; + let placeholders: Vec = body + .repos + .iter() + .enumerate() + .map(|(i, _)| { + if state.db_backend == crate::db::DatabaseBackend::Postgres { + format!("(${}, ${})", i * 2 + 1, i * 2 + 2) + } else { + "(?, ?)".to_string() + } + }) + .collect(); + let sql = format!( + "{base_sql}{} ON CONFLICT DO NOTHING", + placeholders.join(", ") ); - let _ = sqlx::query(&sql) - .bind(job_id) - .bind(&repo.did) - .execute(&state.db) - .await; + + let mut query = sqlx::query(&sql); + for repo in &body.repos { + query = query.bind(job_id).bind(&repo.did); + } + let _ = query.execute(&state.db).await; } - let total = count_repos(state, job_id).await; - update_job_counter(state, job_id, "total_repos", total).await; + running_total += page_count as i32; + update_job_counter(state, job_id, "total_repos", running_total).await; if is_cancelled(state, job_id).await { return Ok(()); @@ -413,8 +405,22 @@ async fn run_fetching_phase(state: &AppState, job_id: &str, collections: &[Strin // Reset processed_repos for the fetching phase update_job_counter(state, job_id, "processed_repos", already_completed).await; + // Seed total_records from DB so a resumed job doesn't lose its prior count + let existing_records: i32 = { + let sql = adapt_sql( + "SELECT total_records FROM backfill_jobs WHERE id = ?", + state.db_backend, + ); + sqlx::query_as::<_, (Option,)>(&sql) + .bind(job_id) + .fetch_one(&state.db) + .await + .map(|(c,)| c.unwrap_or(0)) + .unwrap_or(0) + }; + let processed_repos = Arc::new(AtomicI32::new(already_completed)); - let total_records = Arc::new(AtomicI32::new(0)); + let total_records = Arc::new(AtomicI32::new(existing_records)); let cancelled = Arc::new(AtomicBool::new(false)); let state = Arc::new(state.clone()); let collections = Arc::new(collections.to_vec()); diff --git a/src/http_retry.rs b/src/http_retry.rs new file mode 100644 index 0000000..67c9898 --- /dev/null +++ b/src/http_retry.rs @@ -0,0 +1,23 @@ +/// Parse rate-limit sleep duration from response headers. +/// Checks `RateLimit-Reset` (Unix timestamp, used by XRPC servers) first, +/// then `retry-after` (seconds), defaulting to 5s. +pub fn parse_retry_after(headers: &reqwest::header::HeaderMap) -> u64 { + if let Some(reset) = headers + .get("ratelimit-reset") + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.parse::().ok()) + { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() as i64; + let wait = (reset - now).max(1) as u64; + return wait.min(120); + } + + headers + .get("retry-after") + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.parse::().ok()) + .unwrap_or(5) +} diff --git a/src/lib.rs b/src/lib.rs index d467b65..fe19b38 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -12,6 +12,7 @@ pub mod event_log; pub mod external_auth; pub mod feature_flags; pub mod feature_middleware; +pub mod http_retry; pub mod jetstream; pub mod labeler; pub mod lexicon; diff --git a/src/profile.rs b/src/profile.rs index 298ea01..1a1d350 100644 --- a/src/profile.rs +++ b/src/profile.rs @@ -1,27 +1,7 @@ use serde::{Deserialize, Serialize}; use crate::error::AppError; - -fn parse_retry_after(headers: &reqwest::header::HeaderMap) -> u64 { - if let Some(reset) = headers - .get("ratelimit-reset") - .and_then(|v| v.to_str().ok()) - .and_then(|v| v.parse::().ok()) - { - let now = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_secs() as i64; - let wait = (reset - now).max(1) as u64; - return wait.min(120); - } - - headers - .get("retry-after") - .and_then(|v| v.to_str().ok()) - .and_then(|v| v.parse::().ok()) - .unwrap_or(5) -} +use crate::http_retry::parse_retry_after; #[derive(Serialize)] pub struct Profile { diff --git a/tests/e2e_admin.rs b/tests/e2e_admin.rs index 0911bc8..8952859 100644 --- a/tests/e2e_admin.rs +++ b/tests/e2e_admin.rs @@ -561,6 +561,137 @@ async fn backfill_list_jobs() { assert_eq!(json.as_array().unwrap().len(), 1); } +#[tokio::test] +#[serial] +async fn backfill_cancel_running_job() { + common::require_db!(); + let app = TestApp::new().await; + let backend = app.state.db_backend; + + // Insert a running job directly so we don't need a real relay. + let job_id = uuid::Uuid::new_v4().to_string(); + let now = now_rfc3339(); + let sql = adapt_sql( + "INSERT INTO backfill_jobs (id, status, stage, started_at, created_at) VALUES (?, 'running', 'discovering_repos', ?, ?)", + backend, + ); + sqlx::query(&sql) + .bind(&job_id) + .bind(&now) + .bind(&now) + .execute(&app.state.db) + .await + .unwrap(); + + let resp = app + .router + .clone() + .oneshot(admin_post( + &format!("/admin/backfill/{job_id}/cancel"), + app.admin_cookie(), + &json!({}), + )) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + let json = json_body(resp).await; + assert_eq!(json["id"], job_id); + assert_eq!(json["status"], "cancelling"); +} + +#[tokio::test] +#[serial] +async fn backfill_cancel_already_cancelling_is_idempotent() { + common::require_db!(); + let app = TestApp::new().await; + let backend = app.state.db_backend; + + let job_id = uuid::Uuid::new_v4().to_string(); + let now = now_rfc3339(); + let sql = adapt_sql( + "INSERT INTO backfill_jobs (id, status, stage, started_at, created_at) VALUES (?, 'cancelling', 'fetching_records', ?, ?)", + backend, + ); + sqlx::query(&sql) + .bind(&job_id) + .bind(&now) + .bind(&now) + .execute(&app.state.db) + .await + .unwrap(); + + let resp = app + .router + .clone() + .oneshot(admin_post( + &format!("/admin/backfill/{job_id}/cancel"), + app.admin_cookie(), + &json!({}), + )) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + let json = json_body(resp).await; + assert_eq!(json["status"], "cancelling"); +} + +#[tokio::test] +#[serial] +async fn backfill_cancel_completed_returns_400() { + common::require_db!(); + let app = TestApp::new().await; + let backend = app.state.db_backend; + + let job_id = uuid::Uuid::new_v4().to_string(); + let now = now_rfc3339(); + let sql = adapt_sql( + "INSERT INTO backfill_jobs (id, status, stage, completed_at, created_at) VALUES (?, 'completed', 'completed', ?, ?)", + backend, + ); + sqlx::query(&sql) + .bind(&job_id) + .bind(&now) + .bind(&now) + .execute(&app.state.db) + .await + .unwrap(); + + let resp = app + .router + .clone() + .oneshot(admin_post( + &format!("/admin/backfill/{job_id}/cancel"), + app.admin_cookie(), + &json!({}), + )) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); +} + +#[tokio::test] +#[serial] +async fn backfill_cancel_not_found_returns_404() { + common::require_db!(); + let app = TestApp::new().await; + + let resp = app + .router + .clone() + .oneshot(admin_post( + "/admin/backfill/nonexistent-id/cancel", + app.admin_cookie(), + &json!({}), + )) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::NOT_FOUND); +} + // --------------------------------------------------------------------------- // Admin management // --------------------------------------------------------------------------- -- 2.51.2 From f25ade7675cc356babba5ce05030f64e37f4e1df Mon Sep 17 00:00:00 2001 From: Trezy Date: Thu, 14 May 2026 11:39:45 -0500 Subject: [PATCH 14/15] fix: chunk bulk inserts for sqlite, use rows_affected for counts, faster cancel checks, users page a11y Signed-off-by: Trezy --- src/admin/backfill.rs | 55 +++++++++++-------- web/src/app/dashboard/settings/users/page.tsx | 17 +++++- 2 files changed, 46 insertions(+), 26 deletions(-) diff --git a/src/admin/backfill.rs b/src/admin/backfill.rs index 7534db5..238b796 100644 --- a/src/admin/backfill.rs +++ b/src/admin/backfill.rs @@ -266,32 +266,41 @@ async fn discover_repos_from_relay( let page_count = body.repos.len(); if !body.repos.is_empty() { - let base_sql = "INSERT INTO backfill_repos (job_id, did) VALUES "; - let placeholders: Vec = body - .repos - .iter() - .enumerate() - .map(|(i, _)| { - if state.db_backend == crate::db::DatabaseBackend::Postgres { - format!("(${}, ${})", i * 2 + 1, i * 2 + 2) - } else { - "(?, ?)".to_string() - } - }) - .collect(); - let sql = format!( - "{base_sql}{} ON CONFLICT DO NOTHING", - placeholders.join(", ") - ); + // SQLite has a 999 bound-parameter limit; each row uses 2 params + let chunk_size = if state.db_backend == crate::db::DatabaseBackend::Sqlite { + 499 + } else { + 1000 + }; + + for chunk in body.repos.chunks(chunk_size) { + let base_sql = "INSERT INTO backfill_repos (job_id, did) VALUES "; + let placeholders: Vec = chunk + .iter() + .enumerate() + .map(|(i, _)| { + if state.db_backend == crate::db::DatabaseBackend::Postgres { + format!("(${}, ${})", i * 2 + 1, i * 2 + 2) + } else { + "(?, ?)".to_string() + } + }) + .collect(); + let sql = format!( + "{base_sql}{} ON CONFLICT DO NOTHING", + placeholders.join(", ") + ); - let mut query = sqlx::query(&sql); - for repo in &body.repos { - query = query.bind(job_id).bind(&repo.did); + let mut query = sqlx::query(&sql); + for repo in chunk { + query = query.bind(job_id).bind(&repo.did); + } + if let Ok(result) = query.execute(&state.db).await { + running_total += result.rows_affected() as i32; + } } - let _ = query.execute(&state.db).await; } - running_total += page_count as i32; update_job_counter(state, job_id, "total_repos", running_total).await; if is_cancelled(state, job_id).await { @@ -491,7 +500,7 @@ async fn run_fetching_phase(state: &AppState, job_id: &str, collections: &[Strin let repos = processed_repos.fetch_add(1, Ordering::Relaxed) + 1; - if repos % 100 == 0 { + if repos % 10 == 0 { let records = total_records.load(Ordering::Relaxed); let backend = state.db_backend; let sql = adapt_sql( diff --git a/web/src/app/dashboard/settings/users/page.tsx b/web/src/app/dashboard/settings/users/page.tsx index 6bb9f43..81e7d3d 100644 --- a/web/src/app/dashboard/settings/users/page.tsx +++ b/web/src/app/dashboard/settings/users/page.tsx @@ -53,7 +53,6 @@ import { SheetFooter, SheetHeader, SheetTitle, - SheetDescription, } from "@/components/ui/sheet"; type BskyProfile = { @@ -282,7 +281,19 @@ export default function UsersPage() { )} {users.map((user) => ( - setSelectedUserId(user.id)}> + setSelectedUserId(user.id)} + onKeyDown={(e) => { + if (e.key === "Enter" || e.key === " ") { + e.preventDefault(); + setSelectedUserId(user.id); + } + }} + >
@@ -433,7 +444,7 @@ export default function UsersPage() {
- + Date: Thu, 14 May 2026 11:53:24 -0500 Subject: [PATCH 15/15] fix: only count successful PDS resolutions, cap retry-after, update backfill docs Signed-off-by: Trezy --- packages/docs/content/docs/guides/backfill.md | 4 ++-- src/admin/backfill.rs | 6 ++++-- src/http_retry.rs | 1 + 3 files changed, 7 insertions(+), 4 deletions(-) diff --git a/packages/docs/content/docs/guides/backfill.md b/packages/docs/content/docs/guides/backfill.md index da7cc42..642343a 100644 --- a/packages/docs/content/docs/guides/backfill.md +++ b/packages/docs/content/docs/guides/backfill.md @@ -34,7 +34,7 @@ A backfill job has both a `status` (overall state) and a `stage` (current phase) | `running` | Job is actively processing | | `cancelling` | Cancel requested, waiting for the worker to stop | | `cancelled` | Worker has stopped and cleaned up | -| `completed` | All repos processed successfully | +| `completed` | Worker finished processing all resolvable repos | | `failed` | An error occurred | The `stage` field tracks which phase the job is in: `pending`, `discovering_repos`, `resolving_pds`, `fetching_records`, `completed`, `failed`, or `cancelled`. @@ -44,7 +44,7 @@ The `stage` field tracks which phase the job is in: `pending`, `discovering_repo Running jobs can be cancelled via `POST /admin/backfill/{id}/cancel` or the Cancel button in the dashboard. Cancellation is two-phase: 1. The endpoint sets the job status to `cancelling`. -2. The worker checks for cancellation at natural checkpoints (between relay pages, every 100 DIDs during resolution, every 100 repos during fetching). When it detects the `cancelling` status, it stops work and sets the final status to `cancelled`. +2. The worker checks for cancellation at natural checkpoints (between relay pages, every 100 DIDs during resolution, every 10 repos during fetching). When it detects the `cancelling` status, it stops work and sets the final status to `cancelled`. This means there may be a short delay between clicking Cancel and the job fully stopping, depending on what the worker is doing at that moment. diff --git a/src/admin/backfill.rs b/src/admin/backfill.rs index 238b796..553c3a5 100644 --- a/src/admin/backfill.rs +++ b/src/admin/backfill.rs @@ -346,6 +346,7 @@ async fn run_resolution_phase(state: &AppState, job_id: &str) { let mut resolved_count = already_resolved; + let mut attempted = already_resolved; for (did,) in &unresolved { match profile::resolve_pds_endpoint(&state.http, &state.config.plc_url, did).await { Ok(pds) => { @@ -359,13 +360,14 @@ async fn run_resolution_phase(state: &AppState, job_id: &str) { .bind(did) .execute(&state.db) .await; + resolved_count += 1; } Err(e) => { tracing::warn!(did, error = %e, "failed to resolve PDS endpoint, skipping DID"); } } - resolved_count += 1; - if resolved_count % 100 == 0 { + attempted += 1; + if attempted % 100 == 0 { update_job_counter(state, job_id, "processed_repos", resolved_count).await; if is_cancelled(state, job_id).await { return; diff --git a/src/http_retry.rs b/src/http_retry.rs index 67c9898..c1edb28 100644 --- a/src/http_retry.rs +++ b/src/http_retry.rs @@ -20,4 +20,5 @@ pub fn parse_retry_after(headers: &reqwest::header::HeaderMap) -> u64 { .and_then(|v| v.to_str().ok()) .and_then(|v| v.parse::().ok()) .unwrap_or(5) + .min(120) } -- 2.51.2