From fc92b489b6d824a766d177f4fad6f3810f8b011a Mon Sep 17 00:00:00 2001 From: Orual Date: Sat, 13 Dec 2025 20:45:27 -0500 Subject: [PATCH] more endpoints, service auth options --- Cargo.lock | 26 +- crates/weaver-index/Cargo.toml | 6 +- crates/weaver-index/src/bin/weaver_indexer.rs | 64 ++- .../src/clickhouse/queries/notebooks.rs | 474 +++++++++++++++++- crates/weaver-index/src/endpoints/actor.rs | 341 ++++++++++++- crates/weaver-index/src/endpoints/notebook.rs | 362 ++++++++++++- crates/weaver-index/src/lib.rs | 2 + crates/weaver-index/src/parallel_tap.rs | 10 +- crates/weaver-index/src/server.rs | 81 ++- crates/weaver-index/src/service_identity.rs | 173 +++++++ lexicons/notebook/resolveNotebook.json | 4 +- scripts/test-indexer.sh | 75 ++- 12 files changed, 1567 insertions(+), 51 deletions(-) create mode 100644 crates/weaver-index/src/service_identity.rs diff --git a/Cargo.lock b/Cargo.lock index 2db799d..3522dd1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1234,9 +1234,9 @@ dependencies = [ [[package]] name = "cmake" -version = "0.1.54" +version = "0.1.56" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e7caa3f9de89ddbe2c607f4101924c5abec803763ae9534e4f4d7d8f84aa81f0" +checksum = "b042e5d8a74ae91bb0961acd039822472ec99f8ab0948cbf6d1369588f8be586" dependencies = [ "cc", ] @@ -5567,7 +5567,7 @@ checksum = "4a5f13b858c8d314ee3e8f639011f7ccefe71f97f96e50151fb991f267928e2c" [[package]] name = "jacquard" version = "0.9.4" -source = "git+https://tangled.org/@nonbinary.computer/jacquard#5bd87b462e1b020feb232fefaf4926ff004eefe4" +source = "git+https://tangled.org/@nonbinary.computer/jacquard#d5d29a337d8f08ae52fe7bb4e31f91f8b029ff48" dependencies = [ "bytes", "getrandom 0.2.16", @@ -5599,7 +5599,7 @@ dependencies = [ [[package]] name = "jacquard-api" version = "0.9.2" -source = "git+https://tangled.org/@nonbinary.computer/jacquard#5bd87b462e1b020feb232fefaf4926ff004eefe4" +source = "git+https://tangled.org/@nonbinary.computer/jacquard#d5d29a337d8f08ae52fe7bb4e31f91f8b029ff48" dependencies = [ "bon", "bytes", @@ -5618,7 +5618,7 @@ dependencies = [ [[package]] name = "jacquard-axum" version = "0.9.2" -source = "git+https://tangled.org/@nonbinary.computer/jacquard#5bd87b462e1b020feb232fefaf4926ff004eefe4" +source = "git+https://tangled.org/@nonbinary.computer/jacquard#d5d29a337d8f08ae52fe7bb4e31f91f8b029ff48" dependencies = [ "axum", "bytes", @@ -5640,7 +5640,7 @@ dependencies = [ [[package]] name = "jacquard-common" version = "0.9.2" -source = "git+https://tangled.org/@nonbinary.computer/jacquard#5bd87b462e1b020feb232fefaf4926ff004eefe4" +source = "git+https://tangled.org/@nonbinary.computer/jacquard#d5d29a337d8f08ae52fe7bb4e31f91f8b029ff48" dependencies = [ "base64 0.22.1", "bon", @@ -5688,7 +5688,7 @@ dependencies = [ [[package]] name = "jacquard-derive" version = "0.9.4" -source = "git+https://tangled.org/@nonbinary.computer/jacquard#5bd87b462e1b020feb232fefaf4926ff004eefe4" +source = "git+https://tangled.org/@nonbinary.computer/jacquard#d5d29a337d8f08ae52fe7bb4e31f91f8b029ff48" dependencies = [ "heck 0.5.0", "jacquard-lexicon", @@ -5700,7 +5700,7 @@ dependencies = [ [[package]] name = "jacquard-identity" version = "0.9.2" -source = "git+https://tangled.org/@nonbinary.computer/jacquard#5bd87b462e1b020feb232fefaf4926ff004eefe4" +source = "git+https://tangled.org/@nonbinary.computer/jacquard#d5d29a337d8f08ae52fe7bb4e31f91f8b029ff48" dependencies = [ "bon", "bytes", @@ -5729,7 +5729,7 @@ dependencies = [ [[package]] name = "jacquard-lexicon" version = "0.9.2" -source = "git+https://tangled.org/@nonbinary.computer/jacquard#5bd87b462e1b020feb232fefaf4926ff004eefe4" +source = "git+https://tangled.org/@nonbinary.computer/jacquard#d5d29a337d8f08ae52fe7bb4e31f91f8b029ff48" dependencies = [ "cid", "dashmap 6.1.0", @@ -5755,7 +5755,7 @@ dependencies = [ [[package]] name = "jacquard-oauth" version = "0.9.2" -source = "git+https://tangled.org/@nonbinary.computer/jacquard#5bd87b462e1b020feb232fefaf4926ff004eefe4" +source = "git+https://tangled.org/@nonbinary.computer/jacquard#d5d29a337d8f08ae52fe7bb4e31f91f8b029ff48" dependencies = [ "base64 0.22.1", "bytes", @@ -5788,7 +5788,7 @@ dependencies = [ [[package]] name = "jacquard-repo" version = "0.9.4" -source = "git+https://tangled.org/@nonbinary.computer/jacquard#5bd87b462e1b020feb232fefaf4926ff004eefe4" +source = "git+https://tangled.org/@nonbinary.computer/jacquard#d5d29a337d8f08ae52fe7bb4e31f91f8b029ff48" dependencies = [ "bytes", "cid", @@ -6746,7 +6746,7 @@ dependencies = [ [[package]] name = "mini-moka" version = "0.10.99" -source = "git+https://tangled.org/@nonbinary.computer/jacquard#5bd87b462e1b020feb232fefaf4926ff004eefe4" +source = "git+https://tangled.org/@nonbinary.computer/jacquard#d5d29a337d8f08ae52fe7bb4e31f91f8b029ff48" dependencies = [ "crossbeam-channel", "crossbeam-utils", @@ -11787,8 +11787,10 @@ dependencies = [ "jacquard-axum", "jacquard-common", "jacquard-repo", + "k256", "miette 7.6.0", "n0-future 0.1.3", + "rand 0.8.5", "regex", "rusqlite", "rusqlite_migration", diff --git a/crates/weaver-index/Cargo.toml b/crates/weaver-index/Cargo.toml index 3ea61a9..b394dd4 100644 --- a/crates/weaver-index/Cargo.toml +++ b/crates/weaver-index/Cargo.toml @@ -25,10 +25,14 @@ weaver-common = { path = "../weaver-common", features = ["telemetry"] } # AT Protocol / Jacquard jacquard = { workspace = true, features = ["websocket", "zstd", "dns", "cache"] } -jacquard-common = { workspace = true } +jacquard-common = { workspace = true, features = ["crypto-k256"] } jacquard-repo = { workspace = true } jacquard-axum = { workspace = true } +# Crypto (for service identity keypair) +k256 = { version = "0.13", features = ["ecdsa"] } +rand = "0.8" + # ClickHouse clickhouse = { version = "0.14", features = ["inserter", "chrono", "rustls-tls-ring", "rustls-tls-webpki-roots"] } diff --git a/crates/weaver-index/src/bin/weaver_indexer.rs b/crates/weaver-index/src/bin/weaver_indexer.rs index fb50314..c17e200 100644 --- a/crates/weaver-index/src/bin/weaver_indexer.rs +++ b/crates/weaver-index/src/bin/weaver_indexer.rs @@ -1,13 +1,15 @@ +use std::path::PathBuf; + use clap::{Parser, Subcommand}; use tracing::{error, info, warn}; +use weaver_index::clickhouse::InserterConfig; use weaver_index::clickhouse::{Client, Migrator}; use weaver_index::config::{ ClickHouseConfig, FirehoseConfig, IndexerConfig, ShardConfig, SourceMode, TapConfig, }; use weaver_index::firehose::FirehoseConsumer; use weaver_index::server::{AppState, ServerConfig, TelemetryConfig, telemetry}; -use weaver_index::clickhouse::InserterConfig; -use weaver_index::{FirehoseIndexer, TapIndexer, load_cursor}; +use weaver_index::{FirehoseIndexer, ServiceIdentity, TapIndexer, load_cursor}; #[derive(Parser)] #[command(name = "indexer")] @@ -140,12 +142,39 @@ async fn run_full() -> miette::Result<()> { ); info!("SQLite shards at {}", shard_config.base_path.display()); + // Load or generate service identity keypair + let key_path = std::env::var("SERVICE_KEY_PATH") + .map(PathBuf::from) + .unwrap_or_else(|_| PathBuf::from("./data/service.key")); + let identity = ServiceIdentity::load_or_generate(&key_path)?; + info!( + public_key = %identity.public_key_multibase(), + "Service identity loaded" + ); + + // Generate DID document with service endpoint + let service_endpoint = std::env::var("SERVICE_ENDPOINT").unwrap_or_else(|_| { + format!( + "https://{}", + server_config + .service_did + .as_str() + .strip_prefix("did:web:") + .unwrap_or("index.weaver.sh") + ) + }); + let did_doc = identity.did_document_with_service(&server_config.service_did, &service_endpoint); + // Create separate clients for indexer and server let indexer_client = Client::new(&ch_config)?; let server_client = Client::new(&ch_config)?; // Build AppState for server - let state = AppState::new(server_client, shard_config); + let state = AppState::new( + server_client, + shard_config, + server_config.service_did.clone(), + ); // Spawn the indexer task let indexer_handle = match source_mode { @@ -182,7 +211,7 @@ async fn run_full() -> miette::Result<()> { // Run server, monitoring indexer health tokio::select! { - result = weaver_index::server::run(state, server_config) => { + result = weaver_index::server::run(state, server_config, did_doc) => { result?; } result = indexer_handle => { @@ -273,10 +302,33 @@ async fn run_server_only() -> miette::Result<()> { ); info!("SQLite shards at {}", shard_config.base_path.display()); + // Load or generate service identity keypair + let key_path = std::env::var("SERVICE_KEY_PATH") + .map(PathBuf::from) + .unwrap_or_else(|_| PathBuf::from("./data/service.key")); + let identity = ServiceIdentity::load_or_generate(&key_path)?; + info!( + public_key = %identity.public_key_multibase(), + "Service identity loaded" + ); + + // Generate DID document with service endpoint + let service_endpoint = std::env::var("SERVICE_ENDPOINT").unwrap_or_else(|_| { + format!( + "https://{}", + server_config + .service_did + .as_str() + .strip_prefix("did:web:") + .unwrap_or("localhost") + ) + }); + let did_doc = identity.did_document_with_service(&server_config.service_did, &service_endpoint); + let client = Client::new(&ch_config)?; - let state = AppState::new(client, shard_config); - weaver_index::server::run(state, server_config).await?; + let state = AppState::new(client, shard_config, server_config.service_did.clone()); + weaver_index::server::run(state, server_config, did_doc).await?; Ok(()) } diff --git a/crates/weaver-index/src/clickhouse/queries/notebooks.rs b/crates/weaver-index/src/clickhouse/queries/notebooks.rs index 683de42..7acd650 100644 --- a/crates/weaver-index/src/clickhouse/queries/notebooks.rs +++ b/crates/weaver-index/src/clickhouse/queries/notebooks.rs @@ -19,6 +19,8 @@ pub struct NotebookRow { pub tags: Vec, pub author_dids: Vec, #[serde(with = "clickhouse::serde::chrono::datetime64::millis")] + pub created_at: chrono::DateTime, + #[serde(with = "clickhouse::serde::chrono::datetime64::millis")] pub indexed_at: chrono::DateTime, pub record: SmolStr, } @@ -35,6 +37,8 @@ pub struct EntryRow { pub tags: Vec, pub author_dids: Vec, #[serde(with = "clickhouse::serde::chrono::datetime64::millis")] + pub created_at: chrono::DateTime, + #[serde(with = "clickhouse::serde::chrono::datetime64::millis")] pub indexed_at: chrono::DateTime, pub record: SmolStr, } @@ -58,13 +62,14 @@ impl Client { path, tags, author_dids, + created_at, indexed_at, record FROM notebooks WHERE did = ? AND (path = ? OR title = ?) AND deleted_at = toDateTime64(0, 3) - ORDER BY event_time DESC + ORDER BY toStartOfFiveMinutes(event_time) DESC, created_at DESC LIMIT 1 "#; @@ -102,13 +107,14 @@ impl Client { path, tags, author_dids, + created_at, indexed_at, record FROM notebooks WHERE did = ? AND rkey = ? AND deleted_at = toDateTime64(0, 3) - ORDER BY event_time DESC + ORDER BY toStartOfFiveMinutes(event_time) DESC, created_at DESC LIMIT 1 "#; @@ -131,13 +137,14 @@ impl Client { /// /// Note: This is a simplified version. The full implementation would /// need to join with notebook's entryList to get proper ordering. - /// For now, we just list entries by the same author. + /// For now, we just list entries by the same author, ordered by rkey (notebook order). pub async fn list_notebook_entries( &self, did: &str, limit: u32, cursor: Option<&str>, ) -> Result, IndexError> { + // Note: rkey ordering is intentional here - it's the notebook's entry order let query = if cursor.is_some() { r#" SELECT @@ -149,6 +156,7 @@ impl Client { path, tags, author_dids, + created_at, indexed_at, record FROM entries @@ -169,6 +177,7 @@ impl Client { path, tags, author_dids, + created_at, indexed_at, record FROM entries @@ -200,7 +209,7 @@ impl Client { /// Get an entry by rkey, picking the most recent version across collaborators. /// /// For collaborative entries, the same rkey may exist in multiple repos. - /// This returns the most recently updated version, with indexed_at as tiebreaker. + /// This returns the most recently updated version. /// /// `candidate_dids` should include the notebook owner + all collaborator DIDs. pub async fn get_entry( @@ -225,13 +234,14 @@ impl Client { path, tags, author_dids, + created_at, indexed_at, record FROM entries WHERE rkey = ? AND did IN ({}) AND deleted_at = toDateTime64(0, 3) - ORDER BY updated_at DESC, indexed_at DESC + ORDER BY toStartOfFiveMinutes(event_time) DESC, created_at DESC LIMIT 1 "#, placeholders.join(", ") @@ -271,13 +281,14 @@ impl Client { path, tags, author_dids, + created_at, indexed_at, record FROM entries WHERE did = ? AND rkey = ? AND deleted_at = toDateTime64(0, 3) - ORDER BY updated_at DESC, indexed_at DESC + ORDER BY toStartOfFiveMinutes(event_time) DESC, created_at DESC LIMIT 1 "#; @@ -312,13 +323,14 @@ impl Client { path, tags, author_dids, + created_at, indexed_at, record FROM entries WHERE did = ? AND (path = ? OR title = ?) AND deleted_at = toDateTime64(0, 3) - ORDER BY event_time DESC + ORDER BY toStartOfFiveMinutes(event_time) DESC, created_at DESC LIMIT 1 "#; @@ -337,4 +349,452 @@ impl Client { Ok(row) } + + /// List notebooks for an actor. + /// + /// Returns notebooks owned by the given DID, ordered by created_at DESC. + /// Cursor is created_at timestamp in milliseconds. + pub async fn list_actor_notebooks( + &self, + did: &str, + limit: u32, + cursor: Option, + ) -> Result, IndexError> { + let query = if cursor.is_some() { + r#" + SELECT + did, + rkey, + cid, + uri, + title, + path, + tags, + author_dids, + created_at, + indexed_at, + record + FROM notebooks + WHERE did = ? + AND deleted_at = toDateTime64(0, 3) + AND created_at < fromUnixTimestamp64Milli(?) + ORDER BY toStartOfFiveMinutes(event_time) DESC, created_at DESC + LIMIT ? + "# + } else { + r#" + SELECT + did, + rkey, + cid, + uri, + title, + path, + tags, + author_dids, + created_at, + indexed_at, + record + FROM notebooks + WHERE did = ? + AND deleted_at = toDateTime64(0, 3) + ORDER BY toStartOfFiveMinutes(event_time) DESC, created_at DESC + LIMIT ? + "# + }; + + let mut q = self.inner().query(query).bind(did); + + if let Some(c) = cursor { + q = q.bind(c); + } + + let rows = q + .bind(limit) + .fetch_all::() + .await + .map_err(|e| ClickHouseError::Query { + message: "failed to list actor notebooks".into(), + source: e, + })?; + + Ok(rows) + } + + /// List entries for an actor. + /// + /// Returns entries owned by the given DID, ordered by created_at DESC. + /// Cursor is created_at timestamp in milliseconds. + pub async fn list_actor_entries( + &self, + did: &str, + limit: u32, + cursor: Option, + ) -> Result, IndexError> { + let query = if cursor.is_some() { + r#" + SELECT + did, + rkey, + cid, + uri, + title, + path, + tags, + author_dids, + created_at, + indexed_at, + record + FROM entries + WHERE did = ? + AND deleted_at = toDateTime64(0, 3) + AND created_at < fromUnixTimestamp64Milli(?) + ORDER BY toStartOfFiveMinutes(event_time) DESC, created_at DESC + LIMIT ? + "# + } else { + r#" + SELECT + did, + rkey, + cid, + uri, + title, + path, + tags, + author_dids, + created_at, + indexed_at, + record + FROM entries + WHERE did = ? + AND deleted_at = toDateTime64(0, 3) + ORDER BY toStartOfFiveMinutes(event_time) DESC, created_at DESC + LIMIT ? + "# + }; + + let mut q = self.inner().query(query).bind(did); + + if let Some(c) = cursor { + q = q.bind(c); + } + + let rows = + q.bind(limit) + .fetch_all::() + .await + .map_err(|e| ClickHouseError::Query { + message: "failed to list actor entries".into(), + source: e, + })?; + + Ok(rows) + } + + /// Get a global feed of notebooks. + /// + /// Returns notebooks ordered by created_at DESC (chronological) or by + /// popularity metrics if algorithm is "popular". + /// Cursor is created_at timestamp in milliseconds. + pub async fn get_notebook_feed( + &self, + algorithm: &str, + tags: Option<&[&str]>, + limit: u32, + cursor: Option, + ) -> Result, IndexError> { + // For now, just chronological. Popular would need join with counts. + let base_query = if tags.is_some() && cursor.is_some() { + r#" + SELECT + did, + rkey, + cid, + uri, + title, + path, + tags, + author_dids, + created_at, + indexed_at, + record + FROM notebooks + WHERE deleted_at = toDateTime64(0, 3) + AND hasAny(tags, ?) + AND created_at < fromUnixTimestamp64Milli(?) + ORDER BY toStartOfFiveMinutes(event_time) DESC, created_at DESC + LIMIT ? + "# + } else if tags.is_some() { + r#" + SELECT + did, + rkey, + cid, + uri, + title, + path, + tags, + author_dids, + created_at, + indexed_at, + record + FROM notebooks + WHERE deleted_at = toDateTime64(0, 3) + AND hasAny(tags, ?) + ORDER BY toStartOfFiveMinutes(event_time) DESC, created_at DESC + LIMIT ? + "# + } else if cursor.is_some() { + r#" + SELECT + did, + rkey, + cid, + uri, + title, + path, + tags, + author_dids, + created_at, + indexed_at, + record + FROM notebooks + WHERE deleted_at = toDateTime64(0, 3) + AND created_at < fromUnixTimestamp64Milli(?) + ORDER BY toStartOfFiveMinutes(event_time) DESC, created_at DESC + LIMIT ? + "# + } else { + r#" + SELECT + did, + rkey, + cid, + uri, + title, + path, + tags, + author_dids, + created_at, + indexed_at, + record + FROM notebooks + WHERE deleted_at = toDateTime64(0, 3) + ORDER BY toStartOfFiveMinutes(event_time) DESC, created_at DESC + LIMIT ? + "# + }; + + let _ = algorithm; // TODO: implement popular sorting + + let mut q = self.inner().query(base_query); + + if let Some(t) = tags { + q = q.bind(t); + } + if let Some(c) = cursor { + q = q.bind(c); + } + + let rows = q + .bind(limit) + .fetch_all::() + .await + .map_err(|e| ClickHouseError::Query { + message: "failed to get notebook feed".into(), + source: e, + })?; + + Ok(rows) + } + + /// Get a global feed of entries. + /// + /// Returns entries ordered by created_at DESC (chronological). + /// Cursor is created_at timestamp in milliseconds. + pub async fn get_entry_feed( + &self, + algorithm: &str, + tags: Option<&[&str]>, + limit: u32, + cursor: Option, + ) -> Result, IndexError> { + let base_query = if tags.is_some() && cursor.is_some() { + r#" + SELECT + did, + rkey, + cid, + uri, + title, + path, + tags, + author_dids, + created_at, + indexed_at, + record + FROM entries + WHERE deleted_at = toDateTime64(0, 3) + AND hasAny(tags, ?) + AND created_at < fromUnixTimestamp64Milli(?) + ORDER BY toStartOfFiveMinutes(event_time) DESC, created_at DESC + LIMIT ? + "# + } else if tags.is_some() { + r#" + SELECT + did, + rkey, + cid, + uri, + title, + path, + tags, + author_dids, + created_at, + indexed_at, + record + FROM entries + WHERE deleted_at = toDateTime64(0, 3) + AND hasAny(tags, ?) + ORDER BY toStartOfFiveMinutes(event_time) DESC, created_at DESC + LIMIT ? + "# + } else if cursor.is_some() { + r#" + SELECT + did, + rkey, + cid, + uri, + title, + path, + tags, + author_dids, + created_at, + indexed_at, + record + FROM entries + WHERE deleted_at = toDateTime64(0, 3) + AND created_at < fromUnixTimestamp64Milli(?) + ORDER BY toStartOfFiveMinutes(event_time) DESC, created_at DESC + LIMIT ? + "# + } else { + r#" + SELECT + did, + rkey, + cid, + uri, + title, + path, + tags, + author_dids, + created_at, + indexed_at, + record + FROM entries + WHERE deleted_at = toDateTime64(0, 3) + ORDER BY toStartOfFiveMinutes(event_time) DESC, created_at DESC + LIMIT ? + "# + }; + + let _ = algorithm; // TODO: implement popular sorting + + let mut q = self.inner().query(base_query); + + if let Some(t) = tags { + q = q.bind(t); + } + if let Some(c) = cursor { + q = q.bind(c); + } + + let rows = + q.bind(limit) + .fetch_all::() + .await + .map_err(|e| ClickHouseError::Query { + message: "failed to get entry feed".into(), + source: e, + })?; + + Ok(rows) + } + + /// Get an entry at a specific index within a notebook. + /// + /// Returns the entry at the given 0-based index, plus adjacent entries for prev/next. + pub async fn get_book_entry_at_index( + &self, + notebook_did: &str, + notebook_rkey: &str, + index: u32, + ) -> Result, Option)>, IndexError> { + // Fetch entries for this notebook with index context + // We need 3 entries: prev (index-1), current (index), next (index+1) + let offset = if index > 0 { index - 1 } else { 0 }; + let fetch_count = if index > 0 { 3u32 } else { 2u32 }; + + let query = r#" + SELECT + e.did, + e.rkey, + e.cid, + e.uri, + e.title, + e.path, + e.tags, + e.author_dids, + e.created_at, + e.indexed_at, + e.record + FROM notebook_entries ne FINAL + INNER JOIN entries e ON + e.did = ne.entry_did + AND e.rkey = ne.entry_rkey + AND e.deleted_at = toDateTime64(0, 3) + WHERE ne.notebook_did = ? + AND ne.notebook_rkey = ? + ORDER BY ne.position ASC + LIMIT ? OFFSET ? + "#; + + let rows: Vec = self + .inner() + .query(query) + .bind(notebook_did) + .bind(notebook_rkey) + .bind(fetch_count) + .bind(offset) + .fetch_all() + .await + .map_err(|e| ClickHouseError::Query { + message: "failed to get book entry at index".into(), + source: e, + })?; + + if rows.is_empty() { + return Ok(None); + } + + // Determine which row is which based on the offset + let mut iter = rows.into_iter(); + if index == 0 { + // No prev, rows[0] is current, rows[1] is next (if exists) + let current = iter.next(); + let next = iter.next(); + Ok(current.map(|c| (c, None, next))) + } else { + // rows[0] is prev, rows[1] is current, rows[2] is next + let prev = iter.next(); + let current = iter.next(); + let next = iter.next(); + Ok(current.map(|c| (c, prev, next))) + } + } } diff --git a/crates/weaver-index/src/endpoints/actor.rs b/crates/weaver-index/src/endpoints/actor.rs index 5b5a33a..4d45a81 100644 --- a/crates/weaver-index/src/endpoints/actor.rs +++ b/crates/weaver-index/src/endpoints/actor.rs @@ -1,27 +1,42 @@ //! sh.weaver.actor.* endpoint handlers +use std::collections::{HashMap, HashSet}; + use axum::{Json, extract::State}; use jacquard::IntoStatic; use jacquard::cowstr::ToCowStr; use jacquard::identity::resolver::IdentityResolver; use jacquard::types::ident::AtIdentifier; -use jacquard::types::string::{Did, Handle}; +use jacquard::types::string::{AtUri, Cid, Did, Handle}; use jacquard_axum::ExtractXrpc; +use jacquard_axum::service_auth::{ExtractOptionalServiceAuth, VerifiedServiceAuth}; +use smol_str::SmolStr; use weaver_api::sh_weaver::actor::{ ProfileDataView, ProfileDataViewInner, ProfileView, + get_actor_entries::{GetActorEntriesOutput, GetActorEntriesRequest}, + get_actor_notebooks::{GetActorNotebooksOutput, GetActorNotebooksRequest}, get_profile::{GetProfileOutput, GetProfileRequest}, }; +use weaver_api::sh_weaver::notebook::{AuthorListView, EntryView, NotebookView}; +use crate::clickhouse::ProfileRow; use crate::endpoints::repo::XrpcErrorResponse; use crate::server::AppState; +/// Authenticated viewer context (if present) +pub type Viewer = Option>; + /// Handle sh.weaver.actor.getProfile /// /// Returns a profile view with counts for the requested actor. pub async fn get_profile( State(state): State, + ExtractOptionalServiceAuth(viewer): ExtractOptionalServiceAuth, ExtractXrpc(args): ExtractXrpc, ) -> Result>, XrpcErrorResponse> { + // viewer contains Some(auth) if the request has valid service auth + // can be used later for viewer-specific state (e.g., "you follow this person") + let _viewer = viewer; // Resolve identifier to DID let did = resolve_actor(&state, &args.actor).await?; let did_str = did.as_str(); @@ -156,3 +171,327 @@ fn non_empty_str(s: &smol_str::SmolStr) -> Option> { Some(s.to_cowstr().into_static()) } } + +/// Parse cursor string to i64 timestamp millis +fn parse_cursor(cursor: Option<&str>) -> Result, XrpcErrorResponse> { + cursor + .map(|c| { + c.parse::() + .map_err(|_| XrpcErrorResponse::invalid_request("Invalid cursor format")) + }) + .transpose() +} + +/// Handle sh.weaver.actor.getActorNotebooks +/// +/// Returns notebooks owned by the given actor. +pub async fn get_actor_notebooks( + State(state): State, + ExtractOptionalServiceAuth(viewer): ExtractOptionalServiceAuth, + ExtractXrpc(args): ExtractXrpc, +) -> Result>, XrpcErrorResponse> { + let _viewer: Viewer = viewer; + + // Resolve actor to DID + let did = resolve_actor(&state, &args.actor).await?; + let did_str = did.as_str(); + + // Fetch notebooks for this actor + let limit = args.limit.unwrap_or(50).clamp(1, 100) as u32; + let cursor = parse_cursor(args.cursor.as_deref())?; + + let notebook_rows = state + .clickhouse + .list_actor_notebooks(did_str, limit + 1, cursor) + .await + .map_err(|e| { + tracing::error!("Failed to list actor notebooks: {}", e); + XrpcErrorResponse::internal_error("Database query failed") + })?; + + // Check if there are more + let has_more = notebook_rows.len() > limit as usize; + let notebook_rows: Vec<_> = notebook_rows.into_iter().take(limit as usize).collect(); + + // Collect author DIDs for hydration + let mut all_author_dids: HashSet<&str> = HashSet::new(); + for nb in ¬ebook_rows { + for did in &nb.author_dids { + all_author_dids.insert(did.as_str()); + } + } + + // Batch fetch profiles + let author_dids_vec: Vec<&str> = all_author_dids.into_iter().collect(); + let profiles = state + .clickhouse + .get_profiles_batch(&author_dids_vec) + .await + .map_err(|e| { + tracing::error!("Failed to batch fetch profiles: {}", e); + XrpcErrorResponse::internal_error("Database query failed") + })?; + + let profile_map: HashMap<&str, &ProfileRow> = + profiles.iter().map(|p| (p.did.as_str(), p)).collect(); + + // Build NotebookViews + let mut notebooks: Vec> = Vec::with_capacity(notebook_rows.len()); + for nb_row in ¬ebook_rows { + let notebook_uri = AtUri::new(&nb_row.uri).map_err(|e| { + tracing::error!("Invalid notebook URI in db: {}", e); + XrpcErrorResponse::internal_error("Invalid URI stored") + })?; + + let notebook_cid = Cid::new(nb_row.cid.as_bytes()).map_err(|e| { + tracing::error!("Invalid notebook CID in db: {}", e); + XrpcErrorResponse::internal_error("Invalid CID stored") + })?; + + let authors = hydrate_authors(&nb_row.author_dids, &profile_map)?; + let record = parse_record_json(&nb_row.record)?; + + let notebook = NotebookView::new() + .uri(notebook_uri.into_static()) + .cid(notebook_cid.into_static()) + .authors(authors) + .record(record) + .indexed_at(nb_row.indexed_at.fixed_offset()) + .maybe_title(non_empty_str(&nb_row.title)) + .maybe_path(non_empty_str(&nb_row.path)) + .build(); + + notebooks.push(notebook); + } + + // Build cursor for pagination (created_at millis) + let next_cursor = if has_more { + notebook_rows + .last() + .map(|nb| nb.created_at.timestamp_millis().to_cowstr().into_static()) + } else { + None + }; + + Ok(Json( + GetActorNotebooksOutput { + notebooks, + cursor: next_cursor, + extra_data: None + } + .into_static(), + )) +} + +/// Handle sh.weaver.actor.getActorEntries +/// +/// Returns entries owned by the given actor. +pub async fn get_actor_entries( + State(state): State, + ExtractOptionalServiceAuth(viewer): ExtractOptionalServiceAuth, + ExtractXrpc(args): ExtractXrpc, +) -> Result>, XrpcErrorResponse> { + let _viewer: Viewer = viewer; + + // Resolve actor to DID + let did = resolve_actor(&state, &args.actor).await?; + let did_str = did.as_str(); + + // Fetch entries for this actor + let limit = args.limit.unwrap_or(50).clamp(1, 100) as u32; + let cursor = parse_cursor(args.cursor.as_deref())?; + + let entry_rows = state + .clickhouse + .list_actor_entries(did_str, limit + 1, cursor) + .await + .map_err(|e| { + tracing::error!("Failed to list actor entries: {}", e); + XrpcErrorResponse::internal_error("Database query failed") + })?; + + // Check if there are more + let has_more = entry_rows.len() > limit as usize; + let entry_rows: Vec<_> = entry_rows.into_iter().take(limit as usize).collect(); + + // Collect author DIDs for hydration + let mut all_author_dids: HashSet<&str> = HashSet::new(); + for entry in &entry_rows { + for did in &entry.author_dids { + all_author_dids.insert(did.as_str()); + } + } + + // Batch fetch profiles + let author_dids_vec: Vec<&str> = all_author_dids.into_iter().collect(); + let profiles = state + .clickhouse + .get_profiles_batch(&author_dids_vec) + .await + .map_err(|e| { + tracing::error!("Failed to batch fetch profiles: {}", e); + XrpcErrorResponse::internal_error("Database query failed") + })?; + + let profile_map: HashMap<&str, &ProfileRow> = + profiles.iter().map(|p| (p.did.as_str(), p)).collect(); + + // Build EntryViews + let mut entries: Vec> = Vec::with_capacity(entry_rows.len()); + for entry_row in &entry_rows { + let entry_uri = AtUri::new(&entry_row.uri).map_err(|e| { + tracing::error!("Invalid entry URI in db: {}", e); + XrpcErrorResponse::internal_error("Invalid URI stored") + })?; + + let entry_cid = Cid::new(entry_row.cid.as_bytes()).map_err(|e| { + tracing::error!("Invalid entry CID in db: {}", e); + XrpcErrorResponse::internal_error("Invalid CID stored") + })?; + + let authors = hydrate_authors(&entry_row.author_dids, &profile_map)?; + let record = parse_record_json(&entry_row.record)?; + + let entry = EntryView::new() + .uri(entry_uri.into_static()) + .cid(entry_cid.into_static()) + .authors(authors) + .record(record) + .indexed_at(entry_row.indexed_at.fixed_offset()) + .maybe_title(non_empty_str(&entry_row.title)) + .maybe_path(non_empty_str(&entry_row.path)) + .build(); + + entries.push(entry); + } + + // Build cursor for pagination (created_at millis) + let next_cursor = if has_more { + entry_rows + .last() + .map(|e| e.created_at.timestamp_millis().to_cowstr().into_static()) + } else { + None + }; + + Ok(Json( + GetActorEntriesOutput { + entries, + cursor: next_cursor, + extra_data: None, + } + .into_static(), + )) +} + +/// Hydrate author list from DIDs using profile map +fn hydrate_authors( + author_dids: &[SmolStr], + profile_map: &HashMap<&str, &ProfileRow>, +) -> Result>, XrpcErrorResponse> { + let mut authors = Vec::with_capacity(author_dids.len()); + + for (idx, did_str) in author_dids.iter().enumerate() { + let profile_data = if let Some(profile) = profile_map.get(did_str.as_str()) { + profile_to_data_view(profile)? + } else { + // No profile found - create minimal view with just the DID + let did = Did::new(did_str).map_err(|e| { + tracing::error!("Invalid DID in author_dids: {}", e); + XrpcErrorResponse::internal_error("Invalid DID stored") + })?; + + let inner_profile = ProfileView::new() + .did(did.into_static()) + .handle( + Handle::new(did_str) + .unwrap_or_else(|_| Handle::new("unknown.invalid").unwrap()), + ) + .build(); + + ProfileDataView::new() + .inner(ProfileDataViewInner::ProfileView(Box::new(inner_profile))) + .build() + }; + + let author_view = AuthorListView::new() + .index(idx as i64) + .record(profile_data.into_static()) + .build(); + + authors.push(author_view); + } + + Ok(authors) +} + +/// Convert ProfileRow to ProfileDataView +fn profile_to_data_view( + profile: &ProfileRow, +) -> Result, XrpcErrorResponse> { + use jacquard::types::string::Uri; + + let did = Did::new(&profile.did).map_err(|e| { + tracing::error!("Invalid DID in profile: {}", e); + XrpcErrorResponse::internal_error("Invalid DID stored") + })?; + + let handle = if profile.handle.is_empty() { + Handle::new(&profile.did).unwrap_or_else(|_| Handle::new("unknown.invalid").unwrap()) + } else { + Handle::new(&profile.handle).map_err(|e| { + tracing::error!("Invalid handle in profile: {}", e); + XrpcErrorResponse::internal_error("Invalid handle stored") + })? + }; + + // Build avatar URL from CID if present + let avatar = if !profile.avatar_cid.is_empty() { + let url = format!( + "https://cdn.bsky.app/img/avatar/plain/{}/{}@jpeg", + profile.did, profile.avatar_cid + ); + Uri::new_owned(url).ok() + } else { + None + }; + + // Build banner URL from CID if present + let banner = if !profile.banner_cid.is_empty() { + let url = format!( + "https://cdn.bsky.app/img/banner/plain/{}/{}@jpeg", + profile.did, profile.banner_cid + ); + Uri::new_owned(url).ok() + } else { + None + }; + + let inner_profile = ProfileView::new() + .did(did.into_static()) + .handle(handle.into_static()) + .maybe_display_name(non_empty_str(&profile.display_name)) + .maybe_description(non_empty_str(&profile.description)) + .maybe_avatar(avatar) + .maybe_banner(banner) + .build(); + + let profile_data = ProfileDataView::new() + .inner(ProfileDataViewInner::ProfileView(Box::new(inner_profile))) + .build(); + + Ok(profile_data) +} + +/// Parse record JSON string into owned Data +fn parse_record_json( + json: &str, +) -> Result, XrpcErrorResponse> { + use jacquard::types::value::Data; + + let data: Data<'_> = serde_json::from_str(json).map_err(|e| { + tracing::error!("Failed to parse record JSON: {}", e); + XrpcErrorResponse::internal_error("Invalid record JSON stored") + })?; + Ok(data.into_static()) +} diff --git a/crates/weaver-index/src/endpoints/notebook.rs b/crates/weaver-index/src/endpoints/notebook.rs index 6bf8552..f6f3745 100644 --- a/crates/weaver-index/src/endpoints/notebook.rs +++ b/crates/weaver-index/src/endpoints/notebook.rs @@ -8,17 +8,21 @@ use jacquard::cowstr::ToCowStr; use jacquard::types::string::{AtUri, Cid, Did, Handle, Uri}; use jacquard::types::value::Data; use jacquard_axum::ExtractXrpc; +use jacquard_axum::service_auth::ExtractOptionalServiceAuth; use smol_str::SmolStr; use weaver_api::sh_weaver::actor::{ProfileDataView, ProfileDataViewInner, ProfileView}; use weaver_api::sh_weaver::notebook::{ - AuthorListView, BookEntryView, EntryView, NotebookView, + AuthorListView, BookEntryRef, BookEntryView, EntryView, FeedEntryView, NotebookView, + get_book_entry::{GetBookEntryOutput, GetBookEntryRequest}, get_entry::{GetEntryOutput, GetEntryRequest}, + get_entry_feed::{GetEntryFeedOutput, GetEntryFeedRequest}, + get_notebook_feed::{GetNotebookFeedOutput, GetNotebookFeedRequest}, resolve_entry::{ResolveEntryOutput, ResolveEntryRequest}, resolve_notebook::{ResolveNotebookOutput, ResolveNotebookRequest}, }; -use crate::clickhouse::ProfileRow; -use crate::endpoints::actor::resolve_actor; +use crate::clickhouse::{EntryRow, ProfileRow}; +use crate::endpoints::actor::{Viewer, resolve_actor}; use crate::endpoints::repo::XrpcErrorResponse; use crate::server::AppState; @@ -27,8 +31,12 @@ use crate::server::AppState; /// Resolves a notebook by actor + path/title, returns notebook with entries. pub async fn resolve_notebook( State(state): State, + ExtractOptionalServiceAuth(viewer): ExtractOptionalServiceAuth, ExtractXrpc(args): ExtractXrpc, ) -> Result>, XrpcErrorResponse> { + // viewer can be used later for viewer state (bookmarks, read status, etc.) + let _viewer: Viewer = viewer; + // Resolve actor to DID let did = resolve_actor(&state, &args.actor).await?; let did_str = did.as_str(); @@ -195,8 +203,11 @@ pub async fn resolve_notebook( /// Gets an entry by AT URI. pub async fn get_entry( State(state): State, + ExtractOptionalServiceAuth(viewer): ExtractOptionalServiceAuth, ExtractXrpc(args): ExtractXrpc, ) -> Result>, XrpcErrorResponse> { + let _viewer: Viewer = viewer; + // Parse the AT URI to extract authority and rkey let uri = &args.uri; let authority = uri.authority(); @@ -273,8 +284,11 @@ pub async fn get_entry( /// Resolves an entry by actor + notebook name + entry name. pub async fn resolve_entry( State(state): State, + ExtractOptionalServiceAuth(viewer): ExtractOptionalServiceAuth, ExtractXrpc(args): ExtractXrpc, ) -> Result>, XrpcErrorResponse> { + let _viewer: Viewer = viewer; + // Resolve actor to DID let did = resolve_actor(&state, &args.actor).await?; let did_str = did.as_str(); @@ -294,6 +308,7 @@ pub async fn resolve_entry( XrpcErrorResponse::internal_error("Database query failed") }) }, + // TODO: fix this, as we do need the entries to know for sure which, in case of collisions async { state .clickhouse @@ -522,3 +537,344 @@ fn profile_to_data_view( Ok(profile_data) } + +/// Parse cursor string to i64 timestamp millis +fn parse_cursor(cursor: Option<&str>) -> Result, XrpcErrorResponse> { + cursor + .map(|c| { + c.parse::() + .map_err(|_| XrpcErrorResponse::invalid_request("Invalid cursor format")) + }) + .transpose() +} + +/// Handle sh.weaver.notebook.getNotebookFeed +/// +/// Returns a global feed of notebooks. +pub async fn get_notebook_feed( + State(state): State, + ExtractOptionalServiceAuth(viewer): ExtractOptionalServiceAuth, + ExtractXrpc(args): ExtractXrpc, +) -> Result>, XrpcErrorResponse> { + let _viewer: Viewer = viewer; + + let limit = args.limit.unwrap_or(50).clamp(1, 100) as u32; + let cursor = parse_cursor(args.cursor.as_deref())?; + let algorithm = args.algorithm.as_deref().unwrap_or("chronological"); + + // Convert tags to &[&str] if present + let tags_vec: Vec<&str> = args + .tags + .as_ref() + .map(|t| t.iter().map(|s| s.as_ref()).collect()) + .unwrap_or_default(); + let tags = if tags_vec.is_empty() { + None + } else { + Some(tags_vec.as_slice()) + }; + + let notebook_rows = state + .clickhouse + .get_notebook_feed(algorithm, tags, limit + 1, cursor) + .await + .map_err(|e| { + tracing::error!("Failed to get notebook feed: {}", e); + XrpcErrorResponse::internal_error("Database query failed") + })?; + + // Check if there are more + let has_more = notebook_rows.len() > limit as usize; + let notebook_rows: Vec<_> = notebook_rows.into_iter().take(limit as usize).collect(); + + // Collect author DIDs for hydration + let mut all_author_dids: HashSet<&str> = HashSet::new(); + for nb in ¬ebook_rows { + for did in &nb.author_dids { + all_author_dids.insert(did.as_str()); + } + } + + // Batch fetch profiles + let author_dids_vec: Vec<&str> = all_author_dids.into_iter().collect(); + let profiles = state + .clickhouse + .get_profiles_batch(&author_dids_vec) + .await + .map_err(|e| { + tracing::error!("Failed to batch fetch profiles: {}", e); + XrpcErrorResponse::internal_error("Database query failed") + })?; + + let profile_map: HashMap<&str, &ProfileRow> = + profiles.iter().map(|p| (p.did.as_str(), p)).collect(); + + // Build NotebookViews + let mut notebooks: Vec> = Vec::with_capacity(notebook_rows.len()); + for nb_row in ¬ebook_rows { + let notebook_uri = AtUri::new(&nb_row.uri).map_err(|e| { + tracing::error!("Invalid notebook URI in db: {}", e); + XrpcErrorResponse::internal_error("Invalid URI stored") + })?; + + let notebook_cid = Cid::new(nb_row.cid.as_bytes()).map_err(|e| { + tracing::error!("Invalid notebook CID in db: {}", e); + XrpcErrorResponse::internal_error("Invalid CID stored") + })?; + + let authors = hydrate_authors(&nb_row.author_dids, &profile_map)?; + let record = parse_record_json(&nb_row.record)?; + + let notebook = NotebookView::new() + .uri(notebook_uri.into_static()) + .cid(notebook_cid.into_static()) + .authors(authors) + .record(record) + .indexed_at(nb_row.indexed_at.fixed_offset()) + .maybe_title(non_empty_cowstr(&nb_row.title)) + .maybe_path(non_empty_cowstr(&nb_row.path)) + .build(); + + notebooks.push(notebook); + } + + // Build cursor for pagination (created_at millis) + let next_cursor = if has_more { + notebook_rows + .last() + .map(|nb| nb.created_at.timestamp_millis().to_cowstr().into_static()) + } else { + None + }; + + Ok(Json( + GetNotebookFeedOutput { + notebooks, + cursor: next_cursor, + extra_data: None, + } + .into_static(), + )) +} + +/// Handle sh.weaver.notebook.getEntryFeed +/// +/// Returns a global feed of entries. +pub async fn get_entry_feed( + State(state): State, + ExtractOptionalServiceAuth(viewer): ExtractOptionalServiceAuth, + ExtractXrpc(args): ExtractXrpc, +) -> Result>, XrpcErrorResponse> { + let _viewer: Viewer = viewer; + + let limit = args.limit.unwrap_or(50).clamp(1, 100) as u32; + let cursor = parse_cursor(args.cursor.as_deref())?; + let algorithm = args.algorithm.as_deref().unwrap_or("chronological"); + + // Convert tags to &[&str] if present + let tags_vec: Vec<&str> = args + .tags + .as_ref() + .map(|t| t.iter().map(|s| s.as_ref()).collect()) + .unwrap_or_default(); + let tags = if tags_vec.is_empty() { + None + } else { + Some(tags_vec.as_slice()) + }; + + let entry_rows = state + .clickhouse + .get_entry_feed(algorithm, tags, limit + 1, cursor) + .await + .map_err(|e| { + tracing::error!("Failed to get entry feed: {}", e); + XrpcErrorResponse::internal_error("Database query failed") + })?; + + // Check if there are more + let has_more = entry_rows.len() > limit as usize; + let entry_rows: Vec<_> = entry_rows.into_iter().take(limit as usize).collect(); + + // Collect author DIDs for hydration + let mut all_author_dids: HashSet<&str> = HashSet::new(); + for entry in &entry_rows { + for did in &entry.author_dids { + all_author_dids.insert(did.as_str()); + } + } + + // Batch fetch profiles + let author_dids_vec: Vec<&str> = all_author_dids.into_iter().collect(); + let profiles = state + .clickhouse + .get_profiles_batch(&author_dids_vec) + .await + .map_err(|e| { + tracing::error!("Failed to batch fetch profiles: {}", e); + XrpcErrorResponse::internal_error("Database query failed") + })?; + + let profile_map: HashMap<&str, &ProfileRow> = + profiles.iter().map(|p| (p.did.as_str(), p)).collect(); + + // Build FeedEntryViews + let mut feed: Vec> = Vec::with_capacity(entry_rows.len()); + for entry_row in &entry_rows { + let entry_view = build_entry_view(entry_row, &profile_map)?; + + let feed_entry = FeedEntryView::new().entry(entry_view).build(); + + feed.push(feed_entry); + } + + // Build cursor for pagination (created_at millis) + let next_cursor = if has_more { + entry_rows + .last() + .map(|e| e.created_at.timestamp_millis().to_cowstr().into_static()) + } else { + None + }; + + Ok(Json( + GetEntryFeedOutput { + feed, + cursor: next_cursor, + extra_data: None, + } + .into_static(), + )) +} + +/// Handle sh.weaver.notebook.getBookEntry +/// +/// Returns an entry at a specific index within a notebook, with prev/next navigation. +pub async fn get_book_entry( + State(state): State, + ExtractOptionalServiceAuth(viewer): ExtractOptionalServiceAuth, + ExtractXrpc(args): ExtractXrpc, +) -> Result>, XrpcErrorResponse> { + let _viewer: Viewer = viewer; + + // Parse the notebook URI + let notebook_uri = &args.notebook; + let authority = notebook_uri.authority(); + let notebook_rkey = notebook_uri + .rkey() + .ok_or_else(|| XrpcErrorResponse::invalid_request("Notebook URI must include rkey"))?; + + // Resolve authority to DID + let notebook_did = resolve_actor(&state, authority).await?; + let notebook_did_str = notebook_did.as_str(); + let notebook_rkey_str = notebook_rkey.as_ref(); + + let index = args.index.unwrap_or(0).max(0) as u32; + + // Fetch entry at index with prev/next + let result = state + .clickhouse + .get_book_entry_at_index(notebook_did_str, notebook_rkey_str, index) + .await + .map_err(|e| { + tracing::error!("Failed to get book entry: {}", e); + XrpcErrorResponse::internal_error("Database query failed") + })?; + + let (current_row, prev_row, next_row) = + result.ok_or_else(|| XrpcErrorResponse::not_found("Entry not found at index"))?; + + // Collect all author DIDs for hydration + let mut all_author_dids: HashSet<&str> = HashSet::new(); + for did in ¤t_row.author_dids { + all_author_dids.insert(did.as_str()); + } + if let Some(ref prev) = prev_row { + for did in &prev.author_dids { + all_author_dids.insert(did.as_str()); + } + } + if let Some(ref next) = next_row { + for did in &next.author_dids { + all_author_dids.insert(did.as_str()); + } + } + + // Batch fetch profiles + let author_dids_vec: Vec<&str> = all_author_dids.into_iter().collect(); + let profiles = state + .clickhouse + .get_profiles_batch(&author_dids_vec) + .await + .map_err(|e| { + tracing::error!("Failed to fetch profiles: {}", e); + XrpcErrorResponse::internal_error("Database query failed") + })?; + + let profile_map: HashMap<&str, &ProfileRow> = + profiles.iter().map(|p| (p.did.as_str(), p)).collect(); + + // Build the current entry view + let entry_view = build_entry_view(¤t_row, &profile_map)?; + + // Build prev/next refs if present + let prev_ref = if let Some(ref prev) = prev_row { + let prev_view = build_entry_view(prev, &profile_map)?; + Some(BookEntryRef::new().entry(prev_view).build()) + } else { + None + }; + + let next_ref = if let Some(ref next) = next_row { + let next_view = build_entry_view(next, &profile_map)?; + Some(BookEntryRef::new().entry(next_view).build()) + } else { + None + }; + + let book_entry = BookEntryView::new() + .entry(entry_view) + .index(index as i64) + .maybe_prev(prev_ref) + .maybe_next(next_ref) + .build(); + + Ok(Json( + GetBookEntryOutput { + value: book_entry, + extra_data: None, + } + .into_static(), + )) +} + +/// Build an EntryView from an EntryRow +fn build_entry_view( + entry_row: &EntryRow, + profile_map: &HashMap<&str, &ProfileRow>, +) -> Result, XrpcErrorResponse> { + let entry_uri = AtUri::new(&entry_row.uri).map_err(|e| { + tracing::error!("Invalid entry URI in db: {}", e); + XrpcErrorResponse::internal_error("Invalid URI stored") + })?; + + let entry_cid = Cid::new(entry_row.cid.as_bytes()).map_err(|e| { + tracing::error!("Invalid entry CID in db: {}", e); + XrpcErrorResponse::internal_error("Invalid CID stored") + })?; + + let authors = hydrate_authors(&entry_row.author_dids, profile_map)?; + let record = parse_record_json(&entry_row.record)?; + + let entry_view = EntryView::new() + .uri(entry_uri.into_static()) + .cid(entry_cid.into_static()) + .authors(authors) + .record(record) + .indexed_at(entry_row.indexed_at.fixed_offset()) + .maybe_title(non_empty_cowstr(&entry_row.title)) + .maybe_path(non_empty_cowstr(&entry_row.path)) + .build(); + + Ok(entry_view) +} diff --git a/crates/weaver-index/src/lib.rs b/crates/weaver-index/src/lib.rs index 6bb66ee..d07dee8 100644 --- a/crates/weaver-index/src/lib.rs +++ b/crates/weaver-index/src/lib.rs @@ -6,6 +6,7 @@ pub mod firehose; pub mod indexer; pub mod parallel_tap; pub mod server; +pub mod service_identity; pub mod sqlite; pub mod tap; @@ -14,4 +15,5 @@ pub use error::{IndexError, Result}; pub use indexer::{FirehoseIndexer, load_cursor}; pub use parallel_tap::TapIndexer; pub use server::{AppState, ServerConfig}; +pub use service_identity::ServiceIdentity; pub use sqlite::{ShardKey, ShardRouter, SqliteShard}; diff --git a/crates/weaver-index/src/parallel_tap.rs b/crates/weaver-index/src/parallel_tap.rs index 93ac006..164d282 100644 --- a/crates/weaver-index/src/parallel_tap.rs +++ b/crates/weaver-index/src/parallel_tap.rs @@ -14,10 +14,10 @@ use crate::config::{IndexerConfig, TapConfig}; use crate::error::{ClickHouseError, Result}; use crate::tap::{TapConfig as TapConsumerConfig, TapConsumer, TapEvent}; -/// TAP indexer with multiple parallel websocket connections +/// Tap indexer with multiple parallel websocket connections /// -/// Each worker maintains its own websocket connection to TAP and its own -/// ClickHouse inserter. TAP distributes events across connected clients, +/// Each worker maintains its own websocket connection to Tap and its own +/// ClickHouse inserter. Tap distributes events across connected clients, /// and its ack-gating mechanism ensures per-DID ordering is preserved /// regardless of which worker handles which events. pub struct TapIndexer { @@ -297,8 +297,8 @@ async fn run_tap_worker( /// then runs INSERT queries to populate target tables for incremental MVs. async fn run_backfill(client: Arc) { // Wait for in-flight inserts to settle - info!("backfill: waiting 10s for in-flight inserts to settle"); - tokio::time::sleep(Duration::from_secs(10)).await; + info!("backfill: waiting 100s for in-flight inserts to settle"); + tokio::time::sleep(Duration::from_secs(100)).await; let mvs = Migrator::incremental_mvs(); if mvs.is_empty() { diff --git a/crates/weaver-index/src/server.rs b/crates/weaver-index/src/server.rs index 7c3631b..b9e90a1 100644 --- a/crates/weaver-index/src/server.rs +++ b/crates/weaver-index/src/server.rs @@ -5,18 +5,25 @@ use axum::{Json, Router, extract::State, http::StatusCode, response::IntoRespons use jacquard::api::com_atproto::repo::{ get_record::GetRecordRequest, list_records::ListRecordsRequest, }; -use weaver_api::sh_weaver::actor::get_profile::GetProfileRequest; -use weaver_api::sh_weaver::notebook::{ - get_entry::GetEntryRequest, - resolve_entry::ResolveEntryRequest, - resolve_notebook::ResolveNotebookRequest, -}; use jacquard::client::UnauthenticatedSession; use jacquard::identity::JacquardResolver; +use jacquard::types::did_doc::DidDocument; +use jacquard::types::string::Did; use jacquard_axum::IntoRouter; +use jacquard_axum::did_web::did_web_router; +use jacquard_axum::service_auth::ServiceAuth; use serde::Serialize; use tower_http::trace::TraceLayer; use tracing::info; +use weaver_api::sh_weaver::actor::{ + get_actor_entries::GetActorEntriesRequest, get_actor_notebooks::GetActorNotebooksRequest, + get_profile::GetProfileRequest, +}; +use weaver_api::sh_weaver::notebook::{ + get_book_entry::GetBookEntryRequest, get_entry::GetEntryRequest, + get_entry_feed::GetEntryFeedRequest, get_notebook_feed::GetNotebookFeedRequest, + resolve_entry::ResolveEntryRequest, resolve_notebook::ResolveNotebookRequest, +}; use crate::clickhouse::Client; use crate::config::ShardConfig; @@ -35,21 +42,41 @@ pub struct AppState { pub clickhouse: Arc, pub shards: Arc, pub resolver: Resolver, + /// Our service DID (expected audience for service auth JWTs) + pub service_did: Did<'static>, } impl AppState { - pub fn new(clickhouse: Client, shard_config: ShardConfig) -> Self { + pub fn new(clickhouse: Client, shard_config: ShardConfig, service_did: Did<'static>) -> Self { Self { clickhouse: Arc::new(clickhouse), shards: Arc::new(ShardRouter::new(shard_config.base_path)), resolver: UnauthenticatedSession::new_slingshot(), + service_did, } } } +impl ServiceAuth for AppState { + type Resolver = UnauthenticatedSession; + + fn service_did(&self) -> &Did<'_> { + &self.service_did + } + + fn resolver(&self) -> &Self::Resolver { + &self.resolver + } + + fn require_lxm(&self) -> bool { + true + } +} + /// Build the axum router with all XRPC endpoints -pub fn router(state: AppState) -> Router { +pub fn router(state: AppState, did_doc: DidDocument<'static>) -> Router { Router::new() + // did:web document .route("/xrpc/_health", get(health)) .route("/metrics", get(metrics)) // com.atproto.repo.* endpoints (record cache) @@ -57,12 +84,26 @@ pub fn router(state: AppState) -> Router { .merge(ListRecordsRequest::into_router(repo::list_records)) // sh.weaver.actor.* endpoints .merge(GetProfileRequest::into_router(actor::get_profile)) + .merge(GetActorNotebooksRequest::into_router( + actor::get_actor_notebooks, + )) + .merge(GetActorEntriesRequest::into_router( + actor::get_actor_entries, + )) // sh.weaver.notebook.* endpoints - .merge(ResolveNotebookRequest::into_router(notebook::resolve_notebook)) + .merge(ResolveNotebookRequest::into_router( + notebook::resolve_notebook, + )) .merge(GetEntryRequest::into_router(notebook::get_entry)) .merge(ResolveEntryRequest::into_router(notebook::resolve_entry)) + .merge(GetNotebookFeedRequest::into_router( + notebook::get_notebook_feed, + )) + .merge(GetEntryFeedRequest::into_router(notebook::get_entry_feed)) + .merge(GetBookEntryRequest::into_router(notebook::get_book_entry)) .layer(TraceLayer::new_for_http()) .with_state(state) + .merge(did_web_router(did_doc)) } /// Prometheus metrics endpoint @@ -105,6 +146,8 @@ async fn health(State(state): State) -> impl IntoResponse { pub struct ServerConfig { pub host: String, pub port: u16, + /// Service DID for this indexer (used as expected audience for service auth) + pub service_did: Did<'static>, } impl Default for ServerConfig { @@ -112,6 +155,8 @@ impl Default for ServerConfig { Self { host: "0.0.0.0".to_string(), port: 3000, + // Default to a placeholder - should be overridden in production + service_did: Did::new_static("did:web:index.weaver.sh").unwrap(), } } } @@ -123,8 +168,16 @@ impl ServerConfig { .ok() .and_then(|s| s.parse().ok()) .unwrap_or(3000); + let service_did = std::env::var("SERVICE_DID") + .ok() + .and_then(|s| Did::new_owned(s).ok()) + .unwrap_or_else(|| Did::new_static("did:web:index.weaver.sh").unwrap()); - Self { host, port } + Self { + host, + port, + service_did, + } } pub fn addr(&self) -> SocketAddr { @@ -135,9 +188,13 @@ impl ServerConfig { } /// Run the HTTP server -pub async fn run(state: AppState, config: ServerConfig) -> Result<(), IndexError> { +pub async fn run( + state: AppState, + config: ServerConfig, + did_doc: DidDocument<'static>, +) -> Result<(), IndexError> { let addr = config.addr(); - let app = router(state); + let app = router(state, did_doc); info!("Starting HTTP server on {}", addr); diff --git a/crates/weaver-index/src/service_identity.rs b/crates/weaver-index/src/service_identity.rs new file mode 100644 index 0000000..57f27b0 --- /dev/null +++ b/crates/weaver-index/src/service_identity.rs @@ -0,0 +1,173 @@ +//! Service identity management +//! +//! Handles keypair generation, persistence, and DID document creation +//! for service authentication. + +use std::path::Path; + +use jacquard::from_json_value; +use jacquard::types::crypto::multikey; +use jacquard::types::did_doc::DidDocument; +use jacquard::types::string::Did; +use k256::ecdsa::SigningKey; +use miette::{IntoDiagnostic, Result, WrapErr}; +use serde_json::json; +use tracing::info; + +/// Service identity containing the signing keypair +pub struct ServiceIdentity { + signing_key: SigningKey, + public_key_multibase: String, +} + +impl ServiceIdentity { + /// Load or generate a service identity keypair + /// + /// If the key file exists, loads it. Otherwise generates a new keypair + /// and saves it to the specified path. + pub fn load_or_generate(key_path: &Path) -> Result { + if key_path.exists() { + Self::load(key_path) + } else { + let identity = Self::generate()?; + identity.save(key_path)?; + Ok(identity) + } + } + + /// Generate a new random keypair + pub fn generate() -> Result { + info!("generating new service identity keypair"); + let signing_key = SigningKey::random(&mut rand::thread_rng()); + let public_key_multibase = Self::encode_public_key(&signing_key); + Ok(Self { + signing_key, + public_key_multibase, + }) + } + + /// Load keypair from file + fn load(key_path: &Path) -> Result { + info!(?key_path, "loading service identity keypair"); + let key_bytes = std::fs::read(key_path) + .into_diagnostic() + .wrap_err("failed to read service key file")?; + + if key_bytes.len() != 32 { + miette::bail!( + "invalid key file: expected 32 bytes, got {}", + key_bytes.len() + ); + } + + let signing_key = SigningKey::from_slice(&key_bytes) + .map_err(|e| miette::miette!("invalid key data: {}", e))?; + let public_key_multibase = Self::encode_public_key(&signing_key); + + Ok(Self { + signing_key, + public_key_multibase, + }) + } + + /// Save keypair to file + fn save(&self, key_path: &Path) -> Result<()> { + info!(?key_path, "saving service identity keypair"); + + // Ensure parent directory exists + if let Some(parent) = key_path.parent() { + std::fs::create_dir_all(parent) + .into_diagnostic() + .wrap_err("failed to create key directory")?; + } + + let key_bytes = self.signing_key.to_bytes(); + std::fs::write(key_path, key_bytes.to_vec()) + .into_diagnostic() + .wrap_err("failed to write service key file")?; + + // Set restrictive permissions on Unix + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + let perms = std::fs::Permissions::from_mode(0o600); + std::fs::set_permissions(key_path, perms) + .into_diagnostic() + .wrap_err("failed to set key file permissions")?; + } + + Ok(()) + } + + /// Encode the public key as a multikey string + fn encode_public_key(signing_key: &SigningKey) -> String { + let verifying_key = signing_key.verifying_key(); + let point = verifying_key.to_encoded_point(true); // compressed + let bytes = point.as_bytes(); + // 0xE7 is the multicodec for secp256k1-pub + multikey(0xE7, bytes) + } + + /// Get the public key multibase string + pub fn public_key_multibase(&self) -> &str { + &self.public_key_multibase + } + + /// Get the signing key (for signing JWTs) + pub fn signing_key(&self) -> &SigningKey { + &self.signing_key + } + + /// Build a DID document for this service identity + pub fn did_document(&self, service_did: &Did<'_>) -> DidDocument<'static> { + let did_str = service_did.as_str(); + + let doc = json!({ + "@context": [ + "https://www.w3.org/ns/did/v1", + "https://w3id.org/security/multikey/v1", + "https://w3id.org/security/suites/secp256k1-2019/v1" + ], + "id": did_str, + "verificationMethod": [{ + "id": format!("{}#atproto", did_str), + "type": "Multikey", + "controller": did_str, + "publicKeyMultibase": self.public_key_multibase + }] + }); + + from_json_value::(doc).expect("valid DID document") + } + + /// Build a DID document with a service endpoint + pub fn did_document_with_service( + &self, + service_did: &Did<'_>, + service_endpoint: &str, + ) -> DidDocument<'static> { + let did_str = service_did.as_str(); + + let doc = json!({ + "@context": [ + "https://www.w3.org/ns/did/v1", + "https://w3id.org/security/multikey/v1", + "https://w3id.org/security/suites/secp256k1-2019/v1" + ], + "id": did_str, + "verificationMethod": [{ + "id": format!("{}#atproto", did_str), + "type": "Multikey", + "controller": did_str, + "publicKeyMultibase": self.public_key_multibase + }], + "service": [{ + "id": "#atproto_index", + "type": "AtprotoRecordIndex", + "serviceEndpoint": service_endpoint + }] + }); + + from_json_value::(doc).expect("valid DID document") + } +} diff --git a/lexicons/notebook/resolveNotebook.json b/lexicons/notebook/resolveNotebook.json index 701d144..2aaa16f 100644 --- a/lexicons/notebook/resolveNotebook.json +++ b/lexicons/notebook/resolveNotebook.json @@ -30,9 +30,7 @@ } } }, - "errors": [ - { "name": "NotebookNotFound" } - ] + "errors": [{ "name": "NotebookNotFound" }] } } } diff --git a/scripts/test-indexer.sh b/scripts/test-indexer.sh index 65c4320..c25691c 100755 --- a/scripts/test-indexer.sh +++ b/scripts/test-indexer.sh @@ -56,6 +56,54 @@ urlencode() { python3 -c "import urllib.parse; print(urllib.parse.quote('$1', safe=''))" } +# Get actor notebooks +test_actor_notebooks() { + local actor="${1:-$DID}" + local limit="${2:-10}" + info "Testing sh.weaver.actor.getActorNotebooks (actor=${actor}, limit=${limit})..." + curl -s "${BASE_URL}/xrpc/sh.weaver.actor.getActorNotebooks?actor=${actor}&limit=${limit}" | jq . +} + +# Get actor notebooks with cursor +test_actor_notebooks_cursor() { + local cursor="$1" + local actor="${2:-$DID}" + local limit="${3:-10}" + info "Testing sh.weaver.actor.getActorNotebooks with cursor..." + curl -s "${BASE_URL}/xrpc/sh.weaver.actor.getActorNotebooks?actor=${actor}&limit=${limit}&cursor=${cursor}" | jq . +} + +# Get actor entries +test_actor_entries() { + local actor="${1:-$DID}" + local limit="${2:-10}" + info "Testing sh.weaver.actor.getActorEntries (actor=${actor}, limit=${limit})..." + curl -s "${BASE_URL}/xrpc/sh.weaver.actor.getActorEntries?actor=${actor}&limit=${limit}" | jq . +} + +# Get notebook feed +test_notebook_feed() { + local limit="${1:-10}" + info "Testing sh.weaver.notebook.getNotebookFeed (limit=${limit})..." + curl -s "${BASE_URL}/xrpc/sh.weaver.notebook.getNotebookFeed?limit=${limit}" | jq . +} + +# Get entry feed +test_entry_feed() { + local limit="${1:-10}" + info "Testing sh.weaver.notebook.getEntryFeed (limit=${limit})..." + curl -s "${BASE_URL}/xrpc/sh.weaver.notebook.getEntryFeed?limit=${limit}" | jq . +} + +# Get book entry by index +test_book_entry() { + local notebook_rkey="${1:-weaver}" + local index="${2:-0}" + local notebook_uri="at://${DID}/sh.weaver.notebook.book/${notebook_rkey}" + info "Testing sh.weaver.notebook.getBookEntry (notebook=${notebook_uri}, index=${index})..." + curl -s "${BASE_URL}/xrpc/sh.weaver.notebook.getBookEntry?notebook=$(urlencode "${notebook_uri}")&index=${index}" | jq . +} + # Test all entry rkeys test_all_entries() { local rkeys=( @@ -91,6 +139,16 @@ test_all() { test_resolve_entry "weaver" "drafts_privacy" echo test_get_entry "3m7tg3ni77tqx" + echo + test_actor_notebooks + echo + test_actor_entries + echo + test_notebook_feed + echo + test_entry_feed + echo + test_book_entry "3m4rbphheug2b" 0 } # Main @@ -113,11 +171,26 @@ case "${1:-all}" in entries) test_all_entries ;; + actor-notebooks) + test_actor_notebooks "${2:-$DID}" "${3:-10}" + ;; + actor-entries) + test_actor_entries "${2:-$DID}" "${3:-10}" + ;; + notebook-feed) + test_notebook_feed "${2:-10}" + ;; + entry-feed) + test_entry_feed "${2:-10}" + ;; + book-entry) + test_book_entry "${2:-3m4rbphheug2b}" "${3:-0}" + ;; all) test_all ;; *) - echo "Usage: $0 {health|profile|notebook [name]|entry [rkey]|resolve [notebook] [entry]|entries|all}" + echo "Usage: $0 {health|profile|notebook [name]|entry [rkey]|resolve [notebook] [entry]|entries|actor-notebooks [actor] [limit]|actor-entries [actor] [limit]|notebook-feed [limit]|entry-feed [limit]|book-entry [notebook] [index]|all}" echo echo "Environment:" echo " INDEXER_URL Base URL (default: http://localhost:3000)" -- knotmirror 0.0.1