Something went wrong. Try again.
A lexicon-driven AppView for ATProto.
Something went wrong. Try again.
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520mod 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::db_api::register_db_api;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(), 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(()); 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(), ))), backfill_db: pool.clone(), backfill_events_tx: tokio::sync::broadcast::channel(16).0, verbose_event_logging: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)), }}
async fn seed_records(pool: &sqlx::AnyPool, backend: DatabaseBackend) { let records = [ ( "at://did:plc:test/test.collection/rkey1", "did:plc:test", "test.collection", "rkey1", serde_json::json!({"name": "Test One", "value": 1}), "bafyone", ), ( "at://did:plc:test/test.collection/rkey2", "did:plc:test", "test.collection", "rkey2", serde_json::json!({"name": "Test Two", "value": 2}), "bafytwo", ), ( "at://did:plc:other/test.collection/rkey3", "did:plc:other", "test.collection", "rkey3", serde_json::json!({"name": "Other Record", "value": 3}), "bafythree", ), ];
let now = now_rfc3339(); let sql = adapt_sql( "INSERT INTO happyview_records (uri, did, collection, rkey, record, cid, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)", backend, ); for (uri, did, collection, rkey, record, cid) in &records { happyview::db::query(&sql) .bind(uri) .bind(did) .bind(collection) .bind(rkey) .bind(serde_json::to_string(record).unwrap_or_default()) .bind(cid) .bind(&now) .execute(pool) .await .expect("failed to seed record"); }}
fn setup_lua(state: &AppState) -> Lua { let lua = Lua::new(); register_db_api(&lua, Arc::new(state.clone())).unwrap(); lua}
// ---------------------------------------------------------------------------// Tests// ---------------------------------------------------------------------------
#[tokio::test]#[serial]async fn db_get_returns_record() { common::require_db!(); let pool = db::test_pool().await; let backend = db::test_backend(); db::truncate_all(&pool).await; seed_records(&pool, backend).await; let state = test_state_with_pool(pool, backend).await; let lua = setup_lua(&state);
let result: mlua::Table = lua .load(r#"return db.get("at://did:plc:test/test.collection/rkey1")"#) .eval_async() .await .unwrap();
assert_eq!( result.get::<String>("uri").unwrap(), "at://did:plc:test/test.collection/rkey1" ); assert_eq!(result.get::<String>("name").unwrap(), "Test One");}
#[tokio::test]#[serial]async fn db_get_returns_nil_for_missing() { common::require_db!(); 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_lua(&state);
let result: mlua::Value = lua .load(r#"return db.get("at://did:plc:nonexistent/test.collection/nope")"#) .eval_async() .await .unwrap();
assert!(result.is_nil());}
#[tokio::test]#[serial]async fn db_query_returns_records() { common::require_db!(); let pool = db::test_pool().await; let backend = db::test_backend(); db::truncate_all(&pool).await; seed_records(&pool, backend).await; let state = test_state_with_pool(pool, backend).await; let lua = setup_lua(&state);
let result: mlua::Table = lua .load(r#"return db.query({ collection = "test.collection" })"#) .eval_async() .await .unwrap();
let records: mlua::Table = result.get("records").unwrap(); assert_eq!(records.raw_len(), 3);}
#[tokio::test]#[serial]async fn db_query_respects_limit() { common::require_db!(); let pool = db::test_pool().await; let backend = db::test_backend(); db::truncate_all(&pool).await; seed_records(&pool, backend).await; let state = test_state_with_pool(pool, backend).await; let lua = setup_lua(&state);
let result: mlua::Table = lua .load(r#"return db.query({ collection = "test.collection", limit = 1 })"#) .eval_async() .await .unwrap();
let records: mlua::Table = result.get("records").unwrap(); assert_eq!(records.raw_len(), 1);
// Should have a cursor since there are more records let cursor: String = result.get("cursor").unwrap(); assert!(!cursor.is_empty());}
#[tokio::test]#[serial]async fn db_count_returns_total() { common::require_db!(); let pool = db::test_pool().await; let backend = db::test_backend(); db::truncate_all(&pool).await; seed_records(&pool, backend).await; let state = test_state_with_pool(pool, backend).await; let lua = setup_lua(&state);
let count: i64 = lua .load(r#"return db.count("test.collection")"#) .eval_async() .await .unwrap();
assert_eq!(count, 3);}
#[tokio::test]#[serial]async fn db_count_with_did_filter() { common::require_db!(); let pool = db::test_pool().await; let backend = db::test_backend(); db::truncate_all(&pool).await; seed_records(&pool, backend).await; let state = test_state_with_pool(pool, backend).await; let lua = setup_lua(&state);
let count: i64 = lua .load(r#"return db.count("test.collection", "did:plc:test")"#) .eval_async() .await .unwrap();
assert_eq!(count, 2);}
#[tokio::test]#[serial]async fn db_search_finds_matching() { common::require_db!(); let pool = db::test_pool().await; let backend = db::test_backend(); db::truncate_all(&pool).await; seed_records(&pool, backend).await; let state = test_state_with_pool(pool, backend).await; let lua = setup_lua(&state);
let result: mlua::Table = lua .load( r#"return db.search({ collection = "test.collection", field = "name", query = "Test" })"#, ) .eval_async() .await .unwrap();
let records: mlua::Table = result.get("records").unwrap(); // "Test One" and "Test Two" match; "Other Record" does not assert_eq!(records.raw_len(), 2);}
#[tokio::test]#[serial]async fn db_raw_select_works() { common::require_db!(); 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_lua(&state);
// db.raw operates freely on the caller's own tables (internal HappyView // tables are protected — see raw_blocks_internal_table). let result: mlua::Table = lua .load( r#" db.raw("DROP TABLE IF EXISTS raw_probe") db.raw("CREATE TABLE raw_probe (n INT)") db.raw("INSERT INTO raw_probe (n) VALUES (1), (2), (3)") local rows = db.raw("SELECT COUNT(*) as cnt FROM raw_probe WHERE n >= $1", {2}) db.raw("DROP TABLE raw_probe") return rows "#, ) .eval_async() .await .unwrap();
// Result is an array of row tables let first_row: mlua::Table = result.get(1).unwrap(); let cnt: i64 = first_row.get("cnt").unwrap(); assert_eq!(cnt, 2);}
#[tokio::test]#[serial]async fn db_raw_blocks_internal_table() { common::require_db!(); 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_lua(&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("db.raw must reject internal HappyView tables"); assert!( err.to_string().contains("internal HappyView table"), "unexpected error: {err}" );}
#[tokio::test]#[serial]async fn db_query_filter_equals() { common::require_db!(); let pool = db::test_pool().await; let backend = db::test_backend(); db::truncate_all(&pool).await; seed_records(&pool, backend).await; let state = test_state_with_pool(pool, backend).await; let lua = setup_lua(&state);
let result: mlua::Table = lua .load( r#"return db.query({ collection = "test.collection", filter = { field = "name", value = "Test One" } })"#, ) .eval_async() .await .unwrap();
let records: mlua::Table = result.get("records").unwrap(); assert_eq!(records.raw_len(), 1); let first: mlua::Table = records.get(1).unwrap(); assert_eq!(first.get::<String>("name").unwrap(), "Test One");}
#[tokio::test]#[serial]async fn db_query_filter_not_equals() { common::require_db!(); let pool = db::test_pool().await; let backend = db::test_backend(); db::truncate_all(&pool).await; seed_records(&pool, backend).await; let state = test_state_with_pool(pool, backend).await; let lua = setup_lua(&state);
let result: mlua::Table = lua .load( r#"return db.query({ collection = "test.collection", filter = { field = "name", op = "!=", value = "Test One" } })"#, ) .eval_async() .await .unwrap();
let records: mlua::Table = result.get("records").unwrap(); assert_eq!(records.raw_len(), 2);}
#[tokio::test]#[serial]async fn db_query_filter_no_match_returns_empty() { common::require_db!(); let pool = db::test_pool().await; let backend = db::test_backend(); db::truncate_all(&pool).await; seed_records(&pool, backend).await; let state = test_state_with_pool(pool, backend).await; let lua = setup_lua(&state);
let result: mlua::Table = lua .load( r#"return db.query({ collection = "test.collection", filter = { field = "name", value = "Nonexistent" } })"#, ) .eval_async() .await .unwrap();
let records: mlua::Table = result.get("records").unwrap(); assert_eq!(records.raw_len(), 0);}
#[tokio::test]#[serial]async fn db_query_filter_and_group() { common::require_db!(); let pool = db::test_pool().await; let backend = db::test_backend(); db::truncate_all(&pool).await; seed_records(&pool, backend).await; let state = test_state_with_pool(pool, backend).await; let lua = setup_lua(&state);
let result: mlua::Table = lua .load( r#"return db.query({ collection = "test.collection", filter = { combine = "AND", { field = "name", op = "LIKE", value = "Test%" }, { field = "value", op = ">", value = "1" } } })"#, ) .eval_async() .await .unwrap();
let records: mlua::Table = result.get("records").unwrap(); assert_eq!(records.raw_len(), 1); let first: mlua::Table = records.get(1).unwrap(); assert_eq!(first.get::<String>("name").unwrap(), "Test Two");}
#[tokio::test]#[serial]async fn db_query_sort_by_json_field() { common::require_db!(); let pool = db::test_pool().await; let backend = db::test_backend(); db::truncate_all(&pool).await; seed_records(&pool, backend).await; let state = test_state_with_pool(pool, backend).await; let lua = setup_lua(&state);
let result: mlua::Table = lua .load( r#"return db.query({ collection = "test.collection", sort = "name", sortDirection = "asc" })"#, ) .eval_async() .await .unwrap();
let records: mlua::Table = result.get("records").unwrap(); assert_eq!(records.raw_len(), 3); let first: mlua::Table = records.get(1).unwrap(); let last: mlua::Table = records.get(3).unwrap(); assert_eq!(first.get::<String>("name").unwrap(), "Other Record"); assert_eq!(last.get::<String>("name").unwrap(), "Test Two");}