Something went wrong. Try again.
A lexicon-driven AppView for ATProto.
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268use axum::Json;use axum::response::{IntoResponse, Response};use mlua::LuaSerdeExt;use serde_json::Value;use std::collections::HashMap;use std::sync::Arc;use std::time::Instant;
use crate::AppState;use crate::auth::Claims;use crate::db::{DatabaseBackend, adapt_sql, now_rfc3339};use crate::error::{AppError, ScriptErrorType, parse_lua_line};use crate::event_log::{EventLog, Severity, log_event};use crate::lexicon::ParsedLexicon;use crate::repo;
use super::atproto_api;use super::context;use super::db_api;use super::http_api;use super::record;use super::sandbox;
/// Load all script variables from the database as a key-value map.async fn load_env_vars(db: &sqlx::AnyPool, backend: DatabaseBackend) -> HashMap<String, String> { let sql = adapt_sql("SELECT key, value FROM script_variables", backend); sqlx::query_as::<_, (String, String)>(&sql) .fetch_all(db) .await .unwrap_or_default() .into_iter() .collect()}
/// Execute a Lua script for a procedure endpoint.pub async fn execute_procedure_script( state: &AppState, method: &str, claims: &Claims, input: &Value, params: &std::collections::HashMap<String, Value>, lexicon: &ParsedLexicon, script: &str,) -> Result<Response, AppError> { let start = Instant::now(); let backend = state.db_backend; let span = tracing::info_span!( "script.execute", method = method, script_type = "procedure", caller_did = %claims.did(), ); span.in_scope(|| tracing::info!("script execution started")); let collection = lexicon.target_collection.as_deref().unwrap_or_default();
// Capture script source and input for error logging before anything is consumed. let script_source = script.to_string(); let input_json = input.clone();
let pds_auth = if let Some(client_key) = claims.client_key() { let encryption_key = state .config .token_encryption_key .as_ref() .ok_or_else(|| AppError::Internal("TOKEN_ENCRYPTION_KEY not configured".into()))?; let api_client_id = match repo::get_dpop_client_id(state, client_key).await { Ok(id) => id, Err(e) => { let error_message = format!("{e}"); log_event( &state.db, EventLog { event_type: "script.error".to_string(), severity: Severity::Error, actor_did: Some(claims.did().to_string()), subject: Some(method.to_string()), detail: serde_json::json!({ "error": error_message, "script_source": script_source, "input": input_json, "caller_did": claims.did(), "method": method, "duration_ms": start.elapsed().as_millis() as u64, }), }, backend, ) .await; return Err(e); } }; repo::PdsAuth::Dpop { api_client_id, encryption_key: *encryption_key, } } else { match repo::get_oauth_session(state, claims.did()).await { Ok(s) => repo::PdsAuth::OAuth(Arc::new(s)), Err(e) => { let error_message = format!("{e}"); log_event( &state.db, EventLog { event_type: "script.error".to_string(), severity: Severity::Error, actor_did: Some(claims.did().to_string()), subject: Some(method.to_string()), detail: serde_json::json!({ "error": error_message, "script_source": script_source, "input": input_json, "caller_did": claims.did(), "method": method, "duration_ms": start.elapsed().as_millis() as u64, }), }, backend, ) .await; return Err(e); } } };
let lua = match sandbox::create_sandbox() { Ok(l) => l, Err(e) => { let error_message = format!("failed to create Lua VM: {e}"); log_event( &state.db, EventLog { event_type: "script.error".to_string(), severity: Severity::Error, actor_did: Some(claims.did().to_string()), subject: Some(method.to_string()), detail: serde_json::json!({ "error": error_message, "script_source": script_source, "input": input_json, "caller_did": claims.did(), "method": method, "duration_ms": start.elapsed().as_millis() as u64, }), }, backend, ) .await; return Err(AppError::Internal(error_message)); } };
let state_arc = Arc::new(state.clone()); let claims_arc = Arc::new(claims.clone()); let pds_auth_arc = Arc::new(pds_auth);
if let Err(e) = db_api::register_db_api(&lua, state_arc.clone()) { let error_message = format!("failed to register db API: {e}"); log_event( &state.db, EventLog { event_type: "script.error".to_string(), severity: Severity::Error, actor_did: Some(claims.did().to_string()), subject: Some(method.to_string()), detail: serde_json::json!({ "error": error_message, "script_source": script_source, "input": input_json, "caller_did": claims.did(), "method": method, "duration_ms": start.elapsed().as_millis() as u64, }), }, backend, ) .await; return Err(AppError::Internal(error_message)); }
if let Err(e) = http_api::register_http_api(&lua, state_arc.clone()) { let error_message = format!("failed to register http API: {e}"); log_event( &state.db, EventLog { event_type: "script.error".to_string(), severity: Severity::Error, actor_did: Some(claims.did().to_string()), subject: Some(method.to_string()), detail: serde_json::json!({ "error": error_message, "script_source": script_source, "input": input_json, "caller_did": claims.did(), "method": method, "duration_ms": start.elapsed().as_millis() as u64, }), }, backend, ) .await; return Err(AppError::Internal(error_message)); }
if let Err(e) = super::xrpc_api::register_xrpc_api(&lua, state_arc.clone(), Some(claims.did().to_string())) { let error_message = format!("failed to register xrpc API: {e}"); log_event( &state.db, EventLog { event_type: "script.error".to_string(), severity: Severity::Error, actor_did: Some(claims.did().to_string()), subject: Some(method.to_string()), detail: serde_json::json!({ "error": error_message, "script_source": script_source, "input": input_json, "caller_did": claims.did(), "method": method, "duration_ms": start.elapsed().as_millis() as u64, }), }, backend, ) .await; return Err(AppError::Internal(error_message)); }
if let Err(e) = atproto_api::register_atproto_api(&lua, state_arc.clone(), Some(claims.did())) { let error_message = format!("failed to register atproto API: {e}"); log_event( &state.db, EventLog { event_type: "script.error".to_string(), severity: Severity::Error, actor_did: Some(claims.did().to_string()), subject: Some(method.to_string()), detail: serde_json::json!({ "error": error_message, "script_source": script_source, "input": input_json, "caller_did": claims.did(), "method": method, "duration_ms": start.elapsed().as_millis() as u64, }), }, backend, ) .await; return Err(AppError::Internal(error_message)); }
if let Err(e) = record::register_record_api(&lua, state_arc, claims_arc, pds_auth_arc) { let error_message = format!("failed to register Record API: {e}"); log_event( &state.db, EventLog { event_type: "script.error".to_string(), severity: Severity::Error, actor_did: Some(claims.did().to_string()), subject: Some(method.to_string()), detail: serde_json::json!({ "error": error_message, "script_source": script_source, "input": input_json, "caller_did": claims.did(), "method": method, "duration_ms": start.elapsed().as_millis() as u64, }), }, backend, ) .await; return Err(AppError::Internal(error_message)); }
if let Err(e) = context::set_procedure_context(&lua, method, input, params, claims.did(), collection) { let error_message = format!("failed to set context: {e}"); log_event( &state.db, EventLog { event_type: "script.error".to_string(), severity: Severity::Error, actor_did: Some(claims.did().to_string()), subject: Some(method.to_string()), detail: serde_json::json!({ "error": error_message, "script_source": script_source, "input": input_json, "caller_did": claims.did(), "method": method, "duration_ms": start.elapsed().as_millis() as u64, }), }, backend, ) .await; return Err(AppError::Internal(error_message)); }
if let Err(e) = context::set_env_context(&lua, &load_env_vars(&state.db, backend).await) { let error_message = format!("failed to set env context: {e}"); log_event( &state.db, EventLog { event_type: "script.error".to_string(), severity: Severity::Error, actor_did: Some(claims.did().to_string()), subject: Some(method.to_string()), detail: serde_json::json!({ "error": error_message, "script_source": script_source, "input": input_json, "caller_did": claims.did(), "method": method, "duration_ms": start.elapsed().as_millis() as u64, }), }, backend, ) .await; return Err(AppError::Internal(error_message)); }
if let Err(e) = lua.load(script).exec() { let error_message = format!("{e}"); tracing::error!(method, error = %e, "lua script load failed"); log_event( &state.db, EventLog { event_type: "script.error".to_string(), severity: Severity::Error, actor_did: Some(claims.did().to_string()), subject: Some(method.to_string()), detail: serde_json::json!({ "error": error_message, "script_source": script_source, "input": input_json, "caller_did": claims.did(), "method": method, "duration_ms": start.elapsed().as_millis() as u64, }), }, backend, ) .await; let (line, clean_msg) = parse_lua_line(&error_message); return Err(AppError::ScriptError { error_type: ScriptErrorType::Syntax, message: clean_msg, method: method.to_string(), line, }); }
let handle: mlua::Function = match lua.globals().get("handle") { Ok(f) => f, Err(e) => { let error_message = format!("{e}"); tracing::error!(method, error = %e, "lua script missing handle function"); log_event( &state.db, EventLog { event_type: "script.error".to_string(), severity: Severity::Error, actor_did: Some(claims.did().to_string()), subject: Some(method.to_string()), detail: serde_json::json!({ "error": error_message, "script_source": script_source, "input": input_json, "caller_did": claims.did(), "method": method, "duration_ms": start.elapsed().as_millis() as u64, }), }, backend, ) .await; return Err(AppError::ScriptError { error_type: ScriptErrorType::MissingHandle, message: "script does not define a handle() function".to_string(), method: method.to_string(), line: None, }); } };
let result: mlua::Value = match handle.call_async(()).await { Ok(r) => r, Err(e) => { let msg = e.to_string(); tracing::error!(method, error = %msg, "lua script execution failed"); let (line, clean_msg) = parse_lua_line(&msg); let app_error = if msg.contains("execution limit") { AppError::ScriptError { error_type: ScriptErrorType::Timeout, message: "script exceeded execution time limit".to_string(), method: method.to_string(), line, } } else { AppError::ScriptError { error_type: ScriptErrorType::Runtime, message: clean_msg, method: method.to_string(), line, } }; log_event( &state.db, EventLog { event_type: "script.error".to_string(), severity: Severity::Error, actor_did: Some(claims.did().to_string()), subject: Some(method.to_string()), detail: serde_json::json!({ "error": msg, "script_source": script_source, "input": input_json, "caller_did": claims.did(), "method": method, "duration_ms": start.elapsed().as_millis() as u64, }), }, backend, ) .await; return Err(app_error); } };
let json_value: Value = match lua.from_value(result) { Ok(v) => v, Err(e) => { let error_message = format!("{e}"); tracing::error!(method, error = %e, "failed to convert lua result to JSON"); log_event( &state.db, EventLog { event_type: "script.error".to_string(), severity: Severity::Error, actor_did: Some(claims.did().to_string()), subject: Some(method.to_string()), detail: serde_json::json!({ "error": error_message, "script_source": script_source, "input": input_json, "caller_did": claims.did(), "method": method, "duration_ms": start.elapsed().as_millis() as u64, }), }, backend, ) .await; return Err(AppError::ScriptError { error_type: ScriptErrorType::Runtime, message: error_message, method: method.to_string(), line: None, }); } };
span.in_scope(|| { tracing::info!( duration_ms = start.elapsed().as_millis() as u64, "script execution completed" ); }); log_event( &state.db, EventLog { event_type: "script.executed".to_string(), severity: Severity::Info, actor_did: Some(claims.did().to_string()), subject: Some(method.to_string()), detail: serde_json::json!({ "method": method, "caller_did": claims.did(), "duration_ms": start.elapsed().as_millis() as u64, "response_size": json_value.to_string().len(), "input": input_json, "response": json_value, }), }, backend, ) .await;
Ok(Json(json_value).into_response())}
/// Execute a Lua script for a query endpoint.pub async fn execute_query_script( state: &AppState, method: &str, params: &HashMap<String, serde_json::Value>, lexicon: &ParsedLexicon, script: &str, claims: Option<&Claims>,) -> Result<Response, AppError> { let start = Instant::now(); let backend = state.db_backend; let span = tracing::info_span!("script.execute", method = method, script_type = "query",); span.in_scope(|| tracing::info!("script execution started")); let collection = lexicon.target_collection.as_deref().unwrap_or_default();
// Capture script source for error logging. let script_source = script.to_string();
let lua = match sandbox::create_sandbox() { Ok(l) => l, Err(e) => { let error_message = format!("failed to create Lua VM: {e}"); log_event( &state.db, EventLog { event_type: "script.error".to_string(), severity: Severity::Error, actor_did: None, subject: Some(method.to_string()), detail: serde_json::json!({ "error": error_message, "script_source": script_source, "method": method, "duration_ms": start.elapsed().as_millis() as u64, }), }, backend, ) .await; return Err(AppError::Internal(error_message)); } };
let state_arc = Arc::new(state.clone());
if let Err(e) = db_api::register_db_api(&lua, state_arc.clone()) { let error_message = format!("failed to register db API: {e}"); log_event( &state.db, EventLog { event_type: "script.error".to_string(), severity: Severity::Error, actor_did: None, subject: Some(method.to_string()), detail: serde_json::json!({ "error": error_message, "script_source": script_source, "method": method, "duration_ms": start.elapsed().as_millis() as u64, }), }, backend, ) .await; return Err(AppError::Internal(error_message)); }
if let Err(e) = http_api::register_http_api(&lua, state_arc.clone()) { let error_message = format!("failed to register http API: {e}"); log_event( &state.db, EventLog { event_type: "script.error".to_string(), severity: Severity::Error, actor_did: None, subject: Some(method.to_string()), detail: serde_json::json!({ "error": error_message, "script_source": script_source, "method": method, "duration_ms": start.elapsed().as_millis() as u64, }), }, backend, ) .await; return Err(AppError::Internal(error_message)); }
if let Err(e) = super::xrpc_api::register_xrpc_api( &lua, state_arc.clone(), claims.map(|c| c.did().to_string()), ) { let error_message = format!("failed to register xrpc API: {e}"); log_event( &state.db, EventLog { event_type: "script.error".to_string(), severity: Severity::Error, actor_did: None, subject: Some(method.to_string()), detail: serde_json::json!({ "error": error_message, "script_source": script_source, "method": method, "duration_ms": start.elapsed().as_millis() as u64, }), }, backend, ) .await; return Err(AppError::Internal(error_message)); }
if let Err(e) = atproto_api::register_atproto_api(&lua, state_arc, claims.map(|c| c.did())) { let error_message = format!("failed to register atproto API: {e}"); log_event( &state.db, EventLog { event_type: "script.error".to_string(), severity: Severity::Error, actor_did: None, subject: Some(method.to_string()), detail: serde_json::json!({ "error": error_message, "script_source": script_source, "method": method, "duration_ms": start.elapsed().as_millis() as u64, }), }, backend, ) .await; return Err(AppError::Internal(error_message)); }
if let Err(e) = context::set_query_context(&lua, method, params, collection, claims.map(|c| c.did())) { let error_message = format!("failed to set context: {e}"); log_event( &state.db, EventLog { event_type: "script.error".to_string(), severity: Severity::Error, actor_did: None, subject: Some(method.to_string()), detail: serde_json::json!({ "error": error_message, "script_source": script_source, "method": method, "duration_ms": start.elapsed().as_millis() as u64, }), }, backend, ) .await; return Err(AppError::Internal(error_message)); }
if let Err(e) = context::set_env_context(&lua, &load_env_vars(&state.db, backend).await) { let error_message = format!("failed to set env context: {e}"); log_event( &state.db, EventLog { event_type: "script.error".to_string(), severity: Severity::Error, actor_did: None, subject: Some(method.to_string()), detail: serde_json::json!({ "error": error_message, "script_source": script_source, "method": method, "duration_ms": start.elapsed().as_millis() as u64, }), }, backend, ) .await; return Err(AppError::Internal(error_message)); }
if let Err(e) = lua.load(script).exec() { let error_message = format!("{e}"); tracing::error!(method, error = %e, "lua script load failed"); log_event( &state.db, EventLog { event_type: "script.error".to_string(), severity: Severity::Error, actor_did: None, subject: Some(method.to_string()), detail: serde_json::json!({ "error": error_message, "script_source": script_source, "method": method, "duration_ms": start.elapsed().as_millis() as u64, }), }, backend, ) .await; let (line, clean_msg) = parse_lua_line(&error_message); return Err(AppError::ScriptError { error_type: ScriptErrorType::Syntax, message: clean_msg, method: method.to_string(), line, }); }
let handle: mlua::Function = match lua.globals().get("handle") { Ok(f) => f, Err(e) => { let error_message = format!("{e}"); tracing::error!(method, error = %e, "lua script missing handle function"); log_event( &state.db, EventLog { event_type: "script.error".to_string(), severity: Severity::Error, actor_did: None, subject: Some(method.to_string()), detail: serde_json::json!({ "error": error_message, "script_source": script_source, "method": method, "duration_ms": start.elapsed().as_millis() as u64, }), }, backend, ) .await; return Err(AppError::ScriptError { error_type: ScriptErrorType::MissingHandle, message: "script does not define a handle() function".to_string(), method: method.to_string(), line: None, }); } };
let result: mlua::Value = match handle.call_async(()).await { Ok(r) => r, Err(e) => { let msg = e.to_string(); tracing::error!(method, error = %msg, "lua script execution failed"); let (line, clean_msg) = parse_lua_line(&msg); let app_error = if msg.contains("execution limit") { AppError::ScriptError { error_type: ScriptErrorType::Timeout, message: "script exceeded execution time limit".to_string(), method: method.to_string(), line, } } else { AppError::ScriptError { error_type: ScriptErrorType::Runtime, message: clean_msg, method: method.to_string(), line, } }; log_event( &state.db, EventLog { event_type: "script.error".to_string(), severity: Severity::Error, actor_did: None, subject: Some(method.to_string()), detail: serde_json::json!({ "error": msg, "script_source": script_source, "method": method, "duration_ms": start.elapsed().as_millis() as u64, }), }, backend, ) .await; return Err(app_error); } };
let json_value: Value = match lua.from_value(result) { Ok(v) => v, Err(e) => { let error_message = format!("{e}"); tracing::error!(method, error = %e, "failed to convert lua result to JSON"); log_event( &state.db, EventLog { event_type: "script.error".to_string(), severity: Severity::Error, actor_did: None, subject: Some(method.to_string()), detail: serde_json::json!({ "error": error_message, "script_source": script_source, "method": method, "duration_ms": start.elapsed().as_millis() as u64, }), }, backend, ) .await; return Err(AppError::ScriptError { error_type: ScriptErrorType::Runtime, message: error_message, method: method.to_string(), line: None, }); } };
span.in_scope(|| { tracing::info!( duration_ms = start.elapsed().as_millis() as u64, "script execution completed" ); }); log_event( &state.db, EventLog { event_type: "script.executed".to_string(), severity: Severity::Info, actor_did: None, subject: Some(method.to_string()), detail: serde_json::json!({ "method": method, "duration_ms": start.elapsed().as_millis() as u64, "response_size": json_value.to_string().len(), "params": params, "response": json_value, }), }, backend, ) .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.////// 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, 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<Value> { let max_attempts: i32 = 4; // 1 initial + 3 retries let mut last_error = String::new(); let backend = event.state.db_backend;
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(hook_result) => { 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, }), }, backend, ) .await; return hook_result; } 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 and fail-open with the // original record so indexing is not permanently blocked. tracing::error!( uri = event.uri, lexicon_id = event.lexicon_id, "hook dead-lettered after {max_attempts} attempts" );
let record_str = event .record .map(|r| serde_json::to_string(r).unwrap_or_default()); let dead_letter_sql = adapt_sql( r#" INSERT INTO dead_letter_hooks (lexicon_id, uri, did, collection, rkey, action, record, error, attempts, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) "#, backend, ); if let Err(e) = sqlx::query(&dead_letter_sql) .bind(event.lexicon_id) .bind(event.uri) .bind(event.did) .bind(event.collection) .bind(event.rkey) .bind(event.action) .bind(&record_str) .bind(&last_error) .bind(max_attempts) .bind(now_rfc3339()) .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, }), }, backend, ) .await;
// Fail-open: return the original record so indexing proceeds. event.record.cloned()}
/// 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.pub async fn run_hook_once(event: &HookEvent<'_>) -> Result<Option<Value>, String> { let lua = sandbox::create_sandbox().map_err(|e| format!("failed to create Lua VM: {e}"))?; let backend = event.state.db_backend;
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.clone()) .map_err(|e| format!("failed to register http API: {e}"))?;
super::xrpc_api::register_xrpc_api(&lua, state_arc.clone(), Some(event.did.to_string())) .map_err(|e| format!("failed to register xrpc API: {e}"))?;
atproto_api::register_atproto_api(&lua, state_arc, None) .map_err(|e| format!("failed to register atproto 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}"))?;
context::set_env_context(&lua, &load_env_vars(&event.state.db, backend).await) .map_err(|e| format!("failed to set env 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}"))?;
let result: mlua::Value = handle .call_async::<mlua::Value>(()) .await .map_err(|e| e.to_string())?;
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)]mod tests { use super::*; use crate::config::Config; use crate::db::DatabaseBackend; use crate::lexicon::LexiconRegistry; use serde_json::json; use tokio::sync::watch;
fn test_state() -> AppState { let config = Config { host: "127.0.0.1".into(), port: 3000, database_url: String::new(), database_backend: crate::db::DatabaseBackend::Sqlite, public_url: String::new(), session_secret: "test-secret".into(), jetstream_url: String::new(), relay_url: String::new(), plc_url: String::new(), static_dir: String::new(), event_log_retention_days: 30, app_name: None, logo_uri: None, tos_uri: None, policy_uri: None, token_encryption_key: None, default_rate_limit_capacity: 100, default_rate_limit_refill_rate: 2.0, }; let (tx, _) = watch::channel(vec![]); let (labeler_tx, _) = watch::channel(()); sqlx::any::install_default_drivers(); let test_db = sqlx::AnyPool::connect_lazy("sqlite::memory:").unwrap(); let atrium_http = std::sync::Arc::new(atrium_oauth::DefaultHttpClient::default()); let did_resolver = atrium_identity::did::CommonDidResolver::new( atrium_identity::did::CommonDidResolverConfig { plc_directory_url: "https://plc.directory".into(), http_client: std::sync::Arc::clone(&atrium_http), }, ); let handle_resolver = atrium_identity::handle::AtprotoHandleResolver::new( atrium_identity::handle::AtprotoHandleResolverConfig { dns_txt_resolver: crate::dns::NativeDnsResolver::new(), http_client: atrium_http, }, ); let oauth = atrium_oauth::OAuthClient::new(atrium_oauth::OAuthClientConfig { client_metadata: atrium_oauth::AtprotoLocalhostClientMetadata { redirect_uris: Some(vec!["http://127.0.0.1:0/auth/callback".into()]), scopes: Some(vec![atrium_oauth::Scope::Known( atrium_oauth::KnownScope::Atproto, )]), }, keys: None, state_store: crate::auth::oauth_store::DbStateStore::new( test_db.clone(), crate::db::DatabaseBackend::Sqlite, ), session_store: crate::auth::oauth_store::DbSessionStore::new( test_db.clone(), crate::db::DatabaseBackend::Sqlite, ), resolver: atrium_oauth::OAuthResolverConfig { did_resolver, handle_resolver, authorization_server_metadata: Default::default(), protected_resource_metadata: Default::default(), }, }) .expect("Failed to create test OAuth client"); AppState { config, http: reqwest::Client::new(), db: test_db.clone(), db_backend: DatabaseBackend::Sqlite, domain_cache: crate::domain::DomainCache::new(), lexicons: LexiconRegistry::new(), collections_tx: tx, labeler_subscriptions_tx: labeler_tx, rate_limiter: crate::rate_limit::RateLimiter::new( crate::rate_limit::RateLimitDefaults { query_cost: 1, procedure_cost: 1, proxy_cost: 1, }, ), oauth: std::sync::Arc::new(crate::auth::OAuthClientRegistry::new(std::sync::Arc::new( oauth, ))), oauth_state_store: crate::auth::oauth_store::DbStateStore::new( test_db.clone(), crate::db::DatabaseBackend::Sqlite, ), cookie_key: axum_extra::extract::cookie::Key::derive_from( b"test-secret-for-tests-only-not-production", ), plugin_registry: std::sync::Arc::new(crate::plugin::PluginRegistry::new()), wasm_runtime: std::sync::Arc::new( crate::plugin::WasmRuntime::new().expect("wasm runtime"), ), attestation_signer: None, official_registry: std::sync::Arc::new(tokio::sync::RwLock::new( crate::plugin::official_registry::OfficialRegistryState::default(), )), official_registry_config: crate::plugin::official_registry::RegistryConfig::production( ), } }
fn make_event<'a>( state: &'a AppState, script: &'a str, action: &'a str, record: Option<&'a Value>, ) -> HookEvent<'a> { HookEvent { state, lexicon_id: "test.lexicon", script, action, uri: "at://did:plc:test/test.collection/rkey1", did: "did:plc:test", collection: "test.collection", rkey: "rkey1", record, } }
#[tokio::test] async fn hook_runs_simple_script() { let state = test_state(); 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] async fn hook_fails_on_missing_handle() { let state = test_state(); let event = make_event(&state, "function other() end", "create", None); let result = run_hook_once(&event).await; assert!(result.is_err()); let err = result.unwrap_err(); assert!(err.contains("handle"), "expected handle error, got: {err}"); }
#[tokio::test] async fn hook_fails_on_syntax_error() { let state = test_state(); let event = make_event(&state, "function handle(", "create", None); let result = run_hook_once(&event).await; assert!(result.is_err()); }
#[tokio::test] async fn hook_has_access_to_context_globals() { let state = test_state(); let script = r#" function handle() if action ~= "create" then error("wrong action: " .. tostring(action)) end if uri ~= "at://did:plc:test/test.collection/rkey1" then error("wrong uri") end if did ~= "did:plc:test" then error("wrong did") end if collection ~= "test.collection" then error("wrong collection") end if rkey ~= "rkey1" then error("wrong rkey") end end "#; let event = make_event(&state, script, "create", None); let result = run_hook_once(&event).await; assert!(result.is_ok(), "expected Ok, got: {:?}", result); }
#[tokio::test] async fn hook_has_access_to_record() { let state = test_state(); let record = json!({"name": "Test"}); let script = r#" function handle() if record.name ~= "Test" then error("wrong name: " .. tostring(record.name)) end 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); }
#[tokio::test] async fn hook_record_nil_on_delete() { let state = test_state(); let script = r#" function handle() if record ~= nil then error("expected nil record") end end "#; let event = make_event(&state, script, "delete", None); let result = run_hook_once(&event).await; assert!(result.is_ok(), "expected Ok, got: {:?}", result); }}