From 68e7920bb9a64509838219e0ccd02a31d79f2c5f Mon Sep 17 00:00:00 2001 From: Chris Pardy Date: Fri, 01 May 2026 11:46:55 +0000 Subject: [PATCH] feat(scripts): trigger-keyed scripts subsystem Replaces the legacy `index_hook` lookup with a trigger-keyed scripts table where each row's PK IS the trigger string the dispatcher resolves on. There's no separate "name" or "host column" — bindings, names, and inline-vs-named all collapse into one column. Trigger grammar: record.index: — wildcard for any record event record.create: — specifically a create event record.update: — specifically an update event record.delete: — specifically a delete event xrpc.query: — XRPC query handler xrpc.procedure: — XRPC procedure handler labeler.apply: — label whose uri is at://// labeler.apply:_actor — label whose uri is a bare DID Cascade for record events ONLY: the dispatcher tries `record.:` first, falls back to `record.index:`. No cascade for XRPC or labeler triggers. Schema (one new migration, no data copy — origin/dev had no host columns to migrate from): - New `scripts` table (id PK + body / description / script_type + timestamps). - New `dead_letter_scripts` table (script_ref carries the trigger id). - Re-grants `scripts:read` to lexicon-readers, `scripts:manage` to lexicon-managers. Code surface: - New `src/lua/scripts.rs` dispatcher: ParsedTrigger grammar + validator, ScriptRow / ResolvedScript, `resolve` / `resolve_record_event` (with cascade), `run_record_event_script` (fail-open + retry + dead-letter), `run_label_applied_script`, `trigger_for_label_uri` (at:// → nsid, bare DID → _actor), and `run_record_event_once` for the dead-letter retry path. - New `src/admin/scripts.rs` CRUD (list / get / upsert / patch / delete) with trigger-id grammar validation at write-time and Lua body validation via `crate::lua::validate_script`. - New `Permission::ScriptsRead` / `Permission::ScriptsManage`. - `src/labeler.rs::apply_label` computes the trigger from `label.uri` and runs the script chain before persisting; rewritten labels persist; nil-returning scripts skip. - `src/record_handler.rs::handle_record_event` calls `run_record_event_script(state, &collection, &action, ...)` for both create/update and delete actions. - `src/xrpc/procedure.rs` and `src/xrpc/query.rs` look up via the new dispatcher (trigger `xrpc.procedure:` / `xrpc.query:`) before falling through to the default PDS-write / list flows. - `src/admin/dead_letters.rs::retry_single` resolves via the new dispatcher's cascade (404s if no script matches now). - Legacy `execute_hook_script`, `HookEvent`, and `run_hook_once` removed from `src/lua/execute.rs`. - `register_record_api` refactored to take Optional Claims / PdsAuth; new `register_record_api_no_auth` public wrapper for the no-auth contexts (record / labeler / query). PDS-touching methods (`r:save`, `r:delete`, `Record.save_all`) error cleanly with "no PDS auth in this script context" when registered without auth. - `r:save_local()` / `r:delete_local()` instance methods + `Record.delete_local(uri)` static for local-only mutation in any script context. - `r:delete()` proceeds with the local delete even on PDS failure (operator's logical action is removal regardless of PDS state). Legacy `lexicons.index_hook` column is retained but inert — operators with existing data manually port into a `record.index:` script row via the dashboard. Tests: - 12 new lib tests in `src/lua/scripts.rs` covering trigger grammar, NSID validation, label URI routing, and the `_actor` special case. - 8 integration tests in `tests/lua_record_api.rs` covering the Record local API + the no-PDS-auth boundary. - 18 e2e tests in `tests/e2e_scripts.rs` covering admin CRUD, trigger validation, the cascade rule, label-script + record-event Record local mutation, and the dead-letter behavior when a label script reaches for `r:save()`. - The 4 query-script lib tests in `src/lua/xrpc_api.rs` updated to seed scripts via the new table (plus a single-connection test pool so the in-memory sqlite is shared). --- migrations/postgres/20260501000000_scripts_by_trigger.sql | 58 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ migrations/sqlite/20260501000000_scripts_by_trigger.sql | 52 ++++++++++++++++++++++++++++++++++++++++++++++++++++ src/admin/dead_letters.rs | 56 ++++++++++++++++++++++++-------------------------------- src/admin/mod.rs | 8 ++++++++ src/admin/permissions.rs | 13 ++++++++++++- src/admin/scripts.rs | 373 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ src/labeler.rs | 59 +++++++++++++++++++++++++++++++++++++++++++++-------------- src/lua/execute.rs | 468 +++++++++++++++++++++++++++++++++--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- src/lua/mod.rs | 12 ++++++++---- src/lua/record.rs | 262 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------------------- src/lua/scripts.rs | 855 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ src/lua/xrpc_api.rs | 90 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++--------------------- src/record_handler.rs | 147 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------------------------------------------------------------------------------- src/xrpc/procedure.rs | 12 ++++++++++-- src/xrpc/query.rs | 14 ++++++++++++-- tests/common/db.rs | 4 +++- tests/e2e_scripts.rs | 833 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ tests/lua_record_api.rs | 474 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 18 file(s) changed, 3180 insertion(s)(+), 610 deletion(s)(-) diff --git a/migrations/postgres/20260501000000_scripts_by_trigger.sql b/migrations/postgres/20260501000000_scripts_by_trigger.sql new file mode 100644 --- /dev/null +++ b/migrations/postgres/20260501000000_scripts_by_trigger.sql @@ -0,0 +1,58 @@ +-- Trigger-keyed scripts subsystem. +-- +-- Each row's `id` IS the trigger string the dispatcher resolves on: +-- record.index: — wildcard for any record event +-- record.create: — specifically a create event (cascades to wildcard) +-- record.update: — specifically an update event +-- record.delete: — specifically a delete event +-- xrpc.query: — XRPC query handler +-- xrpc.procedure: — XRPC procedure handler +-- labeler.apply: — label arrives whose subject is at://// +-- labeler.apply:_actor — label arrives whose subject is a bare DID +-- +-- Cascade rule (record events ONLY): the dispatcher tries +-- `record.:` first, falls back to `record.index:` if no +-- specific row exists. No cascade for XRPC or labeler triggers — those +-- resolve directly. Operators express per-action surgical control by +-- creating action-specific rows; the wildcard `record.index:` covers +-- "one body for everything" with branching on `event.action`. +CREATE TABLE scripts ( + id TEXT PRIMARY KEY, + body TEXT NOT NULL, + description TEXT, + script_type TEXT NOT NULL DEFAULT 'lua', + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +-- Permanently-failed runs from the firehose-driven runners (record / label +-- events). XRPC scripts fail-closed and never land here. +CREATE TABLE dead_letter_scripts ( + id BIGSERIAL PRIMARY KEY, + script_ref TEXT NOT NULL, -- = the trigger id whose script failed + host_kind TEXT NOT NULL, -- 'record' | 'label' + host_id TEXT NOT NULL, -- e.g. ':' for record, '' for label + payload JSONB NOT NULL, -- event payload for re-run + error TEXT NOT NULL, + attempts INT NOT NULL DEFAULT 0, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + resolved_at TIMESTAMPTZ +); + +CREATE INDEX idx_dead_letter_scripts_host + ON dead_letter_scripts (host_kind, host_id, created_at DESC); +CREATE INDEX idx_dead_letter_scripts_resolved_at + ON dead_letter_scripts (resolved_at); + +-- Permissions: management for users who can manage lexicons; read for those who can read. +INSERT INTO user_permissions (user_id, permission) +SELECT user_id, 'scripts:manage' + FROM user_permissions + WHERE permission = 'lexicons:create' +ON CONFLICT (user_id, permission) DO NOTHING; + +INSERT INTO user_permissions (user_id, permission) +SELECT user_id, 'scripts:read' + FROM user_permissions + WHERE permission = 'lexicons:read' +ON CONFLICT (user_id, permission) DO NOTHING; diff --git a/migrations/sqlite/20260501000000_scripts_by_trigger.sql b/migrations/sqlite/20260501000000_scripts_by_trigger.sql new file mode 100644 --- /dev/null +++ b/migrations/sqlite/20260501000000_scripts_by_trigger.sql @@ -0,0 +1,52 @@ +-- Trigger-keyed scripts subsystem. +-- +-- Each row's `id` IS the trigger string the dispatcher resolves on: +-- record.index: — wildcard for any record event +-- record.create: — specifically a create event (cascades to wildcard) +-- record.update: — specifically an update event +-- record.delete: — specifically a delete event +-- xrpc.query: — XRPC query handler +-- xrpc.procedure: — XRPC procedure handler +-- labeler.apply: — label arrives whose subject is at://// +-- labeler.apply:_actor — label arrives whose subject is a bare DID +-- +-- See migrations/postgres/20260501000000_scripts_by_trigger.sql for design notes. +-- SQLite mirror. +CREATE TABLE scripts ( + id TEXT PRIMARY KEY, + body TEXT NOT NULL, + description TEXT, + script_type TEXT NOT NULL DEFAULT 'lua', + created_at TEXT NOT NULL DEFAULT (datetime('now')), + updated_at TEXT NOT NULL DEFAULT (datetime('now')) +); + +CREATE TABLE dead_letter_scripts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + script_ref TEXT NOT NULL, -- = the trigger id whose script failed + host_kind TEXT NOT NULL, -- 'record' | 'label' (xrpc fails-closed; never dead-letters) + host_id TEXT NOT NULL, -- e.g. ':' for record, '' for label + payload TEXT NOT NULL, -- JSON-serialized event for re-run + error TEXT NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + created_at TEXT NOT NULL DEFAULT (datetime('now')), + resolved_at TEXT +); + +CREATE INDEX idx_dead_letter_scripts_host + ON dead_letter_scripts (host_kind, host_id, created_at DESC); +CREATE INDEX idx_dead_letter_scripts_resolved_at + ON dead_letter_scripts (resolved_at); + +-- Permissions: management for users who can manage lexicons; read for those who can read. +INSERT INTO user_permissions (user_id, permission, granted_at) +SELECT user_id, 'scripts:manage', datetime('now') + FROM user_permissions + WHERE permission = 'lexicons:create' +ON CONFLICT (user_id, permission) DO NOTHING; + +INSERT INTO user_permissions (user_id, permission, granted_at) +SELECT user_id, 'scripts:read', datetime('now') + FROM user_permissions + WHERE permission = 'lexicons:read' +ON CONFLICT (user_id, permission) DO NOTHING; diff --git a/src/admin/dead_letters.rs b/src/admin/dead_letters.rs --- a/src/admin/dead_letters.rs +++ b/src/admin/dead_letters.rs @@ -10,7 +10,7 @@ use super::permissions::Permission; use crate::AppState; use crate::db::{adapt_sql, now_rfc3339, parse_dt}; use crate::error::AppError; -use crate::lua::{HookEvent, run_hook_once}; +use crate::lua::{resolve_record_event, run_record_event_once}; use crate::record_handler::RecordEvent; // --------------------------------------------------------------------------- @@ -487,29 +487,22 @@ )) } } -/// Fetch the index_hook script directly from the lexicons table, bypassing the in-memory registry. -async fn get_index_hook_from_db( - state: &AppState, - lexicon_id: &str, -) -> Result, AppError> { - let backend = state.db_backend; - let sql = adapt_sql("SELECT index_hook FROM lexicons WHERE id = ?", backend); - let row: Option<(Option,)> = sqlx::query_as(&sql) - .bind(lexicon_id) - .fetch_optional(&state.db) - .await - .map_err(|e| AppError::Internal(format!("failed to fetch index hook: {e}")))?; - Ok(row.and_then(|r| r.0)) -} - -/// Retry a single dead letter by re-running its hook script. +/// Retry a single dead letter by re-running its trigger-keyed script. +/// +/// Resolves the script via the new dispatcher's cascade +/// (`record.:` → `record.index:`). If no script is +/// bound for the cascade now, returns 404 — the operator either deleted +/// the script or never re-bound it under the new naming. async fn retry_single(state: &AppState, id: &str) -> Result<(), AppError> { let dl = fetch_dead_letter_for_action(state, id).await?; - let script = get_index_hook_from_db(state, &dl.lexicon_id) - .await? + let resolved = resolve_record_event(state, &dl.collection, &dl.action) + .await .ok_or_else(|| { - AppError::NotFound(format!("no index hook found for lexicon {}", dl.lexicon_id)) + AppError::NotFound(format!( + "no script bound for record.{}:{} (or record.index:{})", + dl.action, dl.collection, dl.collection + )) })?; let record: Option = dl @@ -517,19 +510,18 @@ .record .as_deref() .and_then(|r| serde_json::from_str(r).ok()); - let event = HookEvent { + match run_record_event_once( state, - lexicon_id: &dl.lexicon_id, - script: &script, - action: &dl.action, - uri: &dl.uri, - did: &dl.did, - collection: &dl.collection, - rkey: &dl.rkey, - record: record.as_ref(), - }; - - match run_hook_once(&event).await { + &resolved, + &dl.action, + &dl.uri, + &dl.did, + &dl.collection, + &dl.rkey, + record.as_ref(), + ) + .await + { Ok(_) => { mark_resolved(state, id).await?; Ok(()) diff --git a/src/admin/mod.rs b/src/admin/mod.rs --- a/src/admin/mod.rs +++ b/src/admin/mod.rs @@ -14,6 +14,7 @@ mod plugins; mod proxy_config; mod records; mod script_variables; +mod scripts; pub mod settings; mod stats; pub(crate) mod types; @@ -68,6 +69,13 @@ "/script-variables", post(script_variables::upsert).get(script_variables::list), ) .route("/script-variables/{key}", delete(script_variables::delete)) + .route("/scripts", get(scripts::list).post(scripts::upsert)) + .route( + "/scripts/{id}", + get(scripts::get) + .patch(scripts::patch) + .delete(scripts::delete), + ) .route("/labelers", post(labelers::add).get(labelers::list)) .route( "/labelers/{did}", diff --git a/src/admin/permissions.rs b/src/admin/permissions.rs --- a/src/admin/permissions.rs +++ b/src/admin/permissions.rs @@ -2,7 +2,7 @@ use std::collections::HashSet; use serde::{Deserialize, Serialize}; -/// All 37 permissions in the system. +/// All permissions in the system. #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] pub enum Permission { #[serde(rename = "lexicons:create")] @@ -100,6 +100,11 @@ #[serde(rename = "spaces:manage-records")] SpacesManageRecords, #[serde(rename = "spaces:manage-credentials")] SpacesManageCredentials, + + #[serde(rename = "scripts:read")] + ScriptsRead, + #[serde(rename = "scripts:manage")] + ScriptsManage, } impl Permission { @@ -147,6 +152,8 @@ Self::SpacesManageMembers => "spaces:manage-members", Self::SpacesManageInvites => "spaces:manage-invites", Self::SpacesManageRecords => "spaces:manage-records", Self::SpacesManageCredentials => "spaces:manage-credentials", + Self::ScriptsRead => "scripts:read", + Self::ScriptsManage => "scripts:manage", } } @@ -194,6 +201,8 @@ Self::SpacesManageMembers, Self::SpacesManageInvites, Self::SpacesManageRecords, Self::SpacesManageCredentials, + Self::ScriptsRead, + Self::ScriptsManage, ]) } } @@ -215,6 +224,7 @@ Self::Viewer => HashSet::from([ Permission::LexiconsRead, Permission::RecordsRead, Permission::ScriptVariablesRead, + Permission::ScriptsRead, Permission::UsersRead, Permission::ApiKeysRead, Permission::BackfillRead, @@ -236,6 +246,7 @@ perms.insert(Permission::LexiconsCreate); perms.insert(Permission::LexiconsDelete); perms.insert(Permission::ScriptVariablesCreate); perms.insert(Permission::ScriptVariablesDelete); + perms.insert(Permission::ScriptsManage); perms.insert(Permission::RecordsDelete); perms.insert(Permission::LabelersCreate); perms.insert(Permission::LabelersRead); diff --git a/src/admin/scripts.rs b/src/admin/scripts.rs new file mode 100644 --- /dev/null +++ b/src/admin/scripts.rs @@ -0,0 +1,373 @@ +//! `/admin/scripts` CRUD — trigger-keyed scripts. +//! +//! Each script row's `id` IS its trigger string (e.g. +//! `record.create:com.example.thing`, `xrpc.query:com.foo.list`, +//! `labeler.apply:_actor`). The dispatcher in [`crate::lua::scripts`] +//! looks up scripts by id at firing time; this admin surface lets +//! operators CRUD those rows. +//! +//! Validation: +//! - On create / patch the body is parsed against the script_type +//! (lua → [`crate::lua::validate_script`]). Invalid bodies are +//! rejected at write-time with a 400. +//! - The trigger id is parsed against +//! [`crate::lua::ParsedTrigger::parse`]; unknown prefixes / invalid +//! NSIDs are rejected at write-time with a 400. +//! +//! Permissions: `scripts:read` for GETs; `scripts:manage` for the +//! mutating endpoints. + +use axum::Json; +use axum::extract::{Path, State}; +use axum::http::StatusCode; +use serde::{Deserialize, Serialize}; + +use crate::AppState; +use crate::db::{adapt_sql, now_rfc3339}; +use crate::error::AppError; +use crate::event_log::{EventLog, Severity, log_event}; +use crate::lua::{ParsedTrigger, ScriptLanguage}; + +use super::auth::UserAuth; +use super::permissions::Permission; + +// --------------------------------------------------------------------------- +// Wire types +// --------------------------------------------------------------------------- + +/// One row from the `scripts` table — what GET endpoints return. +#[derive(Debug, Clone, Serialize)] +pub(super) struct ScriptResponse { + /// The trigger id; identifies the row. + pub id: String, + pub script_type: String, + pub body: String, + pub description: Option, + pub created_at: String, + pub updated_at: String, +} + +/// Body for `POST /admin/scripts` (create or replace by `id`). +#[derive(Debug, Deserialize)] +pub(super) struct UpsertBody { + pub id: String, + /// Defaults to `"lua"` server-side if omitted. + #[serde(default)] + pub script_type: Option, + pub body: String, + #[serde(default)] + pub description: Option, +} + +/// Body for `PATCH /admin/scripts/{id}`. All fields optional. +#[derive(Debug, Deserialize)] +pub(super) struct PatchBody { + #[serde(default)] + pub script_type: Option, + #[serde(default)] + pub body: Option, + /// Set to `Some(None)` to clear via JSON `null`. + #[serde(default, deserialize_with = "deserialize_optional_field")] + pub description: Option>, +} + +/// Three-state field deserializer: missing → `None`, `null` → `Some(None)`, +/// string → `Some(Some(s))`. Lets PATCH distinguish "leave as-is" from +/// "clear to NULL". +fn deserialize_optional_field<'de, D>(d: D) -> Result>, D::Error> +where + D: serde::Deserializer<'de>, +{ + let v: Option = Option::deserialize(d)?; + Ok(Some(v)) +} + +// --------------------------------------------------------------------------- +// Routes +// --------------------------------------------------------------------------- + +/// `GET /admin/scripts` — list all rows. Clients group by trigger family +/// in the UI. +pub(super) async fn list( + State(state): State, + auth: UserAuth, +) -> Result>, AppError> { + auth.require(Permission::ScriptsRead).await?; + + let backend = state.db_backend; + let sql = adapt_sql( + "SELECT id, script_type, body, description, created_at, updated_at + FROM scripts + ORDER BY id", + backend, + ); + #[allow(clippy::type_complexity)] + let rows: Vec<(String, String, String, Option, String, String)> = sqlx::query_as(&sql) + .fetch_all(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to list scripts: {e}")))?; + + let scripts: Vec = rows + .into_iter() + .map( + |(id, script_type, body, description, created_at, updated_at)| ScriptResponse { + id, + script_type, + body, + description, + created_at, + updated_at, + }, + ) + .collect(); + + Ok(Json(scripts)) +} + +/// `GET /admin/scripts/{id}` — fetch one row. +pub(super) async fn get( + State(state): State, + auth: UserAuth, + Path(id): Path, +) -> Result, AppError> { + auth.require(Permission::ScriptsRead).await?; + fetch_one(&state, &id).await.map(Json) +} + +/// `POST /admin/scripts` — create or replace a row by `id`. Returns the +/// upserted row. Status `201 Created` for a new row, `200 OK` for an +/// update. +pub(super) async fn upsert( + State(state): State, + auth: UserAuth, + Json(body): Json, +) -> Result<(StatusCode, Json), AppError> { + auth.require(Permission::ScriptsManage).await?; + + // Validate the trigger id grammar up-front (400 with a clear message). + let _trigger = ParsedTrigger::parse(&body.id).map_err(AppError::BadRequest)?; + + let script_type = body.script_type.unwrap_or_default(); + validate_body_for_type(&body.body, script_type)?; + + let backend = state.db_backend; + let now = now_rfc3339(); + let description = body.description.as_deref().filter(|s| !s.is_empty()); + + // Distinguish create vs update so we can return 201 vs 200. + let pre_exists: Option<(String,)> = + sqlx::query_as(&adapt_sql("SELECT id FROM scripts WHERE id = ?", backend)) + .bind(&body.id) + .fetch_optional(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to check script existence: {e}")))?; + let was_new = pre_exists.is_none(); + + let sql = adapt_sql( + r#" + INSERT INTO scripts (id, script_type, body, description, created_at, updated_at) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT (id) DO UPDATE SET + script_type = EXCLUDED.script_type, + body = EXCLUDED.body, + description = EXCLUDED.description, + updated_at = EXCLUDED.updated_at + "#, + backend, + ); + sqlx::query(&sql) + .bind(&body.id) + .bind(script_type.as_str()) + .bind(&body.body) + .bind(description) + .bind(&now) + .bind(&now) + .execute(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to upsert script: {e}")))?; + + log_event( + &state.db, + EventLog { + event_type: if was_new { + "script.created".to_string() + } else { + "script.updated".to_string() + }, + severity: Severity::Info, + actor_did: Some(auth.did.clone()), + subject: Some(body.id.clone()), + detail: serde_json::json!({ + "script_type": script_type.as_str(), + }), + }, + backend, + ) + .await; + + let row = fetch_one(&state, &body.id).await?; + let status = if was_new { + StatusCode::CREATED + } else { + StatusCode::OK + }; + Ok((status, Json(row))) +} + +/// `PATCH /admin/scripts/{id}` — partial update. At least one of +/// `script_type` / `body` / `description` must be present. +pub(super) async fn patch( + State(state): State, + auth: UserAuth, + Path(id): Path, + Json(body): Json, +) -> Result, AppError> { + auth.require(Permission::ScriptsManage).await?; + + if body.script_type.is_none() && body.body.is_none() && body.description.is_none() { + return Err(AppError::BadRequest( + "patch requires at least one of: script_type, body, description".into(), + )); + } + // Patching a body or script_type? We need a body to validate against + // the (possibly new) language. Patching script_type alone is + // ambiguous (we'd be validating the existing body against the new + // language without re-checking it makes sense), so reject it. + if body.script_type.is_some() && body.body.is_none() { + return Err(AppError::BadRequest( + "patching script_type requires body alongside (so the server can re-validate)".into(), + )); + } + if let Some(ref new_body) = body.body { + let lang = body.script_type.unwrap_or_default(); + validate_body_for_type(new_body, lang)?; + } + + // Existence check + fetch current values. + let existing = fetch_one(&state, &id).await?; + + let backend = state.db_backend; + let now = now_rfc3339(); + let new_script_type = body + .script_type + .map(|s| s.as_str().to_string()) + .unwrap_or(existing.script_type); + let new_body = body.body.unwrap_or(existing.body); + let new_description = match body.description { + Some(desc_opt) => desc_opt, + None => existing.description, + }; + + let sql = adapt_sql( + r#" + UPDATE scripts + SET script_type = ?, + body = ?, + description = ?, + updated_at = ? + WHERE id = ? + "#, + backend, + ); + sqlx::query(&sql) + .bind(&new_script_type) + .bind(&new_body) + .bind(new_description.as_deref()) + .bind(&now) + .bind(&id) + .execute(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to patch script: {e}")))?; + + log_event( + &state.db, + EventLog { + event_type: "script.updated".to_string(), + severity: Severity::Info, + actor_did: Some(auth.did.clone()), + subject: Some(id.clone()), + detail: serde_json::json!({ + "script_type": new_script_type, + }), + }, + backend, + ) + .await; + + let row = fetch_one(&state, &id).await?; + Ok(Json(row)) +} + +/// `DELETE /admin/scripts/{id}` — remove a row. 204 on success, 404 if +/// no row matched. +pub(super) async fn delete( + State(state): State, + auth: UserAuth, + Path(id): Path, +) -> Result { + auth.require(Permission::ScriptsManage).await?; + + let backend = state.db_backend; + let sql = adapt_sql("DELETE FROM scripts WHERE id = ?", backend); + let result = sqlx::query(&sql) + .bind(&id) + .execute(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to delete script: {e}")))?; + if result.rows_affected() == 0 { + return Err(AppError::NotFound(format!("script '{id}' not found"))); + } + + log_event( + &state.db, + EventLog { + event_type: "script.deleted".to_string(), + severity: Severity::Info, + actor_did: Some(auth.did.clone()), + subject: Some(id), + detail: serde_json::json!({}), + }, + backend, + ) + .await; + Ok(StatusCode::NO_CONTENT) +} + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +/// Look up a single script row; 404 if missing. +async fn fetch_one(state: &AppState, id: &str) -> Result { + let backend = state.db_backend; + let sql = adapt_sql( + "SELECT id, script_type, body, description, created_at, updated_at + FROM scripts WHERE id = ?", + backend, + ); + #[allow(clippy::type_complexity)] + let row: Option<(String, String, String, Option, String, String)> = + sqlx::query_as(&sql) + .bind(id) + .fetch_optional(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to fetch script: {e}")))?; + let (id, script_type, body, description, created_at, updated_at) = + row.ok_or_else(|| AppError::NotFound(format!("script '{id}' not found")))?; + Ok(ScriptResponse { + id, + script_type, + body, + description, + created_at, + updated_at, + }) +} + +/// Validate the script body against its declared language. Rejects +/// invalid bodies with a 400 at write-time. +fn validate_body_for_type(body: &str, lang: ScriptLanguage) -> Result<(), AppError> { + match lang { + ScriptLanguage::Lua => crate::lua::validate_script(body).map_err(AppError::BadRequest), + } +} diff --git a/src/labeler.rs b/src/labeler.rs --- a/src/labeler.rs +++ b/src/labeler.rs @@ -253,7 +253,7 @@ last_seq = message.seq; for label in &message.labels { - apply_label(&state.db, label, state.db_backend).await; + apply_label(state, label).await; } events_since_cursor_save += 1; @@ -323,22 +323,53 @@ format!("ws://{base}") } } -async fn apply_label(db: &sqlx::AnyPool, label: &Label, backend: DatabaseBackend) { - if label.neg { +/// Persist a label received from a subscribed upstream labeler. +/// +/// Before touching the DB we run the trigger-keyed script chain (computed +/// from `label.uri` — `labeler.apply:` for at-uri subjects, +/// `labeler.apply:_actor` for bare DIDs). The script can rewrite any field +/// of the label (including `val` or `neg`) or return nil to skip +/// persistence. Failure is fail-open: a dead-lettered script proceeds with +/// the original label, so a buggy script can't permanently break the +/// firehose. +async fn apply_label(state: &AppState, label: &Label) { + let event = crate::lua::LabelAppliedEvent { + src: label.src.clone(), + uri: label.uri.clone(), + val: label.val.clone(), + neg: label.neg, + cts: label.cts.clone(), + exp: label.exp.clone(), + }; + let final_label = match crate::lua::run_label_applied_script(state, event).await { + crate::lua::LabelHookOutcome::Continue(next) => next, + crate::lua::LabelHookOutcome::Skip => { + tracing::debug!( + src = %label.src, uri = %label.uri, val = %label.val, + "label.applied script skipped persistence" + ); + return; + } + }; + + let db = &state.db; + let backend = state.db_backend; + + if final_label.neg { // Negation label — remove it. let delete_sql = adapt_sql( "DELETE FROM labels WHERE src = ? AND uri = ? AND val = ?", backend, ); if let Err(e) = sqlx::query(&delete_sql) - .bind(&label.src) - .bind(&label.uri) - .bind(&label.val) + .bind(&final_label.src) + .bind(&final_label.uri) + .bind(&final_label.val) .execute(db) .await { tracing::warn!( - src = %label.src, uri = %label.uri, val = %label.val, + src = %final_label.src, uri = %final_label.uri, val = %final_label.val, "failed to delete negated label: {e}" ); } @@ -356,16 +387,16 @@ backend, ); if let Err(e) = sqlx::query(&insert_sql) - .bind(&label.src) - .bind(&label.uri) - .bind(&label.val) - .bind(&label.cts) - .bind(&label.exp) + .bind(&final_label.src) + .bind(&final_label.uri) + .bind(&final_label.val) + .bind(&final_label.cts) + .bind(&final_label.exp) .execute(db) .await { tracing::warn!( - src = %label.src, uri = %label.uri, val = %label.val, + src = %final_label.src, uri = %final_label.uri, val = %final_label.val, "failed to upsert label: {e}" ); } @@ -452,7 +483,7 @@ let response: QueryLabelsResponse = resp.json().await?; for label in &response.labels { - apply_label(&state.db, label, state.db_backend).await; + apply_label(state, label).await; } Ok(()) diff --git a/src/lua/execute.rs b/src/lua/execute.rs --- a/src/lua/execute.rs +++ b/src/lua/execute.rs @@ -8,7 +8,7 @@ use std::time::Instant; use crate::AppState; use crate::auth::Claims; -use crate::db::{DatabaseBackend, adapt_sql, now_rfc3339}; +use crate::db::{DatabaseBackend, adapt_sql}; use crate::error::{AppError, ScriptErrorType, parse_lua_line}; use crate::event_log::{EventLog, Severity, log_event}; use crate::lexicon::ParsedLexicon; @@ -256,9 +256,9 @@ } if let Err(e) = record::register_record_api( &lua, - state_arc, - claims_arc, - pds_auth_arc, + state_arc.clone(), + Some(claims_arc), + Some(pds_auth_arc), delegate_did.map(|s| s.to_string()), ) { let error_message = format!("failed to register Record API: {e}"); @@ -627,7 +627,9 @@ .await; return Err(AppError::Internal(error_message)); } - if let Err(e) = atproto_api::register_atproto_api(&lua, state_arc, claims.map(|c| c.did())) { + if let Err(e) = + atproto_api::register_atproto_api(&lua, state_arc.clone(), claims.map(|c| c.did())) + { let error_message = format!("failed to register atproto API: {e}"); log_event( &state.db, @@ -649,6 +651,32 @@ .await; return Err(AppError::Internal(error_message)); } + // Register the Record API in no-auth mode. Queries don't have a PDS + // auth context — the local-only methods (Record.load, :save_local, + // :delete_local, Record.delete_local) work; PDS-touching variants + // error with the no-PDS-auth message. + if let Err(e) = record::register_record_api_no_auth(&lua, state_arc) { + let error_message = format!("failed to register Record API: {e}"); + log_event( + &state.db, + EventLog { + event_type: "script.error".to_string(), + severity: Severity::Error, + actor_did: None, + subject: Some(method.to_string()), + detail: serde_json::json!({ + "error": error_message, + "script_source": script_source, + "method": method, + "duration_ms": start.elapsed().as_millis() as u64, + }), + }, + backend, + ) + .await; + return Err(AppError::Internal(error_message)); + } + if let Err(e) = context::set_query_context( &lua, method, @@ -860,433 +888,3 @@ .await; Ok(Json(json_value).into_response()) } - -/// Context for a hook execution triggered by a record index event. -pub struct HookEvent<'a> { - pub state: &'a AppState, - pub lexicon_id: &'a str, - pub script: &'a str, - pub action: &'a str, - pub uri: &'a str, - pub did: &'a str, - pub collection: &'a str, - pub rkey: &'a str, - pub record: Option<&'a Value>, -} - -/// Execute a Lua hook script triggered by a record index event. -/// -/// Runs **before** the record is indexed. The return value determines what -/// gets stored: -/// - `None` → skip the DB operation entirely -/// - `Some(value)` → use that value for the insert/update -/// -/// Retries up to 3 times with exponential backoff (1s, 2s, 4s). -/// On final failure, dead-letters the event and returns `Some(original_record)` -/// (fail-open so indexing is not permanently blocked). -pub async fn execute_hook_script(event: &HookEvent<'_>) -> Option { - let max_attempts: i32 = 4; // 1 initial + 3 retries - let mut last_error = String::new(); - let backend = event.state.db_backend; - - for attempt in 0..max_attempts { - if attempt > 0 { - let delay = std::time::Duration::from_secs(1 << (attempt - 1)); // 1s, 2s, 4s - tokio::time::sleep(delay).await; - } - - match run_hook_once(event).await { - Ok(hook_result) => { - log_event( - &event.state.db, - EventLog { - event_type: "hook.executed".to_string(), - severity: Severity::Info, - actor_did: None, - subject: Some(event.uri.to_string()), - detail: serde_json::json!({ - "lexicon_id": event.lexicon_id, - "action": event.action, - "collection": event.collection, - "attempts": attempt + 1, - }), - }, - backend, - ) - .await; - return hook_result; - } - Err(e) => { - last_error = e; - tracing::warn!( - uri = event.uri, - lexicon_id = event.lexicon_id, - attempt = attempt + 1, - "hook execution failed: {last_error}" - ); - } - } - } - - // All retries exhausted — dead-letter the event and fail-open with the - // original record so indexing is not permanently blocked. - tracing::error!( - uri = event.uri, - lexicon_id = event.lexicon_id, - "hook dead-lettered after {max_attempts} attempts" - ); - - let record_str = event - .record - .map(|r| serde_json::to_string(r).unwrap_or_default()); - let dead_letter_sql = adapt_sql( - r#" - INSERT INTO dead_letter_hooks (lexicon_id, uri, did, collection, rkey, action, record, error, attempts, created_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) - "#, - backend, - ); - if let Err(e) = sqlx::query(&dead_letter_sql) - .bind(event.lexicon_id) - .bind(event.uri) - .bind(event.did) - .bind(event.collection) - .bind(event.rkey) - .bind(event.action) - .bind(&record_str) - .bind(&last_error) - .bind(max_attempts) - .bind(now_rfc3339()) - .execute(&event.state.db) - .await - { - tracing::error!(uri = event.uri, "failed to insert dead letter hook: {e}"); - } - - log_event( - &event.state.db, - EventLog { - event_type: "hook.dead_lettered".to_string(), - severity: Severity::Error, - actor_did: None, - subject: Some(event.uri.to_string()), - detail: serde_json::json!({ - "lexicon_id": event.lexicon_id, - "action": event.action, - "collection": event.collection, - "error": last_error, - "attempts": max_attempts, - }), - }, - backend, - ) - .await; - - // Fail-open: return the original record so indexing proceeds. - event.record.cloned() -} - -/// Execute a hook script once. -/// -/// Returns `Ok(None)` when `handle()` returns nil (meaning "skip indexing"), -/// `Ok(Some(value))` when it returns a table (use that as the record), or -/// `Ok(Some(original))` for other non-nil types. -pub async fn run_hook_once(event: &HookEvent<'_>) -> Result, String> { - let lua = sandbox::create_sandbox().map_err(|e| format!("failed to create Lua VM: {e}"))?; - let backend = event.state.db_backend; - - let state_arc = Arc::new(event.state.clone()); - - db_api::register_db_api(&lua, state_arc.clone()) - .map_err(|e| format!("failed to register db API: {e}"))?; - - http_api::register_http_api(&lua, state_arc.clone()) - .map_err(|e| format!("failed to register http API: {e}"))?; - - super::xrpc_api::register_xrpc_api(&lua, state_arc.clone(), Some(event.did.to_string())) - .map_err(|e| format!("failed to register xrpc API: {e}"))?; - - atproto_api::register_atproto_api(&lua, state_arc, None) - .map_err(|e| format!("failed to register atproto API: {e}"))?; - - context::set_hook_context( - &lua, - event.action, - event.uri, - event.did, - event.collection, - event.rkey, - event.record, - ) - .map_err(|e| format!("failed to set hook context: {e}"))?; - - context::set_env_context(&lua, &load_env_vars(&event.state.db, backend).await) - .map_err(|e| format!("failed to set env context: {e}"))?; - - lua.load(event.script) - .exec() - .map_err(|e| format!("script load failed: {e}"))?; - - let handle: mlua::Function = lua - .globals() - .get("handle") - .map_err(|e| format!("script missing handle function: {e}"))?; - - let result: mlua::Value = handle - .call_async::(()) - .await - .map_err(|e| e.to_string())?; - - match result { - mlua::Value::Nil => Ok(None), - mlua::Value::Table(_) => { - let json_value: Value = lua - .from_value(result) - .map_err(|e| format!("failed to convert lua table to JSON: {e}"))?; - Ok(Some(json_value)) - } - _ => { - // Non-nil, non-table return — proceed with the original record. - Ok(event.record.cloned()) - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::config::Config; - use crate::db::DatabaseBackend; - use crate::lexicon::LexiconRegistry; - use serde_json::json; - use tokio::sync::watch; - - fn test_state() -> AppState { - let config = Config { - host: "127.0.0.1".into(), - port: 3000, - database_url: String::new(), - database_backend: crate::db::DatabaseBackend::Sqlite, - public_url: String::new(), - session_secret: "test-secret".into(), - jetstream_url: String::new(), - relay_url: String::new(), - plc_url: String::new(), - static_dir: String::new(), - event_log_retention_days: 30, - app_name: None, - logo_uri: None, - tos_uri: None, - policy_uri: None, - token_encryption_key: None, - default_rate_limit_capacity: 100, - default_rate_limit_refill_rate: 2.0, - }; - let (tx, _) = watch::channel(vec![]); - let (labeler_tx, _) = watch::channel(()); - sqlx::any::install_default_drivers(); - let test_db = sqlx::AnyPool::connect_lazy("sqlite::memory:").unwrap(); - let atrium_http = std::sync::Arc::new(atrium_oauth::DefaultHttpClient::default()); - let did_resolver = atrium_identity::did::CommonDidResolver::new( - atrium_identity::did::CommonDidResolverConfig { - plc_directory_url: "https://plc.directory".into(), - http_client: std::sync::Arc::clone(&atrium_http), - }, - ); - let handle_resolver = atrium_identity::handle::AtprotoHandleResolver::new( - atrium_identity::handle::AtprotoHandleResolverConfig { - dns_txt_resolver: crate::dns::NativeDnsResolver::new(), - http_client: atrium_http, - }, - ); - let oauth = atrium_oauth::OAuthClient::new(atrium_oauth::OAuthClientConfig { - client_metadata: atrium_oauth::AtprotoLocalhostClientMetadata { - redirect_uris: Some(vec!["http://127.0.0.1:0/auth/callback".into()]), - scopes: Some(vec![atrium_oauth::Scope::Known( - atrium_oauth::KnownScope::Atproto, - )]), - }, - keys: None, - state_store: crate::auth::oauth_store::DbStateStore::new( - test_db.clone(), - crate::db::DatabaseBackend::Sqlite, - ), - session_store: crate::auth::oauth_store::DbSessionStore::new( - test_db.clone(), - crate::db::DatabaseBackend::Sqlite, - ), - resolver: atrium_oauth::OAuthResolverConfig { - did_resolver, - handle_resolver, - authorization_server_metadata: Default::default(), - protected_resource_metadata: Default::default(), - }, - }) - .expect("Failed to create test OAuth client"); - AppState { - config, - http: reqwest::Client::new(), - db: test_db.clone(), - db_backend: DatabaseBackend::Sqlite, - domain_cache: crate::domain::DomainCache::new(), - lexicons: LexiconRegistry::new(), - collections_tx: tx, - labeler_subscriptions_tx: labeler_tx, - rate_limiter: crate::rate_limit::RateLimiter::new( - crate::rate_limit::RateLimitDefaults { - query_cost: 1, - procedure_cost: 1, - proxy_cost: 1, - }, - ), - oauth: std::sync::Arc::new(crate::auth::OAuthClientRegistry::new(std::sync::Arc::new( - oauth, - ))), - oauth_state_store: crate::auth::oauth_store::DbStateStore::new( - test_db.clone(), - crate::db::DatabaseBackend::Sqlite, - ), - cookie_key: axum_extra::extract::cookie::Key::derive_from( - b"test-secret-for-tests-only-not-production", - ), - plugin_registry: std::sync::Arc::new(crate::plugin::PluginRegistry::new()), - wasm_runtime: std::sync::Arc::new( - crate::plugin::WasmRuntime::new().expect("wasm runtime"), - ), - attestation_signer: None, - official_registry: std::sync::Arc::new(tokio::sync::RwLock::new( - crate::plugin::official_registry::OfficialRegistryState::default(), - )), - official_registry_config: crate::plugin::official_registry::RegistryConfig::production( - ), - proxy_config: std::sync::Arc::new(arc_swap::ArcSwap::new(std::sync::Arc::new( - crate::proxy_config::ProxyConfig::default(), - ))), - } - } - - fn make_event<'a>( - state: &'a AppState, - script: &'a str, - action: &'a str, - record: Option<&'a Value>, - ) -> HookEvent<'a> { - HookEvent { - state, - lexicon_id: "test.lexicon", - script, - action, - uri: "at://did:plc:test/test.collection/rkey1", - did: "did:plc:test", - collection: "test.collection", - rkey: "rkey1", - record, - } - } - - #[tokio::test] - async fn hook_runs_simple_script() { - let state = test_state(); - let event = make_event(&state, "function handle() end", "create", None); - let result = run_hook_once(&event).await; - assert!(result.is_ok(), "expected Ok, got: {:?}", result); - // handle() returns nil implicitly, so result should be None (skip). - assert!(result.unwrap().is_none()); - } - - #[tokio::test] - async fn hook_returns_nil_to_skip() { - let state = test_state(); - let record = json!({"name": "Test"}); - let event = make_event( - &state, - "function handle() return nil end", - "create", - Some(&record), - ); - let result = run_hook_once(&event).await; - assert!(result.is_ok(), "expected Ok, got: {:?}", result); - assert!(result.unwrap().is_none(), "nil return should produce None"); - } - - #[tokio::test] - async fn hook_returns_modified_record() { - let state = test_state(); - let record = json!({"name": "Original"}); - let script = r#" - function handle() - return { name = "Modified", extra = true } - end - "#; - let event = make_event(&state, script, "create", Some(&record)); - let result = run_hook_once(&event).await; - assert!(result.is_ok(), "expected Ok, got: {:?}", result); - let value = result.unwrap(); - assert!(value.is_some(), "table return should produce Some"); - let v = value.unwrap(); - assert_eq!(v["name"], "Modified"); - assert_eq!(v["extra"], true); - } - - #[tokio::test] - async fn hook_fails_on_missing_handle() { - let state = test_state(); - let event = make_event(&state, "function other() end", "create", None); - let result = run_hook_once(&event).await; - assert!(result.is_err()); - let err = result.unwrap_err(); - assert!(err.contains("handle"), "expected handle error, got: {err}"); - } - - #[tokio::test] - async fn hook_fails_on_syntax_error() { - let state = test_state(); - let event = make_event(&state, "function handle(", "create", None); - let result = run_hook_once(&event).await; - assert!(result.is_err()); - } - - #[tokio::test] - async fn hook_has_access_to_context_globals() { - let state = test_state(); - let script = r#" - function handle() - if action ~= "create" then error("wrong action: " .. tostring(action)) end - if uri ~= "at://did:plc:test/test.collection/rkey1" then error("wrong uri") end - if did ~= "did:plc:test" then error("wrong did") end - if collection ~= "test.collection" then error("wrong collection") end - if rkey ~= "rkey1" then error("wrong rkey") end - end - "#; - let event = make_event(&state, script, "create", None); - let result = run_hook_once(&event).await; - assert!(result.is_ok(), "expected Ok, got: {:?}", result); - } - - #[tokio::test] - async fn hook_has_access_to_record() { - let state = test_state(); - let record = json!({"name": "Test"}); - let script = r#" - function handle() - if record.name ~= "Test" then error("wrong name: " .. tostring(record.name)) end - end - "#; - let event = make_event(&state, script, "create", Some(&record)); - let result = run_hook_once(&event).await; - assert!(result.is_ok(), "expected Ok, got: {:?}", result); - } - - #[tokio::test] - async fn hook_record_nil_on_delete() { - let state = test_state(); - let script = r#" - function handle() - if record ~= nil then error("expected nil record") end - end - "#; - let event = make_event(&state, script, "delete", None); - let result = run_hook_once(&event).await; - assert!(result.is_ok(), "expected Ok, got: {:?}", result); - } -} diff --git a/src/lua/mod.rs b/src/lua/mod.rs --- a/src/lua/mod.rs +++ b/src/lua/mod.rs @@ -3,14 +3,18 @@ mod context; pub mod db_api; mod execute; mod http_api; -mod record; +pub mod record; pub(crate) mod sandbox; +pub mod scripts; mod tid; mod xrpc_api; #[allow(unused_imports)] pub(crate) use context::SpaceContext; -pub(crate) use execute::{ - HookEvent, execute_hook_script, execute_procedure_script, execute_query_script, run_hook_once, -}; +pub(crate) use execute::{execute_procedure_script, execute_query_script}; pub(crate) use sandbox::validate_script; +pub use scripts::{ + LabelAppliedEvent, LabelHookOutcome, ParsedTrigger, ResolvedScript, ScriptLanguage, ScriptRow, + TriggerKind, resolve, resolve_record_event, run_label_applied_script, run_record_event_once, + run_record_event_script, trigger_for_label_uri, +}; diff --git a/src/lua/record.rs b/src/lua/record.rs --- a/src/lua/record.rs +++ b/src/lua/record.rs @@ -21,16 +21,44 @@ "_rkey", "_repo_override", ]; +/// Error message returned when a script calls a PDS-touching method +/// (`:save()` / `:delete()` / `Record.save_all`) from a context without +/// caller credentials — e.g. label scripts, record-event scripts, or +/// query handlers. +const NO_PDS_AUTH_MSG: &str = "no PDS auth in this script context — \ + use :save_local() / :delete_local() / Record.delete_local(uri) for local-only mutation"; + +/// Register the `Record` global with only the local-only surface +/// (`Record.load`, `:save_local`, `:delete_local`, `Record.delete_local`). +/// PDS-touching methods (`:save`, `:delete`, `Record.save_all`) are still +/// exposed but error with [`NO_PDS_AUTH_MSG`] when called. +/// +/// This is the entry point for label scripts, record-event scripts, and +/// query handlers — contexts that have no caller credentials to round-trip +/// records through a PDS. +pub fn register_record_api_no_auth(lua: &Lua, state: Arc) -> LuaResult<()> { + register_record_api(lua, state, None, None, None) +} + /// Register the `Record` global constructor and static methods. -/// Only registered for procedure scripts (not queries). /// -/// When `delegate_did` is `Some`, record writes default to the delegate's repo -/// instead of the caller's DID. Scripts can still override via `record:set_repo()`. -pub fn register_record_api( +/// `claims` / `pds_auth` are optional: when both are `Some`, the full +/// surface (`:save()`, `:delete()`, `Record.save_all`) round-trips through +/// the PDS. When either is `None` (label scripts, record-event scripts, +/// query handlers), only the local-only methods are usable +/// (`:save_local()`, `:delete_local()`, `Record.delete_local(uri)`); the +/// PDS-touching methods error with [`NO_PDS_AUTH_MSG`]. Most callers want +/// [`register_record_api_no_auth`] instead — this lower-level entry point +/// exposes the internal `PdsAuth` type and is only public to the crate. +/// +/// When `delegate_did` is `Some`, record writes default to the delegate's +/// repo instead of the caller's DID. Scripts can still override via +/// `record:set_repo()`. +pub(crate) fn register_record_api( lua: &Lua, state: Arc, - claims: Arc, - pds_auth: Arc, + claims: Option>, + pds_auth: Option>, delegate_did: Option, ) -> LuaResult<()> { // -- methods table (shared by all Record instances) -- @@ -48,6 +76,8 @@ let claims = claims.clone(); let pds_auth = pds_auth.clone(); let delegate_did = delegate_did.clone(); async move { + let claims = claims.ok_or_else(|| mlua::Error::runtime(NO_PDS_AUTH_MSG))?; + let pds_auth = pds_auth.ok_or_else(|| mlua::Error::runtime(NO_PDS_AUTH_MSG))?; let backend = state.db_backend; let collection: String = this.raw_get("_collection")?; let schema: mlua::Value = this.raw_get("_schema")?; @@ -223,6 +253,8 @@ let claims = claims.clone(); let pds_auth = pds_auth.clone(); let delegate_did = delegate_did.clone(); async move { + let claims = claims.ok_or_else(|| mlua::Error::runtime(NO_PDS_AUTH_MSG))?; + let pds_auth = pds_auth.ok_or_else(|| mlua::Error::runtime(NO_PDS_AUTH_MSG))?; let backend = state.db_backend; let uri: String = this.raw_get::>("_uri")?.ok_or_else(|| { mlua::Error::runtime("cannot delete a Record that has no _uri") @@ -246,24 +278,39 @@ "collection": collection, "rkey": rkey, }); - let resp = pds_auth + // Try the PDS delete. We log-and-continue on failure so the + // operator's intent ("remove this record") is still + // reflected in the local DB even when the PDS is down or + // refuses the call. The local row is the source of truth + // for the index. + match pds_auth .post_json(&state, repo, "com.atproto.repo.deleteRecord", &pds_body) .await - .map_err(|e| mlua::Error::runtime(format!("PDS deleteRecord failed: {e}")))?; - - if !resp.status().is_success() { - let status = resp.status(); - let body = resp.text().await.unwrap_or_default(); - return Err(mlua::Error::runtime(format!( - "PDS deleteRecord returned {status}: {body}" - ))); + { + Ok(resp) if resp.status().is_success() => {} + Ok(resp) => { + let status = resp.status(); + let body = resp.text().await.unwrap_or_default(); + tracing::warn!( + uri = %uri, + "PDS deleteRecord returned {status}: {body} \ + — proceeding with local delete anyway" + ); + } + Err(e) => { + tracing::warn!( + uri = %uri, + "PDS deleteRecord failed: {e} \ + — proceeding with local delete anyway" + ); + } } - // Delete from local DB + // Always delete locally — operator's logical action is + // "remove this record from view" regardless of PDS outcome. let delete_sql = adapt_sql("DELETE FROM records WHERE uri = ?", backend); let _ = sqlx::query(&delete_sql).bind(&uri).execute(&state.db).await; - // Clear _uri and _cid this.raw_set("_uri", mlua::Value::Nil)?; this.raw_set("_cid", mlua::Value::Nil)?; @@ -273,6 +320,156 @@ })?; methods.set("delete", delete_fn)?; } + // :save_local() — upsert into local DB only, never touches a PDS. + // Works in any script context (no auth required). + // + // For records loaded via `Record.load(uri)` or saved via `:save()` + // (i.e. those with an `_uri`), the repo+rkey are parsed back out of + // the URI. For brand-new records (no `_uri`), we fall back to + // `_repo_override` / `claims.did()` for the repo, and generate an + // rkey via `_key_type` if `_rkey` isn't set. Errors clearly when no + // DID can be determined. + { + let state = state.clone(); + let claims = claims.clone(); + let save_local_fn = lua.create_async_function(move |lua, this: mlua::Table| { + let state = state.clone(); + let claims = claims.clone(); + async move { + let backend = state.db_backend; + let collection: String = this.raw_get("_collection")?; + let schema: mlua::Value = this.raw_get("_schema")?; + + if let mlua::Value::Table(ref schema_table) = schema { + validate_required_fields(&this, schema_table)?; + } + + let data = extract_record_data(&lua, &this, &collection)?; + let data_str = serde_json::to_string(&data).unwrap_or_default(); + let now = now_rfc3339(); + + let existing_uri: Option = this.raw_get("_uri")?; + let (uri, repo, rkey) = if let Some(uri) = existing_uri { + // Parse repo (DID) and rkey out of the URI: + // at://// + let trimmed = uri + .strip_prefix("at://") + .ok_or_else(|| mlua::Error::runtime(format!("invalid AT URI: {uri}")))?; + let mut parts = trimmed.splitn(3, '/'); + let repo = parts + .next() + .ok_or_else(|| mlua::Error::runtime(format!("invalid AT URI: {uri}")))? + .to_string(); + let _col = parts.next(); + let rkey = parts + .next() + .ok_or_else(|| mlua::Error::runtime(format!("invalid AT URI: {uri}")))? + .to_string(); + (uri, repo, rkey) + } else { + // CREATE path — no URI yet. Compute repo + rkey, build URI. + let repo_override: Option = this.raw_get("_repo_override")?; + let repo = repo_override + .or_else(|| claims.as_ref().map(|c| c.did().to_string())) + .ok_or_else(|| { + mlua::Error::runtime( + "save_local() needs a DID — call :set_repo(\"did:plc:...\") \ + or load the record first", + ) + })?; + + let rkey: Option = this.raw_get("_rkey")?; + let rkey = if let Some(rk) = rkey { + rk + } else { + let key_type: Option = this.raw_get("_key_type")?; + match key_type.as_deref() { + Some("tid") | Some("any") | None => generate_tid(), + Some(s) if s.starts_with("literal:") => { + s["literal:".len()..].to_string() + } + Some("nsid") => { + return Err(mlua::Error::runtime( + "cannot auto-generate rkey for nsid key type — \ + call set_rkey() first", + )); + } + Some(other) => { + return Err(mlua::Error::runtime(format!( + "unknown key type '{other}'" + ))); + } + } + }; + let uri = format!("at://{repo}/{collection}/{rkey}"); + (uri, repo, rkey) + }; + + // Upsert. Sentinel CID `""` — no PDS round-trip means we + // have no real CID to record; consumers reading the row + // should treat empty CID as "local-only write". + let upsert_sql = adapt_sql( + r#"INSERT INTO records (uri, did, collection, rkey, record, cid, indexed_at, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT (uri) DO UPDATE + SET record = EXCLUDED.record, + cid = EXCLUDED.cid, + indexed_at = ?"#, + backend, + ); + sqlx::query(&upsert_sql) + .bind(&uri) + .bind(&repo) + .bind(&collection) + .bind(&rkey) + .bind(&data_str) + .bind("") + .bind(&now) + .bind(&now) + .bind(&now) + .execute(&state.db) + .await + .map_err(|e| mlua::Error::runtime(format!("save_local upsert failed: {e}")))?; + + let _ = sync_refs(&state.db, &uri, &collection, &data, backend).await; + + this.raw_set("_uri", uri.as_str())?; + this.raw_set("_cid", "")?; + + Ok(this) + } + })?; + methods.set("save_local", save_local_fn)?; + } + + // :delete_local() — local DB delete only, never touches a PDS. + // Idempotent: succeeds whether or not a row existed at the URI. + { + let state = state.clone(); + let delete_local_fn = lua.create_async_function(move |_lua, this: mlua::Table| { + let state = state.clone(); + async move { + let backend = state.db_backend; + let uri: String = this.raw_get::>("_uri")?.ok_or_else(|| { + mlua::Error::runtime("cannot delete_local a Record that has no _uri") + })?; + + let delete_sql = adapt_sql("DELETE FROM records WHERE uri = ?", backend); + sqlx::query(&delete_sql) + .bind(&uri) + .execute(&state.db) + .await + .map_err(|e| mlua::Error::runtime(format!("delete_local failed: {e}")))?; + + this.raw_set("_uri", mlua::Value::Nil)?; + this.raw_set("_cid", mlua::Value::Nil)?; + + Ok(this) + } + })?; + methods.set("delete_local", delete_local_fn)?; + } + // :set_key_type(type) { let set_key_type_fn = @@ -461,6 +658,8 @@ let claims = claims.clone(); let pds_auth = pds_auth.clone(); let delegate_did = delegate_did.clone(); async move { + let claims = claims.ok_or_else(|| mlua::Error::runtime(NO_PDS_AUTH_MSG))?; + let pds_auth = pds_auth.ok_or_else(|| mlua::Error::runtime(NO_PDS_AUTH_MSG))?; let backend = state.db_backend; // Extract save data from each record (sync) type SaveItem = (mlua::Table, String, Option, Option, Option, Value); @@ -722,8 +921,8 @@ } // Record.load_all(uris) { - let state = state; - let metatable_c = metatable; + let state = state.clone(); + let metatable_c = metatable.clone(); let load_all_fn = lua.create_async_function(move |lua, uris_table: mlua::Table| { let state = state.clone(); let metatable = metatable_c.clone(); @@ -802,6 +1001,31 @@ Ok(mlua::Value::Table(out)) } })?; record_table.set("load_all", load_all_fn)?; + } + + // Record.delete_local(uri) — fire-and-forget local-only delete by URI. + // The common one-liner for label-script reactions like: + // if event.val == "spam" then Record.delete_local(event.uri) end + // Returns true if a row was deleted, false if no row matched. + // Always succeeds (no error) regardless of whether the row existed. + { + let state = state; + let delete_local_static_fn = lua.create_async_function(move |_lua, uri: String| { + let state = state.clone(); + async move { + let backend = state.db_backend; + let delete_sql = adapt_sql("DELETE FROM records WHERE uri = ?", backend); + let res = sqlx::query(&delete_sql) + .bind(&uri) + .execute(&state.db) + .await + .map_err(|e| { + mlua::Error::runtime(format!("Record.delete_local failed: {e}")) + })?; + Ok(res.rows_affected() > 0) + } + })?; + record_table.set("delete_local", delete_local_static_fn)?; } // -- Make Record callable via __call metamethod -- diff --git a/src/lua/scripts.rs b/src/lua/scripts.rs new file mode 100644 --- /dev/null +++ b/src/lua/scripts.rs @@ -0,0 +1,855 @@ +//! 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:` — fires for any record event on `` +//! (wildcard fallback). +//! - `record.create:` / `record.update:` / +//! `record.delete:` — fires only for that specific action. +//! - `xrpc.query:` / `xrpc.procedure:` — fires when the +//! matching XRPC method is invoked. +//! - `labeler.apply:` — fires when a label arrives whose `uri` +//! is `at:////`. +//! - `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.:` first, falls back to `record.index:` +//! 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 serde::{Deserialize, Serialize}; +use serde_json::Value; +use std::sync::Arc; + +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, +} + +/// A trigger id parsed into `(kind, suffix)`. The suffix is either an NSID +/// (`record.*`, `xrpc.*`, `labeler.apply:`) 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), + } + } + + /// 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 { + let (prefix, suffix) = id.split_once(':').ok_or_else(|| { + format!( + "trigger id '{id}' must contain a ':' separator; \ + valid prefixes: record.{{index,create,update,delete}}:, \ + xrpc.{{query,procedure}}:, labeler.apply:" + ) + })?; + + 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, + other => { + return Err(format!( + "unknown trigger prefix '{other}'; valid prefixes: \ + record.{{index,create,update,delete}}, xrpc.{{query,procedure}}, \ + labeler.apply" + )); + } + }; + + // Suffix validation: NSID for everything except `labeler.apply:_actor`. + match (kind, suffix) { + (TriggerKind::LabelerApply, "_actor") => {} + _ => validate_nsid(suffix)?, + } + + Ok(Self { + kind, + suffix: suffix.to_string(), + }) + } +} + +/// 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 { + 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, + 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 { + let sql = adapt_sql( + "SELECT id, body, script_type FROM scripts WHERE id = ?", + state.db_backend, + ); + let row: Option<(String, String, String)> = match sqlx::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.:` first, then `record.index:`. +pub async fn resolve_record_event( + state: &AppState, + nsid: &str, + action: &str, +) -> Option { + 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) +// --------------------------------------------------------------------------- + +/// 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. +#[allow(clippy::too_many_arguments)] +pub async fn run_record_event_script( + state: &AppState, + nsid: &str, + action: &str, + uri: &str, + did: &str, + rkey: &str, + record: Option<&Value>, +) -> Option { + let resolved = match resolve_record_event(state, nsid, action).await { + Some(s) => s, + // No script for this trigger → indexer keeps the original record. + None => return record.cloned(), + }; + + let host_id = format!("{nsid}:{action}"); + let payload = serde_json::json!({ + "trigger": resolved.id, + "action": action, + "uri": uri, + "did": did, + "collection": nsid, + "rkey": rkey, + "record": 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, action, uri, did, nsid, rkey, record).await { + Ok(outcome) => { + log_event( + &state.db, + EventLog { + event_type: "script.executed".to_string(), + severity: Severity::Info, + actor_did: None, + subject: Some(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, + trigger = %resolved.id, + attempt = attempt + 1, + "record script attempt failed: {last_error}" + ); + } + } + } + + write_dead_letter( + state, + &resolved, + "record", + &host_id, + &payload, + &last_error, + MAX_ATTEMPTS, + ) + .await; + log_event( + &state.db, + EventLog { + event_type: "script.dead_lettered".to_string(), + severity: Severity::Error, + actor_did: None, + subject: Some(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. + 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. +#[allow(clippy::too_many_arguments)] +pub async fn run_record_event_once( + state: &AppState, + script: &ResolvedScript, + action: &str, + uri: &str, + did: &str, + collection: &str, + rkey: &str, + record: Option<&Value>, +) -> Result, 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, Some(did))?; + + // Legacy globals (action, uri, did, collection, rkey, record) for + // back-compat with scripts written against the old `index_hook` + // surface. + context::set_hook_context(&lua, action, uri, did, collection, rkey, 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": action, + "uri": uri, + "did": did, + "collection": collection, + "rkey": rkey, + "record": 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::(()) + .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(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, +} + +/// 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:` (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) => 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}" + ); + } + } + } + write_dead_letter( + state, + &resolved, + "label", + &host_id, + &payload, + &last_error, + MAX_ATTEMPTS, + ) + .await; + LabelHookOutcome::Continue(original) +} + +async fn run_label_lua_once( + state: &AppState, + script: &ResolvedScript, + event: &LabelAppliedEvent, +) -> Result { + 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, 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::(()) + .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 { + v.get(key).and_then(|x| x.as_str()).map(String::from) +} + +fn extract_bool(v: &Value, key: &str) -> Option { + 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, + 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}"))?; + record::register_record_api_no_auth(lua, state.clone()) + .map_err(|e| format!("record api: {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 { + let sql = adapt_sql("SELECT key, value FROM script_variables", backend); + sqlx::query_as::<_, (String, String)>(&sql) + .fetch_all(db) + .await + .unwrap_or_default() + .into_iter() + .collect() +} + +/// Persist a permanently-failed run for later admin triage. +async fn write_dead_letter( + state: &AppState, + script: &ResolvedScript, + host_kind: &str, + host_id: &str, + payload: &Value, + error: &str, + attempts: u32, +) { + let payload_str = serde_json::to_string(payload).unwrap_or_else(|_| "{}".to_string()); + let sql = adapt_sql( + "INSERT INTO dead_letter_scripts + (script_ref, host_kind, host_id, payload, error, attempts, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?)", + state.db_backend, + ); + if let Err(e) = sqlx::query(&sql) + .bind(script.id.as_str()) + .bind(host_kind) + .bind(host_id) + .bind(&payload_str) + .bind(error) + .bind(attempts as i64) + .bind(now_rfc3339()) + .execute(&state.db) + .await + { + tracing::error!( + host_kind, + host_id, + trigger = %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 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:///` → 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); + } +} diff --git a/src/lua/xrpc_api.rs b/src/lua/xrpc_api.rs --- a/src/lua/xrpc_api.rs +++ b/src/lua/xrpc_api.rs @@ -214,7 +214,13 @@ }; 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(); + // Single-connection pool so the in-memory DB is shared across the + // pool's queries (separate connections to `sqlite::memory:` get + // independent DBs otherwise). + let test_db = sqlx::pool::PoolOptions::::new() + .max_connections(1) + .connect_lazy("sqlite::memory:") + .unwrap(); let atrium_http = std::sync::Arc::new(atrium_oauth::DefaultHttpClient::default()); let did_resolver = atrium_identity::did::CommonDidResolver::new( atrium_identity::did::CommonDidResolverConfig { @@ -292,6 +298,38 @@ proxy_config: std::sync::Arc::new(arc_swap::ArcSwap::new(std::sync::Arc::new( crate::proxy_config::ProxyConfig::default(), ))), } + } + + /// Create the `scripts` table on the in-memory test DB and insert one + /// row keyed by trigger id. The trigger-keyed dispatcher reads from + /// here at firing time; `make_*_lexicon`'s `script` field is now + /// inert (kept on the struct for forward-compat with row loaders but + /// not consulted by dispatch). + async fn seed_script(state: &AppState, trigger: &str, body: &str) { + sqlx::query( + r#" + CREATE TABLE IF NOT EXISTS scripts ( + id TEXT PRIMARY KEY, + body TEXT NOT NULL, + description TEXT, + script_type TEXT NOT NULL DEFAULT 'lua', + created_at TEXT NOT NULL DEFAULT (datetime('now')), + updated_at TEXT NOT NULL DEFAULT (datetime('now')) + ) + "#, + ) + .execute(&state.db) + .await + .unwrap(); + sqlx::query( + "INSERT OR REPLACE INTO scripts (id, body, script_type, created_at, updated_at) + VALUES (?, ?, 'lua', datetime('now'), datetime('now'))", + ) + .bind(trigger) + .bind(body) + .execute(&state.db) + .await + .unwrap(); } fn make_query_lexicon(id: &str, script: Option<&str>) -> ParsedLexicon { @@ -421,11 +459,14 @@ async fn query_local_script_returns_json() { let state = test_state(); // Register a scripted query that returns a static response - let lexicon = make_query_lexicon( - "test.echo", - Some(r#"function handle() return { greeting = "hello" } end"#), - ); + let lexicon = make_query_lexicon("test.echo", None); state.lexicons.upsert(lexicon).await; + seed_script( + &state, + "xrpc.query:test.echo", + r#"function handle() return { greeting = "hello" } end"#, + ) + .await; let mut params = HashMap::new(); let result = execute_local_query(&state, "test.echo", &mut params, None).await; @@ -443,11 +484,14 @@ #[tokio::test] async fn query_local_script_receives_params() { let state = test_state(); - let lexicon = make_query_lexicon( - "test.greet", - Some(r#"function handle() return { greeting = "hello " .. params.name } end"#), - ); + let lexicon = make_query_lexicon("test.greet", None); state.lexicons.upsert(lexicon).await; + seed_script( + &state, + "xrpc.query:test.greet", + r#"function handle() return { greeting = "hello " .. params.name } end"#, + ) + .await; let mut params = HashMap::new(); params.insert("name".into(), Value::String("world".into())); @@ -469,15 +513,16 @@ #[tokio::test] async fn query_local_script_receives_caller_did() { let state = test_state(); - let lexicon = make_query_lexicon( - "test.whoami", - Some( - r#"function handle() - return { did = caller_did or "anonymous" } - end"#, - ), - ); + let lexicon = make_query_lexicon("test.whoami", None); state.lexicons.upsert(lexicon).await; + seed_script( + &state, + "xrpc.query:test.whoami", + r#"function handle() + return { did = caller_did or "anonymous" } + end"#, + ) + .await; // With caller_did let claims = Claims::internal("did:plc:testuser".into()); @@ -548,11 +593,14 @@ async fn lua_script_calls_xrpc_query() { let state = test_state(); // Register a simple query that the outer script will call - let inner_lexicon = make_query_lexicon( - "test.inner", - Some(r#"function handle() return { value = 42 } end"#), - ); + let inner_lexicon = make_query_lexicon("test.inner", None); state.lexicons.upsert(inner_lexicon).await; + seed_script( + &state, + "xrpc.query:test.inner", + r#"function handle() return { value = 42 } end"#, + ) + .await; let state_arc = Arc::new(state); let lua = sandbox::create_sandbox().unwrap(); diff --git a/src/record_handler.rs b/src/record_handler.rs --- a/src/record_handler.rs +++ b/src/record_handler.rs @@ -56,51 +56,45 @@ None => return, }; let cid = record.cid.as_deref().unwrap_or_default(); - // Run index hook before storing, if configured. The hook's return - // value determines what (if anything) gets written to the DB. - let rec_to_store = - if let Some(script) = state.lexicons.get_index_hook(&record.collection).await { - let hook_result = crate::lua::execute_hook_script(&crate::lua::HookEvent { - state, - lexicon_id: &record.collection, - script: &script, - action: &record.action, - uri: &uri, - did: &record.did, - collection: &record.collection, - rkey: &record.rkey, - record: Some(rec), - }) + // Run record-event script (if any) before storing. The script's + // return value determines what gets written: + // None → skip indexing entirely + // Some(record) → upsert with that record body + // The dispatcher cascades `record.:` → + // `record.index:`; failures are dead-lettered fail-open. + let hook_result = crate::lua::run_record_event_script( + state, + &record.collection, + &record.action, + &uri, + &record.did, + &record.rkey, + Some(rec), + ) + .await; + let rec_to_store = match hook_result { + None => { + log_event( + db, + EventLog { + event_type: "record.skipped".to_string(), + severity: Severity::Info, + actor_did: None, + subject: Some(uri.clone()), + detail: serde_json::json!({ + "collection": record.collection, + "did": record.did, + "rkey": record.rkey, + "reason": "script returned nil", + }), + }, + state.db_backend, + ) .await; - - match hook_result { - None => { - // Hook returned nil — skip indexing this record. - log_event( - db, - EventLog { - event_type: "record.skipped".to_string(), - severity: Severity::Info, - actor_did: None, - subject: Some(uri.clone()), - detail: serde_json::json!({ - "collection": record.collection, - "did": record.did, - "rkey": record.rkey, - "reason": "hook returned nil", - }), - }, - state.db_backend, - ) - .await; - return; - } - Some(v) => v, - } - } else { - // No hook — store the original record as-is. - rec.clone() - }; + return; + } + Some(v) => v, + }; let now = now_rfc3339(); let backend = state.db_backend; @@ -182,42 +176,37 @@ } "delete" => { let backend = state.db_backend; - // Run index hook before deleting, if configured. - if let Some(script) = state.lexicons.get_index_hook(&record.collection).await { - let hook_result = crate::lua::execute_hook_script(&crate::lua::HookEvent { - state, - lexicon_id: &record.collection, - script: &script, - action: "delete", - uri: &uri, - did: &record.did, - collection: &record.collection, - rkey: &record.rkey, - record: None, - }) + // Run record-event script (if any) before deleting. A nil + // return aborts the delete; any other return continues. + let hook_result = crate::lua::run_record_event_script( + state, + &record.collection, + "delete", + &uri, + &record.did, + &record.rkey, + None, + ) + .await; + if hook_result.is_none() { + log_event( + db, + EventLog { + event_type: "record.skipped".to_string(), + severity: Severity::Info, + actor_did: None, + subject: Some(uri.clone()), + detail: serde_json::json!({ + "collection": record.collection, + "did": record.did, + "rkey": record.rkey, + "reason": "script returned nil", + }), + }, + backend, + ) .await; - - if hook_result.is_none() { - // Hook returned nil — skip the delete. - log_event( - db, - EventLog { - event_type: "record.skipped".to_string(), - severity: Severity::Info, - actor_did: None, - subject: Some(uri.clone()), - detail: serde_json::json!({ - "collection": record.collection, - "did": record.did, - "rkey": record.rkey, - "reason": "hook returned nil", - }), - }, - backend, - ) - .await; - return; - } + return; } let delete_sql = adapt_sql("DELETE FROM records WHERE uri = ?", backend); diff --git a/src/xrpc/procedure.rs b/src/xrpc/procedure.rs --- a/src/xrpc/procedure.rs +++ b/src/xrpc/procedure.rs @@ -18,7 +18,15 @@ input: &Value, params: &std::collections::HashMap, lexicon: &crate::lexicon::ParsedLexicon, ) -> Result { - if let Some(ref script) = lexicon.script { + // Trigger-keyed dispatch: a script bound at `xrpc.procedure:` + // overrides the default PDS-write flow. The legacy `lexicon.script` + // column is no longer read. + let trigger = format!("xrpc.procedure:{}", lexicon.id); + if let Some(resolved) = crate::lua::resolve(state, &trigger).await { + // Delegation guard preserved from origin/dev: scripts that run + // under a `delegateDid` must come from a caller who is an + // active write-capable delegate of that account, scoped to the + // calling api_client. let delegate_did = input .get("delegateDid") .and_then(|v| v.as_str()) @@ -71,7 +79,7 @@ claims, &script_input, params, lexicon, - script, + &resolved.body, None, delegate_did.as_deref(), ) diff --git a/src/xrpc/query.rs b/src/xrpc/query.rs --- a/src/xrpc/query.rs +++ b/src/xrpc/query.rs @@ -15,9 +15,19 @@ params: &HashMap, lexicon: &crate::lexicon::ParsedLexicon, claims: Option<&Claims>, ) -> Result { - if let Some(ref script) = lexicon.script { + // Trigger-keyed dispatch: a script bound at `xrpc.query:` + // overrides the default list / get-record flow. The legacy + // `lexicon.script` column is no longer read. + let trigger = format!("xrpc.query:{}", lexicon.id); + if let Some(resolved) = crate::lua::resolve(state, &trigger).await { return crate::lua::execute_query_script( - state, method, params, lexicon, script, claims, None, + state, + method, + params, + lexicon, + &resolved.body, + claims, + None, ) .await; } diff --git a/tests/common/db.rs b/tests/common/db.rs --- a/tests/common/db.rs +++ b/tests/common/db.rs @@ -20,7 +20,7 @@ let backend = test_backend(); match backend { DatabaseBackend::Postgres => { sqlx::query( - "TRUNCATE records, lexicons, backfill_jobs, users, user_permissions, api_keys, event_logs, script_variables, dead_letter_hooks, record_refs, labeler_subscriptions, labels, instance_settings, domains, dpop_sessions, dpop_keys, api_clients, delegated_accounts, account_delegates RESTART IDENTITY CASCADE", + "TRUNCATE records, lexicons, backfill_jobs, users, user_permissions, api_keys, event_logs, script_variables, scripts, dead_letter_scripts, dead_letter_hooks, record_refs, labeler_subscriptions, labels, instance_settings, domains, dpop_sessions, dpop_keys, api_clients, delegated_accounts, account_delegates RESTART IDENTITY CASCADE", ) .execute(pool) .await @@ -41,6 +41,8 @@ "user_permissions", "api_keys", "event_logs", "script_variables", + "scripts", + "dead_letter_scripts", "dead_letter_hooks", "record_refs", "labeler_subscriptions", diff --git a/tests/e2e_scripts.rs b/tests/e2e_scripts.rs new file mode 100644 --- /dev/null +++ b/tests/e2e_scripts.rs @@ -0,0 +1,833 @@ +//! End-to-end tests for the trigger-keyed scripts subsystem. +//! +//! Covers: +//! - Admin CRUD on `/admin/scripts` with trigger-id validation. +//! - Dispatcher cascade for record events +//! (`record.:` → `record.index:`). +//! - Label scripts: URI-routed dispatch + Record local mutation +//! (`Record.delete_local`, `:save_local`). +//! - The no-PDS-auth boundary: a label script that calls `r:save()` +//! gets dead-lettered fail-open with the original record untouched. + +mod common; + +use axum::body::Body; +use axum::http::{Request, StatusCode}; +use happyview::db::{adapt_sql, now_rfc3339}; +use happyview::lua::{LabelAppliedEvent, LabelHookOutcome, run_label_applied_script}; +use happyview::record_handler::{RecordEvent, handle_record_event}; +use http_body_util::BodyExt; +use serde_json::{Value, json}; +use serial_test::serial; +use tower::ServiceExt; + +use common::app::TestApp; +use common::fixtures; + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +async fn json_body(resp: axum::response::Response) -> Value { + let body = resp.into_body().collect().await.unwrap().to_bytes(); + serde_json::from_slice(&body).unwrap() +} + +fn admin_get( + uri: &str, + cookie: (axum::http::HeaderName, axum::http::HeaderValue), +) -> Request { + Request::builder() + .uri(uri) + .header(cookie.0, cookie.1) + .body(Body::empty()) + .unwrap() +} + +fn admin_post( + uri: &str, + cookie: (axum::http::HeaderName, axum::http::HeaderValue), + body: &Value, +) -> Request { + Request::builder() + .method("POST") + .uri(uri) + .header(cookie.0, cookie.1) + .header("content-type", "application/json") + .body(Body::from(serde_json::to_vec(body).unwrap())) + .unwrap() +} + +fn admin_patch( + uri: &str, + cookie: (axum::http::HeaderName, axum::http::HeaderValue), + body: &Value, +) -> Request { + Request::builder() + .method("PATCH") + .uri(uri) + .header(cookie.0, cookie.1) + .header("content-type", "application/json") + .body(Body::from(serde_json::to_vec(body).unwrap())) + .unwrap() +} + +fn admin_delete( + uri: &str, + cookie: (axum::http::HeaderName, axum::http::HeaderValue), +) -> Request { + Request::builder() + .method("DELETE") + .uri(uri) + .header(cookie.0, cookie.1) + .body(Body::empty()) + .unwrap() +} + +/// Seed a record-type lexicon (no scripts bound — scripts live in their +/// own table now, addressed by trigger id). +async fn seed_lexicon(app: &TestApp, lexicon: Value) { + let resp = app + .router + .clone() + .oneshot(admin_post( + "/admin/lexicons", + app.admin_cookie(), + &json!({ "lexicon_json": lexicon }), + )) + .await + .unwrap(); + assert!( + resp.status().is_success(), + "seeding lexicon failed: {:?}", + resp.status() + ); +} + +/// Create a script via the admin API. Returns the created row. +async fn create_script(app: &TestApp, id: &str, body: &str) -> Value { + let resp = app + .router + .clone() + .oneshot(admin_post( + "/admin/scripts", + app.admin_cookie(), + &json!({ "id": id, "body": body }), + )) + .await + .unwrap(); + assert_eq!( + resp.status(), + StatusCode::CREATED, + "create '{id}' failed; body: {:?}", + json_body(resp).await + ); + let resp = app + .router + .clone() + .oneshot(admin_get( + &format!("/admin/scripts/{}", urlencoding::encode(id)), + app.admin_cookie(), + )) + .await + .unwrap(); + json_body(resp).await +} + +async fn seed_record_row( + app: &TestApp, + uri: &str, + did: &str, + collection: &str, + rkey: &str, + body: Value, +) { + let sql = adapt_sql( + "INSERT INTO records (uri, did, collection, rkey, record, cid, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?)", + app.state.db_backend, + ); + sqlx::query(&sql) + .bind(uri) + .bind(did) + .bind(collection) + .bind(rkey) + .bind(serde_json::to_string(&body).unwrap_or_default()) + .bind("bafyseed") + .bind(now_rfc3339()) + .execute(&app.state.db) + .await + .expect("failed to seed records row"); +} + +async fn count_records(app: &TestApp, uri: &str) -> i64 { + let (count,): (i64,) = sqlx::query_as(&adapt_sql( + "SELECT COUNT(*) FROM records WHERE uri = ?", + app.state.db_backend, + )) + .bind(uri) + .fetch_one(&app.state.db) + .await + .unwrap(); + count +} + +async fn fetch_record_body(app: &TestApp, uri: &str) -> Option { + let row: Option<(String,)> = sqlx::query_as(&adapt_sql( + "SELECT record FROM records WHERE uri = ?", + app.state.db_backend, + )) + .bind(uri) + .fetch_optional(&app.state.db) + .await + .unwrap(); + row.map(|(s,)| serde_json::from_str(&s).unwrap()) +} + +// --------------------------------------------------------------------------- +// Admin CRUD +// --------------------------------------------------------------------------- + +#[tokio::test] +#[serial] +#[ignore] +async fn create_then_get_script_round_trips() { + let app = TestApp::new().await; + let id = "record.create:com.example.thing"; + create_script(&app, id, "function handle() return event.record end").await; + + let resp = app + .router + .clone() + .oneshot(admin_get( + &format!("/admin/scripts/{}", urlencoding::encode(id)), + app.admin_cookie(), + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let row = json_body(resp).await; + assert_eq!(row["id"], id); + assert_eq!(row["script_type"], "lua"); +} + +#[tokio::test] +#[serial] +#[ignore] +async fn list_scripts_returns_all_rows() { + let app = TestApp::new().await; + create_script( + &app, + "record.create:com.example.thing", + "function handle() return event.record end", + ) + .await; + create_script( + &app, + "labeler.apply:_actor", + "function handle() return event end", + ) + .await; + + let resp = app + .router + .clone() + .oneshot(admin_get("/admin/scripts", app.admin_cookie())) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let rows = json_body(resp).await; + let arr = rows.as_array().unwrap(); + assert_eq!(arr.len(), 2); +} + +#[tokio::test] +#[serial] +#[ignore] +async fn create_rejects_invalid_trigger_prefix() { + let app = TestApp::new().await; + let resp = app + .router + .clone() + .oneshot(admin_post( + "/admin/scripts", + app.admin_cookie(), + &json!({ + "id": "garbage:com.example.thing", + "body": "function handle() end", + }), + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); + let err = json_body(resp).await; + let msg = err["error"].as_str().unwrap_or(""); + assert!( + msg.contains("unknown trigger prefix"), + "expected validation error, got: {msg}" + ); +} + +#[tokio::test] +#[serial] +#[ignore] +async fn create_rejects_invalid_nsid_suffix() { + let app = TestApp::new().await; + // Single-segment NSID — too few segments. + let resp = app + .router + .clone() + .oneshot(admin_post( + "/admin/scripts", + app.admin_cookie(), + &json!({ + "id": "record.create:foo", + "body": "function handle() end", + }), + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); +} + +#[tokio::test] +#[serial] +#[ignore] +async fn create_allows_labeler_apply_actor_special_case() { + let app = TestApp::new().await; + let resp = app + .router + .clone() + .oneshot(admin_post( + "/admin/scripts", + app.admin_cookie(), + &json!({ + "id": "labeler.apply:_actor", + "body": "function handle() return event end", + }), + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::CREATED); +} + +#[tokio::test] +#[serial] +#[ignore] +async fn create_rejects_invalid_lua_body() { + let app = TestApp::new().await; + let resp = app + .router + .clone() + .oneshot(admin_post( + "/admin/scripts", + app.admin_cookie(), + &json!({ + "id": "record.create:com.example.thing", + "body": "function handle(", // syntax error + }), + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); +} + +#[tokio::test] +#[serial] +#[ignore] +async fn patch_updates_body() { + let app = TestApp::new().await; + let id = "record.create:com.example.thing"; + create_script(&app, id, "function handle() return event.record end").await; + + let resp = app + .router + .clone() + .oneshot(admin_patch( + &format!("/admin/scripts/{}", urlencoding::encode(id)), + app.admin_cookie(), + &json!({ "body": "function handle() return nil end" }), + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let row = json_body(resp).await; + assert!(row["body"].as_str().unwrap().contains("return nil")); +} + +#[tokio::test] +#[serial] +#[ignore] +async fn delete_removes_script() { + let app = TestApp::new().await; + let id = "record.delete:com.example.thing"; + create_script(&app, id, "function handle() return event.record end").await; + + let resp = app + .router + .clone() + .oneshot(admin_delete( + &format!("/admin/scripts/{}", urlencoding::encode(id)), + app.admin_cookie(), + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::NO_CONTENT); + + let resp = app + .router + .clone() + .oneshot(admin_get( + &format!("/admin/scripts/{}", urlencoding::encode(id)), + app.admin_cookie(), + )) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::NOT_FOUND); +} + +// --------------------------------------------------------------------------- +// Cascade resolution: action-specific row wins over wildcard +// --------------------------------------------------------------------------- + +#[tokio::test] +#[serial] +#[ignore] +async fn cascade_wildcard_runs_when_no_action_specific() { + let app = TestApp::new().await; + seed_lexicon(&app, fixtures::game_record_lexicon()).await; + + create_script( + &app, + "record.index:games.gamesgamesgamesgames.game", + // Wildcard — uppercases the title for any action. + "function handle() event.record.title = string.upper(event.record.title); return event.record end", + ) + .await; + + handle_record_event( + &app.state, + &RecordEvent { + did: "did:plc:test".into(), + collection: "games.gamesgamesgamesgames.game".into(), + rkey: "rkey1".into(), + action: "create".into(), + record: Some(json!({"title": "test game"})), + cid: Some("bafy".into()), + }, + ) + .await; + + let body = fetch_record_body( + &app, + "at://did:plc:test/games.gamesgamesgamesgames.game/rkey1", + ) + .await + .unwrap(); + assert_eq!(body["title"], "TEST GAME"); +} + +#[tokio::test] +#[serial] +#[ignore] +async fn cascade_action_specific_wins_over_wildcard() { + let app = TestApp::new().await; + seed_lexicon(&app, fixtures::game_record_lexicon()).await; + + create_script( + &app, + "record.index:games.gamesgamesgamesgames.game", + "function handle() event.record.title = 'WILDCARD'; return event.record end", + ) + .await; + create_script( + &app, + "record.create:games.gamesgamesgamesgames.game", + "function handle() event.record.title = 'CREATE-SPECIFIC'; return event.record end", + ) + .await; + + // Create action — specific should win. + handle_record_event( + &app.state, + &RecordEvent { + did: "did:plc:test".into(), + collection: "games.gamesgamesgamesgames.game".into(), + rkey: "rk-create".into(), + action: "create".into(), + record: Some(json!({"title": "x"})), + cid: Some("bafy".into()), + }, + ) + .await; + let body = fetch_record_body( + &app, + "at://did:plc:test/games.gamesgamesgamesgames.game/rk-create", + ) + .await + .unwrap(); + assert_eq!(body["title"], "CREATE-SPECIFIC"); + + // Update action — no record.update binding → cascades to wildcard. + handle_record_event( + &app.state, + &RecordEvent { + did: "did:plc:test".into(), + collection: "games.gamesgamesgamesgames.game".into(), + rkey: "rk-update".into(), + action: "update".into(), + record: Some(json!({"title": "x"})), + cid: Some("bafy".into()), + }, + ) + .await; + let body = fetch_record_body( + &app, + "at://did:plc:test/games.gamesgamesgamesgames.game/rk-update", + ) + .await + .unwrap(); + assert_eq!(body["title"], "WILDCARD"); +} + +#[tokio::test] +#[serial] +#[ignore] +async fn no_script_passes_record_through_unchanged() { + let app = TestApp::new().await; + seed_lexicon(&app, fixtures::game_record_lexicon()).await; + + handle_record_event( + &app.state, + &RecordEvent { + did: "did:plc:test".into(), + collection: "games.gamesgamesgamesgames.game".into(), + rkey: "rk1".into(), + action: "create".into(), + record: Some(json!({"title": "untouched"})), + cid: Some("bafy".into()), + }, + ) + .await; + let body = fetch_record_body( + &app, + "at://did:plc:test/games.gamesgamesgamesgames.game/rk1", + ) + .await + .unwrap(); + assert_eq!(body["title"], "untouched"); +} + +#[tokio::test] +#[serial] +#[ignore] +async fn record_create_returning_nil_skips_indexing() { + let app = TestApp::new().await; + seed_lexicon(&app, fixtures::game_record_lexicon()).await; + + create_script( + &app, + "record.create:games.gamesgamesgamesgames.game", + "function handle() return nil end", + ) + .await; + + handle_record_event( + &app.state, + &RecordEvent { + did: "did:plc:test".into(), + collection: "games.gamesgamesgamesgames.game".into(), + rkey: "rk1".into(), + action: "create".into(), + record: Some(json!({"title": "doomed"})), + cid: Some("bafy".into()), + }, + ) + .await; + assert_eq!( + count_records( + &app, + "at://did:plc:test/games.gamesgamesgamesgames.game/rk1" + ) + .await, + 0, + "nil return should drop the record" + ); +} + +// --------------------------------------------------------------------------- +// Label scripts via URI routing +// --------------------------------------------------------------------------- + +#[tokio::test] +#[serial] +#[ignore] +async fn label_script_can_drop_record_via_record_delete_local() { + let app = TestApp::new().await; + + let uri = "at://did:plc:victim/app.bsky.feed.post/rkey1"; + seed_record_row( + &app, + uri, + "did:plc:victim", + "app.bsky.feed.post", + "rkey1", + json!({"text": "hello"}), + ) + .await; + + create_script( + &app, + "labeler.apply:app.bsky.feed.post", + "function handle() \ + if event.val == 'spam' then Record.delete_local(event.uri) end \ + return event \ + end", + ) + .await; + + let outcome = run_label_applied_script( + &app.state, + LabelAppliedEvent { + src: "did:plc:labeler".into(), + uri: uri.into(), + val: "spam".into(), + neg: false, + cts: now_rfc3339(), + exp: None, + }, + ) + .await; + assert!(matches!(outcome, LabelHookOutcome::Continue(_))); + assert_eq!(count_records(&app, uri).await, 0); +} + +#[tokio::test] +#[serial] +#[ignore] +async fn label_script_can_redact_record_via_save_local() { + let app = TestApp::new().await; + + let uri = "at://did:plc:author/app.bsky.feed.post/rkey1"; + seed_record_row( + &app, + uri, + "did:plc:author", + "app.bsky.feed.post", + "rkey1", + json!({"text": "original content"}), + ) + .await; + + create_script( + &app, + "labeler.apply:app.bsky.feed.post", + "function handle() \ + if event.val == 'redact' then \ + local r = Record.load(event.uri) \ + if r then r.text = '[redacted by ' .. event.src .. ']'; r:save_local() end \ + end; \ + return event \ + end", + ) + .await; + + let outcome = run_label_applied_script( + &app.state, + LabelAppliedEvent { + src: "did:plc:labeler".into(), + uri: uri.into(), + val: "redact".into(), + neg: false, + cts: now_rfc3339(), + exp: None, + }, + ) + .await; + assert!(matches!(outcome, LabelHookOutcome::Continue(_))); + + let body = fetch_record_body(&app, uri).await.unwrap(); + assert_eq!(body["text"], "[redacted by did:plc:labeler]"); +} + +#[tokio::test] +#[serial] +#[ignore] +async fn label_script_uri_routes_actor_special_case() { + let app = TestApp::new().await; + + create_script( + &app, + "labeler.apply:_actor", + // Sentinel: write a row into records-table-as-flag so we can + // detect that the script ran. + "function handle() \ + db.raw('INSERT INTO records (uri, did, collection, rkey, record, cid, created_at) \ + VALUES (?, ?, ?, ?, ?, ?, ?)', \ + {'at://did:plc:flag/flag.col/k', 'did:plc:flag', 'flag.col', 'k', '{}', 'b', '2026-05-01'}) \ + return event \ + end", + ) + .await; + + // Bare DID URI should route to `labeler.apply:_actor`. + let outcome = run_label_applied_script( + &app.state, + LabelAppliedEvent { + src: "did:plc:labeler".into(), + uri: "did:plc:somebody".into(), + val: "imposter".into(), + neg: false, + cts: now_rfc3339(), + exp: None, + }, + ) + .await; + assert!(matches!(outcome, LabelHookOutcome::Continue(_))); + + // Sentinel row should exist if the script ran. + assert_eq!( + count_records(&app, "at://did:plc:flag/flag.col/k").await, + 1, + "labeler.apply:_actor should have fired for bare-DID label" + ); +} + +#[tokio::test] +#[serial] +#[ignore] +async fn label_script_calling_record_save_dead_letters_with_clear_message() { + let app = TestApp::new().await; + + let uri = "at://did:plc:author/app.bsky.feed.post/rkey1"; + seed_record_row( + &app, + uri, + "did:plc:author", + "app.bsky.feed.post", + "rkey1", + json!({"text": "untouched"}), + ) + .await; + + create_script( + &app, + "labeler.apply:app.bsky.feed.post", + "function handle() \ + local r = Record.load(event.uri) \ + if r then r.text = 'should fail'; r:save() end \ + return event \ + end", + ) + .await; + + let outcome = run_label_applied_script( + &app.state, + LabelAppliedEvent { + src: "did:plc:labeler".into(), + uri: uri.into(), + val: "anything".into(), + neg: false, + cts: now_rfc3339(), + exp: None, + }, + ) + .await; + // Fail-open: the original label still continues even after the script + // fails its retry budget. + assert!(matches!(outcome, LabelHookOutcome::Continue(_))); + + // The original record is unchanged. + let body = fetch_record_body(&app, uri).await.unwrap(); + assert_eq!(body["text"], "untouched"); + + // A dead-letter row exists with the NO_PDS_AUTH message. + let dl: (String,) = sqlx::query_as(&adapt_sql( + "SELECT error FROM dead_letter_scripts WHERE host_kind = 'label' \ + AND host_id = 'did:plc:labeler' ORDER BY id DESC LIMIT 1", + app.state.db_backend, + )) + .fetch_one(&app.state.db) + .await + .expect("expected a dead_letter_scripts row"); + assert!( + dl.0.contains("no PDS auth"), + "expected NO_PDS_AUTH message in dead-letter, got: {}", + dl.0 + ); +} + +#[tokio::test] +#[serial] +#[ignore] +async fn record_event_script_can_call_record_delete_local() { + let app = TestApp::new().await; + seed_lexicon(&app, fixtures::game_record_lexicon()).await; + + let victim_uri = "at://did:plc:test/games.gamesgamesgamesgames.game/old"; + seed_record_row( + &app, + victim_uri, + "did:plc:test", + "games.gamesgamesgamesgames.game", + "old", + json!({"title": "should-be-gone"}), + ) + .await; + + create_script( + &app, + "record.create:games.gamesgamesgamesgames.game", + "function handle() \ + Record.delete_local('at://did:plc:test/games.gamesgamesgamesgames.game/old') \ + return event.record \ + end", + ) + .await; + + handle_record_event( + &app.state, + &RecordEvent { + did: "did:plc:test".into(), + collection: "games.gamesgamesgamesgames.game".into(), + rkey: "new1".into(), + action: "create".into(), + record: Some(json!({"title": "fresh game"})), + cid: Some("bafy".into()), + }, + ) + .await; + + assert_eq!(count_records(&app, victim_uri).await, 0); + assert_eq!( + count_records( + &app, + "at://did:plc:test/games.gamesgamesgamesgames.game/new1" + ) + .await, + 1 + ); +} + +// --------------------------------------------------------------------------- +// Permission gating +// --------------------------------------------------------------------------- + +#[tokio::test] +#[serial] +#[ignore] +async fn no_auth_returns_401() { + let app = TestApp::new().await; + let resp = app + .router + .clone() + .oneshot( + Request::builder() + .uri("/admin/scripts") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); +} diff --git a/tests/lua_record_api.rs b/tests/lua_record_api.rs new file mode 100644 --- /dev/null +++ b/tests/lua_record_api.rs @@ -0,0 +1,474 @@ +//! Integration tests for the local-only Record API surface +//! (`Record.delete_local`, `r:save_local`, `r:delete_local`) and the +//! auth-boundary errors when label / record-event / query scripts reach +//! for PDS-touching methods (`r:save`, `r:delete`). + +mod common; + +use atrium_identity::did::{CommonDidResolver, CommonDidResolverConfig}; +use atrium_identity::handle::{AtprotoHandleResolver, AtprotoHandleResolverConfig}; +use atrium_oauth::{ + AtprotoLocalhostClientMetadata, DefaultHttpClient, KnownScope, OAuthClientConfig, + OAuthResolverConfig, Scope, +}; +use happyview::AppState; +use happyview::config::Config; +use happyview::db::{DatabaseBackend, adapt_sql, now_rfc3339}; +use happyview::lexicon::LexiconRegistry; +use happyview::lua::record::register_record_api_no_auth; +use mlua::Lua; +use serial_test::serial; +use std::sync::Arc; +use tokio::sync::watch; + +use common::db; + +async fn test_state_with_pool(pool: sqlx::AnyPool, backend: DatabaseBackend) -> AppState { + let config = Config { + host: "127.0.0.1".into(), + port: 3000, + database_url: String::new(), + database_backend: backend, + public_url: String::new(), + session_secret: "test-secret".into(), + jetstream_url: String::new(), + relay_url: String::new(), + plc_url: String::new(), + static_dir: String::new(), + event_log_retention_days: 30, + app_name: None, + logo_uri: None, + tos_uri: None, + policy_uri: None, + token_encryption_key: None, + default_rate_limit_capacity: 100, + default_rate_limit_refill_rate: 2.0, + }; + let (tx, _) = watch::channel(vec![]); + let (labeler_tx, _) = watch::channel(()); + let atrium_http = std::sync::Arc::new(DefaultHttpClient::default()); + let did_resolver = CommonDidResolver::new(CommonDidResolverConfig { + plc_directory_url: "https://plc.directory".into(), + http_client: std::sync::Arc::clone(&atrium_http), + }); + let handle_resolver = AtprotoHandleResolver::new(AtprotoHandleResolverConfig { + dns_txt_resolver: happyview::dns::NativeDnsResolver::new(), + http_client: atrium_http, + }); + let oauth_pool = db::test_pool().await; + let oauth = atrium_oauth::OAuthClient::new(OAuthClientConfig { + client_metadata: AtprotoLocalhostClientMetadata { + redirect_uris: Some(vec!["http://127.0.0.1:0/auth/callback".into()]), + scopes: Some(vec![Scope::Known(KnownScope::Atproto)]), + }, + keys: None, + state_store: happyview::auth::oauth_store::DbStateStore::new(oauth_pool.clone(), backend), + session_store: happyview::auth::oauth_store::DbSessionStore::new(oauth_pool, backend), + resolver: OAuthResolverConfig { + did_resolver, + handle_resolver, + authorization_server_metadata: Default::default(), + protected_resource_metadata: Default::default(), + }, + }) + .expect("Failed to create test OAuth client"); + AppState { + config, + http: reqwest::Client::new(), + db: pool.clone(), + db_backend: backend, + lexicons: LexiconRegistry::new(), + collections_tx: tx, + labeler_subscriptions_tx: labeler_tx, + rate_limiter: happyview::rate_limit::RateLimiter::new( + happyview::rate_limit::RateLimitDefaults { + query_cost: 1, + procedure_cost: 1, + proxy_cost: 1, + }, + ), + oauth: std::sync::Arc::new(happyview::auth::OAuthClientRegistry::new( + std::sync::Arc::new(oauth), + )), + oauth_state_store: happyview::auth::oauth_store::DbStateStore::new(pool.clone(), backend), + cookie_key: axum_extra::extract::cookie::Key::derive_from( + b"test-secret-that-is-at-least-32-bytes-long", + ), + plugin_registry: std::sync::Arc::new(happyview::plugin::PluginRegistry::new()), + wasm_runtime: std::sync::Arc::new( + happyview::plugin::WasmRuntime::new().expect("wasm runtime"), + ), + attestation_signer: None, + official_registry: std::sync::Arc::new(tokio::sync::RwLock::new( + happyview::plugin::official_registry::OfficialRegistryState::default(), + )), + official_registry_config: happyview::plugin::official_registry::RegistryConfig::production( + ), + domain_cache: happyview::domain::DomainCache::new(), + proxy_config: std::sync::Arc::new(arc_swap::ArcSwap::new(std::sync::Arc::new( + happyview::proxy_config::ProxyConfig::default(), + ))), + } +} + +async fn seed_record( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + uri: &str, + did: &str, + collection: &str, + rkey: &str, + record: serde_json::Value, +) { + let sql = adapt_sql( + "INSERT INTO records (uri, did, collection, rkey, record, cid, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?)", + backend, + ); + sqlx::query(&sql) + .bind(uri) + .bind(did) + .bind(collection) + .bind(rkey) + .bind(serde_json::to_string(&record).unwrap_or_default()) + .bind("bafyseed") + .bind(now_rfc3339()) + .execute(pool) + .await + .expect("failed to seed record"); +} + +async fn count_records(pool: &sqlx::AnyPool, backend: DatabaseBackend, uri: &str) -> i64 { + let sql = adapt_sql("SELECT COUNT(*) FROM records WHERE uri = ?", backend); + let (count,): (i64,) = sqlx::query_as(&sql) + .bind(uri) + .fetch_one(pool) + .await + .expect("count query"); + count +} + +async fn fetch_record_body( + pool: &sqlx::AnyPool, + backend: DatabaseBackend, + uri: &str, +) -> Option { + let sql = adapt_sql("SELECT record FROM records WHERE uri = ?", backend); + let row: Option<(String,)> = sqlx::query_as(&sql) + .bind(uri) + .fetch_optional(pool) + .await + .expect("fetch record"); + row.map(|(s,)| serde_json::from_str(&s).expect("record json")) +} + +/// Build a sandbox with the Record API registered in **no-auth mode** — +/// the same shape label / record-event / query scripts get. +fn setup_no_auth_lua(state: &AppState) -> Lua { + let lua = Lua::new(); + register_record_api_no_auth(&lua, Arc::new(state.clone())).expect("register record api"); + lua +} + +// --------------------------------------------------------------------------- +// Record.delete_local(uri) static +// --------------------------------------------------------------------------- + +#[tokio::test] +#[serial] +#[ignore] +async fn record_static_delete_local_returns_true_when_row_existed() { + let pool = db::test_pool().await; + let backend = db::test_backend(); + db::truncate_all(&pool).await; + + let uri = "at://did:plc:test/test.collection/rkey1"; + seed_record( + &pool, + backend, + uri, + "did:plc:test", + "test.collection", + "rkey1", + serde_json::json!({"name": "kept"}), + ) + .await; + + let state = test_state_with_pool(pool.clone(), backend).await; + let lua = setup_no_auth_lua(&state); + + let deleted: bool = lua + .load(format!(r#"return Record.delete_local("{uri}")"#)) + .eval_async() + .await + .expect("delete_local call"); + assert!(deleted, "expected true (row existed before)"); + + let after = count_records(&pool, backend, uri).await; + assert_eq!(after, 0, "row should be gone"); +} + +#[tokio::test] +#[serial] +#[ignore] +async fn record_static_delete_local_returns_false_when_row_absent() { + let pool = db::test_pool().await; + let backend = db::test_backend(); + db::truncate_all(&pool).await; + let state = test_state_with_pool(pool, backend).await; + let lua = setup_no_auth_lua(&state); + + let deleted: bool = lua + .load(r#"return Record.delete_local("at://did:plc:nope/test.collection/none")"#) + .eval_async() + .await + .expect("delete_local call"); + assert!(!deleted, "no row → false (idempotent)"); +} + +// --------------------------------------------------------------------------- +// r:delete_local() instance method +// --------------------------------------------------------------------------- + +#[tokio::test] +#[serial] +#[ignore] +async fn record_instance_delete_local_removes_row_and_clears_uri() { + let pool = db::test_pool().await; + let backend = db::test_backend(); + db::truncate_all(&pool).await; + + let uri = "at://did:plc:test/test.collection/rkey1"; + seed_record( + &pool, + backend, + uri, + "did:plc:test", + "test.collection", + "rkey1", + serde_json::json!({"name": "doomed"}), + ) + .await; + + let state = test_state_with_pool(pool.clone(), backend).await; + let lua = setup_no_auth_lua(&state); + + // Load via Record.load(), then call :delete_local() — this is the + // primary shape we expect from a label-script reaction. + let uri_after: mlua::Value = lua + .load(format!( + r#" + local r = Record.load("{uri}") + assert(r ~= nil, "record not loaded") + r:delete_local() + return r._uri + "# + )) + .eval_async() + .await + .expect("delete_local instance call"); + + assert!(matches!(uri_after, mlua::Value::Nil), "_uri should be nil"); + assert_eq!(count_records(&pool, backend, uri).await, 0); +} + +// --------------------------------------------------------------------------- +// r:save_local() instance method +// --------------------------------------------------------------------------- + +#[tokio::test] +#[serial] +#[ignore] +async fn record_instance_save_local_updates_existing_row() { + let pool = db::test_pool().await; + let backend = db::test_backend(); + db::truncate_all(&pool).await; + + let uri = "at://did:plc:test/test.collection/rkey1"; + seed_record( + &pool, + backend, + uri, + "did:plc:test", + "test.collection", + "rkey1", + serde_json::json!({"text": "original"}), + ) + .await; + + let state = test_state_with_pool(pool.clone(), backend).await; + let lua = setup_no_auth_lua(&state); + + // Redact-style flow: load, mutate, save_local. + lua.load(format!( + r#" + local r = Record.load("{uri}") + assert(r ~= nil) + r.text = "[redacted]" + r:save_local() + "# + )) + .exec_async() + .await + .expect("save_local instance call"); + + let body = fetch_record_body(&pool, backend, uri).await.unwrap(); + assert_eq!(body["text"], "[redacted]"); + // $type is injected automatically by the serializer. + assert_eq!(body["$type"], "test.collection"); + // Row count unchanged (upsert). + assert_eq!(count_records(&pool, backend, uri).await, 1); +} + +#[tokio::test] +#[serial] +#[ignore] +async fn record_save_local_creates_new_row_when_repo_set() { + let pool = db::test_pool().await; + let backend = db::test_backend(); + db::truncate_all(&pool).await; + let state = test_state_with_pool(pool.clone(), backend).await; + let lua = setup_no_auth_lua(&state); + + // Build a fresh record. `:set_repo` provides the DID for the URI; + // without auth there's no fallback. We also manually `:set_rkey` + // since there's no key_type from a (missing) lexicon. + let uri: String = lua + .load( + r#" + local r = Record.new("test.collection", { value = 42 }) + r:set_repo("did:plc:newowner") + r:set_rkey("brandnew") + r:save_local() + return r._uri + "#, + ) + .eval_async() + .await + .expect("save_local creating call"); + + assert_eq!(uri, "at://did:plc:newowner/test.collection/brandnew"); + let body = fetch_record_body(&pool, backend, &uri).await.unwrap(); + assert_eq!(body["value"], 42); + assert_eq!(body["$type"], "test.collection"); +} + +#[tokio::test] +#[serial] +#[ignore] +async fn record_save_local_errors_without_did_when_no_uri() { + let pool = db::test_pool().await; + let backend = db::test_backend(); + db::truncate_all(&pool).await; + let state = test_state_with_pool(pool, backend).await; + let lua = setup_no_auth_lua(&state); + + // No `:set_repo` and no claims → :save_local() must error. + let err = lua + .load( + r#" + local r = Record.new("test.collection", { value = 1 }) + r:save_local() + "#, + ) + .exec_async() + .await + .expect_err("expected error: no DID resolvable"); + let msg = err.to_string(); + assert!( + msg.contains("save_local() needs a DID"), + "expected DID-required message, got: {msg}" + ); +} + +// --------------------------------------------------------------------------- +// PDS-touching methods error cleanly when no auth present +// --------------------------------------------------------------------------- + +#[tokio::test] +#[serial] +#[ignore] +async fn record_save_errors_without_pds_auth() { + let pool = db::test_pool().await; + let backend = db::test_backend(); + db::truncate_all(&pool).await; + + let uri = "at://did:plc:test/test.collection/rkey1"; + seed_record( + &pool, + backend, + uri, + "did:plc:test", + "test.collection", + "rkey1", + serde_json::json!({"value": 1}), + ) + .await; + + let state = test_state_with_pool(pool.clone(), backend).await; + let lua = setup_no_auth_lua(&state); + + let err = lua + .load(format!( + r#" + local r = Record.load("{uri}") + r:save() + "# + )) + .exec_async() + .await + .expect_err("expected NO_PDS_AUTH error"); + let msg = err.to_string(); + assert!( + msg.contains("no PDS auth"), + "expected NO_PDS_AUTH message, got: {msg}" + ); + + // The original row should be untouched. + let body = fetch_record_body(&pool, backend, uri).await.unwrap(); + assert_eq!(body["value"], 1); +} + +#[tokio::test] +#[serial] +#[ignore] +async fn record_delete_errors_without_pds_auth() { + let pool = db::test_pool().await; + let backend = db::test_backend(); + db::truncate_all(&pool).await; + + let uri = "at://did:plc:test/test.collection/rkey1"; + seed_record( + &pool, + backend, + uri, + "did:plc:test", + "test.collection", + "rkey1", + serde_json::json!({"value": 1}), + ) + .await; + + let state = test_state_with_pool(pool.clone(), backend).await; + let lua = setup_no_auth_lua(&state); + + let err = lua + .load(format!( + r#" + local r = Record.load("{uri}") + r:delete() + "# + )) + .exec_async() + .await + .expect_err("expected NO_PDS_AUTH error"); + let msg = err.to_string(); + assert!( + msg.contains("no PDS auth"), + "expected NO_PDS_AUTH message, got: {msg}" + ); + + // No-auth :delete() must NOT touch the local DB either — + // the row should still be there. + assert_eq!(count_records(&pool, backend, uri).await, 1); +} -- tangled.sh