diff --git a/Cargo.lock b/Cargo.lock index 05ff37be38..150be25dd8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1088,10 +1088,12 @@ version = "0.1.0" dependencies = [ "buzz-core", "chrono", + "hex", "nostr", "reqwest 0.13.3", "serde", "serde_json", + "sqlx", "thiserror 2.0.18", "tokio", "tracing", diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index d6c45f2d3e..8600e9faa5 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -198,6 +198,14 @@ impl Db { Self { pool } } + /// Returns a clone of the underlying connection pool. + /// + /// Used by the Postgres FTS backend in `buzz-search` so it can run + /// queries against the same pool as the rest of the relay. + pub fn pool(&self) -> PgPool { + self.pool.clone() + } + /// Run pending database migrations. pub async fn migrate(&self) -> Result<()> { migration::run_migrations(&self.pool).await diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index f4f3c9ab3e..1188bd9170 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -128,7 +128,7 @@ mod tests { fn embedded_migrator_contains_all_schema_migrations() { let migrations: Vec<_> = MIGRATOR.iter().collect(); - assert_eq!(migrations.len(), 3); + assert_eq!(migrations.len(), 4); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!( @@ -160,6 +160,17 @@ mod tests { && migrations[2].sql.as_str().contains("idx_events_not_before"), "third migration should add the NIP-ER reminder columns and index" ); + + assert_eq!(migrations[3].version, 4); + assert_eq!(&*migrations[3].description, "search fts"); + assert!( + migrations[3].sql.as_str().contains("content_tsv tsvector") + && migrations[3] + .sql + .as_str() + .contains("idx_events_content_tsv"), + "fourth migration should add the generated tsvector column and GIN index" + ); } async fn connect_test_pool() -> PgPool { @@ -192,24 +203,33 @@ mod tests { .expect("read applied migrations") } - /// Returns `schema/schema.sql` with the NIP-ER reminder DDL removed, so it - /// models a pre-stack deployment whose `events` table lacks the reminder - /// columns and index. The strip is asserted: if the snapshot text drifts so - /// these fragments no longer match, the test fails loudly rather than - /// silently loading a snapshot that already carries the reminder columns - /// (which would make migration 0003 collide on re-add). + /// Returns `schema/schema.sql` with the NIP-ER reminder DDL and the + /// search-FTS DDL removed, so it models a pre-stack deployment whose + /// `events` table lacks the reminder columns and the generated tsvector + /// column. The strip is asserted: if the snapshot text drifts so these + /// fragments no longer match, the test fails loudly rather than silently + /// loading a snapshot that already carries the columns (which would make + /// migration 0003 or 0004 collide on re-add). fn pre_reminder_schema_snapshot() -> String { const REMINDER_COLUMNS: &str = " not_before BIGINT,\n delivered_at BIGINT,\n"; const REMINDER_INDEX: &str = "CREATE INDEX idx_events_not_before ON events (not_before)\n WHERE not_before IS NOT NULL AND deleted_at IS NULL AND delivered_at IS NULL;\n"; + const FTS_COLUMN: &str = " content_tsv tsvector\n GENERATED ALWAYS AS (to_tsvector('simple', content)) STORED,\n"; + const FTS_INDEX: &str = + "CREATE INDEX idx_events_content_tsv ON events USING GIN (content_tsv);\n"; assert!( - SCHEMA_SQL.contains(REMINDER_COLUMNS) && SCHEMA_SQL.contains(REMINDER_INDEX), - "schema.sql reminder DDL drifted; update pre_reminder_schema_snapshot to match" + SCHEMA_SQL.contains(REMINDER_COLUMNS) + && SCHEMA_SQL.contains(REMINDER_INDEX) + && SCHEMA_SQL.contains(FTS_COLUMN) + && SCHEMA_SQL.contains(FTS_INDEX), + "schema.sql reminder/FTS DDL drifted; update pre_reminder_schema_snapshot to match" ); SCHEMA_SQL .replace(REMINDER_COLUMNS, "") .replace(REMINDER_INDEX, "") + .replace(FTS_COLUMN, "") + .replace(FTS_INDEX, "") } #[tokio::test] @@ -220,7 +240,7 @@ mod tests { run_migrations(&pool).await.expect("run migrations"); - assert_eq!(applied_versions(&pool).await, vec![1, 2, 3]); + assert_eq!(applied_versions(&pool).await, vec![1, 2, 3, 4]); let events_exists = sqlx::query_scalar::<_, bool>( "SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = 'public' AND table_name = 'events')", ) @@ -253,7 +273,7 @@ mod tests { run_migrations(&pool).await.expect("baseline migrations"); - assert_eq!(applied_versions(&pool).await, vec![1, 2, 3]); + assert_eq!(applied_versions(&pool).await, vec![1, 2, 3, 4]); let allowlist_count = sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM pubkey_allowlist") .fetch_one(&pool) .await diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 6006f78a14..788ff7accf 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -719,9 +719,10 @@ async fn handle_bridge_search( continue; } - // Build Typesense filter — push channel scope + NIP-01 constraints. + // Build the backend-neutral search filter — push channel scope + + // NIP-01 constraints into structured fields. let h_tag = nostr::SingleLetterTag::lowercase(nostr::Alphabet::H); - let filter_channel_scope = + let filter_channel_scope: Vec = if let Some(vs) = filter.generic_tags.get(&h_tag).filter(|vs| !vs.is_empty()) { let valid: Vec = vs .iter() @@ -732,37 +733,31 @@ async fn handle_bridge_search( if valid.is_empty() { continue; // All #h values inaccessible — skip filter. } - format!("channel_id:=[{}]", valid.join(",")) + valid } else { channel_scope.clone() }; - let mut filter_parts = vec![filter_channel_scope]; - if let Some(ref kinds) = filter.kinds { - if !kinds.is_empty() { - let kind_vals: Vec = kinds.iter().map(|k| k.as_u16().to_string()).collect(); - filter_parts.push(format!("kind:=[{}]", kind_vals.join(","))); - } - } - if let Some(ref authors) = filter.authors { - if !authors.is_empty() { - let author_vals: Vec = authors.iter().map(|a| a.to_hex()).collect(); - filter_parts.push(format!("pubkey:=[{}]", author_vals.join(","))); - } - } - if let Some(since) = filter.since { - filter_parts.push(format!("created_at:>={}", since.as_secs())); - } - if let Some(until) = filter.until { - filter_parts.push(format!("created_at:<={}", until.as_secs())); - } - - let filter_by = filter_parts.join(" && "); + let kinds_vec: Vec = filter + .kinds + .as_ref() + .map(|ks| ks.iter().map(|k| k.as_u16()).collect()) + .unwrap_or_default(); + let authors_vec: Vec = filter + .authors + .as_ref() + .map(|auths| auths.iter().map(|a| a.to_hex()).collect()) + .unwrap_or_default(); + let since_secs = filter.since.map(|t| t.as_secs() as i64); + let until_secs = filter.until.map(|t| t.as_secs() as i64); let search_query = buzz_search::SearchQuery { q: search_text, - filter_by: Some(filter_by), - sort_by: None, // Typesense default = relevance + kinds: kinds_vec, + authors: authors_vec, + channel_ids: filter_channel_scope, + since: since_secs, + until: until_secs, page: 1, per_page: limit, }; diff --git a/crates/buzz-relay/src/config.rs b/crates/buzz-relay/src/config.rs index 714ced400f..df00563146 100644 --- a/crates/buzz-relay/src/config.rs +++ b/crates/buzz-relay/src/config.rs @@ -35,6 +35,8 @@ pub struct Config { pub typesense_url: String, /// Typesense API key. pub typesense_key: String, + /// Which search backend the relay should use. + pub search_backend: buzz_search::SearchBackend, /// Public WebSocket URL of this relay, advertised in NIP-11. pub relay_url: String, /// Maximum number of concurrent WebSocket connections. @@ -151,6 +153,15 @@ impl Config { let typesense_key = std::env::var("TYPESENSE_API_KEY").unwrap_or_else(|_| "buzz_dev_key".to_string()); + let search_backend = match std::env::var("BUZZ_SEARCH_BACKEND") { + Ok(raw) => buzz_search::SearchBackend::parse(&raw).map_err(|bad| { + ConfigError::InvalidValue(format!( + "BUZZ_SEARCH_BACKEND={bad:?} (expected `typesense`, `postgres`, or `disabled`)" + )) + })?, + Err(_) => buzz_search::SearchBackend::Typesense, + }; + let relay_url = std::env::var("RELAY_URL").unwrap_or_else(|_| "ws://localhost:3000".to_string()); @@ -377,6 +388,7 @@ impl Config { redis_url, typesense_url, typesense_key, + search_backend, relay_url, max_connections, max_concurrent_handlers, @@ -548,4 +560,39 @@ mod tests { Some("custom.example.com") ); } + + #[test] + fn search_backend_defaults_to_typesense() { + let _guard = ENV_MUTEX.lock().unwrap(); + std::env::remove_var("BUZZ_SEARCH_BACKEND"); + let config = Config::from_env().expect("default config"); + assert_eq!(config.search_backend, buzz_search::SearchBackend::Typesense); + } + + #[test] + fn search_backend_parses_postgres_and_disabled() { + let _guard = ENV_MUTEX.lock().unwrap(); + + std::env::set_var("BUZZ_SEARCH_BACKEND", "postgres"); + let config = Config::from_env().expect("config"); + assert_eq!(config.search_backend, buzz_search::SearchBackend::Postgres); + + std::env::set_var("BUZZ_SEARCH_BACKEND", "disabled"); + let config = Config::from_env().expect("config"); + assert_eq!(config.search_backend, buzz_search::SearchBackend::Disabled); + + std::env::remove_var("BUZZ_SEARCH_BACKEND"); + } + + #[test] + fn search_backend_rejects_unknown_value() { + let _guard = ENV_MUTEX.lock().unwrap(); + std::env::set_var("BUZZ_SEARCH_BACKEND", "postgress"); // typo + let result = Config::from_env(); + std::env::remove_var("BUZZ_SEARCH_BACKEND"); + assert!( + matches!(result, Err(ConfigError::InvalidValue(ref msg)) if msg.contains("BUZZ_SEARCH_BACKEND")), + "expected InvalidValue for BUZZ_SEARCH_BACKEND, got {result:?}", + ); + } } diff --git a/crates/buzz-relay/src/handlers/req.rs b/crates/buzz-relay/src/handlers/req.rs index 1893c3c2bc..5a18541473 100644 --- a/crates/buzz-relay/src/handlers/req.rs +++ b/crates/buzz-relay/src/handlers/req.rs @@ -277,27 +277,23 @@ const MAX_SEARCH_PAGES: u32 = 10; pub(crate) fn build_search_channel_scope_filter( accessible_channels: &[uuid::Uuid], include_global: bool, -) -> Option { +) -> Option> { if accessible_channels.is_empty() { return if include_global { - Some("channel_id:=__global__".to_string()) + Some(vec![buzz_search::GLOBAL_CHANNEL_SENTINEL.to_string()]) } else { None }; } - let ids: Vec = accessible_channels + let mut ids: Vec = accessible_channels .iter() .map(|id| id.to_string()) .collect(); - Some(if include_global { - format!( - "(channel_id:=[{}] || channel_id:=__global__)", - ids.join(",") - ) - } else { - format!("channel_id:=[{}]", ids.join(",")) - }) + if include_global { + ids.push(buzz_search::GLOBAL_CHANNEL_SENTINEL.to_string()); + } + Some(ids) } #[allow(clippy::too_many_arguments)] @@ -337,17 +333,17 @@ async fn handle_search_req( continue; // NIP-01: limit 0 means "no results from this filter" } - // Push as many NIP-01 constraints into Typesense as possible so - // post-filtering is a correction step, not the primary filter. + // Push as many NIP-01 constraints into the search backend as possible + // so post-filtering is a correction step, not the primary filter. // - // If the filter has a #h tag, push the specific channel(s) into Typesense - // instead of the full accessible set. This prevents cross-channel hits from - // consuming pagination budget and causing under-fetch. - // If the filter has #h, intersect with accessible channels. If all #h - // values are invalid/inaccessible, skip the filter entirely (match nothing) - // rather than broadening to all channels. + // If the filter has a #h tag, push the specific channel(s) into the + // search instead of the full accessible set. This prevents + // cross-channel hits from consuming pagination budget and causing + // under-fetch. If all #h values are invalid/inaccessible, skip the + // filter entirely (match nothing) rather than broadening to all + // channels. let h_tag = nostr::SingleLetterTag::lowercase(nostr::Alphabet::H); - let channel_scope = + let channel_scope: Vec = if let Some(vs) = filter.generic_tags.get(&h_tag).filter(|vs| !vs.is_empty()) { let valid: Vec = vs .iter() @@ -358,31 +354,22 @@ async fn handle_search_req( if valid.is_empty() { continue; // all #h values invalid/inaccessible — skip filter } - format!("channel_id:=[{}]", valid.join(",")) + valid } else { all_channels_filter.clone() }; - let mut filter_parts = vec![channel_scope]; - if let Some(ref kinds) = filter.kinds { - if !kinds.is_empty() { - let kind_vals: Vec = kinds.iter().map(|k| k.as_u16().to_string()).collect(); - filter_parts.push(format!("kind:=[{}]", kind_vals.join(","))); - } - } - if let Some(ref authors) = filter.authors { - if !authors.is_empty() { - let author_vals: Vec = authors.iter().map(|a| a.to_hex()).collect(); - filter_parts.push(format!("pubkey:=[{}]", author_vals.join(","))); - } - } - if let Some(since) = filter.since { - filter_parts.push(format!("created_at:>={}", since.as_secs())); - } - if let Some(until) = filter.until { - filter_parts.push(format!("created_at:<={}", until.as_secs())); - } - - let filter_by = filter_parts.join(" && "); + let kinds_vec: Vec = filter + .kinds + .as_ref() + .map(|ks| ks.iter().map(|k| k.as_u16()).collect()) + .unwrap_or_default(); + let authors_vec: Vec = filter + .authors + .as_ref() + .map(|auths| auths.iter().map(|a| a.to_hex()).collect()) + .unwrap_or_default(); + let since_secs = filter.since.map(|t| t.as_secs() as i64); + let until_secs = filter.until.map(|t| t.as_secs() as i64); // Paginate: keep fetching pages until we've emitted `limit` results // or exhausted the search result set. This ensures post-filtering @@ -399,8 +386,11 @@ async fn handle_search_req( let search_query = buzz_search::SearchQuery { q: search_text.clone(), - filter_by: Some(filter_by.clone()), - sort_by: None, // Typesense default = relevance (text_match score) + kinds: kinds_vec.clone(), + authors: authors_vec.clone(), + channel_ids: channel_scope.clone(), + since: since_secs, + until: until_secs, page, per_page, }; @@ -1071,7 +1061,7 @@ mod tests { let scope = build_search_channel_scope_filter(&[channel_id], false) .expect("restricted tokens with channel access should still search that channel"); - assert_eq!(scope, format!("channel_id:=[{channel_id}]")); + assert_eq!(scope, vec![channel_id.to_string()]); } #[test] diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index b555c3d9de..bfeea5f2b5 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -8,7 +8,7 @@ use buzz_audit::AuditService; use buzz_auth::AuthService; use buzz_db::{Db, DbConfig}; use buzz_pubsub::PubSubManager; -use buzz_search::{SearchConfig, SearchService}; +use buzz_search::{SearchBackend, SearchConfig, SearchService}; use buzz_relay::config::Config; use buzz_relay::metrics as relay_metrics; @@ -189,15 +189,29 @@ async fn main() -> anyhow::Result<()> { let auth = AuthService::new(config.auth.clone()); - let search_config = SearchConfig { - url: config.typesense_url.clone(), - api_key: config.typesense_key.clone(), - collection: std::env::var("TYPESENSE_COLLECTION").unwrap_or_else(|_| "events".to_string()), + let search = match config.search_backend { + SearchBackend::Typesense => { + let search_config = SearchConfig { + url: config.typesense_url.clone(), + api_key: config.typesense_key.clone(), + collection: std::env::var("TYPESENSE_COLLECTION") + .unwrap_or_else(|_| "events".to_string()), + }; + let service = SearchService::new(search_config); + if let Err(e) = service.ensure_collection().await { + error!("Typesense collection setup failed (non-fatal): {e}"); + } + service + } + SearchBackend::Postgres => { + info!("Search backend: postgres (content_tsv generated column)"); + SearchService::with_postgres(db.pool()) + } + SearchBackend::Disabled => { + info!("Search backend: disabled (NIP-50 queries will return empty)"); + SearchService::disabled() + } }; - let search = SearchService::new(search_config); - if let Err(e) = search.ensure_collection().await { - error!("Typesense collection setup failed (non-fatal): {e}"); - } let workflow_config = buzz_workflow::WorkflowConfig::default(); let workflow_engine = Arc::new(WorkflowEngine::new(db.clone(), workflow_config)); diff --git a/crates/buzz-search/Cargo.toml b/crates/buzz-search/Cargo.toml index 3c571733a6..5e9e2d5e80 100644 --- a/crates/buzz-search/Cargo.toml +++ b/crates/buzz-search/Cargo.toml @@ -18,3 +18,5 @@ chrono = { workspace = true } tracing = { workspace = true } thiserror = { workspace = true } nostr = { workspace = true } +sqlx = { workspace = true } +hex = { workspace = true } diff --git a/crates/buzz-search/src/error.rs b/crates/buzz-search/src/error.rs index e17fb5f065..fd02f0958b 100644 --- a/crates/buzz-search/src/error.rs +++ b/crates/buzz-search/src/error.rs @@ -36,4 +36,8 @@ pub enum SearchError { /// The provided event ID is not valid hex. #[error("Invalid event_id: {0}")] InvalidEventId(String), + + /// A database error from sqlx (Postgres backend only). + #[error("Database error: {0}")] + Database(#[from] sqlx::Error), } diff --git a/crates/buzz-search/src/lib.rs b/crates/buzz-search/src/lib.rs index 2a9c3b4182..f52e404bbf 100644 --- a/crates/buzz-search/src/lib.rs +++ b/crates/buzz-search/src/lib.rs @@ -1,30 +1,77 @@ #![deny(unsafe_code)] #![warn(missing_docs)] -//! Buzz search — Typesense integration for full-text event search. +//! Buzz search — pluggable full-text event search. +//! +//! Two production backends and a no-op: +//! +//! - [`SearchService::new`] (Typesense): mirrors event content into a Typesense +//! collection via the indexing worker in `buzz-relay/src/state.rs`. The +//! `search()` path returns event IDs that the relay then refetches from +//! Postgres. +//! - [`SearchService::with_postgres`] (Postgres FTS): runs `plainto_tsquery` +//! against a generated `content_tsv` column on `events`. No write-path +//! indexing needed — the generated stored column auto-populates on INSERT. +//! - [`SearchService::disabled`]: returns empty results for every query and +//! accepts indexing calls as no-ops. Used when NIP-50 search is intentionally +//! off (e.g. for tenants who opted out). +//! +//! The choice is driven by [`SearchBackend`] on [`SearchConfig`]. /// Typesense collection schema management. pub mod collection; /// Search error types. pub mod error; -/// Event indexing helpers. +/// Event indexing helpers (Typesense). pub mod index; -/// Search query execution. +/// Postgres full-text-search backend. +pub mod postgres; +/// Search query types and Typesense execution. pub mod query; pub use error::SearchError; -pub use query::{SearchHit, SearchQuery, SearchResult}; +pub use query::{SearchHit, SearchQuery, SearchResult, GLOBAL_CHANNEL_SENTINEL}; use buzz_core::event::StoredEvent; +use sqlx::PgPool; -/// Configuration for the Typesense search backend. +/// Which search backend the relay should use. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SearchBackend { + /// Typesense (current production default). + Typesense, + /// Postgres full-text search via the `content_tsv` generated column. + Postgres, + /// NIP-50 search is disabled; every query returns empty. + Disabled, +} + +impl SearchBackend { + /// Parses a backend name from a string. Case-insensitive. Accepted values: + /// `typesense`, `postgres` (alias `pg`), `disabled` (aliases `off`, `none`). + /// Returns `Err()` on any other value so misconfiguration surfaces + /// loudly rather than silently falling back. + pub fn parse(s: &str) -> Result { + match s.to_ascii_lowercase().as_str() { + "typesense" => Ok(Self::Typesense), + "postgres" | "pg" => Ok(Self::Postgres), + "disabled" | "off" | "none" => Ok(Self::Disabled), + other => Err(other.to_string()), + } + } +} + +/// Configuration for the search backend. /// -/// [`SearchConfig::default`] reads from environment variables so that no -/// credentials are ever hardcoded in source: +/// Reading from env via [`SearchConfig::default`] selects the Typesense +/// backend by default. For Postgres or Disabled backends, set +/// `BUZZ_SEARCH_BACKEND` in `buzz-relay/src/config.rs` and pass the resolved +/// [`SearchBackend`] into the constructors directly — this struct only carries +/// Typesense-specific fields. /// /// | Field | Environment variable | Default (dev only) | /// |--------------|-------------------------|--------------------------| /// | `url` | `TYPESENSE_URL` | `http://localhost:8108` | -/// | `api_key` | `TYPESENSE_API_KEY` | `buzz_dev_key` | +/// | `api_key` | `TYPESENSE_API_KEY` | `buzz_dev_key` | /// | `collection` | `TYPESENSE_COLLECTION` | `events` | /// /// In production, always set `TYPESENSE_API_KEY` explicitly. The fallback @@ -50,15 +97,30 @@ impl Default for SearchConfig { } } +/// Internal Typesense client + config bundle. Construction is handled via +/// [`SearchService::new`] / [`SearchService::with_client`]; this type is +/// exposed only because it appears in a public enum variant. #[derive(Debug, Clone)] -/// Typesense search client. -pub struct SearchService { +pub struct TypesenseInner { client: reqwest::Client, config: SearchConfig, } +/// Pluggable search client. Construct via [`SearchService::new`] (Typesense), +/// [`SearchService::with_postgres`] (Postgres FTS), or +/// [`SearchService::disabled`] (no-op). +#[derive(Debug, Clone)] +pub enum SearchService { + /// Typesense backend. + Typesense(TypesenseInner), + /// Postgres FTS backend, holding the relay's existing `PgPool`. + Postgres(PgPool), + /// No-op backend; every search returns empty. + Disabled, +} + impl SearchService { - /// Creates a new `SearchService` with a default HTTP client. + /// Creates a Typesense `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() @@ -66,89 +128,161 @@ impl SearchService { .connect_timeout(std::time::Duration::from_secs(5)) .build() .expect("SAFETY: default builder with only timeout config cannot fail"); - Self { client, config } + Self::Typesense(TypesenseInner { client, config }) } - /// Creates a `SearchService` with an explicit HTTP client (useful in tests). + /// Creates a Typesense `SearchService` with an explicit HTTP client + /// (useful in tests). pub fn with_client(client: reqwest::Client, config: SearchConfig) -> Self { - Self { client, config } + Self::Typesense(TypesenseInner { client, config }) } + /// Creates a Postgres FTS `SearchService` backed by the supplied pool. + /// No indexing setup is required — the `content_tsv` generated column + /// populates on every INSERT. + pub fn with_postgres(pool: PgPool) -> Self { + Self::Postgres(pool) + } + + /// Creates a no-op `SearchService` that returns empty results for every + /// query. Useful when search is intentionally disabled. + pub fn disabled() -> Self { + Self::Disabled + } + + /// Ensures the backend is ready to serve search. + /// + /// - **Typesense**: creates the configured collection if it doesn't exist. + /// - **Postgres / Disabled**: no-op. + /// /// 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 + match self { + Self::Typesense(t) => { + collection::ensure_collection( + &t.client, + &t.config.url, + &t.config.api_key, + &t.config.collection, + ) + .await + } + Self::Postgres(_) | Self::Disabled => Ok(()), + } } /// Indexes a single event (upsert semantics). + /// + /// - **Typesense**: writes a document to the collection. + /// - **Postgres**: no-op — the `content_tsv` generated stored column is + /// populated automatically on the original INSERT. + /// - **Disabled**: no-op. 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 + match self { + Self::Typesense(t) => { + index::index_event( + &t.client, + &t.config.url, + &t.config.api_key, + &t.config.collection, + event, + ) + .await + } + Self::Postgres(_) | Self::Disabled => Ok(()), + } } /// Indexes a batch of events. Returns the number successfully indexed. + /// For Postgres and Disabled backends, returns `events.len()` (no work). 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 + match self { + Self::Typesense(t) => { + index::index_batch( + &t.client, + &t.config.url, + &t.config.api_key, + &t.config.collection, + events, + ) + .await + } + Self::Postgres(_) | Self::Disabled => Ok(events.len()), + } } /// Executes a search query and returns matching results. pub async fn search(&self, query: &SearchQuery) -> Result { - query::search( - &self.client, - &self.config.url, - &self.config.api_key, - &self.config.collection, - query, - ) - .await + match self { + Self::Typesense(t) => { + query::search( + &t.client, + &t.config.url, + &t.config.api_key, + &t.config.collection, + query, + ) + .await + } + Self::Postgres(pool) => postgres::search(pool, query).await, + Self::Disabled => Ok(SearchResult { + hits: Vec::new(), + found: 0, + page: query.page, + }), + } } /// Removes an event from the search index by its event ID hex string. + /// + /// - **Typesense**: deletes the document. + /// - **Postgres**: no-op — `content_tsv` is tied to the event row; + /// removing the row removes the index entry, and the relay's event + /// deletion path already handles that. + /// - **Disabled**: no-op. 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 + match self { + Self::Typesense(t) => { + index::delete_event( + &t.client, + &t.config.url, + &t.config.api_key, + &t.config.collection, + event_id, + ) + .await + } + Self::Postgres(_) | Self::Disabled => Ok(()), + } } - /// Checks that the Typesense server is reachable and healthy. + /// Checks that the backend 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?; + match self { + Self::Typesense(t) => { + let url = format!("{}/health", t.config.url); + let resp = t + .client + .get(&url) + .header("X-TYPESENSE-API-KEY", &t.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 }) + let status = resp.status().as_u16(); + if status == 200 { + Ok(()) + } else { + let body = resp.text().await.unwrap_or_default(); + Err(SearchError::Api { status, body }) + } + } + Self::Postgres(pool) => { + sqlx::query_scalar::<_, i32>("SELECT 1") + .fetch_one(pool) + .await?; + Ok(()) + } + Self::Disabled => Ok(()), } } } @@ -171,12 +305,19 @@ mod integration_tests { .unwrap_or(false) } - fn make_service(collection: &str) -> SearchService { - SearchService::new(SearchConfig { + fn make_service(collection: &str) -> (SearchService, SearchConfig, reqwest::Client) { + let config = SearchConfig { url: "http://localhost:8108".into(), api_key: "buzz_dev_key".into(), collection: collection.to_string(), - }) + }; + let client = reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(10)) + .connect_timeout(std::time::Duration::from_secs(5)) + .build() + .expect("client"); + let service = SearchService::with_client(client.clone(), config.clone()); + (service, config, client) } fn make_stored_event(content: &str, kind: Kind) -> StoredEvent { @@ -188,15 +329,11 @@ mod integration_tests { StoredEvent::new(event, None) } - async fn drop_collection(service: &SearchService) { - let url = format!( - "{}/collections/{}", - service.config.url, service.config.collection - ); - let _ = service - .client + async fn drop_collection(config: &SearchConfig, client: &reqwest::Client) { + let url = format!("{}/collections/{}", config.url, config.collection); + let _ = client .delete(&url) - .header("X-TYPESENSE-API-KEY", &service.config.api_key) + .header("X-TYPESENSE-API-KEY", &config.api_key) .send() .await; } @@ -208,13 +345,13 @@ mod integration_tests { return; } let collection = format!("events_test_{}", Uuid::new_v4().simple()); - let service = make_service(&collection); + let (service, config, client) = make_service(&collection); service.ensure_collection().await.expect("first call"); service .ensure_collection() .await .expect("idempotency check"); - drop_collection(&service).await; + drop_collection(&config, &client).await; } #[tokio::test] @@ -224,7 +361,7 @@ mod integration_tests { return; } let collection = format!("events_test_{}", Uuid::new_v4().simple()); - let service = make_service(&collection); + let (service, config, client) = make_service(&collection); service.ensure_collection().await.unwrap(); let unique_token = format!("buzz_search_test_{}", Uuid::new_v4().simple()); @@ -246,7 +383,7 @@ mod integration_tests { assert_eq!(result.hits[0].event_id, event_id); assert!(result.hits[0].content.contains(&unique_token)); - drop_collection(&service).await; + drop_collection(&config, &client).await; } #[tokio::test] @@ -256,7 +393,7 @@ mod integration_tests { return; } let collection = format!("events_test_{}", Uuid::new_v4().simple()); - let service = make_service(&collection); + let (service, config, client) = make_service(&collection); service.ensure_collection().await.unwrap(); let events: Vec = (0..5) @@ -272,7 +409,7 @@ mod integration_tests { service.delete_event(&event_id).await.unwrap(); service.delete_event(&event_id).await.unwrap(); // idempotent - drop_collection(&service).await; + drop_collection(&config, &client).await; } #[tokio::test] @@ -282,7 +419,7 @@ mod integration_tests { return; } let collection = format!("events_test_{}", Uuid::new_v4().simple()); - let service = make_service(&collection); + let (service, config, client) = make_service(&collection); service.ensure_collection().await.unwrap(); let unique = format!("filter_test_{}", Uuid::new_v4().simple()); @@ -294,7 +431,7 @@ mod integration_tests { let result = service .search(&SearchQuery { q: unique.clone(), - filter_by: Some("kind:=1".into()), + kinds: vec![1], ..Default::default() }) .await @@ -304,6 +441,28 @@ mod integration_tests { assert_eq!(hit.kind, 1); } - drop_collection(&service).await; + drop_collection(&config, &client).await; + } + + #[tokio::test] + async fn disabled_backend_returns_empty_results() { + let service = SearchService::disabled(); + let result = service.search(&SearchQuery::default()).await.unwrap(); + assert_eq!(result.found, 0); + assert!(result.hits.is_empty()); + + // index_* and delete_* are no-ops, but exercise them so future changes + // that re-introduce side effects break this test loudly. + let keys = nostr::Keys::generate(); + let event = nostr::EventBuilder::new(nostr::Kind::TextNote, "hello") + .tags([]) + .sign_with_keys(&keys) + .expect("sign"); + let stored = StoredEvent::new(event, None); + service.index_event(&stored).await.unwrap(); + assert_eq!(service.index_batch(&[stored]).await.unwrap(), 1); + service.delete_event("0".repeat(64).as_str()).await.unwrap(); + service.health_check().await.unwrap(); + service.ensure_collection().await.unwrap(); } } diff --git a/crates/buzz-search/src/postgres.rs b/crates/buzz-search/src/postgres.rs new file mode 100644 index 0000000000..00e6aa3cf5 --- /dev/null +++ b/crates/buzz-search/src/postgres.rs @@ -0,0 +1,296 @@ +//! Postgres full-text-search backend. +//! +//! Mirrors the read shape of the Typesense backend: returns event IDs (plus +//! the trivial metadata callers need to fetch the canonical event from +//! Postgres) so the relay's existing +//! `db.get_events_by_ids` → `filters_match` → auth post-filter chain in +//! `crates/buzz-relay/src/handlers/req.rs` is unchanged. +//! +//! Matching uses `plainto_tsquery('simple', $q)` against the `content_tsv` +//! generated column added in migration `0004_search_fts.sql`. Pushdowns: +//! +//! - `kinds` → `kind = ANY($kinds)` +//! - `authors` → `pubkey = ANY($authors)` (hex-decoded) +//! - `channel_ids` → `channel_id = ANY($chans)` with optional +//! `OR channel_id IS NULL` when the global sentinel +//! is present +//! - `since` / `until` → `created_at >= … AND created_at <= …` +//! +//! Results are ordered by `created_at DESC` — NIP-50 does not require strict +//! relevance ordering, and chronological ordering is what the existing Buzz +//! clients expect. + +use chrono::{DateTime, Utc}; +use sqlx::{PgPool, Row}; +use tracing::debug; + +use crate::error::SearchError; +use crate::query::{SearchHit, SearchQuery, SearchResult, GLOBAL_CHANNEL_SENTINEL}; + +/// Maximum `per_page` value the Postgres backend will honor in a single +/// query. The relay paginates internally; this is a guardrail against +/// pathological client input rather than a hard NIP-50 limit. +const MAX_PER_PAGE: u32 = 250; + +/// Executes a search query against Postgres FTS and returns parsed results. +/// +/// Returns `found` as the total number of matching rows (`COUNT(*) OVER ()` +/// window) so the relay's pagination logic can compute exhaustion the same +/// way it does against Typesense. +pub async fn search(pool: &PgPool, query: &SearchQuery) -> Result { + debug!( + q = %query.q, + page = query.page, + per_page = query.per_page, + "Executing Postgres FTS search" + ); + + let per_page = query.per_page.min(MAX_PER_PAGE).max(1) as i64; + let page = query.page.max(1) as i64; + let offset = (page - 1) * per_page; + + // Decode author hex once; reject any malformed hex by silently dropping it + // (matches Typesense behavior where an invalid pubkey simply fails to + // match anything). + let author_bytes: Vec> = query + .authors + .iter() + .filter_map(|hex_str| hex::decode(hex_str.trim()).ok()) + .filter(|b| b.len() == 32) + .collect(); + + // Channel filter splits the sentinel out so we can render + // (channel_id = ANY($chans) OR channel_id IS NULL) + let (channel_uuids, include_global) = split_channel_filter(&query.channel_ids); + + // Build the SQL with literal `ANY` arrays. Using QueryBuilder would be + // cleaner, but the dynamic shape (optional clauses) is small enough that + // hand-rolling a query with a stable parameter ordering is more readable + // and easier to audit. + let mut sql = String::from( + "SELECT id, pubkey, kind, channel_id, created_at, content, \ + COUNT(*) OVER () AS total \ + FROM events \ + WHERE deleted_at IS NULL", + ); + + // `simple` matches the tokenizer used by the `content_tsv` generated + // column in migration 0004. plainto_tsquery treats the input as a plain + // string (handles spaces, ignores punctuation) — closest analogue to + // Typesense's default `query_by=content` behavior. + let mut binds = Binds::new(); + let q_trim = query.q.trim(); + let has_text = !q_trim.is_empty() && q_trim != "*"; + if has_text { + let q_idx = binds.push_text(q_trim); + sql.push_str(&format!( + " AND content_tsv @@ plainto_tsquery('simple', ${q_idx})" + )); + } + + if !query.kinds.is_empty() { + let kinds_i32: Vec = query.kinds.iter().map(|k| *k as i32).collect(); + let idx = binds.push_kinds(kinds_i32); + sql.push_str(&format!(" AND kind = ANY(${idx})")); + } + + if !author_bytes.is_empty() { + let idx = binds.push_authors(author_bytes); + sql.push_str(&format!(" AND pubkey = ANY(${idx})")); + } + + if !channel_uuids.is_empty() && include_global { + let idx = binds.push_uuids(channel_uuids); + sql.push_str(&format!( + " AND (channel_id = ANY(${idx}) OR channel_id IS NULL)" + )); + } else if !channel_uuids.is_empty() { + let idx = binds.push_uuids(channel_uuids); + sql.push_str(&format!(" AND channel_id = ANY(${idx})")); + } else if include_global { + sql.push_str(" AND channel_id IS NULL"); + } + + if let Some(since) = query.since { + let idx = binds.push_timestamp(unix_to_timestamptz(since)); + sql.push_str(&format!(" AND created_at >= ${idx}")); + } + if let Some(until) = query.until { + let idx = binds.push_timestamp(unix_to_timestamptz(until)); + sql.push_str(&format!(" AND created_at <= ${idx}")); + } + + sql.push_str(" ORDER BY created_at DESC"); + let limit_idx = binds.push_i64(per_page); + sql.push_str(&format!(" LIMIT ${limit_idx}")); + let offset_idx = binds.push_i64(offset); + sql.push_str(&format!(" OFFSET ${offset_idx}")); + + // The SQL string is built only from static fragments and `$N` parameter + // placeholders — every dynamic value flows through the bind list. Wrap + // with `AssertSqlSafe` to satisfy sqlx 0.9's static-SQL lint. + let mut q = sqlx::query(sqlx::AssertSqlSafe(sql)); + q = binds.apply(q); + let rows = q.fetch_all(pool).await?; + + let mut found: u64 = 0; + let mut hits: Vec = Vec::with_capacity(rows.len()); + for row in rows { + let total: i64 = row.try_get("total").unwrap_or(0); + if found == 0 && total > 0 { + found = total as u64; + } + + let id: Vec = row.try_get("id")?; + let pubkey: Vec = row.try_get("pubkey")?; + let kind_i32: i32 = row.try_get("kind")?; + let channel_uuid: Option = row.try_get("channel_id")?; + let created_at: DateTime = row.try_get("created_at")?; + let content: String = row.try_get("content")?; + + hits.push(SearchHit { + event_id: hex::encode(&id), + content, + kind: u16::try_from(kind_i32).unwrap_or(0), + pubkey: hex::encode(&pubkey), + channel_id: channel_uuid.map(|u| u.to_string()), + created_at: created_at.timestamp(), + // Postgres backend doesn't expose a relevance score on the hit; + // ts_rank_cd is available but unused by the consumer (req.rs + // ignores `score` entirely). Leave at 0.0 for honesty. + score: 0.0, + }); + } + + Ok(SearchResult { + hits, + found, + page: query.page, + }) +} + +fn split_channel_filter(channel_ids: &[String]) -> (Vec, bool) { + let mut include_global = false; + let mut uuids: Vec = Vec::with_capacity(channel_ids.len()); + for id in channel_ids { + if id == GLOBAL_CHANNEL_SENTINEL { + include_global = true; + } else if let Ok(u) = uuid::Uuid::parse_str(id) { + uuids.push(u); + } + // silently drop malformed UUID strings — matches Typesense's behavior + // of treating an unknown channel as a non-match + } + (uuids, include_global) +} + +fn unix_to_timestamptz(seconds: i64) -> DateTime { + DateTime::::from_timestamp(seconds, 0).unwrap_or_else(|| { + // Either MIN or MAX, whichever side overflow lands on. Use epoch as + // a defensive fallback — Buzz events have second-precision created_at + // bound to NIP-01's u32 range, so this branch is effectively unreachable. + DateTime::::from_timestamp(0, 0).expect("epoch is valid") + }) +} + +/// Bookkeeping for SQL parameter ordering. Each `push_*` records the binding +/// and returns the 1-indexed parameter slot. `apply` re-binds them in order +/// on the final `sqlx::query` so the index labels in the SQL string line up. +struct Binds { + items: Vec, +} + +enum BoundValue { + Text(String), + KindList(Vec), + AuthorList(Vec>), + UuidList(Vec), + Timestamp(DateTime), + Int8(i64), +} + +impl Binds { + fn new() -> Self { + Self { items: Vec::new() } + } + fn next_idx(&self) -> usize { + self.items.len() + 1 + } + fn push_text(&mut self, s: &str) -> usize { + let i = self.next_idx(); + self.items.push(BoundValue::Text(s.to_string())); + i + } + fn push_kinds(&mut self, v: Vec) -> usize { + let i = self.next_idx(); + self.items.push(BoundValue::KindList(v)); + i + } + fn push_authors(&mut self, v: Vec>) -> usize { + let i = self.next_idx(); + self.items.push(BoundValue::AuthorList(v)); + i + } + fn push_uuids(&mut self, v: Vec) -> usize { + let i = self.next_idx(); + self.items.push(BoundValue::UuidList(v)); + i + } + fn push_timestamp(&mut self, t: DateTime) -> usize { + let i = self.next_idx(); + self.items.push(BoundValue::Timestamp(t)); + i + } + fn push_i64(&mut self, n: i64) -> usize { + let i = self.next_idx(); + self.items.push(BoundValue::Int8(n)); + i + } + fn apply<'q>( + self, + mut q: sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments>, + ) -> sqlx::query::Query<'q, sqlx::Postgres, sqlx::postgres::PgArguments> { + for item in self.items { + q = match item { + BoundValue::Text(s) => q.bind(s), + BoundValue::KindList(v) => q.bind(v), + BoundValue::AuthorList(v) => q.bind(v), + BoundValue::UuidList(v) => q.bind(v), + BoundValue::Timestamp(t) => q.bind(t), + BoundValue::Int8(n) => q.bind(n), + }; + } + q + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn split_channel_filter_extracts_global_sentinel() { + let (uuids, global) = split_channel_filter(&[ + "11111111-1111-1111-1111-111111111111".into(), + GLOBAL_CHANNEL_SENTINEL.to_string(), + "not-a-uuid".into(), + "22222222-2222-2222-2222-222222222222".into(), + ]); + assert!(global, "sentinel should be detected"); + assert_eq!(uuids.len(), 2, "malformed UUID should be dropped silently"); + } + + #[test] + fn split_channel_filter_no_sentinel_means_no_global() { + let (uuids, global) = + split_channel_filter(&["11111111-1111-1111-1111-111111111111".into()]); + assert!(!global); + assert_eq!(uuids.len(), 1); + } + + #[test] + fn unix_to_timestamptz_round_trips() { + let ts = unix_to_timestamptz(1_700_000_000); + assert_eq!(ts.timestamp(), 1_700_000_000); + } +} diff --git a/crates/buzz-search/src/query.rs b/crates/buzz-search/src/query.rs index c440d7bc04..0981eaaeac 100644 --- a/crates/buzz-search/src/query.rs +++ b/crates/buzz-search/src/query.rs @@ -5,15 +5,34 @@ use tracing::debug; use crate::error::SearchError; -/// Parameters for a Typesense search request. +/// Sentinel channel identifier used for events that are not scoped to any +/// channel. Mirrored verbatim from `crates/buzz-search/src/index.rs` so the +/// query layer can pass it through without depending on the indexer. +pub const GLOBAL_CHANNEL_SENTINEL: &str = "__global__"; + +/// Backend-neutral search query. +/// +/// Constructed by `req.rs` from a NIP-50 filter and passed to whichever +/// [`crate::SearchService`] backend is active. Each backend renders these +/// structured fields into its own filter syntax (Typesense `filter_by` string, +/// Postgres `WHERE` clause, …) so the call site doesn't have to know which +/// backend is in use. #[derive(Debug, Clone)] pub struct SearchQuery { - /// The search query string (`"*"` matches all documents). + /// The full-text query string. Empty string is treated as "match all". 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, + /// Nostr kinds to restrict to. Empty = no restriction. + pub kinds: Vec, + /// Event author pubkeys (hex). Empty = no restriction. + pub authors: Vec, + /// Channel UUID strings to restrict to. Empty = no restriction. The + /// [`GLOBAL_CHANNEL_SENTINEL`] value (`"__global__"`) selects events that + /// have no `channel_id` set. + pub channel_ids: Vec, + /// Lower bound on `created_at` (Unix seconds, inclusive). + pub since: Option, + /// Upper bound on `created_at` (Unix seconds, inclusive). + pub until: Option, /// Page number (1-indexed). pub page: u32, /// Number of results per page. @@ -24,8 +43,11 @@ impl Default for SearchQuery { fn default() -> Self { Self { q: "*".into(), - filter_by: None, - sort_by: Some("created_at:desc".into()), + kinds: Vec::new(), + authors: Vec::new(), + channel_ids: Vec::new(), + since: None, + until: None, page: 1, per_page: 20, } @@ -33,7 +55,60 @@ impl Default for SearchQuery { } impl SearchQuery { - /// Converts the query into Typesense HTTP query parameters. + /// Renders the structured filters into a Typesense `filter_by` string. + /// Returns `None` when no constraints apply. + pub(crate) fn typesense_filter_by(&self) -> Option { + let mut parts: Vec = Vec::new(); + + if !self.channel_ids.is_empty() { + // Split global sentinel from real UUIDs so we emit + // (channel_id:=[uuid1,uuid2] || channel_id:=__global__) + // instead of an invalid `channel_id:=[__global__,uuid]` mix. + let (globals, uuids): (Vec<&String>, Vec<&String>) = self + .channel_ids + .iter() + .partition(|id| id.as_str() == GLOBAL_CHANNEL_SENTINEL); + + let include_global = !globals.is_empty(); + if !uuids.is_empty() { + let joined: Vec = uuids.iter().map(|s| (*s).clone()).collect(); + if include_global { + parts.push(format!( + "(channel_id:=[{}] || channel_id:=__global__)", + joined.join(",") + )); + } else { + parts.push(format!("channel_id:=[{}]", joined.join(","))); + } + } else if include_global { + parts.push("channel_id:=__global__".to_string()); + } + } + + if !self.kinds.is_empty() { + let vs: Vec = self.kinds.iter().map(|k| k.to_string()).collect(); + parts.push(format!("kind:=[{}]", vs.join(","))); + } + if !self.authors.is_empty() { + parts.push(format!("pubkey:=[{}]", self.authors.join(","))); + } + if let Some(since) = self.since { + parts.push(format!("created_at:>={}", since)); + } + if let Some(until) = self.until { + parts.push(format!("created_at:<={}", until)); + } + + if parts.is_empty() { + None + } else { + Some(parts.join(" && ")) + } + } + + /// Renders the Typesense HTTP query parameters. Kept for compatibility + /// with `to_query_params`-based callers; the actual production path uses + /// `multi_search` via [`search`]. pub fn to_query_params(&self) -> Vec<(String, String)> { let mut params = vec![ ("q".into(), self.q.clone()), @@ -42,12 +117,8 @@ impl SearchQuery { ("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())); + if let Some(filter) = self.typesense_filter_by() { + params.push(("filter_by".into(), filter)); } params @@ -167,11 +238,8 @@ pub async fn search( "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()); + if let Some(filter) = query.typesense_filter_by() { + search_params["filter_by"] = serde_json::Value::String(filter); } let body = serde_json::json!({ "searches": [search_params] }); @@ -241,8 +309,11 @@ mod tests { 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()), + kinds: vec![1], + authors: Vec::new(), + channel_ids: Vec::new(), + since: None, + until: None, page: 2, per_page: 10, }; @@ -259,16 +330,20 @@ mod tests { 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"); + assert_eq!(get("filter_by").unwrap(), "kind:=[1]"); + // sort_by is no longer emitted — Typesense default = relevance. + assert!(params.iter().all(|(k, _)| k != "sort_by")); } #[test] fn test_search_query_no_optional_fields() { let q = SearchQuery { q: "*".into(), - filter_by: None, - sort_by: None, + kinds: Vec::new(), + authors: Vec::new(), + channel_ids: Vec::new(), + since: None, + until: None, page: 1, per_page: 20, }; @@ -284,6 +359,50 @@ mod tests { assert!(!has_key("sort_by")); } + #[test] + fn test_typesense_filter_by_renders_structured_fields() { + let q = SearchQuery { + q: "hello".into(), + kinds: vec![1, 42], + authors: vec!["deadbeef".into()], + channel_ids: vec!["11111111-1111-1111-1111-111111111111".into()], + since: Some(1_700_000_000), + until: Some(1_700_000_100), + ..Default::default() + }; + let filter = q.typesense_filter_by().expect("non-empty filter"); + assert!(filter.contains("channel_id:=[11111111-1111-1111-1111-111111111111]")); + assert!(filter.contains("kind:=[1,42]")); + assert!(filter.contains("pubkey:=[deadbeef]")); + assert!(filter.contains("created_at:>=1700000000")); + assert!(filter.contains("created_at:<=1700000100")); + } + + #[test] + fn test_typesense_filter_by_handles_global_sentinel() { + let with_global_only = SearchQuery { + q: "*".into(), + channel_ids: vec![GLOBAL_CHANNEL_SENTINEL.to_string()], + ..Default::default() + }; + assert_eq!( + with_global_only.typesense_filter_by().as_deref(), + Some("channel_id:=__global__") + ); + + let with_mix = SearchQuery { + q: "*".into(), + channel_ids: vec![ + "11111111-1111-1111-1111-111111111111".into(), + GLOBAL_CHANNEL_SENTINEL.to_string(), + ], + ..Default::default() + }; + let filter = with_mix.typesense_filter_by().expect("non-empty"); + assert!(filter.contains("|| channel_id:=__global__")); + assert!(filter.contains("channel_id:=[11111111-1111-1111-1111-111111111111]")); + } + #[test] fn test_search_result_parsing() { let raw = json!({ diff --git a/migrations/0004_search_fts.sql b/migrations/0004_search_fts.sql new file mode 100644 index 0000000000..277e30e8f4 --- /dev/null +++ b/migrations/0004_search_fts.sql @@ -0,0 +1,35 @@ +-- Add a generated full-text-search column + GIN index to the events table so +-- the relay can serve NIP-50 search directly from Postgres, eliminating the +-- Typesense dependency. +-- +-- `content_tsv` is `GENERATED ALWAYS AS (to_tsvector('simple', content)) STORED` +-- so every INSERT/UPDATE populates it automatically — no application-level +-- index maintenance needed (matches the pattern Typesense filled today via the +-- worker pipeline in `buzz-relay/src/state.rs`). +-- +-- Tokenizer choice: `simple` does no stemming and preserves identifiers like +-- agent handles, nip05 strings, and slugs. The `english` config would stem +-- ("running" → "run") but mangle handles ("alice42" tokenizes fine, but +-- something like "agents" → "agent" would break exact-handle search). Chat +-- content is heterogeneous; `simple` is the safer default for v1. +-- +-- kind:0 metadata flattening: the existing Typesense pipeline appends parsed +-- display_name/name/nip05 to event content before indexing +-- (`buzz-search/src/index.rs::flatten_kind0_for_indexing`). With FTS on raw +-- `content`, those strings still tokenize because they live in the kind:0 JSON +-- body — `to_tsvector('simple', '{"name":"alice"}')` matches `q=alice` after +-- json-aware tokenization. Validated by the NIP-50 e2e suite. +-- +-- `events` is partitioned by RANGE (created_at); ADD COLUMN on the parent +-- cascades the generated column to every partition, and CREATE INDEX on the +-- parent builds a partitioned GIN index that propagates to each partition. +-- Partition pruning on since/until queries narrows the GIN scan further than +-- Typesense's full-collection scan does today. +-- +-- Managed by sqlx migrations. + +ALTER TABLE events + ADD COLUMN content_tsv tsvector + GENERATED ALWAYS AS (to_tsvector('simple', content)) STORED; + +CREATE INDEX idx_events_content_tsv ON events USING GIN (content_tsv); diff --git a/schema/schema.sql b/schema/schema.sql index 2a839adf6b..6fea78b667 100644 --- a/schema/schema.sql +++ b/schema/schema.sql @@ -107,6 +107,8 @@ CREATE TABLE events ( d_tag TEXT, not_before BIGINT, delivered_at BIGINT, + content_tsv tsvector + GENERATED ALWAYS AS (to_tsvector('simple', content)) STORED, PRIMARY KEY (created_at, id) ) PARTITION BY RANGE (created_at); @@ -136,6 +138,7 @@ CREATE INDEX idx_events_addressable ON events (kind, pubkey, channel_id, deleted CREATE INDEX idx_events_parameterized ON events (kind, pubkey, d_tag, deleted_at) WHERE d_tag IS NOT NULL; CREATE INDEX idx_events_not_before ON events (not_before) WHERE not_before IS NOT NULL AND deleted_at IS NULL AND delivered_at IS NULL; +CREATE INDEX idx_events_content_tsv ON events USING GIN (content_tsv); -- ── Event mentions ────────────────────────────────────────────────────────────