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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945//! 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 [`crate::catalog::ScriptCatalog::resolve_in`] and call//! [`super::execute::execute_procedure_script`] ///! [`super::execute::execute_query_script`] directly.
use serde::{Deserialize, Serialize};use serde_json::Value;use std::path::PathBuf;use std::sync::Arc;use std::sync::atomic::Ordering;
use crate::AppState;use crate::catalog::Layer;use crate::db::{adapt_sql, now_rfc3339};use crate::event_log::{EventLog, Severity, log_event};use crate::plugin::{ScriptExecuteOutput, ScriptValueKind};use crate::project::ProjectSnapshot;use crate::script::{Invocation, ScriptOrigin, Trigger, dispatch};
/// Number of attempts (1 initial + 3 retries) before dead-lettering.const MAX_ATTEMPTS: u32 = 4;
/// A test run reports a script's failure rather than waiting out seven/// seconds of backoff for a failure retrying cannot change.fn attempts(state: &AppState) -> u32 { if state.test.is_some() { 1 } else { MAX_ATTEMPTS }}
// ---------------------------------------------------------------------------// 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 { /// A trigger the host builds from an event it is handling, rather than /// reads out of text. It validates nothing: [`parse`](Self::parse) /// answers whether text names a trigger, while a suffix taken from a /// firehose event or a lexicon id is looked up as it stands, and one the /// grammar would refuse simply matches no row. pub fn new(kind: TriggerKind, suffix: impl Into<String>) -> Self { Self { kind, suffix: suffix.into(), } }
/// 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 => { crate::jobs::validate_job_type(suffix)?; // A project names a job by its file, and `*.test.<ext>` is a // test file before it is anything else. Only the grammar // refuses: a job already saved under such a type is still // enqueued and resolved by exact id, so it keeps running. if suffix.ends_with(".test") { return Err(format!( "invalid job type '{suffix}': a final '.test' segment is reserved, \ because src/jobs/{suffix}.<ext> would be read as a test file" )); } } TriggerKind::LabelerApply if suffix == "_actor" => {} _ => happyview_nsid::validate_nsid(suffix).map_err(|e| e.to_string())?, }
Ok(Self { kind, suffix: suffix.to_string(), }) }}
// ---------------------------------------------------------------------------// Script row + resolution// ---------------------------------------------------------------------------
/// The language the reference implementation in this crate reads, which is/// what the codemod and the save-time compile check are written against./// Which other languages a script may be written in is a question about/// installed interpreters, so only this one is a constant.pub const NATIVE_LANGUAGE: &str = "lua";
/// A script ready to execute.#[derive(Clone, Debug)]pub struct ResolvedScript { pub id: String, /// The grammar kind of the trigger this row was looked up by, which is /// what decides the contract its result is read under. It comes from the /// lookup rather than from re-reading `id`, because a stored id the /// grammar refuses still resolves by exact match and still has to run. pub trigger: TriggerKind, /// Which interpreter runs `body`, as the row stamped it. pub script_type: String, pub body: String, /// The layer the script came from. pub layer: Layer, /// Project-relative file path, for a project script only. pub path: Option<PathBuf>, /// What the interpreter names the chunk, so an error points at the file a /// project script lives in. Dashboard scripts use `DASHBOARD_CHUNK_NAME`. pub chunk_name: String, /// XRPC methods the script calls, checked against a service-auth token's /// scope before a procedure runs. pub outbound_xrpcs: Vec<String>, /// The snapshot this script was resolved in, whichever layer answered. /// The run carries it so everything it resolves later, such as a local /// module, comes from the same build. pub snapshot: Arc<ProjectSnapshot>,}
// ---------------------------------------------------------------------------// 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 nothing. 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 project = state.project.load_full(); let resolved = match state .scripts .resolve_record_event_in(&project, 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, });
if no_interpreter_for(state, &resolved).await { dead_letter( state, &DeadLetterEntry { script: &resolved, host_kind: "record", host_id: &host_id, payload: &event_payload, error: &crate::script::no_interpreter_message(&resolved.script_type), attempts: 1, collection: Some(payload.nsid), }, payload.uri, ) .await; return RecordHookOutcome::Proceed; }
let mut last_error = String::new(); for attempt in 0..attempts(state) { 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) => { 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(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}" ); } } }
dead_letter( state, &DeadLetterEntry { script: &resolved, host_kind: "record", host_id: &host_id, payload: &event_payload, error: &last_error, attempts: attempts(state), collection: Some(payload.nsid), }, payload.uri, ) .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 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 nothing, `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> { let event = serde_json::json!({ "action": payload.action, "uri": payload.uri, "did": payload.did, "collection": payload.nsid, "rkey": payload.rkey, "record": payload.record, }); let trigger = trigger_of(script, None)?; let outcome = dispatch( state, &Invocation { trigger_id: &script.id, trigger, language: &script.script_type, source: &script.body, input: &event, // The record's author is who a library reads as the caller, and a // firehose event carries none of their credentials. caller_did: Some(payload.did), has_pds_auth: false, method: None, collection: Some(payload.nsid), params: None, delegate_did: None, space: None, origin: ScriptOrigin::of(script), }, None, ) .await;
match outcome.map_err(|e| e.to_string())? { ScriptExecuteOutput::Returned { value, value_kind } => Ok(match value_kind { ScriptValueKind::None => RecordHookOutcome::Skip, ScriptValueKind::Object => RecordHookOutcome::Replace(value), ScriptValueKind::Other => RecordHookOutcome::Proceed, }), ScriptExecuteOutput::Error { kind, raw, chunk, .. } => Err(failure_text( kind, crate::error::file_of(chunk).as_deref(), &raw, )), }}
// ---------------------------------------------------------------------------// 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 nothing. Skip,}
/// Compute the trigger 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) -> ParsedTrigger { let suffix = match uri.strip_prefix("at://").and_then(|rest| { rest.split('/') .nth(1) .filter(|nsid| !nsid.is_empty()) .map(str::to_string) }) { Some(nsid) => nsid, None => "_actor".to_string(), }; ParsedTrigger::new(TriggerKind::LabelerApply, suffix)}
/// 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 project = state.project.load_full(); let resolved = match state.scripts.resolve_in(&project, &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(); // `_actor` names a bare DID rather than a collection, so it is the one // suffix with nothing to record here. let collection = Some(trigger.suffix.as_str()).filter(|s| *s != "_actor");
if no_interpreter_for(state, &resolved).await { dead_letter( state, &DeadLetterEntry { script: &resolved, host_kind: "label", host_id: &host_id, payload: &payload, error: &crate::script::no_interpreter_message(&resolved.script_type), attempts: 1, collection, }, &event.uri, ) .await; return LabelHookOutcome::Continue(original); }
let mut last_error = String::new(); for attempt in 0..attempts(state) { if attempt > 0 { let delay = std::time::Duration::from_secs(1 << (attempt - 1)); tokio::time::sleep(delay).await; } match run_label_once(state, &resolved, &event).await { Ok(outcome) => { if matches!(outcome, LabelHookOutcome::Skip) { tracing::debug!( target: "happyview::labeler", src = %event.src, uri = %event.uri, val = %event.val, "label.applied script skipped persistence" ); } 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(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}" ); } } } dead_letter( state, &DeadLetterEntry { script: &resolved, host_kind: "label", host_id: &host_id, payload: &payload, error: &last_error, attempts: attempts(state), collection, }, &event.uri, ) .await; LabelHookOutcome::Continue(original)}
async fn run_label_once( state: &AppState, script: &ResolvedScript, event: &LabelAppliedEvent,) -> Result<LabelHookOutcome, String> { let input = serde_json::to_value(event).map_err(|e| format!("encode event: {e}"))?; let trigger = trigger_of(script, None)?; let outcome = dispatch( state, &Invocation { trigger_id: &script.id, trigger, language: &script.script_type, source: &script.body, input: &input, // A label arrives from a subscription, so there is nobody for the // run to act as and no collection of its own to read. caller_did: None, has_pds_auth: false, method: None, collection: None, params: None, delegate_did: None, space: None, origin: ScriptOrigin::of(script), }, None, ) .await;
match outcome.map_err(|e| e.to_string())? { ScriptExecuteOutput::Returned { value, value_kind } => Ok(match value_kind { ScriptValueKind::None => LabelHookOutcome::Skip, // 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. ScriptValueKind::Object => LabelHookOutcome::Continue(LabelAppliedEvent { src: extract_string(&value, "src").unwrap_or_else(|| event.src.clone()), uri: extract_string(&value, "uri").unwrap_or_else(|| event.uri.clone()), val: extract_string(&value, "val").unwrap_or_else(|| event.val.clone()), neg: extract_bool(&value, "neg").unwrap_or(event.neg), cts: extract_string(&value, "cts").unwrap_or_else(|| event.cts.clone()), exp: extract_string(&value, "exp").or_else(|| event.exp.clone()), }), ScriptValueKind::Other => LabelHookOutcome::Continue(event.clone()), }), ScriptExecuteOutput::Error { kind, raw, chunk, .. } => Err(failure_text( kind, crate::error::file_of(chunk).as_deref(), &raw, )), }}
/// How a failed run reads to an operator: the category it arrived with, then/// the interpreter's unparsed text.////// Every reader of this is an operator — the retry's warn, the dead-letter/// row's only error column, a job row's `error` column, the/// `script.dead_lettered` row — and neither half is recoverable from the/// other. The text carries the line that points at the script and no category;/// a spent budget and an exhausted heap describe themselves in no text at all,/// so without the category they read as any other runtime failure.pub(crate) fn failure_text( kind: crate::plugin::ScriptErrorKind, file: Option<&str>, raw: &str,) -> String { match file { Some(file) => format!("{} in {file}: {raw}", kind.as_str()), None => format!("{}: {raw}", kind.as_str()), }}
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// ---------------------------------------------------------------------------
struct DeadLetterEntry<'a> { script: &'a ResolvedScript, host_kind: &'a str, host_id: &'a str, payload: &'a Value, error: &'a str, /// How many runs it took to get here. A retry budget that was spent /// reports the whole budget; a failure no retry could have changed /// reports the one attempt it was worth. attempts: u32, collection: Option<&'a str>,}
/// The trigger a resolved row's run carries, read from the kind the row was/// looked up by. A kind and a job that cannot describe one run end the/// attempt, rather than running the script under a contract it was not/// written against.pub(crate) fn trigger_of<'a>( script: &'a ResolvedScript, job: Option<&'a str>,) -> Result<Trigger<'a>, String> { Trigger::for_trigger_kind(script.trigger, job) .ok_or_else(|| format!("no run can be built for trigger '{}'", script.id))}
/// Whether nothing installed claims this row's language.////// The runners ask before spending a retry budget, because a missing plugin/// is not a transient failure: four attempts and seven seconds of backoff/// cannot install one, and on a busy collection that is ingest held up behind/// a wait nothing can satisfy. [`DispatchError`] would say the same thing/// from inside the loop, but by then the budget is already being spent.////// [`DispatchError`]: crate::script::DispatchErrorasync fn no_interpreter_for(state: &AppState, script: &ResolvedScript) -> bool { state .plugin_registry .get_interpreter_by_language(&script.script_type) .await .is_none()}
/// Record a failed run where an operator reads it: the row they retry from/// and the event their feed shows. One call because they describe one failure/// and a run that wrote only one of them would be a half-told story.async fn dead_letter(state: &AppState, entry: &DeadLetterEntry<'_>, subject: &str) { if let Some(test) = &state.test { test.note_script_failure(entry.error); } write_dead_letter(state, entry).await; log_event( &state.db, EventLog { event_type: "script.dead_lettered".to_string(), severity: Severity::Error, actor_did: None, subject: Some(subject.to_string()), detail: serde_json::json!({ "host_kind": entry.host_kind, "host_id": entry.host_id, "trigger": entry.script.id, "error": entry.error, }), }, state.db_backend, ) .await;}
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 { #[test] fn a_failure_text_leads_with_its_file() { assert_eq!( failure_text( crate::plugin::ScriptErrorKind::Runtime, Some("src/jobs/tick.lua"), "src/jobs/tick.lua:4: boom" ), "runtime in src/jobs/tick.lua: src/jobs/tick.lua:4: boom" ); assert_eq!( failure_text(crate::plugin::ScriptErrorKind::Runtime, None, "x"), "runtime: x" ); }
use super::*;
#[test] fn the_job_trigger_grammar_refuses_a_final_test_segment() { let err = ParsedTrigger::parse("job.run:instance.operation.test").unwrap_err(); assert!(err.contains(".test"), "{err}"); assert!(err.contains("test file"), "{err}");
for fine in [ "test", "test.export", "instance.testing", "contest", "a.tests", ] { assert!( ParsedTrigger::parse(&format!("job.run:{fine}")).is_ok(), "{fine}" ); } }
#[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()); }
/// A lookup's kind and the kind its id names are the same kind, for every /// kind. A resolved row is found by exact id match, so this is what makes /// [`ResolvedScript::trigger`] the kind the stored id names rather than a /// guess about it — and what fails if the two prefix tables drift. #[test] fn a_lookup_id_names_the_kind_it_was_built_from() { for (kind, suffix) in [ (TriggerKind::RecordIndex, "com.example.thing"), (TriggerKind::RecordCreate, "com.example.thing"), (TriggerKind::RecordUpdate, "com.example.thing"), (TriggerKind::RecordDelete, "com.example.thing"), (TriggerKind::XrpcQuery, "com.example.list"), (TriggerKind::XrpcProcedure, "com.example.post"), (TriggerKind::LabelerApply, "app.bsky.feed.post"), (TriggerKind::JobRun, "test.export"), ] { let id = ParsedTrigger::new(kind, suffix).id(); assert_eq!(ParsedTrigger::parse(&id).expect(&id).kind, kind, "{id}"); } }
#[test] fn label_uri_routes_at_uri_to_nsid() { assert_eq!( trigger_for_label_uri("at://did:plc:abc/app.bsky.feed.post/rkey1").id(), "labeler.apply:app.bsky.feed.post" ); }
#[test] fn label_uri_routes_bare_did_to_actor() { assert_eq!( trigger_for_label_uri("did:plc:abc").id(), "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").id(), "labeler.apply:_actor" ); // `at://<did>/` → second segment exists but is empty → actor. assert_eq!( trigger_for_label_uri("at://did:plc:abc/").id(), "labeler.apply:_actor" ); }
#[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); }}