+13

migrations/20260216000000_create_backfill_jobs.sql

reviewed

··· 1 1 + CREATE TABLE IF NOT EXISTS backfill_jobs ( 2 2 + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), 3 3 + collection TEXT, 4 4 + did TEXT, 5 5 + status TEXT NOT NULL DEFAULT 'pending', 6 6 + total_repos INT DEFAULT 0, 7 7 + processed_repos INT DEFAULT 0, 8 8 + total_records INT DEFAULT 0, 9 9 + error TEXT, 10 10 + started_at TIMESTAMPTZ, 11 11 + completed_at TIMESTAMPTZ, 12 12 + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() 13 13 + );

+99

src/admin.rs

reviewed

··· 19 19 .route("/lexicons", post(upload_lexicon).get(list_lexicons)) 20 20 .route("/lexicons/{id}", get(get_lexicon).delete(delete_lexicon)) 21 21 .route("/stats", get(stats)) 22 22 + .route("/backfill", post(create_backfill)) 23 23 + .route("/backfill/status", get(backfill_status)) 22 24 } 23 25 24 26 /// Extract and validate the admin Bearer token from request headers. ··· 299 301 .collect(), 300 302 })) 301 303 } 304 304 + 305 305 + // --------------------------------------------------------------------------- 306 306 + // Backfill endpoints 307 307 + // --------------------------------------------------------------------------- 308 308 + 309 309 + #[derive(Deserialize)] 310 310 + struct CreateBackfillBody { 311 311 + collection: Option<String>, 312 312 + did: Option<String>, 313 313 + } 314 314 + 315 315 + #[derive(Serialize)] 316 316 + struct BackfillJob { 317 317 + id: String, 318 318 + collection: Option<String>, 319 319 + did: Option<String>, 320 320 + status: String, 321 321 + total_repos: Option<i32>, 322 322 + processed_repos: Option<i32>, 323 323 + total_records: Option<i32>, 324 324 + error: Option<String>, 325 325 + started_at: Option<chrono::DateTime<chrono::Utc>>, 326 326 + completed_at: Option<chrono::DateTime<chrono::Utc>>, 327 327 + created_at: chrono::DateTime<chrono::Utc>, 328 328 + } 329 329 + 330 330 + /// POST /admin/backfill — create a new backfill job. 331 331 + async fn create_backfill( 332 332 + State(state): State<AppState>, 333 333 + _admin: AdminAuth, 334 334 + Json(body): Json<CreateBackfillBody>, 335 335 + ) -> Result<(StatusCode, Json<Value>), AppError> { 336 336 + let row: (String,) = sqlx::query_as( 337 337 + "INSERT INTO backfill_jobs (collection, did) VALUES ($1, $2) RETURNING id::text", 338 338 + ) 339 339 + .bind(&body.collection) 340 340 + .bind(&body.did) 341 341 + .fetch_one(&state.db) 342 342 + .await 343 343 + .map_err(|e| AppError::Internal(format!("failed to create backfill job: {e}")))?; 344 344 + 345 345 + Ok(( 346 346 + StatusCode::CREATED, 347 347 + Json(serde_json::json!({ 348 348 + "id": row.0, 349 349 + "status": "pending", 350 350 + })), 351 351 + )) 352 352 + } 353 353 + 354 354 + /// GET /admin/backfill/status — list all backfill jobs. 355 355 + async fn backfill_status( 356 356 + State(state): State<AppState>, 357 357 + _admin: AdminAuth, 358 358 + ) -> Result<Json<Vec<BackfillJob>>, AppError> { 359 359 + let rows: Vec<( 360 360 + String, 361 361 + Option<String>, 362 362 + Option<String>, 363 363 + String, 364 364 + Option<i32>, 365 365 + Option<i32>, 366 366 + Option<i32>, 367 367 + Option<String>, 368 368 + Option<chrono::DateTime<chrono::Utc>>, 369 369 + Option<chrono::DateTime<chrono::Utc>>, 370 370 + chrono::DateTime<chrono::Utc>, 371 371 + )> = sqlx::query_as( 372 372 + "SELECT id::text, collection, did, status, total_repos, processed_repos, total_records, error, started_at, completed_at, created_at FROM backfill_jobs ORDER BY created_at DESC", 373 373 + ) 374 374 + .fetch_all(&state.db) 375 375 + .await 376 376 + .map_err(|e| AppError::Internal(format!("failed to list backfill jobs: {e}")))?; 377 377 + 378 378 + let jobs: Vec<BackfillJob> = rows 379 379 + .into_iter() 380 380 + .map( 381 381 + |(id, collection, did, status, total_repos, processed_repos, total_records, error, started_at, completed_at, created_at)| { 382 382 + BackfillJob { 383 383 + id, 384 384 + collection, 385 385 + did, 386 386 + status, 387 387 + total_repos, 388 388 + processed_repos, 389 389 + total_records, 390 390 + error, 391 391 + started_at, 392 392 + completed_at, 393 393 + created_at, 394 394 + } 395 395 + }, 396 396 + ) 397 397 + .collect(); 398 398 + 399 399 + Ok(Json(jobs)) 400 400 + }

