Something went wrong. Try again.
A lexicon-driven AppView for ATProto. happyview.dev
backfill firehose jetstream atproto appview oauth lexicon
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387#[cfg(feature = "lua-reference")]use mlua::{Lua, LuaSerdeExt, Result as LuaResult};#[cfg(feature = "lua-reference")]use serde_json::Value;#[cfg(feature = "lua-reference")]use std::collections::HashMap;
/// Optional space context passed to Lua scripts when the request is space-scoped.#[derive(Debug, Clone)]pub struct SpaceContext { pub space: String, pub space_id: String, pub did: String, pub authority_did: String, pub type_nsid: String, pub skey: String,}
impl From<&SpaceContext> for crate::plugin::ScriptSpace { fn from(space: &SpaceContext) -> Self { Self { uri: space.space.clone(), id: space.space_id.clone(), did: space.did.clone(), authority_did: space.authority_did.clone(), type_nsid: space.type_nsid.clone(), skey: space.skey.clone(), } }}
/// Everything the runtime knows about one script invocation, built into the/// `ctx` argument of `handle(input, ctx)`. Keys are snake_case, the same/// convention as everything else a Lua script sees.#[cfg(feature = "lua-reference")]pub struct Invocation<'a> { pub trigger_id: &'a str, pub caller_did: Option<&'a str>, pub has_pds_auth: bool, pub env: &'a HashMap<String, String>, pub method: Option<&'a str>, pub collection: Option<&'a str>, pub params: Option<&'a HashMap<String, Value>>, pub delegate_did: Option<&'a str>, pub space: Option<&'a SpaceContext>, pub job: Option<mlua::Table>,}
/// Build the `ctx` table passed as the second argument to `handle`.#[cfg(feature = "lua-reference")]pub fn build_ctx(lua: &Lua, inv: &Invocation<'_>) -> LuaResult<mlua::Table> { let ctx = lua.create_table()?; ctx.set("trigger", inv.trigger_id)?; ctx.set("caller_did", inv.caller_did.map(str::to_string))?; ctx.set("has_pds_auth", inv.has_pds_auth)?; ctx.set("env", lua.to_value(inv.env)?)?; ctx.set("method", inv.method.map(str::to_string))?; ctx.set("collection", inv.collection.map(str::to_string))?; ctx.set( "params", match inv.params { Some(p) => lua.to_value(p)?, None => mlua::Value::Nil, }, )?; ctx.set("delegate_did", inv.delegate_did.map(str::to_string))?; if let Some(s) = inv.space { let t = lua.create_table()?; t.set("uri", s.space.as_str())?; t.set("id", s.space_id.as_str())?; t.set("did", s.did.as_str())?; t.set("authority_did", s.authority_did.as_str())?; t.set("spaceType", s.type_nsid.as_str())?; t.set("skey", s.skey.as_str())?; ctx.set("space", t)?; } if let Some(job) = &inv.job { let t = lua.create_table()?; for canonical in ["id", "progress", "should_stop", "wait"] { t.set(canonical, job.get::<mlua::Value>(canonical)?)?; } ctx.set("job", t)?; } Ok(ctx)}
/// The `ctx.job` table a job script receives from a VM in this process: the/// job's id, progress reporting, cooperative cancellation and sleep.////// An interpreter reaches the same three controls through its `script:host`/// imports, which act on the run the host is holding rather than on a job/// named here, so this builds the surface only where there is no such run.#[cfg(feature = "lua-reference")]pub fn job_ctx( lua: &Lua, state: std::sync::Arc<crate::AppState>, job_id: String,) -> LuaResult<mlua::Table> { let job_table = lua.create_table()?; job_table.set("id", job_id.clone())?;
{ let state = state.clone(); let job_id = job_id.clone(); let progress_fn = lua.create_async_function(move |lua, data: mlua::Value| { let state = state.clone(); let job_id = job_id.clone(); let json_data: serde_json::Value = lua.from_value(data).unwrap_or(serde_json::json!({})); async move { crate::jobs::db::update_progress(&state, &job_id, &json_data) .await .map_err(|e| mlua::Error::runtime(format!("job.progress failed: {e}")))?; Ok(()) } })?; job_table.set("progress", progress_fn)?; }
{ let state = state.clone(); let job_id = job_id.clone(); let should_stop_fn = lua.create_async_function(move |_lua, ()| { let state = state.clone(); let job_id = job_id.clone(); async move { let result = crate::jobs::db::should_stop(&state, &job_id).await; Ok(result.is_some()) } })?; job_table.set("should_stop", should_stop_fn)?; }
{ let wait_fn = lua.create_async_function(move |_lua, seconds: f64| { let state = state.clone(); async move { let duration = std::time::Duration::from_secs_f64(seconds.clamp(0.0, 3600.0)); tokio::time::sleep(duration).await; let elapsed_ms = u64::try_from(duration.as_millis()).unwrap_or(u64::MAX); crate::telemetry::counters::add_saturating( &state.telemetry_counters.job_wait_ms, elapsed_ms, ); Ok(()) } })?; job_table.set("wait", wait_fn)?; }
Ok(job_table)}
#[cfg(test)]mod tests { use super::*;
/// Six same-typed strings, where `space`/`space_id` and `uri`/`id` are the /// pair that do not share a name: a swap would compile and surface only as /// a space a script cannot find. #[test] fn a_space_context_maps_field_for_field_onto_the_wire_shape() { let sent = crate::plugin::ScriptSpace::from(&SpaceContext { space: "at://did:plc:a/space/com.example.type/sk".into(), space_id: "sid".into(), did: "did:plc:a".into(), authority_did: "did:plc:auth".into(), type_nsid: "com.example.type".into(), skey: "sk".into(), }); assert_eq!( sent, crate::plugin::ScriptSpace { uri: "at://did:plc:a/space/com.example.type/sk".into(), id: "sid".into(), did: "did:plc:a".into(), authority_did: "did:plc:auth".into(), type_nsid: "com.example.type".into(), skey: "sk".into(), } ); }}
#[cfg(all(test, feature = "lua-reference"))]mod reference_tests { use super::*; use crate::lua::sandbox::create_sandbox; use serde_json::json;
#[tokio::test] async fn ctx_carries_snake_case_keys_and_nils() { let lua = create_sandbox().unwrap(); let env = HashMap::from([("API".to_string(), "k".to_string())]); let params = HashMap::from([("q".to_string(), json!("x"))]); let space = SpaceContext { space: "at://s".into(), space_id: "sid".into(), did: "did:plc:a".into(), authority_did: "did:plc:auth".into(), type_nsid: "t".into(), skey: "k".into(), }; let ctx = build_ctx( &lua, &Invocation { trigger_id: "xrpc.procedure:app.test.p", caller_did: Some("did:plc:me"), has_pds_auth: true, env: &env, method: Some("app.test.p"), collection: Some("app.test.rec"), params: Some(¶ms), delegate_did: None, space: Some(&space), job: None, }, ) .unwrap(); lua.globals().set("ctx", ctx).unwrap(); let s: String = lua .load( r#"return ctx.caller_did .. "|" .. tostring(ctx.has_pds_auth) .. "|" .. ctx.env.API .. "|" .. ctx.trigger .. "|" .. ctx.method .. "|" .. ctx.collection .. "|" .. ctx.params.q .. "|" .. tostring(ctx.delegate_did) .. "|" .. ctx.space.authority_did .. "|" .. tostring(ctx.job)"#, ) .eval() .unwrap(); assert_eq!( s, "did:plc:me|true|k|xrpc.procedure:app.test.p|app.test.p|app.test.rec|x|nil|did:plc:auth|nil" ); }
#[tokio::test] async fn ctx_for_anonymous_query_has_nil_caller() { let lua = create_sandbox().unwrap(); let env = HashMap::new(); let ctx = build_ctx( &lua, &Invocation { trigger_id: "xrpc.query:app.test.q", caller_did: None, has_pds_auth: false, env: &env, method: Some("app.test.q"), collection: None, params: None, delegate_did: None, space: None, job: None, }, ) .unwrap(); lua.globals().set("ctx", ctx).unwrap(); let s: String = lua .load( r#"return tostring(ctx.caller_did) .. "|" .. tostring(ctx.space) .. "|" .. tostring(ctx.collection)"#, ) .eval() .unwrap(); assert_eq!(s, "nil|nil|nil"); }
#[tokio::test] async fn ctx_carries_a_delegate_and_the_space_fields() { let lua = create_sandbox().unwrap(); let env = HashMap::new(); let space = SpaceContext { space: "at://did:plc:owner/space/com.example.forum/main".into(), space_id: "space-123".into(), did: "did:plc:owner".into(), authority_did: "did:plc:owner".into(), type_nsid: "com.example.forum".into(), skey: "main".into(), }; let ctx = build_ctx( &lua, &Invocation { trigger_id: "xrpc.procedure:com.example.post", caller_did: Some("did:plc:caller"), has_pds_auth: true, env: &env, method: Some("com.example.post"), collection: Some("com.example.forum.post"), params: None, delegate_did: Some("did:plc:delegate"), space: Some(&space), job: None, }, ) .unwrap(); lua.globals().set("ctx", ctx).unwrap(); let s: String = lua .load( r#"return ctx.delegate_did .. "|" .. ctx.space.uri .. "|" .. ctx.space.id .. "|" .. ctx.space.did .. "|" .. ctx.space.spaceType .. "|" .. ctx.space.skey"#, ) .eval() .unwrap(); assert_eq!( s, "did:plc:delegate|at://did:plc:owner/space/com.example.forum/main|space-123|did:plc:owner|com.example.forum|main" ); }
#[tokio::test] async fn job_ctx_carries_the_id_and_the_three_functions() { let lua = create_sandbox().unwrap(); let state = crate::test_support::test_state_with_pool(crate::test_support::memory_pool().await); let job = job_ctx(&lua, std::sync::Arc::new(state), "test-id".into()).unwrap(); lua.globals().set("job", job).unwrap();
let ok: bool = lua .load( r#" return job.id == 'test-id' and type(job.progress) == 'function' and type(job.should_stop) == 'function' and type(job.wait) == 'function' "#, ) .eval_async() .await .unwrap(); assert!(ok); }
#[tokio::test] async fn job_wait_clamps_to_the_allowed_range_and_counts_the_time() { let lua = create_sandbox().unwrap(); let state = crate::test_support::test_state_with_pool(crate::test_support::memory_pool().await); let counters = state.telemetry_counters.clone(); let job = job_ctx(&lua, std::sync::Arc::new(state), "test-id".into()).unwrap(); lua.globals().set("job", job).unwrap();
lua.load("job.wait(-5)").exec_async().await.unwrap(); assert_eq!( counters .job_wait_ms .load(std::sync::atomic::Ordering::Relaxed), 0 ); }
#[tokio::test] async fn ctx_job_carries_only_the_canonical_keys() { let lua = create_sandbox().unwrap(); let env = HashMap::new(); let job = lua.create_table().unwrap(); job.set("id", "job-1").unwrap(); job.set("progress", lua.create_function(|_, ()| Ok(())).unwrap()) .unwrap(); job.set( "should_stop", lua.create_function(|_, ()| Ok(false)).unwrap(), ) .unwrap(); job.set("wait", lua.create_function(|_, _s: f64| Ok(())).unwrap()) .unwrap(); job.set("extra", "dropped").unwrap(); let ctx = build_ctx( &lua, &Invocation { trigger_id: "job.run:test.export", caller_did: Some("did:plc:me"), has_pds_auth: false, env: &env, method: None, collection: None, params: None, delegate_did: None, space: None, job: Some(job), }, ) .unwrap(); lua.globals().set("ctx", ctx).unwrap(); let s: String = lua .load( r#"return ctx.job.id .. "|" .. type(ctx.job.progress) .. "|" .. tostring(ctx.job.should_stop()) .. "|" .. type(ctx.job.wait) .. "|" .. tostring(ctx.job.extra)"#, ) .eval() .unwrap(); assert_eq!(s, "job-1|function|false|function|nil"); }}