A lexicon-driven AppView for ATProto.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185//! 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" => {} _ => happyview_nsid::validate_nsid(suffix).map_err(|e| e.to_string())?, }
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._-]*$/" )); } if crate::jobs::native::is_reserved(job_type) { return Err(format!( "invalid job type '{job_type}': the '{}' prefix is reserved for built-in jobs", crate::jobs::native::RESERVED_PREFIX )); } 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>,}
/// What a record-event script chain decided for an event.////// Deliberately not `Option<Value>`. A delete carries no record body, so/// "no script ran" and "the script returned `nil`" both used to spell/// themselves `None`, and the delete path read the first as the second —/// an instance with no scripts at all silently skipped every delete (#80).#[derive(Debug, Clone, PartialEq)]pub enum RecordHookOutcome { /// Index the event with the body it arrived with. Either no script ran, /// or the script waved the event through without rewriting it. Proceed, /// Index this body in place of the one that arrived. Only meaningful for /// create/update — a delete has no body to replace. Replace(Value), /// Skip the event entirely — the script returned `nil`. Only ever /// produced by a script that actually ran. Skip,}
/// Run the record-event script (if any) for a given event.////// Failure mode is fail-open: a script that exhausts its retry budget is/// dead-lettered and the event proceeds as if no script had run.pub async fn run_record_event_script( state: &AppState, payload: RecordEventPayload<'_>,) -> RecordHookOutcome { let resolved = match resolve_record_event(state, payload.nsid, payload.action).await { Some(s) => s, // No script for this trigger → index the event unchanged. None => return RecordHookOutcome::Proceed, };
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: index the event as if no script had run. For a delete this // means the delete still happens — the record body it lacks is not a // reason to keep a record its PDS no longer has. RecordHookOutcome::Proceed}
/// Single attempt at the record-event Lua script. Used internally by the/// retry loop and externally by admin retry endpoints.////// Returns `Ok(Replace(value))` to continue indexing with `value`,/// `Ok(Skip)` when the script returned `nil`, `Ok(Proceed)` when it waved/// the event through, or `Err(msg)` on any execution failure.pub async fn run_record_event_once( state: &AppState, script: &ResolvedScript, payload: RecordEventPayload<'_>,) -> Result<RecordHookOutcome, 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(RecordHookOutcome::Skip), mlua::Value::Table(_) => { let v: Value = lua .from_value(result) .map_err(|e| format!("convert lua return to JSON: {e}"))?; Ok(RecordHookOutcome::Replace(v)) } // Non-nil, non-table return (`return true`) — pass-through. _ => Ok(RecordHookOutcome::Proceed), }}
// ---------------------------------------------------------------------------// 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 / linked_repos / jobs / Record. `caller_did` flows into xrpc so/// authenticated calls work; pass `None` for unauthenticated contexts.////// `linked_repos` needs no caller: a grant carries its own PDS session, so a/// record-event script can mirror into a linked repo with no credentials of its/// own. It is registered here for the same reason the spaces write surface is —/// every script context is meant to have it.////// 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}"))?; crate::lua::linked_repos_api::register_linked_repos_api(lua, state.clone()) .map_err(|e| format!("linked repos 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::*;
use crate::AppState; use crate::config::Config; use crate::lexicon::LexiconRegistry; use std::sync::Arc; use tokio::sync::watch;
fn registration_test_state() -> AppState { let config = Config { host: "127.0.0.1".into(), port: 3000, database_url: String::new(), database_backend: crate::db::DatabaseBackend::Sqlite, sqlite_journal_size_limit: crate::db::DEFAULT_JOURNAL_SIZE_LIMIT, public_url: String::new(), user_agent: String::new(), session_secret: "test-secret".into(), jetstream_url: String::new(), relay_url: String::new(), plc_url: String::new(), static_dir: String::new(), base_path: None, 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, telemetry_collector_url: String::new(), }; 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 backend = crate::db::DatabaseBackend::Sqlite; let atrium_http = Arc::new(crate::http_retry::HappyViewHttpClient::default()); let did_resolver = atrium_identity::did::CommonDidResolver::new( atrium_identity::did::CommonDidResolverConfig { plc_directory_url: "https://plc.directory".into(), http_client: 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 scopes = vec![atrium_oauth::Scope::Known( atrium_oauth::KnownScope::Atproto, )]; 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(scopes.clone()), }, keys: None, state_store: crate::auth::oauth_store::DbStateStore::new(test_db.clone(), backend), session_store: crate::auth::oauth_store::DbSessionStore::new(test_db.clone(), backend), resolver: atrium_oauth::OAuthResolverConfig { did_resolver, handle_resolver, authorization_server_metadata: Default::default(), protected_resource_metadata: Default::default(), }, http_client: crate::http_retry::HappyViewHttpClient::default(), }) .expect("test OAuth client"); AppState { config, http: reqwest::Client::new(), db: test_db.clone(), backfill_db: test_db.clone(), db_backend: backend, 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: Arc::new(crate::auth::OAuthClientRegistry::new(Arc::new(oauth))), oauth_state_store: crate::auth::oauth_store::DbStateStore::new( test_db.clone(), backend, ), linked_repos_client: Arc::new( crate::linked_repos::client::build( "https://plc.directory", "http://127.0.0.1:0/oauth-client-metadata.json", "http://127.0.0.1:0", "http://127.0.0.1:0/auth/callback".into(), true, scopes, crate::auth::oauth_store::DbStateStore::new(test_db.clone(), backend), test_db.clone(), backend, None, ) .expect("test linked-repo OAuth client"), ), linked_repos_client_kid: None, cookie_key: axum_extra::extract::cookie::Key::derive_from( b"test-secret-that-is-at-least-32-bytes-long", ), plugin_registry: Arc::new(crate::plugin::PluginRegistry::new()), wasm_runtime: Arc::new(crate::plugin::WasmRuntime::new().expect("wasm runtime")), attestation_signer: None, official_registry: Arc::new(tokio::sync::RwLock::new( crate::plugin::official_registry::OfficialRegistryState::default(), )), official_registry_config: crate::plugin::official_registry::RegistryConfig::production( ), proxy_config: Arc::new(arc_swap::ArcSwap::new(Arc::new( crate::proxy_config::ProxyConfig::default(), ))), backfill_events_tx: tokio::sync::broadcast::channel(16).0, verbose_event_logging: Arc::new(std::sync::atomic::AtomicBool::new(false)), client_jwks: Vec::new(), telemetry_counters: Arc::new(crate::telemetry::counters::Counters::new()), } }
#[tokio::test] async fn default_apis_include_linked_repos() { let state = Arc::new(registration_test_state()); let lua = mlua::Lua::new();
register_default_apis(&lua, &state, "record.create:com.example.thing", None) .expect("default APIs should register");
let (kind, get_kind, list_kind): (String, String, String) = lua .load("return type(linked_repos), type(linked_repos.get), type(linked_repos.list)") .eval_async() .await .expect("linked_repos should be present for record-event and label scripts");
assert_eq!(kind, "table"); assert_eq!(get_kind, "function"); assert_eq!(list_kind, "function"); }
#[tokio::test] async fn default_apis_still_include_the_existing_globals() { let state = Arc::new(registration_test_state()); let lua = mlua::Lua::new();
register_default_apis(&lua, &state, "labeler.apply:app.bsky.feed.post", None) .expect("default APIs should register");
let ok: bool = lua .load( "return type(db) == 'table' and type(http) == 'table' \ and type(xrpc) == 'table' and type(atproto) == 'table' \ and type(atproto.spaces) == 'table' and type(jobs) == 'table'", ) .eval_async() .await .unwrap(); assert!(ok, "existing script globals must survive the addition"); }
#[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 delegates_nsid_validation_to_the_shared_crate() { // Shape rules are the crate's job and are pinned by the interop corpus // there. This only proves the wiring. assert!(ParsedTrigger::parse("xrpc.query:pics.2bit.feed.getPhotos").is_ok()); assert!(ParsedTrigger::parse("record.index:1.foo").is_err()); assert!(ParsedTrigger::parse("record.index:").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); }}