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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353use axum::Json;use axum::response::{IntoResponse, Response};use mlua::LuaSerdeExt;use serde_json::Value;use std::collections::HashMap;use std::sync::Arc;use std::sync::atomic::Ordering;use std::time::Instant;
use crate::AppState;use crate::auth::Claims;use crate::db::{DatabaseBackend, adapt_sql};use crate::error::{AppError, LUA_AUTH_ERROR_PREFIX, ScriptErrorType, parse_lua_line};use crate::event_log::{EventLog, Severity, log_event};use crate::lexicon::ParsedLexicon;use crate::repo;use crate::telemetry::counters::Counters;
use super::atproto_api;use super::context;use super::db_api;use super::http_api;use super::record;use super::sandbox;
struct ScriptTimingGuard { counters: Arc<Counters>, start: Instant,}
impl Drop for ScriptTimingGuard { fn drop(&mut self) { let elapsed_us = u64::try_from(self.start.elapsed().as_micros()).unwrap_or(u64::MAX); crate::telemetry::counters::add_saturating(&self.counters.script_runtime_us, elapsed_us); self.counters .script_executions .fetch_add(1, Ordering::Relaxed); }}
/// 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 happyview_script_variables", backend); crate::db::query_as::<(String, String)>(&sql) .fetch_all(db) .await .unwrap_or_default() .into_iter() .collect()}
/// Execute a Lua script for a procedure endpoint.#[allow(clippy::too_many_arguments)]pub async fn execute_procedure_script( state: &AppState, method: &str, claims: &Claims, input: &Value, params: &std::collections::HashMap<String, Value>, lexicon: &ParsedLexicon, script: &str, space_ctx: Option<&context::SpaceContext>, delegate_did: Option<&str>,) -> Result<Response, AppError> { let start = Instant::now(); let _script_timing = ScriptTimingGuard { counters: state.telemetry_counters.clone(), start, }; 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: Option<repo::PdsAuth> = 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); } }; let dpop_key_id = claims .dpop_key_id() .ok_or_else(|| AppError::Internal("DPoP key ID not available in claims".into()))? .to_string(); Some(repo::PdsAuth::Dpop { api_client_id, dpop_key_id, encryption_key: *encryption_key, }) } else { repo::get_oauth_session(state, claims.did()) .await .ok() .map(|s| repo::PdsAuth::OAuth(Arc::new(s))) };
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 = pds_auth.map(Arc::new);
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) = crate::lua::spaces_api::register_spaces_write_api( &lua, state_arc.clone(), Some(claims.did()), ) { let error_message = format!("failed to register spaces write 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) = crate::lua::linked_repos_api::register_linked_repos_api(&lua, state_arc.clone()) { let error_message = format!("failed to register linked repos 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 Some(ref pds_auth) = pds_auth_arc && let Err(e) = atproto_api::register_atproto_blob_api( &lua, state_arc.clone(), claims_arc.clone(), pds_auth.clone(), ) { let error_message = format!("failed to register atproto blob 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::jobs_api::register_jobs_api( &lua, state_arc.clone(), Some(super::jobs_api::JobsCaller { did: claims.did().to_string(), api_client_id: pds_auth_arc.as_ref().and_then(|a| match a.as_ref() { repo::PdsAuth::Dpop { api_client_id, .. } => Some(api_client_id.clone()), _ => None, }), dpop_key_id: pds_auth_arc.as_ref().and_then(|a| match a.as_ref() { repo::PdsAuth::Dpop { dpop_key_id, .. } => Some(dpop_key_id.clone()), _ => None, }), }), ) { let error_message = format!("failed to register jobs 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.clone(), Some(claims_arc), pds_auth_arc, delegate_did.map(|s| s.to_string()), ) { 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)); }
// Override the sandbox's tracing-only `log()` with a version that // also writes a `script.log` row to `event_logs` so operators can // see script output from the dashboard. The xrpc trigger id is // computed from the lexicon's id + procedure type. let trigger_id = format!("xrpc.procedure:{}", lexicon.id); if let Err(e) = super::scripts::register_log_event_api(&lua, &state_arc, &trigger_id, Some(claims.did())) { let error_message = format!("failed to register log 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, space_ctx, delegate_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: 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(LUA_AUTH_ERROR_PREFIX) || clean_msg.contains(LUA_AUTH_ERROR_PREFIX) { let auth_msg = clean_msg .strip_prefix(LUA_AUTH_ERROR_PREFIX) .unwrap_or(&clean_msg) .to_string(); AppError::Auth(auth_msg) } else 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" ); }); if state.verbose_event_logging.load(Ordering::Relaxed) { 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>, space_ctx: Option<&context::SpaceContext>,) -> Result<Response, AppError> { let start = Instant::now(); let _script_timing = ScriptTimingGuard { counters: state.telemetry_counters.clone(), start, }; 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.clone(), 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) = crate::lua::spaces_api::register_spaces_write_api( &lua, state_arc.clone(), claims.map(|c| c.did()), ) { let error_message = format!("failed to register spaces write 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) = crate::lua::linked_repos_api::register_linked_repos_api(&lua, state_arc.clone()) { let error_message = format!("failed to register linked repos 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)); }
// Register the Record API in no-auth mode. Queries don't have a PDS // auth context — the local-only methods (Record.load, :save_local, // :delete_local, Record.delete_local) work; PDS-touching variants // error with the no-PDS-auth message. if let Err(e) = record::register_record_api_no_auth(&lua, state_arc.clone()) { 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: 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)); }
// Override the sandbox's tracing-only `log()` with a version that // also writes a `script.log` row to `event_logs`. let trigger_id = format!("xrpc.query:{}", lexicon.id); if let Err(e) = super::scripts::register_log_event_api( &lua, &state_arc, &trigger_id, claims.map(|c| c.did()), ) { let error_message = format!("failed to register log 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()), space_ctx, ) { 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(LUA_AUTH_ERROR_PREFIX) || clean_msg.contains(LUA_AUTH_ERROR_PREFIX) { let auth_msg = clean_msg .strip_prefix(LUA_AUTH_ERROR_PREFIX) .unwrap_or(&clean_msg) .to_string(); AppError::Auth(auth_msg) } else 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" ); }); if state.verbose_event_logging.load(Ordering::Relaxed) { 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())}
#[cfg(test)]mod tests { use super::*; use crate::lexicon::{LexiconType, ProcedureAction}; use crate::test_support::{memory_pool, migrated_memory_pool, test_state_with_pool};
fn query_lexicon() -> ParsedLexicon { ParsedLexicon { id: "com.example.probe".to_string(), lexicon_type: LexiconType::Query, record_key: None, parameters: None, input: None, output: None, record_schema: None, raw: serde_json::json!({ "id": "com.example.probe" }), revision: 1, target_collection: Some("com.example.probe".to_string()), action: ProcedureAction::Upsert, token_cost: None, space_type: None, space_name: None, space_collections: None, } }
#[tokio::test] async fn successful_script_execution_moves_the_script_counters() { let state = test_state_with_pool(memory_pool().await); let lexicon = query_lexicon(); let params = HashMap::new(); let counters = state.telemetry_counters.clone();
let result = execute_query_script( &state, "com.example.probe", ¶ms, &lexicon, "function handle() return { ok = true } end", None, None, ) .await;
assert!( result.is_ok(), "script should have executed: {:?}", result.err() ); assert_eq!(counters.script_executions.load(Ordering::Relaxed), 1); }
#[tokio::test] async fn a_script_missing_handle_still_moves_the_script_counters() { let state = test_state_with_pool(memory_pool().await); let lexicon = query_lexicon(); let params = HashMap::new(); let counters = state.telemetry_counters.clone();
let result = execute_query_script( &state, "com.example.probe", ¶ms, &lexicon, "local unused = 1", None, None, ) .await;
assert!( result.is_err(), "script has no handle() function, so this must error" ); assert_eq!(counters.script_executions.load(Ordering::Relaxed), 1); }
#[tokio::test] async fn a_lua_runtime_error_still_moves_the_script_counters() { // `handle()` runs and raises — a different early return than the // "missing handle" case (this one comes from `handle.call_async` // failing, not from `lua.globals().get("handle")` failing). let state = test_state_with_pool(memory_pool().await); let lexicon = query_lexicon(); let params = HashMap::new(); let counters = state.telemetry_counters.clone();
let result = execute_query_script( &state, "com.example.probe", ¶ms, &lexicon, "function handle() error('boom') end", None, None, ) .await;
assert!(result.is_err()); assert_eq!(counters.script_executions.load(Ordering::Relaxed), 1); }
#[tokio::test] async fn script_counters_accumulate_across_calls_on_the_same_counters() { let state = test_state_with_pool(memory_pool().await); let lexicon = query_lexicon(); let params = HashMap::new(); let counters = state.telemetry_counters.clone();
for _ in 0..20 { let _ = execute_query_script( &state, "com.example.probe", ¶ms, &lexicon, "function handle() return {} end", None, None, ) .await; }
assert_eq!(counters.script_executions.load(Ordering::Relaxed), 20); assert!( counters.script_runtime_us.load(Ordering::Relaxed) > 0, "20 script executions should accumulate measurable wall-clock time" ); }
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 }
/// Query and procedure scripts run on every XRPC call, and `script.executed` /// stores the full request and response, so the row is gated like the other /// per-request telemetry. #[tokio::test] async fn script_executed_is_not_logged_while_verbose_logging_is_off() { let state = test_state_with_pool(migrated_memory_pool().await); let lexicon = query_lexicon(); let params = HashMap::new();
let result = execute_query_script( &state, "com.example.probe", ¶ms, &lexicon, "function handle() return { ok = true } end", None, None, ) .await;
assert!( result.is_ok(), "script should have executed: {:?}", result.err() ); 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 = test_state_with_pool(migrated_memory_pool().await); state.verbose_event_logging.store(true, Ordering::Relaxed); let lexicon = query_lexicon(); let params = HashMap::new();
let result = execute_query_script( &state, "com.example.probe", ¶ms, &lexicon, "function handle() return { ok = true } end", None, None, ) .await;
assert!( result.is_ok(), "script should have executed: {:?}", result.err() ); assert_eq!( executed_event_count(&state).await, 1, "script.executed must still be logged when verbose event logging is on" ); }}