Something went wrong. Try again.
A lexicon-driven AppView for ATProto.
Something went wrong. Try again.
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800use 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<TapRecordEvent>, identity: Option<TapIdentityEvent>,}
#[derive(Deserialize)]struct TapRecordEvent { did: String, collection: String, rkey: String, action: String, record: Option<Value>, cid: Option<String>, #[allow(dead_code)] live: Option<bool>,}
#[derive(Deserialize)]#[allow(dead_code)]struct TapIdentityEvent { did: String, handle: Option<String>, #[serde(rename = "isActive")] is_active: Option<bool>, status: Option<String>,}
// ---------------------------------------------------------------------------// 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<T: serde::de::DeserializeOwned>( http: &reqwest::Client, tap_url: &str, path: &str, password: Option<&str>,) -> Result<T, String> { 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::<T>() .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<TapStats, String> { let (repo, record, outbox) = tokio::try_join!( tap_get::<RepoCountResponse>(http, tap_url, "/stats/repo-count", tap_admin_password), tap_get::<RecordCountResponse>(http, tap_url, "/stats/record-count", tap_admin_password), tap_get::<OutboxBufferResponse>(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<Vec<String>>) { 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<Vec<String>>,) -> Result<(), Box<dyn std::error::Error + Send + Sync>> { 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::net::TcpStream>, >, _, ) = 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<String>,)> = 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); } } _ => {} }}