jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-27 20:41:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

rewrite(search): Postgres FTS, community-scoped, drop Typesense

Browse Source 
The Lane-0 freeze landed `events.search_tsv TSVECTOR GENERATED ALWAYS AS
(to_tsvector('simple', content)) STORED` + `GIN (search_tsv)` directly in
the schema. With that in place the entire Typesense apparatus is dead
weight: there is nothing to index out-of-band, no consistency window to
reason about, no client-forgeable index/content drift. Indexing is the
SQL write.

This rewrites `crates/buzz-search/` from scratch around that:

  - `query.rs`: one SQL builder. `community_id = $ctx` is the first
    predicate of every executed statement and is unconditional —
    `SearchQuery` requires a `CommunityId` at the type level (no
    construction path omits it). `search_tsv @@ websearch_to_tsquery(...)`
    is the FTS predicate; `ts_rank_cd DESC, created_at DESC, id` is the
    order. Channel scope replaces today's `__global__` sentinel with
    `channel_id IS NULL`. Empty query short-circuits without a roundtrip.

  - `lib.rs`: thin `SearchService { pool }`. Takes `&PgPool` directly so
    the crate stays a leaf — no buzz-db dependency. Re-exports
    `CommunityId` for callers that need to mint the fence.

  - `error.rs`: collapsed to one variant (`Db(sqlx::Error)`); empty
    queries are not errors.

  - Deleted `collection.rs` and `index.rs` (Typesense HTTP client and
    indexer). Dropped `reqwest`/`serde`/`serde_json`/`chrono`/`nostr`
    from `Cargo.toml`.

  - Added `tests/fts_integration.rs` — 8 integration tests against real
    Postgres, each on its own throwaway schema applying the frozen
    `migrations/0001_initial_schema.sql` via `include_str!`. The
    load-bearing one is `search_does_not_return_other_community_events`:
    mutating the `community_id = $ctx` predicate to `1=1` makes that
    test go red (verified, then reverted) — the fence bites where it
    has to.

Conformance row 50 — search re-auth and one-shot NIP-50 — is unchanged
in shape: the relay refetches canonical events per hit through buzz-db's
scoped fetcher and runs the access predicate. Search is never the
access boundary; this crate just returns candidate ids. The row's
Typesense prose rewrite is owned by Eva's integration lane (one writer
per path).

EXPLAIN ANALYZE evidence on a 200k-row community confirms the planner
picks `Bitmap Index Scan on events_p<...>_search_tsv_idx` for the
populated partition (full plan in RESEARCH/SEARCH_LANE_FTS_EXPLAIN.md
in the workspace). Single-column `GIN (search_tsv)` is sufficient at
this scale — no `btree_gin` needed (Max's caveat holds).

Cross-lane removals owed to Eva (relay-wiring lane, not this commit):
  - relay state.rs: remove `search_index_tx` mpsc + worker
  - relay main.rs: remove `search.ensure_collection()` call
  - relay handlers/event.rs: remove `search_index_tx.send()`
  - relay api/bridge.rs::handle_bridge_search: rewrite to new API
  - relay handlers/req.rs::handle_search_req: rewrite to new API
  - relay handlers/req.rs::build_search_channel_scope_filter: delete
  - relay bin/reindex_kind0.rs: delete
  - docker-compose.yml: drop typesense service + volume
  - docs/multi-tenant-conformance.md row 50: rewrite Typesense prose

Tests: `cargo test -p buzz-search --test fts_integration --
--include-ignored --test-threads=1` — 8 passed, 0 failed.
Clippy: `cargo clippy -p buzz-search --all-targets -- -D warnings` — clean.

Co-authored-by: Tyler Longwell <tlongwell@block.xyz>
Signed-off-by: Tyler Longwell <tlongwell@block.xyz>
This commit is contained in:
npub1jmc9dt2lyvzu3h0kxlwxt5zg4fxp9476awyxw6gwxn72g6cw7exqs64whm
2026-06-26 16:54:02 -04:00
parent e349d76498
commit e31c098ce5
8 changed files with 829 additions and 1375 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
Cargo.lock generated
View File

@@ -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",
]

15
crates/buzz-search/Cargo.toml
View File

@@ -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 }

129
crates/buzz-search/src/collection.rs
View File

@@ -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[]");
}
}

38
crates/buzz-search/src/error.rs
View File

@@ -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),
}

529
crates/buzz-search/src/index.rs
View File

@@ -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<Value, SearchError> {
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<String> = 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::<Value>(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<usize, SearchError> {
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::<Value>(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<Uuid>) -> 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");
}
}

319
crates/buzz-search/src/lib.rs
View File

@@ -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<usize, SearchError> {
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<SearchResult, SearchError> {
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<StoredEvent> = (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
}
}

569
crates/buzz-search/src/query.rs
View File

@@ -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<String>,
/// Optional sort expression (e.g. `"created_at:desc"`).
pub sort_by: Option<String>,
/// 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<Vec<Uuid>>,
/// 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<Vec<i32>>,
/// NIP-01 authors filter (32-byte pubkeys). None = no author constraint.
pub authors: Option<Vec<Vec<u8>>>,
/// NIP-01 since (Unix seconds). Inclusive lower bound on created_at.
pub since: Option<i64>,
/// NIP-01 until (Unix seconds). Inclusive upper bound on created_at.
pub until: Option<i64>,
/// 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<String>,
/// 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<Uuid>,
/// 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<SearchHit>,
/// 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<TypesenseSearchResult>,
}
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<SearchResult, SearchError> {
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<TypesenseHit>,
}
#[derive(Debug, Deserialize)]
struct TypesenseHit {
document: TypesenseDocument,
#[serde(rename = "text_match")]
text_match: Option<u64>,
}
#[derive(Debug, Deserialize)]
struct TypesenseDocument {
id: String,
content: String,
kind: i32,
pubkey: String,
channel_id: Option<String>,
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<SearchResult, SearchError> {
debug!(
q = %query.q,
page = query.page,
per_page = query.per_page,
collection = collection_name,
"Executing search"
let mut qb: QueryBuilder<sqlx::Postgres> = 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": [<search_response>]}. 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<SearchResult, SearchError> {
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<String> {
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<u8> = row.try_get("id")?;
let pk_bytes: Vec<u8> = row.try_get("pubkey")?;
let id: [u8; 32] = id_bytes.try_into().map_err(|v: Vec<u8>| {
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<u8>| {
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 })
}

598
crates/buzz-search/tests/fts_integration.rs Normal file
View File

@@ -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<Uuid>,
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;
}