From 66fe12ac0ea1d2f87c6642fedcd4d56027b272d2 Mon Sep 17 00:00:00 2001 From: Trezy Date: Thu, 21 May 2026 11:32:30 -0500 Subject: [PATCH] feat: add more detailed information to backfill UI Signed-off-by: Trezy --- Cargo.lock | 39 +- Cargo.toml | 1 + .../20260521000000_backfill_diagnostics.sql | 2 + .../20260521000000_backfill_diagnostics.sql | 2 + src/admin/backfill.rs | 396 +++++++++++- src/admin/mod.rs | 14 + src/admin/settings.rs | 1 + src/admin/types.rs | 71 ++ src/lib.rs | 1 + src/lua/atproto_api.rs | 1 + src/lua/db_api.rs | 1 + src/lua/execute.rs | 1 + src/lua/http_api.rs | 1 + src/lua/xrpc_api.rs | 1 + src/main.rs | 10 + tests/common/app.rs | 1 + tests/lua_atproto_api.rs | 1 + tests/lua_db_api.rs | 1 + web/package-lock.json | 28 + web/package.json | 1 + web/src/app/dashboard/backfill/page.tsx | 610 ++++++++++++++++-- .../app/dashboard/settings/general/page.tsx | 44 ++ web/src/lib/api.ts | 32 +- web/src/types/backfill.ts | 45 ++ 24 files changed, 1231 insertions(+), 74 deletions(-) create mode 100644 migrations/postgres/20260521000000_backfill_diagnostics.sql create mode 100644 migrations/sqlite/20260521000000_backfill_diagnostics.sql diff --git a/Cargo.lock b/Cargo.lock index 29026e6..69eeee1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -157,6 +157,28 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "async-stream" +version = "0.3.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b5a71a6f37880a80d1d7f19efd781e4b5de42c88f0722cc13bcb6cc2cfe8476" +dependencies = [ + "async-stream-impl", + "futures-core", + "pin-project-lite", +] + +[[package]] +name = "async-stream-impl" +version = "0.3.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7c24de15d275a1ecfd47a380fb4d5ec9bfe0933f309ed5e705b775596a3574d" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "async-trait" version = "0.1.89" @@ -483,7 +505,7 @@ dependencies = [ "cap-primitives", "cap-std", "io-lifetimes", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -512,7 +534,7 @@ dependencies = [ "maybe-owned", "rustix 1.1.3", "rustix-linux-procfs", - "windows-sys 0.52.0", + "windows-sys 0.59.0", "winx", ] @@ -1293,7 +1315,7 @@ checksum = "0ce92ff622d6dadf7349484f42c93271a0d49b7cc4d466a936405bacbe10aa78" dependencies = [ "cfg-if", "rustix 1.1.3", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -1377,7 +1399,7 @@ checksum = "94e7099f6313ecacbe1256e8ff9d617b75d1bcb16a6fddef94866d225a01a14a" dependencies = [ "io-lifetimes", "rustix 1.1.3", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -1626,6 +1648,7 @@ dependencies = [ "aes-gcm", "anyhow", "arc-swap", + "async-stream", "atrium-api", "atrium-common", "atrium-identity", @@ -2095,7 +2118,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2285ddfe3054097ef4b2fe909ef8c3bcd1ea52a8f0d274416caebeef39f04a65" dependencies = [ "io-lifetimes", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -3264,7 +3287,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.4.15", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -3975,7 +3998,7 @@ dependencies = [ "fd-lock", "io-lifetimes", "rustix 0.38.44", - "windows-sys 0.52.0", + "windows-sys 0.59.0", "winx", ] @@ -5544,7 +5567,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f3fd376f71958b862e7afb20cfe5a22830e1963462f3a17f49d82a6c1d1f42d" dependencies = [ "bitflags", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index aaf858d..40bc34f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -55,6 +55,7 @@ wasmtime = { version = "29", features = ["async"] } wasmtime-wasi = "29" regex = "1.12.3" semver = "1.0" +async-stream = "0.3.6" [[bin]] name = "migrate-lua-sql" diff --git a/migrations/postgres/20260521000000_backfill_diagnostics.sql b/migrations/postgres/20260521000000_backfill_diagnostics.sql new file mode 100644 index 0000000..f80f7c7 --- /dev/null +++ b/migrations/postgres/20260521000000_backfill_diagnostics.sql @@ -0,0 +1,2 @@ +ALTER TABLE backfill_repos ADD COLUMN records_fetched INTEGER NOT NULL DEFAULT 0; +CREATE INDEX IF NOT EXISTS idx_backfill_repos_job_status ON backfill_repos (job_id, status); diff --git a/migrations/sqlite/20260521000000_backfill_diagnostics.sql b/migrations/sqlite/20260521000000_backfill_diagnostics.sql new file mode 100644 index 0000000..f80f7c7 --- /dev/null +++ b/migrations/sqlite/20260521000000_backfill_diagnostics.sql @@ -0,0 +1,2 @@ +ALTER TABLE backfill_repos ADD COLUMN records_fetched INTEGER NOT NULL DEFAULT 0; +CREATE INDEX IF NOT EXISTS idx_backfill_repos_job_status ON backfill_repos (job_id, status); diff --git a/src/admin/backfill.rs b/src/admin/backfill.rs index 26b7253..c9c3d8c 100644 --- a/src/admin/backfill.rs +++ b/src/admin/backfill.rs @@ -12,6 +12,8 @@ use serde_json::Value; use tokio::sync::mpsc; use uuid::Uuid; +use rand::Rng; + use crate::AppState; use crate::db::{adapt_sql, now_rfc3339}; use crate::error::AppError; @@ -66,6 +68,13 @@ async fn set_stage(state: &AppState, job_id: &str, stage: &str) { .bind(job_id) .execute(&state.db) .await; + publish_event( + state, + super::types::BackfillEvent::JobStageChanged { + job_id: job_id.to_string(), + stage: stage.to_string(), + }, + ); } async fn update_job_counter(state: &AppState, job_id: &str, column: &str, value: i32) { @@ -103,12 +112,13 @@ async fn count_repos(state: &AppState, job_id: &str) -> i32 { .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; +fn publish_event(state: &AppState, event: super::types::BackfillEvent) { + let _ = state.backfill_events_tx.send(event); +} + +fn random_batch_threshold(base: i32) -> i32 { + let low = base - base / 10; + rand::rng().random_range(low..=base) } async fn fail_job(state: &AppState, job_id: &str, error: &str) { @@ -123,7 +133,14 @@ async fn fail_job(state: &AppState, job_id: &str, error: &str) { .bind(job_id) .execute(&state.db) .await; - cleanup_repos(state, job_id).await; + publish_event( + state, + super::types::BackfillEvent::JobCompleted { + job_id: job_id.to_string(), + status: "failed".to_string(), + error: Some(error.to_string()), + }, + ); } async fn is_cancelled(state: &AppState, job_id: &str) -> bool { @@ -159,7 +176,14 @@ async fn finalise_cancel(state: &AppState, job_id: &str) { .bind(job_id) .execute(&state.db) .await; - cleanup_repos(state, job_id).await; + publish_event( + state, + super::types::BackfillEvent::JobCompleted { + job_id: job_id.to_string(), + status: "cancelled".to_string(), + error: None, + }, + ); } async fn complete_job( @@ -182,7 +206,14 @@ async fn complete_job( .bind(job_id) .execute(&state.db) .await; - cleanup_repos(state, job_id).await; + publish_event( + state, + super::types::BackfillEvent::JobCompleted { + job_id: job_id.to_string(), + status: "completed".to_string(), + error: error.map(|e| e.to_string()), + }, + ); } // --------------------------------------------------------------------------- @@ -207,6 +238,13 @@ async fn run_discovery_phase( .bind(did) .execute(&state.db) .await; + publish_event( + state, + super::types::BackfillEvent::RepoDiscovered { + job_id: job_id.to_string(), + did: did.to_string(), + }, + ); } else { for collection in collections { if is_cancelled(state, job_id).await { @@ -301,6 +339,15 @@ async fn discover_repos_from_relay( if let Ok(result) = query.execute(&state.db).await { running_total += result.rows_affected() as i32; } + for repo in chunk { + publish_event( + state, + super::types::BackfillEvent::RepoDiscovered { + job_id: job_id.to_string(), + did: repo.did.clone(), + }, + ); + } } } @@ -401,6 +448,8 @@ async fn run_pipelined_resolve_and_fetch( .unwrap_or_default(); let mut attempted: i32 = 0; + let mut next_flush = random_batch_threshold(100); + let mut next_cancel_check = random_batch_threshold(100); for (did,) in &unresolved { if resolver_cancelled.load(Ordering::Relaxed) { break; @@ -425,8 +474,17 @@ async fn run_pipelined_resolve_and_fetch( .execute(&resolver_state.db) .await; + publish_event( + &resolver_state, + super::types::BackfillEvent::RepoResolved { + job_id: resolver_job_id.clone(), + did: did.clone(), + pds_endpoint: pds.clone(), + }, + ); + let count = resolver_resolved.fetch_add(1, Ordering::Relaxed) + 1; - if count % 100 == 0 { + if count >= next_flush { update_job_counter( &resolver_state, &resolver_job_id, @@ -434,7 +492,18 @@ async fn run_pipelined_resolve_and_fetch( count, ) .await; + next_flush = count + random_batch_threshold(100); } + publish_event( + &resolver_state, + super::types::BackfillEvent::JobCounters { + job_id: resolver_job_id.clone(), + total_repos: None, + resolved_repos: Some(count), + processed_repos: None, + total_records: None, + }, + ); if tx_resolver.send((did.clone(), pds)).await.is_err() { break; @@ -446,9 +515,12 @@ async fn run_pipelined_resolve_and_fetch( } attempted += 1; - if attempted % 100 == 0 && is_cancelled(&resolver_state, &resolver_job_id).await { - resolver_cancelled.store(true, Ordering::Relaxed); - break; + if attempted >= next_cancel_check { + if is_cancelled(&resolver_state, &resolver_job_id).await { + resolver_cancelled.store(true, Ordering::Relaxed); + break; + } + next_cancel_check = attempted + random_batch_threshold(100); } } @@ -657,6 +729,7 @@ async fn run_pds_worker(ctx: FetchContext, pds_endpoint: String, mut rx: mpsc::R } = ctx; let mut fetches = FuturesUnordered::new(); let mut rx_open = true; + let mut next_flush = random_batch_threshold(10); loop { tokio::select! { @@ -668,18 +741,26 @@ async fn run_pds_worker(ctx: FetchContext, pds_endpoint: String, mut rx: mpsc::R // Mark DID as completed let sql = adapt_sql( - "UPDATE backfill_repos SET status = 'completed' WHERE job_id = ? AND did = ?", + "UPDATE backfill_repos SET status = 'completed', records_fetched = ? WHERE job_id = ? AND did = ?", state.db_backend, ); let _ = sqlx::query(&sql) + .bind(records) .bind(job_id.as_str()) .bind(&did) .execute(&state.db) .await; + publish_event(&state, super::types::BackfillEvent::RepoFetched { + job_id: job_id.to_string(), + did: did.clone(), + pds_endpoint: pds_endpoint.clone(), + records_fetched: records, + }); + let repos = processed_repos.fetch_add(1, Ordering::Relaxed) + 1; - if repos % 10 == 0 { - let records = total_records.load(Ordering::Relaxed); + let records = total_records.load(Ordering::Relaxed); + if repos >= next_flush { let sql = adapt_sql( "UPDATE backfill_jobs SET processed_repos = ?, total_records = ? WHERE id = ?", state.db_backend, @@ -695,7 +776,15 @@ async fn run_pds_worker(ctx: FetchContext, pds_endpoint: String, mut rx: mpsc::R cancelled.store(true, Ordering::Relaxed); break; } + next_flush = repos + random_batch_threshold(10); } + publish_event(&state, super::types::BackfillEvent::JobCounters { + job_id: job_id.to_string(), + total_repos: None, + resolved_repos: None, + processed_repos: Some(repos), + total_records: Some(records), + }); } did = rx.recv(), if rx_open && fetches.len() < 3 => { @@ -747,15 +836,26 @@ async fn run_pds_worker(ctx: FetchContext, pds_endpoint: String, mut rx: mpsc::R total_records.fetch_add(records, Ordering::Relaxed); let sql = adapt_sql( - "UPDATE backfill_repos SET status = 'completed' WHERE job_id = ? AND did = ?", + "UPDATE backfill_repos SET status = 'completed', records_fetched = ? WHERE job_id = ? AND did = ?", state.db_backend, ); let _ = sqlx::query(&sql) + .bind(records) .bind(job_id.as_str()) .bind(&did) .execute(&state.db) .await; + publish_event( + &state, + super::types::BackfillEvent::RepoFetched { + job_id: job_id.to_string(), + did: did.clone(), + pds_endpoint: pds_endpoint.clone(), + records_fetched: records, + }, + ); + processed_repos.fetch_add(1, Ordering::Relaxed); } } @@ -815,6 +915,9 @@ 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(existing_records)); let cancelled = Arc::new(AtomicBool::new(false)); + let next_flush = Arc::new(AtomicI32::new( + already_completed + random_batch_threshold(10), + )); let state = Arc::new(state.clone()); let collections = Arc::new(collections.to_vec()); let job_id_arc = Arc::new(job_id.to_string()); @@ -828,6 +931,7 @@ async fn run_fetching_phase(state: &AppState, job_id: &str, collections: &[Strin let processed_repos = Arc::clone(&processed_repos); let total_records = Arc::clone(&total_records); let cancelled = Arc::clone(&cancelled); + let next_flush = Arc::clone(&next_flush); let job_id = Arc::clone(&job_id_arc); async move { @@ -838,6 +942,7 @@ async fn run_fetching_phase(state: &AppState, job_id: &str, collections: &[Strin let processed_repos = Arc::clone(&processed_repos); let total_records = Arc::clone(&total_records); let cancelled = Arc::clone(&cancelled); + let next_flush = Arc::clone(&next_flush); let pds_endpoint = pds_endpoint.clone(); let job_id = Arc::clone(&job_id); @@ -846,6 +951,7 @@ async fn run_fetching_phase(state: &AppState, job_id: &str, collections: &[Strin return; } + let mut did_records: i32 = 0; for collection in collections.iter() { match fetch_records_from_pds( &state, @@ -856,6 +962,7 @@ async fn run_fetching_phase(state: &AppState, job_id: &str, collections: &[Strin .await { Ok(count) => { + did_records += count as i32; total_records .fetch_add(count as i32, Ordering::Relaxed); } @@ -873,19 +980,23 @@ async fn run_fetching_phase(state: &AppState, job_id: &str, collections: &[Strin // Mark DID as completed let sql = adapt_sql( - "UPDATE backfill_repos SET status = 'completed' WHERE job_id = ? AND did = ?", + "UPDATE backfill_repos SET status = 'completed', records_fetched = ? WHERE job_id = ? AND did = ?", state.db_backend, ); let _ = sqlx::query(&sql) + .bind(did_records) .bind(job_id.as_str()) .bind(&did) .execute(&state.db) .await; let repos = processed_repos.fetch_add(1, Ordering::Relaxed) + 1; + let records = total_records.load(Ordering::Relaxed); - if repos % 10 == 0 { - let records = total_records.load(Ordering::Relaxed); + let threshold = next_flush.load(Ordering::Relaxed); + if repos >= threshold + && next_flush.compare_exchange(threshold, repos + random_batch_threshold(10), Ordering::Relaxed, Ordering::Relaxed).is_ok() + { let backend = state.db_backend; let sql = adapt_sql( "UPDATE backfill_jobs SET processed_repos = ?, total_records = ? WHERE id = ?", @@ -902,6 +1013,14 @@ async fn run_fetching_phase(state: &AppState, job_id: &str, collections: &[Strin cancelled.store(true, Ordering::Relaxed); } } + + publish_event(&state, super::types::BackfillEvent::JobCounters { + job_id: job_id.to_string(), + total_repos: None, + resolved_repos: None, + processed_repos: Some(repos), + total_records: Some(records), + }); } }) .await; @@ -1311,6 +1430,243 @@ pub(super) async fn backfill_status( Ok(Json(jobs)) } +// --------------------------------------------------------------------------- +// SSE events endpoint +// --------------------------------------------------------------------------- + +pub(super) async fn backfill_events( + State(state): State, + Path(job_id): Path, + auth: UserAuth, +) -> Result< + axum::response::sse::Sse< + impl futures_util::Stream>, + >, + AppError, +> { + auth.require(Permission::BackfillRead).await?; + + let mut rx = state.backfill_events_tx.subscribe(); + + let stream = async_stream::stream! { + loop { + match rx.recv().await { + Ok(event) => { + let event_job_id = match &event { + super::types::BackfillEvent::RepoDiscovered { job_id, .. } + | super::types::BackfillEvent::RepoResolved { job_id, .. } + | super::types::BackfillEvent::RepoFetched { job_id, .. } + | super::types::BackfillEvent::JobCounters { job_id, .. } + | super::types::BackfillEvent::JobStageChanged { job_id, .. } + | super::types::BackfillEvent::JobCompleted { job_id, .. } => job_id, + }; + if *event_job_id != job_id { + continue; + } + if let Ok(json) = serde_json::to_string(&event) { + yield Ok(axum::response::sse::Event::default().event("event").data(json)); + } + } + Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { + tracing::warn!(job_id, skipped = n, "SSE client lagged behind"); + continue; + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + } + } + }; + + Ok(axum::response::sse::Sse::new(stream).keep_alive(axum::response::sse::KeepAlive::default())) +} + +// --------------------------------------------------------------------------- +// REST detail endpoints +// --------------------------------------------------------------------------- + +#[derive(Deserialize)] +pub(super) struct ReposQuery { + phase: Option, + cursor: Option, + limit: Option, +} + +pub(super) async fn backfill_repos( + State(state): State, + Path(job_id): Path, + auth: UserAuth, + axum::extract::Query(query): axum::extract::Query, +) -> Result, AppError> { + auth.require(Permission::BackfillRead).await?; + + let limit = query.limit.unwrap_or(50).min(100); + let phase_filter = match query.phase.as_deref() { + Some("resolved") => " AND pds_endpoint IS NOT NULL", + Some("fetched") => " AND status = 'completed'", + _ => "", + }; + let cursor_filter = if query.cursor.is_some() { + " AND did > ?" + } else { + "" + }; + + let sql_str = format!( + "SELECT did, pds_endpoint, status, records_fetched FROM backfill_repos WHERE job_id = ?{phase_filter}{cursor_filter} ORDER BY did ASC LIMIT ?", + ); + let sql = adapt_sql(&sql_str, state.db_backend); + + let mut q = sqlx::query_as::<_, (String, Option, String, i32)>(&sql).bind(&job_id); + if let Some(ref cursor) = query.cursor { + q = q.bind(cursor); + } + q = q.bind(limit + 1); + + let rows: Vec<(String, Option, String, i32)> = q + .fetch_all(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to query backfill repos: {e}")))?; + + let has_more = rows.len() > limit as usize; + let repos: Vec = rows + .into_iter() + .take(limit as usize) + .map( + |(did, pds_endpoint, status, records_fetched)| super::types::BackfillRepoEntry { + did, + pds_endpoint, + status, + records_fetched, + }, + ) + .collect(); + + let cursor = if has_more { + repos.last().map(|r| r.did.clone()) + } else { + None + }; + + Ok(Json(super::types::BackfillReposResponse { repos, cursor })) +} + +pub(super) async fn backfill_pds_summary( + State(state): State, + Path(job_id): Path, + auth: UserAuth, +) -> Result, AppError> { + auth.require(Permission::BackfillRead).await?; + + let sql = adapt_sql( + "SELECT pds_endpoint, COUNT(*) as total_repos, SUM(CASE WHEN status = 'completed' THEN 1 ELSE 0 END) as completed_repos, SUM(records_fetched) as total_records FROM backfill_repos WHERE job_id = ? AND pds_endpoint IS NOT NULL GROUP BY pds_endpoint ORDER BY COUNT(*) DESC", + state.db_backend, + ); + + let rows: Vec<(String, i32, i32, i64)> = sqlx::query_as(&sql) + .bind(&job_id) + .fetch_all(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to query PDS summary: {e}")))?; + + let pds_endpoints: Vec = rows + .into_iter() + .map( + |(pds_endpoint, total_repos, completed_repos, total_records)| { + super::types::PdsSummaryEntry { + pds_endpoint, + total_repos, + completed_repos, + total_records: total_records as i32, + } + }, + ) + .collect(); + + Ok(Json(super::types::PdsSummaryResponse { pds_endpoints })) +} + +// --------------------------------------------------------------------------- +// Flush endpoints +// --------------------------------------------------------------------------- + +pub(super) async fn flush_backfill_details( + State(state): State, + Path(job_id): Path, + auth: UserAuth, +) -> Result { + auth.require(Permission::BackfillCreate).await?; + + 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; + + Ok(StatusCode::NO_CONTENT) +} + +pub(super) async fn flush_all_backfill_details( + State(state): State, + auth: UserAuth, +) -> Result { + auth.require(Permission::BackfillCreate).await?; + + let sql = adapt_sql( + "DELETE FROM backfill_repos WHERE job_id IN (SELECT id FROM backfill_jobs WHERE status IN ('completed', 'cancelled', 'failed'))", + state.db_backend, + ); + let _ = sqlx::query(&sql).execute(&state.db).await; + + Ok(StatusCode::NO_CONTENT) +} + +// --------------------------------------------------------------------------- +// Retention cleanup +// --------------------------------------------------------------------------- + +pub async fn run_backfill_retention_cleanup(state: &AppState) { + use super::settings::get_setting; + + let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(86400)); + interval.tick().await; // first tick is immediate — run once on startup + + loop { + interval.tick().await; + + let retention_days: i64 = + get_setting(&state.db, "backfill_retention_days", state.db_backend) + .await + .and_then(|v| v.parse().ok()) + .unwrap_or(28); + + if retention_days == 0 { + continue; + } + + let cutoff = chrono::Utc::now() - chrono::Duration::days(retention_days); + let cutoff_str = cutoff.to_rfc3339(); + + let sql = adapt_sql( + "DELETE FROM backfill_repos WHERE job_id IN (SELECT id FROM backfill_jobs WHERE completed_at IS NOT NULL AND completed_at < ?)", + state.db_backend, + ); + match sqlx::query(&sql).bind(&cutoff_str).execute(&state.db).await { + Ok(result) => { + let deleted = result.rows_affected(); + if deleted > 0 { + tracing::info!( + deleted, + retention_days, + "cleaned up old backfill detail rows" + ); + } + } + Err(e) => { + tracing::warn!(error = %e, "backfill retention cleanup failed"); + } + } + } +} + // --------------------------------------------------------------------------- // Startup resumption // --------------------------------------------------------------------------- diff --git a/src/admin/mod.rs b/src/admin/mod.rs index fa7722a..5cba8b4 100644 --- a/src/admin/mod.rs +++ b/src/admin/mod.rs @@ -37,7 +37,21 @@ 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/details", + delete(backfill::flush_all_backfill_details), + ) .route("/backfill/{id}/cancel", post(backfill::cancel_backfill)) + .route("/backfill/{id}/events", get(backfill::backfill_events)) + .route("/backfill/{id}/repos", get(backfill::backfill_repos)) + .route( + "/backfill/{id}/pds-summary", + get(backfill::backfill_pds_summary), + ) + .route( + "/backfill/{id}/details", + delete(backfill::flush_backfill_details), + ) .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/src/admin/settings.rs b/src/admin/settings.rs index f86c065..8c6a138 100644 --- a/src/admin/settings.rs +++ b/src/admin/settings.rs @@ -16,6 +16,7 @@ use super::types::{SettingEntry, UpsertSettingBody}; const ENV_FALLBACKS: &[(&str, &str)] = &[ ("app_name", "APP_NAME"), + ("backfill_retention_days", "BACKFILL_RETENTION_DAYS"), ("client_uri", "CLIENT_URI"), ("feature.spaces_enabled", "FEATURE_SPACES_ENABLED"), ("logo_uri", "LOGO_URI"), diff --git a/src/admin/types.rs b/src/admin/types.rs index 3a5c0d1..2e903a8 100644 --- a/src/admin/types.rs +++ b/src/admin/types.rs @@ -85,6 +85,77 @@ pub(crate) struct BackfillJob { pub(crate) created_at: String, } +// --------------------------------------------------------------------------- +// Backfill event types +// --------------------------------------------------------------------------- + +#[derive(Clone, Debug, Serialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum BackfillEvent { + RepoDiscovered { + job_id: String, + did: String, + }, + RepoResolved { + job_id: String, + did: String, + pds_endpoint: String, + }, + RepoFetched { + job_id: String, + did: String, + pds_endpoint: String, + records_fetched: i32, + }, + JobCounters { + job_id: String, + total_repos: Option, + resolved_repos: Option, + processed_repos: Option, + total_records: Option, + }, + JobStageChanged { + job_id: String, + stage: String, + }, + JobCompleted { + job_id: String, + status: String, + error: Option, + }, +} + +// --------------------------------------------------------------------------- +// Backfill detail response types +// --------------------------------------------------------------------------- + +#[derive(Serialize)] +pub(crate) struct BackfillRepoEntry { + pub(crate) did: String, + pub(crate) pds_endpoint: Option, + pub(crate) status: String, + pub(crate) records_fetched: i32, +} + +#[derive(Serialize)] +pub(crate) struct BackfillReposResponse { + pub(crate) repos: Vec, + pub(crate) cursor: Option, +} + +#[derive(Serialize)] +pub(crate) struct PdsSummaryEntry { + pub(crate) pds_endpoint: String, + pub(crate) total_repos: i32, + pub(crate) completed_repos: i32, + pub(crate) total_records: i32, +} + +#[derive(Serialize)] +pub(crate) struct PdsSummaryResponse { + pub(crate) pds_endpoints: Vec, +} + // --------------------------------------------------------------------------- // Network lexicon types // --------------------------------------------------------------------------- diff --git a/src/lib.rs b/src/lib.rs index fe19b38..9ea8e48 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -78,6 +78,7 @@ pub struct AppState { pub official_registry: SharedRegistry, pub official_registry_config: RegistryConfig, pub proxy_config: Arc>, + pub backfill_events_tx: tokio::sync::broadcast::Sender, } impl axum::extract::FromRef for axum_extra::extract::cookie::Key { diff --git a/src/lua/atproto_api.rs b/src/lua/atproto_api.rs index e70e488..560e638 100644 --- a/src/lua/atproto_api.rs +++ b/src/lua/atproto_api.rs @@ -582,6 +582,7 @@ mod tests { proxy_config: std::sync::Arc::new(arc_swap::ArcSwap::new(std::sync::Arc::new( crate::proxy_config::ProxyConfig::default(), ))), + backfill_events_tx: tokio::sync::broadcast::channel(16).0, } } diff --git a/src/lua/db_api.rs b/src/lua/db_api.rs index 5899bbd..f22725b 100644 --- a/src/lua/db_api.rs +++ b/src/lua/db_api.rs @@ -790,6 +790,7 @@ mod tests { proxy_config: std::sync::Arc::new(arc_swap::ArcSwap::new(std::sync::Arc::new( crate::proxy_config::ProxyConfig::default(), ))), + backfill_events_tx: tokio::sync::broadcast::channel(16).0, } } diff --git a/src/lua/execute.rs b/src/lua/execute.rs index 2e32cb9..97f5e6b 100644 --- a/src/lua/execute.rs +++ b/src/lua/execute.rs @@ -1180,6 +1180,7 @@ mod tests { proxy_config: std::sync::Arc::new(arc_swap::ArcSwap::new(std::sync::Arc::new( crate::proxy_config::ProxyConfig::default(), ))), + backfill_events_tx: tokio::sync::broadcast::channel(16).0, } } diff --git a/src/lua/http_api.rs b/src/lua/http_api.rs index 39b7a22..2c20331 100644 --- a/src/lua/http_api.rs +++ b/src/lua/http_api.rs @@ -188,6 +188,7 @@ mod tests { proxy_config: std::sync::Arc::new(arc_swap::ArcSwap::new(std::sync::Arc::new( crate::proxy_config::ProxyConfig::default(), ))), + backfill_events_tx: tokio::sync::broadcast::channel(16).0, } } diff --git a/src/lua/xrpc_api.rs b/src/lua/xrpc_api.rs index 308be59..6cb7860 100644 --- a/src/lua/xrpc_api.rs +++ b/src/lua/xrpc_api.rs @@ -292,6 +292,7 @@ mod tests { proxy_config: std::sync::Arc::new(arc_swap::ArcSwap::new(std::sync::Arc::new( crate::proxy_config::ProxyConfig::default(), ))), + backfill_events_tx: tokio::sync::broadcast::channel(16).0, } } diff --git a/src/main.rs b/src/main.rs index 16f92fe..595bb0a 100644 --- a/src/main.rs +++ b/src/main.rs @@ -612,6 +612,8 @@ async fn main() { std::sync::Arc::new(arc_swap::ArcSwap::new(std::sync::Arc::new(config))) }; + let (backfill_events_tx, _) = tokio::sync::broadcast::channel(1024); + let state = AppState { config: config.clone(), http, @@ -631,6 +633,7 @@ async fn main() { official_registry, official_registry_config, proxy_config, + backfill_events_tx, }; jetstream::spawn(state.clone(), collections_rx); @@ -646,6 +649,13 @@ async fn main() { happyview::admin::backfill::resume_backfill_jobs(&state).await; + { + let state = state.clone(); + tokio::spawn(async move { + happyview::admin::backfill::run_backfill_retention_cleanup(&state).await; + }); + } + let app = server::router(state); let addr = config.listen_addr(); diff --git a/tests/common/app.rs b/tests/common/app.rs index f39914b..2331947 100644 --- a/tests/common/app.rs +++ b/tests/common/app.rs @@ -167,6 +167,7 @@ impl TestApp { proxy_config: std::sync::Arc::new(arc_swap::ArcSwap::new(std::sync::Arc::new( happyview::proxy_config::ProxyConfig::default(), ))), + backfill_events_tx: tokio::sync::broadcast::channel(16).0, }; let router = server::router(state.clone()).layer(axum::middleware::from_fn( diff --git a/tests/lua_atproto_api.rs b/tests/lua_atproto_api.rs index 4946ced..459a9b6 100644 --- a/tests/lua_atproto_api.rs +++ b/tests/lua_atproto_api.rs @@ -101,6 +101,7 @@ async fn test_state_with_pool(pool: sqlx::AnyPool, backend: DatabaseBackend) -> proxy_config: std::sync::Arc::new(arc_swap::ArcSwap::new(std::sync::Arc::new( happyview::proxy_config::ProxyConfig::default(), ))), + backfill_events_tx: tokio::sync::broadcast::channel(16).0, } } diff --git a/tests/lua_db_api.rs b/tests/lua_db_api.rs index 1eb8de3..cd6a621 100644 --- a/tests/lua_db_api.rs +++ b/tests/lua_db_api.rs @@ -104,6 +104,7 @@ async fn test_state_with_pool(pool: sqlx::AnyPool, backend: DatabaseBackend) -> proxy_config: std::sync::Arc::new(arc_swap::ArcSwap::new(std::sync::Arc::new( happyview::proxy_config::ProxyConfig::default(), ))), + backfill_events_tx: tokio::sync::broadcast::channel(16).0, } } diff --git a/web/package-lock.json b/web/package-lock.json index ed30f1a..df10472 100644 --- a/web/package-lock.json +++ b/web/package-lock.json @@ -13,6 +13,7 @@ "@tabler/icons-react": "^3.36.1", "@tailwindcss/typography": "^0.5.19", "@tanstack/react-table": "^8.21.3", + "@tanstack/react-virtual": "^3.13.25", "class-variance-authority": "^0.7.1", "clsx": "^2.1.1", "cmdk": "^1.1.1", @@ -4143,6 +4144,23 @@ "react-dom": ">=16.8" } }, + "node_modules/@tanstack/react-virtual": { + "version": "3.13.25", + "resolved": "https://registry.npmjs.org/@tanstack/react-virtual/-/react-virtual-3.13.25.tgz", + "integrity": "sha512-bmNoqMu6gcAW9JGrKVB0Q1tN1i5RONZF8r1fW0bbE4Oyf3DwEGnzzQJ2OW+Ozg1P4s8PyugkHg2ULZoFQN+cqw==", + "license": "MIT", + "dependencies": { + "@tanstack/virtual-core": "3.15.0" + }, + "funding": { + "type": "github", + "url": "https://github.com/sponsors/tannerlinsley" + }, + "peerDependencies": { + "react": "^16.8.0 || ^17.0.0 || ^18.0.0 || ^19.0.0", + "react-dom": "^16.8.0 || ^17.0.0 || ^18.0.0 || ^19.0.0" + } + }, "node_modules/@tanstack/table-core": { "version": "8.21.3", "resolved": "https://registry.npmjs.org/@tanstack/table-core/-/table-core-8.21.3.tgz", @@ -4156,6 +4174,16 @@ "url": "https://github.com/sponsors/tannerlinsley" } }, + "node_modules/@tanstack/virtual-core": { + "version": "3.15.0", + "resolved": "https://registry.npmjs.org/@tanstack/virtual-core/-/virtual-core-3.15.0.tgz", + "integrity": "sha512-0AwPGx0I8QxPYjAxShT/+z+ZOe9u8mW5rsXvivCTjRfRmz9a43+3mRyi4wwlyoUqOC56q/jatKa0Bh9M99BEHQ==", + "license": "MIT", + "funding": { + "type": "github", + "url": "https://github.com/sponsors/tannerlinsley" + } + }, "node_modules/@ts-morph/common": { "version": "0.27.0", "resolved": "https://registry.npmjs.org/@ts-morph/common/-/common-0.27.0.tgz", diff --git a/web/package.json b/web/package.json index 6286c50..517f9d7 100644 --- a/web/package.json +++ b/web/package.json @@ -14,6 +14,7 @@ "@tabler/icons-react": "^3.36.1", "@tailwindcss/typography": "^0.5.19", "@tanstack/react-table": "^8.21.3", + "@tanstack/react-virtual": "^3.13.25", "class-variance-authority": "^0.7.1", "clsx": "^2.1.1", "cmdk": "^1.1.1", diff --git a/web/src/app/dashboard/backfill/page.tsx b/web/src/app/dashboard/backfill/page.tsx index e3f1060..3331075 100644 --- a/web/src/app/dashboard/backfill/page.tsx +++ b/web/src/app/dashboard/backfill/page.tsx @@ -1,18 +1,44 @@ "use client"; -import { useCallback, useEffect, useState } from "react"; +import { useCallback, useEffect, useMemo, useRef, useState } from "react"; import { useCurrentUser } from "@/hooks/use-current-user"; import { cancelBackfillJob, createBackfillJob, getBackfillJobs, + getBackfillRepos, + getBackfillPdsSummary, + flushBackfillDetails, + flushAllBackfillDetails, getLexicons, } from "@/lib/api"; -import type { BackfillJob } from "@/types/backfill"; -import { CheckCircle2, Circle, Loader2 } from "lucide-react"; +import type { + BackfillJob, + BackfillRepoEntry, + PdsSummaryEntry, + BackfillEvent, + BlueskyProfile, +} from "@/types/backfill"; +import { CheckCircle2, ChevronRight, Circle, Loader2 } from "lucide-react"; import { SiteHeader } from "@/components/site-header"; +import { + AlertDialog, + AlertDialogAction, + AlertDialogCancel, + AlertDialogContent, + AlertDialogDescription, + AlertDialogFooter, + AlertDialogHeader, + AlertDialogTitle, + AlertDialogTrigger, +} from "@/components/ui/alert-dialog"; import { Badge } from "@/components/ui/badge"; +import { + Collapsible, + CollapsibleContent, + CollapsibleTrigger, +} from "@/components/ui/collapsible"; import { Button } from "@/components/ui/button"; import { Combobox, @@ -97,6 +123,257 @@ function phaseIndex(stage: string): number { return idx; } +// SSE hook for backfill events +function useBackfillSSE(jobId: string | null, active: boolean): BackfillEvent[] { + const [events, setEvents] = useState([]); + + useEffect(() => { + if (!jobId || !active) { + setEvents([]); + return; + } + + const basePath = process.env.NEXT_PUBLIC_BASE_PATH || ""; + const es = new EventSource(`${basePath}/admin/backfill/${jobId}/events`, { + withCredentials: true, + }); + + es.addEventListener("event", (e) => { + try { + const event: BackfillEvent = JSON.parse((e as MessageEvent).data); + setEvents((prev) => [...prev, event]); + } catch { /* ignore parse errors */ } + }); + + return () => es.close(); + }, [jobId, active]); + + return events; +} + +// Batch Bluesky profile resolution hook +function useBlueskyProfiles(dids: string[]): Map { + const [profiles, setProfiles] = useState>(new Map()); + const resolvedRef = useRef>(new Set()); + const pendingRef = useRef(false); + + useEffect(() => { + const unresolved = dids.filter((d) => !resolvedRef.current.has(d)); + if (unresolved.length === 0 || pendingRef.current) return; + + pendingRef.current = true; + + const batches: string[][] = []; + for (let i = 0; i < unresolved.length; i += 25) { + batches.push(unresolved.slice(i, i + 25)); + } + + // Mark all as resolved immediately to prevent re-fetching + for (const did of unresolved) { + resolvedRef.current.add(did); + } + + Promise.all( + batches.map(async (batch) => { + const params = batch.map((d) => `actors=${encodeURIComponent(d)}`).join("&"); + try { + const resp = await fetch( + `https://public.api.bsky.app/xrpc/app.bsky.actor.getProfiles?${params}` + ); + if (!resp.ok) return []; + const data = await resp.json(); + return (data.profiles || []) as BlueskyProfile[]; + } catch { + return []; + } + }) + ).then((results) => { + setProfiles((prev) => { + const newProfiles = new Map(prev); + for (const batch of results) { + for (const p of batch) { + newProfiles.set(p.did, p); + } + } + return newProfiles; + }); + pendingRef.current = false; + }); + }, [dids]); + + return profiles; +} + +const BSKY_PDS_SUFFIX = ".bsky.network"; +const BSKY_PDS_HOSTNAMES = ["bsky.social", "staging.bsky.dev"]; +const failedFaviconUrls = new Set(); + +function isBskyPds(pdsEndpoint: string): boolean { + try { + const hostname = new URL(pdsEndpoint).hostname; + return BSKY_PDS_HOSTNAMES.includes(hostname) || hostname.endsWith(BSKY_PDS_SUFFIX); + } catch { + return false; + } +} + +function PdsFavicon({ pdsEndpoint }: { pdsEndpoint: string }) { + let hostname: string; + try { + hostname = new URL(pdsEndpoint).hostname; + } catch { + return ; + } + + if (isBskyPds(pdsEndpoint)) { + return ( + + + + ); + } + + const faviconUrl = `https://twenty-icons.com/${hostname}`; + + if (failedFaviconUrls.has(faviconUrl)) { + return ; + } + + return ; +} + +function PdsFaviconImg({ url }: { url: string }) { + const [failed, setFailed] = useState(false); + + if (failed) return ; + + return ( + { failedFaviconUrls.add(url); setFailed(true); }} + /> + ); +} + +function PdsPlaceholderIcon() { + return ( + + + + + + + + ); +} + +function ScrollSentinel({ onVisible }: { onVisible: () => void }) { + const ref = useRef(null); + const onVisibleRef = useRef(onVisible); + onVisibleRef.current = onVisible; + + useEffect(() => { + const el = ref.current; + if (!el) return; + const observer = new IntersectionObserver( + ([entry]) => { if (entry.isIntersecting) onVisibleRef.current(); }, + { rootMargin: "100px" }, + ); + observer.observe(el); + return () => observer.disconnect(); + }, []); + + return
; +} + +function AnimatedNumber({ value }: { value: number }) { + const targetRef = useRef(value); + const displayedRef = useRef(value); + const [displayed, setDisplayed] = useState(value); + const rafRef = useRef(0); + + targetRef.current = value; + + useEffect(() => { + cancelAnimationFrame(rafRef.current); + + function tick() { + const current = displayedRef.current; + const target = targetRef.current; + const diff = target - current; + + if (Math.abs(diff) < 0.5) { + displayedRef.current = target; + setDisplayed(target); + return; + } + + const next = current + diff * 0.06; + displayedRef.current = next; + setDisplayed(next); + rafRef.current = requestAnimationFrame(tick); + } + + tick(); + return () => cancelAnimationFrame(rafRef.current); + }, [value]); + + return <>{Math.round(displayed).toLocaleString()}; +} + +function CompactRepoRow({ did, profile }: { + did: string; + profile?: BlueskyProfile; +}) { + return ( +
+
+ {profile?.avatar && ( + + )} +
+ + {profile?.handle ? `@${profile.handle}` : {did}} + +
+ ); +} + +function ProfileRow({ did, profile, suffix }: { + did: string; + profile?: BlueskyProfile; + suffix?: React.ReactNode; +}) { + return ( +
+
+ {profile?.avatar && ( + + )} +
+
+

+ {profile?.displayName || profile?.handle || did} +

+ {profile?.handle && ( +

@{profile.handle}

+ )} + {!profile?.handle && ( +

{did}

+ )} +
+ {suffix && ( + {suffix} + )} +
+ ); +} + export default function BackfillPage() { const { hasPermission } = useCurrentUser(); const [jobs, setJobs] = useState([]); @@ -119,6 +396,7 @@ export default function BackfillPage() { }, [load]); const selectedJob = jobs.find((j) => j.id === selectedJobId) ?? null; + const canFlush = hasPermission("backfill:create"); return ( <> @@ -128,9 +406,32 @@ export default function BackfillPage() {

Backfill Jobs

- {hasPermission("backfill:create") && ( - - )} +
+ {canFlush && ( + + + + + + + Clear all job details? + + This will permanently delete per-repo detail data for all backfill jobs. + + + + Cancel + { + await flushAllBackfillDetails(); + }}>Clear + + + + )} + {hasPermission("backfill:create") && ( + + )} +
@@ -201,6 +502,7 @@ export default function BackfillPage() { { await cancelBackfillJob(selectedJob.id); load(); @@ -217,10 +519,12 @@ export default function BackfillPage() { function JobDetail({ job, canCancel, + canFlush, onCancel, }: { job: BackfillJob; canCancel: boolean; + canFlush: boolean; onCancel: () => Promise; }) { const [cancelling, setCancelling] = useState(false); @@ -228,6 +532,21 @@ function JobDetail({ const allDone = job.status === "completed"; const isActive = job.status === "running" || job.status === "cancelling"; + // Detail data state + const [discoveredRepos, setDiscoveredRepos] = useState([]); + const [discoveredCursor, setDiscoveredCursor] = useState(null); + const [discoveredLoaded, setDiscoveredLoaded] = useState(false); + + const [pdsSummary, setPdsSummary] = useState([]); + const [pdsLoaded, setPdsLoaded] = useState(false); + + const [fetchedRepos, setFetchedRepos] = useState([]); + const [fetchedCursor, setFetchedCursor] = useState(null); + const [fetchedLoaded, setFetchedLoaded] = useState(false); + + // SSE events for active jobs + const sseEvents = useBackfillSSE(job.id, isActive); + function hasReached(phase: (typeof PROGRESS_PHASES)[number]): boolean { if (allDone) return true; if (job.stage === "resolving_and_fetching") { @@ -245,6 +564,102 @@ function JobDetail({ } } + // Auto-load detail data when phases are reached + const discoveredReached = hasReached("discovering_repos"); + const pdsReached = hasReached("resolving_pds"); + const fetchedReached = hasReached("fetching_records") || job.stage === "resolving_and_fetching"; + + useEffect(() => { + if (discoveredReached && !discoveredLoaded) { + getBackfillRepos(job.id, { phase: "discovered", limit: 50 }) + .then((resp) => { setDiscoveredRepos(resp.repos); setDiscoveredCursor(resp.cursor); setDiscoveredLoaded(true); }) + .catch(() => {}); + } + }, [job.id, discoveredReached, discoveredLoaded]); + + useEffect(() => { + if (pdsReached && !pdsLoaded) { + getBackfillPdsSummary(job.id) + .then((resp) => { setPdsSummary(resp.pds_endpoints); setPdsLoaded(true); }) + .catch(() => {}); + } + }, [job.id, pdsReached, pdsLoaded]); + + useEffect(() => { + if (fetchedReached && !fetchedLoaded) { + getBackfillRepos(job.id, { phase: "fetched", limit: 50 }) + .then((resp) => { setFetchedRepos(resp.repos); setFetchedCursor(resp.cursor); setFetchedLoaded(true); }) + .catch(() => {}); + } + }, [job.id, fetchedReached, fetchedLoaded]); + + const loadMoreDiscovered = useCallback(async () => { + if (!discoveredCursor) return; + try { + const resp = await getBackfillRepos(job.id, { phase: "discovered", cursor: discoveredCursor, limit: 50 }); + setDiscoveredRepos((prev) => [...prev, ...resp.repos]); + setDiscoveredCursor(resp.cursor); + } catch { /* ignore */ } + }, [job.id, discoveredCursor]); + + const loadMoreFetched = useCallback(async () => { + if (!fetchedCursor) return; + try { + const resp = await getBackfillRepos(job.id, { phase: "fetched", cursor: fetchedCursor, limit: 50 }); + setFetchedRepos((prev) => [...prev, ...resp.repos]); + setFetchedCursor(resp.cursor); + } catch { /* ignore */ } + }, [job.id, fetchedCursor]); + + // Process SSE events + useEffect(() => { + for (const event of sseEvents) { + if (event.type === "repo_discovered" && event.did) { + setDiscoveredRepos((prev) => { + if (prev.some((r) => r.did === event.did)) return prev; + return [{ did: event.did!, pds_endpoint: null, status: "pending", records_fetched: 0 }, ...prev]; + }); + } + if (event.type === "repo_resolved" && event.did && event.pds_endpoint) { + setDiscoveredRepos((prev) => + prev.map((r) => r.did === event.did ? { ...r, pds_endpoint: event.pds_endpoint! } : r) + ); + setPdsSummary((prev) => { + const idx = prev.findIndex((p) => p.pds_endpoint === event.pds_endpoint); + if (idx >= 0) { + const updated = [...prev]; + updated[idx] = { ...updated[idx], total_repos: updated[idx].total_repos + 1 }; + return updated; + } + return [...prev, { pds_endpoint: event.pds_endpoint!, total_repos: 1, completed_repos: 0, total_records: 0 }]; + }); + } + if (event.type === "repo_fetched" && event.did) { + setFetchedRepos((prev) => { + if (prev.some((r) => r.did === event.did)) return prev; + return [{ did: event.did!, pds_endpoint: event.pds_endpoint ?? null, status: "completed", records_fetched: event.records_fetched ?? 0 }, ...prev]; + }); + setPdsSummary((prev) => { + if (!event.pds_endpoint) return prev; + return prev.map((p) => p.pds_endpoint === event.pds_endpoint + ? { ...p, completed_repos: p.completed_repos + 1, total_records: p.total_records + (event.records_fetched ?? 0) } + : p + ); + }); + } + } + }, [sseEvents]); + + // Collect all visible DIDs for profile resolution + const allVisibleDids = useMemo(() => { + const dids = new Set(); + for (const r of discoveredRepos) dids.add(r.did); + for (const r of fetchedRepos) dids.add(r.did); + return Array.from(dids); + }, [discoveredRepos, fetchedRepos]); + + const profiles = useBlueskyProfiles(allVisibleDids); + return ( <> @@ -306,9 +721,27 @@ function JobDetail({ label="Discovering repos" active={isActive && job.stage === "discovering_repos"} reached={hasReached("discovering_repos")} - value={job.total_repos?.toLocaleString()} + value={job.total_repos != null ? : undefined} suffix="repos found" - /> + loading={discoveredReached && !discoveredLoaded} + > + {discoveredRepos.length > 0 ? ( +
+ {discoveredRepos.map((repo) => ( + + ))} + {discoveredCursor && ( + + )} +
+ ) : discoveredLoaded ? ( +

No repos discovered yet.

+ ) : null} + / : undefined } suffix="resolved" - /> + loading={pdsReached && !pdsLoaded} + > + {pdsSummary.length > 0 ? ( +
+ {pdsSummary + .sort((a, b) => b.total_repos - a.total_repos) + .map((pds) => ( +
+ + {new URL(pds.pds_endpoint).hostname} + + / repos · records + +
+ ))} +
+ ) : pdsLoaded ? ( +

No PDS data yet.

+ ) : null} + / repos : undefined } suffix={ hasReached("fetching_records") || job.stage === "resolving_and_fetching" - ? `${job.total_records?.toLocaleString() ?? "0"} records` + ? <> records : undefined } - /> + loading={fetchedReached && !fetchedLoaded} + > + {fetchedRepos.filter((r) => r.records_fetched > 0).length > 0 ? ( +
+ {fetchedRepos.filter((r) => r.records_fetched > 0).map((repo) => ( + records} + /> + ))} + {fetchedCursor && ( + + )} +
+ ) : fetchedLoaded ? ( +

No repos fetched yet.

+ ) : null} +
- {canCancel && isActive && ( - + + {canFlush && !isActive && ( + + + + + + + Clear job details? + + This will permanently delete per-repo detail data for this backfill job. + + + + Cancel + { + await flushBackfillDetails(job.id); + setDiscoveredRepos([]); + setDiscoveredCursor(null); + setDiscoveredLoaded(false); + setPdsSummary([]); + setPdsLoaded(false); + setFetchedRepos([]); + setFetchedCursor(null); + setFetchedLoaded(false); + }}>Clear + + + + )} + {canCancel && isActive && ( - - )} + )} + ); } @@ -370,42 +870,64 @@ function ProgressRow({ reached, value, suffix, + loading, + children, }: { label: string; active: boolean; reached: boolean; - value?: string; - suffix?: string; + value?: React.ReactNode; + suffix?: React.ReactNode; + loading?: boolean; + children?: React.ReactNode; }) { + const [open, setOpen] = useState(false); const done = reached && !active; + const expandable = reached; return ( -
- - {active ? ( - - ) : done ? ( - - ) : ( - - )} - - {label} - {reached && value && ( - - {value} - {suffix ? ` · ${suffix}` : ""} - + + +
+ + {active ? ( + + ) : done ? ( + + ) : ( + + )} + + {label} + {reached && value && ( + + {value} + {suffix && <> · {suffix}} + + )} + {expandable && ( + + )} +
+ + {expandable && ( + +
+ {loading ? ( +
+ +
+ ) : children} +
+ )} -
+ ); } diff --git a/web/src/app/dashboard/settings/general/page.tsx b/web/src/app/dashboard/settings/general/page.tsx index 731ac34..9d04415 100644 --- a/web/src/app/dashboard/settings/general/page.tsx +++ b/web/src/app/dashboard/settings/general/page.tsx @@ -23,6 +23,7 @@ const SETTING_KEYS = [ "logo_uri", "tos_uri", "policy_uri", + "backfill_retention_days", ] as const type FieldKey = (typeof SETTING_KEYS)[number] @@ -80,6 +81,7 @@ export default function GeneralSettingsPage() { logo_uri: "", tos_uri: "", policy_uri: "", + backfill_retention_days: "28", }) const [sources, setSources] = useState>({ app_name: "unset", @@ -87,6 +89,7 @@ export default function GeneralSettingsPage() { logo_uri: "unset", tos_uri: "unset", policy_uri: "unset", + backfill_retention_days: "unset", }) const [logoUploaded, setLogoUploaded] = useState(false) const [error, setError] = useState(null) @@ -104,6 +107,7 @@ export default function GeneralSettingsPage() { logo_uri: byKey.get("logo_uri")?.value ?? "", tos_uri: byKey.get("tos_uri")?.value ?? "", policy_uri: byKey.get("policy_uri")?.value ?? "", + backfill_retention_days: byKey.get("backfill_retention_days")?.value ?? "28", }) setSources({ app_name: (byKey.get("app_name")?.source as "database" | "env" | undefined) ?? "unset", @@ -111,6 +115,7 @@ export default function GeneralSettingsPage() { logo_uri: (byKey.get("logo_uri")?.source as "database" | "env" | undefined) ?? "unset", tos_uri: (byKey.get("tos_uri")?.source as "database" | "env" | undefined) ?? "unset", policy_uri: (byKey.get("policy_uri")?.source as "database" | "env" | undefined) ?? "unset", + backfill_retention_days: (byKey.get("backfill_retention_days")?.source as "database" | "env" | undefined) ?? "unset", }) setLogoUploaded(byKey.has("logo_data")) } catch (e: unknown) { @@ -137,6 +142,14 @@ export default function GeneralSettingsPage() { await upsertSetting(field.key, value) } } + const retentionValue = values["backfill_retention_days"] + if (retentionValue === "") { + if (sources["backfill_retention_days"] === "database") { + await deleteSetting("backfill_retention_days") + } + } else { + await upsertSetting("backfill_retention_days", retentionValue) + } setNotice("Settings saved.") await load() } catch (e: unknown) { @@ -252,6 +265,37 @@ export default function GeneralSettingsPage() { +
+

Data Retention

+

+ Configure how long HappyView retains detailed data from completed backfill jobs. +

+
+ +
+
+ + {sources["backfill_retention_days"] === "env" && ( + + from env var + + )} +
+ + setValues((v) => ({ ...v, backfill_retention_days: e.target.value })) + } + placeholder="28" + disabled={!canManage} + /> +

+ How long to keep per-repo detail data from completed backfill jobs. Set to 0 to keep indefinitely. +

+
+