diff --git a/Cargo.lock b/Cargo.lock index f6db8aa573..acdc8f4bd2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1096,14 +1096,9 @@ name = "buzz-search" version = "0.1.0" dependencies = [ "buzz-core", - "chrono", - "nostr", - "reqwest 0.13.3", - "serde", - "serde_json", + "sqlx", "thiserror 2.0.18", "tokio", - "tracing", "uuid", ] diff --git a/crates/buzz-search/Cargo.toml b/crates/buzz-search/Cargo.toml index 3c571733a6..e42bcc8041 100644 --- a/crates/buzz-search/Cargo.toml +++ b/crates/buzz-search/Cargo.toml @@ -5,16 +5,13 @@ edition.workspace = true rust-version.workspace = true license.workspace = true repository.workspace = true -description = "Typesense-backed full-text search for Buzz" +description = "Postgres full-text search for Buzz, scoped by community" [dependencies] -buzz-core = { workspace = true } -reqwest = { workspace = true } -tokio = { workspace = true } -serde = { workspace = true } -serde_json = { workspace = true } +buzz-core = { workspace = true } +sqlx = { workspace = true } uuid = { workspace = true } -chrono = { workspace = true } -tracing = { workspace = true } thiserror = { workspace = true } -nostr = { workspace = true } + +[dev-dependencies] +tokio = { workspace = true } diff --git a/crates/buzz-search/src/collection.rs b/crates/buzz-search/src/collection.rs deleted file mode 100644 index 8f7dbc5e4c..0000000000 --- a/crates/buzz-search/src/collection.rs +++ /dev/null @@ -1,129 +0,0 @@ -//! Typesense collection schema management. - -use serde_json::json; -use tracing::{debug, info, warn}; - -use crate::error::SearchError; - -/// Returns the Typesense collection schema JSON for the events collection. -pub fn events_schema(collection_name: &str) -> serde_json::Value { - json!({ - "name": collection_name, - "fields": [ - {"name": "id", "type": "string"}, - {"name": "content", "type": "string"}, - {"name": "kind", "type": "int32"}, - {"name": "pubkey", "type": "string", "facet": true}, - {"name": "channel_id", "type": "string", "facet": true, "optional": true}, - {"name": "created_at", "type": "int64"}, - {"name": "tags_flat", "type": "string[]", "optional": true} - ], - "default_sorting_field": "created_at" - }) -} - -/// Ensures the Typesense collection exists, creating it if absent (idempotent). -pub async fn ensure_collection( - client: &reqwest::Client, - base_url: &str, - api_key: &str, - collection_name: &str, -) -> Result<(), SearchError> { - let check_url = format!("{}/collections/{}", base_url, collection_name); - let resp = client - .get(&check_url) - .header("X-TYPESENSE-API-KEY", api_key) - .send() - .await?; - - match resp.status().as_u16() { - 200 => { - debug!(collection = collection_name, "Collection already exists"); - return Ok(()); - } - 404 => { - debug!( - collection = collection_name, - "Collection not found, creating" - ); - } - status => { - let body = resp.text().await.unwrap_or_default(); - return Err(SearchError::Api { status, body }); - } - } - - let schema = events_schema(collection_name); - let create_url = format!("{}/collections", base_url); - let resp = client - .post(&create_url) - .header("X-TYPESENSE-API-KEY", api_key) - .header("Content-Type", "application/json") - .json(&schema) - .send() - .await?; - - let status = resp.status().as_u16(); - match status { - 200 | 201 => { - info!( - collection = collection_name, - "Collection created successfully" - ); - Ok(()) - } - 409 => { - // Race condition: another process created it between our check and create. - warn!( - collection = collection_name, - "Collection created concurrently (409 conflict), treating as success" - ); - Ok(()) - } - _ => { - let body = resp.text().await.unwrap_or_default(); - Err(SearchError::Api { status, body }) - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_events_schema_structure() { - let schema = events_schema("events"); - assert_eq!(schema["name"], "events"); - assert_eq!(schema["default_sorting_field"], "created_at"); - - let fields = schema["fields"].as_array().unwrap(); - assert_eq!(fields.len(), 7); - - let field_names: Vec<&str> = fields.iter().map(|f| f["name"].as_str().unwrap()).collect(); - for expected in [ - "id", - "content", - "kind", - "pubkey", - "channel_id", - "created_at", - "tags_flat", - ] { - assert!(field_names.contains(&expected)); - } - } - - #[test] - fn test_events_schema_field_types() { - let schema = events_schema("test"); - let fields = schema["fields"].as_array().unwrap(); - let find = |name: &str| fields.iter().find(|f| f["name"] == name).unwrap().clone(); - - assert_eq!(find("kind")["type"], "int32"); - assert_eq!(find("pubkey")["facet"], true); - assert_eq!(find("channel_id")["optional"], true); - assert_eq!(find("created_at")["type"], "int64"); - assert_eq!(find("tags_flat")["type"], "string[]"); - } -} diff --git a/crates/buzz-search/src/error.rs b/crates/buzz-search/src/error.rs index e17fb5f065..4a71a01daa 100644 --- a/crates/buzz-search/src/error.rs +++ b/crates/buzz-search/src/error.rs @@ -1,39 +1,9 @@ use thiserror::Error; -/// Errors produced by the search service. +/// Errors produced by the FTS service. #[derive(Debug, Error)] pub enum SearchError { - /// An HTTP transport error from reqwest. - #[error("HTTP error: {0}")] - Http(#[from] reqwest::Error), - - /// Typesense returned a non-success HTTP status. - #[error("Typesense API error (status {status}): {body}")] - Api { - /// HTTP status code returned by Typesense. - status: u16, - /// Response body from Typesense. - body: String, - }, - - /// JSON serialization or deserialization failed. - #[error("JSON error: {0}")] - Json(#[from] serde_json::Error), - - /// A batch import partially failed. - #[error("Batch import partial failure: {succeeded} succeeded, {failed} failed")] - BatchPartial { - /// Number of documents successfully imported. - succeeded: usize, - /// Number of documents that failed to import. - failed: usize, - }, - - /// A Nostr event could not be converted to a Typesense document. - #[error("Event conversion error: {0}")] - Conversion(String), - - /// The provided event ID is not valid hex. - #[error("Invalid event_id: {0}")] - InvalidEventId(String), + /// A database error from sqlx. + #[error("database error: {0}")] + Db(#[from] sqlx::Error), } diff --git a/crates/buzz-search/src/index.rs b/crates/buzz-search/src/index.rs deleted file mode 100644 index 53fa151a19..0000000000 --- a/crates/buzz-search/src/index.rs +++ /dev/null @@ -1,529 +0,0 @@ -//! Event indexing — Nostr events → Typesense documents. Upsert semantics. - -use serde_json::{json, Value}; -use tracing::{debug, warn}; - -use buzz_core::event::StoredEvent; -use buzz_core::kind::event_kind_i32; - -use crate::error::SearchError; - -/// Converts a [`StoredEvent`] into a Typesense document JSON value. -pub fn event_to_document(event: &StoredEvent) -> Result { - let nostr_event = &event.event; - - // Use ASCII unit separator (U+001F) as delimiter to avoid ambiguity with - // tag values that contain colons (e.g. URLs in "r" tags). - let tags_flat: Vec = nostr_event - .tags - .iter() - .flat_map(|tag| { - let tag_vec = tag.as_slice(); - if tag_vec.len() >= 2 { - vec![format!("{}\x1f{}", tag_vec[0], tag_vec[1])] - } else if tag_vec.len() == 1 { - vec![tag_vec[0].to_string()] - } else { - vec![] - } - }) - .collect(); - - // Global events use a sentinel value instead of NULL/absent. Typesense 27.1's - // `__missing__` filter does not reliably match absent optional fields, so we use - // an explicit `__global__` sentinel that can be matched with `channel_id:=__global__`. - // NOTE: Historical docs indexed before this change have channel_id absent/null and - // won't match the sentinel filter. A full reindex (`just reindex-search`) is needed - // after deploy to backfill. Pre-existing global events (kind:0 only) were already - // excluded from search results by the old `.filter(|h| h.channel_id.is_some())`, so - // this is not a regression — those docs were never returned. - let channel_id_val = event - .channel_id - .as_ref() - .map(|id| id.to_string()) - .unwrap_or_else(|| "__global__".to_string()); - - // For kind:0 (user metadata) we append the parsed JSON values to `content` - // so Typesense's word-tokenizer can index them cleanly. Without this, a - // raw blob like `{"display_name":"alice","about":"loves cats"}` does not - // produce a clean `alice` token — the leading `"` glues onto the next - // word, so the doc is unreachable for the obvious `q=alice` search. - // - // We only flatten kind:0 (the structured-metadata kind defined by NIP-01) - // and only the small set of fields the member-picker uses. Bio / about / - // website are intentionally left out so they don't pollute name-prefix - // searches with false positives. Stays consistent with `display_name > - // nip05 > pubkey` ranking applied on the desktop side. - // - // The Typesense `content` field is write-only as far as the relay's read - // paths go (the bridge fetches the canonical event from Postgres by id - // after Typesense returns hits), so appending derived tokens here doesn't - // affect any consumer's view of the event's actual content. - // - // NOTE: existing kind:0 docs indexed before this change won't have the - // appended tokens. Running `just reindex-search` (or the - // `buzz-relay reindex-search` admin path) repopulates them. New / - // updated profiles get the tokens automatically. - let content_indexed = if event_kind_i32(nostr_event) == 0 { - flatten_kind0_for_indexing(nostr_event.content.as_str()) - } else { - nostr_event.content.as_str().to_string() - }; - - let doc = json!({ - "id": nostr_event.id.to_string(), - "content": content_indexed, - // Cast to i32 for Typesense schema (int32 field). nostr Kind is u16; all Buzz kinds fit in i32. - "kind": event_kind_i32(nostr_event), - "pubkey": nostr_event.pubkey.to_string(), - "channel_id": channel_id_val, - "created_at": nostr_event.created_at.as_secs() as i64, - "tags_flat": tags_flat, - }); - - Ok(doc) -} - -/// For kind:0 events, return the original content with the searchable fields -/// (`display_name`, `name`, `nip05`) appended as space-separated plain words. -/// -/// Tolerant of malformed input: anything that fails JSON parsing returns the -/// original content unchanged, never an error. -fn flatten_kind0_for_indexing(raw_content: &str) -> String { - let Ok(parsed) = serde_json::from_str::(raw_content) else { - return raw_content.to_string(); - }; - let Some(obj) = parsed.as_object() else { - return raw_content.to_string(); - }; - - let mut extracted: Vec<&str> = Vec::with_capacity(3); - for key in ["display_name", "name", "nip05"] { - if let Some(val) = obj.get(key).and_then(Value::as_str) { - let trimmed = val.trim(); - if !trimmed.is_empty() { - extracted.push(trimmed); - } - } - } - - if extracted.is_empty() { - raw_content.to_string() - } else { - // Single leading space ensures we don't smash the closing `}` of the - // original JSON into the first appended token. - format!("{} {}", raw_content, extracted.join(" ")) - } -} - -/// Indexes a single event via Typesense upsert. -pub async fn index_event( - client: &reqwest::Client, - base_url: &str, - api_key: &str, - collection_name: &str, - event: &StoredEvent, -) -> Result<(), SearchError> { - let doc = event_to_document(event)?; - let url = format!( - "{}/collections/{}/documents?action=upsert", - base_url, collection_name - ); - - debug!(event_id = %event.event.id, collection = collection_name, "indexing event"); - - let resp = client - .post(&url) - .header("X-TYPESENSE-API-KEY", api_key) - .header("Content-Type", "application/json") - .json(&doc) - .send() - .await?; - - let status = resp.status().as_u16(); - if status == 200 || status == 201 { - Ok(()) - } else { - let body = resp.text().await.unwrap_or_default(); - Err(SearchError::Api { status, body }) - } -} - -/// Indexes a batch of events via Typesense JSONL import. -pub async fn index_batch( - client: &reqwest::Client, - base_url: &str, - api_key: &str, - collection_name: &str, - events: &[StoredEvent], -) -> Result { - if events.is_empty() { - return Ok(0); - } - - let mut jsonl = String::new(); - for event in events { - let doc = event_to_document(event)?; - jsonl.push_str(&serde_json::to_string(&doc)?); - jsonl.push('\n'); - } - - let url = format!( - "{}/collections/{}/documents/import?action=upsert", - base_url, collection_name - ); - - debug!( - count = events.len(), - collection = collection_name, - "batch indexing events" - ); - - let resp = client - .post(&url) - .header("X-TYPESENSE-API-KEY", api_key) - .header("Content-Type", "text/plain") - .body(jsonl) - .send() - .await?; - - let status = resp.status().as_u16(); - if status != 200 { - let body = resp.text().await.unwrap_or_default(); - return Err(SearchError::Api { status, body }); - } - - let body = resp.text().await.unwrap_or_default(); - let mut succeeded = 0usize; - let mut failed = 0usize; - - for line in body.lines() { - if line.trim().is_empty() { - continue; - } - match serde_json::from_str::(line) { - Ok(result) => { - if result["success"].as_bool().unwrap_or(false) { - succeeded += 1; - } else { - failed += 1; - warn!( - error = result["error"].as_str().unwrap_or("unknown"), - "batch import document failure" - ); - } - } - Err(e) => { - warn!(line = line, error = %e, "could not parse batch import result line"); - failed += 1; - } - } - } - - if failed > 0 { - Err(SearchError::BatchPartial { succeeded, failed }) - } else { - Ok(succeeded) - } -} - -/// Validate that `event_id` is a 64-character lowercase hex string, as -/// required by the Nostr protocol (SHA-256 of the serialised event). -/// -/// Rejects the input early to avoid sending a malformed path segment to -/// Typesense, which could otherwise produce confusing 404 or 400 responses, -/// or — if the value contains `/` or `?` — accidentally hit a different API -/// endpoint. -fn validate_event_id(event_id: &str) -> Result<(), SearchError> { - if event_id.len() != 64 { - return Err(SearchError::InvalidEventId(format!( - "event_id must be 64 hex characters, got {} characters", - event_id.len() - ))); - } - if !event_id.chars().all(|c| c.is_ascii_hexdigit()) { - return Err(SearchError::InvalidEventId( - "event_id must contain only hex characters (0-9, a-f)".into(), - )); - } - Ok(()) -} - -/// Deletes an event from the index by event ID hex string. -pub async fn delete_event( - client: &reqwest::Client, - base_url: &str, - api_key: &str, - collection_name: &str, - event_id: &str, -) -> Result<(), SearchError> { - validate_event_id(event_id)?; - - let url = format!( - "{}/collections/{}/documents/{}", - base_url, collection_name, event_id - ); - - debug!( - event_id = event_id, - collection = collection_name, - "deleting event from index" - ); - - let resp = client - .delete(&url) - .header("X-TYPESENSE-API-KEY", api_key) - .send() - .await?; - - match resp.status().as_u16() { - 200 => Ok(()), - 404 => { - debug!( - event_id = event_id, - "event not found in index (already deleted)" - ); - Ok(()) - } - status => { - let body = resp.text().await.unwrap_or_default(); - Err(SearchError::Api { status, body }) - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - use buzz_core::event::StoredEvent; - use nostr::{EventBuilder, Keys, Kind}; - use uuid::Uuid; - - fn make_stored_event(content: &str, kind: Kind, channel_id: Option) -> StoredEvent { - let keys = Keys::generate(); - let event = EventBuilder::new(kind, content) - .tags([]) - .sign_with_keys(&keys) - .expect("signing failed"); - StoredEvent::new(event, channel_id) - } - - #[test] - fn document_fields_correct() { - let channel_id = Uuid::new_v4(); - let stored = make_stored_event("hello world", Kind::TextNote, Some(channel_id)); - let doc = event_to_document(&stored).unwrap(); - - assert_eq!(doc["id"].as_str().unwrap(), stored.event.id.to_string()); - assert_eq!(doc["content"].as_str().unwrap(), "hello world"); - assert_eq!(doc["kind"].as_i64().unwrap(), 1i64); - assert_eq!(doc["channel_id"].as_str().unwrap(), channel_id.to_string()); - assert!(doc["created_at"].as_i64().is_some()); - assert!(doc["channel_id"].is_string()); - } - - #[test] - fn document_no_channel_id_uses_global_sentinel() { - let stored = make_stored_event("no channel", Kind::TextNote, None); - let doc = event_to_document(&stored).unwrap(); - assert_eq!(doc["channel_id"].as_str().unwrap(), "__global__"); - } - - #[test] - fn kind0_appends_display_name_for_tokenization() { - let stored = make_stored_event( - r#"{"display_name":"alice","about":"loves cats"}"#, - Kind::Metadata, - None, - ); - let doc = event_to_document(&stored).unwrap(); - let content = doc["content"].as_str().unwrap(); - // Original JSON is preserved (read paths don't depend on this but it - // costs nothing and makes debugging the index cheaper). - assert!(content.contains(r#""display_name":"alice""#)); - // The display name is also present as a free-standing token so the - // default Typesense tokenizer can index it without the leading-quote - // gluing onto the next character. - assert!(content.ends_with(" alice"), "got: {content:?}"); - } - - #[test] - fn kind0_appends_name_when_display_name_absent() { - // NIP-01 allows `name` as the canonical display field too. - let stored = make_stored_event(r#"{"name":"bob","about":"x"}"#, Kind::Metadata, None); - let doc = event_to_document(&stored).unwrap(); - let content = doc["content"].as_str().unwrap(); - assert!(content.ends_with(" bob"), "got: {content:?}"); - } - - #[test] - fn kind0_includes_both_display_name_and_name_when_present() { - let stored = make_stored_event( - r#"{"display_name":"Alice","name":"alice"}"#, - Kind::Metadata, - None, - ); - let doc = event_to_document(&stored).unwrap(); - let content = doc["content"].as_str().unwrap(); - assert!(content.ends_with(" Alice alice"), "got: {content:?}"); - } - - #[test] - fn kind0_includes_nip05_in_appended_tokens() { - let stored = make_stored_event( - r#"{"display_name":"alice","nip05":"alice@example.com"}"#, - Kind::Metadata, - None, - ); - let doc = event_to_document(&stored).unwrap(); - let content = doc["content"].as_str().unwrap(); - assert!( - content.ends_with(" alice alice@example.com"), - "got: {content:?}" - ); - } - - #[test] - fn kind0_excludes_about_and_website_from_appended_tokens() { - // `about` and `website` deliberately do not get appended — including - // them would cause name-prefix searches to return false positives from - // bios. The user's own display_name still appears. - let stored = make_stored_event( - r#"{"display_name":"alice","about":"I work with bob on x","website":"https://carol.example"}"#, - Kind::Metadata, - None, - ); - let doc = event_to_document(&stored).unwrap(); - let content = doc["content"].as_str().unwrap(); - assert!(content.ends_with(" alice"), "got: {content:?}"); - // Sanity: the about/website are still in the doc because we preserve - // the original JSON — they just don't appear in the trailing tokens. - assert!(content.contains("bob")); - assert!(content.contains("carol")); - } - - #[test] - fn kind0_malformed_json_is_passed_through_unchanged() { - let stored = make_stored_event("not json at all", Kind::Metadata, None); - let doc = event_to_document(&stored).unwrap(); - assert_eq!(doc["content"].as_str().unwrap(), "not json at all"); - } - - #[test] - fn kind0_non_object_json_is_passed_through_unchanged() { - // Defensive: NIP-01 says content is a JSON object, but a malformed - // client could publish e.g. a JSON array. We don't crash, we just - // skip the flattening for that doc. - let stored = make_stored_event(r#"["nope"]"#, Kind::Metadata, None); - let doc = event_to_document(&stored).unwrap(); - assert_eq!(doc["content"].as_str().unwrap(), r#"["nope"]"#); - } - - #[test] - fn kind0_empty_string_values_skipped() { - let stored = make_stored_event( - r#"{"display_name":"","name":"alice","nip05":" "}"#, - Kind::Metadata, - None, - ); - let doc = event_to_document(&stored).unwrap(); - let content = doc["content"].as_str().unwrap(); - // Only `name` is non-empty; whitespace-only `nip05` is also skipped. - assert!(content.ends_with(" alice"), "got: {content:?}"); - } - - #[test] - fn kind0_no_searchable_fields_is_passed_through() { - // Profile with only fields we don't extract. - let stored = make_stored_event( - r#"{"about":"just a bio","picture":"https://x"}"#, - Kind::Metadata, - None, - ); - let doc = event_to_document(&stored).unwrap(); - let content = doc["content"].as_str().unwrap(); - // No trailing space-separated tokens added; original content unchanged. - assert_eq!(content, r#"{"about":"just a bio","picture":"https://x"}"#); - } - - #[test] - fn non_kind0_events_not_flattened() { - // kind:1 (note) with a JSON-looking body must be left strictly alone. - let json_looking = r#"{"display_name":"alice"}"#; - let stored = make_stored_event(json_looking, Kind::TextNote, None); - let doc = event_to_document(&stored).unwrap(); - assert_eq!(doc["content"].as_str().unwrap(), json_looking); - } - - #[test] - fn tag_flattening_uses_unit_separator() { - let keys = Keys::generate(); - let tag = nostr::Tag::parse(["e", "abc123def456"]).expect("tag parse"); - let event = EventBuilder::new(Kind::TextNote, "tagged") - .tags([tag]) - .sign_with_keys(&keys) - .expect("sign"); - let stored = StoredEvent::new(event, None); - let doc = event_to_document(&stored).unwrap(); - - let tags_flat = doc["tags_flat"].as_array().unwrap(); - assert!(!tags_flat.is_empty()); - // Must use \x1f, not colon, to avoid ambiguity with values containing colons. - let entry = tags_flat[0].as_str().unwrap(); - assert!( - entry.contains('\x1f'), - "expected unit separator in tag entry: {entry:?}" - ); - assert!( - !entry.contains(':') || entry.starts_with("http"), - "colon used as delimiter" - ); - assert!(entry.contains("abc123def456")); - } - - #[test] - fn delete_event_rejects_invalid_id() { - // Too short - assert!(matches!( - validate_event_id("abc123"), - Err(SearchError::InvalidEventId(_)) - )); - // Right length but non-hex character - let bad = "g".repeat(64); - assert!(matches!( - validate_event_id(&bad), - Err(SearchError::InvalidEventId(_)) - )); - // Valid 64-char hex - let good = "a".repeat(64); - assert!(validate_event_id(&good).is_ok()); - // Uppercase hex should also be accepted - let upper = "A".repeat(64); - assert!(validate_event_id(&upper).is_ok()); - // Path injection attempt - assert!(matches!( - validate_event_id("../admin"), - Err(SearchError::InvalidEventId(_)) - )); - } - - #[test] - fn tag_with_colon_value_not_ambiguous() { - let keys = Keys::generate(); - // "r" tag with a URL value containing colons - let tag = nostr::Tag::parse(["r", "wss://relay.example.com"]).expect("tag parse"); - let event = EventBuilder::new(Kind::TextNote, "relay ref") - .tags([tag]) - .sign_with_keys(&keys) - .expect("sign"); - let stored = StoredEvent::new(event, None); - let doc = event_to_document(&stored).unwrap(); - - let tags_flat = doc["tags_flat"].as_array().unwrap(); - let entry = tags_flat[0].as_str().unwrap(); - // With \x1f delimiter, splitting on \x1f gives exactly ["r", "wss://relay.example.com"] - let parts: Vec<&str> = entry.splitn(2, '\x1f').collect(); - assert_eq!(parts[0], "r"); - assert_eq!(parts[1], "wss://relay.example.com"); - } -} diff --git a/crates/buzz-search/src/lib.rs b/crates/buzz-search/src/lib.rs index 2a9c3b4182..ca5e14ec1e 100644 --- a/crates/buzz-search/src/lib.rs +++ b/crates/buzz-search/src/lib.rs @@ -1,309 +1,54 @@ #![deny(unsafe_code)] #![warn(missing_docs)] -//! Buzz search — Typesense integration for full-text event search. +//! Buzz search — community-scoped Postgres full-text search over Buzz events. +//! +//! The index lives in the `events` table: `search_tsv TSVECTOR GENERATED +//! ALWAYS AS (to_tsvector('simple', content)) STORED`, with `GIN +//! (search_tsv)` as the access path. Because the column is `GENERATED ALWAYS`, +//! every row write *is* the index update — there is no separate indexer, no +//! mpsc queue, no reindex job, no consistency window to reason about. A +//! client cannot forge the tsvector out of sync with the content it signed. +//! +//! This crate is the **query** side. Indexing is the SQL row insert — owned +//! by `buzz-db`. The relay refetches canonical events through `buzz-db`'s +//! scoped fetcher and runs access checks per hit; search is never the access +//! boundary (conformance row 50). +//! +//! ## Multi-tenant fence +//! +//! Every [`SearchQuery`] carries a [`CommunityId`]. There is no construction +//! path through this crate that omits it, and every SQL execution binds +//! `community_id = $ctx` as the first predicate. A query bound to community +//! A cannot return events stored under community B, by construction. -/// Typesense collection schema management. -pub mod collection; /// Search error types. pub mod error; -/// Event indexing helpers. -pub mod index; /// Search query execution. pub mod query; +pub use buzz_core::CommunityId; pub use error::SearchError; -pub use query::{SearchHit, SearchQuery, SearchResult}; +pub use query::{search, SearchHit, SearchQuery, SearchResult}; -use buzz_core::event::StoredEvent; +use sqlx::PgPool; -/// Configuration for the Typesense search backend. +/// Thin handle around a `PgPool` for community-scoped FTS. /// -/// [`SearchConfig::default`] reads from environment variables so that no -/// credentials are ever hardcoded in source: -/// -/// | Field | Environment variable | Default (dev only) | -/// |--------------|-------------------------|--------------------------| -/// | `url` | `TYPESENSE_URL` | `http://localhost:8108` | -/// | `api_key` | `TYPESENSE_API_KEY` | `buzz_dev_key` | -/// | `collection` | `TYPESENSE_COLLECTION` | `events` | -/// -/// In production, always set `TYPESENSE_API_KEY` explicitly. The fallback -/// value `buzz_dev_key` is intentionally weak and only suitable for local -/// development with a locally-running Typesense instance. +/// Holds nothing the pool itself doesn't already own. The whole purpose of +/// this type is a stable injection point for the relay's `AppState`. #[derive(Debug, Clone)] -pub struct SearchConfig { - /// Typesense base URL (e.g. `http://localhost:8108`). - pub url: String, - /// Typesense API key. - pub api_key: String, - /// Collection name to use for events. - pub collection: String, -} - -impl Default for SearchConfig { - fn default() -> Self { - Self { - url: std::env::var("TYPESENSE_URL").unwrap_or_else(|_| "http://localhost:8108".into()), - api_key: std::env::var("TYPESENSE_API_KEY").unwrap_or_else(|_| "buzz_dev_key".into()), - collection: std::env::var("TYPESENSE_COLLECTION").unwrap_or_else(|_| "events".into()), - } - } -} - -#[derive(Debug, Clone)] -/// Typesense search client. pub struct SearchService { - client: reqwest::Client, - config: SearchConfig, + pool: PgPool, } impl SearchService { - /// Creates a new `SearchService` with a default HTTP client. - pub fn new(config: SearchConfig) -> Self { - // SAFETY: default builder with only timeout/connect_timeout config cannot fail - let client = reqwest::Client::builder() - .timeout(std::time::Duration::from_secs(10)) - .connect_timeout(std::time::Duration::from_secs(5)) - .build() - .expect("SAFETY: default builder with only timeout config cannot fail"); - Self { client, config } + /// Build a search service over an existing Postgres pool. + pub fn new(pool: PgPool) -> Self { + Self { pool } } - /// Creates a `SearchService` with an explicit HTTP client (useful in tests). - pub fn with_client(client: reqwest::Client, config: SearchConfig) -> Self { - Self { client, config } - } - - /// Idempotent — safe to call on every startup. - pub async fn ensure_collection(&self) -> Result<(), SearchError> { - collection::ensure_collection( - &self.client, - &self.config.url, - &self.config.api_key, - &self.config.collection, - ) - .await - } - - /// Indexes a single event (upsert semantics). - pub async fn index_event(&self, event: &StoredEvent) -> Result<(), SearchError> { - index::index_event( - &self.client, - &self.config.url, - &self.config.api_key, - &self.config.collection, - event, - ) - .await - } - - /// Indexes a batch of events. Returns the number successfully indexed. - pub async fn index_batch(&self, events: &[StoredEvent]) -> Result { - index::index_batch( - &self.client, - &self.config.url, - &self.config.api_key, - &self.config.collection, - events, - ) - .await - } - - /// Executes a search query and returns matching results. + /// Execute a community-scoped FTS query. pub async fn search(&self, query: &SearchQuery) -> Result { - query::search( - &self.client, - &self.config.url, - &self.config.api_key, - &self.config.collection, - query, - ) - .await - } - - /// Removes an event from the search index by its event ID hex string. - pub async fn delete_event(&self, event_id: &str) -> Result<(), SearchError> { - index::delete_event( - &self.client, - &self.config.url, - &self.config.api_key, - &self.config.collection, - event_id, - ) - .await - } - - /// Checks that the Typesense server is reachable and healthy. - pub async fn health_check(&self) -> Result<(), SearchError> { - let url = format!("{}/health", self.config.url); - let resp = self - .client - .get(&url) - .header("X-TYPESENSE-API-KEY", &self.config.api_key) - .send() - .await?; - - let status = resp.status().as_u16(); - if status == 200 { - Ok(()) - } else { - let body = resp.text().await.unwrap_or_default(); - Err(SearchError::Api { status, body }) - } - } -} - -#[cfg(test)] -mod integration_tests { - use super::*; - use nostr::{EventBuilder, Keys, Kind}; - use uuid::Uuid; - - async fn typesense_available() -> bool { - let client = reqwest::Client::new(); - client - .get("http://localhost:8108/health") - .header("X-TYPESENSE-API-KEY", "buzz_dev_key") - .timeout(std::time::Duration::from_secs(2)) - .send() - .await - .map(|r| r.status().is_success()) - .unwrap_or(false) - } - - fn make_service(collection: &str) -> SearchService { - SearchService::new(SearchConfig { - url: "http://localhost:8108".into(), - api_key: "buzz_dev_key".into(), - collection: collection.to_string(), - }) - } - - fn make_stored_event(content: &str, kind: Kind) -> StoredEvent { - let keys = Keys::generate(); - let event = EventBuilder::new(kind, content) - .tags([]) - .sign_with_keys(&keys) - .expect("signing failed"); - StoredEvent::new(event, None) - } - - async fn drop_collection(service: &SearchService) { - let url = format!( - "{}/collections/{}", - service.config.url, service.config.collection - ); - let _ = service - .client - .delete(&url) - .header("X-TYPESENSE-API-KEY", &service.config.api_key) - .send() - .await; - } - - #[tokio::test] - #[ignore = "requires Typesense"] - async fn ensure_collection_idempotent() { - if !typesense_available().await { - return; - } - let collection = format!("events_test_{}", Uuid::new_v4().simple()); - let service = make_service(&collection); - service.ensure_collection().await.expect("first call"); - service - .ensure_collection() - .await - .expect("idempotency check"); - drop_collection(&service).await; - } - - #[tokio::test] - #[ignore = "requires Typesense"] - async fn index_and_search_roundtrip() { - if !typesense_available().await { - return; - } - let collection = format!("events_test_{}", Uuid::new_v4().simple()); - let service = make_service(&collection); - service.ensure_collection().await.unwrap(); - - let unique_token = format!("buzz_search_test_{}", Uuid::new_v4().simple()); - let stored = make_stored_event(&format!("hello {}", unique_token), Kind::TextNote); - let event_id = stored.event.id.to_string(); - - service.index_event(&stored).await.unwrap(); - tokio::time::sleep(std::time::Duration::from_millis(500)).await; - - let result = service - .search(&SearchQuery { - q: unique_token.clone(), - ..Default::default() - }) - .await - .unwrap(); - - assert!(result.found >= 1); - assert_eq!(result.hits[0].event_id, event_id); - assert!(result.hits[0].content.contains(&unique_token)); - - drop_collection(&service).await; - } - - #[tokio::test] - #[ignore = "requires Typesense"] - async fn index_batch_and_delete() { - if !typesense_available().await { - return; - } - let collection = format!("events_test_{}", Uuid::new_v4().simple()); - let service = make_service(&collection); - service.ensure_collection().await.unwrap(); - - let events: Vec = (0..5) - .map(|i| make_stored_event(&format!("batch event {i}"), Kind::TextNote)) - .collect(); - let count = service.index_batch(&events).await.unwrap(); - assert_eq!(count, 5); - - let stored = make_stored_event("to be deleted", Kind::TextNote); - let event_id = stored.event.id.to_string(); - service.index_event(&stored).await.unwrap(); - tokio::time::sleep(std::time::Duration::from_millis(300)).await; - service.delete_event(&event_id).await.unwrap(); - service.delete_event(&event_id).await.unwrap(); // idempotent - - drop_collection(&service).await; - } - - #[tokio::test] - #[ignore = "requires Typesense"] - async fn search_with_kind_filter() { - if !typesense_available().await { - return; - } - let collection = format!("events_test_{}", Uuid::new_v4().simple()); - let service = make_service(&collection); - service.ensure_collection().await.unwrap(); - - let unique = format!("filter_test_{}", Uuid::new_v4().simple()); - let event_k1 = make_stored_event(&format!("{unique} kind1"), Kind::TextNote); - let event_k42 = make_stored_event(&format!("{unique} kind42"), Kind::from(42u16)); - service.index_batch(&[event_k1, event_k42]).await.unwrap(); - tokio::time::sleep(std::time::Duration::from_millis(500)).await; - - let result = service - .search(&SearchQuery { - q: unique.clone(), - filter_by: Some("kind:=1".into()), - ..Default::default() - }) - .await - .unwrap(); - - for hit in &result.hits { - assert_eq!(hit.kind, 1); - } - - drop_collection(&service).await; + query::search(&self.pool, query).await } } diff --git a/crates/buzz-search/src/query.rs b/crates/buzz-search/src/query.rs index c440d7bc04..d794c95d33 100644 --- a/crates/buzz-search/src/query.rs +++ b/crates/buzz-search/src/query.rs @@ -1,411 +1,218 @@ -//! Search query building and result parsing. +//! NIP-50 search query against Postgres FTS, community-scoped. +//! +//! The relay never trusts a hit by itself: this layer returns canonical event +//! ids ordered by relevance, the relay refetches `StoredEvent`s through +//! buzz-db's `(community_id, event_id)` scoped fetcher, and runs the access +//! predicate (`search_hit_accepted` in `bridge.rs`) per hit. Search is never +//! the access boundary — it cannot widen visibility. +//! +//! See conformance row 50. -use serde::Deserialize; -use tracing::debug; +use buzz_core::CommunityId; +use sqlx::{PgPool, QueryBuilder, Row}; +use uuid::Uuid; use crate::error::SearchError; -/// Parameters for a Typesense search request. +/// A community-scoped FTS query. +/// +/// The community is REQUIRED at the type level — there is no construction path +/// that omits it. This is the search-side expression of conformance row zero: +/// every search call carries the server-resolved tenant, never client input. #[derive(Debug, Clone)] pub struct SearchQuery { - /// The search query string (`"*"` matches all documents). + /// Server-resolved community. Required. + pub community: CommunityId, + /// NIP-50 search text. Empty string is rejected by `search()` early + /// (no hits, no SQL roundtrip). pub q: String, - /// Optional Typesense filter expression (e.g. `"kind:=1"`). - pub filter_by: Option, - /// Optional sort expression (e.g. `"created_at:desc"`). - pub sort_by: Option, - /// Page number (1-indexed). + /// Restrict hits to one of these channel UUIDs. `None` = no channel + /// constraint (community-global within the community). An empty `Some(vec![])` + /// is also treated as "no channel constraint" — call sites that mean + /// "no channels are accessible" must short-circuit before calling. + pub channel_ids: Option>, + /// If `true`, include channel-less events (channel_id IS NULL) in addition + /// to any `channel_ids` filter. If `channel_ids` is `None`, this is + /// implicitly satisfied. Maps to today's `__global__` sentinel semantic. + pub include_channel_less: bool, + /// NIP-01 kinds filter. None = no kind constraint. + pub kinds: Option>, + /// NIP-01 authors filter (32-byte pubkeys). None = no author constraint. + pub authors: Option>>, + /// NIP-01 since (Unix seconds). Inclusive lower bound on created_at. + pub since: Option, + /// NIP-01 until (Unix seconds). Inclusive upper bound on created_at. + pub until: Option, + /// 1-indexed page number. pub page: u32, - /// Number of results per page. + /// Page size. Clamped at 500 internally. pub per_page: u32, } -impl Default for SearchQuery { - fn default() -> Self { - Self { - q: "*".into(), - filter_by: None, - sort_by: Some("created_at:desc".into()), - page: 1, - per_page: 20, - } - } -} - -impl SearchQuery { - /// Converts the query into Typesense HTTP query parameters. - pub fn to_query_params(&self) -> Vec<(String, String)> { - let mut params = vec![ - ("q".into(), self.q.clone()), - ("query_by".into(), "content".into()), - ("page".into(), self.page.to_string()), - ("per_page".into(), self.per_page.to_string()), - ]; - - if let Some(ref filter) = self.filter_by { - params.push(("filter_by".into(), filter.clone())); - } - - if let Some(ref sort) = self.sort_by { - params.push(("sort_by".into(), sort.clone())); - } - - params - } -} - -/// A single search result hit. +/// A single FTS hit. The relay refetches the canonical `StoredEvent` and +/// re-authorizes; this struct is just enough to drive that fetch and preserve +/// relevance ordering. #[derive(Debug, Clone)] pub struct SearchHit { - /// Hex event ID of the matching event. - pub event_id: String, - /// Event content text **as indexed in Typesense** — not necessarily the - /// canonical event content. - /// - /// For kind:0 (user metadata) events, `flatten_kind0_for_indexing` in - /// `index.rs` appends the parsed `display_name` / `name` / `nip05` values - /// to the original JSON content (space-separated) so the default - /// tokenizer can produce clean word tokens. That doctored string is what - /// lands here. - /// - /// All production read paths (`bridge.rs::handle_bridge_search`, - /// `handlers/req.rs` WS REQ) refetch the canonical `StoredEvent` from - /// Postgres by `event_id` and ignore this field — which is why the - /// append-to-content trick is safe. If you're adding a new feature that - /// reads this field directly, do the same: fetch the canonical event by - /// id rather than trusting `content` to round-trip. - pub content: String, - /// Nostr kind number. - pub kind: u16, - /// Hex public key of the event author. - pub pubkey: String, - /// Channel UUID string, if the event is scoped to a channel. - pub channel_id: Option, - /// Unix timestamp of event creation. + /// 32-byte event id. + pub event_id: [u8; 32], + /// Nostr kind. + pub kind: i32, + /// 32-byte pubkey of author. + pub pubkey: [u8; 32], + /// Optional channel UUID. `None` = channel-less event. + pub channel_id: Option, + /// Unix seconds. pub created_at: i64, - /// Typesense relevance score. - pub score: f64, + /// `ts_rank_cd` relevance score (higher = better). + pub rank: f32, } -/// The result of a search query. +/// Result of a search. #[derive(Debug, Clone)] pub struct SearchResult { - /// Matching hits for this page. + /// Hits on this page, ordered by relevance then created_at desc. pub hits: Vec, - /// Total number of matching documents across all pages. - pub found: u64, - /// Current page number. + /// 1-indexed page returned. pub page: u32, } -#[derive(Debug, Deserialize)] -struct TypesenseMultiSearchResponse { - results: Vec, -} +const PER_PAGE_MAX: u32 = 500; +const PER_PAGE_DEFAULT: u32 = 100; -#[derive(Debug, Deserialize)] -#[serde(untagged)] -enum TypesenseSearchResult { - Ok(TypesenseSearchResponse), - Error(TypesenseSearchError), -} +/// Execute a community-scoped FTS query. +/// +/// SQL shape (always): +/// ```sql +/// SELECT id, kind, pubkey, channel_id, EXTRACT(EPOCH FROM created_at)::bigint AS created_at_s, +/// ts_rank_cd(search_tsv, query) AS rank +/// FROM events, +/// websearch_to_tsquery('simple', $q) AS query +/// WHERE community_id = $ctx +/// AND deleted_at IS NULL +/// AND search_tsv @@ query +/// [+ channel scope, kinds, authors, since, until] +/// ORDER BY rank DESC, created_at DESC, id +/// LIMIT $per_page OFFSET (($page - 1) * $per_page) +/// ``` +/// +/// `community_id = $ctx` is the first predicate and is non-negotiable. There +/// is no code path through this function that omits it. +pub async fn search(pool: &PgPool, query: &SearchQuery) -> Result { + let trimmed = query.q.trim(); + if trimmed.is_empty() { + return Ok(SearchResult { + hits: Vec::new(), + page: query.page.max(1), + }); + } -#[derive(Debug, Deserialize)] -struct TypesenseSearchError { - code: u16, - error: String, -} + let per_page = query.per_page.clamp(1, PER_PAGE_MAX); + let per_page_actual = if query.per_page == 0 { + PER_PAGE_DEFAULT + } else { + per_page + }; + let page = query.page.max(1); + let offset = ((page - 1) as i64) * (per_page_actual as i64); -#[derive(Debug, Deserialize)] -struct TypesenseSearchResponse { - found: u64, - page: u32, - hits: Vec, -} - -#[derive(Debug, Deserialize)] -struct TypesenseHit { - document: TypesenseDocument, - #[serde(rename = "text_match")] - text_match: Option, -} - -#[derive(Debug, Deserialize)] -struct TypesenseDocument { - id: String, - content: String, - kind: i32, - pubkey: String, - channel_id: Option, - created_at: i64, -} - -/// Executes a search query against Typesense and returns parsed results. -pub async fn search( - client: &reqwest::Client, - base_url: &str, - api_key: &str, - collection_name: &str, - query: &SearchQuery, -) -> Result { - debug!( - q = %query.q, - page = query.page, - per_page = query.per_page, - collection = collection_name, - "Executing search" + let mut qb: QueryBuilder = QueryBuilder::new( + "SELECT id, kind, pubkey, channel_id, \ + EXTRACT(EPOCH FROM created_at)::bigint AS created_at_s, \ + ts_rank_cd(search_tsv, query) AS rank \ + FROM events, websearch_to_tsquery('simple', ", ); + qb.push_bind(trimmed); + qb.push(") AS query WHERE community_id = "); + qb.push_bind(*query.community.as_uuid()); + qb.push(" AND deleted_at IS NULL AND search_tsv @@ query"); - // Typesense GET search has a 4000-char query string limit. When filter_by - // contains hundreds of channel UUIDs, the URL exceeds this. Use the - // /multi_search POST endpoint which accepts the same params in a JSON body. - let url = format!("{}/multi_search", base_url); - let mut search_params = serde_json::json!({ - "collection": collection_name, - "q": query.q, - "query_by": "content", - "page": query.page, - "per_page": query.per_page, - }); - if let Some(ref filter) = query.filter_by { - search_params["filter_by"] = serde_json::Value::String(filter.clone()); - } - if let Some(ref sort) = query.sort_by { - search_params["sort_by"] = serde_json::Value::String(sort.clone()); - } - let body = serde_json::json!({ "searches": [search_params] }); - - let resp = client - .post(&url) - .header("X-TYPESENSE-API-KEY", api_key) - .json(&body) - .send() - .await?; - - let status = resp.status().as_u16(); - if status != 200 { - let body = resp.text().await.unwrap_or_default(); - return Err(SearchError::Api { status, body }); - } - - // multi_search wraps results: {"results": []}. Individual - // searches can fail inside an HTTP 200 response as `{code, error}`; surface - // those as API errors instead of deserializing them as JSON errors so callers - // and logs show the actual Typesense failure. - let wrapper: TypesenseMultiSearchResponse = resp.json().await?; - let ts_resp = wrapper.results.into_iter().next().ok_or(SearchError::Api { - status: 200, - body: "empty multi_search results".into(), - })?; - match ts_resp { - TypesenseSearchResult::Ok(response) => parse_response(response), - TypesenseSearchResult::Error(error) => Err(SearchError::Api { - status: error.code, - body: error.error, - }), - } -} - -fn parse_response(ts_resp: TypesenseSearchResponse) -> Result { - let hits = ts_resp - .hits - .into_iter() - .map(|hit| { - // Raw Typesense text_match relevance score (not normalized). - let score = hit.text_match.unwrap_or(0) as f64; - SearchHit { - event_id: hit.document.id, - content: hit.document.content, - kind: u16::try_from(hit.document.kind).unwrap_or(0), - pubkey: hit.document.pubkey, - channel_id: hit.document.channel_id.filter(|id| id != "__global__"), - created_at: hit.document.created_at, - score, + // Channel scope. Three shapes: + // - channel_ids = Some([..]) + include_channel_less = true: (channel_id = ANY($) OR channel_id IS NULL) + // - channel_ids = Some([..]) + include_channel_less = false: channel_id = ANY($) + // - channel_ids = None + include_channel_less = true: (no constraint — also covers None/false for callers + // that explicitly want "no channel scope at all") + // - channel_ids = None + include_channel_less = false: caller meant "nothing accessible" but didn't + // short-circuit; we conservatively return no hits + match (&query.channel_ids, query.include_channel_less) { + (Some(ids), include_global) if !ids.is_empty() => { + qb.push(" AND (channel_id = ANY("); + qb.push_bind(ids.clone()); + if include_global { + qb.push(") OR channel_id IS NULL)"); + } else { + qb.push("))"); } - }) - .collect(); + } + (Some(_), true) | (None, true) => { + // No channel constraint — include everything in the community. + // (channel_ids = Some(empty) falls here because no IDs to filter + // by and channel-less events are included.) + } + (Some(_), false) | (None, false) => { + // Caller said "no accessible channels and exclude channel-less" — + // produces an empty result. + return Ok(SearchResult { + hits: Vec::new(), + page, + }); + } + } - Ok(SearchResult { - hits, - found: ts_resp.found, - page: ts_resp.page, - }) -} - -#[cfg(test)] -mod tests { - use super::*; - use serde_json::json; - - #[test] - fn test_search_query_building() { - let q = SearchQuery { - q: "hello world".into(), - filter_by: Some("kind:=1".into()), - sort_by: Some("created_at:desc".into()), - page: 2, - per_page: 10, - }; - - let params = q.to_query_params(); - let get = |key: &str| -> Option { - params - .iter() - .find(|(k, _)| k == key) - .map(|(_, v)| v.clone()) - }; - - assert_eq!(get("q").unwrap(), "hello world"); - assert_eq!(get("query_by").unwrap(), "content"); - assert_eq!(get("page").unwrap(), "2"); - assert_eq!(get("per_page").unwrap(), "10"); - assert_eq!(get("filter_by").unwrap(), "kind:=1"); - assert_eq!(get("sort_by").unwrap(), "created_at:desc"); - } - - #[test] - fn test_search_query_no_optional_fields() { - let q = SearchQuery { - q: "*".into(), - filter_by: None, - sort_by: None, - page: 1, - per_page: 20, - }; - - let params = q.to_query_params(); - let has_key = |key: &str| params.iter().any(|(k, _)| k == key); - - assert!(has_key("q")); - assert!(has_key("query_by")); - assert!(has_key("page")); - assert!(has_key("per_page")); - assert!(!has_key("filter_by")); - assert!(!has_key("sort_by")); - } - - #[test] - fn test_search_result_parsing() { - let raw = json!({ - "found": 42, - "page": 1, - "hits": [ - { - "document": { - "id": "abc123", - "content": "hello buzz", - "kind": 1, - "pubkey": "deadbeef", - "channel_id": "chan-uuid", - "created_at": 1700000000i64, - "tags_flat": ["e:ref123"] - }, - "text_match": 578730123i64 - }, - { - "document": { - "id": "def456", - "content": "another message", - "kind": 42, - "pubkey": "cafebabe", - "channel_id": null, - "created_at": 1700000100i64, - "tags_flat": [] - }, - "text_match": null - } - ] - }); - - let ts_resp: TypesenseSearchResponse = serde_json::from_value(raw).expect("should parse"); - let result = parse_response(ts_resp).expect("should succeed"); - - assert_eq!(result.found, 42); - assert_eq!(result.page, 1); - assert_eq!(result.hits.len(), 2); - - let h0 = &result.hits[0]; - assert_eq!(h0.event_id, "abc123"); - assert_eq!(h0.content, "hello buzz"); - assert_eq!(h0.kind, 1); - assert_eq!(h0.pubkey, "deadbeef"); - assert_eq!(h0.channel_id.as_deref(), Some("chan-uuid")); - assert_eq!(h0.created_at, 1700000000); - assert!(h0.score > 0.0); - - let h1 = &result.hits[1]; - assert_eq!(h1.event_id, "def456"); - assert_eq!(h1.kind, 42); - assert!(h1.channel_id.is_none()); - assert_eq!(h1.score, 0.0); // null text_match → 0 - } - - #[test] - fn test_search_result_empty() { - let raw = json!({ - "found": 0, - "page": 1, - "hits": [] - }); - - let ts_resp: TypesenseSearchResponse = serde_json::from_value(raw).expect("should parse"); - let result = parse_response(ts_resp).expect("should succeed"); - - assert_eq!(result.found, 0); - assert!(result.hits.is_empty()); - } - - #[test] - fn test_multi_search_result_success_parses() { - let raw = json!({ - "results": [{ - "found": 1, - "page": 1, - "hits": [{ - "document": { - "id": "abc123", - "content": "hello buzz", - "kind": 1, - "pubkey": "deadbeef", - "channel_id": "chan-uuid", - "created_at": 1700000000i64, - "tags_flat": [] - }, - "text_match": 578730123i64 - }] - }] - }); - - let wrapper: TypesenseMultiSearchResponse = - serde_json::from_value(raw).expect("should parse multi_search success result"); - let response = match wrapper.results.into_iter().next().expect("one result") { - TypesenseSearchResult::Ok(response) => response, - TypesenseSearchResult::Error(err) => panic!("expected success result, got {err:?}"), - }; - - let result = parse_response(response).expect("should parse response"); - assert_eq!(result.found, 1); - assert_eq!(result.hits.len(), 1); - assert_eq!(result.hits[0].event_id, "abc123"); - } - - #[test] - fn test_multi_search_result_error_parses() { - let raw = json!({ - "results": [{ - "code": 400, - "error": "Could not find a filter field named `channel_id` in the schema." - }] - }); - - let wrapper: TypesenseMultiSearchResponse = - serde_json::from_value(raw).expect("should parse multi_search error result"); - let err = match wrapper.results.into_iter().next().expect("one result") { - TypesenseSearchResult::Ok(_) => panic!("expected error result"), - TypesenseSearchResult::Error(err) => err, - }; - - assert_eq!(err.code, 400); - assert!(err.error.contains("channel_id")); - } + if let Some(ref kinds) = query.kinds { + if !kinds.is_empty() { + qb.push(" AND kind = ANY("); + qb.push_bind(kinds.clone()); + qb.push(")"); + } + } + + if let Some(ref authors) = query.authors { + if !authors.is_empty() { + qb.push(" AND pubkey = ANY("); + qb.push_bind(authors.clone()); + qb.push(")"); + } + } + + if let Some(since) = query.since { + qb.push(" AND created_at >= to_timestamp("); + qb.push_bind(since); + qb.push(")"); + } + + if let Some(until) = query.until { + qb.push(" AND created_at <= to_timestamp("); + qb.push_bind(until); + qb.push(")"); + } + + qb.push(" ORDER BY rank DESC, created_at DESC, id LIMIT "); + qb.push_bind(per_page_actual as i64); + qb.push(" OFFSET "); + qb.push_bind(offset); + + let rows = qb.build().fetch_all(pool).await?; + + let mut hits = Vec::with_capacity(rows.len()); + for row in rows { + let id_bytes: Vec = row.try_get("id")?; + let pk_bytes: Vec = row.try_get("pubkey")?; + let id: [u8; 32] = id_bytes.try_into().map_err(|v: Vec| { + sqlx::Error::Decode(format!("event id column is {} bytes, expected 32", v.len()).into()) + })?; + let pubkey: [u8; 32] = pk_bytes.try_into().map_err(|v: Vec| { + sqlx::Error::Decode(format!("pubkey column is {} bytes, expected 32", v.len()).into()) + })?; + hits.push(SearchHit { + event_id: id, + kind: row.try_get("kind")?, + pubkey, + channel_id: row.try_get("channel_id")?, + created_at: row.try_get("created_at_s")?, + rank: row.try_get("rank")?, + }); + } + + Ok(SearchResult { hits, page }) } diff --git a/crates/buzz-search/tests/fts_integration.rs b/crates/buzz-search/tests/fts_integration.rs new file mode 100644 index 0000000000..99992892f6 --- /dev/null +++ b/crates/buzz-search/tests/fts_integration.rs @@ -0,0 +1,598 @@ +//! Integration tests for community-scoped Postgres FTS. +//! +//! Run with a local PG: `BUZZ_TEST_DATABASE_URL=postgres://buzz:buzz_dev@localhost:5432/buzz cargo test -p buzz-search --tests -- --include-ignored` +//! +//! Each test creates a uniquely-named schema, applies the consolidated `0001` +//! migration into it, exercises a scenario, and drops it. Tests are +//! parallel-safe. + +use buzz_core::CommunityId; +use buzz_search::{SearchQuery, SearchService}; +use sqlx::{postgres::PgPoolOptions, Executor, PgPool}; +use uuid::Uuid; + +const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; +const MIGRATION_SQL: &str = include_str!("../../../migrations/0001_initial_schema.sql"); + +async fn setup() -> (PgPool, String) { + let url = std::env::var("BUZZ_TEST_DATABASE_URL").unwrap_or_else(|_| TEST_DB_URL.to_string()); + let schema = format!("fts_test_{}", Uuid::new_v4().simple()); + // Connect to the default schema first to create the test schema. + let admin_pool = PgPoolOptions::new() + .max_connections(1) + .connect(&url) + .await + .expect("connect"); + let create_sql = format!("CREATE SCHEMA \"{schema}\""); + sqlx::query(sqlx::AssertSqlSafe(create_sql)) + .execute(&admin_pool) + .await + .expect("create schema"); + admin_pool.close().await; + + // Connect with search_path set so the migration's CREATE TABLE lands here. + let url_with_search_path = format!("{url}?options=-c%20search_path%3D{schema}"); + let pool = PgPoolOptions::new() + .max_connections(2) + .connect(&url_with_search_path) + .await + .expect("connect with search_path"); + pool.execute(MIGRATION_SQL) + .await + .expect("apply 0001 migration"); + (pool, schema) +} + +async fn teardown(pool: PgPool, schema: &str) { + pool.close().await; + let admin_pool = PgPoolOptions::new() + .max_connections(1) + .connect( + &std::env::var("BUZZ_TEST_DATABASE_URL").unwrap_or_else(|_| TEST_DB_URL.to_string()), + ) + .await + .expect("reconnect for drop"); + let drop_sql = format!("DROP SCHEMA \"{schema}\" CASCADE"); + sqlx::query(sqlx::AssertSqlSafe(drop_sql)) + .execute(&admin_pool) + .await + .expect("drop schema"); + admin_pool.close().await; +} + +/// Insert a community row, return its id. +async fn mk_community(pool: &PgPool, host: &str) -> CommunityId { + let id = Uuid::new_v4(); + sqlx::query("INSERT INTO communities (id, host, signing_key) VALUES ($1, $2, $3)") + .bind(id) + .bind(host) + .bind(b"signingkey".as_slice()) + .execute(pool) + .await + .expect("insert community"); + CommunityId::from_uuid(id) +} + +/// Insert a minimal event. `created_at_secs` is unix seconds. +#[allow(clippy::too_many_arguments)] +async fn insert_event( + pool: &PgPool, + community: CommunityId, + id: [u8; 32], + pubkey: [u8; 32], + kind: i32, + content: &str, + channel_id: Option, + created_at_secs: i64, +) { + sqlx::query( + "INSERT INTO events (community_id, id, pubkey, created_at, kind, tags, content, sig, channel_id) \ + VALUES ($1, $2, $3, to_timestamp($4), $5, '[]'::jsonb, $6, $7, $8)", + ) + .bind(community.as_uuid()) + .bind(&id[..]) + .bind(&pubkey[..]) + .bind(created_at_secs) + .bind(kind) + .bind(content) + .bind(b"signature".as_slice()) + .bind(channel_id) + .execute(pool) + .await + .expect("insert event"); +} + +fn rand_bytes32() -> [u8; 32] { + let mut out = [0u8; 32]; + let u = Uuid::new_v4(); + let bytes = u.as_bytes(); + out[..16].copy_from_slice(bytes); + out[16..].copy_from_slice(bytes); + out +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn search_finds_event_in_same_community() { + let (pool, schema) = setup().await; + + let c_a = mk_community(&pool, "a.example").await; + let evt_id = rand_bytes32(); + let pk = rand_bytes32(); + insert_event( + &pool, + c_a, + evt_id, + pk, + 1, + "hello wonderland — buzz everyone", + None, + 1700000000, + ) + .await; + + let svc = SearchService::new(pool.clone()); + let result = svc + .search(&SearchQuery { + community: c_a, + q: "wonderland".into(), + channel_ids: None, + include_channel_less: true, + kinds: None, + authors: None, + since: None, + until: None, + page: 1, + per_page: 10, + }) + .await + .expect("search ok"); + + assert_eq!(result.hits.len(), 1); + assert_eq!(result.hits[0].event_id, evt_id); + assert_eq!(result.hits[0].kind, 1); + assert_eq!(result.hits[0].created_at, 1700000000); + assert!(result.hits[0].rank > 0.0); + + teardown(pool, &schema).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn search_does_not_return_other_community_events() { + // The load-bearing test: event indexed under community A, query bound to + // community B → zero hits. + let (pool, schema) = setup().await; + + let c_a = mk_community(&pool, "a.example").await; + let c_b = mk_community(&pool, "b.example").await; + let pk = rand_bytes32(); + insert_event( + &pool, + c_a, + rand_bytes32(), + pk, + 1, + "only-in-a unique-token-xyz", + None, + 1700000000, + ) + .await; + + let svc = SearchService::new(pool.clone()); + let result_a = svc + .search(&SearchQuery { + community: c_a, + q: "unique-token-xyz".into(), + channel_ids: None, + include_channel_less: true, + kinds: None, + authors: None, + since: None, + until: None, + page: 1, + per_page: 10, + }) + .await + .unwrap(); + assert_eq!(result_a.hits.len(), 1, "A should see its own event"); + + let result_b = svc + .search(&SearchQuery { + community: c_b, + q: "unique-token-xyz".into(), + channel_ids: None, + include_channel_less: true, + kinds: None, + authors: None, + since: None, + until: None, + page: 1, + per_page: 10, + }) + .await + .unwrap(); + assert_eq!(result_b.hits.len(), 0, "B must not see A's event"); + + teardown(pool, &schema).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn kind0_search_by_display_name_works_without_flattening() { + // The case I worried might regress without the kind:0 content-flattening + // hack. Postgres FTS over raw JSON content tokenizes through the + // punctuation and finds display_name/nip05 values. + let (pool, schema) = setup().await; + + let c = mk_community(&pool, "a.example").await; + let pk = rand_bytes32(); + insert_event( + &pool, + c, + rand_bytes32(), + pk, + 0, + r#"{"display_name":"Alice Wonderland","name":"alice","nip05":"alice@buzz.app","about":"hello"}"#, + None, + 1700000000, + ) + .await; + + let svc = SearchService::new(pool.clone()); + for q in ["wonderland", "alice", "alice@buzz.app"] { + let r = svc + .search(&SearchQuery { + community: c, + q: q.to_string(), + channel_ids: None, + include_channel_less: true, + kinds: Some(vec![0]), + authors: None, + since: None, + until: None, + page: 1, + per_page: 10, + }) + .await + .unwrap(); + assert_eq!(r.hits.len(), 1, "kind:0 query {q:?} should find Alice"); + } + + teardown(pool, &schema).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn channel_scope_restricts_results() { + let (pool, schema) = setup().await; + + let c = mk_community(&pool, "a.example").await; + // Channels need community_id since channels PK is (community_id, id). + let ch_a = Uuid::new_v4(); + let ch_b = Uuid::new_v4(); + sqlx::query("INSERT INTO channels (community_id, id, name, channel_type, created_by) VALUES ($1, $2, $3, 'stream'::channel_type, $4), ($1, $5, $6, 'stream'::channel_type, $4)") + .bind(c.as_uuid()) + .bind(ch_a) + .bind("ch-a") + .bind(b"\x01".repeat(32)) + .bind(ch_b) + .bind("ch-b") + .execute(&pool) + .await + .expect("insert channels"); + + let pk = rand_bytes32(); + insert_event( + &pool, + c, + rand_bytes32(), + pk, + 1, + "shared-token in ch-a", + Some(ch_a), + 1700000000, + ) + .await; + insert_event( + &pool, + c, + rand_bytes32(), + pk, + 1, + "shared-token in ch-b", + Some(ch_b), + 1700000001, + ) + .await; + insert_event( + &pool, + c, + rand_bytes32(), + pk, + 1, + "shared-token global", + None, + 1700000002, + ) + .await; + + let svc = SearchService::new(pool.clone()); + + // restrict to ch_a, exclude global + let r = svc + .search(&SearchQuery { + community: c, + q: "shared-token".into(), + channel_ids: Some(vec![ch_a]), + include_channel_less: false, + kinds: None, + authors: None, + since: None, + until: None, + page: 1, + per_page: 10, + }) + .await + .unwrap(); + assert_eq!(r.hits.len(), 1); + assert_eq!(r.hits[0].channel_id, Some(ch_a)); + + // restrict to ch_a + include global + let r = svc + .search(&SearchQuery { + community: c, + q: "shared-token".into(), + channel_ids: Some(vec![ch_a]), + include_channel_less: true, + kinds: None, + authors: None, + since: None, + until: None, + page: 1, + per_page: 10, + }) + .await + .unwrap(); + assert_eq!(r.hits.len(), 2); + + // no channel constraint + let r = svc + .search(&SearchQuery { + community: c, + q: "shared-token".into(), + channel_ids: None, + include_channel_less: true, + kinds: None, + authors: None, + since: None, + until: None, + page: 1, + per_page: 10, + }) + .await + .unwrap(); + assert_eq!(r.hits.len(), 3); + + // empty accessible channels + exclude global = zero + let r = svc + .search(&SearchQuery { + community: c, + q: "shared-token".into(), + channel_ids: Some(vec![]), + include_channel_less: false, + kinds: None, + authors: None, + since: None, + until: None, + page: 1, + per_page: 10, + }) + .await + .unwrap(); + assert_eq!(r.hits.len(), 0); + + teardown(pool, &schema).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn deleted_events_are_excluded() { + let (pool, schema) = setup().await; + + let c = mk_community(&pool, "a.example").await; + let evt_id = rand_bytes32(); + let pk = rand_bytes32(); + insert_event(&pool, c, evt_id, pk, 1, "deleted-token-q", None, 1700000000).await; + + // Soft-delete + sqlx::query("UPDATE events SET deleted_at = NOW() WHERE community_id = $1 AND id = $2") + .bind(c.as_uuid()) + .bind(&evt_id[..]) + .execute(&pool) + .await + .expect("delete"); + + let svc = SearchService::new(pool.clone()); + let r = svc + .search(&SearchQuery { + community: c, + q: "deleted-token-q".into(), + channel_ids: None, + include_channel_less: true, + kinds: None, + authors: None, + since: None, + until: None, + page: 1, + per_page: 10, + }) + .await + .unwrap(); + assert!( + r.hits.is_empty(), + "soft-deleted events must not appear in FTS" + ); + + teardown(pool, &schema).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn empty_query_returns_empty_result_no_roundtrip() { + let (pool, schema) = setup().await; + let c = mk_community(&pool, "a.example").await; + let svc = SearchService::new(pool.clone()); + + for q in ["", " "] { + let r = svc + .search(&SearchQuery { + community: c, + q: q.into(), + channel_ids: None, + include_channel_less: true, + kinds: None, + authors: None, + since: None, + until: None, + page: 1, + per_page: 10, + }) + .await + .unwrap(); + assert!( + r.hits.is_empty(), + "empty/whitespace query must return no hits" + ); + } + + teardown(pool, &schema).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn since_until_filters() { + let (pool, schema) = setup().await; + + let c = mk_community(&pool, "a.example").await; + let pk = rand_bytes32(); + insert_event( + &pool, + c, + rand_bytes32(), + pk, + 1, + "time-token-zz at A", + None, + 1_700_000_000, + ) + .await; + insert_event( + &pool, + c, + rand_bytes32(), + pk, + 1, + "time-token-zz at B", + None, + 1_700_010_000, + ) + .await; + insert_event( + &pool, + c, + rand_bytes32(), + pk, + 1, + "time-token-zz at C", + None, + 1_700_020_000, + ) + .await; + + let svc = SearchService::new(pool.clone()); + + let r = svc + .search(&SearchQuery { + community: c, + q: "time-token-zz".into(), + channel_ids: None, + include_channel_less: true, + kinds: None, + authors: None, + since: Some(1_700_005_000), + until: Some(1_700_015_000), + page: 1, + per_page: 10, + }) + .await + .unwrap(); + assert_eq!(r.hits.len(), 1); + assert_eq!(r.hits[0].created_at, 1_700_010_000); + + teardown(pool, &schema).await; +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn pagination_works() { + let (pool, schema) = setup().await; + + let c = mk_community(&pool, "a.example").await; + let pk = rand_bytes32(); + for i in 0..7 { + insert_event( + &pool, + c, + rand_bytes32(), + pk, + 1, + "page-token-qq", + None, + 1_700_000_000 + i, + ) + .await; + } + + let svc = SearchService::new(pool.clone()); + + let r = svc + .search(&SearchQuery { + community: c, + q: "page-token-qq".into(), + channel_ids: None, + include_channel_less: true, + kinds: None, + authors: None, + since: None, + until: None, + page: 1, + per_page: 3, + }) + .await + .unwrap(); + assert_eq!(r.hits.len(), 3); + + let r2 = svc + .search(&SearchQuery { + community: c, + q: "page-token-qq".into(), + channel_ids: None, + include_channel_less: true, + kinds: None, + authors: None, + since: None, + until: None, + page: 3, + per_page: 3, + }) + .await + .unwrap(); + assert_eq!( + r2.hits.len(), + 1, + "page 3 of 7 with per_page=3 should have 1 hit" + ); + + teardown(pool, &schema).await; +}