Something went wrong. Try again.
A lexicon-driven AppView for ATProto.
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340use base64::{Engine, engine::general_purpose::STANDARD as BASE64};use mlua::{Lua, LuaSerdeExt, Result as LuaResult};use serde_json::{Value, json};use sqlx::{Column, Row};use std::sync::Arc;
use crate::AppState;use crate::db::{DatabaseBackend, adapt_sql, decode_cursor, encode_cursor};
const MAX_FILTER_DEPTH: u8 = 5;const ALLOWED_OPS: &[&str] = &["=", "!=", "<", ">", "<=", ">=", "LIKE", "NOT LIKE"];
/// Table-name prefix reserved for HappyView's own internal tables. `db.raw`/// blocks these **by default** — so a table added in a future migration is/// protected until it is deliberately allowed — except for the data tables in/// [`ALLOWED_INTERNAL_TABLES`].const PROTECTED_TABLE_PREFIX: &str = "happyview_";
/// Internal tables that don't carry the `happyview_` prefix but are still/// off-limits (SQLx's migration bookkeeping).const PROTECTED_EXACT_TABLES: &[&str] = &["_sqlx_migrations"];
/// Internal tables `db.raw` is allowed to read and write despite the reserved/// prefix: public AppView data and space *data*. Everything else `happyview_*`/// stays blocked — secrets and tokens (`happyview_dpop_keys`/`_sessions`,/// `happyview_api_keys`, `happyview_oauth_*`, `happyview_script_variables`),/// auth/privilege state (`happyview_users`/`_user_permissions`, the delegation/// tables), trust config (`happyview_domains`, `happyview_instance_settings`),/// and cryptographic material (`happyview_space_credentials`, and/// `happyview_space_repo_state` which holds commit-signature key material).////// Space membership/records are exposed because a space defines *access*, not/// confidentiality — whether to expose otherwise-private space data through the/// AppView is left to the admin.const ALLOWED_INTERNAL_TABLES: &[&str] = &[ // Public AppView data. "happyview_records", "happyview_record_refs", "happyview_labels", "happyview_lexicons", // Background jobs. "happyview_jobs", // Space data (not the credential/key-material tables). "happyview_spaces", "happyview_space_members", "happyview_space_records", "happyview_space_record_oplog", "happyview_space_notify_registrations", "happyview_space_dids",];
/// Reject a `db.raw` SQL string that references a protected internal table.////// Tokenizes the SQL (so string literals and comments containing the prefix are/// ignored, and quoted / schema-qualified identifiers are still caught) and/// blocks any `happyview_*` (or `_sqlx_migrations`) identifier that is not in/// [`ALLOWED_INTERNAL_TABLES`]. Unicode-escaped identifiers (`U&"…"`) are refused/// outright as an evasion vector, and SQL that cannot be tokenized fails closed.fn check_raw_sql_tables(sql: &str) -> Result<(), String> { use sqlparser::dialect::GenericDialect; use sqlparser::tokenizer::{Token, Tokenizer};
// `U&'…'` / `U&"…"` unicode-escaped literals could smuggle a protected // identifier past tokenization (the escapes decode to letters); there is no // legitimate need for them in `db.raw`, so refuse them outright. let lowered = sql.to_ascii_lowercase(); if lowered.contains("u&\"") || lowered.contains("u&'") { return Err("db.raw does not allow unicode-escaped identifiers".into()); }
// Tokenizing (rather than substring matching) means the prefix inside string // literals or comments is ignored, while quoted and schema-qualified // identifiers are still seen. SQL we cannot tokenize fails closed. let tokens = Tokenizer::new(&GenericDialect {}, sql) .tokenize() .map_err(|e| format!("db.raw could not parse SQL: {e}"))?;
for token in tokens { if let Token::Word(word) = token { let name = word.value.to_ascii_lowercase(); let is_internal = name.starts_with(PROTECTED_TABLE_PREFIX) || PROTECTED_EXACT_TABLES.contains(&name.as_str()); if is_internal && !ALLOWED_INTERNAL_TABLES.contains(&name.as_str()) { return Err(format!( "db.raw cannot reference the protected internal HappyView table '{}'", word.value )); } } }
Ok(())}
fn is_valid_json_field_path(path: &str) -> bool { if path.is_empty() { return false; } for segment in path.split('.') { if segment.is_empty() { return false; } let bracket_start = segment.find('[').unwrap_or(segment.len()); let ident = &segment[..bracket_start]; if ident.is_empty() || !ident.chars().all(|c| c.is_ascii_alphanumeric() || c == '_') { return false; } let mut rest = &segment[bracket_start..]; while !rest.is_empty() { if !rest.starts_with('[') { return false; } let close = match rest.find(']') { Some(i) => i, None => return false, }; let idx = &rest[1..close]; if idx.is_empty() || !idx.chars().all(|c| c.is_ascii_digit()) { return false; } rest = &rest[close + 1..]; } } true}
#[derive(Debug)]enum FilterNode { Condition { field: String, op: String, value: String, }, Group { combine: String, children: Vec<FilterNode>, },}
fn parse_filter_node(table: &mlua::Table, depth: u8) -> LuaResult<FilterNode> { if depth >= MAX_FILTER_DEPTH { return Err(mlua::Error::runtime(format!( "filter nesting too deep (max {MAX_FILTER_DEPTH} levels)", ))); }
if let Ok(field) = table.get::<String>("field") { if !is_valid_json_field_path(&field) { return Err(mlua::Error::runtime(format!( "invalid filter field '{field}': use alphanumeric names with optional dot notation and array indices (e.g. 'name', 'author.handle', 'tags[0]')", ))); }
let op: String = table .get::<String>("op") .unwrap_or_else(|_| "=".to_string()); let op_upper = op.to_uppercase(); if !ALLOWED_OPS.contains(&op_upper.as_str()) { return Err(mlua::Error::runtime(format!( "invalid filter op '{op}': must be one of {ALLOWED_OPS:?}", ))); }
let val: mlua::Value = table.get("value")?; let value = match val { mlua::Value::String(s) => s.to_str()?.to_string(), mlua::Value::Integer(n) => n.to_string(), mlua::Value::Number(n) => n.to_string(), mlua::Value::Boolean(b) => (if b { "true" } else { "false" }).to_string(), other => { return Err(mlua::Error::runtime(format!( "unsupported filter value type for '{field}': {}", other.type_name() ))); } };
return Ok(FilterNode::Condition { field, op: op_upper, value, }); }
let combine: String = table .get::<String>("combine") .unwrap_or_else(|_| "AND".to_string()) .to_uppercase(); if combine != "AND" && combine != "OR" { return Err(mlua::Error::runtime(format!( "invalid filter combine '{combine}': must be 'AND' or 'OR'", ))); }
let mut children = Vec::new(); for child in table.sequence_values::<mlua::Table>() { children.push(parse_filter_node(&child?, depth + 1)?); }
if children.is_empty() { return Err(mlua::Error::runtime("filter group has no conditions")); }
Ok(FilterNode::Group { combine, children })}
fn build_filter_sql(node: &FilterNode, binds: &mut Vec<String>) -> String { match node { FilterNode::Condition { field, op, value } => { binds.push(value.clone()); format!("json_extract(record, '$.{field}') {op} ?") } FilterNode::Group { combine, children } => { let parts: Vec<String> = children .iter() .map(|c| build_filter_sql(c, binds)) .collect(); if parts.len() == 1 { parts.into_iter().next().unwrap() } else { format!("({})", parts.join(&format!(" {combine} "))) } } }}
/// Register the `db` table with database query functions.pub fn register_db_api(lua: &Lua, state: Arc<AppState>) -> LuaResult<()> { let db_table = lua.create_table()?;
// db.query({ collection, did?, limit?, offset?, cursor?, sort?, sortDirection?, filter? }) -> { records, cursor? } let state_query = state.clone(); let query_fn = lua.create_async_function(move |lua, opts: mlua::Table| { let state = state_query.clone(); async move { let backend = state.db_backend; let collection: String = opts.get("collection")?; let did: Option<String> = opts.get("did").ok(); let limit: i64 = opts.get::<i64>("limit").unwrap_or(20).min(100); let sort: Option<String> = opts.get("sort").ok(); let sort_direction: Option<String> = opts.get("sortDirection").ok(); let cursor_str: Option<String> = opts.get("cursor").ok();
if let Some(ref field) = sort && !is_valid_json_field_path(field) { return Err(mlua::Error::runtime( "invalid sort field: use alphanumeric names with optional dot notation and array indices (e.g. 'name', 'author.handle', 'tags[0]')", )); }
let direction = match sort_direction.as_deref() { Some("asc") => "ASC", Some("desc") => "DESC", None => "DESC", Some(other) => { return Err(mlua::Error::runtime(format!( "invalid sortDirection '{other}': must be 'asc' or 'desc'" ))); } };
let filter_table: Option<mlua::Table> = opts.get("filter").ok(); let mut filter_binds: Vec<String> = Vec::new(); let filter_clause = if let Some(ref tbl) = filter_table { let node = parse_filter_node(tbl, 0)?; let sql = build_filter_sql(&node, &mut filter_binds); format!(" AND {sql}") } else { String::new() };
let result_table = lua.create_table()?;
if let Some(ref sort_field) = sort { // Custom sort: use OFFSET/LIMIT with base64-encoded offset cursor let offset: i64 = if let Some(ref cursor) = cursor_str { BASE64.decode(cursor).ok() .and_then(|b| String::from_utf8(b).ok()) .and_then(|s| s.parse::<i64>().ok()) .unwrap_or(0) } else { opts.get::<i64>("offset").unwrap_or(0) };
let top_level_columns = ["indexed_at", "did", "uri"]; let order_expr = if top_level_columns.contains(&sort_field.as_str()) { format!("{sort_field} {direction}") } else { format!("json_extract(record, '$.{sort_field}') {direction}") };
let did_clause = if did.is_some() { " AND did = ?" } else { "" }; let sql = adapt_sql( &format!("SELECT uri, did, record FROM happyview_records WHERE collection = ?{did_clause}{filter_clause} ORDER BY {order_expr} LIMIT ? OFFSET ?"), backend, ); let mut q = sqlx::query_as(&sql).bind(&collection); if let Some(ref did) = did { q = q.bind(did); } for val in &filter_binds { q = q.bind(val); } let rows: Vec<(String, String, String)> = q .bind(limit) .bind(offset) .fetch_all(&state.db) .await .map_err(|e| mlua::Error::runtime(format!("DB query failed: {e}")))?;
let has_next = rows.len() as i64 == limit;
if has_next { let next_offset = offset + limit; result_table.set("cursor", BASE64.encode(next_offset.to_string()))?; }
let records: Vec<Value> = rows .into_iter() .map(|(uri, _did, record_str)| { let mut record: Value = serde_json::from_str(&record_str).unwrap_or(json!({})); if let Some(obj) = record.as_object_mut() { obj.insert("uri".to_string(), json!(uri)); } record }) .collect();
let record_values: Vec<mlua::Value> = records .iter() .map(|r| lua.to_value(r)) .collect::<LuaResult<_>>()?; let records_table = lua.create_sequence_from(record_values)?; records_table.set_metatable(Some(lua.array_metatable()))?; result_table.set("records", records_table)?; } else { // Cursor-based pagination on (created_at, uri) let cursor_parts = cursor_str.as_ref().and_then(|c| decode_cursor(c));
type RowType = (String, String, String, String);
let did_clause = if did.is_some() { " AND did = ?" } else { "" }; let cursor_clause = if cursor_parts.is_some() { " AND (created_at < ? OR (created_at = ? AND uri < ?))" } else { "" }; let sql = adapt_sql( &format!( "SELECT uri, did, record, created_at FROM happyview_records \ WHERE collection = ?{did_clause}{cursor_clause}{filter_clause} \ ORDER BY created_at DESC, uri DESC \ LIMIT ?" ), backend, ); let mut q = sqlx::query_as::<_, RowType>(&sql).bind(&collection); if let Some(ref did) = did { q = q.bind(did); } if let Some((cursor_ts, cursor_uri)) = &cursor_parts { q = q.bind(cursor_ts).bind(cursor_ts).bind(cursor_uri); } for val in &filter_binds { q = q.bind(val); } let rows_raw: Vec<RowType> = q .bind(limit) .fetch_all(&state.db) .await .map_err(|e| mlua::Error::runtime(format!("DB query failed: {e}")))?;
let has_next = rows_raw.len() as i64 == limit;
if has_next && let Some((last_uri, _, _, last_created_at)) = rows_raw.last() { let cursor = encode_cursor(last_created_at, last_uri); result_table.set("cursor", cursor)?; }
let records: Vec<Value> = rows_raw .into_iter() .map(|(uri, _did, record_str, _created_at)| { let mut record: Value = serde_json::from_str(&record_str).unwrap_or(json!({})); if let Some(obj) = record.as_object_mut() { obj.insert("uri".to_string(), json!(uri)); } record }) .collect();
let record_values: Vec<mlua::Value> = records .iter() .map(|r| lua.to_value(r)) .collect::<LuaResult<_>>()?; let records_table = lua.create_sequence_from(record_values)?; records_table.set_metatable(Some(lua.array_metatable()))?; result_table.set("records", records_table)?; }
Ok(mlua::Value::Table(result_table)) } })?; db_table.set("query", query_fn)?;
// db.get(uri) -> record table or nil let state_get = state.clone(); let get_fn = lua.create_async_function(move |lua, uri: String| { let state = state_get.clone(); async move { let backend = state.db_backend; let sql = adapt_sql( "SELECT record FROM happyview_records WHERE uri = ?", backend, ); let row: Option<(String,)> = sqlx::query_as(&sql) .bind(&uri) .fetch_optional(&state.db) .await .map_err(|e| mlua::Error::runtime(format!("DB query failed: {e}")))?;
match row { Some((record_str,)) => { let mut record: Value = serde_json::from_str(&record_str).unwrap_or(json!({})); if let Some(obj) = record.as_object_mut() { obj.insert("uri".to_string(), json!(uri)); } lua.to_value(&record) } None => Ok(mlua::Value::Nil), } } })?; db_table.set("get", get_fn)?;
// db.search({ collection, field, query, limit? }) -> { records } let state_search = state.clone(); let search_fn = lua.create_async_function(move |lua, opts: mlua::Table| { let state = state_search.clone(); async move { let backend = state.db_backend; let collection: String = opts.get("collection")?; let field: String = opts.get("field")?; let query: String = opts.get("query")?; let limit: i64 = opts.get::<i64>("limit").unwrap_or(10).min(100);
if !is_valid_json_field_path(&field) { return Err(mlua::Error::runtime( "invalid search field: use alphanumeric names with optional dot notation and array indices (e.g. 'name', 'author.handle', 'tags[0]')", )); }
let like_pattern = format!("%{query}%");
// Cannot use adapt_sql: Postgres reuses $3 for two bind positions, // while SQLite needs separate ? for each. Different bind counts. let rows: Vec<(String, String, String)> = match backend { DatabaseBackend::Sqlite => { let sql = format!( "SELECT uri, did, record FROM happyview_records \ WHERE collection = ? \ AND json_extract(record, '$.{field}') LIKE ? COLLATE NOCASE \ ORDER BY \ CASE \ WHEN LOWER(json_extract(record, '$.{field}')) = LOWER(?) THEN 0 \ WHEN LOWER(json_extract(record, '$.{field}')) LIKE LOWER(?) || '%' THEN 1 \ ELSE 2 \ END, \ json_extract(record, '$.{field}') \ LIMIT ?" ); sqlx::query_as(&sql) .bind(&collection) .bind(&like_pattern) .bind(&query) .bind(&query) .bind(limit) .fetch_all(&state.db) .await .map_err(|e| mlua::Error::runtime(format!("DB search failed: {e}")))? } DatabaseBackend::Postgres => { let sql = format!( "SELECT uri, did, record FROM happyview_records \ WHERE collection = $1 \ AND record::jsonb->>'{field}' ILIKE $2 \ ORDER BY \ CASE \ WHEN LOWER(record::jsonb->>'{field}') = LOWER($3) THEN 0 \ WHEN LOWER(record::jsonb->>'{field}') LIKE LOWER($3) || '%' THEN 1 \ ELSE 2 \ END, \ record::jsonb->>'{field}' \ LIMIT $4" ); sqlx::query_as(&sql) .bind(&collection) .bind(&like_pattern) .bind(&query) .bind(limit) .fetch_all(&state.db) .await .map_err(|e| mlua::Error::runtime(format!("DB search failed: {e}")))? } };
let records: Vec<Value> = rows .into_iter() .map(|(uri, _did, record_str)| { let mut record: Value = serde_json::from_str(&record_str).unwrap_or(json!({})); if let Some(obj) = record.as_object_mut() { obj.insert("uri".to_string(), json!(uri)); } record }) .collect();
let record_values: Vec<mlua::Value> = records .iter() .map(|r| lua.to_value(r)) .collect::<LuaResult<_>>()?; let records_table = lua.create_sequence_from(record_values)?; records_table.set_metatable(Some(lua.array_metatable()))?;
let result_table = lua.create_table()?; result_table.set("records", records_table)?;
Ok(mlua::Value::Table(result_table)) } })?; db_table.set("search", search_fn)?;
// db.count(collection, did?) -> integer let state_count = state.clone(); let count_fn = lua.create_async_function(move |_, (collection, did): (String, Option<String>)| { let state = state_count.clone(); async move { let backend = state.db_backend; let count: (i64,) = if let Some(ref did) = did { let sql = adapt_sql( "SELECT COUNT(*) FROM happyview_records WHERE collection = ? AND did = ?", backend, ); sqlx::query_as(&sql) .bind(&collection) .bind(did) .fetch_one(&state.db) .await .map_err(|e| mlua::Error::runtime(format!("DB count failed: {e}")))? } else { let sql = adapt_sql( "SELECT COUNT(*) FROM happyview_records WHERE collection = ?", backend, ); sqlx::query_as(&sql) .bind(&collection) .fetch_one(&state.db) .await .map_err(|e| mlua::Error::runtime(format!("DB count failed: {e}")))? }; Ok(count.0) } })?; db_table.set("count", count_fn)?;
// db.backlinks({ collection, uri, did?, limit?, cursor? }) -> { records, cursor? } // Find records in `collection` that reference the given AT URI via record_refs. let state_backlinks = state.clone(); let backlinks_fn = lua.create_async_function(move |lua, opts: mlua::Table| { let state = state_backlinks.clone(); async move { let backend = state.db_backend; let collection: String = opts.get("collection")?; let uri: String = opts.get("uri")?; let did: Option<String> = opts.get("did").ok(); let limit: i64 = opts.get::<i64>("limit").unwrap_or(20).min(100); let cursor_str: Option<String> = opts.get("cursor").ok();
let cursor_parts = cursor_str.as_ref().and_then(|c| decode_cursor(c));
type RowType = (String, String, String, String);
let rows_raw: Vec<RowType> = match (&did, &cursor_parts) { (Some(did), Some((cursor_ts, cursor_uri))) => { let sql = adapt_sql( "SELECT r.uri, r.did, r.record, r.created_at FROM happyview_records r \ INNER JOIN happyview_record_refs ref ON ref.source_uri = r.uri \ WHERE ref.target_uri = ? AND ref.collection = ? AND r.did = ? \ AND (r.created_at < ? OR (r.created_at = ? AND r.uri < ?)) \ ORDER BY r.created_at DESC, r.uri DESC \ LIMIT ?", backend, ); sqlx::query_as(&sql) .bind(&uri) .bind(&collection) .bind(did) .bind(cursor_ts) .bind(cursor_ts) .bind(cursor_uri) .bind(limit) .fetch_all(&state.db) .await .map_err(|e| mlua::Error::runtime(format!("DB backlinks failed: {e}")))? } (Some(did), None) => { let sql = adapt_sql( "SELECT r.uri, r.did, r.record, r.created_at FROM happyview_records r \ INNER JOIN happyview_record_refs ref ON ref.source_uri = r.uri \ WHERE ref.target_uri = ? AND ref.collection = ? AND r.did = ? \ ORDER BY r.created_at DESC, r.uri DESC \ LIMIT ?", backend, ); sqlx::query_as(&sql) .bind(&uri) .bind(&collection) .bind(did) .bind(limit) .fetch_all(&state.db) .await .map_err(|e| mlua::Error::runtime(format!("DB backlinks failed: {e}")))? } (None, Some((cursor_ts, cursor_uri))) => { let sql = adapt_sql( "SELECT r.uri, r.did, r.record, r.created_at FROM happyview_records r \ INNER JOIN happyview_record_refs ref ON ref.source_uri = r.uri \ WHERE ref.target_uri = ? AND ref.collection = ? \ AND (r.created_at < ? OR (r.created_at = ? AND r.uri < ?)) \ ORDER BY r.created_at DESC, r.uri DESC \ LIMIT ?", backend, ); sqlx::query_as(&sql) .bind(&uri) .bind(&collection) .bind(cursor_ts) .bind(cursor_ts) .bind(cursor_uri) .bind(limit) .fetch_all(&state.db) .await .map_err(|e| mlua::Error::runtime(format!("DB backlinks failed: {e}")))? } (None, None) => { let sql = adapt_sql( "SELECT r.uri, r.did, r.record, r.created_at FROM happyview_records r \ INNER JOIN happyview_record_refs ref ON ref.source_uri = r.uri \ WHERE ref.target_uri = ? AND ref.collection = ? \ ORDER BY r.created_at DESC, r.uri DESC \ LIMIT ?", backend, ); sqlx::query_as(&sql) .bind(&uri) .bind(&collection) .bind(limit) .fetch_all(&state.db) .await .map_err(|e| mlua::Error::runtime(format!("DB backlinks failed: {e}")))? } };
let has_next = rows_raw.len() as i64 == limit;
let result_table = lua.create_table()?;
if has_next && let Some((last_uri, _, _, last_created_at)) = rows_raw.last() { let cursor = encode_cursor(last_created_at, last_uri); result_table.set("cursor", cursor)?; }
let records: Vec<Value> = rows_raw .into_iter() .map(|(uri, _did, record_str, _created_at)| { let mut record: Value = serde_json::from_str(&record_str).unwrap_or(json!({})); if let Some(obj) = record.as_object_mut() { obj.insert("uri".to_string(), json!(uri)); } record }) .collect();
let record_values: Vec<mlua::Value> = records .iter() .map(|r| lua.to_value(r)) .collect::<LuaResult<_>>()?; let records_table = lua.create_sequence_from(record_values)?; records_table.set_metatable(Some(lua.array_metatable()))?; result_table.set("records", records_table)?;
Ok(mlua::Value::Table(result_table)) } })?; db_table.set("backlinks", backlinks_fn)?;
// db.raw(sql, params?) -> rows[] let state_raw = state.clone(); let raw_fn = lua.create_async_function(move |lua, (sql, params): (String, Option<mlua::Table>)| { let state = state_raw.clone(); async move { // Protect HappyView's internal tables (secrets, auth, config, // AppView bookkeeping) from raw access; own tables are fine. check_raw_sql_tables(&sql).map_err(mlua::Error::runtime)?;
let mut query = sqlx::query(&sql); if let Some(ref params_table) = params { for value in params_table.sequence_values::<mlua::Value>() { let value = value?; query = match value { mlua::Value::String(s) => query.bind(s.to_str()?.to_string()), mlua::Value::Integer(n) => query.bind(n), mlua::Value::Number(n) => query.bind(n), mlua::Value::Boolean(b) => query.bind(if b { 1_i32 } else { 0_i32 }), mlua::Value::Nil => query.bind(Option::<String>::None), other => { return Err(mlua::Error::runtime(format!( "unsupported parameter type: {}", other.type_name() ))); } }; } }
let rows = query .fetch_all(&state.db) .await .map_err(|e| mlua::Error::runtime(format!("db.raw query failed: {e}")))?;
// Convert rows to Lua tables let mut lua_rows: Vec<mlua::Value> = Vec::with_capacity(rows.len()); for row in &rows { let row_table = lua.create_table()?; for col in row.columns() { let name = col.name(); let lua_val: mlua::Value = match row.try_get::<String, _>(name) { Ok(s) => mlua::Value::String(lua.create_string(&s)?), Err(_) => match row.try_get::<i64, _>(name) { Ok(n) => mlua::Value::Integer(n), Err(_) => match row.try_get::<i32, _>(name) { Ok(n) => mlua::Value::Integer(n as i64), Err(_) => match row.try_get::<f64, _>(name) { Ok(n) => mlua::Value::Number(n), Err(_) => match row.try_get::<bool, _>(name) { Ok(b) => mlua::Value::Boolean(b), Err(_) => mlua::Value::Nil, }, }, }, }, }; row_table.set(name, lua_val)?; } lua_rows.push(mlua::Value::Table(row_table)); }
let result = lua.create_sequence_from(lua_rows)?; result.set_metatable(Some(lua.array_metatable()))?; Ok(result) } })?; db_table.set("raw", raw_fn)?;
// db.backend() -> "sqlite" | "postgres" let backend = state.db_backend; let backend_fn = lua.create_function(move |_, ()| { Ok(match backend { DatabaseBackend::Sqlite => "sqlite", DatabaseBackend::Postgres => "postgres", }) })?; db_table.set("backend", backend_fn)?;
lua.globals().set("db", db_table)?; Ok(())}
#[cfg(test)]mod tests { use super::*; use crate::config::Config; use crate::db::DatabaseBackend; use crate::lexicon::LexiconRegistry; 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(), base_path: None, event_log_retention_days: 30, app_name: None, logo_uri: None, tos_uri: None, policy_uri: None, token_encryption_key: None, default_rate_limit_capacity: 100, default_rate_limit_refill_rate: 2.0, }; 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(), backfill_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(), ))), backfill_events_tx: tokio::sync::broadcast::channel(16).0, verbose_event_logging: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), } }
fn setup(state: &AppState) -> Lua { let lua = Lua::new(); register_db_api(&lua, Arc::new(state.clone())).unwrap(); lua }
#[tokio::test] async fn raw_allows_non_select_on_own_tables() { let state = test_state(); let lua = setup(&state); // Non-SELECT statements are allowed against non-internal tables. Passes // table validation; may then fail on the (empty in-memory) DB. let result: Result<mlua::Value, _> = lua .load(r#"return db.raw("DELETE FROM my_table")"#) .eval_async() .await; if let Err(e) = &result { let err = e.to_string(); assert!( !err.contains("internal HappyView table"), "should have passed table validation but got: {err}" ); } }
#[tokio::test] async fn raw_blocks_internal_tables() { let state = test_state(); let lua = setup(&state); let result: Result<mlua::Value, _> = lua .load(r#"return db.raw("SELECT * FROM happyview_dpop_keys")"#) .eval_async() .await; let err = result.expect_err("querying an internal table must be rejected"); assert!( err.to_string().contains("internal HappyView table"), "expected an internal-table error, got: {err}" ); }
#[tokio::test] async fn raw_allows_select() { let state = test_state(); let lua = setup(&state); let result: Result<mlua::Value, _> = lua.load(r#"return db.raw("SELECT 1")"#).eval_async().await; // Should either succeed (SQLite in-memory) or fail with a DB connection error, // but NOT a validation error. if let Err(e) = &result { let err = e.to_string(); assert!( !err.contains("only supports SELECT"), "should have passed validation but got: {err}" ); } }
#[test] fn raw_sql_allows_non_protected_tables() { // Admins can get wild with their own tables. assert!(super::check_raw_sql_tables("SELECT * FROM my_table").is_ok()); assert!(super::check_raw_sql_tables("CREATE TABLE analytics (id INT)").is_ok()); assert!(super::check_raw_sql_tables("INSERT INTO analytics VALUES (1)").is_ok()); assert!(super::check_raw_sql_tables("UPDATE analytics SET id = 2").is_ok()); assert!(super::check_raw_sql_tables("DELETE FROM analytics WHERE id = 1").is_ok()); assert!(super::check_raw_sql_tables("DROP TABLE analytics").is_ok()); // A table that merely *contains* the prefix mid-name is fine. assert!(super::check_raw_sql_tables("SELECT * FROM myhappyview_data").is_ok()); // The prefix appearing inside a string literal is not a table reference. assert!( super::check_raw_sql_tables("INSERT INTO logs (msg) VALUES ('happyview_started')") .is_ok() ); }
#[test] fn raw_sql_allows_allowlisted_internal_tables() { // Public AppView data and background jobs are readable/writable. assert!(super::check_raw_sql_tables("SELECT * FROM happyview_records").is_ok()); assert!( super::check_raw_sql_tables("DELETE FROM happyview_records WHERE uri = $1").is_ok() ); assert!(super::check_raw_sql_tables("SELECT * FROM happyview_record_refs").is_ok()); assert!(super::check_raw_sql_tables("SELECT * FROM happyview_labels").is_ok()); assert!(super::check_raw_sql_tables("SELECT * FROM happyview_lexicons").is_ok()); assert!(super::check_raw_sql_tables("SELECT * FROM happyview_jobs").is_ok()); // Space data (access, not confidentiality). assert!(super::check_raw_sql_tables("SELECT * FROM happyview_space_records").is_ok()); assert!(super::check_raw_sql_tables("SELECT * FROM happyview_space_members").is_ok()); }
#[test] fn raw_sql_blocks_protected_tables() { // Secrets / tokens / keys. assert!(super::check_raw_sql_tables("SELECT * FROM happyview_dpop_keys").is_err()); assert!(super::check_raw_sql_tables("DROP TABLE happyview_api_keys").is_err()); assert!(super::check_raw_sql_tables("SELECT * FROM happyview_script_variables").is_err()); // Auth / privilege / trust config. assert!(super::check_raw_sql_tables("UPDATE happyview_users SET is_super = true").is_err()); assert!(super::check_raw_sql_tables("SELECT * FROM happyview_domains").is_err()); assert!(super::check_raw_sql_tables("SELECT * FROM happyview_instance_settings").is_err()); // Space credential / key-material tables stay blocked even though other // space tables are allowed. assert!(super::check_raw_sql_tables("SELECT * FROM happyview_space_credentials").is_err()); assert!(super::check_raw_sql_tables("SELECT * FROM happyview_space_repo_state").is_err()); // The migration bookkeeping table is off-limits too. assert!(super::check_raw_sql_tables("SELECT * FROM _sqlx_migrations").is_err()); }
#[test] fn raw_sql_blocks_protected_tables_evasion() { // Case-insensitive. assert!(super::check_raw_sql_tables("SELECT * FROM HAPPYVIEW_USERS").is_err()); // Double-quoted identifier. assert!(super::check_raw_sql_tables(r#"SELECT * FROM "happyview_api_keys""#).is_err()); // Schema-qualified. assert!( super::check_raw_sql_tables("SELECT * FROM public.happyview_dpop_sessions").is_err() ); // Second statement in a batch. assert!(super::check_raw_sql_tables("SELECT 1; SELECT * FROM happyview_users").is_err()); // JOIN / subquery position. assert!( super::check_raw_sql_tables( "SELECT * FROM my_table JOIN happyview_api_clients USING (id)" ) .is_err() ); // Unicode-escaped identifier evasion is refused outright. assert!(super::check_raw_sql_tables(r#"SELECT * FROM U&"happyview_dpop_keys""#).is_err()); }
#[test] fn valid_json_field_paths() { assert!(super::is_valid_json_field_path("name")); assert!(super::is_valid_json_field_path("author_name")); assert!(super::is_valid_json_field_path("author.handle")); assert!(super::is_valid_json_field_path("tags[0]")); assert!(super::is_valid_json_field_path("data[0][1]")); assert!(super::is_valid_json_field_path("author.websites[0].url")); assert!(super::is_valid_json_field_path("a.b.c.d.e")); }
#[test] fn invalid_json_field_paths() { assert!(!super::is_valid_json_field_path("")); assert!(!super::is_valid_json_field_path(".name")); assert!(!super::is_valid_json_field_path("name.")); assert!(!super::is_valid_json_field_path("name..foo")); assert!(!super::is_valid_json_field_path("[0]")); assert!(!super::is_valid_json_field_path("name[]")); assert!(!super::is_valid_json_field_path("name[abc]")); assert!(!super::is_valid_json_field_path("name; DROP TABLE")); assert!(!super::is_valid_json_field_path("name'OR 1=1")); assert!(!super::is_valid_json_field_path("na-me")); }
#[tokio::test] async fn query_accepts_nested_sort_field() { let state = test_state(); let lua = setup(&state); let result: Result<mlua::Value, _> = lua .load(r#"return db.query({ collection = "test", sort = "author.handle" })"#) .eval_async() .await; assert!(result.is_err()); let err = result.unwrap_err().to_string(); assert!( !err.contains("invalid sort field"), "nested sort field should be accepted, got: {err}" ); }
#[tokio::test] async fn query_accepts_array_index_sort_field() { let state = test_state(); let lua = setup(&state); let result: Result<mlua::Value, _> = lua .load(r#"return db.query({ collection = "test", sort = "tags[0]" })"#) .eval_async() .await; assert!(result.is_err()); let err = result.unwrap_err().to_string(); assert!( !err.contains("invalid sort field"), "array index sort field should be accepted, got: {err}" ); }
#[tokio::test] async fn query_rejects_invalid_sort_field() { let state = test_state(); let lua = setup(&state); let result: Result<mlua::Value, _> = lua .load(r#"return db.query({ collection = "test", sort = "name; DROP TABLE" })"#) .eval_async() .await; assert!(result.is_err()); let err = result.unwrap_err().to_string(); assert!( err.contains("invalid sort field"), "expected sort field error, got: {err}" ); }
#[tokio::test] async fn query_rejects_invalid_sort_direction() { let state = test_state(); let lua = setup(&state); let result: Result<mlua::Value, _> = lua .load(r#"return db.query({ collection = "test", sortDirection = "sideways" })"#) .eval_async() .await; assert!(result.is_err()); let err = result.unwrap_err().to_string(); assert!( err.contains("invalid sortDirection"), "expected sortDirection error, got: {err}" ); }
#[test] fn cursor_round_trip() { let encoded = super::encode_cursor("2026-03-12T10:00:00Z", "at://did:plc:abc/col/rkey"); let (ts, uri) = super::decode_cursor(&encoded).unwrap(); assert_eq!(ts, "2026-03-12T10:00:00Z"); assert_eq!(uri, "at://did:plc:abc/col/rkey"); }
#[test] fn decode_invalid_cursor_returns_none() { assert!(super::decode_cursor("not-valid-base64!!!").is_none()); }
#[test] fn decode_cursor_missing_pipe_returns_none() { use base64::{Engine, engine::general_purpose::STANDARD as BASE64}; let encoded = BASE64.encode("no-pipe-here"); assert!(super::decode_cursor(&encoded).is_none()); }
// ----------------------------------------------------------------------- // parse_filter_node / build_filter_sql // -----------------------------------------------------------------------
fn make_condition_table(lua: &Lua, field: &str, op: &str, value: &str) -> mlua::Table { let t = lua.create_table().unwrap(); t.set("field", field).unwrap(); t.set("op", op).unwrap(); t.set("value", value).unwrap(); t }
#[test] fn filter_simple_condition() { let lua = Lua::new(); let t = make_condition_table(&lua, "name", "=", "alice"); let node = parse_filter_node(&t, 0).unwrap(); let mut binds = Vec::new(); let sql = build_filter_sql(&node, &mut binds); assert_eq!(sql, "json_extract(record, '$.name') = ?"); assert_eq!(binds, vec!["alice"]); }
#[test] fn filter_defaults_op_to_equals() { let lua = Lua::new(); let t = lua.create_table().unwrap(); t.set("field", "status").unwrap(); t.set("value", "active").unwrap(); let node = parse_filter_node(&t, 0).unwrap(); let mut binds = Vec::new(); let sql = build_filter_sql(&node, &mut binds); assert_eq!(sql, "json_extract(record, '$.status') = ?"); }
#[test] fn filter_rejects_invalid_op() { let lua = Lua::new(); let t = make_condition_table(&lua, "name", "DROP", "x"); let err = parse_filter_node(&t, 0).unwrap_err(); assert!(err.to_string().contains("invalid filter op")); }
#[test] fn filter_rejects_invalid_field() { let lua = Lua::new(); let t = make_condition_table(&lua, "name; DROP TABLE", "=", "x"); let err = parse_filter_node(&t, 0).unwrap_err(); assert!(err.to_string().contains("invalid filter field")); }
#[test] fn filter_and_group() { let lua = Lua::new(); let group = lua.create_table().unwrap(); group.set("combine", "AND").unwrap(); let c1 = make_condition_table(&lua, "status", "=", "active"); let c2 = make_condition_table(&lua, "age", ">", "18"); group.set(1, c1).unwrap(); group.set(2, c2).unwrap(); let node = parse_filter_node(&group, 0).unwrap(); let mut binds = Vec::new(); let sql = build_filter_sql(&node, &mut binds); assert_eq!( sql, "(json_extract(record, '$.status') = ? AND json_extract(record, '$.age') > ?)" ); assert_eq!(binds, vec!["active", "18"]); }
#[test] fn filter_or_group() { let lua = Lua::new(); let group = lua.create_table().unwrap(); group.set("combine", "OR").unwrap(); let c1 = make_condition_table(&lua, "role", "=", "admin"); let c2 = make_condition_table(&lua, "role", "=", "mod"); group.set(1, c1).unwrap(); group.set(2, c2).unwrap(); let node = parse_filter_node(&group, 0).unwrap(); let mut binds = Vec::new(); let sql = build_filter_sql(&node, &mut binds); assert_eq!( sql, "(json_extract(record, '$.role') = ? OR json_extract(record, '$.role') = ?)" ); assert_eq!(binds, vec!["admin", "mod"]); }
#[test] fn filter_single_child_group_unwraps() { let lua = Lua::new(); let group = lua.create_table().unwrap(); group.set("combine", "AND").unwrap(); let c1 = make_condition_table(&lua, "x", "=", "1"); group.set(1, c1).unwrap(); let node = parse_filter_node(&group, 0).unwrap(); let mut binds = Vec::new(); let sql = build_filter_sql(&node, &mut binds); assert_eq!(sql, "json_extract(record, '$.x') = ?"); }
#[test] fn filter_rejects_invalid_combine() { let lua = Lua::new(); let group = lua.create_table().unwrap(); group.set("combine", "XOR").unwrap(); let c1 = make_condition_table(&lua, "x", "=", "1"); group.set(1, c1).unwrap(); let err = parse_filter_node(&group, 0).unwrap_err(); assert!(err.to_string().contains("invalid filter combine")); }
#[test] fn filter_rejects_empty_group() { let lua = Lua::new(); let group = lua.create_table().unwrap(); group.set("combine", "AND").unwrap(); let err = parse_filter_node(&group, 0).unwrap_err(); assert!(err.to_string().contains("filter group has no conditions")); }
#[test] fn filter_rejects_excessive_depth() { let lua = Lua::new(); let c = make_condition_table(&lua, "x", "=", "1"); let err = parse_filter_node(&c, MAX_FILTER_DEPTH).unwrap_err(); assert!(err.to_string().contains("filter nesting too deep")); }
#[test] fn filter_accepts_all_ops() { let lua = Lua::new(); for op in ALLOWED_OPS { let t = make_condition_table(&lua, "field", op, "val"); assert!( parse_filter_node(&t, 0).is_ok(), "op '{op}' should be accepted" ); } }
#[test] fn filter_op_case_insensitive() { let lua = Lua::new(); let t = make_condition_table(&lua, "name", "like", "alice%"); let node = parse_filter_node(&t, 0).unwrap(); let mut binds = Vec::new(); let sql = build_filter_sql(&node, &mut binds); assert_eq!(sql, "json_extract(record, '$.name') LIKE ?"); }
#[test] fn filter_integer_value() { let lua = Lua::new(); let t = lua.create_table().unwrap(); t.set("field", "count").unwrap(); t.set("op", ">").unwrap(); t.set("value", 42).unwrap(); let node = parse_filter_node(&t, 0).unwrap(); let mut binds = Vec::new(); build_filter_sql(&node, &mut binds); assert_eq!(binds, vec!["42"]); }
#[test] fn filter_boolean_value() { let lua = Lua::new(); let t = lua.create_table().unwrap(); t.set("field", "active").unwrap(); t.set("value", true).unwrap(); let node = parse_filter_node(&t, 0).unwrap(); let mut binds = Vec::new(); build_filter_sql(&node, &mut binds); assert_eq!(binds, vec!["true"]); }
#[test] fn filter_nested_field_path() { let lua = Lua::new(); let t = make_condition_table(&lua, "author.websites[0].url", "=", "https://example.com"); let node = parse_filter_node(&t, 0).unwrap(); let mut binds = Vec::new(); let sql = build_filter_sql(&node, &mut binds); assert_eq!(sql, "json_extract(record, '$.author.websites[0].url') = ?"); }
// ----------------------------------------------------------------------- // query sort direction // -----------------------------------------------------------------------
#[tokio::test] async fn query_accepts_valid_sort_direction() { let state = test_state(); let lua = setup(&state); let result: Result<mlua::Value, _> = lua .load(r#"return db.query({ collection = "test", sortDirection = "asc" })"#) .eval_async() .await; // Should fail with a DB connection error, NOT a validation error assert!(result.is_err()); let err = result.unwrap_err().to_string(); assert!( !err.contains("invalid sortDirection"), "should have passed validation but got: {err}" ); }}