Something went wrong. Try again.
A lexicon-driven AppView for ATProto.
Something went wrong. Try again.
Rust
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022//! Trigger-keyed scripts dispatcher.//!//! Each row in the `scripts` table is identified by a TRIGGER STRING — the//! `id` column IS the binding. There's no separate "name" or "host column."//!//! Trigger grammar://!//! - `record.index:<nsid>` — fires for any record event on `<nsid>`//! (wildcard fallback).//! - `record.create:<nsid>` / `record.update:<nsid>` ///! `record.delete:<nsid>` — fires only for that specific action.//! - `xrpc.query:<nsid>` / `xrpc.procedure:<nsid>` — fires when the//! matching XRPC method is invoked.//! - `labeler.apply:<nsid>` — fires when a label arrives whose `uri`//! is `at://<did>/<nsid>/<rkey>`.//! - `labeler.apply:_actor` — fires when a label arrives whose `uri`//! is a bare DID (actor-level label).//!//! **Cascade for record events ONLY**: the dispatcher tries//! `record.<action>:<nsid>` first, falls back to `record.index:<nsid>`//! if no specific row exists. No cascade for XRPC or labeler triggers.//!//! Fail mode varies by host://!//! - **Record / label events**: fail-OPEN — a buggy script eats its//! retry budget then dead-letters; the upstream operation proceeds//! with whatever the dispatcher returns (original record / original//! label). The firehose has no caller to surface errors to.//! - **XRPC procedures / queries**: fail-CLOSED, single-shot — a script//! error becomes a 5xx response. The XRPC dispatchers in//! [`crate::xrpc`] resolve the script via [`resolve`] and call//! [`super::execute::execute_procedure_script`] ///! [`super::execute::execute_query_script`] directly.
use regex::Regex;use serde::{Deserialize, Serialize};use serde_json::Value;use std::sync::{Arc, LazyLock};
static JOB_TYPE_RE: LazyLock<Regex> = LazyLock::new(|| Regex::new(r"^[a-z0-9][a-z0-9._-]*$").unwrap());
use crate::AppState;use crate::db::{DatabaseBackend, adapt_sql, now_rfc3339};use crate::event_log::{EventLog, Severity, log_event};
use super::{atproto_api, context, db_api, http_api, record, sandbox, xrpc_api};
/// Number of attempts (1 initial + 3 retries) before dead-lettering.const MAX_ATTEMPTS: u32 = 4;
// ---------------------------------------------------------------------------// Trigger grammar// ---------------------------------------------------------------------------
/// Which family a trigger belongs to. Determines auth context, fail mode,/// and which event payload shape the script expects.#[derive(Clone, Copy, Debug, PartialEq, Eq)]pub enum TriggerKind { RecordIndex, RecordCreate, RecordUpdate, RecordDelete, XrpcQuery, XrpcProcedure, LabelerApply, JobRun,}
/// A trigger id parsed into `(kind, suffix)`. The suffix is either an NSID/// (`record.*`, `xrpc.*`, `labeler.apply:<nsid>`) or the literal `"_actor"`/// for `labeler.apply:_actor`.#[derive(Clone, Debug, PartialEq, Eq)]pub struct ParsedTrigger { pub kind: TriggerKind, pub suffix: String,}
impl ParsedTrigger { /// Reconstruct the canonical trigger id from `(kind, suffix)`. pub fn id(&self) -> String { match self.kind { TriggerKind::RecordIndex => format!("record.index:{}", self.suffix), TriggerKind::RecordCreate => format!("record.create:{}", self.suffix), TriggerKind::RecordUpdate => format!("record.update:{}", self.suffix), TriggerKind::RecordDelete => format!("record.delete:{}", self.suffix), TriggerKind::XrpcQuery => format!("xrpc.query:{}", self.suffix), TriggerKind::XrpcProcedure => format!("xrpc.procedure:{}", self.suffix), TriggerKind::LabelerApply => format!("labeler.apply:{}", self.suffix), TriggerKind::JobRun => format!("job.run:{}", self.suffix), } }
/// Parse a trigger id. Returns a structured error message naming the /// valid prefixes when the input doesn't match the grammar. pub fn parse(id: &str) -> Result<Self, String> { let (prefix, suffix) = id.split_once(':').ok_or_else(|| { format!( "trigger id '{id}' must contain a ':' separator; \ valid prefixes: record.{{index,create,update,delete}}:<nsid>, \ xrpc.{{query,procedure}}:<nsid>, labeler.apply:<nsid|_actor>, \ job.run:<type>" ) })?;
if suffix.is_empty() { return Err(format!("trigger id '{id}' has empty suffix")); }
let kind = match prefix { "record.index" => TriggerKind::RecordIndex, "record.create" => TriggerKind::RecordCreate, "record.update" => TriggerKind::RecordUpdate, "record.delete" => TriggerKind::RecordDelete, "xrpc.query" => TriggerKind::XrpcQuery, "xrpc.procedure" => TriggerKind::XrpcProcedure, "labeler.apply" => TriggerKind::LabelerApply, "job.run" => TriggerKind::JobRun, other => { return Err(format!( "unknown trigger prefix '{other}'; valid prefixes: \ record.{{index,create,update,delete}}, xrpc.{{query,procedure}}, \ labeler.apply, job.run" )); } };
// Suffix validation: NSID for most triggers, but `labeler.apply:_actor` // and `job.run:<type>` have their own formats. match kind { TriggerKind::JobRun => validate_job_type(suffix)?, TriggerKind::LabelerApply if suffix == "_actor" => {} _ => validate_nsid(suffix)?, }
Ok(Self { kind, suffix: suffix.to_string(), }) }}
fn validate_job_type(job_type: &str) -> Result<(), String> { if job_type.is_empty() || job_type.len() > 128 { return Err(format!( "invalid job type '{job_type}': must be 1–128 characters" )); } if !JOB_TYPE_RE.is_match(job_type) { return Err(format!( "invalid job type '{job_type}': must match /^[a-z0-9][a-z0-9._-]*$/" )); } Ok(())}
/// Minimal NSID validation: at least two dot-separated segments, each/// non-empty and matching `[a-zA-Z][a-zA-Z0-9-]*`. Mirrors the AT Protocol/// spec's character class for everyday use; full Unicode strictness lives/// in atrium downstream.fn validate_nsid(nsid: &str) -> Result<(), String> { let segments: Vec<&str> = nsid.split('.').collect(); if segments.len() < 2 { return Err(format!( "invalid NSID '{nsid}': need at least 2 dot-separated segments" )); } for (idx, seg) in segments.iter().enumerate() { if seg.is_empty() { return Err(format!("invalid NSID '{nsid}': empty segment")); } let mut chars = seg.chars(); let first = chars.next().unwrap(); if !first.is_ascii_alphabetic() { return Err(format!( "invalid NSID '{nsid}': segment {idx} must start with a letter" )); } for c in chars { if !c.is_ascii_alphanumeric() && c != '-' { return Err(format!( "invalid NSID '{nsid}': segment {idx} contains invalid character '{c}'" )); } } } Ok(())}
// ---------------------------------------------------------------------------// ScriptLanguage// ---------------------------------------------------------------------------
/// Runtime a script is written for. Today only [`ScriptLanguage::Lua`] ships;/// the column is stamped per row so a future runtime (e.g. TypeScript) can/// land without a schema migration.#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]#[serde(rename_all = "lowercase")]pub enum ScriptLanguage { #[default] Lua,}
impl ScriptLanguage { pub fn as_str(&self) -> &'static str { match self { Self::Lua => "lua", } }
pub fn parse_str(s: &str) -> Option<Self> { match s { "lua" => Some(Self::Lua), _ => None, } }
pub fn supported() -> &'static [&'static str] { &["lua"] }}
// ---------------------------------------------------------------------------// Script row + resolution// ---------------------------------------------------------------------------
/// A row from the `scripts` table — the wire shape the admin API returns.#[derive(Clone, Debug, Serialize, Deserialize)]pub struct ScriptRow { pub id: String, pub body: String, pub description: Option<String>, pub script_type: String, pub created_at: String, pub updated_at: String,}
/// A script ready to execute.#[derive(Clone, Debug)]pub struct ResolvedScript { pub id: String, pub language: ScriptLanguage, pub body: String,}
/// Look up a single trigger id. Returns `None` when no row matches OR when/// the row's `script_type` is unknown to this binary (logged at warn).pub async fn resolve(state: &AppState, trigger_id: &str) -> Option<ResolvedScript> { let sql = adapt_sql( "SELECT id, body, script_type FROM happyview_scripts WHERE id = ?", state.db_backend, ); let row: Option<(String, String, String)> = match crate::db::query_as(&sql) .bind(trigger_id) .fetch_optional(&state.db) .await { Ok(r) => r, Err(e) => { tracing::warn!(trigger_id, "scripts lookup failed: {e}"); return None; } }; let (id, body, script_type) = row?; let language = match ScriptLanguage::parse_str(&script_type) { Some(l) => l, None => { tracing::warn!( id, script_type, "unknown script_type; this binary supports: {}", ScriptLanguage::supported().join(", ") ); return None; } }; Some(ResolvedScript { id, language, body })}
/// Resolve a record-event trigger with the cascade rule:/// `record.<action>:<nsid>` first, then `record.index:<nsid>`.pub async fn resolve_record_event( state: &AppState, nsid: &str, action: &str,) -> Option<ResolvedScript> { let action_trigger = match action { "create" => Some(format!("record.create:{nsid}")), "update" => Some(format!("record.update:{nsid}")), "delete" => Some(format!("record.delete:{nsid}")), _ => None, }; if let Some(t) = action_trigger && let Some(s) = resolve(state, &t).await { return Some(s); } resolve(state, &format!("record.index:{nsid}")).await}
// ---------------------------------------------------------------------------// Record-event runner (fail-open, retry + dead-letter)// ---------------------------------------------------------------------------
/// All the contextual fields a record-event script needs at execution/// time. Bundled into a struct so the runner doesn't take 6+ `&str`/// positional arguments — easy to swap `did` and `uri` and have the/// type checker shrug.#[derive(Clone, Copy, Debug)]pub struct RecordEventPayload<'a> { pub nsid: &'a str, pub action: &'a str, pub uri: &'a str, pub did: &'a str, pub rkey: &'a str, pub record: Option<&'a Value>,}
/// Run the record-event script (if any) for a given event. Returns the/// record body the indexer should store: `Some(record)` to proceed,/// `None` to skip indexing.////// Failure mode is fail-open: a script that exhausts its retry budget is/// dead-lettered and the indexer proceeds with the original record.pub async fn run_record_event_script( state: &AppState, payload: RecordEventPayload<'_>,) -> Option<Value> { let resolved = match resolve_record_event(state, payload.nsid, payload.action).await { Some(s) => s, // No script for this trigger → indexer keeps the original record. None => return payload.record.cloned(), };
let host_id = format!("{}:{}", payload.nsid, payload.action); let event_payload = serde_json::json!({ "trigger": resolved.id, "action": payload.action, "uri": payload.uri, "did": payload.did, "collection": payload.nsid, "rkey": payload.rkey, "record": payload.record, });
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)); tokio::time::sleep(delay).await; } match run_record_event_once(state, &resolved, payload).await { Ok(outcome) => { log_event( &state.db, EventLog { event_type: "script.executed".to_string(), severity: Severity::Info, actor_did: None, subject: Some(payload.uri.to_string()), detail: serde_json::json!({ "host_kind": "record", "host_id": host_id, "trigger": resolved.id, "attempts": attempt + 1, }), }, state.db_backend, ) .await; return outcome; } Err(e) => { last_error = e; tracing::warn!( uri = %payload.uri, trigger = %resolved.id, attempt = attempt + 1, "record script attempt failed: {last_error}" ); } } }
write_dead_letter( state, &DeadLetterEntry { script: &resolved, host_kind: "record", host_id: &host_id, payload: &event_payload, error: &last_error, attempts: MAX_ATTEMPTS, collection: Some(payload.nsid), }, ) .await; log_event( &state.db, EventLog { event_type: "script.dead_lettered".to_string(), severity: Severity::Error, actor_did: None, subject: Some(payload.uri.to_string()), detail: serde_json::json!({ "host_kind": "record", "host_id": host_id, "trigger": resolved.id, "error": last_error, }), }, state.db_backend, ) .await;
// Fail-open: indexer proceeds with the original record. payload.record.cloned()}
/// Single attempt at the record-event Lua script. Used internally by the/// retry loop and externally by admin retry endpoints.////// Returns `Ok(Some(value))` to continue indexing with `value`,/// `Ok(None)` when the script returned `nil` (skip), or `Err(msg)` on/// any execution failure.pub async fn run_record_event_once( state: &AppState, script: &ResolvedScript, payload: RecordEventPayload<'_>,) -> Result<Option<Value>, String> { if script.language != ScriptLanguage::Lua { return Err(format!( "this binary cannot run {} scripts", script.language.as_str() )); } let lua = sandbox::create_sandbox().map_err(|e| format!("create sandbox: {e}"))?; let state_arc = Arc::new(state.clone()); register_default_apis(&lua, &state_arc, &script.id, Some(payload.did))?;
// Legacy globals (action, uri, did, collection, rkey, record) for // convenience / backwards compatibility. context::set_hook_context( &lua, payload.action, payload.uri, payload.did, payload.nsid, payload.rkey, payload.record, ) .map_err(|e| format!("set hook context: {e}"))?;
// Also expose an `event` table — same fields, different idiom. New // scripts can read `event.action` / `event.record.title` instead of // the bare globals; both styles work. use mlua::LuaSerdeExt; let event_value = serde_json::json!({ "action": payload.action, "uri": payload.uri, "did": payload.did, "collection": payload.nsid, "rkey": payload.rkey, "record": payload.record, }); lua.globals() .set( "event", lua.to_value(&event_value) .map_err(|e| format!("event lua-conv: {e}"))?, ) .map_err(|e| format!("set event global: {e}"))?;
context::set_env_context(&lua, &load_env_vars(&state.db, state.db_backend).await) .map_err(|e| format!("set env: {e}"))?;
lua.load(script.body.as_str()) .exec() .map_err(|e| format!("script load: {e}"))?; let handle: mlua::Function = lua .globals() .get("handle") .map_err(|e| format!("missing handle(): {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 v: Value = lua .from_value(result) .map_err(|e| format!("convert lua return to JSON: {e}"))?; Ok(Some(v)) } // Non-nil, non-table return — pass-through: keep the original record. _ => Ok(payload.record.cloned()), }}
// ---------------------------------------------------------------------------// Label-applied dispatcher (fail-open, retry + dead-letter)// ---------------------------------------------------------------------------
/// Payload passed to `labeler.apply:*` scripts. Mirrors the AT Proto label/// shape from `com.atproto.label.subscribeLabels`.#[derive(Clone, Debug, Serialize, Deserialize)]pub struct LabelAppliedEvent { pub src: String, pub uri: String, pub val: String, #[serde(default)] pub neg: bool, pub cts: String, #[serde(default, skip_serializing_if = "Option::is_none")] pub exp: Option<String>,}
/// What an `on_label_applied` script chain decided.#[derive(Debug)]pub enum LabelHookOutcome { /// Persist the (possibly rewritten) label. Continue(LabelAppliedEvent), /// Skip persistence — the script returned `nil`. Skip,}
/// Compute the trigger string for a given label. `at://` URIs route to/// `labeler.apply:<nsid>` (using the second path segment); everything else/// (bare DIDs, malformed) routes to `labeler.apply:_actor`.pub fn trigger_for_label_uri(uri: &str) -> String { if let Some(rest) = uri.strip_prefix("at://") { match rest.split('/').nth(1) { Some(nsid) if !nsid.is_empty() => format!("labeler.apply:{nsid}"), _ => "labeler.apply:_actor".to_string(), } } else { "labeler.apply:_actor".to_string() }}
/// Run the label-applied script (if any) for an inbound label. Fail-open:/// dead-lettered failures fall through with the original label.pub async fn run_label_applied_script( state: &AppState, event: LabelAppliedEvent,) -> LabelHookOutcome { let trigger = trigger_for_label_uri(&event.uri); let resolved = match resolve(state, &trigger).await { Some(s) => s, None => return LabelHookOutcome::Continue(event), };
let payload = serde_json::to_value(&event).unwrap_or(Value::Null); let host_id = event.src.clone(); let original = event.clone();
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)); tokio::time::sleep(delay).await; } match run_label_lua_once(state, &resolved, &event).await { Ok(outcome) => { log_event( &state.db, EventLog { event_type: "script.executed".to_string(), severity: Severity::Info, actor_did: None, subject: Some(event.uri.clone()), detail: serde_json::json!({ "host_kind": "label", "host_id": host_id, "trigger": resolved.id, "attempts": attempt + 1, }), }, state.db_backend, ) .await; return outcome; } Err(e) => { last_error = e; tracing::warn!( src = %event.src, uri = %event.uri, trigger = %resolved.id, attempt = attempt + 1, "label script attempt failed: {last_error}" ); } } } let collection = resolved .id .split_once(':') .map(|(_, suf)| suf) .filter(|s| *s != "_actor"); write_dead_letter( state, &DeadLetterEntry { script: &resolved, host_kind: "label", host_id: &host_id, payload: &payload, error: &last_error, attempts: MAX_ATTEMPTS, collection, }, ) .await; log_event( &state.db, EventLog { event_type: "script.dead_lettered".to_string(), severity: Severity::Error, actor_did: None, subject: Some(event.uri.clone()), detail: serde_json::json!({ "host_kind": "label", "host_id": host_id, "trigger": resolved.id, "error": last_error, }), }, state.db_backend, ) .await; LabelHookOutcome::Continue(original)}
async fn run_label_lua_once( state: &AppState, script: &ResolvedScript, event: &LabelAppliedEvent,) -> Result<LabelHookOutcome, String> { if script.language != ScriptLanguage::Lua { return Err(format!( "this binary cannot run {} scripts", script.language.as_str() )); } let lua = sandbox::create_sandbox().map_err(|e| format!("create sandbox: {e}"))?; let state_arc = Arc::new(state.clone()); register_default_apis(&lua, &state_arc, &script.id, None)?;
use mlua::LuaSerdeExt; let globals = lua.globals(); globals .set("src", event.src.clone()) .map_err(|e| format!("set src: {e}"))?; globals .set("uri", event.uri.clone()) .map_err(|e| format!("set uri: {e}"))?; globals .set("val", event.val.clone()) .map_err(|e| format!("set val: {e}"))?; globals .set("neg", event.neg) .map_err(|e| format!("set neg: {e}"))?; globals .set("cts", event.cts.clone()) .map_err(|e| format!("set cts: {e}"))?; match &event.exp { Some(exp) => globals.set("exp", exp.clone()), None => globals.set("exp", mlua::Value::Nil), } .map_err(|e| format!("set exp: {e}"))?; let event_value = serde_json::to_value(event).map_err(|e| format!("encode event: {e}"))?; globals .set( "event", lua.to_value(&event_value) .map_err(|e| format!("event lua-conv: {e}"))?, ) .map_err(|e| format!("set event: {e}"))?; context::set_env_context(&lua, &load_env_vars(&state.db, state.db_backend).await) .map_err(|e| format!("set env: {e}"))?;
lua.load(script.body.as_str()) .exec() .map_err(|e| format!("script load: {e}"))?; let handle: mlua::Function = lua .globals() .get("handle") .map_err(|e| format!("missing handle(): {e}"))?; let result: mlua::Value = handle .call_async::<mlua::Value>(()) .await .map_err(|e| e.to_string())?;
match result { mlua::Value::Nil => Ok(LabelHookOutcome::Skip), mlua::Value::Table(_) => { let v: Value = lua .from_value(result) .map_err(|e| format!("convert lua return: {e}"))?; // Merge: any field the script omitted falls back to the // original. This makes "filter only" scripts (return `event`) // and "rewrite val" scripts (return `{ val = "..." }`) both // ergonomic. let next = LabelAppliedEvent { src: extract_string(&v, "src").unwrap_or_else(|| event.src.clone()), uri: extract_string(&v, "uri").unwrap_or_else(|| event.uri.clone()), val: extract_string(&v, "val").unwrap_or_else(|| event.val.clone()), neg: extract_bool(&v, "neg").unwrap_or(event.neg), cts: extract_string(&v, "cts").unwrap_or_else(|| event.cts.clone()), exp: extract_string(&v, "exp").or_else(|| event.exp.clone()), }; Ok(LabelHookOutcome::Continue(next)) } _ => Ok(LabelHookOutcome::Continue(event.clone())), }}
fn extract_string(v: &Value, key: &str) -> Option<String> { v.get(key).and_then(|x| x.as_str()).map(String::from)}
fn extract_bool(v: &Value, key: &str) -> Option<bool> { v.get(key).and_then(|x| x.as_bool())}
// ---------------------------------------------------------------------------// Helpers// ---------------------------------------------------------------------------
/// Register the default API surface on a fresh sandbox: db / http / xrpc //// atproto / Record. `caller_did` flows into xrpc so authenticated calls/// work; pass `None` for unauthenticated contexts.////// The Record API is registered in **no-auth mode** here — fine for/// record-event and label scripts which have no caller credentials./// Calling `:save()` / `:delete()` (the PDS-touching variants) errors/// clearly with the no-PDS-auth message; the local-only variants/// (`:save_local`, `:delete_local`, `Record.delete_local`) work.fn register_default_apis( lua: &mlua::Lua, state: &Arc<AppState>, trigger_id: &str, caller_did: Option<&str>,) -> Result<(), String> { db_api::register_db_api(lua, state.clone()).map_err(|e| format!("db api: {e}"))?; http_api::register_http_api(lua, state.clone()).map_err(|e| format!("http api: {e}"))?; xrpc_api::register_xrpc_api(lua, state.clone(), caller_did.map(String::from)) .map_err(|e| format!("xrpc api: {e}"))?; atproto_api::register_atproto_api(lua, state.clone(), None) .map_err(|e| format!("atproto api: {e}"))?; crate::lua::spaces_api::register_spaces_write_api(lua, state.clone(), caller_did) .map_err(|e| format!("spaces write api: {e}"))?; super::jobs_api::register_jobs_api( lua, state.clone(), caller_did.map(|d| super::jobs_api::JobsCaller { did: d.to_string(), api_client_id: None, dpop_key_id: None, }), ) .map_err(|e| format!("jobs api: {e}"))?; record::register_record_api_no_auth(lua, state.clone()) .map_err(|e| format!("record api: {e}"))?; register_log_event_api(lua, state, trigger_id, caller_did)?; Ok(())}
/// Register `log(msg)` as a Lua global that writes a `script.log` row to/// `event_logs` (so operators can inspect script output from/// `/dashboard/events` without tailing stderr) AND emits a/// `tracing::debug!` for ops who do tail.////// `trigger_id` is recorded as the row's `subject` so events for a/// specific script can be filtered by trigger. `caller_did` is recorded/// as `actor_did` when the runner has one (XRPC handlers); record //// label runners pass `None`.////// This intentionally **overrides** the basic `log()` helper that/// `sandbox::create_sandbox()` registers (which only writes to/// `tracing::debug!`). All script runners call this helper so every/// trigger family lands its `log()` calls in the event log.pub(crate) fn register_log_event_api( lua: &mlua::Lua, state: &Arc<AppState>, trigger_id: &str, caller_did: Option<&str>,) -> Result<(), String> { let state = state.clone(); let trigger_id = trigger_id.to_string(); let caller_did = caller_did.map(String::from); let log_fn = lua .create_async_function(move |_, msg: String| { let state = state.clone(); let trigger_id = trigger_id.clone(); let caller_did = caller_did.clone(); async move { tracing::debug!(lua_log = %msg, trigger = %trigger_id, "lua script log"); log_event( &state.db, EventLog { event_type: "script.log".to_string(), severity: Severity::Info, actor_did: caller_did, subject: Some(trigger_id.clone()), detail: serde_json::json!({ "trigger": trigger_id, "message": msg, }), }, state.db_backend, ) .await; Ok(()) } }) .map_err(|e| format!("log api: {e}"))?; lua.globals() .set("log", log_fn) .map_err(|e| format!("set log global: {e}"))?; Ok(())}
/// Load `script_variables` as a flat key→value map for the `env` global.async fn load_env_vars( db: &sqlx::AnyPool, backend: DatabaseBackend,) -> std::collections::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()}
struct DeadLetterEntry<'a> { script: &'a ResolvedScript, host_kind: &'a str, host_id: &'a str, payload: &'a Value, error: &'a str, attempts: u32, collection: Option<&'a str>,}
async fn write_dead_letter(state: &AppState, entry: &DeadLetterEntry<'_>) { let payload_str = serde_json::to_string(entry.payload).unwrap_or_else(|_| "{}".to_string()); let sql = adapt_sql( "INSERT INTO happyview_dead_letter_scripts (script_ref, host_kind, host_id, payload, error, attempts, created_at, collection) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", state.db_backend, ); if let Err(e) = crate::db::query(&sql) .bind(entry.script.id.as_str()) .bind(entry.host_kind) .bind(entry.host_id) .bind(&payload_str) .bind(entry.error) .bind(entry.attempts as i64) .bind(now_rfc3339()) .bind(entry.collection) .execute(&state.db) .await { tracing::error!( host_kind = entry.host_kind, host_id = entry.host_id, trigger = %entry.script.id, "failed to write dead_letter_scripts: {e}" ); }}
// ---------------------------------------------------------------------------// Tests// ---------------------------------------------------------------------------
#[cfg(test)]mod tests { use super::*;
#[test] fn parse_record_index_trigger() { let t = ParsedTrigger::parse("record.index:com.example.thing").unwrap(); assert_eq!(t.kind, TriggerKind::RecordIndex); assert_eq!(t.suffix, "com.example.thing"); assert_eq!(t.id(), "record.index:com.example.thing"); }
#[test] fn parse_record_action_triggers() { for (prefix, kind) in [ ("record.create", TriggerKind::RecordCreate), ("record.update", TriggerKind::RecordUpdate), ("record.delete", TriggerKind::RecordDelete), ] { let id = format!("{prefix}:com.example.thing"); let t = ParsedTrigger::parse(&id).unwrap(); assert_eq!(t.kind, kind); assert_eq!(t.id(), id); } }
#[test] fn parse_xrpc_triggers() { let q = ParsedTrigger::parse("xrpc.query:com.example.list").unwrap(); assert_eq!(q.kind, TriggerKind::XrpcQuery); let p = ParsedTrigger::parse("xrpc.procedure:com.example.create").unwrap(); assert_eq!(p.kind, TriggerKind::XrpcProcedure); }
#[test] fn parse_labeler_apply_with_nsid() { let t = ParsedTrigger::parse("labeler.apply:app.bsky.feed.post").unwrap(); assert_eq!(t.kind, TriggerKind::LabelerApply); assert_eq!(t.suffix, "app.bsky.feed.post"); }
#[test] fn parse_labeler_apply_actor_special_case() { let t = ParsedTrigger::parse("labeler.apply:_actor").unwrap(); assert_eq!(t.kind, TriggerKind::LabelerApply); assert_eq!(t.suffix, "_actor"); }
#[test] fn rejects_no_colon() { let err = ParsedTrigger::parse("record.index").unwrap_err(); assert!(err.contains("must contain a ':' separator")); assert!(err.contains("valid prefixes")); }
#[test] fn parse_job_run_trigger() { let t = ParsedTrigger::parse("job.run:test.export").unwrap(); assert_eq!(t.kind, TriggerKind::JobRun); assert_eq!(t.suffix, "test.export"); assert_eq!(t.id(), "job.run:test.export"); }
#[test] fn rejects_bad_job_type() { assert!(ParsedTrigger::parse("job.run:UPPER").is_err()); assert!(ParsedTrigger::parse("job.run:has space").is_err()); assert!(ParsedTrigger::parse("job.run:").is_err()); }
#[test] fn rejects_unknown_prefix() { let err = ParsedTrigger::parse("garbage:com.example.thing").unwrap_err(); assert!(err.contains("unknown trigger prefix 'garbage'")); }
#[test] fn rejects_bad_nsid() { // single segment assert!(ParsedTrigger::parse("record.index:foo").is_err()); // empty suffix assert!(ParsedTrigger::parse("record.index:").is_err()); // non-letter start assert!(ParsedTrigger::parse("record.index:1.foo").is_err()); // invalid char assert!(ParsedTrigger::parse("record.index:com.foo!bar").is_err()); }
#[test] fn allows_only_actor_special_case_for_labeler() { // _actor is not a valid NSID, but it's the literal special case. assert!(ParsedTrigger::parse("labeler.apply:_actor").is_ok()); // Other prefixes don't get the _actor escape hatch. assert!(ParsedTrigger::parse("record.index:_actor").is_err()); }
#[test] fn label_uri_routes_at_uri_to_nsid() { assert_eq!( trigger_for_label_uri("at://did:plc:abc/app.bsky.feed.post/rkey1"), "labeler.apply:app.bsky.feed.post" ); }
#[test] fn label_uri_routes_bare_did_to_actor() { assert_eq!(trigger_for_label_uri("did:plc:abc"), "labeler.apply:_actor"); }
#[test] fn label_uri_routes_malformed_at_uri_to_actor() { // `at://` with no path → no second segment → actor. assert_eq!( trigger_for_label_uri("at://did:plc:abc"), "labeler.apply:_actor" ); // `at://<did>/` → second segment exists but is empty → actor. assert_eq!( trigger_for_label_uri("at://did:plc:abc/"), "labeler.apply:_actor" ); }
#[test] fn script_language_round_trip() { assert_eq!(ScriptLanguage::Lua.as_str(), "lua"); assert_eq!(ScriptLanguage::parse_str("lua"), Some(ScriptLanguage::Lua)); assert_eq!(ScriptLanguage::parse_str("typescript"), None); assert_eq!(ScriptLanguage::default(), ScriptLanguage::Lua); }
#[test] fn extract_helpers() { let v = serde_json::json!({"a": "x", "b": true, "c": null}); assert_eq!(extract_string(&v, "a"), Some("x".into())); assert_eq!(extract_string(&v, "missing"), None); assert_eq!(extract_bool(&v, "b"), Some(true)); assert_eq!(extract_bool(&v, "a"), None); }}