A lexicon-driven AppView for ATProto.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026use 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, 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" );
match happyview::spaces::cid_backfill::run_if_needed(&db_pool, db_backend).await { Ok(Some(report)) if report.is_noop() => { info!("space CID backfill: nothing to repair"); } Ok(Some(report)) => { info!( records_updated = report.records_updated, oplog_rows_remapped = report.oplog_rows_remapped, repos_rebuilt = report.repos_rebuilt, records_unencodable = report.records_unencodable, "space CID backfill applied" ); if report.records_unencodable > 0 { tracing::warn!( count = report.records_unencodable, "some space records could not be encoded as DAG-CBOR and kept their existing CID" ); } } Ok(None) => { /* already completed; skipped without scanning */ } Err(e) => tracing::error!( error = %e, "space CID backfill failed; will retry on next startup" ), }
match happyview::maintenance::vacuum::run_if_requested( &db_pool, db_backend, &config.database_url, ) .await { Ok(Some(result)) if result.status == "ok" => { info!( reclaimed = %happyview::maintenance::vacuum::human_bytes(result.reclaimed_bytes), "scheduled vacuum complete" ); } Ok(Some(result)) => { tracing::error!( error = %result.error.clone().unwrap_or_default(), "scheduled vacuum failed" ); } Ok(None) => { /* not scheduled */ } Err(e) => tracing::error!(error = %e, "scheduled vacuum could not be evaluated"), }
happyview::telemetry::collect::health::note_restart(&db_pool, db_backend).await; happyview::maintenance::nsid_audit::run(&db_pool, db_backend).await; happyview::maintenance::lexicon_ids::run(&db_pool, db_backend).await;
// 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 = happyview::http_retry::init_shared_client(&config.user_agent); 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('/') ); // Use our UA-and-timeout-configured client for OAuth traffic. atrium-oauth's // `default-client` feature (which we no longer enable — see Cargo.toml) // used to build its own unconfigured `reqwest::Client::new()` here, with no // seam for headers or timeouts; this identifies OAuth requests and gives // them the shared connect/read timeouts instead. let atrium_http = Arc::new(happyview::http_retry::HappyViewHttpClient::new( http.clone(), ));
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 = happyview::auth::client_registry::is_loopback_url(&config.public_url);
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 linked_repos_scopes = oauth_scopes.clone(); let oauth_scopes_for_by_kid = oauth_scopes.clone();
let instance_key = happyview::oauth::client_keys::ensure_instance_key( &db_pool, db_backend, config.token_encryption_key.as_ref(), ) .await .expect("Failed to load OAuth client authentication key");
let client_jwks = vec![ happyview::oauth::client_keys::to_atrium_jwk(&instance_key) .expect("Failed to convert client key to JWK"), ]; info!(kid = %instance_key.kid, "confidential OAuth client key ready");
let instance_client_id_url = config.instance_client_id_url();
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.clone()]), scopes: Some(oauth_scopes), }, keys: None, state_store: oauth_state_store.clone(), session_store: DbSessionStore::new(db_pool.clone(), db_backend), resolver: resolver_config, http_client: happyview::http_retry::HappyViewHttpClient::new(http.clone()), }) .expect("Failed to create OAuth client") } else { atrium_oauth::OAuthClient::new(OAuthClientConfig { client_metadata: AtprotoClientMetadata { client_id: instance_client_id_url.clone(), client_uri: Some(config.effective_public_url()), redirect_uris: vec![callback_url.clone()], token_endpoint_auth_method: AuthMethod::PrivateKeyJwt, grant_types: vec![GrantType::AuthorizationCode, GrantType::RefreshToken], scopes: oauth_scopes, jwks_uri: None, token_endpoint_auth_signing_alg: Some("ES256".to_string()), }, keys: Some(client_jwks.clone()), state_store: oauth_state_store.clone(), session_store: DbSessionStore::new(db_pool.clone(), db_backend) .with_signing_kid(Some(instance_key.kid.clone())), resolver: resolver_config, http_client: happyview::http_retry::HappyViewHttpClient::new(http.clone()), }) .expect("Failed to create OAuth client") };
// A loopback client never signs at all, so it gets no pin regardless — // same reasoning as `build_instance_client`'s and `linked_repos::client:: // build`'s own `signing_kid` computation, which this must keep matching. // Fixed here, once, alongside `linked_repos_client` itself: that client // is never rebuilt live on rotation (see `oauth::rotation`'s module // doc), so this is the one place its actual kid is ever set. let instance_signing_kid = if is_loopback { None } else { Some(instance_key.kid.clone()) };
let linked_repos_client = Arc::new( happyview::linked_repos::client::build( &config.plc_url, &format!( "{}/oauth-client-metadata.json", config.effective_public_url().trim_end_matches('/') ), &config.effective_public_url(), callback_url.clone(), is_loopback, linked_repos_scopes, oauth_state_store.clone(), db_pool.clone(), db_backend, Some(client_jwks.clone()), ) .expect("Failed to create linked-repo OAuth client"), );
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_with_kid( Arc::clone(&oauth_client_arc), instance_signing_kid.clone(), )); 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), instance_signing_kid.clone(), ); }
// 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('/') );
// Same reasoning as `atrium_http` above. let domain_http = Arc::new(happyview::http_retry::HappyViewHttpClient::new( http.clone(), )); 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::PrivateKeyJwt, grant_types: vec![GrantType::AuthorizationCode, GrantType::RefreshToken], scopes: vec![Scope::Known(KnownScope::Atproto)], jwks_uri: None, token_endpoint_auth_signing_alg: Some("ES256".to_string()), }, keys: Some(client_jwks.clone()), state_store: oauth_state_store.clone(), session_store: DbSessionStore::new(db_pool.clone(), db_backend) .with_signing_kid(Some(instance_key.kid.clone())), resolver: domain_resolver, http_client: happyview::http_retry::HappyViewHttpClient::new(http.clone()), }) { Ok(client) => { info!(domain = %domain.url, "Registered domain OAuth client"); oauth_registry.register_domain_client( domain.url.clone(), domain_client_id, Arc::new(client), Some(instance_key.kid.clone()), ); } Err(e) => { tracing::error!(domain = %domain.url, error = %e, "Failed to create domain OAuth client"); } } }
// Index every non-revoked instance key by (client_id_url, kid) so a // session pinned to a specific key can be resolved unambiguously — see // `OAuthClientRegistry::register_for_kid`. Each key gets its own, // genuinely single-key `OAuthClient` built fresh here: reusing // `oauth_client_arc` (built from only the *current* key) across every // kid is correct only while exactly one key is ever live. The moment a // `retiring` key coexists with `current` — which rotation makes routine // — that reuse would map the retiring kid to a client that can only // sign with the wrong (current) key, silently, since the lookup itself // succeeds. let instance_keys = happyview::oauth::client_keys::load_keys( &db_pool, db_backend, config.token_encryption_key.as_ref(), happyview::oauth::client_keys::INSTANCE_OWNER, ) .await .unwrap_or_default(); for key in &instance_keys { match happyview::auth::client_registry::build_instance_client( &config.plc_url, &instance_client_id_url, &config.effective_public_url(), vec![callback_url.clone()], is_loopback, oauth_scopes_for_by_kid.clone(), oauth_state_store.clone(), db_pool.clone(), db_backend, key, ) { Ok(client) => { oauth_registry.register_for_kid( &instance_client_id_url, &key.kid, Arc::new(client), ); } Err(e) => { warn!(kid = %key.kid, error = %e, "failed to build per-kid instance OAuth client"); } }
// Same reasoning, nested: each domain's `by_kid` entry must also be // a distinct single-key client, not a reused Arc built for a // different kid. for domain in &all_domains { 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('/') ); match happyview::auth::client_registry::build_instance_client( &config.plc_url, &domain_client_id, &domain_base_url, vec![domain_callback_url], false, vec![Scope::Known(KnownScope::Atproto)], oauth_state_store.clone(), db_pool.clone(), db_backend, key, ) { Ok(client) => { oauth_registry.register_for_kid(&domain_client_id, &key.kid, Arc::new(client)); } Err(e) => { warn!(domain = %domain.url, kid = %key.kid, error = %e, "failed to build per-kid domain OAuth client"); } } } }
// Index the linked-repos client by kid too, in its own namespace kept // separate from the loop above — see `OAuthClientRegistry`'s // `linked_repos_by_kid` field doc for why it cannot share a // `(client_id_url, kid)` entry with the instance/domain clients despite // publishing the same `client_id` text. Same distinct-client-per-kid // reasoning applies here too. for key in &instance_keys { match happyview::oauth::client_keys::to_atrium_jwk(key) { Ok(jwk) => match happyview::linked_repos::client::build( &config.plc_url, &format!( "{}/oauth-client-metadata.json", config.effective_public_url().trim_end_matches('/') ), &config.effective_public_url(), callback_url.clone(), is_loopback, oauth_scopes_for_by_kid.clone(), oauth_state_store.clone(), db_pool.clone(), db_backend, Some(vec![jwk]), ) { Ok(client) => { oauth_registry.register_linked_repos_for_kid(&key.kid, Arc::new(client)); } Err(e) => { warn!(kid = %key.kid, error = %e, "failed to build per-kid linked-repos OAuth client"); } }, Err(e) => { warn!(kid = %key.kid, error = %e, "failed to convert instance key to JWK for linked-repos registration"); } } }
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 telemetry_counters = std::sync::Arc::new(happyview::telemetry::counters::Counters::new());
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, linked_repos_client, linked_repos_client_kid: instance_signing_kid, cookie_key, plugin_registry, wasm_runtime, attestation_signer, official_registry, official_registry_config, proxy_config, backfill_events_tx, verbose_event_logging, client_jwks, telemetry_counters, };
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));
{ let gc_db = state.db.clone(); let gc_backend = state.db_backend; tokio::spawn(async move { happyview::auth::state_gc::run_expired_state_gc(gc_db, gc_backend).await; }); }
{ let telemetry_state = state.clone(); tokio::spawn(async move { happyview::telemetry::reporter::run_reporter(telemetry_state).await; }); }
tokio::spawn(happyview::event_log::spawn_retention_cleanup( state.db.clone(), 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 keepalive_state = state.clone(); tokio::spawn(async move { happyview::linked_repos::worker::run_keepalive(keepalive_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, }}