diff --git a/crates/weaver-app/.env-dev b/crates/weaver-app/.env-dev deleted file mode 100644 index 5d82bc4..0000000 --- a/crates/weaver-app/.env-dev +++ /dev/null @@ -1,10 +0,0 @@ -WEAVER_APP_ENV="dev" -WEAVER_APP_HOST="http://localhost" -WEAVER_APP_DOMAIN="" -WEAVER_PORT=8080 -WEAVER_APP_SCOPES="atproto transition:generic" -WEAVER_CLIENT_NAME="Weaver" - -WEAVER_LOGO_URI="" -WEAVER_TOS_URI="" -WEAVER_PRIVACY_POLICY_URI="" diff --git a/crates/weaver-app/.env-prod b/crates/weaver-app/.env-prod new file mode 100644 index 0000000..9b544f2 --- /dev/null +++ b/crates/weaver-app/.env-prod @@ -0,0 +1,13 @@ +WEAVER_APP_ENV="prod" +WEAVER_APP_HOST="https://alpha.weaver.sh" +WEAVER_APP_DOMAIN="https://alpha.weaver.sh" +WEAVER_PORT=8080 +WEAVER_APP_SCOPES="atproto transition:generic" +WEAVER_CLIENT_NAME="Weaver" + +WEAVER_LOGO_URI="https://alpha.weaver.sh/favicon.ico" +WEAVER_TOS_URI="https://alpha.weaver.sh/tos" +WEAVER_PRIVACY_POLICY_URI="https://alpha.weaver.sh/privacy" +WEAVER_OWNER_DID="did:plc:yfvwmnlztr4dwkb7hwz55r2g" +WEAVER_INDEXER_URL="https://index.weaver.sh" +WEAVER_INDEXER_DID="did:web:index.weaver.sh" diff --git a/crates/weaver-app/Cargo.toml b/crates/weaver-app/Cargo.toml index a6e09cf..caeb363 100644 --- a/crates/weaver-app/Cargo.toml +++ b/crates/weaver-app/Cargo.toml @@ -23,11 +23,12 @@ path = "src/bin/embed_worker.rs" required-features = ["web"] [features] -default = ["web", "fullstack-server", "no-app-index"] +default = ["web", "fullstack-server", "use-index"] # Fullstack mode with SSR and server functions fullstack-server = ["dioxus/fullstack"] wasm-split = ["dioxus/wasm-split"] -no-app-index = [] +# Use weaver-index for reads (proxied when authenticated) +use-index = ["weaver-common/use-index"] web = ["dioxus/web", "dioxus-primitives/web"] desktop = ["dioxus/desktop"] diff --git a/crates/weaver-app/build.rs b/crates/weaver-app/build.rs index ff6b3f4..e9856f3 100644 --- a/crates/weaver-app/build.rs +++ b/crates/weaver-app/build.rs @@ -11,6 +11,10 @@ fn main() { dotenv().ok(); f.write_all(b"// This file is automatically generated by build.rs\n\n") .unwrap(); + + // Track which keys we've written + let mut written_keys = std::collections::HashSet::new(); + for (key, value) in env::vars() { if key.starts_with("WEAVER_") { let line = format!( @@ -19,6 +23,15 @@ fn main() { value.replace("\"", "\\\"") ); f.write_all(line.as_bytes()).unwrap(); + written_keys.insert(key); + } + } + + // Ensure index-related constants are always defined (even if empty) + for key in ["WEAVER_INDEXER_URL", "WEAVER_INDEXER_DID"] { + if !written_keys.contains(key) { + let line = format!("#[allow(unused)]\npub const {}: &'static str = \"\";\n", key); + f.write_all(line.as_bytes()).unwrap(); } } } diff --git a/crates/weaver-app/src/env.rs b/crates/weaver-app/src/env.rs index aa0c24e..f3c3e2f 100644 --- a/crates/weaver-app/src/env.rs +++ b/crates/weaver-app/src/env.rs @@ -1,11 +1,11 @@ // This file is automatically generated by build.rs #[allow(unused)] -pub const WEAVER_APP_ENV: &'static str = "prod"; +pub const WEAVER_APP_ENV: &'static str = "dev"; #[allow(unused)] -pub const WEAVER_APP_HOST: &'static str = "https://alpha.weaver.sh"; +pub const WEAVER_APP_HOST: &'static str = "http://localhost"; #[allow(unused)] -pub const WEAVER_APP_DOMAIN: &'static str = "https://alpha.weaver.sh"; +pub const WEAVER_APP_DOMAIN: &'static str = ""; #[allow(unused)] pub const WEAVER_PORT: &'static str = "8080"; #[allow(unused)] @@ -13,10 +13,14 @@ pub const WEAVER_APP_SCOPES: &'static str = "atproto transition:generic"; #[allow(unused)] pub const WEAVER_CLIENT_NAME: &'static str = "Weaver"; #[allow(unused)] -pub const WEAVER_LOGO_URI: &'static str = "https://alpha.weaver.sh/favicon.ico"; +pub const WEAVER_LOGO_URI: &'static str = ""; #[allow(unused)] -pub const WEAVER_TOS_URI: &'static str = "https://alpha.weaver.sh/tos"; +pub const WEAVER_TOS_URI: &'static str = ""; #[allow(unused)] -pub const WEAVER_PRIVACY_POLICY_URI: &'static str = "https://alpha.weaver.sh/privacy"; +pub const WEAVER_PRIVACY_POLICY_URI: &'static str = ""; +#[allow(unused)] +pub const WEAVER_INDEXER_URL: &'static str = "http://localhost:3000"; +#[allow(unused)] +pub const WEAVER_INDEXER_DID: &'static str = "did:web:index.weaver.sh"; #[allow(unused)] pub const WEAVER_OWNER_DID: &'static str = "did:plc:yfvwmnlztr4dwkb7hwz55r2g"; diff --git a/crates/weaver-app/src/fetch.rs b/crates/weaver-app/src/fetch.rs index dea0d6b..d650b82 100644 --- a/crates/weaver-app/src/fetch.rs +++ b/crates/weaver-app/src/fetch.rs @@ -24,7 +24,10 @@ use jacquard::types::string::Handle; use jacquard::types::string::Nsid; use jacquard::xrpc::XrpcResponse; use jacquard::xrpc::*; -use jacquard::{smol_str::{SmolStr, format_smolstr}, types::ident::AtIdentifier}; +use jacquard::{ + smol_str::{SmolStr, format_smolstr}, + types::ident::AtIdentifier, +}; use serde::{Deserialize, Serialize}; use std::future::Future; use std::{sync::Arc, time::Duration}; @@ -112,6 +115,11 @@ impl XrpcClient for Client { if let Some(session) = guard.clone() { session.base_uri().await } else { + // When unauthenticated, use index if configured + #[cfg(feature = "use-index")] + if !crate::env::WEAVER_INDEXER_URL.is_empty() { + return CowStr::from(crate::env::WEAVER_INDEXER_URL); + } self.oauth_client.base_uri().await } } @@ -392,8 +400,19 @@ impl Fetcher { &self, session: OAuthSession, ) { + let agent = Arc::new(Agent::new(session)); + + // When use-index is enabled, set the atproto_proxy header for service proxying + #[cfg(feature = "use-index")] + if !crate::env::WEAVER_INDEXER_DID.is_empty() { + let proxy_value = format!("{}#atproto_index", crate::env::WEAVER_INDEXER_DID); + let mut opts = agent.opts().await; + opts.atproto_proxy = Some(CowStr::from(proxy_value)); + agent.set_opts(opts).await; + } + let mut session_slot = self.client.session.write().await; - *session_slot = Some(Arc::new(Agent::new(session))); + *session_slot = Some(agent); } pub async fn downgrade_to_unauthenticated(&self) { @@ -508,6 +527,63 @@ impl Fetcher { } } + #[cfg(feature = "use-index")] + pub async fn fetch_notebooks_from_ufos( + &self, + ) -> Result, Vec>)>>> { + use weaver_api::sh_weaver::notebook::book::Book; + use weaver_api::sh_weaver::notebook::get_notebook_feed::GetNotebookFeed; + + let client = self.get_client(); + + let resp = client + .send(GetNotebookFeed::new().limit(100).build()) + .await + .map_err(|e| dioxus::CapturedError::from_display(e))?; + + let output = resp + .into_output() + .map_err(|e| dioxus::CapturedError::from_display(e))?; + + let mut notebooks = Vec::new(); + + for notebook in output.notebooks { + // Extract entry_list from the record + let book: Book = jacquard::from_data(¬ebook.record) + .map_err(|e| dioxus::CapturedError::from_display(e))?; + let book = book.into_static(); + + let entries: Vec> = book + .entry_list + .into_iter() + .map(IntoStatic::into_static) + .collect(); + + let ident = notebook.uri.authority().clone().into_static(); + let title = notebook + .title + .as_ref() + .map(|t| SmolStr::new(t.as_ref())) + .unwrap_or_else(|| SmolStr::new("Untitled")); + + let result = Arc::new((notebook.into_static(), entries)); + #[cfg(feature = "server")] + { + cache_impl::insert(&self.notebook_key_cache, title.clone(), ident.clone()); + cache_impl::insert(&self.book_cache, (ident.clone(), title), result.clone()); + if let Some(path) = result.0.path.as_ref() { + let path: SmolStr = path.as_ref().into(); + cache_impl::insert(&self.notebook_key_cache, path.clone(), ident.clone()); + cache_impl::insert(&self.book_cache, (ident, path), result.clone()); + } + } + notebooks.push(result); + } + + Ok(notebooks) + } + + #[cfg(not(feature = "use-index"))] pub async fn fetch_notebooks_from_ufos( &self, ) -> Result, Vec>)>>> { @@ -530,10 +606,13 @@ impl Fetcher { // Construct URI let uri_str = format_smolstr!( "at://{}/{}/{}", - ufos_record.did, ufos_record.collection, ufos_record.rkey + ufos_record.did, + ufos_record.collection, + ufos_record.rkey ); - let uri = AtUri::new_owned(uri_str) - .map_err(|e| dioxus::CapturedError::from_display(format_smolstr!("Invalid URI: {}", e).as_str()))?; + let uri = AtUri::new_owned(uri_str).map_err(|e| { + dioxus::CapturedError::from_display(format_smolstr!("Invalid URI: {}", e).as_str()) + })?; match client.view_notebook(&uri).await { Ok((notebook, entries)) => { let ident = uri.authority().clone().into_static(); @@ -573,7 +652,47 @@ impl Fetcher { Ok(notebooks) } + /// Fetch entries from index feed (reverse chronological) + #[cfg(feature = "use-index")] + pub async fn fetch_entries_from_ufos( + &self, + ) -> Result, Entry<'static>, u64)>>> { + use jacquard::IntoStatic; + use weaver_api::sh_weaver::notebook::entry::Entry; + use weaver_api::sh_weaver::notebook::get_entry_feed::GetEntryFeed; + + let client = self.get_client(); + + let resp = client + .send(GetEntryFeed::new().limit(100).build()) + .await + .map_err(|e| dioxus::CapturedError::from_display(e))?; + + let output = resp + .into_output() + .map_err(|e| dioxus::CapturedError::from_display(e))?; + + let mut entries = Vec::new(); + + for feed_entry in output.feed { + let entry_view = feed_entry.entry; + // indexed_at is ISO datetime, parse to get millisecond timestamp + let timestamp = chrono::DateTime::parse_from_rfc3339(entry_view.indexed_at.as_str()) + .map(|dt| dt.timestamp_millis() as u64) + .unwrap_or(0); + + let entry: Entry = jacquard::from_data(&entry_view.record) + .map_err(|e| dioxus::CapturedError::from_display(e))?; + let entry = entry.into_static(); + + entries.push(Arc::new((entry_view.into_static(), entry, timestamp))); + } + + Ok(entries) + } + /// Fetch entries from UFOS discovery service (reverse chronological) + #[cfg(not(feature = "use-index"))] pub async fn fetch_entries_from_ufos( &self, ) -> Result, Entry<'static>, u64)>>> { @@ -630,6 +749,81 @@ impl Fetcher { Ok(entries) } + #[cfg(feature = "use-index")] + pub async fn fetch_notebooks_for_did( + &self, + ident: &AtIdentifier<'_>, + ) -> Result, Vec>)>>> { + use weaver_api::sh_weaver::actor::get_actor_notebooks::GetActorNotebooks; + use weaver_api::sh_weaver::notebook::book::Book; + + let client = self.get_client(); + + let resp = client + .send( + GetActorNotebooks::new() + .actor(ident.clone()) + .limit(100) + .build(), + ) + .await + .map_err(|e| dioxus::CapturedError::from_display(e))?; + + let output = resp + .into_output() + .map_err(|e| dioxus::CapturedError::from_display(e))?; + + let mut notebooks = Vec::new(); + + for notebook in output.notebooks { + // Extract entry_list from the record + let book: Book = jacquard::from_data(¬ebook.record) + .map_err(|e| dioxus::CapturedError::from_display(e))?; + let book = book.into_static(); + + let entries: Vec> = book + .entry_list + .into_iter() + .map(IntoStatic::into_static) + .collect(); + + let ident_static = notebook.uri.authority().clone().into_static(); + let title = notebook + .title + .as_ref() + .map(|t| SmolStr::new(t.as_ref())) + .unwrap_or_else(|| SmolStr::new("Untitled")); + + let result = Arc::new((notebook.into_static(), entries)); + #[cfg(feature = "server")] + { + cache_impl::insert( + &self.notebook_key_cache, + title.clone(), + ident_static.clone(), + ); + cache_impl::insert( + &self.book_cache, + (ident_static.clone(), title), + result.clone(), + ); + if let Some(path) = result.0.path.as_ref() { + let path: SmolStr = path.as_ref().into(); + cache_impl::insert( + &self.notebook_key_cache, + path.clone(), + ident_static.clone(), + ); + cache_impl::insert(&self.book_cache, (ident_static, path), result.clone()); + } + } + notebooks.push(result); + } + + Ok(notebooks) + } + + #[cfg(not(feature = "use-index"))] pub async fn fetch_notebooks_for_did( &self, ident: &AtIdentifier<'_>, @@ -728,7 +922,11 @@ impl Fetcher { notebooks.push(result); } Err(e) => { - tracing::warn!("fetch_notebooks_for_did: view_notebook failed for {}: {}", record.uri, e); + tracing::warn!( + "fetch_notebooks_for_did: view_notebook failed for {}: {}", + record.uri, + e + ); continue; } } @@ -738,6 +936,45 @@ impl Fetcher { } /// Fetch all entries for a DID (for profile timeline) + #[cfg(feature = "use-index")] + pub async fn fetch_entries_for_did( + &self, + ident: &AtIdentifier<'_>, + ) -> Result, Entry<'static>)>>> { + use weaver_api::sh_weaver::actor::get_actor_entries::GetActorEntries; + + let client = self.get_client(); + + let resp = client + .send( + GetActorEntries::new() + .actor(ident.clone()) + .limit(100) + .build(), + ) + .await + .map_err(|e| dioxus::CapturedError::from_display(e))?; + + let output = resp + .into_output() + .map_err(|e| dioxus::CapturedError::from_display(e))?; + + let mut entries = Vec::new(); + + for entry_view in output.entries { + // Deserialize Entry from the record field + let entry: Entry = jacquard::from_data(&entry_view.record) + .map_err(|e| dioxus::CapturedError::from_display(e))?; + let entry = entry.into_static(); + + entries.push(Arc::new((entry_view.into_static(), entry))); + } + + Ok(entries) + } + + /// Fetch all entries for a DID (for profile timeline) + #[cfg(not(feature = "use-index"))] pub async fn fetch_entries_for_did( &self, ident: &AtIdentifier<'_>, @@ -922,7 +1159,9 @@ impl Fetcher { // Try to find notebook context via constellation let entry_uri = entry_view.uri.clone(); let at_uri = AtUri::new(entry_uri.as_ref()).map_err(|e| { - dioxus::CapturedError::from_display(format_smolstr!("Invalid entry URI: {}", e).as_str()) + dioxus::CapturedError::from_display( + format_smolstr!("Invalid entry URI: {}", e).as_str(), + ) })?; let (total, first_notebook) = client @@ -941,7 +1180,9 @@ impl Fetcher { notebook_id.rkey.0.as_str() ); let notebook_uri = AtUri::new_owned(notebook_uri_str).map_err(|e| { - dioxus::CapturedError::from_display(format_smolstr!("Invalid notebook URI: {}", e).as_str()) + dioxus::CapturedError::from_display( + format_smolstr!("Invalid notebook URI: {}", e).as_str(), + ) })?; // Fetch notebook and find entry position @@ -1032,7 +1273,9 @@ impl Fetcher { // Check if entry is in multiple notebooks - if so, clear prev/next let entry_uri = book_entry_view.entry.uri.clone(); let at_uri = AtUri::new(entry_uri.as_ref()).map_err(|e| { - dioxus::CapturedError::from_display(format_smolstr!("Invalid entry URI: {}", e).as_str()) + dioxus::CapturedError::from_display( + format_smolstr!("Invalid entry URI: {}", e).as_str(), + ) })?; let (total, _) = client diff --git a/crates/weaver-common/Cargo.toml b/crates/weaver-common/Cargo.toml index 5cc22d8..e98ba99 100644 --- a/crates/weaver-common/Cargo.toml +++ b/crates/weaver-common/Cargo.toml @@ -9,6 +9,7 @@ publish = false default = ["dev"] dev = [] native = ["jacquard/dns"] +use-index = [] iroh = ["dep:iroh", "dep:iroh-gossip", "dep:iroh-tickets"] telemetry = ["dep:metrics", "dep:metrics-exporter-prometheus", "dep:tracing-subscriber", "dep:tracing-loki"] diff --git a/crates/weaver-common/src/agent.rs b/crates/weaver-common/src/agent.rs index a4eae6f..05686d3 100644 --- a/crates/weaver-common/src/agent.rs +++ b/crates/weaver-common/src/agent.rs @@ -359,6 +359,41 @@ pub trait WeaverExt: AgentSessionExt + XrpcExt + Send + Sync + Sized { /// View functions - generic versions that work with any Agent /// Fetch a notebook and construct NotebookView with author profiles + #[cfg(feature = "use-index")] + fn view_notebook( + &self, + uri: &AtUri<'_>, + ) -> impl Future, Vec>), WeaverError>> + where + Self: Sized, + { + async move { + use weaver_api::sh_weaver::notebook::get_notebook::GetNotebook; + + let resp = self + .send(GetNotebook::new().notebook(uri.clone()).build()) + .await + .map_err(|e| AgentError::from(ClientError::from(e)))?; + + let output = resp.into_output().map_err(|e| { + AgentError::from(ClientError::invalid_request(format!( + "Failed to get notebook: {}", + e + ))) + })?; + + Ok(( + output.notebook.into_static(), + output + .entries + .into_iter() + .map(IntoStatic::into_static) + .collect(), + )) + } + } + + #[cfg(not(feature = "use-index"))] fn view_notebook( &self, uri: &AtUri<'_>, @@ -608,6 +643,54 @@ pub trait WeaverExt: AgentSessionExt + XrpcExt + Send + Sync + Sized { } /// Search for a notebook by title for a given DID or handle + #[cfg(feature = "use-index")] + fn notebook_by_title( + &self, + ident: &jacquard::types::ident::AtIdentifier<'_>, + title: &str, + ) -> impl Future< + Output = Result, Vec>)>, WeaverError>, + > + where + Self: Sized, + { + async move { + use weaver_api::sh_weaver::notebook::resolve_notebook::ResolveNotebook; + + let resp = self + .send( + ResolveNotebook::new() + .actor(ident.clone()) + .name(title) + .build(), + ) + .await + .map_err(|e| AgentError::from(ClientError::from(e)))?; + + match resp.into_output() { + Ok(output) => { + // Extract StrongRefs from the BookEntryViews for compatibility + let entries: Vec> = output + .entries + .iter() + .map(|bev| { + StrongRef::new() + .uri(bev.entry.uri.clone()) + .cid(bev.entry.cid.clone()) + .build() + .into_static() + }) + .collect(); + + Ok(Some((output.notebook.into_static(), entries))) + } + Err(_) => Ok(None), + } + } + } + + /// Search for a notebook by title for a given DID or handle + #[cfg(not(feature = "use-index"))] fn notebook_by_title( &self, ident: &jacquard::types::ident::AtIdentifier<'_>, @@ -740,6 +823,41 @@ pub trait WeaverExt: AgentSessionExt + XrpcExt + Send + Sync + Sized { } /// Hydrate a profile view from either weaver or bsky profile + #[cfg(feature = "use-index")] + fn hydrate_profile_view( + &self, + did: &Did<'_>, + ) -> impl Future< + Output = Result< + ( + Option>, + weaver_api::sh_weaver::actor::ProfileDataView<'static>, + ), + WeaverError, + >, + > { + async move { + use weaver_api::sh_weaver::actor::get_profile::GetProfile; + + let resp = self + .send(GetProfile::new().actor(did.clone()).build()) + .await + .map_err(|e| AgentError::from(ClientError::from(e)))?; + + let output = resp.into_output().map_err(|e| { + AgentError::from(ClientError::invalid_request(format!( + "Failed to get profile: {}", + e + ))) + })?; + + // URI is goofy in this signature, just return None for now + Ok((None, output.value.into_static())) + } + } + + /// Hydrate a profile view from either weaver or bsky profile + #[cfg(not(feature = "use-index"))] fn hydrate_profile_view( &self, did: &Did<'_>, @@ -885,6 +1003,39 @@ pub trait WeaverExt: AgentSessionExt + XrpcExt + Send + Sync + Sized { } /// View an entry at a specific index with prev/next navigation + #[cfg(feature = "use-index")] + fn view_entry<'a>( + &self, + notebook: &NotebookView<'a>, + _entries: &[StrongRef<'_>], + index: usize, + ) -> impl Future, WeaverError>> { + async move { + use weaver_api::sh_weaver::notebook::get_book_entry::GetBookEntry; + + let resp = self + .send( + GetBookEntry::new() + .notebook(notebook.uri.clone()) + .index(index as i64) + .build(), + ) + .await + .map_err(|e| AgentError::from(ClientError::from(e)))?; + + let output = resp.into_output().map_err(|e| { + AgentError::from(ClientError::invalid_request(format!( + "Failed to get book entry: {}", + e + ))) + })?; + + Ok(output.value.into_static()) + } + } + + /// View an entry at a specific index with prev/next navigation + #[cfg(not(feature = "use-index"))] fn view_entry<'a>( &self, notebook: &NotebookView<'a>, @@ -1146,6 +1297,57 @@ pub trait WeaverExt: AgentSessionExt + XrpcExt + Send + Sync + Sized { /// /// This bypasses notebook context entirely - useful for standalone entries /// or when you have the rkey but not the notebook. + #[cfg(feature = "use-index")] + fn fetch_entry_by_rkey( + &self, + ident: &jacquard::types::ident::AtIdentifier<'_>, + rkey: &str, + ) -> impl Future, entry::Entry<'static>), WeaverError>> + where + Self: Sized, + { + async move { + use jacquard::types::collection::Collection; + use weaver_api::sh_weaver::notebook::get_entry::GetEntry; + + // Build entry URI from ident + rkey + let entry_uri_str = format!("at://{}/{}/{}", ident, entry::Entry::NSID, rkey); + let entry_uri = AtUri::new(&entry_uri_str) + .map_err(|_| AgentError::from(ClientError::invalid_request("Invalid entry URI")))? + .into_static(); + + let resp = self + .send(GetEntry::new().uri(entry_uri).build()) + .await + .map_err(|e| AgentError::from(ClientError::from(e)))?; + + let output = resp.into_output().map_err(|e| { + AgentError::from(ClientError::invalid_request(format!( + "Failed to get entry: {}", + e + ))) + })?; + + // Clone the record for deserialization so we can consume output.value + let record_clone = output.value.record.clone(); + + // Deserialize Entry from the cloned record + let entry_value: entry::Entry = jacquard::from_data(&record_clone).map_err(|e| { + AgentError::from(ClientError::invalid_request(format!( + "Failed to deserialize entry record: {}", + e + ))) + })?; + + Ok((output.value.into_static(), entry_value.into_static())) + } + } + + /// Fetch an entry directly by its rkey, returning the EntryView and raw Entry. + /// + /// This bypasses notebook context entirely - useful for standalone entries + /// or when you have the rkey but not the notebook. + #[cfg(not(feature = "use-index"))] fn fetch_entry_by_rkey( &self, ident: &jacquard::types::ident::AtIdentifier<'_>, diff --git a/crates/weaver-index/migrations/clickhouse/025_edit_nodes.sql b/crates/weaver-index/migrations/clickhouse/025_edit_nodes.sql index d1090f3..bb628ee 100644 --- a/crates/weaver-index/migrations/clickhouse/025_edit_nodes.sql +++ b/crates/weaver-index/migrations/clickhouse/025_edit_nodes.sql @@ -22,13 +22,15 @@ CREATE TABLE IF NOT EXISTS edit_nodes ( resource_rkey String DEFAULT '', resource_collection LowCardinality(String) DEFAULT '', - -- For diffs: reference to root + -- For diffs: reference to root (StrongRef) root_did String DEFAULT '', root_rkey String DEFAULT '', + root_cid String DEFAULT '', - -- For diffs: reference to previous node + -- For diffs: reference to previous node (StrongRef) prev_did String DEFAULT '', prev_rkey String DEFAULT '', + prev_cid String DEFAULT '', -- Whether this has inline diff data vs blob snapshot has_inline_diff UInt8 DEFAULT 0, diff --git a/crates/weaver-index/migrations/clickhouse/026_edit_roots_mv.sql b/crates/weaver-index/migrations/clickhouse/026_edit_roots_mv.sql index 12d6d5b..5b6e1b8 100644 --- a/crates/weaver-index/migrations/clickhouse/026_edit_roots_mv.sql +++ b/crates/weaver-index/migrations/clickhouse/026_edit_roots_mv.sql @@ -47,8 +47,10 @@ SELECT -- Roots don't have root/prev refs '' as root_did, '' as root_rkey, + '' as root_cid, '' as prev_did, '' as prev_rkey, + '' as prev_cid, -- Roots always have snapshot 0 as has_inline_diff, diff --git a/crates/weaver-index/migrations/clickhouse/027_edit_diffs_mv.sql b/crates/weaver-index/migrations/clickhouse/027_edit_diffs_mv.sql index a1c420f..cf43aa5 100644 --- a/crates/weaver-index/migrations/clickhouse/027_edit_diffs_mv.sql +++ b/crates/weaver-index/migrations/clickhouse/027_edit_diffs_mv.sql @@ -44,17 +44,21 @@ SELECT '' ) as resource_collection, - -- Root reference + -- Root reference (StrongRef: uri + cid) splitByChar('/', replaceOne(toString(record.root.uri), 'at://', ''))[1] as root_did, splitByChar('/', replaceOne(toString(record.root.uri), 'at://', ''))[3] as root_rkey, + toString(record.root.cid) as root_cid, - -- Prev reference (optional) + -- Prev reference (optional StrongRef) if(toString(record.prev.uri) != '', splitByChar('/', replaceOne(toString(record.prev.uri), 'at://', ''))[1], '') as prev_did, if(toString(record.prev.uri) != '', splitByChar('/', replaceOne(toString(record.prev.uri), 'at://', ''))[3], '') as prev_rkey, + if(toString(record.prev.uri) != '', + toString(record.prev.cid), + '') as prev_cid, -- Check for inline diff vs snapshot if(length(toString(record.inlineDiff)) > 0, 1, 0) as has_inline_diff, diff --git a/crates/weaver-index/src/clickhouse.rs b/crates/weaver-index/src/clickhouse.rs index e0e2cbb..b830480 100644 --- a/crates/weaver-index/src/clickhouse.rs +++ b/crates/weaver-index/src/clickhouse.rs @@ -7,7 +7,8 @@ mod schema; pub use client::{Client, TableSize}; pub use migrations::{DbObject, MigrationResult, Migrator, ObjectType}; pub use queries::{ - EntryRow, HandleMappingRow, NotebookRow, ProfileCountsRow, ProfileRow, ProfileWithCounts, + CollaboratorRow, EditHeadRow, EditNodeRow, EntryRow, HandleMappingRow, NotebookRow, + ProfileCountsRow, ProfileRow, ProfileWithCounts, }; pub use resilient_inserter::{InserterConfig, ResilientRecordInserter}; pub use schema::{ diff --git a/crates/weaver-index/src/clickhouse/queries.rs b/crates/weaver-index/src/clickhouse/queries.rs index 8f6c4b9..9fa8cfa 100644 --- a/crates/weaver-index/src/clickhouse/queries.rs +++ b/crates/weaver-index/src/clickhouse/queries.rs @@ -2,11 +2,17 @@ //! //! These modules add query methods to the ClickHouse Client via impl blocks. +mod collab; +mod collab_state; mod contributors; +mod edit; mod identity; mod notebooks; mod profiles; +pub use collab::PermissionRow; +pub use collab_state::{CollaboratorRow, EditHeadRow}; +pub use edit::EditNodeRow; pub use identity::HandleMappingRow; pub use notebooks::{EntryRow, NotebookRow}; pub use profiles::{ProfileCountsRow, ProfileRow, ProfileWithCounts}; diff --git a/crates/weaver-index/src/clickhouse/queries/collab.rs b/crates/weaver-index/src/clickhouse/queries/collab.rs new file mode 100644 index 0000000..4701983 --- /dev/null +++ b/crates/weaver-index/src/clickhouse/queries/collab.rs @@ -0,0 +1,93 @@ +//! Collaboration and permission queries + +use clickhouse::Row; +use serde::Deserialize; +use smol_str::SmolStr; + +use crate::clickhouse::Client; +use crate::error::{ClickHouseError, IndexError}; + +/// Permission row from the permissions materialized view +#[derive(Debug, Clone, Row, Deserialize)] +pub struct PermissionRow { + pub resource_did: SmolStr, + pub resource_collection: SmolStr, + pub resource_rkey: SmolStr, + pub resource_uri: SmolStr, + pub grantee_did: SmolStr, + pub scope: SmolStr, + #[serde(with = "clickhouse::serde::chrono::datetime64::millis")] + pub granted_at: chrono::DateTime, +} + +impl Client { + /// Get all permissions for a resource by URI. + /// + /// Returns owner and all collaborators who have accepted invites. + pub async fn get_resource_permissions( + &self, + resource_uri: &str, + ) -> Result, IndexError> { + let query = r#" + SELECT + resource_did, + resource_collection, + resource_rkey, + resource_uri, + grantee_did, + scope, + granted_at + FROM permissions FINAL + WHERE resource_uri = ? + ORDER BY scope DESC, granted_at ASC + "#; + + let rows = self + .inner() + .query(query) + .bind(resource_uri) + .fetch_all::() + .await + .map_err(|e| ClickHouseError::Query { + message: "failed to get resource permissions".into(), + source: e, + })?; + + Ok(rows) + } + + /// Check if a DID can edit a resource. + /// + /// Returns true if the DID is owner or collaborator. + pub async fn can_edit_resource( + &self, + resource_uri: &str, + did: &str, + ) -> Result { + let query = r#" + SELECT count(*) as cnt + FROM permissions FINAL + WHERE resource_uri = ? + AND grantee_did = ? + "#; + + #[derive(Row, Deserialize)] + struct CountRow { + cnt: u64, + } + + let row = self + .inner() + .query(query) + .bind(resource_uri) + .bind(did) + .fetch_one::() + .await + .map_err(|e| ClickHouseError::Query { + message: "failed to check edit permission".into(), + source: e, + })?; + + Ok(row.cnt > 0) + } +} diff --git a/crates/weaver-index/src/clickhouse/queries/collab_state.rs b/crates/weaver-index/src/clickhouse/queries/collab_state.rs new file mode 100644 index 0000000..cbabfca --- /dev/null +++ b/crates/weaver-index/src/clickhouse/queries/collab_state.rs @@ -0,0 +1,156 @@ +//! Collaboration state queries + +use clickhouse::Row; +use serde::Deserialize; +use smol_str::SmolStr; + +use crate::clickhouse::Client; +use crate::error::{ClickHouseError, IndexError}; + +/// Collaborator row from the collaborators MV +#[derive(Debug, Clone, Row, Deserialize)] +pub struct CollaboratorRow { + pub resource_uri: SmolStr, + pub collaborator_did: SmolStr, + pub inviter_did: SmolStr, + pub invite_uri: SmolStr, + pub accept_uri: SmolStr, + pub scope: SmolStr, + #[serde(with = "clickhouse::serde::chrono::datetime64::millis")] + pub invited_at: chrono::DateTime, + #[serde(with = "clickhouse::serde::chrono::datetime64::millis")] + pub accepted_at: chrono::DateTime, +} + +/// Edit head row from the edit_heads MV +#[derive(Debug, Clone, Row, Deserialize)] +pub struct EditHeadRow { + pub resource_uri: SmolStr, + pub head_did: SmolStr, + pub head_rkey: SmolStr, + pub head_cid: SmolStr, + pub head_uri: SmolStr, + pub head_type: SmolStr, + pub root_did: SmolStr, + pub root_rkey: SmolStr, + pub root_cid: SmolStr, + pub root_uri: SmolStr, + #[serde(with = "clickhouse::serde::chrono::datetime64::millis")] + pub head_created_at: chrono::DateTime, +} + +impl Client { + /// Get collaborators for a resource (matched invite+accept pairs). + pub async fn get_collaborators( + &self, + resource_uri: &str, + ) -> Result, IndexError> { + let query = r#" + SELECT + resource_uri, + collaborator_did, + inviter_did, + invite_uri, + accept_uri, + scope, + invited_at, + accepted_at + FROM collaborators FINAL + WHERE resource_uri = ? + ORDER BY accepted_at ASC + "#; + + let rows = self + .inner() + .query(query) + .bind(resource_uri) + .fetch_all::() + .await + .map_err(|e| ClickHouseError::Query { + message: "failed to get collaborators".into(), + source: e, + })?; + + Ok(rows) + } + + /// Get edit heads for a resource. + /// + /// Multiple heads means divergent branches. + pub async fn get_edit_heads(&self, resource_uri: &str) -> Result, IndexError> { + let query = r#" + SELECT + resource_uri, + head_did, + head_rkey, + head_cid, + head_uri, + head_type, + root_did, + root_rkey, + root_cid, + root_uri, + head_created_at + FROM edit_heads FINAL + WHERE resource_uri = ? + ORDER BY head_created_at DESC + "#; + + let rows = self + .inner() + .query(query) + .bind(resource_uri) + .fetch_all::() + .await + .map_err(|e| ClickHouseError::Query { + message: "failed to get edit heads".into(), + source: e, + })?; + + Ok(rows) + } + + /// Check if resource has divergent branches (more than one head). + pub async fn has_divergence(&self, resource_uri: &str) -> Result { + let heads = self.get_edit_heads(resource_uri).await?; + Ok(heads.len() > 1) + } + + /// Get CID for a record from raw_records. + pub async fn get_record_cid( + &self, + did: &str, + collection: &str, + rkey: &str, + ) -> Result, IndexError> { + #[derive(Row, Deserialize)] + struct CidRow { + cid: SmolStr, + } + + let query = r#" + SELECT cid + FROM raw_records FINAL + WHERE did = ? + AND collection = ? + AND rkey = ? + AND operation != 'delete' + LIMIT 1 + "#; + + let row = self + .inner() + .query(query) + .bind(did) + .bind(collection) + .bind(rkey) + .fetch_optional::() + .await + .map_err(|e| ClickHouseError::Query { + message: "failed to get record cid".into(), + source: e, + })?; + + Ok(row.map(|r| r.cid)) + } +} diff --git a/crates/weaver-index/src/clickhouse/queries/edit.rs b/crates/weaver-index/src/clickhouse/queries/edit.rs new file mode 100644 index 0000000..80820e1 --- /dev/null +++ b/crates/weaver-index/src/clickhouse/queries/edit.rs @@ -0,0 +1,107 @@ +//! Edit history queries + +use clickhouse::Row; +use serde::Deserialize; +use smol_str::SmolStr; + +use crate::clickhouse::Client; +use crate::error::{ClickHouseError, IndexError}; + +/// Edit node row from the edit_nodes table +#[derive(Debug, Clone, Row, Deserialize)] +pub struct EditNodeRow { + pub did: SmolStr, + pub rkey: SmolStr, + pub cid: SmolStr, + pub collection: SmolStr, + pub node_type: SmolStr, + pub root_did: SmolStr, + pub root_rkey: SmolStr, + pub root_cid: SmolStr, + pub prev_did: SmolStr, + pub prev_rkey: SmolStr, + pub prev_cid: SmolStr, + pub has_inline_diff: u8, + pub has_snapshot: u8, + #[serde(with = "clickhouse::serde::chrono::datetime64::millis")] + pub created_at: chrono::DateTime, +} + +impl Client { + /// Get edit history for a resource. + /// + /// Returns roots and diffs separately, ordered by created_at. + /// The resource_uri should be an at:// URI for an entry or notebook. + pub async fn get_edit_history( + &self, + resource_uri: &str, + cursor: Option, + after_rkey: Option<&str>, + limit: i64, + ) -> Result, IndexError> { + // Parse resource URI to extract did/collection/rkey + let parts: Vec<&str> = resource_uri + .strip_prefix("at://") + .unwrap_or(resource_uri) + .split('/') + .collect(); + + if parts.len() < 3 { + return Ok(Vec::new()); + } + + let resource_did = parts[0]; + let resource_collection = parts[1]; + let resource_rkey = parts[2]; + + let query = r#" + SELECT + did, + rkey, + cid, + collection, + node_type, + root_did, + root_rkey, + root_cid, + prev_did, + prev_rkey, + prev_cid, + has_inline_diff, + has_snapshot, + created_at + FROM edit_nodes FINAL + WHERE resource_did = ? + AND resource_collection = ? + AND resource_rkey = ? + AND deleted_at = toDateTime64(0, 3) + AND (? = 0 OR toUnixTimestamp64Milli(created_at) < ?) + AND (? = '' OR rkey > ?) + ORDER BY created_at DESC + LIMIT ? + "#; + + let cursor_val = cursor.unwrap_or(0); + let after_rkey_val = after_rkey.unwrap_or(""); + + let rows = self + .inner() + .query(query) + .bind(resource_did) + .bind(resource_collection) + .bind(resource_rkey) + .bind(cursor_val) + .bind(cursor_val) + .bind(after_rkey_val) + .bind(after_rkey_val) + .bind(limit) + .fetch_all::() + .await + .map_err(|e| ClickHouseError::Query { + message: "failed to get edit history".into(), + source: e, + })?; + + Ok(rows) + } +} diff --git a/crates/weaver-index/src/endpoints/actor.rs b/crates/weaver-index/src/endpoints/actor.rs index 4d45a81..ff4de6b 100644 --- a/crates/weaver-index/src/endpoints/actor.rs +++ b/crates/weaver-index/src/endpoints/actor.rs @@ -163,14 +163,8 @@ pub async fn resolve_actor<'a>( } } -/// Convert SmolStr to Option if non-empty -fn non_empty_str(s: &smol_str::SmolStr) -> Option> { - if s.is_empty() { - None - } else { - Some(s.to_cowstr().into_static()) - } -} +// Re-export from parent for local use +use super::non_empty_str; /// Parse cursor string to i64 timestamp millis fn parse_cursor(cursor: Option<&str>) -> Result, XrpcErrorResponse> { @@ -426,7 +420,7 @@ fn hydrate_authors( } /// Convert ProfileRow to ProfileDataView -fn profile_to_data_view( +pub fn profile_to_data_view( profile: &ProfileRow, ) -> Result, XrpcErrorResponse> { use jacquard::types::string::Uri; diff --git a/crates/weaver-index/src/endpoints/collab.rs b/crates/weaver-index/src/endpoints/collab.rs new file mode 100644 index 0000000..0bcde53 --- /dev/null +++ b/crates/weaver-index/src/endpoints/collab.rs @@ -0,0 +1,327 @@ +//! Collaboration endpoint handlers + +use std::collections::HashMap; + +use axum::{Json, extract::State}; +use jacquard::IntoStatic; +use jacquard::cowstr::ToCowStr; +use jacquard::types::datetime::Datetime; +use jacquard::types::string::{AtUri, Cid, Did, Handle}; +use jacquard_axum::ExtractXrpc; +use jacquard_axum::service_auth::ExtractOptionalServiceAuth; + +use weaver_api::com_atproto::repo::strong_ref::StrongRef; +use weaver_api::sh_weaver::actor::ProfileViewBasic; +use weaver_api::sh_weaver::collab::get_collaboration_state::{ + GetCollaborationStateOutput, GetCollaborationStateRequest, +}; +use weaver_api::sh_weaver::collab::get_resource_participants::{ + GetResourceParticipantsOutput, GetResourceParticipantsRequest, +}; +use weaver_api::sh_weaver::collab::{CollaborationStateView, ParticipantStateView}; + +use crate::clickhouse::{CollaboratorRow, ProfileRow}; +use crate::endpoints::actor::Viewer; +use crate::endpoints::non_empty_str; +use crate::endpoints::repo::XrpcErrorResponse; +use crate::server::AppState; + +/// Handle sh.weaver.collab.getResourceParticipants +/// +/// Returns owner and collaborators who can edit the resource. +pub async fn get_resource_participants( + State(state): State, + ExtractOptionalServiceAuth(viewer): ExtractOptionalServiceAuth, + ExtractXrpc(args): ExtractXrpc, +) -> Result>, XrpcErrorResponse> { + let _viewer: Viewer = viewer; + let viewer_did: Option<&str> = _viewer.as_ref().map(|v| v.did().as_str()); + + let resource_uri = args.resource.as_str(); + + // Get all permissions for the resource + let permissions = state + .clickhouse + .get_resource_permissions(resource_uri) + .await + .map_err(|e| { + tracing::error!("Failed to get resource permissions: {}", e); + XrpcErrorResponse::internal_error("Database query failed") + })?; + + if permissions.is_empty() { + return Err(XrpcErrorResponse::not_found("Resource not found")); + } + + // Collect all DIDs for profile hydration + let all_dids: Vec<&str> = permissions.iter().map(|p| p.grantee_did.as_str()).collect(); + + // Batch fetch profiles + let profiles = state + .clickhouse + .get_profiles_batch(&all_dids) + .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(); + + // Find owner and build participants + let mut owner: Option> = None; + let mut participants: Vec> = Vec::new(); + + for perm in &permissions { + let profile_view = if let Some(profile) = profile_map.get(perm.grantee_did.as_str()) { + profile_to_view_basic(profile)? + } else { + // No profile found - skip (shouldn't happen if permissions table is consistent) + continue; + }; + + if perm.scope == "owner" { + owner = Some(profile_view); + } else { + participants.push(profile_view); + } + } + + let owner = owner.ok_or_else(|| { + tracing::error!("Resource has no owner in permissions"); + XrpcErrorResponse::internal_error("Resource has no owner") + })?; + + // Check if viewer can edit + let viewer_can_edit = viewer_did.map(|v| all_dids.contains(&v)); + + Ok(Json( + GetResourceParticipantsOutput { + owner, + participants, + viewer_can_edit, + extra_data: None, + } + .into_static(), + )) +} + +/// Convert ProfileRow to ProfileViewBasic directly +pub fn profile_to_view_basic( + profile: &ProfileRow, +) -> Result, XrpcErrorResponse> { + let did = Did::new_owned(profile.did.clone()) + .map_err(|_| XrpcErrorResponse::internal_error("Invalid DID in profile"))?; + + let handle = Handle::new_owned(profile.handle.clone()) + .map_err(|_| XrpcErrorResponse::internal_error("Invalid handle in profile"))?; + + Ok(ProfileViewBasic::new() + .did(did) + .handle(handle) + .maybe_display_name(non_empty_str(&profile.display_name)) + .build()) +} + +/// Handle sh.weaver.collab.getCollaborationState +/// +/// Returns full collaboration state for a resource. +pub async fn get_collaboration_state( + State(state): State, + ExtractOptionalServiceAuth(viewer): ExtractOptionalServiceAuth, + ExtractXrpc(args): ExtractXrpc, +) -> Result>, XrpcErrorResponse> { + let _viewer: Viewer = viewer; + + let resource_uri = args.resource.as_str(); + + // Get permissions for the resource + let permissions = state + .clickhouse + .get_resource_permissions(resource_uri) + .await + .map_err(|e| { + tracing::error!("Failed to get resource permissions: {}", e); + XrpcErrorResponse::internal_error("Database query failed") + })?; + + if permissions.is_empty() { + return Err(XrpcErrorResponse::not_found("Resource not found")); + } + + // Get collaborators (invite+accept pairs) for additional data + let collaborators = state + .clickhouse + .get_collaborators(resource_uri) + .await + .map_err(|e| { + tracing::error!("Failed to get collaborators: {}", e); + XrpcErrorResponse::internal_error("Database query failed") + })?; + + // Check for divergence + let has_divergence = state + .clickhouse + .has_divergence(resource_uri) + .await + .map_err(|e| { + tracing::error!("Failed to check divergence: {}", e); + XrpcErrorResponse::internal_error("Database query failed") + })?; + + // Collect all DIDs for profile hydration + let all_dids: Vec<&str> = permissions.iter().map(|p| p.grantee_did.as_str()).collect(); + + // Batch fetch profiles + let profiles = state + .clickhouse + .get_profiles_batch(&all_dids) + .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 collaborator lookup for invite/accept URIs + let collab_map: HashMap<&str, &CollaboratorRow> = collaborators + .iter() + .map(|c| (c.collaborator_did.as_str(), c)) + .collect(); + + // Find owner and get resource CID + let owner_perm = permissions + .iter() + .find(|p| p.scope == "owner") + .ok_or_else(|| { + tracing::error!("Resource has no owner in permissions"); + XrpcErrorResponse::internal_error("Resource has no owner") + })?; + + // Build resource StrongRef - look up the CID from the appropriate table + let resource_uri_parsed = AtUri::new(resource_uri) + .map_err(|_| XrpcErrorResponse::internal_error("Invalid resource URI"))? + .into_static(); + + // Look up the resource CID from raw_records + let resource_cid = state + .clickhouse + .get_record_cid( + &owner_perm.resource_did, + &owner_perm.resource_collection, + &owner_perm.resource_rkey, + ) + .await + .map_err(|e| { + tracing::error!("Failed to get resource CID: {}", e); + XrpcErrorResponse::internal_error("Database query failed") + })? + .ok_or_else(|| XrpcErrorResponse::not_found("Resource not found in database"))?; + + let resource = StrongRef::new() + .uri(resource_uri_parsed.clone()) + .cid( + Cid::new(resource_cid.as_bytes()) + .map_err(|_| XrpcErrorResponse::internal_error("Invalid resource CID"))? + .into_static(), + ) + .build(); + + // Build participants + let mut participants: Vec> = Vec::new(); + let mut first_collab_at: Option> = None; + + for perm in &permissions { + let profile = profile_map + .get(perm.grantee_did.as_str()) + .ok_or_else(|| XrpcErrorResponse::internal_error("Missing profile for participant"))?; + let collab = collab_map.get(perm.grantee_did.as_str()); + + // Track first collaborator time + if perm.scope != "owner" { + if let Some(c) = collab { + match first_collab_at { + None => first_collab_at = Some(c.accepted_at), + Some(t) if c.accepted_at < t => first_collab_at = Some(c.accepted_at), + _ => {} + } + } + } + + let participant = build_participant_state(profile, collab, &perm.scope)?; + participants.push(participant); + } + + // Determine status + let status = if collaborators.is_empty() { + "solo" + } else if has_divergence { + "diverged" + } else { + "synced" + }; + + let collab_state = CollaborationStateView::new() + .resource(resource) + .status(status) + .participants(participants) + .maybe_canonical_uri(Some(resource_uri_parsed)) + .maybe_has_divergence(Some(has_divergence)) + .maybe_first_collaborator_added_at( + first_collab_at.map(|dt| Datetime::new(dt.fixed_offset())), + ) + .build(); + + Ok(Json( + GetCollaborationStateOutput { + value: collab_state, + extra_data: None, + } + .into_static(), + )) +} + +/// Build ParticipantStateView from available data +fn build_participant_state( + profile: &ProfileRow, + collab: Option<&&CollaboratorRow>, + scope: &str, +) -> Result, XrpcErrorResponse> { + let user = profile_to_view_basic(profile)?; + + let role = match scope { + "owner" => "owner", + "collaborator" => "collaborator", + _ => "unknown", + }; + + let status = if collab.is_some() { + "active" + } else { + "pending" + }; + + // Parse URIs if we have collab data + let (invite_uri, accept_uri) = if let Some(c) = collab { + let inv = AtUri::new(c.invite_uri.as_str()) + .map_err(|_| XrpcErrorResponse::internal_error("Invalid invite URI"))? + .into_static(); + let acc = AtUri::new(c.accept_uri.as_str()) + .map_err(|_| XrpcErrorResponse::internal_error("Invalid accept URI"))? + .into_static(); + (Some(inv), Some(acc)) + } else { + (None, None) + }; + + Ok(ParticipantStateView::new() + .role(role) + .status(status) + .user(user) + .maybe_invite_uri(invite_uri) + .maybe_accept_uri(accept_uri) + .build()) +} diff --git a/crates/weaver-index/src/endpoints/edit.rs b/crates/weaver-index/src/endpoints/edit.rs new file mode 100644 index 0000000..91250a7 --- /dev/null +++ b/crates/weaver-index/src/endpoints/edit.rs @@ -0,0 +1,186 @@ +//! Edit endpoint handlers + +use std::collections::HashMap; + +use axum::{Json, extract::State}; +use jacquard::IntoStatic; +use jacquard::cowstr::ToCowStr; +use jacquard::types::datetime::Datetime; +use jacquard::types::string::{AtUri, Cid}; +use jacquard_axum::ExtractXrpc; +use jacquard_axum::service_auth::ExtractOptionalServiceAuth; + +use weaver_api::com_atproto::repo::strong_ref::StrongRef; +use weaver_api::sh_weaver::edit::EditHistoryEntry; +use weaver_api::sh_weaver::edit::get_edit_history::{GetEditHistoryOutput, GetEditHistoryRequest}; + +use crate::clickhouse::{EditNodeRow, ProfileRow}; +use crate::endpoints::actor::Viewer; +use crate::endpoints::collab::profile_to_view_basic; +use crate::endpoints::repo::XrpcErrorResponse; +use crate::server::AppState; + +/// Handle sh.weaver.edit.getEditHistory +/// +/// Returns edit history (roots and diffs) for a resource. +pub async fn get_edit_history( + State(state): State, + ExtractOptionalServiceAuth(viewer): ExtractOptionalServiceAuth, + ExtractXrpc(args): ExtractXrpc, +) -> Result>, XrpcErrorResponse> { + let _viewer: Viewer = viewer; + + let resource_uri = args.resource.as_str(); + let limit = args.limit.unwrap_or(50).min(100).max(1); + + // Parse cursor as millisecond timestamp + let cursor = args + .cursor + .as_deref() + .map(|c| c.parse::()) + .transpose() + .map_err(|_| XrpcErrorResponse::invalid_request("Invalid cursor format"))?; + + let after_rkey = args.after_rkey.as_deref(); + + // Fetch edit nodes + let nodes = state + .clickhouse + .get_edit_history(resource_uri, cursor, after_rkey, limit + 1) + .await + .map_err(|e| { + tracing::error!("Failed to get edit history: {}", e); + XrpcErrorResponse::internal_error("Database query failed") + })?; + + // Check if there are more results + let has_more = nodes.len() > limit as usize; + let nodes: Vec<_> = nodes.into_iter().take(limit as usize).collect(); + + // Collect unique author DIDs + let author_dids: Vec<&str> = nodes.iter().map(|n| n.did.as_str()).collect(); + let unique_dids: Vec<&str> = author_dids + .iter() + .copied() + .collect::>() + .into_iter() + .collect(); + + // Batch fetch profiles + let profiles = state + .clickhouse + .get_profiles_batch(&unique_dids) + .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(); + + // Separate roots and diffs, building EditHistoryEntry for each + let mut roots = Vec::new(); + let mut diffs = Vec::new(); + + for node in &nodes { + let entry = node_to_history_entry(node, &profile_map)?; + + if node.node_type == "root" { + roots.push(entry); + } else { + diffs.push(entry); + } + } + + // Build cursor from last node's created_at + let next_cursor = if has_more { + nodes + .last() + .map(|n| n.created_at.timestamp_millis().to_cowstr().into_static()) + } else { + None + }; + + Ok(Json( + GetEditHistoryOutput { + roots, + diffs, + cursor: next_cursor, + extra_data: None, + } + .into_static(), + )) +} + +/// Convert EditNodeRow to EditHistoryEntry +fn node_to_history_entry( + node: &EditNodeRow, + profile_map: &HashMap<&str, &ProfileRow>, +) -> Result, XrpcErrorResponse> { + let author = profile_map + .get(node.did.as_str()) + .map(|p| profile_to_view_basic(p)) + .transpose()? + .ok_or_else(|| XrpcErrorResponse::internal_error("Author profile not found"))?; + + // Build URI + let uri = AtUri::new(&format!( + "at://{}/{}/{}", + node.did, node.collection, node.rkey + )) + .map_err(|_| XrpcErrorResponse::internal_error("Invalid AT URI"))? + .into_static(); + + let cid = Cid::new(node.cid.as_bytes()) + .map_err(|_| XrpcErrorResponse::internal_error("Invalid CID"))? + .into_static(); + + // Build optional StrongRefs for diffs + let root_ref = if !node.root_cid.is_empty() { + let root_uri = AtUri::new(&format!( + "at://{}/sh.weaver.edit.root/{}", + node.root_did, node.root_rkey + )) + .map_err(|_| XrpcErrorResponse::internal_error("Invalid root URI"))? + .into_static(); + + let root_cid = Cid::new(node.root_cid.as_bytes()) + .map_err(|_| XrpcErrorResponse::internal_error("Invalid root CID"))? + .into_static(); + + Some(StrongRef::new().uri(root_uri).cid(root_cid).build()) + } else { + None + }; + + let prev_ref = if !node.prev_cid.is_empty() { + let prev_uri = AtUri::new(&format!( + "at://{}/sh.weaver.edit.diff/{}", + node.prev_did, node.prev_rkey + )) + .map_err(|_| XrpcErrorResponse::internal_error("Invalid prev URI"))? + .into_static(); + + let prev_cid = Cid::new(node.prev_cid.as_bytes()) + .map_err(|_| XrpcErrorResponse::internal_error("Invalid prev CID"))? + .into_static(); + + Some(StrongRef::new().uri(prev_uri).cid(prev_cid).build()) + } else { + None + }; + + let created_at = Datetime::new(node.created_at.fixed_offset()); + + Ok(EditHistoryEntry::new() + .uri(uri) + .cid(cid) + .author(author) + .created_at(created_at) + .r#type(node.node_type.clone()) + .maybe_has_inline_diff(Some(node.has_inline_diff == 1)) + .maybe_prev_ref(prev_ref) + .maybe_root_ref(root_ref) + .build()) +} diff --git a/crates/weaver-index/src/endpoints/mod.rs b/crates/weaver-index/src/endpoints/mod.rs index 6a54377..126a142 100644 --- a/crates/weaver-index/src/endpoints/mod.rs +++ b/crates/weaver-index/src/endpoints/mod.rs @@ -1,5 +1,21 @@ //! XRPC endpoint handlers for the appview. +use jacquard::CowStr; +use jacquard::IntoStatic; +use jacquard::cowstr::ToCowStr; +use smol_str::SmolStr; + pub mod actor; +pub mod collab; +pub mod edit; pub mod notebook; pub mod repo; + +/// Convert SmolStr to Option if non-empty +pub fn non_empty_str(s: &SmolStr) -> Option> { + if s.is_empty() { + None + } else { + Some(s.to_cowstr().into_static()) + } +} diff --git a/crates/weaver-index/src/endpoints/notebook.rs b/crates/weaver-index/src/endpoints/notebook.rs index f6f3745..55e5dd9 100644 --- a/crates/weaver-index/src/endpoints/notebook.rs +++ b/crates/weaver-index/src/endpoints/notebook.rs @@ -10,12 +10,14 @@ use jacquard::types::value::Data; use jacquard_axum::ExtractXrpc; use jacquard_axum::service_auth::ExtractOptionalServiceAuth; use smol_str::SmolStr; +use weaver_api::com_atproto::repo::strong_ref::StrongRef; use weaver_api::sh_weaver::actor::{ProfileDataView, ProfileDataViewInner, ProfileView}; use weaver_api::sh_weaver::notebook::{ AuthorListView, BookEntryRef, BookEntryView, EntryView, FeedEntryView, NotebookView, get_book_entry::{GetBookEntryOutput, GetBookEntryRequest}, get_entry::{GetEntryOutput, GetEntryRequest}, get_entry_feed::{GetEntryFeedOutput, GetEntryFeedRequest}, + get_notebook::{GetNotebookOutput, GetNotebookRequest}, get_notebook_feed::{GetNotebookFeedOutput, GetNotebookFeedRequest}, resolve_entry::{ResolveEntryOutput, ResolveEntryRequest}, resolve_notebook::{ResolveNotebookOutput, ResolveNotebookRequest}, @@ -198,6 +200,117 @@ pub async fn resolve_notebook( )) } +/// Handle sh.weaver.notebook.getNotebook +/// +/// Gets a notebook by AT URI, returns notebook view with entry refs. +pub async fn get_notebook( + 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.notebook; + let authority = uri.authority(); + let rkey = uri + .rkey() + .ok_or_else(|| XrpcErrorResponse::invalid_request("URI must include rkey"))?; + let rkey_str = rkey.as_ref(); + + // Resolve authority to DID (could be handle or DID) + let did = resolve_actor(&state, authority).await?; + let did_str = did.as_str(); + + // Fetch notebook by DID + rkey + let notebook_row = state + .clickhouse + .get_notebook(did_str, rkey_str) + .await + .map_err(|e| { + tracing::error!("Failed to get notebook: {}", e); + XrpcErrorResponse::internal_error("Database query failed") + })? + .ok_or_else(|| XrpcErrorResponse::not_found("Notebook not found"))?; + + // Fetch notebook contributors + let notebook_contributors = state + .clickhouse + .get_notebook_contributors(did_str, rkey_str) + .await + .map_err(|e| { + tracing::error!("Failed to get notebook contributors: {}", e); + XrpcErrorResponse::internal_error("Database query failed") + })?; + + // Collect all author DIDs for batch hydration + let mut all_author_dids: HashSet<&str> = + notebook_contributors.iter().map(|s| s.as_str()).collect(); + for did in ¬ebook_row.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 NotebookView + let notebook_uri = AtUri::new(¬ebook_row.uri).map_err(|e| { + tracing::error!("Invalid notebook URI in db: {}", e); + XrpcErrorResponse::internal_error("Invalid URI stored") + })?; + + let notebook_cid = Cid::new(notebook_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(¬ebook_contributors, &profile_map)?; + let record = parse_record_json(¬ebook_row.record)?; + + let notebook = NotebookView::new() + .uri(notebook_uri.into_static()) + .cid(notebook_cid.into_static()) + .authors(authors) + .record(record.clone()) + .indexed_at(notebook_row.indexed_at.fixed_offset()) + .maybe_title(non_empty_cowstr(¬ebook_row.title)) + .maybe_path(non_empty_cowstr(¬ebook_row.path)) + .build(); + + // Deserialize Book from record to get entry_list + let book: weaver_api::sh_weaver::notebook::book::Book = + jacquard::from_data(&record).map_err(|e| { + tracing::error!("Failed to deserialize Book record: {}", e); + XrpcErrorResponse::internal_error("Invalid Book record") + })?; + + let entries: Vec> = book + .entry_list + .into_iter() + .map(|r| r.into_static()) + .collect(); + + Ok(Json( + GetNotebookOutput { + notebook, + entries, + extra_data: None, + } + .into_static(), + )) +} + /// Handle sh.weaver.notebook.getEntry /// /// Gets an entry by AT URI. diff --git a/crates/weaver-index/src/server.rs b/crates/weaver-index/src/server.rs index b9e90a1..921fba9 100644 --- a/crates/weaver-index/src/server.rs +++ b/crates/weaver-index/src/server.rs @@ -19,15 +19,19 @@ use weaver_api::sh_weaver::actor::{ get_actor_entries::GetActorEntriesRequest, get_actor_notebooks::GetActorNotebooksRequest, get_profile::GetProfileRequest, }; +use weaver_api::sh_weaver::collab::get_collaboration_state::GetCollaborationStateRequest; +use weaver_api::sh_weaver::collab::get_resource_participants::GetResourceParticipantsRequest; +use weaver_api::sh_weaver::edit::get_edit_history::GetEditHistoryRequest; 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, + get_entry_feed::GetEntryFeedRequest, get_notebook::GetNotebookRequest, + get_notebook_feed::GetNotebookFeedRequest, resolve_entry::ResolveEntryRequest, + resolve_notebook::ResolveNotebookRequest, }; use crate::clickhouse::Client; use crate::config::ShardConfig; -use crate::endpoints::{actor, notebook, repo}; +use crate::endpoints::{actor, collab, edit, notebook, repo}; use crate::error::{IndexError, ServerError}; use crate::sqlite::ShardRouter; @@ -94,6 +98,7 @@ pub fn router(state: AppState, did_doc: DidDocument<'static>) -> Router { .merge(ResolveNotebookRequest::into_router( notebook::resolve_notebook, )) + .merge(GetNotebookRequest::into_router(notebook::get_notebook)) .merge(GetEntryRequest::into_router(notebook::get_entry)) .merge(ResolveEntryRequest::into_router(notebook::resolve_entry)) .merge(GetNotebookFeedRequest::into_router( @@ -101,6 +106,15 @@ pub fn router(state: AppState, did_doc: DidDocument<'static>) -> Router { )) .merge(GetEntryFeedRequest::into_router(notebook::get_entry_feed)) .merge(GetBookEntryRequest::into_router(notebook::get_book_entry)) + // sh.weaver.collab.* endpoints + .merge(GetResourceParticipantsRequest::into_router( + collab::get_resource_participants, + )) + .merge(GetCollaborationStateRequest::into_router( + collab::get_collaboration_state, + )) + // sh.weaver.edit.* endpoints + .merge(GetEditHistoryRequest::into_router(edit::get_edit_history)) .layer(TraceLayer::new_for_http()) .with_state(state) .merge(did_web_router(did_doc))