Something went wrong. Try again.
A lexicon-driven AppView for ATProto.
Something went wrong. Try again.
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389use 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::RateLimiter;use happyview::resolve::{fetch_lexicon_from_pds, resolve_nsid_authority};use happyview::{AppState, labeler, server, tap};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();
tracing_subscriber::fmt() .with_env_filter( tracing_subscriber::EnvFilter::try_from_default_env() .unwrap_or_else(|_| "happyview=debug,tower_http=debug".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;
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,) = sqlx::query_as("SELECT COUNT(*) FROM 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,) = sqlx::query_as("SELECT COUNT(*) FROM 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 records ORDER BY uri LIMIT ? OFFSET ?", backend, );
loop { let batch: Vec<(String, String, String)> = sqlx::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>)> = sqlx::query_as( "SELECT id, authority_did, target_collection FROM 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, None, None, ) { Ok(parsed) => { let now = db::now_rfc3339(); let update_sql = db::adapt_sql( "UPDATE 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) = sqlx::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 (optional) let attestation_signer = match happyview::plugin::attestation::load_from_env() { Ok(Some(signer)) => { tracing::info!("Attestation signing enabled"); Some(Arc::new(signer)) } Ok(None) => { tracing::info!("Attestation signing disabled (no ATTESTATION_PRIVATE_KEY)"); None } Err(e) => { tracing::error!(error = %e, "Failed to load 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::load_from_url(&http, &url, 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"); } } } }
// 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"); } } } } }
// Initialize rate limiter from DB. let rl_state = RateLimiter::load_from_db(&db_pool).await; let rate_limiter = RateLimiter::new(rl_state.enabled, rl_state.global, rl_state.allowlist); tokio::spawn(rate_limiter.clone().spawn_cleanup());
// Build atrium-oauth client let dns = NativeDnsResolver::new(); let callback_url = format!("{}/auth/callback", config.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]");
let resolver_config = OAuthResolverConfig { did_resolver, handle_resolver, authorization_server_metadata: Default::default(), protected_resource_metadata: Default::default(), };
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(vec![ Scope::Known(KnownScope::Atproto), Scope::Known(KnownScope::TransitionGeneric), ]), }, keys: None, state_store: DbStateStore::new(db_pool.clone(), db_backend), 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.public_url.trim_end_matches('/') ), client_uri: Some(config.public_url.clone()), redirect_uris: vec![callback_url], token_endpoint_auth_method: AuthMethod::None, grant_types: vec![GrantType::AuthorizationCode, GrantType::RefreshToken], scopes: vec![ Scope::Known(KnownScope::Atproto), Scope::Known(KnownScope::TransitionGeneric), ], jwks_uri: None, token_endpoint_auth_signing_alg: None, }, keys: None, state_store: DbStateStore::new(db_pool.clone(), db_backend), session_store: DbSessionStore::new(db_pool.clone(), db_backend), resolver: resolver_config, }) .expect("Failed to create OAuth client") };
if config.session_secret == "change-me-in-production-not-secure" { if db_backend == happyview::db::DatabaseBackend::Postgres { tracing::error!( "INSECURE SESSION SECRET — You are using the default session secret with a \ Postgres backend, which likely indicates a production deployment. \ Set SESSION_SECRET to a random string of at least 64 characters." ); } else { warn!( "Using the default session secret. Set SESSION_SECRET to a random \ string in production." ); } }
let cookie_key = axum_extra::extract::cookie::Key::derive_from(config.session_secret.as_bytes());
let initial_collections = lexicons.get_record_collections().await; let initial_collections_for_sync = initial_collections.clone(); let (collections_tx, collections_rx) = watch::channel(initial_collections); let (labeler_subscriptions_tx, labeler_subscriptions_rx) = watch::channel(());
let state = AppState { config: config.clone(), http, db: db_pool, db_backend, lexicons, collections_tx, labeler_subscriptions_tx, rate_limiter, oauth: Arc::new(oauth_client), cookie_key, plugin_registry, wasm_runtime, attestation_signer, };
// Sync initial collections to Tap on startup. { let mut wanted = initial_collections_for_sync; if !wanted.contains(&"com.atproto.lexicon.schema".to_string()) { wanted.push("com.atproto.lexicon.schema".to_string()); } if let Err(e) = tap::sync_collections( &state.http, &config.tap_url, config.tap_admin_password.as_deref(), &wanted, ) .await { warn!("failed to sync initial collections to tap: {e}"); } }
tap::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 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");}