From dbd22337e3216e4cbe0f44172ed0c6363dc208ff Mon Sep 17 00:00:00 2001 From: Trezy Date: Mon, 16 Feb 2026 00:01:47 -0600 Subject: [PATCH] feat: replace Jetstream and backfill with Tap --- docker-compose.yml | 23 +- docker/init-databases.sh | 2 + src/admin/backfill.rs | 191 +++++++++++++- src/admin/lexicons.rs | 10 +- src/admin/network_lexicons.rs | 10 +- src/backfill.rs | 373 -------------------------- src/config.rs | 16 +- src/jetstream.rs | 347 ------------------------- src/lib.rs | 3 +- src/main.rs | 32 ++- src/tap.rs | 475 ++++++++++++++++++++++++++++++++++ tests/common/app.rs | 3 +- 12 files changed, 732 insertions(+), 753 deletions(-) delete mode 100644 src/backfill.rs delete mode 100644 src/jetstream.rs create mode 100644 src/tap.rs diff --git a/docker-compose.yml b/docker-compose.yml index 96ecb6f..6f63801 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -33,6 +33,21 @@ services: postgres: condition: service_healthy + tap: + image: ghcr.io/bluesky-social/indigo/tap:latest + ports: + - "2480:2480" + environment: + TAP_DATABASE_URL: postgres://tap:tap@postgres/tap + TAP_RELAY_URL: https://bsky.network + TAP_PLC_URL: https://plc.directory + TAP_ADMIN_PASSWORD: ${TAP_ADMIN_PASSWORD} + TAP_COLLECTION_FILTERS: "" + TAP_SIGNAL_COLLECTIONS: "" + depends_on: + postgres: + condition: service_healthy + happyview: image: rust:1.93 working_dir: /app @@ -47,12 +62,16 @@ services: environment: DATABASE_URL: postgres://happyview:happyview@postgres/happyview AIP_URL: https://aip.gamesgamesgamesgames.games + TAP_URL: http://tap:2480 + TAP_ADMIN_PASSWORD: ${TAP_ADMIN_PASSWORD} PORT: 3000 depends_on: postgres: condition: service_healthy aip: condition: service_started + tap: + condition: service_started web: image: node:24-alpine @@ -66,8 +85,8 @@ services: environment: - HOSTNAME=0.0.0.0 - API_URL=http://happyview:3000 - - AIP_PROXY_URL=https://aip.gamesgamesgamesgames.games - - NEXT_PUBLIC_AIP_URL=https://aip.gamesgamesgamesgames.games + - AIP_PROXY_URL=http://aip:8080 + - NEXT_PUBLIC_AIP_URL=http://localhost:8080 volumes: pgdata: diff --git a/docker/init-databases.sh b/docker/init-databases.sh index 009ecba..8864b2d 100644 --- a/docker/init-databases.sh +++ b/docker/init-databases.sh @@ -4,4 +4,6 @@ set -e psql -v ON_ERROR_STOP=1 --username "$POSTGRES_USER" --dbname "$POSTGRES_DB" <<-EOSQL CREATE USER aip WITH PASSWORD 'aip'; CREATE DATABASE aip OWNER aip; + CREATE USER tap WITH PASSWORD 'tap'; + CREATE DATABASE tap OWNER tap; EOSQL diff --git a/src/admin/backfill.rs b/src/admin/backfill.rs index 01d4073..e1c495e 100644 --- a/src/admin/backfill.rs +++ b/src/admin/backfill.rs @@ -1,20 +1,90 @@ use axum::Json; use axum::extract::State; use axum::http::StatusCode; +use serde::Deserialize; use serde_json::Value; use crate::AppState; use crate::error::AppError; +use crate::tap; use super::auth::AdminAuth; use super::types::{BackfillJob, CreateBackfillBody}; -/// POST /admin/backfill — create a new backfill job. +// --------------------------------------------------------------------------- +// Relay discovery (reused from old backfill module) +// --------------------------------------------------------------------------- + +#[derive(Deserialize)] +struct ListReposResponse { + repos: Vec, + cursor: Option, +} + +#[derive(Deserialize)] +struct RepoEntry { + did: String, +} + +/// Discover all DIDs that have records in `collection` via the relay's +/// `com.atproto.sync.listReposByCollection` endpoint. Paginates until done. +async fn list_repos_by_collection( + http: &reqwest::Client, + relay_url: &str, + collection: &str, +) -> Result, String> { + let base = relay_url.trim_end_matches('/'); + let mut dids = Vec::new(); + let mut cursor: Option = None; + + loop { + let mut url = format!( + "{base}/xrpc/com.atproto.sync.listReposByCollection?collection={collection}&limit=1000" + ); + if let Some(ref c) = cursor { + url.push_str(&format!("&cursor={c}")); + } + + let resp = http + .get(&url) + .send() + .await + .map_err(|e| format!("relay request failed: {e}"))?; + + if !resp.status().is_success() { + return Err(format!("relay returned {}", resp.status())); + } + + let body: ListReposResponse = resp + .json() + .await + .map_err(|e| format!("invalid relay response: {e}"))?; + + let page_count = body.repos.len(); + for repo in body.repos { + dids.push(repo.did); + } + + match body.cursor { + Some(c) if page_count > 0 => cursor = Some(c), + _ => break, + } + } + + Ok(dids) +} + +// --------------------------------------------------------------------------- +// Admin handlers +// --------------------------------------------------------------------------- + +/// POST /admin/backfill — create a backfill job, discover repos, and add them to Tap. pub(super) async fn create_backfill( State(state): State, _admin: AdminAuth, Json(body): Json, ) -> Result<(StatusCode, Json), AppError> { + // Create a backfill_jobs record for tracking/audit. let row: (String,) = sqlx::query_as( "INSERT INTO backfill_jobs (collection, did) VALUES ($1, $2) RETURNING id::text", ) @@ -24,11 +94,126 @@ pub(super) async fn create_backfill( .await .map_err(|e| AppError::Internal(format!("failed to create backfill job: {e}")))?; + let job_id = row.0.clone(); + + // Mark as running. + let _ = sqlx::query( + "UPDATE backfill_jobs SET status = 'running', started_at = NOW() WHERE id::text = $1", + ) + .bind(&job_id) + .execute(&state.db) + .await; + + // Determine target collections. + let collections: Vec = if let Some(ref col) = body.collection { + vec![col.clone()] + } else { + let rows: Vec<(String,)> = sqlx::query_as( + "SELECT id FROM lexicons WHERE backfill = TRUE AND lexicon_json->'defs'->'main'->>'type' = 'record'", + ) + .fetch_all(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to query backfill-eligible lexicons: {e}")))?; + rows.into_iter().map(|(id,)| id).collect() + }; + + if collections.is_empty() { + let _ = sqlx::query( + "UPDATE backfill_jobs SET status = 'completed', completed_at = NOW(), error = 'no backfill-eligible collections' WHERE id::text = $1", + ) + .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", + })), + )); + } + + // Discover repos and add them to Tap. + let mut all_dids = Vec::new(); + + for collection in &collections { + let dids = if let Some(ref did) = body.did { + vec![did.clone()] + } else { + match list_repos_by_collection(&state.http, &state.config.relay_url, collection).await { + Ok(dids) => dids, + Err(e) => { + tracing::warn!(collection, error = %e, "failed to discover repos, skipping"); + continue; + } + } + }; + + all_dids.extend(dids); + } + + // Deduplicate DIDs. + all_dids.sort(); + all_dids.dedup(); + + let total_repos = all_dids.len() as i32; + + // Update job with total repos. + let _ = sqlx::query("UPDATE backfill_jobs SET total_repos = $2 WHERE id::text = $1") + .bind(&job_id) + .bind(total_repos) + .execute(&state.db) + .await; + + // Add repos to Tap in batches. + 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 _ = sqlx::query( + "UPDATE backfill_jobs SET status = 'failed', completed_at = NOW(), error = $2 WHERE id::text = $1", + ) + .bind(&job_id) + .bind(&e) + .execute(&state.db) + .await; + + return Ok(( + StatusCode::CREATED, + Json(serde_json::json!({ + "id": job_id, + "status": "failed", + "error": e, + })), + )); + } + } + } + + // Mark as completed (Tap handles the actual backfill asynchronously). + let _ = sqlx::query( + "UPDATE backfill_jobs SET status = 'completed', completed_at = NOW(), processed_repos = $2 WHERE id::text = $1", + ) + .bind(&job_id) + .bind(total_repos) + .execute(&state.db) + .await; + Ok(( StatusCode::CREATED, Json(serde_json::json!({ - "id": row.0, - "status": "pending", + "id": job_id, + "status": "completed", + "total_repos": total_repos, })), )) } diff --git a/src/admin/lexicons.rs b/src/admin/lexicons.rs index 7de5cde..c7f28e5 100644 --- a/src/admin/lexicons.rs +++ b/src/admin/lexicons.rs @@ -10,9 +10,9 @@ use crate::lexicon::{LexiconType, ParsedLexicon, ProcedureAction}; use super::auth::AdminAuth; use super::types::{LexiconSummary, UploadLexiconBody}; -/// Send the current record collection list to the Jetstream task so it -/// reconnects with the updated filter. -async fn notify_jetstream(state: &AppState) { +/// Send the current record collection list to the Tap task so it +/// syncs the updated filter. +async fn notify_collections(state: &AppState) { let collections = state.lexicons.get_record_collections().await; let _ = state.collections_tx.send(collections); } @@ -93,7 +93,7 @@ pub(super) async fn upload_lexicon( state.lexicons.upsert(parsed).await; if is_record { - notify_jetstream(&state).await; + notify_collections(&state).await; } let status = if revision == 1 { @@ -197,7 +197,7 @@ pub(super) async fn delete_lexicon( } state.lexicons.remove(&id).await; - notify_jetstream(&state).await; + notify_collections(&state).await; Ok(StatusCode::NO_CONTENT) } diff --git a/src/admin/network_lexicons.rs b/src/admin/network_lexicons.rs index e26547c..64a9f5b 100644 --- a/src/admin/network_lexicons.rs +++ b/src/admin/network_lexicons.rs @@ -11,9 +11,9 @@ use crate::resolve::{fetch_lexicon_from_pds, resolve_nsid_authority}; use super::auth::AdminAuth; use super::types::{AddNetworkLexiconBody, NetworkLexiconSummary}; -/// Send the current record collection list to the Jetstream task so it -/// reconnects with the updated filter. -async fn notify_jetstream(state: &AppState) { +/// Send the current record collection list to the Tap task so it +/// syncs the updated filter. +async fn notify_collections(state: &AppState) { let collections = state.lexicons.get_record_collections().await; let _ = state.collections_tx.send(collections); } @@ -95,7 +95,7 @@ pub(super) async fn add( state.lexicons.upsert(parsed).await; if is_record { - notify_jetstream(&state).await; + notify_collections(&state).await; } Ok(( @@ -165,7 +165,7 @@ pub(super) async fn remove( .await; state.lexicons.remove(&nsid).await; - notify_jetstream(&state).await; + notify_collections(&state).await; Ok(StatusCode::NO_CONTENT) } diff --git a/src/backfill.rs b/src/backfill.rs deleted file mode 100644 index db2b79e..0000000 --- a/src/backfill.rs +++ /dev/null @@ -1,373 +0,0 @@ -use serde::Deserialize; -use serde_json::Value; -use sqlx::PgPool; -use std::sync::Arc; -use tokio::sync::Semaphore; -use tracing::{debug, error, info, warn}; - -use crate::profile; - -// --------------------------------------------------------------------------- -// Relay / PDS response types -// --------------------------------------------------------------------------- - -#[derive(Deserialize)] -struct ListReposResponse { - repos: Vec, - cursor: Option, -} - -#[derive(Deserialize)] -struct RepoEntry { - did: String, -} - -#[derive(Deserialize)] -struct ListRecordsResponse { - records: Vec, - cursor: Option, -} - -#[derive(Deserialize)] -struct RecordEntry { - uri: String, - cid: String, - value: Value, -} - -// --------------------------------------------------------------------------- -// Relay discovery -// --------------------------------------------------------------------------- - -/// Discover all DIDs that have records in `collection` via the relay's -/// `com.atproto.sync.listReposByCollection` endpoint. Paginates until done. -async fn list_repos_by_collection( - http: &reqwest::Client, - relay_url: &str, - collection: &str, -) -> Result, String> { - let base = relay_url.trim_end_matches('/'); - let mut dids = Vec::new(); - let mut cursor: Option = None; - - loop { - let mut url = format!( - "{base}/xrpc/com.atproto.sync.listReposByCollection?collection={collection}&limit=1000" - ); - if let Some(ref c) = cursor { - url.push_str(&format!("&cursor={c}")); - } - - let resp = http - .get(&url) - .send() - .await - .map_err(|e| format!("relay request failed: {e}"))?; - - if !resp.status().is_success() { - return Err(format!("relay returned {}", resp.status())); - } - - let body: ListReposResponse = resp - .json() - .await - .map_err(|e| format!("invalid relay response: {e}"))?; - - let page_count = body.repos.len(); - for repo in body.repos { - dids.push(repo.did); - } - - match body.cursor { - Some(c) if page_count > 0 => cursor = Some(c), - _ => break, - } - } - - Ok(dids) -} - -// --------------------------------------------------------------------------- -// PDS record fetching -// --------------------------------------------------------------------------- - -/// Fetch all records for a DID + collection from their PDS via -/// `com.atproto.repo.listRecords`. Paginates until done. -async fn fetch_records( - http: &reqwest::Client, - pds_url: &str, - did: &str, - collection: &str, -) -> Result, String> { - let base = pds_url.trim_end_matches('/'); - let mut records = Vec::new(); - let mut cursor: Option = None; - - 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 = http - .get(&url) - .send() - .await - .map_err(|e| format!("PDS listRecords failed: {e}"))?; - - if !resp.status().is_success() { - return Err(format!("PDS returned {} for {did}", resp.status())); - } - - let body: ListRecordsResponse = resp - .json() - .await - .map_err(|e| format!("invalid PDS listRecords response: {e}"))?; - - let page_count = body.records.len(); - for entry in body.records { - let rkey = entry - .uri - .split('/') - .next_back() - .unwrap_or_default() - .to_string(); - records.push((entry.uri, rkey, entry.cid, entry.value)); - } - - match body.cursor { - Some(c) if page_count > 0 => cursor = Some(c), - _ => break, - } - } - - Ok(records) -} - -// --------------------------------------------------------------------------- -// Job runner -// --------------------------------------------------------------------------- - -/// Run a single backfill job: discover repos, fetch records, upsert into DB. -async fn run_job( - db: &PgPool, - http: &reqwest::Client, - relay_url: &str, - plc_url: &str, - job_id: &str, -) -> Result<(), String> { - // Fetch the job - let job: (Option, Option) = - sqlx::query_as("SELECT collection, did FROM backfill_jobs WHERE id::text = $1") - .bind(job_id) - .fetch_one(db) - .await - .map_err(|e| format!("failed to fetch job: {e}"))?; - - let (job_collection, job_did) = job; - - // Mark as running - let _ = sqlx::query( - "UPDATE backfill_jobs SET status = 'running', started_at = NOW() WHERE id::text = $1", - ) - .bind(job_id) - .execute(db) - .await; - - // Determine target collections - let collections: Vec = if let Some(ref col) = job_collection { - vec![col.clone()] - } else { - // All backfill-eligible collections - let rows: Vec<(String,)> = sqlx::query_as( - "SELECT id FROM lexicons WHERE backfill = TRUE AND lexicon_json->'defs'->'main'->>'type' = 'record'", - ) - .fetch_all(db) - .await - .map_err(|e| format!("failed to query backfill-eligible lexicons: {e}"))?; - rows.into_iter().map(|(id,)| id).collect() - }; - - if collections.is_empty() { - let _ = sqlx::query( - "UPDATE backfill_jobs SET status = 'completed', completed_at = NOW(), error = 'no backfill-eligible collections' WHERE id::text = $1", - ) - .bind(job_id) - .execute(db) - .await; - return Ok(()); - } - - info!(job = job_id, ?collections, "starting backfill"); - - let semaphore = Arc::new(Semaphore::new(8)); - let mut total_repos = 0i32; - let mut processed_repos = 0i32; - let mut total_records = 0i32; - - for collection in &collections { - // Discover DIDs - let dids = if let Some(ref did) = job_did { - vec![did.clone()] - } else { - match list_repos_by_collection(http, relay_url, collection).await { - Ok(dids) => dids, - Err(e) => { - warn!(collection, error = %e, "failed to discover repos, skipping"); - continue; - } - } - }; - - total_repos += dids.len() as i32; - let _ = sqlx::query("UPDATE backfill_jobs SET total_repos = $2 WHERE id::text = $1") - .bind(job_id) - .bind(total_repos) - .execute(db) - .await; - - // Process each DID concurrently (bounded by semaphore) - let mut tasks = Vec::new(); - - for did in dids { - let permit = semaphore.clone().acquire_owned().await.unwrap(); - let http = http.clone(); - let db = db.clone(); - let collection = collection.clone(); - - let plc_url = plc_url.to_string(); - let task = tokio::spawn(async move { - let _permit = permit; - backfill_repo(&db, &http, &plc_url, &did, &collection).await - }); - tasks.push(task); - } - - for task in tasks { - match task.await { - Ok(Ok(count)) => { - total_records += count; - processed_repos += 1; - } - Ok(Err(e)) => { - warn!(error = %e, "repo backfill failed"); - processed_repos += 1; - } - Err(e) => { - warn!(error = %e, "repo backfill task panicked"); - processed_repos += 1; - } - } - - // Update progress periodically - let _ = sqlx::query( - "UPDATE backfill_jobs SET processed_repos = $2, total_records = $3 WHERE id::text = $1", - ) - .bind(job_id) - .bind(processed_repos) - .bind(total_records) - .execute(db) - .await; - } - } - - // Mark completed - let _ = sqlx::query( - "UPDATE backfill_jobs SET status = 'completed', completed_at = NOW(), processed_repos = $2, total_records = $3 WHERE id::text = $1", - ) - .bind(job_id) - .bind(processed_repos) - .bind(total_records) - .execute(db) - .await; - - info!( - job = job_id, - processed_repos, total_records, "backfill completed" - ); - Ok(()) -} - -/// Backfill a single repo's records for a collection. Returns the number of -/// records upserted. -async fn backfill_repo( - db: &PgPool, - http: &reqwest::Client, - plc_url: &str, - did: &str, - collection: &str, -) -> Result { - // Resolve PDS - let pds = profile::resolve_pds_endpoint(http, plc_url, did) - .await - .map_err(|e| format!("PDS resolution failed for {did}: {e}"))?; - - // Fetch records - let records = fetch_records(http, &pds, did, collection).await?; - let count = records.len() as i32; - - debug!(did, collection, count, "fetched records from PDS"); - - // Upsert into DB - for (uri, rkey, cid, value) in records { - let _ = sqlx::query( - r#" - INSERT INTO records (uri, did, collection, rkey, record, cid) - VALUES ($1, $2, $3, $4, $5, $6) - ON CONFLICT (uri) DO UPDATE - SET record = EXCLUDED.record, - cid = EXCLUDED.cid - "#, - ) - .bind(&uri) - .bind(did) - .bind(collection) - .bind(&rkey) - .bind(&value) - .bind(&cid) - .execute(db) - .await - .map_err(|e| format!("DB upsert failed for {uri}: {e}"))?; - } - - Ok(count) -} - -// --------------------------------------------------------------------------- -// Background worker -// --------------------------------------------------------------------------- - -/// Spawn a background task that polls for pending backfill jobs and runs them. -pub fn spawn_worker(db: PgPool, http: reqwest::Client, relay_url: String, plc_url: String) { - tokio::spawn(async move { - info!("backfill worker started"); - loop { - // Poll for a pending job - let job: Option<(String,)> = sqlx::query_as( - "SELECT id::text FROM backfill_jobs WHERE status = 'pending' ORDER BY created_at ASC LIMIT 1", - ) - .fetch_optional(&db) - .await - .unwrap_or(None); - - if let Some((job_id,)) = job { - info!(job = %job_id, "picked up backfill job"); - if let Err(e) = run_job(&db, &http, &relay_url, &plc_url, &job_id).await { - error!(job = %job_id, error = %e, "backfill job failed"); - let _ = sqlx::query( - "UPDATE backfill_jobs SET status = 'failed', completed_at = NOW(), error = $2 WHERE id::text = $1", - ) - .bind(&job_id) - .bind(&e) - .execute(&db) - .await; - } - } else { - // No pending jobs, wait before polling again - tokio::time::sleep(std::time::Duration::from_secs(5)).await; - } - } - }); -} diff --git a/src/config.rs b/src/config.rs index afa11a8..c26cae0 100644 --- a/src/config.rs +++ b/src/config.rs @@ -7,7 +7,8 @@ pub struct Config { pub port: u16, pub database_url: String, pub aip_url: String, - pub jetstream_url: String, + pub tap_url: String, + pub tap_admin_password: Option, pub relay_url: String, pub plc_url: String, pub static_dir: String, @@ -23,8 +24,8 @@ impl Config { .unwrap_or(3000), database_url: env::var("DATABASE_URL").expect("DATABASE_URL must be set"), aip_url: env::var("AIP_URL").expect("AIP_URL must be set"), - jetstream_url: env::var("JETSTREAM_URL") - .unwrap_or_else(|_| "wss://jetstream2.us-west.bsky.network/subscribe".into()), + tap_url: env::var("TAP_URL").unwrap_or_else(|_| "http://localhost:2480".into()), + tap_admin_password: env::var("TAP_ADMIN_PASSWORD").ok(), 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()), @@ -49,7 +50,8 @@ mod tests { "PORT", "DATABASE_URL", "AIP_URL", - "JETSTREAM_URL", + "TAP_URL", + "TAP_ADMIN_PASSWORD", "RELAY_URL", "PLC_URL", ] { @@ -73,7 +75,8 @@ mod tests { port: 8080, database_url: String::new(), aip_url: String::new(), - jetstream_url: String::new(), + tap_url: String::new(), + tap_admin_password: None, relay_url: String::new(), plc_url: String::new(), static_dir: String::new(), @@ -106,7 +109,8 @@ mod tests { let config = Config::from_env(); assert_eq!(config.host, "0.0.0.0"); assert_eq!(config.port, 3000); - assert!(config.jetstream_url.contains("jetstream")); + assert_eq!(config.tap_url, "http://localhost:2480"); + assert!(config.tap_admin_password.is_none()); 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 deleted file mode 100644 index 49a2987..0000000 --- a/src/jetstream.rs +++ /dev/null @@ -1,347 +0,0 @@ -use futures_util::StreamExt; -use serde::Deserialize; -use serde_json::Value; -use sqlx::PgPool; -use std::sync::Arc; -use std::sync::atomic::{AtomicI64, Ordering}; -use tokio::sync::watch; -use tokio_tungstenite::tungstenite::Message; - -use crate::lexicon::{LexiconRegistry, ParsedLexicon, ProcedureAction}; - -// --------------------------------------------------------------------------- -// Jetstream event types -// --------------------------------------------------------------------------- - -#[derive(Deserialize)] -struct JetstreamEvent { - did: String, - time_us: i64, - kind: String, - commit: Option, -} - -#[derive(Deserialize)] -struct JetstreamCommit { - operation: String, - collection: String, - rkey: String, - record: Option, - cid: Option, -} - -// --------------------------------------------------------------------------- -// Public API -// --------------------------------------------------------------------------- - -/// The static collection we always watch for lexicon schema updates. -const LEXICON_SCHEMA_COLLECTION: &str = "com.atproto.lexicon.schema"; - -/// Spawn a background task that subscribes to the Jetstream firehose and -/// indexes records for collections specified by the watch channel. -/// -/// When the collection list is empty, the task idles without connecting. -/// When collections change, it disconnects and reconnects with the new filter. -pub fn spawn( - db: PgPool, - jetstream_url: String, - mut collections_rx: watch::Receiver>, - lexicons: LexiconRegistry, - collections_tx: watch::Sender>, -) { - tokio::spawn(async move { - let cursor: Arc = Arc::new(AtomicI64::new(0)); - - loop { - // Wait until we have at least one collection to subscribe to. - let collections = collections_rx.borrow_and_update().clone(); - if collections.is_empty() { - tracing::info!("no collections configured, jetstream idle"); - // Block until the collection list changes. - if collections_rx.changed().await.is_err() { - // Sender dropped — shut down. - tracing::info!("jetstream watch channel closed, shutting down"); - return; - } - continue; - } - - // Always include the lexicon schema collection alongside the dynamic ones. - let mut wanted = collections.clone(); - if !wanted.contains(&LEXICON_SCHEMA_COLLECTION.to_string()) { - wanted.push(LEXICON_SCHEMA_COLLECTION.to_string()); - } - - // Connect and process events. If the collection list changes - // mid-stream, `run` returns so we can reconnect with new filters. - match run( - &db, - &jetstream_url, - &cursor, - &wanted, - &mut collections_rx, - &lexicons, - &collections_tx, - ) - .await - { - Ok(()) => { - tracing::info!("jetstream reconnecting due to collection change"); - } - Err(e) => { - tracing::warn!("jetstream disconnected: {e}"); - tokio::time::sleep(std::time::Duration::from_secs(2)).await; - tracing::info!("reconnecting to jetstream..."); - } - } - } - }); -} - -// --------------------------------------------------------------------------- -// Connection loop -// --------------------------------------------------------------------------- - -async fn run( - db: &PgPool, - jetstream_url: &str, - cursor: &Arc, - collections: &[String], - collections_rx: &mut watch::Receiver>, - lexicons: &LexiconRegistry, - collections_tx: &watch::Sender>, -) -> Result<(), Box> { - let wanted: String = collections - .iter() - .map(|c| format!("wantedCollections={c}")) - .collect::>() - .join("&"); - - let mut url = format!("{jetstream_url}?{wanted}"); - - let last = cursor.load(Ordering::Relaxed); - if last > 0 { - // Rewind 5 seconds for gapless playback. - let rewound = last - 5_000_000; - url.push_str(&format!("&cursor={rewound}")); - tracing::info!(cursor = rewound, "resuming jetstream with cursor"); - } - - tracing::info!(collections = ?collections, "connecting to jetstream"); - - let (ws, _) = tokio_tungstenite::connect_async(&url).await?; - tracing::info!("connected to jetstream"); - - let (_, mut read) = ws.split(); - - loop { - tokio::select! { - msg = read.next() => { - let msg = match msg { - Some(Ok(m)) => m, - Some(Err(e)) => return Err(e.into()), - None => break, - }; - - let text = match msg { - Message::Text(t) => t, - Message::Close(_) => break, - _ => continue, - }; - - let event: JetstreamEvent = match serde_json::from_str(&text) { - Ok(e) => e, - Err(e) => { - tracing::debug!("skipping unparseable event: {e}"); - continue; - } - }; - - // Update cursor. - cursor.store(event.time_us, Ordering::Relaxed); - - if event.kind != "commit" { - continue; - } - - let commit = match event.commit { - Some(c) => c, - None => continue, - }; - - let uri = format!( - "at://{}/{}/{}", - event.did, commit.collection, commit.rkey, - ); - - // Handle lexicon schema events for tracked network lexicons. - if commit.collection == LEXICON_SCHEMA_COLLECTION { - handle_lexicon_schema_event( - db, - lexicons, - collections_tx, - &event.did, - &commit, - ) - .await; - continue; - } - - match commit.operation.as_str() { - "create" | "update" => { - let record = match commit.record { - Some(r) => r, - None => continue, - }; - let cid = commit.cid.unwrap_or_default(); - - if let Err(e) = sqlx::query( - r#" - INSERT INTO records (uri, did, collection, rkey, record, cid, indexed_at) - VALUES ($1, $2, $3, $4, $5, $6, NOW()) - ON CONFLICT (uri) DO UPDATE - SET record = EXCLUDED.record, - cid = EXCLUDED.cid, - indexed_at = NOW() - "#, - ) - .bind(&uri) - .bind(&event.did) - .bind(&commit.collection) - .bind(&commit.rkey) - .bind(&record) - .bind(&cid) - .execute(db) - .await - { - tracing::warn!(uri = %uri, "failed to upsert record: {e}"); - } - } - "delete" => { - if let Err(e) = sqlx::query("DELETE FROM records WHERE uri = $1") - .bind(&uri) - .execute(db) - .await - { - tracing::warn!(uri = %uri, "failed to delete record: {e}"); - } - } - _ => {} - } - } - // If the collection list changes, break out to reconnect. - _ = collections_rx.changed() => { - tracing::info!("collection filter changed, will reconnect"); - return Ok(()); - } - } - } - - Ok(()) -} - -// --------------------------------------------------------------------------- -// Lexicon schema event handler -// --------------------------------------------------------------------------- - -/// Handle a `com.atproto.lexicon.schema` commit event for tracked network lexicons. -async fn handle_lexicon_schema_event( - db: &PgPool, - lexicons: &LexiconRegistry, - collections_tx: &watch::Sender>, - did: &str, - commit: &JetstreamCommit, -) { - let nsid = &commit.rkey; - - // Check if this NSID is one we're tracking and the DID matches the authority. - let tracked: Option<(Option,)> = sqlx::query_as( - "SELECT target_collection FROM network_lexicons WHERE nsid = $1 AND authority_did = $2", - ) - .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 commit.operation.as_str() { - "create" | "update" => { - let record = match &commit.record { - Some(r) => r, - None => return, - }; - - let parsed = match ParsedLexicon::parse( - record.clone(), - 1, - target_collection.clone(), - ProcedureAction::Upsert, - ) { - 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. - if let Err(e) = sqlx::query( - r#" - INSERT INTO lexicons (id, lexicon_json, backfill, target_collection) - VALUES ($1, $2, false, $3) - ON CONFLICT (id) DO UPDATE SET - lexicon_json = EXCLUDED.lexicon_json, - target_collection = EXCLUDED.target_collection, - revision = lexicons.revision + 1, - updated_at = NOW() - "#, - ) - .bind(nsid) - .bind(record) - .bind(&target_collection) - .execute(db) - .await - { - tracing::warn!(nsid, "failed to upsert lexicon from event: {e}"); - return; - } - - // Update last_fetched_at. - let _ = - sqlx::query("UPDATE network_lexicons SET last_fetched_at = NOW() WHERE nsid = $1") - .bind(nsid) - .execute(db) - .await; - - lexicons.upsert(parsed).await; - tracing::info!(nsid, "updated network lexicon from jetstream event"); - - if is_record { - let collections = lexicons.get_record_collections().await; - let _ = collections_tx.send(collections); - } - } - "delete" => { - // Remove from lexicons table and registry. - let _ = sqlx::query("DELETE FROM lexicons WHERE id = $1") - .bind(nsid) - .execute(db) - .await; - - let was_present = lexicons.remove(nsid).await; - if was_present { - tracing::info!(nsid, "removed network lexicon from jetstream delete event"); - let collections = lexicons.get_record_collections().await; - let _ = collections_tx.send(collections); - } - } - _ => {} - } -} diff --git a/src/lib.rs b/src/lib.rs index dce7946..6f5f8a1 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,14 +1,13 @@ pub mod admin; pub mod auth; -pub mod backfill; pub mod config; pub mod error; -pub mod jetstream; pub mod lexicon; pub mod profile; pub mod repo; pub mod resolve; pub mod server; +pub mod tap; pub mod xrpc; use config::Config; diff --git a/src/main.rs b/src/main.rs index da5b34c..2d831db 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,7 +1,7 @@ use happyview::config::Config; use happyview::lexicon::{LexiconRegistry, ParsedLexicon, ProcedureAction}; use happyview::resolve::{fetch_lexicon_from_pds, resolve_nsid_authority}; -use happyview::{AppState, backfill, jetstream, server}; +use happyview::{AppState, server, tap}; use tokio::sync::watch; use tracing::{info, warn}; @@ -107,6 +107,7 @@ async fn main() { } 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 state = AppState { @@ -117,19 +118,32 @@ async fn main() { collections_tx, }; - jetstream::spawn( + // 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.db.clone(), - config.jetstream_url.clone(), + config.tap_url.clone(), + config.tap_admin_password.clone(), collections_rx, state.lexicons.clone(), state.collections_tx.clone(), ); - backfill::spawn_worker( - state.db.clone(), - state.http.clone(), - config.relay_url.clone(), - config.plc_url.clone(), - ); let app = server::router(state); let addr = config.listen_addr(); diff --git a/src/tap.rs b/src/tap.rs new file mode 100644 index 0000000..acc2bd6 --- /dev/null +++ b/src/tap.rs @@ -0,0 +1,475 @@ +use futures_util::{SinkExt, StreamExt}; +use serde::Deserialize; +use serde_json::Value; +use sqlx::PgPool; +use tokio::sync::watch; +use tokio_tungstenite::tungstenite::Message; +use tokio_tungstenite::tungstenite::client::IntoClientRequest; + +use crate::lexicon::{LexiconRegistry, 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(()) +} + +/// 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 +} + +// --------------------------------------------------------------------------- +// 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( + db: PgPool, + tap_url: String, + tap_admin_password: Option, + mut collections_rx: watch::Receiver>, + lexicons: LexiconRegistry, + collections_tx: watch::Sender>, +) { + let http = reqwest::Client::new(); + + tokio::spawn(async move { + loop { + // Build WebSocket URL from HTTP URL. + let ws_url = build_ws_url(&tap_url); + + match run( + &db, + &http, + &tap_url, + tap_admin_password.as_deref(), + &ws_url, + &mut collections_rx, + &lexicons, + &collections_tx, + ) + .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 +// --------------------------------------------------------------------------- + +#[allow(clippy::too_many_arguments)] +async fn run( + db: &PgPool, + http: &reqwest::Client, + tap_url: &str, + tap_admin_password: Option<&str>, + ws_url: &str, + collections_rx: &mut watch::Receiver>, + lexicons: &LexiconRegistry, + collections_tx: &watch::Sender>, +) -> Result<(), Box> { + 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"); + + let (mut write, mut read) = ws.split(); + + loop { + tokio::select! { + msg = read.next() => { + let msg = match msg { + Some(Ok(m)) => m, + Some(Err(e)) => return Err(e.into()), + None => break, + }; + + let text = match msg { + Message::Text(t) => t, + Message::Close(_) => break, + _ => continue, + }; + + let event: TapEvent = match serde_json::from_str(&text) { + Ok(e) => e, + Err(e) => { + tracing::debug!("skipping unparseable tap event: {e}"); + continue; + } + }; + + let event_id = event.id; + + match event.event_type.as_str() { + "record" => { + if let Some(record) = event.record { + handle_record_event(db, lexicons, collections_tx, &record).await; + } + } + "identity" => { + if let Some(identity) = event.identity { + tracing::debug!( + did = %identity.did, + handle = ?identity.handle, + "received identity event from tap" + ); + } + } + other => { + tracing::debug!(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()); + } + } + // 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}"); + } + } + } + } + + Ok(()) +} + +// --------------------------------------------------------------------------- +// Record event handler +// --------------------------------------------------------------------------- + +async fn handle_record_event( + db: &PgPool, + lexicons: &LexiconRegistry, + collections_tx: &watch::Sender>, + record: &TapRecordEvent, +) { + 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(db, lexicons, collections_tx, &record.did, record).await; + 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(); + + if let Err(e) = sqlx::query( + r#" + INSERT INTO records (uri, did, collection, rkey, record, cid, indexed_at) + VALUES ($1, $2, $3, $4, $5, $6, NOW()) + ON CONFLICT (uri) DO UPDATE + SET record = EXCLUDED.record, + cid = EXCLUDED.cid, + indexed_at = NOW() + "#, + ) + .bind(&uri) + .bind(&record.did) + .bind(&record.collection) + .bind(&record.rkey) + .bind(rec) + .bind(cid) + .execute(db) + .await + { + tracing::warn!(uri = %uri, "failed to upsert record: {e}"); + } + } + "delete" => { + if let Err(e) = sqlx::query("DELETE FROM records WHERE uri = $1") + .bind(&uri) + .execute(db) + .await + { + tracing::warn!(uri = %uri, "failed to delete record: {e}"); + } + } + _ => {} + } +} + +// --------------------------------------------------------------------------- +// Lexicon schema event handler +// --------------------------------------------------------------------------- + +/// Handle a `com.atproto.lexicon.schema` record event for tracked network lexicons. +async fn handle_lexicon_schema_event( + db: &PgPool, + lexicons: &LexiconRegistry, + collections_tx: &watch::Sender>, + did: &str, + record: &TapRecordEvent, +) { + let nsid = &record.rkey; + + // Check if this NSID is one we're tracking and the DID matches the authority. + let tracked: Option<(Option,)> = sqlx::query_as( + "SELECT target_collection FROM network_lexicons WHERE nsid = $1 AND authority_did = $2", + ) + .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, + ) { + 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. + if let Err(e) = sqlx::query( + r#" + INSERT INTO lexicons (id, lexicon_json, backfill, target_collection) + VALUES ($1, $2, false, $3) + ON CONFLICT (id) DO UPDATE SET + lexicon_json = EXCLUDED.lexicon_json, + target_collection = EXCLUDED.target_collection, + revision = lexicons.revision + 1, + updated_at = NOW() + "#, + ) + .bind(nsid) + .bind(rec) + .bind(&target_collection) + .execute(db) + .await + { + tracing::warn!(nsid, "failed to upsert lexicon from event: {e}"); + return; + } + + // Update last_fetched_at. + let _ = + sqlx::query("UPDATE network_lexicons SET last_fetched_at = NOW() WHERE nsid = $1") + .bind(nsid) + .execute(db) + .await; + + 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 _ = sqlx::query("DELETE FROM lexicons WHERE id = $1") + .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 index 510f604..bb46251 100644 --- a/tests/common/app.rs +++ b/tests/common/app.rs @@ -35,7 +35,8 @@ impl TestApp { port: 0, database_url: String::new(), // not used — pool is already connected aip_url: mock_url.clone(), - jetstream_url: String::new(), + tap_url: "http://localhost:2480".into(), + tap_admin_password: None, relay_url: mock_url.clone(), plc_url: mock_url.clone(), static_dir: "./web/out".into(), -- 2.51.2