diff --git a/migrations/20260304000000_add_on_index_script.sql b/migrations/20260304000000_add_on_index_script.sql new file mode 100644 --- /dev/null +++ b/migrations/20260304000000_add_on_index_script.sql @@ -0,0 +1,1 @@ +ALTER TABLE lexicons ADD COLUMN on_index_script TEXT; diff --git a/migrations/20260304000001_create_dead_letter_hooks.sql b/migrations/20260304000001_create_dead_letter_hooks.sql new file mode 100644 --- /dev/null +++ b/migrations/20260304000001_create_dead_letter_hooks.sql @@ -0,0 +1,16 @@ +CREATE TABLE dead_letter_hooks ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + lexicon_id TEXT NOT NULL, + uri TEXT NOT NULL, + did TEXT NOT NULL, + collection TEXT NOT NULL, + rkey TEXT NOT NULL, + action TEXT NOT NULL, + record JSONB, + error TEXT NOT NULL, + attempts INT NOT NULL DEFAULT 0, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX idx_dead_letter_hooks_collection ON dead_letter_hooks (collection); +CREATE INDEX idx_dead_letter_hooks_created_at ON dead_letter_hooks (created_at); diff --git a/src/admin/lexicons.rs b/src/admin/lexicons.rs --- a/src/admin/lexicons.rs +++ b/src/admin/lexicons.rs @@ -57,6 +57,7 @@ 1, body.target_collection.clone(), action.clone(), body.script.clone(), + body.on_index_script.clone(), ) .map_err(|e| AppError::BadRequest(format!("failed to parse lexicon: {e}")))?; @@ -65,20 +66,26 @@ if let Some(ref script) = body.script { crate::lua::validate_script(script).map_err(AppError::BadRequest)?; } + // Validate on_index_script if provided + if let Some(ref script) = body.on_index_script { + crate::lua::validate_script(script).map_err(AppError::BadRequest)?; + } + let action_str = action.to_optional_str(); let has_script = body.script.is_some(); // Upsert into database let row: (i32,) = sqlx::query_as( r#" - INSERT INTO lexicons (id, lexicon_json, backfill, target_collection, action, script, source) - VALUES ($1, $2, $3, $4, $5, $6, 'manual') + INSERT INTO lexicons (id, lexicon_json, backfill, target_collection, action, script, on_index_script, source) + VALUES ($1, $2, $3, $4, $5, $6, $7, 'manual') ON CONFLICT (id) DO UPDATE SET lexicon_json = EXCLUDED.lexicon_json, backfill = EXCLUDED.backfill, target_collection = EXCLUDED.target_collection, action = EXCLUDED.action, script = EXCLUDED.script, + on_index_script = EXCLUDED.on_index_script, source = 'manual', revision = lexicons.revision + 1, updated_at = NOW() @@ -91,6 +98,7 @@ .bind(body.backfill) .bind(&body.target_collection) .bind(action_str) .bind(&body.script) + .bind(&body.on_index_script) .fetch_one(&state.db) .await .map_err(|e| AppError::Internal(format!("failed to upsert lexicon: {e}")))?; @@ -104,6 +112,7 @@ revision, body.target_collection, action, body.script, + body.on_index_script.clone(), ) .map_err(|e| AppError::Internal(format!("failed to re-parse lexicon: {e}")))?; let is_record = parsed.lexicon_type == LexiconType::Record; @@ -134,6 +143,7 @@ subject: Some(id.clone()), detail: serde_json::json!({ "revision": revision, "has_script": has_script, + "has_on_index_script": body.on_index_script.is_some(), "source": "manual", }), }, @@ -155,9 +165,9 @@ State(state): State, _admin: AdminAuth, ) -> Result>, AppError> { #[allow(clippy::type_complexity)] - let rows: Vec<(String, i32, Value, bool, Option, Option, Option, String, Option, Option>, chrono::DateTime, chrono::DateTime)> = + let rows: Vec<(String, i32, Value, bool, Option, Option, Option, Option, String, Option, Option>, chrono::DateTime, chrono::DateTime)> = sqlx::query_as( - "SELECT id, revision, lexicon_json, backfill, action, target_collection, script, source, authority_did, last_fetched_at, created_at, updated_at FROM lexicons ORDER BY id", + "SELECT id, revision, lexicon_json, backfill, action, target_collection, script, on_index_script, source, authority_did, last_fetched_at, created_at, updated_at FROM lexicons ORDER BY id", ) .fetch_all(&state.db) .await @@ -174,6 +184,7 @@ backfill, action, target_collection, script, + on_index_script, source, authority_did, last_fetched_at, @@ -181,7 +192,7 @@ created_at, updated_at, )| { let parsed = - ParsedLexicon::parse(json, revision, None, ProcedureAction::Upsert, None); + ParsedLexicon::parse(json, revision, None, ProcedureAction::Upsert, None, None); let lexicon_type = parsed .as_ref() .map(|p| format!("{:?}", p.lexicon_type).to_lowercase()) @@ -199,6 +210,7 @@ backfill, action, target_collection, has_script: script.is_some(), + has_on_index_script: on_index_script.is_some(), source, authority_did, last_fetched_at, @@ -220,9 +232,9 @@ _admin: AdminAuth, Path(id): Path, ) -> Result, AppError> { #[allow(clippy::type_complexity)] - let row: Option<(String, i32, Value, bool, Option, Option, Option, String, Option, Option>, chrono::DateTime, chrono::DateTime)> = + let row: Option<(String, i32, Value, bool, Option, Option, Option, Option, String, Option, Option>, chrono::DateTime, chrono::DateTime)> = sqlx::query_as( - "SELECT id, revision, lexicon_json, backfill, action, target_collection, script, source, authority_did, last_fetched_at, created_at, updated_at FROM lexicons WHERE id = $1", + "SELECT id, revision, lexicon_json, backfill, action, target_collection, script, on_index_script, source, authority_did, last_fetched_at, created_at, updated_at FROM lexicons WHERE id = $1", ) .bind(&id) .fetch_optional(&state.db) @@ -237,6 +249,7 @@ backfill, action, target_collection, script, + on_index_script, source, authority_did, last_fetched_at, @@ -250,6 +263,7 @@ revision, None, ProcedureAction::Upsert, None, + None, ) .map(|p| format!("{:?}", p.lexicon_type).to_lowercase()) .unwrap_or_else(|_| "unknown".into()); @@ -266,6 +280,8 @@ "action": action, "target_collection": target_collection, "has_script": has_script, "script": script, + "has_on_index_script": on_index_script.is_some(), + "on_index_script": on_index_script, "source": source, "authority_did": authority_did, "last_fetched_at": last_fetched_at, diff --git a/src/admin/network_lexicons.rs b/src/admin/network_lexicons.rs --- a/src/admin/network_lexicons.rs +++ b/src/admin/network_lexicons.rs @@ -41,6 +41,7 @@ 1, body.target_collection.clone(), ProcedureAction::Upsert, None, + None, ) .map_err(|e| AppError::BadRequest(format!("failed to parse lexicon: {e}")))?; @@ -77,6 +78,7 @@ lexicon_json, revision, body.target_collection, ProcedureAction::Upsert, + None, None, ) .map_err(|e| AppError::Internal(format!("failed to re-parse lexicon: {e}")))?; diff --git a/src/admin/types.rs b/src/admin/types.rs --- a/src/admin/types.rs +++ b/src/admin/types.rs @@ -14,6 +14,7 @@ pub(super) backfill: bool, pub(super) action: Option, pub(super) target_collection: Option, pub(super) has_script: bool, + pub(super) has_on_index_script: bool, pub(super) source: String, pub(super) authority_did: Option, pub(super) last_fetched_at: Option>, @@ -32,6 +33,7 @@ pub(super) backfill: bool, pub(super) target_collection: Option, pub(super) action: Option, pub(super) script: Option, + pub(super) on_index_script: Option, } fn default_backfill() -> bool { diff --git a/src/lexicon.rs b/src/lexicon.rs --- a/src/lexicon.rs +++ b/src/lexicon.rs @@ -81,6 +81,8 @@ /// For procedures: the action this procedure performs (create, update, delete, upsert). pub action: ProcedureAction, /// Optional Lua script that replaces the built-in handler. pub script: Option, + /// Optional Lua script that runs when a record in this collection is indexed. + pub on_index_script: Option, } impl ParsedLexicon { @@ -91,6 +93,7 @@ revision: i32, target_collection: Option, action: ProcedureAction, script: Option, + on_index_script: Option, ) -> Result { let id = raw .get("id") @@ -134,6 +137,7 @@ revision, target_collection, action, script, + on_index_script, }) } } @@ -164,11 +168,12 @@ let rows: Vec<( String, Value, i32, + Option, Option, Option, Option, )> = sqlx::query_as( - "SELECT id, lexicon_json, revision, target_collection, action, script FROM lexicons", + "SELECT id, lexicon_json, revision, target_collection, action, script, on_index_script FROM lexicons", ) .fetch_all(db) .await @@ -178,7 +183,7 @@ let mut inner = self.inner.write().await; inner.clear(); let mut loaded = 0u32; - for (id, json, revision, target_collection, action_str, script) in rows { + for (id, json, revision, target_collection, action_str, script, on_index_script) in rows { let action = match ProcedureAction::from_optional_str(action_str.as_deref()) { Ok(a) => a, Err(e) => { @@ -186,7 +191,14 @@ warn!(%id, "invalid action value: {e}"); ProcedureAction::Upsert } }; - match ParsedLexicon::parse(json, revision, target_collection, action, script) { + match ParsedLexicon::parse( + json, + revision, + target_collection, + action, + script, + on_index_script, + ) { Ok(parsed) => { inner.insert(id, parsed); loaded += 1; @@ -247,6 +259,14 @@ .values() .filter(|lex| lex.lexicon_type == LexiconType::Procedure) .map(|lex| lex.id.clone()) .collect() + } + + /// Get the on_index_script for a record-type lexicon by its collection NSID. + pub async fn get_on_index_script(&self, collection: &str) -> Option { + let inner = self.inner.read().await; + inner + .get(collection) + .and_then(|lex| lex.on_index_script.clone()) } /// Return the total count of registered lexicons. @@ -344,6 +364,7 @@ 1, None, ProcedureAction::Upsert, None, + None, ) .unwrap(); assert_eq!(parsed.id, "games.gamesgamesgamesgames.game"); @@ -361,6 +382,7 @@ query_lexicon_json(), 2, Some("games.gamesgamesgamesgames.game".into()), ProcedureAction::Upsert, + None, None, ) .unwrap(); @@ -382,6 +404,7 @@ 1, None, ProcedureAction::Upsert, None, + None, ) .unwrap(); assert_eq!(parsed.lexicon_type, LexiconType::Procedure); @@ -396,6 +419,7 @@ procedure_lexicon_json(), 1, None, ProcedureAction::Delete, + None, None, ) .unwrap(); @@ -410,6 +434,7 @@ 1, None, ProcedureAction::Upsert, None, + None, ) .unwrap(); assert_eq!(parsed.lexicon_type, LexiconType::Definitions); @@ -418,7 +443,7 @@ #[test] fn parse_missing_id_returns_error() { let raw = json!({"lexicon": 1, "defs": {}}); - let result = ParsedLexicon::parse(raw, 1, None, ProcedureAction::Upsert, None); + let result = ParsedLexicon::parse(raw, 1, None, ProcedureAction::Upsert, None, None); assert!(result.is_err()); assert!(result.unwrap_err().contains("id")); } @@ -427,7 +452,8 @@ #[test] fn parse_preserves_raw_json() { let raw = record_lexicon_json(); let parsed = - ParsedLexicon::parse(raw.clone(), 1, None, ProcedureAction::Upsert, None).unwrap(); + ParsedLexicon::parse(raw.clone(), 1, None, ProcedureAction::Upsert, None, None) + .unwrap(); assert_eq!(parsed.raw, raw); } @@ -438,6 +464,7 @@ query_lexicon_json(), 1, Some("custom.collection".into()), ProcedureAction::Upsert, + None, None, ) .unwrap(); @@ -463,6 +490,7 @@ 1, None, ProcedureAction::Upsert, None, + None, ) .unwrap(); reg.upsert(parsed).await; @@ -480,6 +508,7 @@ record_lexicon_json(), 1, None, ProcedureAction::Upsert, + None, None, ) .unwrap(); @@ -491,6 +520,7 @@ 5, None, ProcedureAction::Upsert, None, + None, ) .unwrap(); reg.upsert(v2).await; @@ -513,6 +543,7 @@ record_lexicon_json(), 1, None, ProcedureAction::Upsert, + None, None, ) .unwrap(); @@ -544,17 +575,25 @@ 1, None, ProcedureAction::Upsert, None, + None, ) .unwrap(); - let query = - ParsedLexicon::parse(query_lexicon_json(), 1, None, ProcedureAction::Upsert, None) - .unwrap(); + let query = ParsedLexicon::parse( + query_lexicon_json(), + 1, + None, + ProcedureAction::Upsert, + None, + None, + ) + .unwrap(); let procedure = ParsedLexicon::parse( procedure_lexicon_json(), 1, None, ProcedureAction::Upsert, None, + None, ) .unwrap(); let defs = ParsedLexicon::parse( @@ -562,6 +601,7 @@ definitions_lexicon_json(), 1, None, ProcedureAction::Upsert, + None, None, ) .unwrap(); @@ -631,5 +671,84 @@ assert_eq!(ProcedureAction::Create.to_optional_str(), Some("create")); assert_eq!(ProcedureAction::Update.to_optional_str(), Some("update")); assert_eq!(ProcedureAction::Delete.to_optional_str(), Some("delete")); assert_eq!(ProcedureAction::Upsert.to_optional_str(), None); + } + + // ----------------------------------------------------------------------- + // on_index_script + // ----------------------------------------------------------------------- + + #[test] + fn parse_preserves_on_index_script() { + let parsed = ParsedLexicon::parse( + record_lexicon_json(), + 1, + None, + ProcedureAction::Upsert, + None, + Some("function handle() end".into()), + ) + .unwrap(); + assert_eq!(parsed.on_index_script, Some("function handle() end".into())); + } + + #[test] + fn parse_on_index_script_none_by_default() { + let parsed = ParsedLexicon::parse( + record_lexicon_json(), + 1, + None, + ProcedureAction::Upsert, + None, + None, + ) + .unwrap(); + assert!(parsed.on_index_script.is_none()); + } + + #[tokio::test] + async fn registry_get_on_index_script_returns_script() { + let reg = LexiconRegistry::new(); + let parsed = ParsedLexicon::parse( + record_lexicon_json(), + 1, + None, + ProcedureAction::Upsert, + None, + Some("function handle() log('hook') end".into()), + ) + .unwrap(); + reg.upsert(parsed).await; + + let script = reg + .get_on_index_script("games.gamesgamesgamesgames.game") + .await; + assert_eq!(script, Some("function handle() log('hook') end".into())); + } + + #[tokio::test] + async fn registry_get_on_index_script_returns_none_when_absent() { + let reg = LexiconRegistry::new(); + let parsed = ParsedLexicon::parse( + record_lexicon_json(), + 1, + None, + ProcedureAction::Upsert, + None, + None, + ) + .unwrap(); + reg.upsert(parsed).await; + + let script = reg + .get_on_index_script("games.gamesgamesgamesgames.game") + .await; + assert!(script.is_none()); + } + + #[tokio::test] + async fn registry_get_on_index_script_returns_none_for_unknown() { + let reg = LexiconRegistry::new(); + let script = reg.get_on_index_script("nonexistent").await; + assert!(script.is_none()); } } diff --git a/src/lua/context.rs b/src/lua/context.rs --- a/src/lua/context.rs +++ b/src/lua/context.rs @@ -31,3 +31,81 @@ globals.set("params", lua.to_value(params)?)?; globals.set("collection", collection.to_string())?; Ok(()) } + +/// Set global context variables for an index hook script. +pub fn set_hook_context( + lua: &Lua, + action: &str, + uri: &str, + did: &str, + collection: &str, + rkey: &str, + record: Option<&Value>, +) -> LuaResult<()> { + let globals = lua.globals(); + globals.set("action", action.to_string())?; + globals.set("uri", uri.to_string())?; + globals.set("did", did.to_string())?; + globals.set("collection", collection.to_string())?; + globals.set("rkey", rkey.to_string())?; + match record { + Some(r) => globals.set("record", lua.to_value(r)?)?, + None => globals.set("record", mlua::Value::Nil)?, + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::lua::sandbox::create_sandbox; + use serde_json::json; + + #[test] + fn hook_context_sets_all_globals() { + let lua = create_sandbox().unwrap(); + let record = json!({"name": "Test Game"}); + set_hook_context( + &lua, + "create", + "at://did:plc:abc/col/rkey", + "did:plc:abc", + "col", + "rkey", + Some(&record), + ) + .unwrap(); + + let globals = lua.globals(); + assert_eq!(globals.get::("action").unwrap(), "create"); + assert_eq!( + globals.get::("uri").unwrap(), + "at://did:plc:abc/col/rkey" + ); + assert_eq!(globals.get::("did").unwrap(), "did:plc:abc"); + assert_eq!(globals.get::("collection").unwrap(), "col"); + assert_eq!(globals.get::("rkey").unwrap(), "rkey"); + + let rec: mlua::Table = globals.get("record").unwrap(); + assert_eq!(rec.get::("name").unwrap(), "Test Game"); + } + + #[test] + fn hook_context_record_nil_on_delete() { + let lua = create_sandbox().unwrap(); + set_hook_context( + &lua, + "delete", + "at://did:plc:abc/col/rkey", + "did:plc:abc", + "col", + "rkey", + None, + ) + .unwrap(); + + let globals = lua.globals(); + assert_eq!(globals.get::("action").unwrap(), "delete"); + assert!(globals.get::("record").unwrap().is_nil()); + } +} diff --git a/src/lua/execute.rs b/src/lua/execute.rs --- a/src/lua/execute.rs +++ b/src/lua/execute.rs @@ -516,3 +516,149 @@ .await; Ok(Json(json_value).into_response()) } + +/// Context for a hook execution triggered by a record index event. +pub struct HookEvent<'a> { + pub state: &'a AppState, + pub lexicon_id: &'a str, + pub script: &'a str, + pub action: &'a str, + pub uri: &'a str, + pub did: &'a str, + pub collection: &'a str, + pub rkey: &'a str, + pub record: Option<&'a Value>, +} + +/// Execute a Lua hook script triggered by a record index event. +/// +/// 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<'_>) { + let max_attempts: i32 = 4; // 1 initial + 3 retries + let mut last_error = String::new(); + + for attempt in 0..max_attempts { + if attempt > 0 { + let delay = std::time::Duration::from_secs(1 << (attempt - 1)); // 1s, 2s, 4s + tokio::time::sleep(delay).await; + } + + match run_hook_once(event).await { + Ok(()) => { + log_event( + &event.state.db, + EventLog { + event_type: "hook.executed".to_string(), + severity: Severity::Info, + actor_did: None, + subject: Some(event.uri.to_string()), + detail: serde_json::json!({ + "lexicon_id": event.lexicon_id, + "action": event.action, + "collection": event.collection, + "attempts": attempt + 1, + }), + }, + ) + .await; + return; + } + Err(e) => { + last_error = e; + tracing::warn!( + uri = event.uri, + lexicon_id = event.lexicon_id, + attempt = attempt + 1, + "hook execution failed: {last_error}" + ); + } + } + } + + // All retries exhausted — dead-letter the event. + tracing::error!( + uri = event.uri, + lexicon_id = event.lexicon_id, + "hook dead-lettered after {max_attempts} attempts" + ); + + if let Err(e) = sqlx::query( + r#" + INSERT INTO dead_letter_hooks (lexicon_id, uri, did, collection, rkey, action, record, error, attempts) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + "#, + ) + .bind(event.lexicon_id) + .bind(event.uri) + .bind(event.did) + .bind(event.collection) + .bind(event.rkey) + .bind(event.action) + .bind(event.record) + .bind(&last_error) + .bind(max_attempts) + .execute(&event.state.db) + .await + { + tracing::error!(uri = event.uri, "failed to insert dead letter hook: {e}"); + } + + log_event( + &event.state.db, + EventLog { + event_type: "hook.dead_lettered".to_string(), + severity: Severity::Error, + actor_did: None, + subject: Some(event.uri.to_string()), + detail: serde_json::json!({ + "lexicon_id": event.lexicon_id, + "action": event.action, + "collection": event.collection, + "error": last_error, + "attempts": max_attempts, + }), + }, + ) + .await; +} + +/// Execute a hook script once. Returns Ok(()) on success or Err(message) on failure. +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()); + + db_api::register_db_api(&lua, state_arc.clone()) + .map_err(|e| format!("failed to register db API: {e}"))?; + + http_api::register_http_api(&lua, state_arc) + .map_err(|e| format!("failed to register http API: {e}"))?; + + context::set_hook_context( + &lua, + event.action, + event.uri, + event.did, + event.collection, + event.rkey, + event.record, + ) + .map_err(|e| format!("failed to set hook context: {e}"))?; + + lua.load(event.script) + .exec() + .map_err(|e| format!("script load failed: {e}"))?; + + let handle: mlua::Function = lua + .globals() + .get("handle") + .map_err(|e| format!("script missing handle function: {e}"))?; + + handle + .call_async::(()) + .await + .map_err(|e| e.to_string())?; + + Ok(()) +} diff --git a/src/lua/mod.rs b/src/lua/mod.rs --- a/src/lua/mod.rs +++ b/src/lua/mod.rs @@ -6,5 +6,7 @@ mod record; pub(crate) mod sandbox; mod tid; -pub(crate) use execute::{execute_procedure_script, execute_query_script}; +pub(crate) use execute::{ + HookEvent, execute_hook_script, execute_procedure_script, execute_query_script, +}; pub(crate) use sandbox::validate_script; diff --git a/src/lua/sandbox.rs b/src/lua/sandbox.rs --- a/src/lua/sandbox.rs +++ b/src/lua/sandbox.rs @@ -59,6 +59,28 @@ Ok(seq) })?; globals.set("toarray", toarray_fn)?; + // JSON utilities: json.encode(table) -> string, json.decode(string) -> table + let json_table = lua.create_table()?; + + let encode_fn = lua.create_function(|lua, value: mlua::Value| { + let json_value: serde_json::Value = lua + .from_value(value) + .map_err(|e| mlua::Error::runtime(format!("json.encode: {e}")))?; + serde_json::to_string(&json_value) + .map_err(|e| mlua::Error::runtime(format!("json.encode: {e}"))) + })?; + json_table.set("encode", encode_fn)?; + + let decode_fn = lua.create_function(|lua, s: String| { + let json_value: serde_json::Value = serde_json::from_str(&s) + .map_err(|e| mlua::Error::runtime(format!("json.decode: {e}")))?; + lua.to_value(&json_value) + .map_err(|e| mlua::Error::runtime(format!("json.decode: {e}"))) + })?; + json_table.set("decode", decode_fn)?; + + globals.set("json", json_table)?; + Ok(lua) } @@ -180,5 +202,46 @@ let lua = create_sandbox().unwrap(); let table: mlua::Table = lua.load(r#"return toarray({})"#).eval().unwrap(); let json: serde_json::Value = lua.from_value(mlua::Value::Table(table)).unwrap(); assert!(json.is_array(), "expected JSON array, got: {json}"); + } + + #[test] + fn sandbox_provides_json_encode() { + let lua = create_sandbox().unwrap(); + let result: String = lua + .load(r#"return json.encode({name = "test", count = 42})"#) + .eval() + .unwrap(); + let parsed: serde_json::Value = serde_json::from_str(&result).unwrap(); + assert_eq!(parsed["name"], "test"); + assert_eq!(parsed["count"], 42); + } + + #[test] + fn sandbox_provides_json_decode() { + let lua = create_sandbox().unwrap(); + let result: mlua::Table = lua + .load(r#"return json.decode('{"name":"test","count":42}')"#) + .eval() + .unwrap(); + assert_eq!(result.get::("name").unwrap(), "test"); + assert_eq!(result.get::("count").unwrap(), 42); + } + + #[test] + fn sandbox_json_encode_array() { + let lua = create_sandbox().unwrap(); + let result: String = lua + .load(r#"return json.encode(toarray({1, 2, 3}))"#) + .eval() + .unwrap(); + assert_eq!(result, "[1,2,3]"); + } + + #[test] + fn sandbox_json_decode_invalid_returns_error() { + let lua = create_sandbox().unwrap(); + let result: Result = + lua.load(r#"return json.decode("not valid json")"#).eval(); + assert!(result.is_err()); } } diff --git a/src/main.rs b/src/main.rs --- a/src/main.rs +++ b/src/main.rs @@ -56,6 +56,7 @@ 1, target_collection.clone(), ProcedureAction::Upsert, None, + None, ) { Ok(parsed) => { if let Err(e) = sqlx::query( @@ -127,14 +128,7 @@ warn!("failed to sync initial collections to tap: {e}"); } } - tap::spawn( - state.db.clone(), - config.tap_url.clone(), - config.tap_admin_password.clone(), - collections_rx, - state.lexicons.clone(), - state.collections_tx.clone(), - ); + tap::spawn(state.clone(), collections_rx); tokio::spawn(happyview::event_log::spawn_retention_cleanup( state.db.clone(), diff --git a/src/tap.rs b/src/tap.rs --- a/src/tap.rs +++ b/src/tap.rs @@ -1,13 +1,13 @@ use futures_util::{SinkExt, StreamExt}; use serde::{Deserialize, Serialize}; use serde_json::Value; -use sqlx::PgPool; use tokio::sync::watch; use tokio_tungstenite::tungstenite::Message; use tokio_tungstenite::tungstenite::client::IntoClientRequest; +use crate::AppState; use crate::event_log::{EventLog, Severity, log_event}; -use crate::lexicon::{LexiconRegistry, LexiconType, ParsedLexicon, ProcedureAction}; +use crate::lexicon::{LexiconType, ParsedLexicon, ProcedureAction}; // --------------------------------------------------------------------------- // Tap event types (matches Tap's outbox JSON format) @@ -229,33 +229,13 @@ /// 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(); - +pub fn spawn(state: AppState, mut collections_rx: watch::Receiver>) { tokio::spawn(async move { loop { // Build WebSocket URL from HTTP URL. - let ws_url = build_ws_url(&tap_url); + let ws_url = build_ws_url(&state.config.tap_url); - match run( - &db, - &http, - &tap_url, - tap_admin_password.as_deref(), - &ws_url, - &mut collections_rx, - &lexicons, - &collections_tx, - ) - .await - { + match run(&state, &ws_url, &mut collections_rx).await { Ok(()) => { tracing::info!("tap reconnecting due to collection change"); } @@ -285,17 +265,16 @@ // --------------------------------------------------------------------------- // Connection loop // --------------------------------------------------------------------------- -#[allow(clippy::too_many_arguments)] async fn run( - db: &PgPool, - http: &reqwest::Client, - tap_url: &str, - tap_admin_password: Option<&str>, + state: &AppState, ws_url: &str, collections_rx: &mut watch::Receiver>, - lexicons: &LexiconRegistry, - collections_tx: &watch::Sender>, ) -> Result<(), Box> { + let db = &state.db; + let http = &state.http; + let tap_url = &state.config.tap_url; + let tap_admin_password = state.config.tap_admin_password.as_deref(); + tracing::info!(url = %ws_url, "connecting to tap"); let mut request = ws_url.to_string().into_client_request()?; @@ -370,7 +349,7 @@ match event.event_type.as_str() { "record" => { if let Some(record) = event.record { - handle_record_event(db, lexicons, collections_tx, &record).await; + handle_record_event(state, &record).await; } } "identity" => { @@ -431,17 +410,15 @@ // --------------------------------------------------------------------------- // Record event handler // --------------------------------------------------------------------------- -async fn handle_record_event( - db: &PgPool, - lexicons: &LexiconRegistry, - collections_tx: &watch::Sender>, - record: &TapRecordEvent, -) { +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(db, lexicons, collections_tx, &record.did, record).await; + handle_lexicon_schema_event(state, &record.did, record).await; return; } @@ -502,6 +479,34 @@ }), }, ) .await; + + // Fire index hook if configured. + if let Some(script) = + state.lexicons.get_on_index_script(&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}"); @@ -546,6 +551,32 @@ }), }, ) .await; + + // Fire index hook if configured. + if let Some(script) = + state.lexicons.get_on_index_script(&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}"); @@ -577,13 +608,10 @@ // 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, -) { +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; // Check if this NSID is one we're tracking and the DID matches the authority. @@ -613,6 +641,7 @@ rec.clone(), 1, target_collection.clone(), ProcedureAction::Upsert, + None, None, ) { Ok(p) => p,