Something went wrong. Try again.
A lexicon-driven AppView for ATProto.
Something went wrong. Try again.
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545mod 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 serial_test::serial;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(), ))), }}
async fn seed_record( pool: &sqlx::AnyPool, backend: DatabaseBackend, uri: &str, did: &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("test.collection") .bind("rkey1") .bind(serde_json::to_string(&record).unwrap_or_default()) .bind("bafytest") .bind(now_rfc3339()) .execute(pool) .await .expect("failed to seed record");}
async fn seed_label( pool: &sqlx::AnyPool, backend: DatabaseBackend, src: &str, uri: &str, val: &str, exp: Option<&str>,) { if let Some(exp) = exp { let sql = adapt_sql( "INSERT INTO labels (src, uri, val, cts, exp) VALUES (?, ?, ?, ?, ?)", backend, ); sqlx::query(&sql) .bind(src) .bind(uri) .bind(val) .bind(now_rfc3339()) .bind(exp) .execute(pool) .await .expect("failed to seed label"); } else { let sql = adapt_sql( "INSERT INTO labels (src, uri, val, cts) VALUES (?, ?, ?, ?)", backend, ); sqlx::query(&sql) .bind(src) .bind(uri) .bind(val) .bind(now_rfc3339()) .execute(pool) .await .expect("failed to seed label"); }}
// ---------------------------------------------------------------------------// get_labels tests// ---------------------------------------------------------------------------
#[tokio::test]#[serial]async fn get_labels_returns_external_labels() { common::require_db!(); 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", serde_json::json!({"name": "test"}), ) .await; seed_label( &pool, backend, "did:plc:labeler1", uri, "adult-content", None, ) .await; seed_label(&pool, backend, "did:plc:labeler1", uri, "violence", None).await;
let state = test_state_with_pool(pool, backend).await;
let now = now_rfc3339(); let sql = adapt_sql( "SELECT src, uri, val FROM labels WHERE uri = ? AND (exp IS NULL OR exp > ?)", backend, ); let rows: Vec<(String, String, String)> = sqlx::query_as(&sql) .bind(uri) .bind(&now) .fetch_all(&state.db) .await .unwrap();
assert_eq!(rows.len(), 2); assert_eq!(rows[0].2, "adult-content"); assert_eq!(rows[1].2, "violence");}
#[tokio::test]#[serial]async fn get_labels_filters_expired() { common::require_db!(); 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", serde_json::json!({"name": "test"}), ) .await;
// Active label seed_label(&pool, backend, "did:plc:labeler1", uri, "nudity", None).await; // Expired label (past date) seed_label( &pool, backend, "did:plc:labeler1", uri, "spam", Some("2020-01-01T00:00:00Z"), ) .await;
let state = test_state_with_pool(pool, backend).await;
let now = now_rfc3339(); let sql = adapt_sql( "SELECT src, uri, val FROM labels WHERE uri = ? AND (exp IS NULL OR exp > ?)", backend, ); let rows: Vec<(String, String, String)> = sqlx::query_as(&sql) .bind(uri) .bind(&now) .fetch_all(&state.db) .await .unwrap();
assert_eq!(rows.len(), 1); assert_eq!(rows[0].2, "nudity");}
#[tokio::test]#[serial]async fn get_labels_includes_self_labels() { common::require_db!(); let pool = db::test_pool().await; let backend = db::test_backend(); db::truncate_all(&pool).await;
let uri = "at://did:plc:author/test.collection/rkey1"; let record = serde_json::json!({ "name": "test", "labels": { "values": [ { "val": "sexual" }, { "val": "graphic-media" } ] } }); seed_record(&pool, backend, uri, "did:plc:author", record.clone()).await;
let sql = adapt_sql("SELECT did, record FROM records WHERE uri = ?", backend); let fetched: Option<(String, String)> = sqlx::query_as(&sql) .bind(uri) .fetch_optional(&pool) .await .unwrap();
let (did, rec_str) = fetched.unwrap(); assert_eq!(did, "did:plc:author");
let rec: serde_json::Value = serde_json::from_str(&rec_str).unwrap(); let self_labels: Vec<&str> = rec .get("labels") .and_then(|l| l.get("values")) .and_then(|v| v.as_array()) .unwrap() .iter() .filter_map(|item| item.get("val").and_then(|v| v.as_str())) .collect();
assert_eq!(self_labels, vec!["sexual", "graphic-media"]);}
#[tokio::test]#[serial]async fn get_labels_empty_for_unlabeled_record() { common::require_db!(); 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", serde_json::json!({"name": "test"}), ) .await;
let now = now_rfc3339(); let sql = adapt_sql( "SELECT src, uri, val FROM labels WHERE uri = ? AND (exp IS NULL OR exp > ?)", backend, ); let rows: Vec<(String, String, String)> = sqlx::query_as(&sql) .bind(uri) .bind(&now) .fetch_all(&pool) .await .unwrap();
assert!(rows.is_empty());}
// ---------------------------------------------------------------------------// get_labels_batch tests// ---------------------------------------------------------------------------
#[tokio::test]#[serial]async fn get_labels_batch_returns_labels_per_uri() { common::require_db!(); let pool = db::test_pool().await; let backend = db::test_backend(); db::truncate_all(&pool).await;
let uri1 = "at://did:plc:test/test.collection/rkey1"; let uri2 = "at://did:plc:test/test.collection/rkey2";
seed_record( &pool, backend, uri1, "did:plc:test", serde_json::json!({"name": "one"}), ) .await;
let sql = adapt_sql( "INSERT INTO records (uri, did, collection, rkey, record, cid, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)", backend, ); sqlx::query(&sql) .bind(uri2) .bind("did:plc:test") .bind("test.collection") .bind("rkey2") .bind(serde_json::to_string(&serde_json::json!({"name": "two"})).unwrap_or_default()) .bind("bafytest2") .bind(now_rfc3339()) .execute(&pool) .await .unwrap();
seed_label(&pool, backend, "did:plc:labeler1", uri1, "nudity", None).await; seed_label(&pool, backend, "did:plc:labeler1", uri2, "spam", None).await; seed_label(&pool, backend, "did:plc:labeler2", uri2, "violence", None).await;
let now = now_rfc3339(); let sql = adapt_sql( "SELECT src, uri, val FROM labels WHERE uri IN (?, ?) AND (exp IS NULL OR exp > ?)", backend, ); let rows: Vec<(String, String, String)> = sqlx::query_as(&sql) .bind(uri1) .bind(uri2) .bind(&now) .fetch_all(&pool) .await .unwrap();
// uri1 has 1 label, uri2 has 2 labels let uri1_labels: Vec<_> = rows.iter().filter(|r| r.1 == uri1).collect(); let uri2_labels: Vec<_> = rows.iter().filter(|r| r.1 == uri2).collect();
assert_eq!(uri1_labels.len(), 1); assert_eq!(uri1_labels[0].2, "nudity"); assert_eq!(uri2_labels.len(), 2);}
#[tokio::test]#[serial]async fn get_labels_batch_empty_for_no_labels() { common::require_db!(); let pool = db::test_pool().await; let backend = db::test_backend(); db::truncate_all(&pool).await;
let uri1 = "at://did:plc:test/test.collection/rkey1"; let uri2 = "at://did:plc:test/test.collection/rkey2";
let now = now_rfc3339(); let sql = adapt_sql( "SELECT src, uri, val FROM labels WHERE uri IN (?, ?) AND (exp IS NULL OR exp > ?)", backend, ); let rows: Vec<(String, String, String)> = sqlx::query_as(&sql) .bind(uri1) .bind(uri2) .bind(&now) .fetch_all(&pool) .await .unwrap();
assert!(rows.is_empty());}
// ---------------------------------------------------------------------------// Label negation (materialized state)// ---------------------------------------------------------------------------
#[tokio::test]#[serial]async fn label_negation_removes_row() { common::require_db!(); 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";
// Add a label seed_label(&pool, backend, "did:plc:labeler1", uri, "nudity", None).await;
// Verify it exists let sql = adapt_sql( "SELECT COUNT(*) FROM labels WHERE src = ? AND uri = ? AND val = ?", backend, ); let count: (i64,) = sqlx::query_as(&sql) .bind("did:plc:labeler1") .bind(uri) .bind("nudity") .fetch_one(&pool) .await .unwrap(); assert_eq!(count.0, 1);
// Simulate negation (same logic as labeler.rs) let sql = adapt_sql( "DELETE FROM labels WHERE src = ? AND uri = ? AND val = ?", backend, ); sqlx::query(&sql) .bind("did:plc:labeler1") .bind(uri) .bind("nudity") .execute(&pool) .await .unwrap();
// Verify it's gone let sql = adapt_sql( "SELECT COUNT(*) FROM labels WHERE src = ? AND uri = ? AND val = ?", backend, ); let count: (i64,) = sqlx::query_as(&sql) .bind("did:plc:labeler1") .bind(uri) .bind("nudity") .fetch_one(&pool) .await .unwrap(); assert_eq!(count.0, 0);}
// ---------------------------------------------------------------------------// Label upsert (idempotent)// ---------------------------------------------------------------------------
#[tokio::test]#[serial]async fn label_upsert_is_idempotent() { common::require_db!(); 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";
let upsert_sql = match backend { DatabaseBackend::Postgres => { "INSERT INTO labels (src, uri, val, cts) VALUES ($1, $2, $3, $4) ON CONFLICT (src, uri, val) DO UPDATE SET cts = EXCLUDED.cts".to_string() } DatabaseBackend::Sqlite => { "INSERT INTO labels (src, uri, val, cts) VALUES (?, ?, ?, ?) ON CONFLICT (src, uri, val) DO UPDATE SET cts = excluded.cts".to_string() } };
// Insert same label twice (upsert pattern from labeler.rs) for _ in 0..2 { sqlx::query(&upsert_sql) .bind("did:plc:labeler1") .bind(uri) .bind("nudity") .bind(now_rfc3339()) .execute(&pool) .await .unwrap(); }
let sql = adapt_sql( "SELECT COUNT(*) FROM labels WHERE src = ? AND uri = ? AND val = ?", backend, ); let count: (i64,) = sqlx::query_as(&sql) .bind("did:plc:labeler1") .bind(uri) .bind("nudity") .fetch_one(&pool) .await .unwrap(); assert_eq!(count.0, 1);}