Something went wrong. Try again.
A lexicon-driven AppView for ATProto.
Something went wrong. Try again.
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777use std::sync::Arc;
use happyview::auth::oauth_store::{DbSessionStore, DbStateStore};use happyview::config::Config;use happyview::db;use happyview::dns::NativeDnsResolver;use happyview::lexicon::{LexiconRegistry, ParsedLexicon, ProcedureAction};use happyview::rate_limit::{RateLimitDefaults, RateLimiter};use happyview::resolve::{fetch_lexicon_from_pds, resolve_nsid_authority};use happyview::{AppState, jetstream, labeler, server};use sqlx::Row;use tokio::sync::watch;use tracing::{info, warn};
use atrium_identity::did::{CommonDidResolver, CommonDidResolverConfig};use atrium_identity::handle::{AtprotoHandleResolver, AtprotoHandleResolverConfig};use atrium_oauth::{ AtprotoClientMetadata, AtprotoLocalhostClientMetadata, AuthMethod, DefaultHttpClient, GrantType, KnownScope, OAuthClientConfig, OAuthResolverConfig, Scope,};
#[tokio::main]async fn main() { dotenvy::dotenv().ok();
// Install rustls crypto provider early so all TLS users (jetstream, labeler, etc.) can find it. let _ = rustls::crypto::ring::default_provider().install_default();
tracing_subscriber::fmt() .with_env_filter( tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| { "happyview=debug,tower_http=debug,sqlx=warn" .parse() .unwrap() }), ) .init();
let config = Config::from_env(); let db_backend = config.database_backend;
// Connect to database and run migrations. let db_pool = db::connect(&config.database_url, db_backend).await; let backfill_db_pool = db::connect_backfill_pool(&config.database_url, db_backend).await;
info!( backend = ?db_backend, "connected to database" );
// Backfill record_refs in the background (first run after upgrade) { let db_bg = db_pool.clone(); let backend = db_backend; tokio::spawn(async move { let count: (i64,) = crate::db::query_as("SELECT COUNT(*) FROM happyview_record_refs") .fetch_one(&db_bg) .await .expect("failed to count record_refs");
if count.0 == 0 { info!("backfilling record_refs table in background..."); let total: (i64,) = crate::db::query_as("SELECT COUNT(*) FROM happyview_records") .fetch_one(&db_bg) .await .expect("failed to count records"); let total = total.0 as usize;
let batch_size = 1000i64; let mut offset = 0i64; let mut processed = 0usize;
let query = db::adapt_sql( "SELECT uri, collection, record FROM happyview_records ORDER BY uri LIMIT ? OFFSET ?", backend, );
loop { let batch: Vec<(String, String, String)> = crate::db::query_as(&query) .bind(batch_size) .bind(offset) .fetch_all(&db_bg) .await .expect("failed to fetch records for backfill");
if batch.is_empty() { break; }
for (uri, collection, record_str) in &batch { let record: serde_json::Value = serde_json::from_str(record_str).unwrap_or(serde_json::Value::Null); if let Err(e) = happyview::record_refs::sync_refs( &db_bg, uri, collection, &record, backend, ) .await { warn!(uri = uri.as_str(), "failed to backfill refs: {e}"); } }
processed += batch.len(); offset += batch_size;
if processed.is_multiple_of(10000) || processed == total { info!("backfill progress: {processed}/{total}"); } }
info!("backfill complete: processed {processed} records"); } }); }
let lexicons = LexiconRegistry::new(); lexicons .load_from_db(&db_pool) .await .expect("failed to load lexicons");
// Re-fetch all network lexicons from their respective PDSes. let http = reqwest::Client::new(); let network_rows: Vec<(String, Option<String>, Option<String>)> = crate::db::query_as( "SELECT id, authority_did, target_collection FROM happyview_lexicons WHERE source = 'network'", ) .fetch_all(&db_pool) .await .unwrap_or_default();
for (nsid, _authority_did, target_collection) in &network_rows { match resolve_nsid_authority(&http, &config.plc_url, nsid).await { Ok((did, pds_endpoint)) => { match fetch_lexicon_from_pds(&http, &pds_endpoint, &did, nsid).await { Ok(lexicon_json) => { match ParsedLexicon::parse( lexicon_json.clone(), 1, target_collection.clone(), ProcedureAction::Upsert, None, ) { Ok(parsed) => { let now = db::now_rfc3339(); let update_sql = db::adapt_sql( "UPDATE happyview_lexicons SET lexicon_json = ?, last_fetched_at = ?, revision = revision + 1, updated_at = ? WHERE id = ? AND source = 'network'", db_backend, ); let lexicon_json_str = serde_json::to_string(&lexicon_json).unwrap_or_default(); if let Err(e) = crate::db::query(&update_sql) .bind(&lexicon_json_str) .bind(&now) .bind(&now) .bind(nsid) .execute(&db_pool) .await { warn!(nsid, "failed to update network lexicon in DB: {e}"); continue; }
lexicons.upsert(parsed).await; info!(nsid, "refreshed network lexicon"); } Err(e) => warn!(nsid, "failed to parse network lexicon: {e}"), } } Err(e) => warn!(nsid, "failed to fetch network lexicon from PDS: {e}"), } } Err(e) => warn!(nsid, "failed to resolve network lexicon authority: {e}"), } }
if !network_rows.is_empty() { info!( count = network_rows.len(), "processed network lexicons on startup" ); }
// Initialize plugin registry (with DB for persistence) let plugin_registry = Arc::new(happyview::plugin::PluginRegistry::with_db( db_pool.clone(), db_backend, ));
// Initialize WASM runtime let wasm_runtime = Arc::new(happyview::plugin::WasmRuntime::new().expect("Failed to create WASM runtime"));
// Initialize attestation signer (auto-generates key if none exists) let attestation_signer = match happyview::plugin::attestation::load_or_generate( &db_pool, db_backend, &config.public_url, ) .await { Ok(signer) => { tracing::info!("Attestation signing enabled"); Some(Arc::new(signer)) } Err(e) => { tracing::error!(error = %e, "Failed to initialize attestation signer"); None } };
// Load plugins from PLUGIN_URLS env var if let Ok(urls) = std::env::var("PLUGIN_URLS") { for (id, url, sha256) in happyview::plugin::loader::parse_plugin_urls(&urls) { match happyview::plugin::loader::fetch_manifest(&http, &url).await { Ok(preview) => { match happyview::plugin::loader::load_from_manifest( &http, &preview, sha256.as_deref(), ) .await { Ok(plugin) => { tracing::info!(id = %id, "Loaded plugin from URL"); plugin_registry.register(plugin).await; } Err(e) => { tracing::error!(id = %id, error = %e, "Failed to load plugin WASM"); } } } Err(e) => { tracing::error!(id = %id, error = %e, "Failed to fetch plugin manifest"); } } } }
// Load plugins from directory let plugin_dir = std::path::Path::new("./plugins"); if plugin_dir.exists() && let Ok(entries) = std::fs::read_dir(plugin_dir) { for entry in entries.flatten() { let path = entry.path(); if path.is_dir() { match happyview::plugin::loader::load_from_file(&path).await { Ok(plugin) => { tracing::info!(id = %plugin.info.id, "Loaded plugin from file"); plugin_registry.register(plugin).await; } Err(e) => { tracing::error!(path = %path.display(), error = %e, "Failed to load plugin"); } } } } }
// Load plugins from database (added via admin UI) match plugin_registry.load_from_db(&http).await { Ok(count) if count > 0 => { tracing::info!(count = count, "Loaded plugins from database"); } Ok(_) => {} Err(e) => { tracing::error!(error = %e, "Failed to load plugins from database"); } }
// Seed and load per-instance default token costs from instance_settings. let defaults = seed_and_load_rate_limit_defaults(&db_pool, db_backend).await; let rate_limiter = RateLimiter::new(defaults); tokio::spawn(rate_limiter.clone().spawn_cleanup());
// Load per-client rate limit configs and identities from api_clients table. { type ClientRow = ( String, String, String, Option<i32>, Option<f64>, Option<String>, ); let client_rows: Vec<ClientRow> = crate::db::query_as( "SELECT client_key, client_secret_hash, client_uri, rate_limit_capacity, rate_limit_refill_rate, parent_client_id FROM happyview_api_clients WHERE is_active = 1", ) .fetch_all(&db_pool) .await .unwrap_or_default();
for (client_key, secret_hash, client_uri, capacity, refill_rate, _) in &client_rows { rate_limiter.register_client_identity( client_key.clone(), happyview::rate_limit::ClientIdentity { secret_hash: secret_hash.clone(), client_uri: client_uri.clone(), }, ); if let (Some(cap), Some(refill)) = (capacity, refill_rate) { rate_limiter.register_client_config( client_key.clone(), happyview::rate_limit::RateLimitConfig { capacity: *cap as u32, refill_rate: *refill, default_query_cost: defaults.query_cost, default_procedure_cost: defaults.procedure_cost, default_proxy_cost: defaults.proxy_cost, }, ); } else { rate_limiter.register_client_config( client_key.clone(), happyview::rate_limit::RateLimitConfig { capacity: config.default_rate_limit_capacity, refill_rate: config.default_rate_limit_refill_rate, default_query_cost: defaults.query_cost, default_procedure_cost: defaults.procedure_cost, default_proxy_cost: defaults.proxy_cost, }, ); } } }
// Seed and load domain cache let domain_cache = happyview::domain::DomainCache::new(); { let count_sql = happyview::db::adapt_sql("SELECT COUNT(*) FROM happyview_domains", db_backend); let row = crate::db::query(&count_sql) .fetch_one(&db_pool) .await .expect("Failed to count domains"); let count: i64 = row.try_get(0).unwrap_or(0);
if count == 0 { let id = uuid::Uuid::new_v4().to_string(); let now = happyview::db::now_rfc3339(); let insert_sql = happyview::db::adapt_sql( "INSERT INTO happyview_domains (id, url, is_primary, created_at, updated_at) VALUES (?, ?, 1, ?, ?)", db_backend, ); crate::db::query(&insert_sql) .bind(&id) .bind(&config.public_url) .bind(&now) .bind(&now) .execute(&db_pool) .await .expect("Failed to insert primary domain"); info!("Seeded primary domain: {}", config.public_url); } else { // Sync the primary domain URL with PUBLIC_URL if it changed let primary_sql = happyview::db::adapt_sql( "SELECT id, url FROM happyview_domains WHERE is_primary = 1", db_backend, ); if let Some(row) = crate::db::query(&primary_sql) .fetch_optional(&db_pool) .await .expect("Failed to check primary domain") { let primary_url: String = row.try_get("url").unwrap_or_default(); if primary_url != config.public_url { let primary_id: String = row.try_get("id").unwrap_or_default(); let now = happyview::db::now_rfc3339(); let update_sql = happyview::db::adapt_sql( "UPDATE happyview_domains SET url = ?, updated_at = ? WHERE id = ?", db_backend, ); crate::db::query(&update_sql) .bind(&config.public_url) .bind(&now) .bind(&primary_id) .execute(&db_pool) .await .expect("Failed to update primary domain URL"); info!( "Updated primary domain URL from {} to {}", primary_url, config.public_url ); } } }
let select_sql = happyview::db::adapt_sql( "SELECT id, url, is_primary, created_at, updated_at FROM happyview_domains", db_backend, ); let rows = crate::db::query(&select_sql) .fetch_all(&db_pool) .await .expect("Failed to load domains");
let domains: Vec<happyview::domain::Domain> = rows .into_iter() .map(|row| { let is_primary_int: i32 = row.try_get("is_primary").unwrap_or(0); happyview::domain::Domain { id: row.try_get("id").unwrap_or_default(), url: row.try_get("url").unwrap_or_default(), is_primary: is_primary_int != 0, created_at: row.try_get("created_at").unwrap_or_default(), updated_at: row.try_get("updated_at").unwrap_or_default(), } }) .collect();
let domain_count = domains.len(); domain_cache.load(domains).await; info!("Loaded {} domain(s) into cache", domain_count); }
// Build atrium-oauth client let dns = NativeDnsResolver::new(); let callback_url = format!( "{}/auth/callback", config.effective_public_url().trim_end_matches('/') ); let atrium_http = Arc::new(DefaultHttpClient::default());
let did_resolver = CommonDidResolver::new(CommonDidResolverConfig { plc_directory_url: config.plc_url.clone(), http_client: Arc::clone(&atrium_http), });
let handle_resolver = AtprotoHandleResolver::new(AtprotoHandleResolverConfig { dns_txt_resolver: dns, http_client: Arc::clone(&atrium_http), });
let is_loopback = config.public_url.contains("127.0.0.1") || config.public_url.contains("[::1]") || config.public_url.contains("localhost");
let resolver_config = OAuthResolverConfig { did_resolver, handle_resolver, authorization_server_metadata: Default::default(), protected_resource_metadata: Default::default(), };
let oauth_state_store = DbStateStore::new(db_pool.clone(), db_backend);
let oauth_scopes = vec![ Scope::Known(KnownScope::Atproto), Scope::Unknown("identity:*".to_string()), ];
let oauth_client = if is_loopback { info!("Using loopback OAuth client metadata (local development)"); atrium_oauth::OAuthClient::new(OAuthClientConfig { client_metadata: AtprotoLocalhostClientMetadata { redirect_uris: Some(vec![callback_url]), scopes: Some(oauth_scopes), }, keys: None, state_store: oauth_state_store.clone(), session_store: DbSessionStore::new(db_pool.clone(), db_backend), resolver: resolver_config, }) .expect("Failed to create OAuth client") } else { atrium_oauth::OAuthClient::new(OAuthClientConfig { client_metadata: AtprotoClientMetadata { client_id: format!( "{}/oauth-client-metadata.json", config.effective_public_url().trim_end_matches('/') ), client_uri: Some(config.effective_public_url()), redirect_uris: vec![callback_url], token_endpoint_auth_method: AuthMethod::None, grant_types: vec![GrantType::AuthorizationCode, GrantType::RefreshToken], scopes: oauth_scopes, jwks_uri: None, token_endpoint_auth_signing_alg: None, }, keys: None, state_store: oauth_state_store.clone(), session_store: DbSessionStore::new(db_pool.clone(), db_backend), resolver: resolver_config, }) .expect("Failed to create OAuth client") };
// Derive the cookie signing key from SESSION_SECRET when it is secure. When // it is not, log the problem loudly and fall back to an ephemeral random key // so no attacker can forge cookies with a known/weak key. Cookie-based auth // is disabled in this state (see the auth extractors and login handlers); // DPoP, service auth, and API-key auth are unaffected. The server still boots // so the dashboard can surface the misconfiguration to an operator. let cookie_key = if config.session_secret_secure() { axum_extra::extract::cookie::Key::derive_from(config.session_secret.as_bytes()) } else { for err in config.config_errors() { tracing::error!("INSECURE CONFIGURATION: {err}"); } tracing::error!( "Cookie-based login is DISABLED until SESSION_SECRET is set securely. \ Other auth (DPoP, service auth, API keys) continues to work." ); axum_extra::extract::cookie::Key::generate() };
let initial_collections = lexicons.get_record_collections().await; let (collections_tx, collections_rx) = watch::channel(initial_collections); let (labeler_subscriptions_tx, labeler_subscriptions_rx) = watch::channel(());
// Build the OAuth client registry and load API clients from DB let oauth_client_arc = Arc::new(oauth_client); let oauth_registry = Arc::new(happyview::auth::OAuthClientRegistry::new(Arc::clone( &oauth_client_arc, ))); oauth_registry .load_from_db( &db_pool, db_backend, &config.plc_url, oauth_state_store.clone(), db_pool.clone(), ) .await;
// Register the primary domain's OAuth client in domain_clients if let Some(ref pd) = domain_cache.primary().await { let primary_client_id_url = format!( "{}/oauth-client-metadata.json", config.url_with_base_path(&pd.url).trim_end_matches('/') ); oauth_registry.register_domain_client( pd.url.clone(), primary_client_id_url, Arc::clone(&oauth_client_arc), ); }
// Build OAuth clients for all non-primary domains let all_domains = domain_cache.all().await; for domain in &all_domains { if domain.is_primary { continue; // Already registered above }
let domain_base_url = config.url_with_base_path(&domain.url); let domain_callback_url = format!("{}/auth/callback", domain_base_url.trim_end_matches('/')); let domain_client_id = format!( "{}/oauth-client-metadata.json", domain_base_url.trim_end_matches('/') );
let domain_http = Arc::new(DefaultHttpClient::default()); let domain_resolver = OAuthResolverConfig { did_resolver: CommonDidResolver::new(CommonDidResolverConfig { plc_directory_url: config.plc_url.clone(), http_client: Arc::clone(&domain_http), }), handle_resolver: AtprotoHandleResolver::new(AtprotoHandleResolverConfig { dns_txt_resolver: NativeDnsResolver::new(), http_client: Arc::clone(&domain_http), }), authorization_server_metadata: Default::default(), protected_resource_metadata: Default::default(), };
match atrium_oauth::OAuthClient::new(OAuthClientConfig { client_metadata: AtprotoClientMetadata { client_id: domain_client_id.clone(), client_uri: Some(domain_base_url.clone()), redirect_uris: vec![domain_callback_url], token_endpoint_auth_method: AuthMethod::None, grant_types: vec![GrantType::AuthorizationCode, GrantType::RefreshToken], scopes: vec![Scope::Known(KnownScope::Atproto)], jwks_uri: None, token_endpoint_auth_signing_alg: None, }, keys: None, state_store: oauth_state_store.clone(), session_store: DbSessionStore::new(db_pool.clone(), db_backend), resolver: domain_resolver, }) { Ok(client) => { info!(domain = %domain.url, "Registered domain OAuth client"); oauth_registry.register_domain_client( domain.url.clone(), domain_client_id, Arc::new(client), ); } Err(e) => { tracing::error!(domain = %domain.url, error = %e, "Failed to create domain OAuth client"); } } }
let official_registry: happyview::plugin::official_registry::SharedRegistry = std::sync::Arc::new(tokio::sync::RwLock::new( happyview::plugin::official_registry::OfficialRegistryState::default(), )); let official_registry_config = happyview::plugin::official_registry::RegistryConfig::production(); happyview::plugin::official_registry::spawn_refresh_task( http.clone(), official_registry_config.clone(), official_registry.clone(), );
let proxy_config = { let json_str = happyview::admin::settings::get_setting(&db_pool, "xrpc_proxy_config", db_backend) .await; let config = json_str .and_then(|s| serde_json::from_str::<happyview::proxy_config::ProxyConfig>(&s).ok()) .unwrap_or_default(); info!(mode = ?config.mode, nsid_count = config.nsids.len(), "Loaded XRPC proxy config"); std::sync::Arc::new(arc_swap::ArcSwap::new(std::sync::Arc::new(config))) };
let (backfill_events_tx, _) = tokio::sync::broadcast::channel(16384);
let verbose_event_logging = { let enabled = happyview::admin::settings::get_setting(&db_pool, "verbose_event_logging", db_backend) .await .map(|v| v.eq_ignore_ascii_case("true")) .unwrap_or(false); std::sync::Arc::new(std::sync::atomic::AtomicBool::new(enabled)) };
let state = AppState { config: config.clone(), http, db: db_pool, backfill_db: backfill_db_pool, db_backend, domain_cache: domain_cache.clone(), lexicons, collections_tx, labeler_subscriptions_tx, rate_limiter, oauth: oauth_registry, oauth_state_store, cookie_key, plugin_registry, wasm_runtime, attestation_signer, official_registry, official_registry_config, proxy_config, backfill_events_tx, verbose_event_logging, };
jetstream::spawn(state.clone(), collections_rx);
labeler::spawn(state.clone(), labeler_subscriptions_rx); tokio::spawn(labeler::spawn_label_gc(state.db.clone(), state.db_backend));
tokio::spawn(happyview::event_log::spawn_retention_cleanup( state.db.clone(), state.config.event_log_retention_days, state.db_backend, ));
{ let db = state.db.clone(); let flag = state.verbose_event_logging.clone(); let backend = state.db_backend; tokio::spawn(async move { loop { tokio::time::sleep(std::time::Duration::from_secs(30)).await; let enabled = happyview::admin::settings::get_setting(&db, "verbose_event_logging", backend) .await .map(|v| v.eq_ignore_ascii_case("true")) .unwrap_or(false); flag.store(enabled, std::sync::atomic::Ordering::Relaxed); } }); }
happyview::admin::backfill::resume_backfill_jobs(&state).await;
// Resume interrupted jobs and start the job worker happyview::jobs::worker::resume_interrupted_jobs(&state).await; { let job_state = state.clone(); tokio::spawn(async move { happyview::jobs::worker::run_worker(job_state).await; }); }
{ let state = state.clone(); tokio::spawn(async move { happyview::admin::backfill::run_backfill_retention_cleanup(&state).await; }); }
let app = server::router(state); let addr = config.listen_addr();
info!(%addr, "HappyView is listening");
let listener = tokio::net::TcpListener::bind(addr) .await .expect("failed to bind");
axum::serve(listener, app).await.expect("server error");}
async fn seed_and_load_rate_limit_defaults( pool: &sqlx::AnyPool, backend: happyview::db::DatabaseBackend,) -> RateLimitDefaults { use happyview::rate_limit::{ SEED_DEFAULT_PROCEDURE_COST, SEED_DEFAULT_PROXY_COST, SEED_DEFAULT_QUERY_COST, SETTING_DEFAULT_PROCEDURE_COST, SETTING_DEFAULT_PROXY_COST, SETTING_DEFAULT_QUERY_COST, };
async fn seed_and_read( pool: &sqlx::AnyPool, backend: happyview::db::DatabaseBackend, key: &str, seed: u32, ) -> u32 { if happyview::admin::settings::get_setting(pool, key, backend) .await .is_none() { let now = happyview::db::now_rfc3339(); let sql = happyview::db::adapt_sql( "INSERT INTO happyview_instance_settings (key, value, updated_at) VALUES (?, ?, ?) ON CONFLICT (key) DO NOTHING", backend, ); if let Err(e) = crate::db::query(&sql) .bind(key) .bind(seed.to_string()) .bind(&now) .execute(pool) .await { warn!(error = %e, key = key, "failed to seed rate-limit default"); } } happyview::admin::settings::get_setting(pool, key, backend) .await .and_then(|s| s.parse::<u32>().ok()) .unwrap_or(seed) }
RateLimitDefaults { query_cost: seed_and_read( pool, backend, SETTING_DEFAULT_QUERY_COST, SEED_DEFAULT_QUERY_COST, ) .await, procedure_cost: seed_and_read( pool, backend, SETTING_DEFAULT_PROCEDURE_COST, SEED_DEFAULT_PROCEDURE_COST, ) .await, proxy_cost: seed_and_read( pool, backend, SETTING_DEFAULT_PROXY_COST, SEED_DEFAULT_PROXY_COST, ) .await, }}