From bcddf4c60ea067e9766dc43aa945e4190a1f6f0d Mon Sep 17 00:00:00 2001 From: Trezy Date: Tue, 24 Mar 2026 15:15:32 +0000 Subject: [PATCH] feat: add backfill and realtime streaming directly to HappyView (no more Tap dependency) --- src/admin/backfill.rs | 448 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++---------------------------------------------------------------------------------------------------------------------------------------- src/admin/mod.rs | 2 -- src/admin/tap_stats.rs | 26 -------------------------- src/admin/types.rs | 2 +- src/config.rs | 19 +++++++++---------- src/jetstream.rs | 475 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ src/lib.rs | 3 ++- src/lua/atproto_api.rs | 3 +-- src/lua/db_api.rs | 3 +-- src/lua/execute.rs | 3 +-- src/lua/http_api.rs | 3 +-- src/main.rs | 26 +++++--------------------- src/record_handler.rs | 373 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ src/tap.rs | 799 ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- tests/common/app.rs | 3 +-- tests/lua_atproto_api.rs | 3 +-- tests/lua_db_api.rs | 3 +-- 17 file(s) changed, 1184 insertion(s)(+), 1010 deletion(s)(-) diff --git a/src/admin/backfill.rs b/src/admin/backfill.rs --- a/src/admin/backfill.rs +++ b/src/admin/backfill.rs @@ -1,6 +1,11 @@ +use std::collections::HashMap; +use std::sync::Arc; +use std::sync::atomic::{AtomicI32, Ordering}; + use axum::Json; use axum::extract::State; use axum::http::StatusCode; +use futures_util::stream::{self, StreamExt}; use serde::Deserialize; use serde_json::Value; use uuid::Uuid; @@ -9,7 +14,8 @@ use crate::AppState; use crate::db::{adapt_sql, now_rfc3339}; use crate::error::AppError; use crate::event_log::{EventLog, Severity, log_event}; -use crate::tap; +use crate::profile; +use crate::record_handler::{self, RecordEvent}; use super::auth::UserAuth; use super::permissions::Permission; @@ -30,6 +36,23 @@ struct RepoEntry { did: String, } +// --------------------------------------------------------------------------- +// PDS record types +// --------------------------------------------------------------------------- + +#[derive(Deserialize)] +struct ListRecordsResponse { + records: Vec, + cursor: Option, +} + +#[derive(Deserialize)] +struct RecordEntry { + uri: String, + cid: String, + 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( @@ -79,10 +102,95 @@ Ok(dids) } // --------------------------------------------------------------------------- +// PDS record fetching +// --------------------------------------------------------------------------- + +/// 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( + state: &AppState, + pds_endpoint: &str, + did: &str, + collection: &str, +) -> Result { + let base = pds_endpoint.trim_end_matches('/'); + let mut cursor: Option = None; + let mut count: u32 = 0; + + loop { + let mut url = format!( + "{base}/xrpc/com.atproto.repo.listRecords?repo={did}&collection={collection}&limit=100" + ); + if let Some(ref c) = cursor { + url.push_str(&format!("&cursor={c}")); + } + + let resp = state + .http + .get(&url) + .send() + .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; + continue; // retry same page + } + + if !resp.status().is_success() { + return Err(format!("PDS returned {}", resp.status())); + } + + let body: ListRecordsResponse = resp + .json() + .await + .map_err(|e| format!("invalid PDS response: {e}"))?; + + let page_count = body.records.len(); + + for entry in &body.records { + let rkey = entry.uri.rsplit('/').next().unwrap_or_default().to_string(); + + let event = RecordEvent { + did: did.to_string(), + collection: collection.to_string(), + rkey, + action: "create".to_string(), + record: Some(entry.value.clone()), + cid: Some(entry.cid.clone()), + }; + + record_handler::handle_record_event(state, &event).await; + count += 1; + } + + match body.cursor { + Some(c) if page_count > 0 => cursor = Some(c), + _ => break, + } + } + + Ok(count) +} + +// --------------------------------------------------------------------------- // Admin handlers // --------------------------------------------------------------------------- -/// POST /admin/backfill — create a backfill job, discover repos, and add them to Tap. +/// POST /admin/backfill — create a backfill job and spawn background work. pub(super) async fn create_backfill( State(state): State, admin: UserAuth, @@ -94,7 +202,7 @@ let now = now_rfc3339(); let job_id = Uuid::new_v4().to_string(); let sql = adapt_sql( - "INSERT INTO backfill_jobs (id, collection, did, created_at) VALUES (?, ?, ?, ?) RETURNING id", + "INSERT INTO backfill_jobs (id, collection, did, status, started_at, created_at) VALUES (?, ?, ?, 'running', ?, ?) RETURNING id", backend, ); let row: (String,) = sqlx::query_as(&sql) @@ -102,22 +210,12 @@ .bind(&job_id) .bind(&body.collection) .bind(&body.did) .bind(&now) + .bind(&now) .fetch_one(&state.db) .await .map_err(|e| AppError::Internal(format!("failed to create backfill job: {e}")))?; let job_id = row.0.clone(); - - let now = now_rfc3339(); - let sql = adapt_sql( - "UPDATE backfill_jobs SET status = 'running', started_at = ? WHERE id = ?", - backend, - ); - let _ = sqlx::query(&sql) - .bind(&now) - .bind(&job_id) - .execute(&state.db) - .await; log_event( &state.db, @@ -134,6 +232,31 @@ backend, ) .await; + // 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; + + // Determine target collections let collections: Vec = if let Some(ref col) = body.collection { let lexicon_exists: bool = state .lexicons @@ -142,26 +265,8 @@ .await .is_some_and(|lex| lex.lexicon_type == crate::lexicon::LexiconType::Record); if !lexicon_exists { let error = format!("no record-type lexicon registered for collection '{col}'"); - let now = now_rfc3339(); - let sql = adapt_sql( - "UPDATE backfill_jobs SET status = 'failed', completed_at = ?, error = ? WHERE id = ?", - backend, - ); - let _ = sqlx::query(&sql) - .bind(&now) - .bind(&error) - .bind(&job_id) - .execute(&state.db) - .await; - - return Ok(( - StatusCode::CREATED, - Json(serde_json::json!({ - "id": job_id, - "status": "failed", - "error": error, - })), - )); + fail_job(&state, &job_id, &error).await; + return; } vec![col.clone()] } else { @@ -169,38 +274,30 @@ let sql = adapt_sql( "SELECT id FROM lexicons WHERE backfill = 1 AND json_extract(lexicon_json, '$.defs.main.type') = 'record'", backend, ); - let rows: Vec<(String,)> = - sqlx::query_as(&sql) - .fetch_all(&state.db) - .await - .map_err(|e| { - AppError::Internal(format!("failed to query backfill-eligible lexicons: {e}")) - })?; + let rows: Vec<(String,)> = match sqlx::query_as(&sql).fetch_all(&state.db).await { + Ok(rows) => rows, + Err(e) => { + let error = format!("failed to query backfill-eligible lexicons: {e}"); + fail_job(&state, &job_id, &error).await; + return; + } + }; rows.into_iter().map(|(id,)| id).collect() }; if collections.is_empty() { - let now = now_rfc3339(); - let sql = adapt_sql( - "UPDATE backfill_jobs SET status = 'completed', completed_at = ?, error = 'no backfill-eligible collections' WHERE id = ?", - backend, - ); - let _ = sqlx::query(&sql) - .bind(&now) - .bind(&job_id) - .execute(&state.db) - .await; - - return Ok(( - StatusCode::CREATED, - Json(serde_json::json!({ - "id": job_id, - "status": "completed", - "error": "no backfill-eligible collections", - })), - )); + complete_job( + &state, + &job_id, + 0, + 0, + Some("no backfill-eligible collections"), + ) + .await; + return; } + // Discover DIDs let mut all_dids = Vec::new(); for collection in &collections { @@ -224,6 +321,7 @@ all_dids.dedup(); let total_repos = all_dids.len() as i32; + // Update total_repos in DB let sql = adapt_sql( "UPDATE backfill_jobs SET total_repos = ? WHERE id = ?", backend, @@ -234,82 +332,125 @@ .bind(&job_id) .execute(&state.db) .await; - for chunk in all_dids.chunks(1000) { - if let Err(e) = tap::remove_repos( - &state.http, - &state.config.tap_url, - state.config.tap_admin_password.as_deref(), - chunk, + 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 - { - tracing::warn!(error = %e, "failed to remove repos from tap, continuing"); - } + .await; + return; } - if !all_dids.is_empty() { - for chunk in all_dids.chunks(1000) { - if let Err(e) = tap::add_repos( - &state.http, - &state.config.tap_url, - state.config.tap_admin_password.as_deref(), - chunk, - ) - .await - { - tracing::warn!(error = %e, "failed to add repos to tap"); - let now = now_rfc3339(); - let sql = adapt_sql( - "UPDATE backfill_jobs SET status = 'failed', completed_at = ?, error = ? WHERE id = ?", - backend, - ); - let _ = sqlx::query(&sql) - .bind(&now) - .bind(&e) - .bind(&job_id) - .execute(&state.db) - .await; - - log_event( - &state.db, - EventLog { - event_type: "backfill.failed".to_string(), - severity: Severity::Error, - actor_did: None, - subject: body.collection.clone(), - detail: serde_json::json!({ - "job_id": job_id.clone(), - "error": e, - }), - }, - backend, - ) - .await; + // Resolve DIDs to PDS endpoints and group by PDS + let mut pds_to_dids: HashMap> = HashMap::new(); - return Ok(( - StatusCode::CREATED, - Json(serde_json::json!({ - "id": job_id, - "status": "failed", - "error": e, - })), - )); + 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 now = now_rfc3339(); - let sql = adapt_sql( - "UPDATE backfill_jobs SET status = 'completed', completed_at = ?, processed_repos = ? WHERE id = ?", - backend, - ); - let _ = sqlx::query(&sql) - .bind(&now) - .bind(total_repos) - .bind(&job_id) - .execute(&state.db) + 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; + let final_processed = processed_repos.load(Ordering::Relaxed); + let final_records = total_records.load(Ordering::Relaxed); + + complete_job(&state, &job_id, final_processed, final_records, None).await; + log_event( &state.db, EventLog { @@ -318,22 +459,57 @@ severity: Severity::Info, actor_did: None, subject: body.collection.clone(), detail: serde_json::json!({ - "job_id": job_id.clone(), - "total_repos": total_repos, + "job_id": job_id, + "total_repos": final_processed, + "total_records": final_records, }), }, backend, ) .await; +} - Ok(( - StatusCode::CREATED, - Json(serde_json::json!({ - "id": job_id, - "status": "completed", - "total_repos": total_repos, - })), - )) +// --------------------------------------------------------------------------- +// Helper functions +// --------------------------------------------------------------------------- + +async fn fail_job(state: &AppState, job_id: &str, error: &str) { + let now = now_rfc3339(); + let backend = state.db_backend; + let sql = adapt_sql( + "UPDATE backfill_jobs SET status = 'failed', completed_at = ?, error = ? WHERE id = ?", + backend, + ); + let _ = sqlx::query(&sql) + .bind(&now) + .bind(error) + .bind(job_id) + .execute(&state.db) + .await; +} + +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 sql = adapt_sql( + "UPDATE backfill_jobs SET status = 'completed', completed_at = ?, processed_repos = ?, total_records = ?, error = ? WHERE id = ?", + backend, + ); + let _ = sqlx::query(&sql) + .bind(&now) + .bind(processed_repos) + .bind(total_records) + .bind(error) + .bind(job_id) + .execute(&state.db) + .await; } /// GET /admin/backfill/status — list all backfill jobs. diff --git a/src/admin/mod.rs b/src/admin/mod.rs --- a/src/admin/mod.rs +++ b/src/admin/mod.rs @@ -12,7 +12,6 @@ mod records; mod script_variables; pub(crate) mod settings; mod stats; -mod tap_stats; mod types; mod users; @@ -55,7 +54,6 @@ .route( "/records/collection", delete(records::delete_collection_records), ) - .route("/tap/stats", get(tap_stats::tap_stats)) .route( "/network-lexicons", post(network_lexicons::add).get(network_lexicons::list), diff --git a/src/admin/tap_stats.rs b/src/admin/tap_stats.rs deleted file mode 100644 --- a/src/admin/tap_stats.rs +++ /dev/null @@ -1,26 +0,0 @@ -use axum::Json; -use axum::extract::State; - -use crate::AppState; -use crate::error::AppError; -use crate::tap; - -use super::auth::UserAuth; -use super::permissions::Permission; - -/// GET /admin/tap/stats — aggregate stats from Tap. -pub(super) async fn tap_stats( - State(state): State, - auth: UserAuth, -) -> Result, AppError> { - auth.require(Permission::StatsRead).await?; - let stats = tap::get_stats( - &state.http, - &state.config.tap_url, - state.config.tap_admin_password.as_deref(), - ) - .await - .map_err(AppError::BadGateway)?; - - Ok(Json(stats)) -} diff --git a/src/admin/types.rs b/src/admin/types.rs --- a/src/admin/types.rs +++ b/src/admin/types.rs @@ -62,7 +62,7 @@ // --------------------------------------------------------------------------- // Backfill types // --------------------------------------------------------------------------- -#[derive(Deserialize)] +#[derive(Deserialize, Clone)] pub(super) struct CreateBackfillBody { pub(super) collection: Option, pub(super) did: Option, diff --git a/src/config.rs b/src/config.rs --- a/src/config.rs +++ b/src/config.rs @@ -11,8 +11,7 @@ pub database_url: String, pub database_backend: DatabaseBackend, pub public_url: String, pub session_secret: String, - pub tap_url: String, - pub tap_admin_password: Option, + pub jetstream_url: String, pub relay_url: String, pub plc_url: String, pub static_dir: String, @@ -43,8 +42,8 @@ database_backend, public_url: env::var("PUBLIC_URL").expect("PUBLIC_URL must be set"), session_secret: env::var("SESSION_SECRET") .unwrap_or_else(|_| "change-me-in-production-not-secure".into()), - tap_url: env::var("TAP_URL").unwrap_or_else(|_| "http://localhost:2480".into()), - tap_admin_password: env::var("TAP_ADMIN_PASSWORD").ok(), + jetstream_url: env::var("JETSTREAM_URL") + .unwrap_or_else(|_| "wss://jetstream1.us-east.bsky.network".into()), relay_url: env::var("RELAY_URL").unwrap_or_else(|_| "https://bsky.network".into()), plc_url: env::var("PLC_URL").unwrap_or_else(|_| "https://plc.directory".into()), static_dir: env::var("STATIC_DIR").unwrap_or_else(|_| "./web/out".into()), @@ -86,8 +85,7 @@ "DATABASE_URL", "DATABASE_BACKEND", "PUBLIC_URL", "SESSION_SECRET", - "TAP_URL", - "TAP_ADMIN_PASSWORD", + "JETSTREAM_URL", "RELAY_URL", "PLC_URL", "EVENT_LOG_RETENTION_DAYS", @@ -118,8 +116,7 @@ database_url: String::new(), database_backend: DatabaseBackend::Postgres, public_url: String::new(), session_secret: String::new(), - tap_url: String::new(), - tap_admin_password: None, + jetstream_url: String::new(), relay_url: String::new(), plc_url: String::new(), static_dir: String::new(), @@ -158,8 +155,10 @@ } let config = Config::from_env(); assert_eq!(config.host, "0.0.0.0"); assert_eq!(config.port, 3000); - assert_eq!(config.tap_url, "http://localhost:2480"); - assert!(config.tap_admin_password.is_none()); + assert_eq!( + config.jetstream_url, + "wss://jetstream1.us-east.bsky.network" + ); assert_eq!(config.relay_url, "https://bsky.network"); assert_eq!(config.plc_url, "https://plc.directory"); } diff --git a/src/jetstream.rs b/src/jetstream.rs new file mode 100644 --- /dev/null +++ b/src/jetstream.rs @@ -0,0 +1,475 @@ +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use futures_util::StreamExt; +use serde::Deserialize; +use serde_json::Value; +use sqlx::AnyPool; +use tokio::sync::{Semaphore, watch}; +use tokio_tungstenite::tungstenite::Message; + +use crate::AppState; +use crate::db::{DatabaseBackend, adapt_sql, now_rfc3339}; +use crate::event_log::{EventLog, Severity, log_event}; +use crate::record_handler::{self, LEXICON_SCHEMA_COLLECTION, RecordEvent}; + +// --------------------------------------------------------------------------- +// Jetstream event types +// --------------------------------------------------------------------------- + +#[derive(Deserialize)] +struct JetstreamEvent { + #[serde(rename = "kind")] + event_type: String, + did: String, + time_us: i64, + commit: Option, + identity: Option, +} + +#[derive(Deserialize)] +struct CommitEvent { + operation: String, + collection: String, + rkey: String, + record: Option, + cid: Option, +} + +#[derive(Deserialize)] +struct IdentityEvent { + did: String, + handle: Option, +} + +// --------------------------------------------------------------------------- +// Cursor persistence +// --------------------------------------------------------------------------- + +async fn load_cursor(db: &AnyPool, backend: DatabaseBackend) -> Option { + let sql = adapt_sql("SELECT value FROM instance_settings WHERE key = ?", backend); + let row: Option<(String,)> = sqlx::query_as(&sql) + .bind("jetstream_cursor") + .fetch_optional(db) + .await + .ok()?; + row.and_then(|(v,)| v.parse::().ok()) +} + +async fn save_cursor(db: &AnyPool, backend: DatabaseBackend, cursor: i64) { + let now = now_rfc3339(); + let sql = adapt_sql( + r#" + INSERT INTO instance_settings (key, value, updated_at) + VALUES (?, ?, ?) + ON CONFLICT (key) DO UPDATE SET value = ?, updated_at = ? + "#, + backend, + ); + if let Err(e) = sqlx::query(&sql) + .bind("jetstream_cursor") + .bind(cursor.to_string()) + .bind(&now) + .bind(cursor.to_string()) + .bind(&now) + .execute(db) + .await + { + tracing::warn!("failed to save jetstream cursor: {e}"); + } +} + +// --------------------------------------------------------------------------- +// URL builder +// --------------------------------------------------------------------------- + +fn build_subscribe_url(base_url: &str, collections: &[String]) -> String { + let base = base_url.trim_end_matches('/'); + let mut url = format!("{base}/subscribe?compress=false"); + + let mut has_lexicon_schema = false; + for col in collections { + url.push_str(&format!("&wantedCollections={col}")); + if col == LEXICON_SCHEMA_COLLECTION { + has_lexicon_schema = true; + } + } + + if !has_lexicon_schema { + url.push_str(&format!("&wantedCollections={LEXICON_SCHEMA_COLLECTION}")); + } + + url +} + +// --------------------------------------------------------------------------- +// Public API +// --------------------------------------------------------------------------- + +/// Spawn a background task that connects to Jetstream's WebSocket firehose and +/// processes record + identity events. Uses exponential backoff on disconnects. +pub fn spawn(state: AppState, mut collections_rx: watch::Receiver>) { + tokio::spawn(async move { + let mut backoff = Duration::from_secs(2); + let max_backoff = Duration::from_secs(60); + + loop { + match run(&state, &mut collections_rx).await { + Ok(()) => { + // Clean reconnect (collection change) — reset backoff. + backoff = Duration::from_secs(2); + tracing::info!("jetstream reconnecting due to collection change"); + } + Err(e) => { + tracing::warn!("jetstream disconnected: {e}"); + tracing::info!( + backoff_secs = backoff.as_secs(), + "reconnecting to jetstream after backoff" + ); + tokio::time::sleep(backoff).await; + backoff = (backoff * 2).min(max_backoff); + } + } + } + }); +} + +// --------------------------------------------------------------------------- +// Connection loop +// --------------------------------------------------------------------------- + +async fn run( + state: &AppState, + collections_rx: &mut watch::Receiver>, +) -> Result<(), Box> { + let db = &state.db; + let backend = state.db_backend; + + // Build subscribe URL with current collections. + let collections = collections_rx.borrow().clone(); + let url = build_subscribe_url(&state.config.jetstream_url, &collections); + + // Load cursor from DB; skip if older than 72 hours. + let cursor = load_cursor(db, backend).await; + let subscribe_url = if let Some(c) = cursor { + let now_us = chrono::Utc::now().timestamp_micros(); + let age_hours = (now_us - c) as f64 / 3_600_000_000.0; + if age_hours > 72.0 { + tracing::warn!( + cursor = c, + age_hours = age_hours, + "jetstream cursor is older than 72 hours, skipping it" + ); + url + } else { + format!("{url}&cursor={c}") + } + } else { + url + }; + + tracing::info!(url = %subscribe_url, "connecting to jetstream"); + + let (ws, _): ( + tokio_tungstenite::WebSocketStream< + tokio_tungstenite::MaybeTlsStream, + >, + _, + ) = tokio::time::timeout( + Duration::from_secs(15), + tokio_tungstenite::connect_async(&subscribe_url), + ) + .await + .map_err(|_| "jetstream websocket connection timed out after 15s")??; + + log_event( + db, + EventLog { + event_type: "jetstream.connected".to_string(), + severity: Severity::Info, + actor_did: None, + subject: None, + detail: serde_json::json!({ "url": subscribe_url }), + }, + backend, + ) + .await; + + let (_write, mut read) = ws.split(); + + let semaphore = Arc::new(Semaphore::new(50)); + + // Cursor batching state: save every 1000 events or 5 seconds. + let mut latest_cursor: Option = cursor; + let mut events_since_flush: u64 = 0; + let mut last_flush = Instant::now(); + let flush_interval = Duration::from_secs(5); + const FLUSH_EVENT_THRESHOLD: u64 = 1000; + + loop { + tokio::select! { + msg = read.next() => { + let msg = match msg { + Some(Ok(m)) => m, + Some(Err(e)) => { + tracing::warn!("jetstream websocket read error: {e}"); + // Flush cursor before returning error. + if let Some(c) = latest_cursor { + save_cursor(db, backend, c).await; + } + log_event( + db, + EventLog { + event_type: "jetstream.disconnected".to_string(), + severity: Severity::Warn, + actor_did: None, + subject: None, + detail: serde_json::json!({ "reason": e.to_string() }), + }, + backend, + ) + .await; + return Err(e.into()); + } + None => { + tracing::info!("jetstream websocket stream ended"); + break; + } + }; + + let text = match msg { + Message::Text(t) => t, + Message::Close(_) => { + tracing::info!("jetstream websocket received close frame"); + break; + } + other => { + tracing::debug!(msg_type = ?other, "ignoring non-text websocket message"); + continue; + } + }; + + let event: JetstreamEvent = match serde_json::from_str(&text) { + Ok(e) => e, + Err(e) => { + tracing::warn!("skipping unparseable jetstream event: {e}"); + continue; + } + }; + + // Update cursor tracking. + latest_cursor = Some(event.time_us); + events_since_flush += 1; + + match event.event_type.as_str() { + "commit" => { + if let Some(commit) = event.commit { + tracing::debug!( + operation = %commit.operation, + collection = %commit.collection, + did = %event.did, + rkey = %commit.rkey, + "received commit event from jetstream" + ); + + let record_event = RecordEvent { + did: event.did.clone(), + collection: commit.collection, + rkey: commit.rkey, + action: commit.operation, + record: commit.record, + cid: commit.cid, + }; + + let sem = semaphore.clone(); + let state = state.clone(); + tokio::spawn(async move { + let _permit = sem.acquire().await.unwrap(); + record_handler::handle_record_event(&state, &record_event).await; + }); + } + } + "identity" => { + if let Some(identity) = event.identity { + tracing::info!( + did = %identity.did, + handle = ?identity.handle, + "received identity event from jetstream" + ); + } + } + "account" => { + tracing::debug!( + did = %event.did, + "received account event from jetstream (ignored)" + ); + } + other => { + tracing::debug!(event_type = %other, "unknown jetstream event type"); + } + } + + // Flush cursor periodically. + if (events_since_flush >= FLUSH_EVENT_THRESHOLD + || last_flush.elapsed() >= flush_interval) + && let Some(c) = latest_cursor + { + save_cursor(db, backend, c).await; + events_since_flush = 0; + last_flush = Instant::now(); + } + } + _ = collections_rx.changed() => { + let new_collections = collections_rx.borrow_and_update().clone(); + tracing::info!(?new_collections, "collection filter changed, reconnecting to jetstream"); + + // Flush cursor before reconnecting. + if let Some(c) = latest_cursor { + save_cursor(db, backend, c).await; + } + + return Ok(()); + } + } + } + + // Stream ended cleanly — flush cursor and log disconnect. + if let Some(c) = latest_cursor { + save_cursor(db, backend, c).await; + } + + log_event( + db, + EventLog { + event_type: "jetstream.disconnected".to_string(), + severity: Severity::Warn, + actor_did: None, + subject: None, + detail: serde_json::json!({ "reason": "connection closed" }), + }, + backend, + ) + .await; + + Ok(()) +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_build_subscribe_url_basic() { + let url = build_subscribe_url( + "https://jetstream.example.com", + &["app.bsky.feed.post".to_string()], + ); + assert!(url.starts_with("https://jetstream.example.com/subscribe?compress=false")); + assert!(url.contains("&wantedCollections=app.bsky.feed.post")); + assert!(url.contains(&format!("&wantedCollections={LEXICON_SCHEMA_COLLECTION}"))); + } + + #[test] + fn test_build_subscribe_url_strips_trailing_slash() { + let url = build_subscribe_url( + "https://jetstream.example.com/", + &["app.bsky.feed.post".to_string()], + ); + assert!(url.starts_with("https://jetstream.example.com/subscribe")); + assert!(!url.contains("//subscribe")); + } + + #[test] + fn test_build_subscribe_url_does_not_duplicate_lexicon_schema() { + let url = build_subscribe_url( + "https://jetstream.example.com", + &[ + "app.bsky.feed.post".to_string(), + LEXICON_SCHEMA_COLLECTION.to_string(), + ], + ); + let count = url.matches(LEXICON_SCHEMA_COLLECTION).count(); + assert_eq!( + count, 1, + "lexicon schema collection should appear exactly once" + ); + } + + #[test] + fn test_build_subscribe_url_empty_collections() { + let url = build_subscribe_url("https://jetstream.example.com", &[]); + assert!(url.contains(&format!("&wantedCollections={LEXICON_SCHEMA_COLLECTION}"))); + } + + #[test] + fn test_deserialize_commit_event() { + let json = r#"{ + "kind": "commit", + "did": "did:plc:abc123", + "time_us": 1700000000000000, + "commit": { + "operation": "create", + "collection": "app.bsky.feed.post", + "rkey": "3k2y6e7wh4k2a", + "record": {"text": "hello world", "$type": "app.bsky.feed.post"}, + "cid": "bafyreiabc123" + } + }"#; + + let event: JetstreamEvent = serde_json::from_str(json).unwrap(); + assert_eq!(event.event_type, "commit"); + assert_eq!(event.did, "did:plc:abc123"); + assert_eq!(event.time_us, 1700000000000000); + + let commit = event.commit.unwrap(); + assert_eq!(commit.operation, "create"); + assert_eq!(commit.collection, "app.bsky.feed.post"); + assert_eq!(commit.rkey, "3k2y6e7wh4k2a"); + assert!(commit.record.is_some()); + assert_eq!(commit.cid.as_deref(), Some("bafyreiabc123")); + } + + #[test] + fn test_deserialize_identity_event() { + let json = r#"{ + "kind": "identity", + "did": "did:plc:abc123", + "time_us": 1700000000000000, + "identity": { + "did": "did:plc:abc123", + "handle": "alice.bsky.social" + } + }"#; + + let event: JetstreamEvent = serde_json::from_str(json).unwrap(); + assert_eq!(event.event_type, "identity"); + + let identity = event.identity.unwrap(); + assert_eq!(identity.did, "did:plc:abc123"); + assert_eq!(identity.handle.as_deref(), Some("alice.bsky.social")); + } + + #[test] + fn test_deserialize_delete_commit_no_record() { + let json = r#"{ + "kind": "commit", + "did": "did:plc:abc123", + "time_us": 1700000000000000, + "commit": { + "operation": "delete", + "collection": "app.bsky.feed.post", + "rkey": "3k2y6e7wh4k2a" + } + }"#; + + let event: JetstreamEvent = serde_json::from_str(json).unwrap(); + let commit = event.commit.unwrap(); + assert_eq!(commit.operation, "delete"); + assert!(commit.record.is_none()); + assert!(commit.cid.is_none()); + } +} diff --git a/src/lib.rs b/src/lib.rs --- a/src/lib.rs +++ b/src/lib.rs @@ -6,17 +6,18 @@ pub mod dns; pub mod error; pub mod event_log; pub mod external_auth; +pub mod jetstream; pub mod labeler; pub mod lexicon; pub mod lua; pub mod plugin; pub mod profile; pub mod rate_limit; +pub mod record_handler; pub mod record_refs; pub mod repo; pub mod resolve; pub mod server; -pub mod tap; pub mod xrpc; use auth::oauth_store::{DbSessionStore, DbStateStore}; diff --git a/src/lua/atproto_api.rs b/src/lua/atproto_api.rs --- a/src/lua/atproto_api.rs +++ b/src/lua/atproto_api.rs @@ -212,8 +212,7 @@ database_url: String::new(), database_backend: crate::db::DatabaseBackend::Sqlite, public_url: String::new(), session_secret: "test-secret".into(), - tap_url: String::new(), - tap_admin_password: None, + jetstream_url: String::new(), relay_url: String::new(), plc_url: plc_url.to_string(), static_dir: String::new(), diff --git a/src/lua/db_api.rs b/src/lua/db_api.rs --- a/src/lua/db_api.rs +++ b/src/lua/db_api.rs @@ -620,8 +620,7 @@ database_url: String::new(), database_backend: crate::db::DatabaseBackend::Sqlite, public_url: String::new(), session_secret: "test-secret".into(), - tap_url: String::new(), - tap_admin_password: None, + jetstream_url: String::new(), relay_url: String::new(), plc_url: String::new(), static_dir: String::new(), diff --git a/src/lua/execute.rs b/src/lua/execute.rs --- a/src/lua/execute.rs +++ b/src/lua/execute.rs @@ -954,8 +954,7 @@ database_url: String::new(), database_backend: crate::db::DatabaseBackend::Sqlite, public_url: String::new(), session_secret: "test-secret".into(), - tap_url: String::new(), - tap_admin_password: None, + jetstream_url: String::new(), relay_url: String::new(), plc_url: String::new(), static_dir: String::new(), diff --git a/src/lua/http_api.rs b/src/lua/http_api.rs --- a/src/lua/http_api.rs +++ b/src/lua/http_api.rs @@ -94,8 +94,7 @@ database_url: String::new(), database_backend: crate::db::DatabaseBackend::Sqlite, public_url: String::new(), session_secret: "test-secret".into(), - tap_url: String::new(), - tap_admin_password: None, + jetstream_url: String::new(), relay_url: String::new(), plc_url: String::new(), static_dir: String::new(), diff --git a/src/main.rs b/src/main.rs --- a/src/main.rs +++ b/src/main.rs @@ -7,7 +7,7 @@ use happyview::dns::NativeDnsResolver; use happyview::lexicon::{LexiconRegistry, ParsedLexicon, ProcedureAction}; use happyview::rate_limit::RateLimiter; use happyview::resolve::{fetch_lexicon_from_pds, resolve_nsid_authority}; -use happyview::{AppState, labeler, server, tap}; +use happyview::{AppState, jetstream, labeler, server}; use tokio::sync::watch; use tracing::{info, warn}; @@ -21,6 +21,9 @@ #[tokio::main] async fn main() { dotenvy::dotenv().ok(); + + // Install rustls crypto provider early so all TLS users (jetstream, labeler, etc.) can find it. + let _ = rustls::crypto::ring::default_provider().install_default(); tracing_subscriber::fmt() .with_env_filter( @@ -350,7 +353,6 @@ let cookie_key = axum_extra::extract::cookie::Key::derive_from(config.session_secret.as_bytes()); let initial_collections = lexicons.get_record_collections().await; - let initial_collections_for_sync = initial_collections.clone(); let (collections_tx, collections_rx) = watch::channel(initial_collections); let (labeler_subscriptions_tx, labeler_subscriptions_rx) = watch::channel(()); @@ -370,25 +372,7 @@ wasm_runtime, attestation_signer, }; - // Sync initial collections to Tap on startup. - { - let mut wanted = initial_collections_for_sync; - if !wanted.contains(&"com.atproto.lexicon.schema".to_string()) { - wanted.push("com.atproto.lexicon.schema".to_string()); - } - if let Err(e) = tap::sync_collections( - &state.http, - &config.tap_url, - config.tap_admin_password.as_deref(), - &wanted, - ) - .await - { - warn!("failed to sync initial collections to tap: {e}"); - } - } - - tap::spawn(state.clone(), collections_rx); + jetstream::spawn(state.clone(), collections_rx); labeler::spawn(state.clone(), labeler_subscriptions_rx); tokio::spawn(labeler::spawn_label_gc(state.db.clone(), state.db_backend)); diff --git a/src/record_handler.rs b/src/record_handler.rs new file mode 100644 --- /dev/null +++ b/src/record_handler.rs @@ -0,0 +1,373 @@ +use std::sync::Arc; + +use serde_json::Value; + +use crate::AppState; +use crate::db::{adapt_sql, now_rfc3339}; +use crate::event_log::{EventLog, Severity, log_event}; +use crate::lexicon::{LexiconType, ParsedLexicon, ProcedureAction}; + +/// The static collection we always include for lexicon schema updates. +pub const LEXICON_SCHEMA_COLLECTION: &str = "com.atproto.lexicon.schema"; + +/// A generic record event that can originate from any source (Jetstream, backfill, etc.). +pub struct RecordEvent { + pub did: String, + pub collection: String, + pub rkey: String, + pub action: String, + pub record: Option, + pub cid: Option, +} + +/// Process a record event: upsert/delete the record in the database, run index +/// hooks, and handle lexicon schema events. +pub async fn handle_record_event(state: &AppState, record: &RecordEvent) { + let db = &state.db; + let lexicons = &state.lexicons; + + let uri = format!("at://{}/{}/{}", record.did, record.collection, record.rkey); + + // Handle lexicon schema events for tracked network lexicons. + if record.collection == LEXICON_SCHEMA_COLLECTION { + handle_lexicon_schema_event(state, &record.did, record).await; + return; + } + + // Skip records whose collection is not tracked by a registered record-type lexicon. + let is_tracked = lexicons + .get(&record.collection) + .await + .is_some_and(|lex| lex.lexicon_type == LexiconType::Record); + + if !is_tracked { + tracing::debug!( + collection = %record.collection, + "skipping record for untracked collection" + ); + return; + } + + match record.action.as_str() { + "create" | "update" => { + let rec = match &record.record { + Some(r) => r, + None => return, + }; + let cid = record.cid.as_deref().unwrap_or_default(); + + // Run index hook before storing, if configured. The hook's return + // value determines what (if anything) gets written to the DB. + let rec_to_store = + if let Some(script) = state.lexicons.get_index_hook(&record.collection).await { + let hook_result = crate::lua::execute_hook_script(&crate::lua::HookEvent { + state, + lexicon_id: &record.collection, + script: &script, + action: &record.action, + uri: &uri, + did: &record.did, + collection: &record.collection, + rkey: &record.rkey, + record: Some(rec), + }) + .await; + + match hook_result { + None => { + // Hook returned nil — skip indexing this record. + log_event( + db, + EventLog { + event_type: "record.skipped".to_string(), + severity: Severity::Info, + actor_did: None, + subject: Some(uri.clone()), + detail: serde_json::json!({ + "collection": record.collection, + "did": record.did, + "rkey": record.rkey, + "reason": "hook returned nil", + }), + }, + state.db_backend, + ) + .await; + return; + } + Some(v) => v, + } + } else { + // No hook — store the original record as-is. + rec.clone() + }; + + let now = now_rfc3339(); + let backend = state.db_backend; + let insert_sql = adapt_sql( + r#" + INSERT INTO records (uri, did, collection, rkey, record, cid, indexed_at, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT (uri) DO UPDATE + SET record = EXCLUDED.record, + cid = EXCLUDED.cid, + indexed_at = ? + "#, + backend, + ); + match sqlx::query(&insert_sql) + .bind(&uri) + .bind(&record.did) + .bind(&record.collection) + .bind(&record.rkey) + .bind(serde_json::to_string(&rec_to_store).unwrap_or_default()) + .bind(cid) + .bind(&now) + .bind(&now) + .bind(&now) + .execute(db) + .await + { + Ok(_) => { + let _ = crate::record_refs::sync_refs( + db, + &uri, + &record.collection, + &rec_to_store, + backend, + ) + .await; + + log_event( + db, + EventLog { + event_type: "record.created".to_string(), + severity: Severity::Info, + actor_did: None, + subject: Some(uri.clone()), + detail: serde_json::json!({ + "collection": record.collection, + "did": record.did, + "rkey": record.rkey, + }), + }, + backend, + ) + .await; + + crate::labeler::backfill_labels_for_uri(Arc::new(state.clone()), uri.clone()); + } + Err(e) => { + tracing::warn!(uri = %uri, "failed to upsert record: {e}"); + log_event( + db, + EventLog { + event_type: "record.created".to_string(), + severity: Severity::Error, + actor_did: None, + subject: Some(uri.clone()), + detail: serde_json::json!({ + "collection": record.collection, + "did": record.did, + "rkey": record.rkey, + "error": e.to_string(), + }), + }, + backend, + ) + .await; + } + } + } + "delete" => { + let backend = state.db_backend; + + // Run index hook before deleting, if configured. + if let Some(script) = state.lexicons.get_index_hook(&record.collection).await { + let hook_result = crate::lua::execute_hook_script(&crate::lua::HookEvent { + state, + lexicon_id: &record.collection, + script: &script, + action: "delete", + uri: &uri, + did: &record.did, + collection: &record.collection, + rkey: &record.rkey, + record: None, + }) + .await; + + if hook_result.is_none() { + // Hook returned nil — skip the delete. + log_event( + db, + EventLog { + event_type: "record.skipped".to_string(), + severity: Severity::Info, + actor_did: None, + subject: Some(uri.clone()), + detail: serde_json::json!({ + "collection": record.collection, + "did": record.did, + "rkey": record.rkey, + "reason": "hook returned nil", + }), + }, + backend, + ) + .await; + return; + } + } + + let delete_sql = adapt_sql("DELETE FROM records WHERE uri = ?", backend); + match sqlx::query(&delete_sql).bind(&uri).execute(db).await { + Ok(_) => { + log_event( + db, + EventLog { + event_type: "record.deleted".to_string(), + severity: Severity::Info, + actor_did: None, + subject: Some(uri.clone()), + detail: serde_json::json!({ + "collection": record.collection, + "did": record.did, + "rkey": record.rkey, + }), + }, + backend, + ) + .await; + } + Err(e) => { + tracing::warn!(uri = %uri, "failed to delete record: {e}"); + log_event( + db, + EventLog { + event_type: "record.deleted".to_string(), + severity: Severity::Error, + actor_did: None, + subject: Some(uri.clone()), + detail: serde_json::json!({ + "collection": record.collection, + "did": record.did, + "rkey": record.rkey, + "error": e.to_string(), + }), + }, + backend, + ) + .await; + } + } + } + _ => {} + } +} + +/// Handle a `com.atproto.lexicon.schema` record event for tracked network lexicons. +pub async fn handle_lexicon_schema_event(state: &AppState, did: &str, record: &RecordEvent) { + let db = &state.db; + let lexicons = &state.lexicons; + let collections_tx = &state.collections_tx; + let nsid = &record.rkey; + + let backend = state.db_backend; + + // Check if this NSID is one we're tracking and the DID matches the authority. + let select_sql = adapt_sql( + "SELECT target_collection FROM lexicons WHERE id = ? AND source = 'network' AND authority_did = ?", + backend, + ); + let tracked: Option<(Option,)> = sqlx::query_as(&select_sql) + .bind(nsid) + .bind(did) + .fetch_optional(db) + .await + .unwrap_or(None); + + let target_collection = match tracked { + Some((tc,)) => tc, + None => return, // Not a tracked network lexicon. + }; + + match record.action.as_str() { + "create" | "update" => { + let rec = match &record.record { + Some(r) => r, + None => return, + }; + + let parsed = match ParsedLexicon::parse( + rec.clone(), + 1, + target_collection.clone(), + ProcedureAction::Upsert, + None, + None, + None, + ) { + Ok(p) => p, + Err(e) => { + tracing::warn!(nsid, "failed to parse lexicon schema event: {e}"); + return; + } + }; + + let is_record = parsed.lexicon_type == crate::lexicon::LexiconType::Record; + + // Upsert into lexicons table with last_fetched_at. + let now = now_rfc3339(); + let upsert_sql = adapt_sql( + r#" + INSERT INTO lexicons (id, lexicon_json, backfill, target_collection, source, authority_did, last_fetched_at, created_at) + VALUES (?, ?, 0, ?, 'network', ?, ?, ?) + ON CONFLICT (id) DO UPDATE SET + lexicon_json = EXCLUDED.lexicon_json, + target_collection = EXCLUDED.target_collection, + last_fetched_at = ?, + revision = lexicons.revision + 1, + updated_at = ? + "#, + backend, + ); + if let Err(e) = sqlx::query(&upsert_sql) + .bind(nsid) + .bind(serde_json::to_string(rec).unwrap_or_default()) + .bind(&target_collection) + .bind(did) + .bind(&now) + .bind(&now) + .bind(&now) + .bind(&now) + .execute(db) + .await + { + tracing::warn!(nsid, "failed to upsert lexicon from event: {e}"); + return; + } + + lexicons.upsert(parsed).await; + tracing::info!(nsid, "updated network lexicon from network event"); + + if is_record { + let collections = lexicons.get_record_collections().await; + let _ = collections_tx.send(collections); + } + } + "delete" => { + // Remove from lexicons table and registry. + let delete_sql = adapt_sql("DELETE FROM lexicons WHERE id = ?", backend); + let _ = sqlx::query(&delete_sql).bind(nsid).execute(db).await; + + let was_present = lexicons.remove(nsid).await; + if was_present { + tracing::info!(nsid, "removed network lexicon from network delete event"); + let collections = lexicons.get_record_collections().await; + let _ = collections_tx.send(collections); + } + } + _ => {} + } +} diff --git a/src/tap.rs b/src/tap.rs deleted file mode 100644 --- a/src/tap.rs +++ /dev/null @@ -1,799 +0,0 @@ -use std::sync::Arc; - -use futures_util::{SinkExt, StreamExt}; -use serde::{Deserialize, Serialize}; -use serde_json::Value; -use tokio::sync::{Semaphore, watch}; -use tokio_tungstenite::tungstenite::Message; -use tokio_tungstenite::tungstenite::client::IntoClientRequest; - -use crate::AppState; -use crate::db::{adapt_sql, now_rfc3339}; -use crate::event_log::{EventLog, Severity, log_event}; -use crate::lexicon::{LexiconType, ParsedLexicon, ProcedureAction}; - -// --------------------------------------------------------------------------- -// Tap event types (matches Tap's outbox JSON format) -// --------------------------------------------------------------------------- - -#[derive(Deserialize)] -struct TapEvent { - id: u64, - #[serde(rename = "type")] - event_type: String, - record: Option, - identity: Option, -} - -#[derive(Deserialize)] -struct TapRecordEvent { - did: String, - collection: String, - rkey: String, - action: String, - record: Option, - cid: Option, - #[allow(dead_code)] - live: Option, -} - -#[derive(Deserialize)] -#[allow(dead_code)] -struct TapIdentityEvent { - did: String, - handle: Option, - #[serde(rename = "isActive")] - is_active: Option, - status: Option, -} - -// --------------------------------------------------------------------------- -// Tap HTTP client helpers -// --------------------------------------------------------------------------- - -async fn tap_put( - http: &reqwest::Client, - tap_url: &str, - path: &str, - password: Option<&str>, - body: &Value, -) -> Result<(), String> { - let url = format!("{}{}", tap_url.trim_end_matches('/'), path); - let mut req = http.put(&url).json(body); - if let Some(pw) = password { - req = req.basic_auth("admin", Some(pw)); - } - let resp = req - .send() - .await - .map_err(|e| format!("tap HTTP request failed: {e}"))?; - if !resp.status().is_success() { - let status = resp.status(); - let body = resp.text().await.unwrap_or_default(); - return Err(format!("tap returned {status}: {body}")); - } - Ok(()) -} - -async fn tap_post( - http: &reqwest::Client, - tap_url: &str, - path: &str, - password: Option<&str>, - body: &Value, -) -> Result<(), String> { - let url = format!("{}{}", tap_url.trim_end_matches('/'), path); - let mut req = http.post(&url).json(body); - if let Some(pw) = password { - req = req.basic_auth("admin", Some(pw)); - } - let resp = req - .send() - .await - .map_err(|e| format!("tap HTTP request failed: {e}"))?; - if !resp.status().is_success() { - let status = resp.status(); - let body = resp.text().await.unwrap_or_default(); - return Err(format!("tap returned {status}: {body}")); - } - Ok(()) -} - -async fn tap_get( - http: &reqwest::Client, - tap_url: &str, - path: &str, - password: Option<&str>, -) -> Result { - let url = format!("{}{}", tap_url.trim_end_matches('/'), path); - let mut req = http.get(&url); - if let Some(pw) = password { - req = req.basic_auth("admin", Some(pw)); - } - let resp = req - .send() - .await - .map_err(|e| format!("tap HTTP request failed: {e}"))?; - if !resp.status().is_success() { - let status = resp.status(); - let body = resp.text().await.unwrap_or_default(); - return Err(format!("tap returned {status}: {body}")); - } - resp.json::() - .await - .map_err(|e| format!("failed to parse tap response: {e}")) -} - -// --------------------------------------------------------------------------- -// Tap stats -// --------------------------------------------------------------------------- - -#[derive(Serialize)] -pub struct TapStats { - pub repo_count: u64, - pub record_count: u64, - pub outbox_buffer: u64, -} - -#[derive(Deserialize)] -struct RepoCountResponse { - repo_count: u64, -} - -#[derive(Deserialize)] -struct RecordCountResponse { - record_count: u64, -} - -#[derive(Deserialize)] -struct OutboxBufferResponse { - outbox_buffer: u64, -} - -/// Fetch aggregate stats from Tap's monitoring endpoints in parallel. -pub async fn get_stats( - http: &reqwest::Client, - tap_url: &str, - tap_admin_password: Option<&str>, -) -> Result { - let (repo, record, outbox) = tokio::try_join!( - tap_get::(http, tap_url, "/stats/repo-count", tap_admin_password), - tap_get::(http, tap_url, "/stats/record-count", tap_admin_password), - tap_get::(http, tap_url, "/stats/outbox-buffer", tap_admin_password), - )?; - - Ok(TapStats { - repo_count: repo.repo_count, - record_count: record.record_count, - outbox_buffer: outbox.outbox_buffer, - }) -} - -/// Sync Tap's collection filters and signal collections with HappyView's -/// current record collections. -pub async fn sync_collections( - http: &reqwest::Client, - tap_url: &str, - tap_admin_password: Option<&str>, - collections: &[String], -) -> Result<(), String> { - let body = serde_json::json!({ "collections": collections }); - tap_put( - http, - tap_url, - "/collection-filters", - tap_admin_password, - &body, - ) - .await?; - tap_put( - http, - tap_url, - "/signal-collections", - tap_admin_password, - &body, - ) - .await?; - Ok(()) -} - -/// Add repos to Tap for backfill via POST /repos/add. -pub async fn add_repos( - http: &reqwest::Client, - tap_url: &str, - tap_admin_password: Option<&str>, - dids: &[String], -) -> Result<(), String> { - let body = serde_json::json!({ "dids": dids }); - tap_post(http, tap_url, "/repos/add", tap_admin_password, &body).await -} - -/// Remove repos from Tap via POST /repos/remove, clearing their cached state -/// so a subsequent add triggers a fresh resync. -pub async fn remove_repos( - http: &reqwest::Client, - tap_url: &str, - tap_admin_password: Option<&str>, - dids: &[String], -) -> Result<(), String> { - let body = serde_json::json!({ "dids": dids }); - tap_post(http, tap_url, "/repos/remove", tap_admin_password, &body).await -} - -// --------------------------------------------------------------------------- -// Public API -// --------------------------------------------------------------------------- - -/// The static collection we always include for lexicon schema updates. -const LEXICON_SCHEMA_COLLECTION: &str = "com.atproto.lexicon.schema"; - -/// Spawn a background task that connects to Tap's WebSocket channel and -/// processes record + identity events. Replaces both jetstream and backfill. -/// -/// When the collection list changes (via `collections_rx`), the task syncs -/// the updated filters to Tap's HTTP API. -pub fn spawn(state: AppState, mut collections_rx: watch::Receiver>) { - tokio::spawn(async move { - loop { - // Build WebSocket URL from HTTP URL. - let ws_url = build_ws_url(&state.config.tap_url); - - match run(&state, &ws_url, &mut collections_rx).await { - Ok(()) => { - tracing::info!("tap reconnecting due to collection change"); - } - Err(e) => { - tracing::warn!("tap disconnected: {e}"); - tokio::time::sleep(std::time::Duration::from_secs(2)).await; - tracing::info!("reconnecting to tap..."); - } - } - } - }); -} - -fn build_ws_url(tap_url: &str) -> String { - let base = tap_url.trim_end_matches('/'); - let ws_base = if let Some(rest) = base.strip_prefix("https://") { - format!("wss://{rest}") - } else if let Some(rest) = base.strip_prefix("http://") { - format!("ws://{rest}") - } else { - format!("ws://{base}") - }; - format!("{ws_base}/channel") -} - -// --------------------------------------------------------------------------- -// Connection loop -// --------------------------------------------------------------------------- - -async fn run( - state: &AppState, - ws_url: &str, - collections_rx: &mut watch::Receiver>, -) -> Result<(), Box> { - let db = &state.db; - let http = &state.http; - let tap_url = &state.config.tap_url; - let tap_admin_password = state.config.tap_admin_password.as_deref(); - - tracing::info!(url = %ws_url, "connecting to tap"); - - let mut request = ws_url.to_string().into_client_request()?; - if let Some(pw) = tap_admin_password { - use base64::Engine; - let encoded = base64::engine::general_purpose::STANDARD.encode(format!("admin:{pw}")); - request - .headers_mut() - .insert("Authorization", format!("Basic {encoded}").parse().unwrap()); - } - - let (ws, _): ( - tokio_tungstenite::WebSocketStream< - tokio_tungstenite::MaybeTlsStream, - >, - _, - ) = tokio_tungstenite::connect_async(request).await?; - tracing::info!("connected to tap"); - - // Re-sync collection filters on every (re)connect so Tap knows which - // collections to track, even if Tap was restarted since the initial sync. - { - let collections = collections_rx.borrow().clone(); - let mut wanted = collections; - if !wanted.contains(&LEXICON_SCHEMA_COLLECTION.to_string()) { - wanted.push(LEXICON_SCHEMA_COLLECTION.to_string()); - } - if let Err(e) = sync_collections(http, tap_url, tap_admin_password, &wanted).await { - tracing::warn!("failed to sync collections to tap on reconnect: {e}"); - } - } - - log_event( - db, - EventLog { - event_type: "tap.connected".to_string(), - severity: Severity::Info, - actor_did: None, - subject: None, - detail: serde_json::json!({ "url": ws_url }), - }, - state.db_backend, - ) - .await; - - let (mut write, mut read) = ws.split(); - - // Allow up to 50 record events to be processed concurrently. - let semaphore = Arc::new(Semaphore::new(50)); - - loop { - tokio::select! { - msg = read.next() => { - let msg = match msg { - Some(Ok(m)) => m, - Some(Err(e)) => { - tracing::warn!("tap websocket read error: {e}"); - return Err(e.into()); - } - None => { - tracing::info!("tap websocket stream ended"); - break; - } - }; - - let text = match msg { - Message::Text(t) => t, - Message::Close(_) => { - tracing::info!("tap websocket received close frame"); - break; - } - other => { - tracing::debug!(msg_type = ?other, "ignoring non-text websocket message"); - continue; - } - }; - - let event: TapEvent = match serde_json::from_str(&text) { - Ok(e) => e, - Err(e) => { - tracing::warn!("skipping unparseable tap event: {e}"); - continue; - } - }; - - let event_id = event.id; - - match event.event_type.as_str() { - "record" => { - if let Some(record) = event.record { - tracing::info!( - id = event_id, - action = %record.action, - collection = %record.collection, - did = %record.did, - rkey = %record.rkey, - "received record event from tap" - ); - let sem = semaphore.clone(); - let state = state.clone(); - tokio::spawn(async move { - let _permit = sem.acquire().await.unwrap(); - handle_record_event(&state, &record).await; - }); - } - } - "identity" => { - if let Some(identity) = event.identity { - tracing::info!( - did = %identity.did, - handle = ?identity.handle, - "received identity event from tap" - ); - } - } - other => { - tracing::warn!(event_type = %other, "unknown tap event type"); - } - } - - // Ack the event. - let ack = serde_json::json!({ "type": "ack", "id": event_id }); - if let Err(e) = write.send(Message::Text(ack.to_string().into())).await { - tracing::warn!("failed to send ack: {e}"); - return Err(e.into()); - } - tracing::debug!(id = event_id, "acked tap event"); - } - // If the collection list changes, sync to Tap and continue. - _ = collections_rx.changed() => { - let collections = collections_rx.borrow_and_update().clone(); - tracing::info!(?collections, "collection filter changed, syncing to tap"); - - // Always include the lexicon schema collection. - let mut wanted = collections; - if !wanted.contains(&LEXICON_SCHEMA_COLLECTION.to_string()) { - wanted.push(LEXICON_SCHEMA_COLLECTION.to_string()); - } - - if let Err(e) = sync_collections(http, tap_url, tap_admin_password, &wanted).await { - tracing::warn!("failed to sync collections to tap: {e}"); - } - } - } - } - - log_event( - db, - EventLog { - event_type: "tap.disconnected".to_string(), - severity: Severity::Warn, - actor_did: None, - subject: None, - detail: serde_json::json!({ "reason": "connection closed" }), - }, - state.db_backend, - ) - .await; - - Ok(()) -} - -// --------------------------------------------------------------------------- -// Record event handler -// --------------------------------------------------------------------------- - -async fn handle_record_event(state: &AppState, record: &TapRecordEvent) { - let db = &state.db; - let lexicons = &state.lexicons; - - let uri = format!("at://{}/{}/{}", record.did, record.collection, record.rkey,); - - // Handle lexicon schema events for tracked network lexicons. - if record.collection == LEXICON_SCHEMA_COLLECTION { - handle_lexicon_schema_event(state, &record.did, record).await; - return; - } - - // Skip records whose collection is not tracked by a registered record-type lexicon. - let is_tracked = lexicons - .get(&record.collection) - .await - .is_some_and(|lex| lex.lexicon_type == LexiconType::Record); - - if !is_tracked { - tracing::debug!( - collection = %record.collection, - "skipping record for untracked collection" - ); - return; - } - - match record.action.as_str() { - "create" | "update" => { - let rec = match &record.record { - Some(r) => r, - None => return, - }; - let cid = record.cid.as_deref().unwrap_or_default(); - - // Run index hook before storing, if configured. The hook's return - // value determines what (if anything) gets written to the DB. - let rec_to_store = - if let Some(script) = state.lexicons.get_index_hook(&record.collection).await { - let hook_result = crate::lua::execute_hook_script(&crate::lua::HookEvent { - state, - lexicon_id: &record.collection, - script: &script, - action: &record.action, - uri: &uri, - did: &record.did, - collection: &record.collection, - rkey: &record.rkey, - record: Some(rec), - }) - .await; - - match hook_result { - None => { - // Hook returned nil — skip indexing this record. - log_event( - db, - EventLog { - event_type: "record.skipped".to_string(), - severity: Severity::Info, - actor_did: None, - subject: Some(uri.clone()), - detail: serde_json::json!({ - "collection": record.collection, - "did": record.did, - "rkey": record.rkey, - "reason": "hook returned nil", - }), - }, - state.db_backend, - ) - .await; - return; - } - Some(v) => v, - } - } else { - // No hook — store the original record as-is. - rec.clone() - }; - - let now = now_rfc3339(); - let backend = state.db_backend; - let insert_sql = adapt_sql( - r#" - INSERT INTO records (uri, did, collection, rkey, record, cid, indexed_at, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?) - ON CONFLICT (uri) DO UPDATE - SET record = EXCLUDED.record, - cid = EXCLUDED.cid, - indexed_at = ? - "#, - backend, - ); - match sqlx::query(&insert_sql) - .bind(&uri) - .bind(&record.did) - .bind(&record.collection) - .bind(&record.rkey) - .bind(serde_json::to_string(&rec_to_store).unwrap_or_default()) - .bind(cid) - .bind(&now) - .bind(&now) - .bind(&now) - .execute(db) - .await - { - Ok(_) => { - let _ = crate::record_refs::sync_refs( - db, - &uri, - &record.collection, - &rec_to_store, - backend, - ) - .await; - - log_event( - db, - EventLog { - event_type: "record.created".to_string(), - severity: Severity::Info, - actor_did: None, - subject: Some(uri.clone()), - detail: serde_json::json!({ - "collection": record.collection, - "did": record.did, - "rkey": record.rkey, - }), - }, - backend, - ) - .await; - - crate::labeler::backfill_labels_for_uri(Arc::new(state.clone()), uri.clone()); - } - Err(e) => { - tracing::warn!(uri = %uri, "failed to upsert record: {e}"); - log_event( - db, - EventLog { - event_type: "record.created".to_string(), - severity: Severity::Error, - actor_did: None, - subject: Some(uri.clone()), - detail: serde_json::json!({ - "collection": record.collection, - "did": record.did, - "rkey": record.rkey, - "error": e.to_string(), - }), - }, - backend, - ) - .await; - } - } - } - "delete" => { - let backend = state.db_backend; - - // Run index hook before deleting, if configured. - if let Some(script) = state.lexicons.get_index_hook(&record.collection).await { - let hook_result = crate::lua::execute_hook_script(&crate::lua::HookEvent { - state, - lexicon_id: &record.collection, - script: &script, - action: "delete", - uri: &uri, - did: &record.did, - collection: &record.collection, - rkey: &record.rkey, - record: None, - }) - .await; - - if hook_result.is_none() { - // Hook returned nil — skip the delete. - log_event( - db, - EventLog { - event_type: "record.skipped".to_string(), - severity: Severity::Info, - actor_did: None, - subject: Some(uri.clone()), - detail: serde_json::json!({ - "collection": record.collection, - "did": record.did, - "rkey": record.rkey, - "reason": "hook returned nil", - }), - }, - backend, - ) - .await; - return; - } - } - - let delete_sql = adapt_sql("DELETE FROM records WHERE uri = ?", backend); - match sqlx::query(&delete_sql).bind(&uri).execute(db).await { - Ok(_) => { - log_event( - db, - EventLog { - event_type: "record.deleted".to_string(), - severity: Severity::Info, - actor_did: None, - subject: Some(uri.clone()), - detail: serde_json::json!({ - "collection": record.collection, - "did": record.did, - "rkey": record.rkey, - }), - }, - backend, - ) - .await; - } - Err(e) => { - tracing::warn!(uri = %uri, "failed to delete record: {e}"); - log_event( - db, - EventLog { - event_type: "record.deleted".to_string(), - severity: Severity::Error, - actor_did: None, - subject: Some(uri.clone()), - detail: serde_json::json!({ - "collection": record.collection, - "did": record.did, - "rkey": record.rkey, - "error": e.to_string(), - }), - }, - backend, - ) - .await; - } - } - } - _ => {} - } -} - -// --------------------------------------------------------------------------- -// Lexicon schema event handler -// --------------------------------------------------------------------------- - -/// Handle a `com.atproto.lexicon.schema` record event for tracked network lexicons. -async fn handle_lexicon_schema_event(state: &AppState, did: &str, record: &TapRecordEvent) { - let db = &state.db; - let lexicons = &state.lexicons; - let collections_tx = &state.collections_tx; - let nsid = &record.rkey; - - let backend = state.db_backend; - - // Check if this NSID is one we're tracking and the DID matches the authority. - let select_sql = adapt_sql( - "SELECT target_collection FROM lexicons WHERE id = ? AND source = 'network' AND authority_did = ?", - backend, - ); - let tracked: Option<(Option,)> = sqlx::query_as(&select_sql) - .bind(nsid) - .bind(did) - .fetch_optional(db) - .await - .unwrap_or(None); - - let target_collection = match tracked { - Some((tc,)) => tc, - None => return, // Not a tracked network lexicon. - }; - - match record.action.as_str() { - "create" | "update" => { - let rec = match &record.record { - Some(r) => r, - None => return, - }; - - let parsed = match ParsedLexicon::parse( - rec.clone(), - 1, - target_collection.clone(), - ProcedureAction::Upsert, - None, - None, - None, - ) { - Ok(p) => p, - Err(e) => { - tracing::warn!(nsid, "failed to parse lexicon schema event: {e}"); - return; - } - }; - - let is_record = parsed.lexicon_type == crate::lexicon::LexiconType::Record; - - // Upsert into lexicons table with last_fetched_at. - let now = now_rfc3339(); - let upsert_sql = adapt_sql( - r#" - INSERT INTO lexicons (id, lexicon_json, backfill, target_collection, source, authority_did, last_fetched_at, created_at) - VALUES (?, ?, 0, ?, 'network', ?, ?, ?) - ON CONFLICT (id) DO UPDATE SET - lexicon_json = EXCLUDED.lexicon_json, - target_collection = EXCLUDED.target_collection, - last_fetched_at = ?, - revision = lexicons.revision + 1, - updated_at = ? - "#, - backend, - ); - if let Err(e) = sqlx::query(&upsert_sql) - .bind(nsid) - .bind(serde_json::to_string(rec).unwrap_or_default()) - .bind(&target_collection) - .bind(did) - .bind(&now) - .bind(&now) - .bind(&now) - .bind(&now) - .execute(db) - .await - { - tracing::warn!(nsid, "failed to upsert lexicon from event: {e}"); - return; - } - - lexicons.upsert(parsed).await; - tracing::info!(nsid, "updated network lexicon from tap event"); - - if is_record { - let collections = lexicons.get_record_collections().await; - let _ = collections_tx.send(collections); - } - } - "delete" => { - // Remove from lexicons table and registry. - let delete_sql = adapt_sql("DELETE FROM lexicons WHERE id = ?", backend); - let _ = sqlx::query(&delete_sql).bind(nsid).execute(db).await; - - let was_present = lexicons.remove(nsid).await; - if was_present { - tracing::info!(nsid, "removed network lexicon from tap delete event"); - let collections = lexicons.get_record_collections().await; - let _ = collections_tx.send(collections); - } - } - _ => {} - } -} diff --git a/tests/common/app.rs b/tests/common/app.rs --- a/tests/common/app.rs +++ b/tests/common/app.rs @@ -41,8 +41,7 @@ database_url: String::new(), database_backend: backend, public_url: "http://127.0.0.1:0".into(), session_secret: "test-secret".into(), - tap_url: "http://localhost:2480".into(), - tap_admin_password: None, + jetstream_url: "wss://jetstream1.us-east.bsky.network".into(), relay_url: mock_url.clone(), plc_url: mock_url.clone(), static_dir: "./web/out".into(), diff --git a/tests/lua_atproto_api.rs b/tests/lua_atproto_api.rs --- a/tests/lua_atproto_api.rs +++ b/tests/lua_atproto_api.rs @@ -23,8 +23,7 @@ database_url: String::new(), database_backend: backend, public_url: String::new(), session_secret: "test-secret".into(), - tap_url: String::new(), - tap_admin_password: None, + jetstream_url: String::new(), relay_url: String::new(), plc_url: String::new(), static_dir: String::new(), diff --git a/tests/lua_db_api.rs b/tests/lua_db_api.rs --- a/tests/lua_db_api.rs +++ b/tests/lua_db_api.rs @@ -26,8 +26,7 @@ database_url: String::new(), database_backend: backend, public_url: String::new(), session_secret: "test-secret".into(), - tap_url: String::new(), - tap_admin_password: None, + jetstream_url: String::new(), relay_url: String::new(), plc_url: String::new(), static_dir: String::new(), -- tangled.sh