+363

src/backfill.rs

reviewed

··· 1 1 + use serde::Deserialize; 2 2 + use serde_json::Value; 3 3 + use sqlx::PgPool; 4 4 + use std::sync::Arc; 5 5 + use tokio::sync::Semaphore; 6 6 + use tracing::{debug, error, info, warn}; 7 7 + 8 8 + use crate::profile; 9 9 + 10 10 + // --------------------------------------------------------------------------- 11 11 + // Relay / PDS response types 12 12 + // --------------------------------------------------------------------------- 13 13 + 14 14 + #[derive(Deserialize)] 15 15 + struct ListReposResponse { 16 16 + repos: Vec<RepoEntry>, 17 17 + cursor: Option<String>, 18 18 + } 19 19 + 20 20 + #[derive(Deserialize)] 21 21 + struct RepoEntry { 22 22 + did: String, 23 23 + } 24 24 + 25 25 + #[derive(Deserialize)] 26 26 + struct ListRecordsResponse { 27 27 + records: Vec<RecordEntry>, 28 28 + cursor: Option<String>, 29 29 + } 30 30 + 31 31 + #[derive(Deserialize)] 32 32 + struct RecordEntry { 33 33 + uri: String, 34 34 + cid: String, 35 35 + value: Value, 36 36 + } 37 37 + 38 38 + // --------------------------------------------------------------------------- 39 39 + // Relay discovery 40 40 + // --------------------------------------------------------------------------- 41 41 + 42 42 + /// Discover all DIDs that have records in `collection` via the relay's 43 43 + /// `com.atproto.sync.listReposByCollection` endpoint. Paginates until done. 44 44 + async fn list_repos_by_collection( 45 45 + http: &reqwest::Client, 46 46 + relay_url: &str, 47 47 + collection: &str, 48 48 + ) -> Result<Vec<String>, String> { 49 49 + let base = relay_url.trim_end_matches('/'); 50 50 + let mut dids = Vec::new(); 51 51 + let mut cursor: Option<String> = None; 52 52 + 53 53 + loop { 54 54 + let mut url = format!( 55 55 + "{base}/xrpc/com.atproto.sync.listReposByCollection?collection={collection}&limit=1000" 56 56 + ); 57 57 + if let Some(ref c) = cursor { 58 58 + url.push_str(&format!("&cursor={c}")); 59 59 + } 60 60 + 61 61 + let resp = http 62 62 + .get(&url) 63 63 + .send() 64 64 + .await 65 65 + .map_err(|e| format!("relay request failed: {e}"))?; 66 66 + 67 67 + if !resp.status().is_success() { 68 68 + return Err(format!("relay returned {}", resp.status())); 69 69 + } 70 70 + 71 71 + let body: ListReposResponse = resp 72 72 + .json() 73 73 + .await 74 74 + .map_err(|e| format!("invalid relay response: {e}"))?; 75 75 + 76 76 + let page_count = body.repos.len(); 77 77 + for repo in body.repos { 78 78 + dids.push(repo.did); 79 79 + } 80 80 + 81 81 + match body.cursor { 82 82 + Some(c) if page_count > 0 => cursor = Some(c), 83 83 + _ => break, 84 84 + } 85 85 + } 86 86 + 87 87 + Ok(dids) 88 88 + } 89 89 + 90 90 + // --------------------------------------------------------------------------- 91 91 + // PDS record fetching 92 92 + // --------------------------------------------------------------------------- 93 93 + 94 94 + /// Fetch all records for a DID + collection from their PDS via 95 95 + /// `com.atproto.repo.listRecords`. Paginates until done. 96 96 + async fn fetch_records( 97 97 + http: &reqwest::Client, 98 98 + pds_url: &str, 99 99 + did: &str, 100 100 + collection: &str, 101 101 + ) -> Result<Vec<(String, String, String, Value)>, String> { 102 102 + let base = pds_url.trim_end_matches('/'); 103 103 + let mut records = Vec::new(); 104 104 + let mut cursor: Option<String> = None; 105 105 + 106 106 + loop { 107 107 + let mut url = format!( 108 108 + "{base}/xrpc/com.atproto.repo.listRecords?repo={did}&collection={collection}&limit=100" 109 109 + ); 110 110 + if let Some(ref c) = cursor { 111 111 + url.push_str(&format!("&cursor={c}")); 112 112 + } 113 113 + 114 114 + let resp = http 115 115 + .get(&url) 116 116 + .send() 117 117 + .await 118 118 + .map_err(|e| format!("PDS listRecords failed: {e}"))?; 119 119 + 120 120 + if !resp.status().is_success() { 121 121 + return Err(format!("PDS returned {} for {did}", resp.status())); 122 122 + } 123 123 + 124 124 + let body: ListRecordsResponse = resp 125 125 + .json() 126 126 + .await 127 127 + .map_err(|e| format!("invalid PDS listRecords response: {e}"))?; 128 128 + 129 129 + let page_count = body.records.len(); 130 130 + for entry in body.records { 131 131 + let rkey = entry.uri.split('/').last().unwrap_or_default().to_string(); 132 132 + records.push((entry.uri, rkey, entry.cid, entry.value)); 133 133 + } 134 134 + 135 135 + match body.cursor { 136 136 + Some(c) if page_count > 0 => cursor = Some(c), 137 137 + _ => break, 138 138 + } 139 139 + } 140 140 + 141 141 + Ok(records) 142 142 + } 143 143 + 144 144 + // --------------------------------------------------------------------------- 145 145 + // Job runner 146 146 + // --------------------------------------------------------------------------- 147 147 + 148 148 + /// Run a single backfill job: discover repos, fetch records, upsert into DB. 149 149 + async fn run_job( 150 150 + db: &PgPool, 151 151 + http: &reqwest::Client, 152 152 + relay_url: &str, 153 153 + job_id: &str, 154 154 + ) -> Result<(), String> { 155 155 + // Fetch the job 156 156 + let job: (Option<String>, Option<String>) = sqlx::query_as( 157 157 + "SELECT collection, did FROM backfill_jobs WHERE id::text = $1", 158 158 + ) 159 159 + .bind(job_id) 160 160 + .fetch_one(db) 161 161 + .await 162 162 + .map_err(|e| format!("failed to fetch job: {e}"))?; 163 163 + 164 164 + let (job_collection, job_did) = job; 165 165 + 166 166 + // Mark as running 167 167 + let _ = sqlx::query( 168 168 + "UPDATE backfill_jobs SET status = 'running', started_at = NOW() WHERE id::text = $1", 169 169 + ) 170 170 + .bind(job_id) 171 171 + .execute(db) 172 172 + .await; 173 173 + 174 174 + // Determine target collections 175 175 + let collections: Vec<String> = if let Some(ref col) = job_collection { 176 176 + vec![col.clone()] 177 177 + } else { 178 178 + // All backfill-eligible collections 179 179 + let rows: Vec<(String,)> = sqlx::query_as( 180 180 + "SELECT id FROM lexicons WHERE backfill = TRUE AND lexicon_json->'defs'->'main'->>'type' = 'record'", 181 181 + ) 182 182 + .fetch_all(db) 183 183 + .await 184 184 + .map_err(|e| format!("failed to query backfill-eligible lexicons: {e}"))?; 185 185 + rows.into_iter().map(|(id,)| id).collect() 186 186 + }; 187 187 + 188 188 + if collections.is_empty() { 189 189 + let _ = sqlx::query( 190 190 + "UPDATE backfill_jobs SET status = 'completed', completed_at = NOW(), error = 'no backfill-eligible collections' WHERE id::text = $1", 191 191 + ) 192 192 + .bind(job_id) 193 193 + .execute(db) 194 194 + .await; 195 195 + return Ok(()); 196 196 + } 197 197 + 198 198 + info!(job = job_id, ?collections, "starting backfill"); 199 199 + 200 200 + let semaphore = Arc::new(Semaphore::new(8)); 201 201 + let mut total_repos = 0i32; 202 202 + let mut processed_repos = 0i32; 203 203 + let mut total_records = 0i32; 204 204 + 205 205 + for collection in &collections { 206 206 + // Discover DIDs 207 207 + let dids = if let Some(ref did) = job_did { 208 208 + vec![did.clone()] 209 209 + } else { 210 210 + match list_repos_by_collection(http, relay_url, collection).await { 211 211 + Ok(dids) => dids, 212 212 + Err(e) => { 213 213 + warn!(collection, error = %e, "failed to discover repos, skipping"); 214 214 + continue; 215 215 + } 216 216 + } 217 217 + }; 218 218 + 219 219 + total_repos += dids.len() as i32; 220 220 + let _ = sqlx::query("UPDATE backfill_jobs SET total_repos = $2 WHERE id::text = $1") 221 221 + .bind(job_id) 222 222 + .bind(total_repos) 223 223 + .execute(db) 224 224 + .await; 225 225 + 226 226 + // Process each DID concurrently (bounded by semaphore) 227 227 + let mut tasks = Vec::new(); 228 228 + 229 229 + for did in dids { 230 230 + let permit = semaphore.clone().acquire_owned().await.unwrap(); 231 231 + let http = http.clone(); 232 232 + let db = db.clone(); 233 233 + let collection = collection.clone(); 234 234 + 235 235 + let task = tokio::spawn(async move { 236 236 + let _permit = permit; 237 237 + backfill_repo(&db, &http, &did, &collection).await 238 238 + }); 239 239 + tasks.push(task); 240 240 + } 241 241 + 242 242 + for task in tasks { 243 243 + match task.await { 244 244 + Ok(Ok(count)) => { 245 245 + total_records += count; 246 246 + processed_repos += 1; 247 247 + } 248 248 + Ok(Err(e)) => { 249 249 + warn!(error = %e, "repo backfill failed"); 250 250 + processed_repos += 1; 251 251 + } 252 252 + Err(e) => { 253 253 + warn!(error = %e, "repo backfill task panicked"); 254 254 + processed_repos += 1; 255 255 + } 256 256 + } 257 257 + 258 258 + // Update progress periodically 259 259 + let _ = sqlx::query( 260 260 + "UPDATE backfill_jobs SET processed_repos = $2, total_records = $3 WHERE id::text = $1", 261 261 + ) 262 262 + .bind(job_id) 263 263 + .bind(processed_repos) 264 264 + .bind(total_records) 265 265 + .execute(db) 266 266 + .await; 267 267 + } 268 268 + } 269 269 + 270 270 + // Mark completed 271 271 + let _ = sqlx::query( 272 272 + "UPDATE backfill_jobs SET status = 'completed', completed_at = NOW(), processed_repos = $2, total_records = $3 WHERE id::text = $1", 273 273 + ) 274 274 + .bind(job_id) 275 275 + .bind(processed_repos) 276 276 + .bind(total_records) 277 277 + .execute(db) 278 278 + .await; 279 279 + 280 280 + info!(job = job_id, processed_repos, total_records, "backfill completed"); 281 281 + Ok(()) 282 282 + } 283 283 + 284 284 + /// Backfill a single repo's records for a collection. Returns the number of 285 285 + /// records upserted. 286 286 + async fn backfill_repo( 287 287 + db: &PgPool, 288 288 + http: &reqwest::Client, 289 289 + did: &str, 290 290 + collection: &str, 291 291 + ) -> Result<i32, String> { 292 292 + // Resolve PDS 293 293 + let pds = profile::resolve_pds_endpoint(http, did) 294 294 + .await 295 295 + .map_err(|e| format!("PDS resolution failed for {did}: {e}"))?; 296 296 + 297 297 + // Fetch records 298 298 + let records = fetch_records(http, &pds, did, collection).await?; 299 299 + let count = records.len() as i32; 300 300 + 301 301 + debug!(did, collection, count, "fetched records from PDS"); 302 302 + 303 303 + // Upsert into DB 304 304 + for (uri, rkey, cid, value) in records { 305 305 + let _ = sqlx::query( 306 306 + r#" 307 307 + INSERT INTO records (uri, did, collection, rkey, record, cid) 308 308 + VALUES ($1, $2, $3, $4, $5, $6) 309 309 + ON CONFLICT (uri) DO UPDATE 310 310 + SET record = EXCLUDED.record, 311 311 + cid = EXCLUDED.cid 312 312 + "#, 313 313 + ) 314 314 + .bind(&uri) 315 315 + .bind(did) 316 316 + .bind(collection) 317 317 + .bind(&rkey) 318 318 + .bind(&value) 319 319 + .bind(&cid) 320 320 + .execute(db) 321 321 + .await 322 322 + .map_err(|e| format!("DB upsert failed for {uri}: {e}"))?; 323 323 + } 324 324 + 325 325 + Ok(count) 326 326 + } 327 327 + 328 328 + // --------------------------------------------------------------------------- 329 329 + // Background worker 330 330 + // --------------------------------------------------------------------------- 331 331 + 332 332 + /// Spawn a background task that polls for pending backfill jobs and runs them. 333 333 + pub fn spawn_worker(db: PgPool, http: reqwest::Client, relay_url: String) { 334 334 + tokio::spawn(async move { 335 335 + info!("backfill worker started"); 336 336 + loop { 337 337 + // Poll for a pending job 338 338 + let job: Option<(String,)> = sqlx::query_as( 339 339 + "SELECT id::text FROM backfill_jobs WHERE status = 'pending' ORDER BY created_at ASC LIMIT 1", 340 340 + ) 341 341 + .fetch_optional(&db) 342 342 + .await 343 343 + .unwrap_or(None); 344 344 + 345 345 + if let Some((job_id,)) = job { 346 346 + info!(job = %job_id, "picked up backfill job"); 347 347 + if let Err(e) = run_job(&db, &http, &relay_url, &job_id).await { 348 348 + error!(job = %job_id, error = %e, "backfill job failed"); 349 349 + let _ = sqlx::query( 350 350 + "UPDATE backfill_jobs SET status = 'failed', completed_at = NOW(), error = $2 WHERE id::text = $1", 351 351 + ) 352 352 + .bind(&job_id) 353 353 + .bind(&e) 354 354 + .execute(&db) 355 355 + .await; 356 356 + } 357 357 + } else { 358 358 + // No pending jobs, wait before polling again 359 359 + tokio::time::sleep(std::time::Duration::from_secs(5)).await; 360 360 + } 361 361 + } 362 362 + }); 363 363 + }

+3

src/config.rs

reviewed

··· 9 9 pub aip_url: String, 10 10 pub jetstream_url: String, 11 11 pub admin_secret: Option<String>, 12 12 + pub relay_url: String, 12 13 } 13 14 14 15 impl Config { ··· 26 27 jetstream_url: env::var("JETSTREAM_URL") 27 28 .unwrap_or_else(|_| "wss://jetstream2.us-west.bsky.network/subscribe".into()), 28 29 admin_secret: env::var("ADMIN_SECRET").ok(), 30 30 + relay_url: env::var("RELAY_URL") 31 31 + .unwrap_or_else(|_| "https://bsky.network".into()), 29 32 } 30 33 } 31 34

+2

src/main.rs

reviewed

··· 1 1 mod admin; 2 2 mod auth; 3 3 + mod backfill; 3 4 mod config; 4 5 mod error; 5 6 mod jetstream; ··· 66 67 }; 67 68 68 69 jetstream::spawn(state.db.clone(), config.jetstream_url.clone(), collections_rx); 70 70 + backfill::spawn_worker(state.db.clone(), state.http.clone(), config.relay_url.clone()); 69 71 70 72 let app = server::router(state); 71 73 let addr = config.listen_addr();