Something went wrong. Try again.
A lexicon-driven AppView for ATProto. happyview.dev
backfill firehose jetstream atproto appview oauth lexicon
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803use std::sync::Arc;use std::sync::atomic::Ordering;
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};use crate::lua::RecordHookOutcome;
/// 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<Value>, pub cid: Option<String>,}
/// The outcome of processing one record event, returned so a caller can/// account for it in telemetry rather than this function reaching into/// `state.telemetry_counters` itself.#[derive(Debug, Clone, Copy, PartialEq, Eq)]#[must_use]pub enum RecordOutcome { Matched, Skipped, SchemaEvent, Errored,}
/// Process a record event: upsert/delete the record in the database, run index/// hooks, and handle lexicon schema events. Returns the outcome so the caller/// can account for it in telemetry — see `RecordOutcome`.pub async fn handle_record_event(state: &AppState, record: &RecordEvent) -> RecordOutcome { 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 RecordOutcome::SchemaEvent; }
// 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 RecordOutcome::Skipped; }
match record.action.as_str() { "create" | "update" => { let rec = match &record.record { Some(r) => r, None => return RecordOutcome::Errored, }; let cid = record.cid.as_deref().unwrap_or_default();
// Reject records whose claimed CID doesn't match their content // (security review L9). A hostile source can otherwise store a // record under a mismatched CID. Skip indexing entirely on // mismatch; `Skipped` (no/unencodable CID) proceeds unchanged. if crate::cid_verify::verify_record_cid(cid, rec) == crate::cid_verify::CidCheck::Mismatch { log_event( db, EventLog { event_type: "record.cid_mismatch".to_string(), severity: Severity::Warn, actor_did: None, subject: Some(uri.clone()), detail: serde_json::json!({ "collection": record.collection, "did": record.did, "rkey": record.rkey, "claimed_cid": cid, "reason": "record content does not match claimed CID", }), }, state.db_backend, ) .await; return RecordOutcome::Skipped; }
// Run record-event script (if any) before storing. The script's // return value determines what gets written: // Skip → skip indexing entirely // Replace(record) → upsert with that record body // Proceed → upsert with the record as it arrived // The dispatcher cascades `record.<action>:<nsid>` → // `record.index:<nsid>`; failures are dead-lettered fail-open. let hook_result = crate::lua::run_record_event_script( state, crate::lua::RecordEventPayload { nsid: &record.collection, action: &record.action, uri: &uri, did: &record.did, rkey: &record.rkey, record: Some(rec), }, ) .await; let rec_to_store = match hook_result { RecordHookOutcome::Skip => { // Gated like `record.created`/`record.deleted` below: a // filtering script skips far more records than it keeps, so // logging every skip writes a row per discarded firehose // record. if state.verbose_event_logging.load(Ordering::Relaxed) { 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": "script returned nil", }), }, state.db_backend, ) .await; } return RecordOutcome::Skipped; } RecordHookOutcome::Replace(v) => v, RecordHookOutcome::Proceed => rec.clone(), };
let now = now_rfc3339(); let backend = state.db_backend; let insert_sql = adapt_sql( r#" INSERT INTO happyview_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 crate::db::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;
if state.verbose_event_logging.load(Ordering::Relaxed) { 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()); RecordOutcome::Matched } 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; RecordOutcome::Errored } } } "delete" => { let backend = state.db_backend;
// Run record-event script (if any) before deleting. Only a // script that actually ran and returned `nil` aborts the // delete — no script, or a dead-lettered one, proceeds. let hook_result = crate::lua::run_record_event_script( state, crate::lua::RecordEventPayload { nsid: &record.collection, action: "delete", uri: &uri, did: &record.did, rkey: &record.rkey, record: None, }, ) .await; if hook_result == RecordHookOutcome::Skip { // Gated for the same reason as the create path above. if state.verbose_event_logging.load(Ordering::Relaxed) { 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": "script returned nil", }), }, backend, ) .await; } return RecordOutcome::Skipped; }
let delete_sql = adapt_sql("DELETE FROM happyview_records WHERE uri = ?", backend); match crate::db::query(&delete_sql).bind(&uri).execute(db).await { Ok(_) => { if state.verbose_event_logging.load(Ordering::Relaxed) { 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; } RecordOutcome::Matched } 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; RecordOutcome::Errored } } } _ => RecordOutcome::Errored, }}
/// 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 happyview_lexicons WHERE id = ? AND source = 'network' AND authority_did = ?", backend, ); let tracked: Option<(Option<String>,)> = crate::db::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, ) { 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 happyview_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 = happyview_lexicons.revision + 1, updated_at = ? "#, backend, ); if let Err(e) = crate::db::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 happyview_lexicons WHERE id = ?", backend); let _ = crate::db::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); } } _ => {} }}
#[cfg(test)]mod tests { use super::*;
use crate::lexicon::ProcedureAction; use crate::test_support::{memory_pool, test_state_with_pool};
const NSID: &str = "com.example.thing"; const URI: &str = "at://did:plc:abc/com.example.thing/rkey1";
/// A state with the record/script/event-log tables and `NSID` registered as /// a record-type lexicon, so `handle_record_event` treats it as tracked. async fn tracked_state() -> AppState { let pool = memory_pool().await; for ddl in [ "CREATE TABLE happyview_records ( uri TEXT PRIMARY KEY, did TEXT NOT NULL, collection TEXT NOT NULL, rkey TEXT NOT NULL, record TEXT NOT NULL, cid TEXT, indexed_at TEXT NOT NULL, created_at TEXT NOT NULL )", "CREATE TABLE happyview_scripts ( id TEXT PRIMARY KEY, body TEXT NOT NULL, script_type TEXT NOT NULL DEFAULT 'lua' )", "CREATE TABLE happyview_event_logs ( id TEXT PRIMARY KEY, event_type TEXT NOT NULL, severity TEXT NOT NULL, actor_did TEXT, subject TEXT, detail TEXT, created_at TEXT NOT NULL )", "CREATE TABLE happyview_record_refs ( source_uri TEXT NOT NULL, target_uri TEXT NOT NULL, field TEXT NOT NULL )", ] { crate::db::query(ddl) .execute(&pool) .await .unwrap_or_else(|e| panic!("create table: {e}")); }
let state = test_state_with_pool(pool); let parsed = ParsedLexicon::parse( serde_json::json!({ "lexicon": 1, "id": NSID, "defs": {"main": {"type": "record", "key": "tid"}}, }), 1, Some(NSID.to_string()), ProcedureAction::Upsert, None, ) .expect("parse test lexicon"); state.lexicons.upsert(parsed).await; state }
async fn insert_record(state: &AppState) { crate::db::query( "INSERT INTO happyview_records (uri, did, collection, rkey, record, cid, indexed_at, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", ) .bind(URI) .bind("did:plc:abc") .bind(NSID) .bind("rkey1") .bind(r#"{"text":"hello"}"#) .bind("bafyreiabc") .bind("2026-01-01T00:00:00+00:00") .bind("2026-01-01T00:00:00+00:00") .execute(&state.db) .await .expect("seed record"); }
async fn record_exists(state: &AppState) -> bool { let row: Option<(String,)> = crate::db::query_as("SELECT uri FROM happyview_records WHERE uri = ?") .bind(URI) .fetch_optional(&state.db) .await .expect("query record"); row.is_some() }
fn delete_event() -> RecordEvent { RecordEvent { did: "did:plc:abc".to_string(), collection: NSID.to_string(), rkey: "rkey1".to_string(), action: "delete".to_string(), record: None, cid: None, } }
fn create_event() -> RecordEvent { RecordEvent { did: "did:plc:abc".to_string(), collection: NSID.to_string(), rkey: "rkey1".to_string(), action: "create".to_string(), record: Some(serde_json::json!({"text": "hello"})), cid: None, } }
async fn skipped_event_count(state: &AppState) -> i64 { let row: (i64,) = crate::db::query_as( "SELECT COUNT(*) FROM happyview_event_logs WHERE event_type = 'record.skipped'", ) .fetch_one(&state.db) .await .expect("count record.skipped events"); row.0 }
async fn executed_event_count(state: &AppState) -> i64 { let row: (i64,) = crate::db::query_as( "SELECT COUNT(*) FROM happyview_event_logs WHERE event_type = 'script.executed'", ) .fetch_one(&state.db) .await .expect("count script.executed events"); row.0 }
async fn install_script(state: &AppState, trigger: &str, body: &str) { crate::db::query( "INSERT INTO happyview_scripts (id, body, script_type) VALUES (?, ?, 'lua')", ) .bind(trigger) .bind(body) .execute(&state.db) .await .expect("install script"); }
/// Regression test for #80: a Jetstream delete for a tracked collection with /// no registered script must delete the row. The "no script ran" and "the /// script returned nil" signals used to be spelled the same way, so an /// instance with no scripts at all skipped every delete. #[tokio::test] async fn delete_without_any_script_removes_the_record() { let state = tracked_state().await; insert_record(&state).await;
let _ = handle_record_event(&state, &delete_event()).await;
assert!( !record_exists(&state).await, "delete with no registered script must remove the record" ); }
/// `return true` is the documented "proceed, I only had side effects" /// return. On a delete it used to fall through to the original record /// body — which is nil for a delete — and abort. #[tokio::test] async fn delete_with_a_script_returning_true_removes_the_record() { let state = tracked_state().await; install_script( &state, &format!("record.delete:{NSID}"), "function handle() return true end", ) .await; insert_record(&state).await;
let _ = handle_record_event(&state, &delete_event()).await;
assert!( !record_exists(&state).await, "a delete script returning true must let the delete proceed" ); }
/// The documented delete gate: a script that runs and returns `nil` /// still keeps the record. This is the one case that must NOT delete. #[tokio::test] async fn delete_with_a_script_returning_nil_keeps_the_record() { let state = tracked_state().await; install_script( &state, &format!("record.delete:{NSID}"), "function handle() return nil end", ) .await; insert_record(&state).await;
let _ = handle_record_event(&state, &delete_event()).await;
assert!( record_exists(&state).await, "a delete script returning nil must keep the record" ); }
/// The create path's pass-through: no script means index the record as it /// arrived, not skip it. #[tokio::test] async fn create_without_any_script_indexes_the_record() { let state = tracked_state().await;
let _ = handle_record_event( &state, &RecordEvent { did: "did:plc:abc".to_string(), collection: NSID.to_string(), rkey: "rkey1".to_string(), action: "create".to_string(), record: Some(serde_json::json!({"text": "hello"})), cid: None, }, ) .await;
assert!( record_exists(&state).await, "create with no registered script must index the record" ); }
/// `record.skipped` is per-record telemetry on the *most common* outcome /// for a filtering script, so it belongs behind the same `verbose_event_logging` /// gate as its `record.created`/`record.deleted` siblings. Ungated, it wrote a /// row for every discarded firehose record: on upvote.at that was 5.7M of the /// 5.8M rows in `happyview_event_logs`, ~2.7 GB, against 397 `record.created`. #[tokio::test] async fn create_skip_is_not_logged_while_verbose_logging_is_off() { let state = tracked_state().await; install_script( &state, &format!("record.create:{NSID}"), "function handle() return nil end", ) .await;
let _ = handle_record_event(&state, &create_event()).await;
assert!( !record_exists(&state).await, "a create script returning nil must skip indexing" ); assert_eq!( skipped_event_count(&state).await, 0, "record.skipped must not be logged while verbose event logging is off" ); }
#[tokio::test] async fn delete_skip_is_not_logged_while_verbose_logging_is_off() { let state = tracked_state().await; install_script( &state, &format!("record.delete:{NSID}"), "function handle() return nil end", ) .await; insert_record(&state).await;
let _ = handle_record_event(&state, &delete_event()).await;
assert!( record_exists(&state).await, "a delete script returning nil must keep the record" ); assert_eq!( skipped_event_count(&state).await, 0, "record.skipped must not be logged while verbose event logging is off" ); }
/// The gate must suppress the log, not remove it: with verbose logging on, /// both skip paths still report. #[tokio::test] async fn create_skip_is_logged_when_verbose_logging_is_on() { let state = tracked_state().await; state.verbose_event_logging.store(true, Ordering::Relaxed); install_script( &state, &format!("record.create:{NSID}"), "function handle() return nil end", ) .await;
let _ = handle_record_event(&state, &create_event()).await;
assert_eq!( skipped_event_count(&state).await, 1, "record.skipped must still be logged when verbose event logging is on" ); }
#[tokio::test] async fn delete_skip_is_logged_when_verbose_logging_is_on() { let state = tracked_state().await; state.verbose_event_logging.store(true, Ordering::Relaxed); install_script( &state, &format!("record.delete:{NSID}"), "function handle() return nil end", ) .await; insert_record(&state).await;
let _ = handle_record_event(&state, &delete_event()).await;
assert_eq!( skipped_event_count(&state).await, 1, "record.skipped must still be logged when verbose event logging is on" ); }
/// `script.executed` fires once per record that reaches a script, so it has /// the same volume as `record.created` and sits behind the same gate. #[tokio::test] async fn script_executed_is_not_logged_while_verbose_logging_is_off() { let state = tracked_state().await; install_script( &state, &format!("record.create:{NSID}"), "function handle() return true end", ) .await;
let _ = handle_record_event(&state, &create_event()).await;
assert!( record_exists(&state).await, "a create script returning true must index the record" ); assert_eq!( executed_event_count(&state).await, 0, "script.executed must not be logged while verbose event logging is off" ); }
#[tokio::test] async fn script_executed_is_logged_when_verbose_logging_is_on() { let state = tracked_state().await; state.verbose_event_logging.store(true, Ordering::Relaxed); install_script( &state, &format!("record.create:{NSID}"), "function handle() return true end", ) .await;
let _ = handle_record_event(&state, &create_event()).await;
assert_eq!( executed_event_count(&state).await, 1, "script.executed must still be logged when verbose event logging is on" ); }}