From 518bfc717454de3309edd964633d88cbcf53c4bf Mon Sep 17 00:00:00 2001 From: Trezy Date: Sun, 8 Mar 2026 23:25:45 -0500 Subject: [PATCH] feat: run index hooks before storing records to enable record bypass and manipulation --- docs/guides/index-hooks.md | 60 +++++++++++++++-- src/lua/execute.rs | 80 +++++++++++++++++++--- src/tap.rs | 134 +++++++++++++++++++++++-------------- 3 files changed, 210 insertions(+), 64 deletions(-) diff --git a/docs/guides/index-hooks.md b/docs/guides/index-hooks.md index d674faa..f4239a5 100644 --- a/docs/guides/index-hooks.md +++ b/docs/guides/index-hooks.md @@ -1,8 +1,8 @@ # Index Hooks -Index hooks are Lua scripts that run automatically whenever a record in a collection is created, updated, or deleted on the network. They let you react to record changes in real time — push data to search engines, sync with external APIs, send notifications, or build materialized views. +Index hooks are Lua scripts that run automatically whenever a record in a collection is created, updated, or deleted on the network. They run **before** the record is indexed, giving you the ability to filter out unwanted records, transform record data before storage, or trigger side effects like syncing with external services. -Unlike [query and procedure scripts](scripting.md) that run in response to XRPC requests, index hooks are triggered by the firehose. They run asynchronously and never block record indexing. +Unlike [query and procedure scripts](scripting.md) that run in response to XRPC requests, index hooks are triggered by the firehose. ## Attaching a hook @@ -22,7 +22,17 @@ function handle() end ``` -The function is called once per record event. There is no return value — index hooks are fire-and-forget from the caller's perspective. +The function is called once per record event. The return value controls what happens next: + +| Return value | Effect | +| ------------ | ------------------------------------------------ | +| `nil` | The record is **not** indexed (skipped entirely) | +| A table | That table is stored as the record instead | +| *(no hook)* | The original record is stored as-is | + +On **delete** events, returning `nil` skips the delete (the record stays in the database). + +If the hook errors after all retries, the system **fails open** — the original record is stored and the failed event is dead-lettered for later inspection. ## Context globals @@ -54,10 +64,14 @@ Index hooks are designed to be resilient: 1. If a hook fails, it retries up to **3 times** with exponential backoff (1s, 2s, 4s delays). 2. If all retries are exhausted, the failed event is inserted into the `dead_letter_hooks` table for later inspection. -3. Hook failures never block record indexing — the record is always indexed regardless of whether the hook succeeds. +3. On failure the system **fails open** — the original record is stored as-is so indexing is not permanently blocked. Failed hooks are logged as errors. Check the [event logs](event-logs.md) or query the `dead_letter_hooks` table directly to find and replay failures. +### Performance considerations + +Because hooks run synchronously before indexing, they block the firehose consumer while executing. With retry logic (1s + 2s + 4s backoff), a persistently failing hook could block for ~7 seconds per record. Keep hook scripts fast and ensure external services they depend on are reliable. + ### Dead letter table The `dead_letter_hooks` table stores events that failed all retry attempts: @@ -78,6 +92,39 @@ The `dead_letter_hooks` table stores events that failed all retry attempts: ## Examples +### Filter out records missing a required field + +Skip indexing any record that doesn't have a `title` field: + +```lua +function handle() + if action == "delete" then + return record -- allow deletes to proceed + end + + if record.title == nil or record.title == "" then + return nil -- skip: no title + end + + return record +end +``` + +### Transform a record before storage + +Enrich a record with a computed field before it is stored: + +```lua +function handle() + if action == "delete" then + return record + end + + record.slug = string.lower(string.gsub(record.title or "", "%s+", "-")) + return record +end +``` + ### Post to a webhook ```lua @@ -91,6 +138,7 @@ function handle() record = record }) }) + return record end ``` @@ -121,6 +169,8 @@ function handle() }) }) end + + return record end ``` @@ -154,6 +204,8 @@ function handle() })) }) end + + return record end ``` diff --git a/src/lua/execute.rs b/src/lua/execute.rs index 46e78f8..74b47b7 100644 --- a/src/lua/execute.rs +++ b/src/lua/execute.rs @@ -584,9 +584,15 @@ pub struct HookEvent<'a> { /// Execute a Lua hook script triggered by a record index event. /// +/// Runs **before** the record is indexed. The return value determines what +/// gets stored: +/// - `None` → skip the DB operation entirely +/// - `Some(value)` → use that value for the insert/update +/// /// Retries up to 3 times with exponential backoff (1s, 2s, 4s). -/// On final failure, inserts into `dead_letter_hooks` table. -pub async fn execute_hook_script(event: &HookEvent<'_>) { +/// On final failure, dead-letters the event and returns `Some(original_record)` +/// (fail-open so indexing is not permanently blocked). +pub async fn execute_hook_script(event: &HookEvent<'_>) -> Option { let max_attempts: i32 = 4; // 1 initial + 3 retries let mut last_error = String::new(); @@ -597,7 +603,7 @@ pub async fn execute_hook_script(event: &HookEvent<'_>) { } match run_hook_once(event).await { - Ok(()) => { + Ok(hook_result) => { log_event( &event.state.db, EventLog { @@ -614,7 +620,7 @@ pub async fn execute_hook_script(event: &HookEvent<'_>) { }, ) .await; - return; + return hook_result; } Err(e) => { last_error = e; @@ -628,7 +634,8 @@ pub async fn execute_hook_script(event: &HookEvent<'_>) { } } - // All retries exhausted — dead-letter the event. + // All retries exhausted — dead-letter the event and fail-open with the + // original record so indexing is not permanently blocked. tracing::error!( uri = event.uri, lexicon_id = event.lexicon_id, @@ -673,10 +680,17 @@ pub async fn execute_hook_script(event: &HookEvent<'_>) { }, ) .await; + + // Fail-open: return the original record so indexing proceeds. + event.record.cloned() } -/// Execute a hook script once. Returns Ok(()) on success or Err(message) on failure. -async fn run_hook_once(event: &HookEvent<'_>) -> Result<(), String> { +/// Execute a hook script once. +/// +/// Returns `Ok(None)` when `handle()` returns nil (meaning "skip indexing"), +/// `Ok(Some(value))` when it returns a table (use that as the record), or +/// `Ok(Some(original))` for other non-nil types. +async fn run_hook_once(event: &HookEvent<'_>) -> Result, String> { let lua = sandbox::create_sandbox().map_err(|e| format!("failed to create Lua VM: {e}"))?; let state_arc = Arc::new(event.state.clone()); @@ -710,12 +724,24 @@ async fn run_hook_once(event: &HookEvent<'_>) -> Result<(), String> { .get("handle") .map_err(|e| format!("script missing handle function: {e}"))?; - handle + let result: mlua::Value = handle .call_async::(()) .await .map_err(|e| e.to_string())?; - Ok(()) + match result { + mlua::Value::Nil => Ok(None), + mlua::Value::Table(_) => { + let json_value: Value = lua + .from_value(result) + .map_err(|e| format!("failed to convert lua table to JSON: {e}"))?; + Ok(Some(json_value)) + } + _ => { + // Non-nil, non-table return — proceed with the original record. + Ok(event.record.cloned()) + } + } } #[cfg(test)] @@ -775,6 +801,42 @@ mod tests { let event = make_event(&state, "function handle() end", "create", None); let result = run_hook_once(&event).await; assert!(result.is_ok(), "expected Ok, got: {:?}", result); + // handle() returns nil implicitly, so result should be None (skip). + assert!(result.unwrap().is_none()); + } + + #[tokio::test] + async fn hook_returns_nil_to_skip() { + let state = test_state(); + let record = json!({"name": "Test"}); + let event = make_event( + &state, + "function handle() return nil end", + "create", + Some(&record), + ); + let result = run_hook_once(&event).await; + assert!(result.is_ok(), "expected Ok, got: {:?}", result); + assert!(result.unwrap().is_none(), "nil return should produce None"); + } + + #[tokio::test] + async fn hook_returns_modified_record() { + let state = test_state(); + let record = json!({"name": "Original"}); + let script = r#" + function handle() + return { name = "Modified", extra = true } + end + "#; + let event = make_event(&state, script, "create", Some(&record)); + let result = run_hook_once(&event).await; + assert!(result.is_ok(), "expected Ok, got: {:?}", result); + let value = result.unwrap(); + assert!(value.is_some(), "table return should produce Some"); + let v = value.unwrap(); + assert_eq!(v["name"], "Modified"); + assert_eq!(v["extra"], true); } #[tokio::test] diff --git a/src/tap.rs b/src/tap.rs index b29de6a..c191b03 100644 --- a/src/tap.rs +++ b/src/tap.rs @@ -444,6 +444,51 @@ async fn handle_record_event(state: &AppState, record: &TapRecordEvent) { }; 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", + }), + }, + ) + .await; + return; + } + Some(v) => v, + } + } else { + // No hook — store the original record as-is. + rec.clone() + }; + match sqlx::query( r#" INSERT INTO records (uri, did, collection, rkey, record, cid, indexed_at) @@ -458,7 +503,7 @@ async fn handle_record_event(state: &AppState, record: &TapRecordEvent) { .bind(&record.did) .bind(&record.collection) .bind(&record.rkey) - .bind(rec) + .bind(&rec_to_store) .bind(cid) .execute(db) .await @@ -479,32 +524,6 @@ async fn handle_record_event(state: &AppState, record: &TapRecordEvent) { }, ) .await; - - // Fire index hook if configured. - if let Some(script) = state.lexicons.get_index_hook(&record.collection).await { - let hook_state = state.clone(); - let hook_lexicon_id = record.collection.clone(); - let hook_uri = uri.clone(); - let hook_did = record.did.clone(); - let hook_collection = record.collection.clone(); - let hook_rkey = record.rkey.clone(); - let hook_action = record.action.clone(); - let hook_rec = rec.clone(); - tokio::spawn(async move { - crate::lua::execute_hook_script(&crate::lua::HookEvent { - state: &hook_state, - lexicon_id: &hook_lexicon_id, - script: &script, - action: &hook_action, - uri: &hook_uri, - did: &hook_did, - collection: &hook_collection, - rkey: &hook_rkey, - record: Some(&hook_rec), - }) - .await; - }); - } } Err(e) => { tracing::warn!(uri = %uri, "failed to upsert record: {e}"); @@ -528,6 +547,43 @@ async fn handle_record_event(state: &AppState, record: &TapRecordEvent) { } } "delete" => { + // 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", + }), + }, + ) + .await; + return; + } + } + match sqlx::query("DELETE FROM records WHERE uri = $1") .bind(&uri) .execute(db) @@ -549,30 +605,6 @@ async fn handle_record_event(state: &AppState, record: &TapRecordEvent) { }, ) .await; - - // Fire index hook if configured. - if let Some(script) = state.lexicons.get_index_hook(&record.collection).await { - let hook_state = state.clone(); - let hook_lexicon_id = record.collection.clone(); - let hook_uri = uri.clone(); - let hook_did = record.did.clone(); - let hook_collection = record.collection.clone(); - let hook_rkey = record.rkey.clone(); - tokio::spawn(async move { - crate::lua::execute_hook_script(&crate::lua::HookEvent { - state: &hook_state, - lexicon_id: &hook_lexicon_id, - script: &script, - action: "delete", - uri: &hook_uri, - did: &hook_did, - collection: &hook_collection, - rkey: &hook_rkey, - record: None, - }) - .await; - }); - } } Err(e) => { tracing::warn!(uri = %uri, "failed to delete record: {e}"); -- 2.51.2