diff --git a/Cargo.lock b/Cargo.lock --- a/Cargo.lock +++ b/Cargo.lock @@ -43,6 +43,17 @@ "serde_json", ] [[package]] +name = "async-trait" +version = "0.1.89" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] name = "atoi" version = "2.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" @@ -255,6 +266,30 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "19d374276b40fb8bbdee95aef7c7fa6b5316ec764510eb64b8dd0e2ed0d7e7f5" [[package]] +name = "critical-section" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "790eea4361631c5e7d22598ecd5723ff611904e3344ce8720784c93e3d83d40b" + +[[package]] +name = "crossbeam-channel" +version = "0.5.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "82b8f8f868b36967f9606790d1903570de9ceaf870a7bf9fbbd3016d636a2cb2" +dependencies = [ + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-epoch" +version = "0.9.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e" +dependencies = [ + "crossbeam-utils", +] + +[[package]] name = "crossbeam-queue" version = "0.3.12" source = "registry+https://github.com/rust-lang/crates.io-index" @@ -417,6 +452,18 @@ "cfg-if", ] [[package]] +name = "enum-as-inner" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a1e6a265c649f3f5979b601d26f1d05ada116434c87741c9493cb56218f76cbc" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "syn", +] + +[[package]] name = "equivalent" version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" @@ -712,6 +759,7 @@ "bytes", "chrono", "dotenvy", "futures-util", + "hickory-resolver", "http-body-util", "jsonwebtoken", "p256", @@ -777,6 +825,52 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f24254aa9a54b5c858eaee2f5bccdb46aaf0e486a595ed5fd8f86ba55232a70" [[package]] +name = "hickory-proto" +version = "0.25.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8a6fe56c0038198998a6f217ca4e7ef3a5e51f46163bd6dd60b5c71ca6c6502" +dependencies = [ + "async-trait", + "cfg-if", + "data-encoding", + "enum-as-inner", + "futures-channel", + "futures-io", + "futures-util", + "idna", + "ipnet", + "once_cell", + "rand 0.9.2", + "ring", + "thiserror", + "tinyvec", + "tokio", + "tracing", + "url", +] + +[[package]] +name = "hickory-resolver" +version = "0.25.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc62a9a99b0bfb44d2ab95a7208ac952d31060efc16241c87eaf36406fecf87a" +dependencies = [ + "cfg-if", + "futures-util", + "hickory-proto", + "ipconfig", + "moka", + "once_cell", + "parking_lot", + "rand 0.9.2", + "resolv-conf", + "smallvec", + "thiserror", + "tokio", + "tracing", +] + +[[package]] name = "hkdf" version = "0.12.4" source = "registry+https://github.com/rust-lang/crates.io-index" @@ -920,7 +1014,7 @@ "ipnet", "libc", "percent-encoding", "pin-project-lite", - "socket2", + "socket2 0.6.2", "system-configuration", "tokio", "tower-service", @@ -1073,6 +1167,18 @@ "serde_core", ] [[package]] +name = "ipconfig" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b58db92f96b720de98181bbbe63c831e87005ab460c1bf306eb2622b4707997f" +dependencies = [ + "socket2 0.5.10", + "widestring", + "windows-sys 0.48.0", + "winreg", +] + +[[package]] name = "ipnet" version = "2.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" @@ -1243,6 +1349,23 @@ "windows-sys 0.61.2", ] [[package]] +name = "moka" +version = "0.12.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b4ac832c50ced444ef6be0767a008b02c106a909ba79d1d830501e94b96f6b7e" +dependencies = [ + "crossbeam-channel", + "crossbeam-epoch", + "crossbeam-utils", + "equivalent", + "parking_lot", + "portable-atomic", + "smallvec", + "tagptr", + "uuid", +] + +[[package]] name = "native-tls" version = "0.2.14" source = "registry+https://github.com/rust-lang/crates.io-index" @@ -1345,6 +1468,10 @@ name = "once_cell" version = "1.21.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "42f5e15c9953c5e4ccceeb2e7382a716482c34515315f7b03532b8b4e8393d2d" +dependencies = [ + "critical-section", + "portable-atomic", +] [[package]] name = "openssl" @@ -1496,6 +1623,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c" [[package]] +name = "portable-atomic" +version = "1.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c33a9471896f1c69cecef8d20cbe2f7accd12527ce60845ff44c153bb2a21b49" + +[[package]] name = "potential_utf" version = "0.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" @@ -1707,6 +1840,12 @@ "wasm-bindgen", "wasm-bindgen-futures", "web-sys", ] + +[[package]] +name = "resolv-conf" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e061d1b48cb8d38042de4ae0a7a6401009d6143dc80d2e2d6f31f0bdd6470c7" [[package]] name = "rfc6979" @@ -2062,6 +2201,16 @@ ] [[package]] name = "socket2" +version = "0.5.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e22376abed350d73dd1cd119b57ffccad95b4e585a7cda43e286245ce23c0678" +dependencies = [ + "libc", + "windows-sys 0.52.0", +] + +[[package]] +name = "socket2" version = "0.6.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "86f4aa3ad99f2088c990dfa82d367e19cb29268ed67c574d10d0a4bfe71f07e0" @@ -2359,6 +2508,12 @@ "libc", ] [[package]] +name = "tagptr" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417" + +[[package]] name = "tempfile" version = "3.25.0" source = "registry+https://github.com/rust-lang/crates.io-index" @@ -2468,7 +2623,7 @@ "mio", "parking_lot", "pin-project-lite", "signal-hook-registry", - "socket2", + "socket2 0.6.2", "tokio-macros", "windows-sys 0.61.2", ] @@ -2966,6 +3121,12 @@ "wasite", ] [[package]] +name = "widestring" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72069c3113ab32ab29e5584db3c6ec55d416895e60715417b5b883a357c3e471" + +[[package]] name = "windows-core" version = "0.62.2" source = "registry+https://github.com/rust-lang/crates.io-index" @@ -3256,6 +3417,16 @@ name = "windows_x86_64_msvc" version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650" + +[[package]] +name = "winreg" +version = "0.50.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "524e57b2c537c0f9b1e69f1965311ec12182b4122e45035b1508cd24d2adadb1" +dependencies = [ + "cfg-if", + "windows-sys 0.48.0", +] [[package]] name = "wiremock" diff --git a/Cargo.toml b/Cargo.toml --- a/Cargo.toml +++ b/Cargo.toml @@ -21,6 +21,7 @@ sqlx = { version = "0.8", features = ["runtime-tokio", "tls-rustls", "postgres", "json", "chrono", "migrate"] } tokio = { version = "1", features = ["full"] } tokio-tungstenite = { version = "0.26", features = ["rustls-tls-webpki-roots"] } tower-http = { version = "0.6", features = ["cors", "trace"] } +hickory-resolver = "0.25" tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] } diff --git a/docs/README.md b/docs/README.md --- a/docs/README.md +++ b/docs/README.md @@ -24,6 +24,7 @@ - [Quickstart](quickstart.md) - get a local instance running - [Configuration](configuration.md) - environment variables reference - [Deployment](deployment.md) - Docker, production, TLS - [Lexicons](lexicons.md) - uploading and managing lexicon definitions +- [Network Lexicons](network-lexicons.md) - loading lexicons from the ATProto network - [Backfill](backfill.md) - bulk-indexing historical records - [XRPC API](xrpc-api.md) - query and procedure endpoints - [Admin API](admin-api.md) - manage lexicons, backfills, and admins diff --git a/docs/admin-api.md b/docs/admin-api.md --- a/docs/admin-api.md +++ b/docs/admin-api.md @@ -102,6 +102,84 @@ **Response**: `204 No Content` --- +## Network Lexicons + +Network lexicons are fetched from the ATProto network via DNS TXT resolution and kept updated via Jetstream. See [Network Lexicons](network-lexicons.md) for background. + +### Add a network lexicon + +``` +POST /admin/network-lexicons +``` + +```sh +curl -X POST http://localhost:3000/admin/network-lexicons \ + -H "$AUTH" \ + -H "Content-Type: application/json" \ + -d '{ + "nsid": "games.gamesgamesgamesgames.game", + "target_collection": null + }' +``` + +| Field | Type | Required | Description | +| ------------------- | ------ | -------- | ------------------------------------------------------------------- | +| `nsid` | string | yes | The NSID of the lexicon to watch | +| `target_collection` | string | no | For query/procedure lexicons, the record collection they operate on | + +HappyView resolves the NSID authority via DNS TXT, fetches the lexicon from the authority's PDS, parses it, and stores it. + +**Response**: `201 Created` + +```json +{ + "nsid": "games.gamesgamesgamesgames.game", + "authority_did": "did:plc:authority", + "revision": 1 +} +``` + +### List network lexicons + +``` +GET /admin/network-lexicons +``` + +```sh +curl http://localhost:3000/admin/network-lexicons -H "$AUTH" +``` + +**Response**: `200 OK` + +```json +[ + { + "nsid": "games.gamesgamesgamesgames.game", + "authority_did": "did:plc:authority", + "target_collection": null, + "last_fetched_at": "2025-01-01T00:00:00Z", + "created_at": "2025-01-01T00:00:00Z" + } +] +``` + +### Remove a network lexicon + +``` +DELETE /admin/network-lexicons/{nsid} +``` + +```sh +curl -X DELETE http://localhost:3000/admin/network-lexicons/games.gamesgamesgamesgames.game \ + -H "$AUTH" +``` + +Removes the network lexicon tracking and also deletes the lexicon from the `lexicons` table and in-memory registry. + +**Response**: `204 No Content` + +--- + ## Stats ### Record counts diff --git a/docs/network-lexicons.md b/docs/network-lexicons.md new file mode 100644 --- /dev/null +++ b/docs/network-lexicons.md @@ -0,0 +1,57 @@ +# Network Lexicons + +Network lexicons are lexicon definitions that HappyView fetches directly from the ATProto network rather than being uploaded manually via the admin API. An admin specifies an NSID, HappyView resolves the authority's repo, fetches the lexicon record, and keeps it updated via Jetstream. + +## How it works + +### NSID authority resolution + +Lexicon records live in repos as `com.atproto.lexicon.schema` with the rkey set to the NSID. To find which repo holds a lexicon, HappyView resolves the NSID's authority: + +1. Extract the authority from the NSID (all segments except the last). For example, `games.gamesgamesgamesgames.game` has authority `games.gamesgamesgamesgames`. +2. Reverse the authority segments to form a domain: `gamesgamesgamesgames.games`. +3. Look up the DNS TXT record at `_lexicon.{domain}` (e.g. `_lexicon.gamesgamesgamesgames.games`). +4. Parse the TXT record for a `did=` value. +5. Resolve the DID to a PDS endpoint via the PLC directory. + +Resolution is **non-hierarchical** --- each authority requires its own explicit TXT record. + +### Fetching + +Once the authority DID and PDS endpoint are known, HappyView calls `com.atproto.repo.getRecord` with: +- `repo` = the authority DID +- `collection` = `com.atproto.lexicon.schema` +- `rkey` = the NSID + +The `value` field of the response is the raw lexicon JSON. + +### Live updates via Jetstream + +Jetstream always subscribes to `com.atproto.lexicon.schema` alongside the dynamic record collections. When a commit event arrives: + +- **create/update**: If the event's DID and rkey match a tracked network lexicon (`authority_did` and `nsid`), the lexicon is parsed, upserted into the `lexicons` table and in-memory registry, and Jetstream is notified if it's a record type. +- **delete**: The lexicon is removed from the `lexicons` table and registry. + +### Startup re-fetch + +On every startup, HappyView re-fetches all network lexicons from their respective PDSes. This ensures consistency even if Jetstream events were missed while offline. Failures are logged as warnings but don't block startup. + +## Admin API + +See [Admin API - Network Lexicons](admin-api.md#network-lexicons) for endpoint details. + +### Quick reference + +```sh +# Add a network lexicon +curl -X POST http://localhost:3000/admin/network-lexicons \ + -H "$AUTH" -H "Content-Type: application/json" \ + -d '{ "nsid": "games.gamesgamesgamesgames.game" }' + +# List tracked network lexicons +curl http://localhost:3000/admin/network-lexicons -H "$AUTH" + +# Remove a network lexicon +curl -X DELETE http://localhost:3000/admin/network-lexicons/games.gamesgamesgamesgames.game \ + -H "$AUTH" +``` diff --git a/migrations/20260219000000_create_network_lexicons.sql b/migrations/20260219000000_create_network_lexicons.sql new file mode 100644 --- /dev/null +++ b/migrations/20260219000000_create_network_lexicons.sql @@ -0,0 +1,8 @@ +CREATE TABLE IF NOT EXISTS network_lexicons ( + id UUID PRIMARY KEY DEFAULT gen_random_uuid(), + nsid TEXT NOT NULL UNIQUE, + authority_did TEXT NOT NULL, + target_collection TEXT, + last_fetched_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); diff --git a/src/admin/mod.rs b/src/admin/mod.rs --- a/src/admin/mod.rs +++ b/src/admin/mod.rs @@ -2,6 +2,7 @@ mod admins; pub(crate) mod auth; mod backfill; mod lexicons; +mod network_lexicons; mod stats; mod types; @@ -28,4 +29,9 @@ "/admins", post(admins::create_admin).get(admins::list_admins), ) .route("/admins/{id}", delete(admins::delete_admin)) + .route( + "/network-lexicons", + post(network_lexicons::add).get(network_lexicons::list), + ) + .route("/network-lexicons/{nsid}", delete(network_lexicons::remove)) } diff --git a/src/admin/network_lexicons.rs b/src/admin/network_lexicons.rs new file mode 100644 --- /dev/null +++ b/src/admin/network_lexicons.rs @@ -0,0 +1,161 @@ +use axum::Json; +use axum::extract::{Path, State}; +use axum::http::StatusCode; +use serde_json::Value; + +use crate::AppState; +use crate::error::AppError; +use crate::lexicon::{LexiconType, ParsedLexicon}; +use crate::resolve::{fetch_lexicon_from_pds, resolve_nsid_authority}; + +use super::auth::AdminAuth; +use super::types::{AddNetworkLexiconBody, NetworkLexiconSummary}; + +/// Send the current record collection list to the Jetstream task so it +/// reconnects with the updated filter. +async fn notify_jetstream(state: &AppState) { + let collections = state.lexicons.get_record_collections().await; + let _ = state.collections_tx.send(collections); +} + +/// POST /admin/network-lexicons — add a network lexicon to watch. +pub(super) async fn add( + State(state): State, + _admin: AdminAuth, + Json(body): Json, +) -> Result<(StatusCode, Json), AppError> { + let nsid = &body.nsid; + + // Resolve NSID authority via DNS TXT lookup. + let (authority_did, pds_endpoint) = + resolve_nsid_authority(&state.http, &state.config.plc_url, nsid).await?; + + // Fetch the lexicon from the authority's PDS. + let lexicon_json = + fetch_lexicon_from_pds(&state.http, &pds_endpoint, &authority_did, nsid).await?; + + // Parse to validate. + let parsed = ParsedLexicon::parse(lexicon_json.clone(), 1, body.target_collection.clone()) + .map_err(|e| AppError::BadRequest(format!("failed to parse lexicon: {e}")))?; + + // Insert into network_lexicons table. + sqlx::query( + r#" + INSERT INTO network_lexicons (nsid, authority_did, target_collection, last_fetched_at) + VALUES ($1, $2, $3, NOW()) + ON CONFLICT (nsid) DO UPDATE SET + authority_did = EXCLUDED.authority_did, + target_collection = EXCLUDED.target_collection, + last_fetched_at = NOW() + "#, + ) + .bind(nsid) + .bind(&authority_did) + .bind(&body.target_collection) + .execute(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to insert network lexicon: {e}")))?; + + // Upsert into lexicons table. + let row: (i32,) = sqlx::query_as( + r#" + INSERT INTO lexicons (id, lexicon_json, backfill, target_collection) + VALUES ($1, $2, false, $3) + ON CONFLICT (id) DO UPDATE SET + lexicon_json = EXCLUDED.lexicon_json, + target_collection = EXCLUDED.target_collection, + revision = lexicons.revision + 1, + updated_at = NOW() + RETURNING revision + "#, + ) + .bind(nsid) + .bind(&lexicon_json) + .bind(&body.target_collection) + .fetch_one(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to upsert lexicon: {e}")))?; + + let revision = row.0; + + // Update in-memory registry. + let is_record = parsed.lexicon_type == LexiconType::Record; + let parsed = ParsedLexicon::parse(lexicon_json, revision, body.target_collection) + .map_err(|e| AppError::Internal(format!("failed to re-parse lexicon: {e}")))?; + state.lexicons.upsert(parsed).await; + + if is_record { + notify_jetstream(&state).await; + } + + Ok(( + StatusCode::CREATED, + Json(serde_json::json!({ + "nsid": nsid, + "authority_did": authority_did, + "revision": revision, + })), + )) +} + +/// GET /admin/network-lexicons — list tracked network lexicons. +pub(super) async fn list( + State(state): State, + _admin: AdminAuth, +) -> Result>, AppError> { + #[allow(clippy::type_complexity)] + let rows: Vec<(String, String, Option, Option>, chrono::DateTime)> = + sqlx::query_as( + "SELECT nsid, authority_did, target_collection, last_fetched_at, created_at FROM network_lexicons ORDER BY nsid", + ) + .fetch_all(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to list network lexicons: {e}")))?; + + let summaries: Vec = rows + .into_iter() + .map( + |(nsid, authority_did, target_collection, last_fetched_at, created_at)| { + NetworkLexiconSummary { + nsid, + authority_did, + target_collection, + last_fetched_at, + created_at, + } + }, + ) + .collect(); + + Ok(Json(summaries)) +} + +/// DELETE /admin/network-lexicons/{nsid} — stop watching a network lexicon. +pub(super) async fn remove( + State(state): State, + _admin: AdminAuth, + Path(nsid): Path, +) -> Result { + let result = sqlx::query("DELETE FROM network_lexicons WHERE nsid = $1") + .bind(&nsid) + .execute(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to delete network lexicon: {e}")))?; + + if result.rows_affected() == 0 { + return Err(AppError::NotFound(format!( + "network lexicon '{nsid}' not found" + ))); + } + + // Also remove from lexicons table and registry. + let _ = sqlx::query("DELETE FROM lexicons WHERE id = $1") + .bind(&nsid) + .execute(&state.db) + .await; + + state.lexicons.remove(&nsid).await; + notify_jetstream(&state).await; + + Ok(StatusCode::NO_CONTENT) +} diff --git a/src/admin/types.rs b/src/admin/types.rs --- a/src/admin/types.rs +++ b/src/admin/types.rs @@ -69,6 +69,25 @@ pub(super) created_at: chrono::DateTime, } // --------------------------------------------------------------------------- +// Network lexicon types +// --------------------------------------------------------------------------- + +#[derive(Deserialize)] +pub(super) struct AddNetworkLexiconBody { + pub(super) nsid: String, + pub(super) target_collection: Option, +} + +#[derive(Serialize)] +pub(super) struct NetworkLexiconSummary { + pub(super) nsid: String, + pub(super) authority_did: String, + pub(super) target_collection: Option, + pub(super) last_fetched_at: Option>, + pub(super) created_at: chrono::DateTime, +} + +// --------------------------------------------------------------------------- // Admin management types // --------------------------------------------------------------------------- diff --git a/src/jetstream.rs b/src/jetstream.rs --- a/src/jetstream.rs +++ b/src/jetstream.rs @@ -7,6 +7,8 @@ use std::sync::atomic::{AtomicI64, Ordering}; use tokio::sync::watch; use tokio_tungstenite::tungstenite::Message; +use crate::lexicon::{LexiconRegistry, ParsedLexicon}; + // --------------------------------------------------------------------------- // Jetstream event types // --------------------------------------------------------------------------- @@ -32,12 +34,21 @@ // --------------------------------------------------------------------------- // Public API // --------------------------------------------------------------------------- +/// The static collection we always watch for lexicon schema updates. +const LEXICON_SCHEMA_COLLECTION: &str = "com.atproto.lexicon.schema"; + /// Spawn a background task that subscribes to the Jetstream firehose and /// indexes records for collections specified by the watch channel. /// /// When the collection list is empty, the task idles without connecting. /// When collections change, it disconnects and reconnects with the new filter. -pub fn spawn(db: PgPool, jetstream_url: String, mut collections_rx: watch::Receiver>) { +pub fn spawn( + db: PgPool, + jetstream_url: String, + mut collections_rx: watch::Receiver>, + lexicons: LexiconRegistry, + collections_tx: watch::Sender>, +) { tokio::spawn(async move { let cursor: Arc = Arc::new(AtomicI64::new(0)); @@ -55,14 +66,22 @@ } continue; } + // Always include the lexicon schema collection alongside the dynamic ones. + let mut wanted = collections.clone(); + if !wanted.contains(&LEXICON_SCHEMA_COLLECTION.to_string()) { + wanted.push(LEXICON_SCHEMA_COLLECTION.to_string()); + } + // Connect and process events. If the collection list changes // mid-stream, `run` returns so we can reconnect with new filters. match run( &db, &jetstream_url, &cursor, - &collections, + &wanted, &mut collections_rx, + &lexicons, + &collections_tx, ) .await { @@ -89,6 +108,8 @@ jetstream_url: &str, cursor: &Arc, collections: &[String], collections_rx: &mut watch::Receiver>, + lexicons: &LexiconRegistry, + collections_tx: &watch::Sender>, ) -> Result<(), Box> { let wanted: String = collections .iter() @@ -153,6 +174,19 @@ "at://{}/{}/{}", event.did, commit.collection, commit.rkey, ); + // Handle lexicon schema events for tracked network lexicons. + if commit.collection == LEXICON_SCHEMA_COLLECTION { + handle_lexicon_schema_event( + db, + lexicons, + collections_tx, + &event.did, + &commit, + ) + .await; + continue; + } + match commit.operation.as_str() { "create" | "update" => { let record = match commit.record { @@ -205,3 +239,104 @@ } Ok(()) } + +// --------------------------------------------------------------------------- +// Lexicon schema event handler +// --------------------------------------------------------------------------- + +/// Handle a `com.atproto.lexicon.schema` commit event for tracked network lexicons. +async fn handle_lexicon_schema_event( + db: &PgPool, + lexicons: &LexiconRegistry, + collections_tx: &watch::Sender>, + did: &str, + commit: &JetstreamCommit, +) { + let nsid = &commit.rkey; + + // Check if this NSID is one we're tracking and the DID matches the authority. + let tracked: Option<(Option,)> = sqlx::query_as( + "SELECT target_collection FROM network_lexicons WHERE nsid = $1 AND authority_did = $2", + ) + .bind(nsid) + .bind(did) + .fetch_optional(db) + .await + .unwrap_or(None); + + let target_collection = match tracked { + Some((tc,)) => tc, + None => return, // Not a tracked network lexicon. + }; + + match commit.operation.as_str() { + "create" | "update" => { + let record = match &commit.record { + Some(r) => r, + None => return, + }; + + let parsed = match ParsedLexicon::parse(record.clone(), 1, target_collection.clone()) { + Ok(p) => p, + Err(e) => { + tracing::warn!(nsid, "failed to parse lexicon schema event: {e}"); + return; + } + }; + + let is_record = parsed.lexicon_type == crate::lexicon::LexiconType::Record; + + // Upsert into lexicons table. + if let Err(e) = sqlx::query( + r#" + INSERT INTO lexicons (id, lexicon_json, backfill, target_collection) + VALUES ($1, $2, false, $3) + ON CONFLICT (id) DO UPDATE SET + lexicon_json = EXCLUDED.lexicon_json, + target_collection = EXCLUDED.target_collection, + revision = lexicons.revision + 1, + updated_at = NOW() + "#, + ) + .bind(nsid) + .bind(record) + .bind(&target_collection) + .execute(db) + .await + { + tracing::warn!(nsid, "failed to upsert lexicon from event: {e}"); + return; + } + + // Update last_fetched_at. + let _ = + sqlx::query("UPDATE network_lexicons SET last_fetched_at = NOW() WHERE nsid = $1") + .bind(nsid) + .execute(db) + .await; + + lexicons.upsert(parsed).await; + tracing::info!(nsid, "updated network lexicon from jetstream event"); + + if is_record { + let collections = lexicons.get_record_collections().await; + let _ = collections_tx.send(collections); + } + } + "delete" => { + // Remove from lexicons table and registry. + let _ = sqlx::query("DELETE FROM lexicons WHERE id = $1") + .bind(nsid) + .execute(db) + .await; + + let was_present = lexicons.remove(nsid).await; + if was_present { + tracing::info!(nsid, "removed network lexicon from jetstream delete event"); + let collections = lexicons.get_record_collections().await; + let _ = collections_tx.send(collections); + } + } + _ => {} + } +} diff --git a/src/lib.rs b/src/lib.rs --- a/src/lib.rs +++ b/src/lib.rs @@ -7,6 +7,7 @@ pub mod jetstream; pub mod lexicon; pub mod profile; pub mod repo; +pub mod resolve; pub mod server; pub mod xrpc; diff --git a/src/main.rs b/src/main.rs --- a/src/main.rs +++ b/src/main.rs @@ -1,8 +1,9 @@ use happyview::config::Config; -use happyview::lexicon::LexiconRegistry; +use happyview::lexicon::{LexiconRegistry, ParsedLexicon}; +use happyview::resolve::{fetch_lexicon_from_pds, resolve_nsid_authority}; use happyview::{AppState, backfill, jetstream, server}; use tokio::sync::watch; -use tracing::info; +use tracing::{info, warn}; #[tokio::main] async fn main() { @@ -35,12 +36,81 @@ .load_from_db(&db) .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, String, Option)> = + sqlx::query_as("SELECT nsid, authority_did, target_collection FROM network_lexicons") + .fetch_all(&db) + .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(), + ) { + Ok(parsed) => { + // Upsert into lexicons table. + if let Err(e) = sqlx::query( + r#" + INSERT INTO lexicons (id, lexicon_json, backfill, target_collection) + VALUES ($1, $2, false, $3) + ON CONFLICT (id) DO UPDATE SET + lexicon_json = EXCLUDED.lexicon_json, + target_collection = EXCLUDED.target_collection, + revision = lexicons.revision + 1, + updated_at = NOW() + "#, + ) + .bind(nsid) + .bind(&lexicon_json) + .bind(target_collection) + .execute(&db) + .await + { + warn!(nsid, "failed to upsert network lexicon into DB: {e}"); + continue; + } + + // Update last_fetched_at. + let _ = sqlx::query( + "UPDATE network_lexicons SET last_fetched_at = NOW() WHERE nsid = $1", + ) + .bind(nsid) + .execute(&db) + .await; + + 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" + ); + } + let initial_collections = lexicons.get_record_collections().await; let (collections_tx, collections_rx) = watch::channel(initial_collections); let state = AppState { config: config.clone(), - http: reqwest::Client::new(), + http, db, lexicons, collections_tx, @@ -50,6 +120,8 @@ jetstream::spawn( state.db.clone(), config.jetstream_url.clone(), collections_rx, + state.lexicons.clone(), + state.collections_tx.clone(), ); backfill::spawn_worker( state.db.clone(), diff --git a/src/resolve.rs b/src/resolve.rs new file mode 100644 --- /dev/null +++ b/src/resolve.rs @@ -0,0 +1,105 @@ +use hickory_resolver::Resolver; + +use crate::error::AppError; +use crate::profile::resolve_pds_endpoint; + +/// Resolve the authority DID and PDS endpoint for the given NSID. +/// +/// 1. Extract the authority from the NSID (all segments except the last). +/// 2. Reverse the authority segments to form a domain name. +/// 3. Look up `_lexicon.{domain}` TXT record for a `did=` value. +/// 4. Resolve the DID → PDS endpoint via the PLC directory. +pub async fn resolve_nsid_authority( + http: &reqwest::Client, + plc_url: &str, + nsid: &str, +) -> Result<(String, String), AppError> { + let segments: Vec<&str> = nsid.split('.').collect(); + if segments.len() < 2 { + return Err(AppError::BadRequest(format!( + "invalid NSID (too few segments): {nsid}" + ))); + } + + // Authority is everything except the last segment, reversed to form a domain. + let authority_segments = &segments[..segments.len() - 1]; + let reversed: Vec<&str> = authority_segments.iter().copied().rev().collect(); + let domain = reversed.join("."); + + let lookup_name = format!("_lexicon.{domain}."); + + let resolver = Resolver::builder_tokio() + .map_err(|e| AppError::Internal(format!("failed to create DNS resolver: {e}")))? + .build(); + + let txt_lookup = resolver.txt_lookup(&lookup_name).await.map_err(|e| { + AppError::BadRequest(format!("DNS TXT lookup failed for {lookup_name}: {e}")) + })?; + + let did = txt_lookup + .iter() + .flat_map(|txt| txt.txt_data().iter()) + .filter_map(|data| { + let s = std::str::from_utf8(data).ok()?; + s.strip_prefix("did=") + }) + .next() + .ok_or_else(|| AppError::BadRequest(format!("no did= TXT record found at {lookup_name}")))? + .to_string(); + + let pds_endpoint = resolve_pds_endpoint(http, plc_url, &did).await?; + + Ok((did, pds_endpoint)) +} + +/// Fetch a lexicon record from a PDS. +/// +/// Calls `com.atproto.repo.getRecord` with collection `com.atproto.lexicon.schema` +/// and the NSID as the rkey. Returns the `value` field (the raw lexicon JSON). +pub async fn fetch_lexicon_from_pds( + http: &reqwest::Client, + pds_endpoint: &str, + did: &str, + nsid: &str, +) -> Result { + let url = format!( + "{}/xrpc/com.atproto.repo.getRecord?repo={}&collection=com.atproto.lexicon.schema&rkey={}", + pds_endpoint.trim_end_matches('/'), + did, + nsid, + ); + + let resp = http + .get(&url) + .send() + .await + .map_err(|e| AppError::Internal(format!("PDS request failed: {e}")))?; + + if !resp.status().is_success() { + return Err(AppError::NotFound(format!( + "lexicon record not found for {nsid} in {did}'s repo" + ))); + } + + let body: serde_json::Value = resp + .json() + .await + .map_err(|e| AppError::Internal(format!("invalid PDS response: {e}")))?; + + body.get("value") + .cloned() + .ok_or_else(|| AppError::Internal("PDS response missing 'value' field".into())) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn nsid_too_few_segments_is_error() { + let rt = tokio::runtime::Runtime::new().unwrap(); + let http = reqwest::Client::new(); + let result = rt.block_on(resolve_nsid_authority(&http, "http://localhost", "single")); + assert!(result.is_err()); + } +} diff --git a/tests/common/db.rs b/tests/common/db.rs --- a/tests/common/db.rs +++ b/tests/common/db.rs @@ -19,7 +19,7 @@ } /// Truncate all application tables, preserving schema. pub async fn truncate_all(pool: &PgPool) { - sqlx::query("TRUNCATE records, lexicons, backfill_jobs, admins RESTART IDENTITY CASCADE") + sqlx::query("TRUNCATE records, lexicons, backfill_jobs, admins, network_lexicons RESTART IDENTITY CASCADE") .execute(pool) .await .expect("failed to truncate tables"); diff --git a/tests/e2e_network_lexicons.rs b/tests/e2e_network_lexicons.rs new file mode 100644 --- /dev/null +++ b/tests/e2e_network_lexicons.rs @@ -0,0 +1,191 @@ +mod common; + +use axum::body::Body; +use axum::http::{Request, StatusCode}; +use http_body_util::BodyExt; +use serde_json::Value; +use serial_test::serial; +use tower::ServiceExt; + +use common::app::TestApp; +use common::auth::admin_auth_header; +use common::fixtures; + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +async fn json_body(resp: axum::response::Response) -> Value { + let body = resp.into_body().collect().await.unwrap().to_bytes(); + serde_json::from_slice(&body).unwrap() +} + +fn admin_get(uri: &str, token: &str) -> Request { + let (hname, hval) = admin_auth_header(token); + Request::builder() + .uri(uri) + .header(hname, hval) + .body(Body::empty()) + .unwrap() +} + +fn admin_delete(uri: &str, token: &str) -> Request { + let (hname, hval) = admin_auth_header(token); + Request::builder() + .method("DELETE") + .uri(uri) + .header(hname, hval) + .body(Body::empty()) + .unwrap() +} + +/// Set up mocks for NSID authority resolution: +/// - DNS TXT is not mockable in e2e, so we test at the API level by mocking +/// the PLC directory and PDS responses and seeding the network_lexicons table directly. +async fn seed_network_lexicon(app: &TestApp, nsid: &str, authority_did: &str) { + sqlx::query( + r#" + INSERT INTO network_lexicons (nsid, authority_did, last_fetched_at) + VALUES ($1, $2, NOW()) + ON CONFLICT (nsid) DO NOTHING + "#, + ) + .bind(nsid) + .bind(authority_did) + .execute(&app.state.db) + .await + .expect("failed to seed network lexicon"); + + // Also seed the lexicons table so it's consistent. + let lexicon_json = fixtures::game_record_lexicon(); + sqlx::query( + r#" + INSERT INTO lexicons (id, lexicon_json, backfill) + VALUES ($1, $2, false) + ON CONFLICT (id) DO NOTHING + "#, + ) + .bind(nsid) + .bind(&lexicon_json) + .execute(&app.state.db) + .await + .expect("failed to seed lexicon"); +} + +// --------------------------------------------------------------------------- +// Network lexicon CRUD +// --------------------------------------------------------------------------- + +#[tokio::test] +#[serial] +async fn network_lexicon_list_empty() { + let app = TestApp::new().await; + app.mock_admin_userinfo().await; + + let resp = app + .router + .oneshot(admin_get("/admin/network-lexicons", &app.admin_token)) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + let json = json_body(resp).await; + assert!(json.as_array().unwrap().is_empty()); +} + +#[tokio::test] +#[serial] +async fn network_lexicon_list_returns_seeded() { + let app = TestApp::new().await; + app.mock_admin_userinfo().await; + + seed_network_lexicon(&app, "games.gamesgamesgamesgames.game", "did:plc:authority").await; + + let resp = app + .router + .oneshot(admin_get("/admin/network-lexicons", &app.admin_token)) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + let json = json_body(resp).await; + let arr = json.as_array().unwrap(); + assert_eq!(arr.len(), 1); + assert_eq!(arr[0]["nsid"], "games.gamesgamesgamesgames.game"); + assert_eq!(arr[0]["authority_did"], "did:plc:authority"); +} + +#[tokio::test] +#[serial] +async fn network_lexicon_delete_removes_tracking_and_lexicon() { + let app = TestApp::new().await; + app.mock_admin_userinfo().await; + + let nsid = "games.gamesgamesgamesgames.game"; + seed_network_lexicon(&app, nsid, "did:plc:authority").await; + + let resp = app + .router + .clone() + .oneshot(admin_delete( + &format!("/admin/network-lexicons/{nsid}"), + &app.admin_token, + )) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::NO_CONTENT); + + // Verify network_lexicons table is empty. + let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM network_lexicons WHERE nsid = $1") + .bind(nsid) + .fetch_one(&app.state.db) + .await + .unwrap(); + assert_eq!(count.0, 0); + + // Verify lexicons table is also cleaned up. + let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM lexicons WHERE id = $1") + .bind(nsid) + .fetch_one(&app.state.db) + .await + .unwrap(); + assert_eq!(count.0, 0); +} + +#[tokio::test] +#[serial] +async fn network_lexicon_delete_not_found() { + let app = TestApp::new().await; + app.mock_admin_userinfo().await; + + let resp = app + .router + .oneshot(admin_delete( + "/admin/network-lexicons/nonexistent.lexicon", + &app.admin_token, + )) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::NOT_FOUND); +} + +#[tokio::test] +#[serial] +async fn network_lexicon_no_auth_returns_401() { + let app = TestApp::new().await; + + let resp = app + .router + .oneshot( + Request::builder() + .uri("/admin/network-lexicons") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::UNAUTHORIZED); +}