From 2d433a4d9911503b2ec6a2bec1b9105ba7d366dc Mon Sep 17 00:00:00 2001 From: npub13n66s06epmqf2kc3v373ez8hj65cuzyvxzjf93vwpervxqn2u7jq2qd9je <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> Date: Sat, 1 Aug 2026 12:45:34 -0400 Subject: [PATCH] feat(relay): add SQLite DM and FTS5 paths Signed-off-by: npub13n66s06epmqf2kc3v373ez8hj65cuzyvxzjf93vwpervxqn2u7jq2qd9je <8cf5a83f590ec0955b11647d1c88f796a98e088c30a492c58e0e46c3026ae7a4@buzz.block.builderlab.xyz> --- crates/buzz-db/src/lib.rs | 65 ++++++++-- crates/buzz-db/src/sqlite.rs | 217 +++++++++++++++++++++++++++++++- crates/buzz-relay/src/main.rs | 3 +- crates/buzz-search/src/lib.rs | 11 +- crates/buzz-search/src/query.rs | 174 ++++++++++++++++++++++++- 5 files changed, 458 insertions(+), 12 deletions(-) diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 9b9a2d32a..183c168ed 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -669,6 +669,12 @@ impl Db { } } + /// Return a clone of the SQLite pool for process-local services that share + /// the same database, such as FTS5 search. + pub fn sqlite_pool_clone(&self) -> Result { + Ok(self.sqlite_pool()?.clone()) + } + /// Creates a local single-node database backed by SQLite. /// /// The embedded local schema is applied before this constructor returns. @@ -2815,7 +2821,14 @@ impl Db { pubkey: &[u8], policy: &str, ) -> Result<()> { - user::set_channel_add_policy(self.pg_pool()?, community_id, pubkey, policy).await + match &self.backend { + DbBackend::SQLite(pool) => { + sqlite::set_channel_add_policy(pool, community_id, pubkey, policy).await + } + DbBackend::Postgres => { + user::set_channel_add_policy(self.pg_pool()?, community_id, pubkey, policy).await + } + } } /// Find an existing DM by its participant hash. @@ -2824,7 +2837,14 @@ impl Db { community_id: CommunityId, participant_hash: &[u8], ) -> Result> { - dm::find_dm_by_participants(self.pg_pool()?, community_id, participant_hash).await + match &self.backend { + DbBackend::SQLite(pool) => { + sqlite::find_dm_by_participants(pool, community_id, participant_hash).await + } + DbBackend::Postgres => { + dm::find_dm_by_participants(self.pg_pool()?, community_id, participant_hash).await + } + } } /// Create or return an existing DM channel. @@ -2834,7 +2854,14 @@ impl Db { participants: &[&[u8]], created_by: &[u8], ) -> Result { - dm::create_dm(self.pg_pool()?, community_id, participants, created_by).await + match &self.backend { + DbBackend::SQLite(pool) => { + sqlite::create_dm(pool, community_id, participants, created_by).await + } + DbBackend::Postgres => { + dm::create_dm(self.pg_pool()?, community_id, participants, created_by).await + } + } } /// List all DMs for a user. @@ -2845,7 +2872,14 @@ impl Db { limit: u32, cursor: Option, ) -> Result> { - dm::list_dms_for_user(self.pg_pool()?, community_id, pubkey, limit, cursor).await + match &self.backend { + DbBackend::SQLite(pool) => { + sqlite::list_dms_for_user(pool, community_id, pubkey, limit, cursor).await + } + DbBackend::Postgres => { + dm::list_dms_for_user(self.pg_pool()?, community_id, pubkey, limit, cursor).await + } + } } /// Open or retrieve a DM for the given participants. @@ -2894,7 +2928,14 @@ impl Db { channel_id: Uuid, pubkey: &[u8], ) -> Result<()> { - dm::hide_dm(self.pg_pool()?, community_id, channel_id, pubkey).await + match &self.backend { + DbBackend::SQLite(pool) => { + sqlite::hide_dm(pool, community_id, channel_id, pubkey).await + } + DbBackend::Postgres => { + dm::hide_dm(self.pg_pool()?, community_id, channel_id, pubkey).await + } + } } /// Unhide a DM channel for a specific user. @@ -2904,7 +2945,14 @@ impl Db { channel_id: Uuid, pubkey: &[u8], ) -> Result<()> { - dm::unhide_dm(self.pg_pool()?, community_id, channel_id, pubkey).await + match &self.backend { + DbBackend::SQLite(pool) => { + sqlite::unhide_dm(pool, community_id, channel_id, pubkey).await + } + DbBackend::Postgres => { + dm::unhide_dm(self.pg_pool()?, community_id, channel_id, pubkey).await + } + } } /// List the channel IDs of all DMs the given user currently has hidden. @@ -2913,7 +2961,10 @@ impl Db { community_id: CommunityId, pubkey: &[u8], ) -> Result> { - dm::list_hidden_dms(self.pg_pool()?, community_id, pubkey).await + match &self.backend { + DbBackend::SQLite(pool) => sqlite::list_hidden_dms(pool, community_id, pubkey).await, + DbBackend::Postgres => dm::list_hidden_dms(self.pg_pool()?, community_id, pubkey).await, + } } /// Insert thread metadata. diff --git a/crates/buzz-db/src/sqlite.rs b/crates/buzz-db/src/sqlite.rs index 922657ad1..cc0c62554 100644 --- a/crates/buzz-db/src/sqlite.rs +++ b/crates/buzz-db/src/sqlite.rs @@ -201,7 +201,24 @@ pub(crate) async fn migrate(pool: &SqlitePool) -> Result<()> { tx.commit().await?; version = 2; } - if version != 2 { + if version < 3 { + let mut tx = pool.begin().await?; + for statement in [ + "CREATE VIRTUAL TABLE IF NOT EXISTS events_fts USING fts5(content, content='events', content_rowid='rowid', tokenize='unicode61')", + "CREATE TRIGGER IF NOT EXISTS events_fts_insert AFTER INSERT ON events WHEN new.kind NOT IN (1059, 30300, 30350, 30622, 44100, 44101, 44200) BEGIN INSERT INTO events_fts(rowid, content) VALUES (new.rowid, new.content); END", + "CREATE TRIGGER IF NOT EXISTS events_fts_delete AFTER DELETE ON events WHEN old.kind NOT IN (1059, 30300, 30350, 30622, 44100, 44101, 44200) BEGIN INSERT INTO events_fts(events_fts, rowid, content) VALUES ('delete', old.rowid, old.content); END", + "CREATE TRIGGER IF NOT EXISTS events_fts_update AFTER UPDATE OF content, kind ON events BEGIN INSERT INTO events_fts(events_fts, rowid, content) SELECT 'delete', old.rowid, old.content WHERE old.kind NOT IN (1059, 30300, 30350, 30622, 44100, 44101, 44200); INSERT INTO events_fts(rowid, content) SELECT new.rowid, new.content WHERE new.kind NOT IN (1059, 30300, 30350, 30622, 44100, 44101, 44200); END", + "INSERT INTO events_fts(rowid, content) SELECT rowid, content FROM events WHERE kind NOT IN (1059, 30300, 30350, 30622, 44100, 44101, 44200)", + ] { + sqlx::query(statement).execute(&mut *tx).await?; + } + sqlx::query("UPDATE schema_version SET version = 3 WHERE singleton = 1") + .execute(&mut *tx) + .await?; + tx.commit().await?; + version = 3; + } + if version != 3 { return Err(crate::DbError::InvalidData(format!( "unsupported SQLite schema version {version}" ))); @@ -667,6 +684,202 @@ pub(crate) async fn remove_reaction_by_source_event_id( .execute(pool).await?.rows_affected() != 0) } +pub(crate) async fn set_channel_add_policy( + pool: &SqlitePool, + community: CommunityId, + pubkey: &[u8], + policy: &str, +) -> Result<()> { + if !matches!(policy, "anyone" | "owner_only" | "nobody") { + return Err(crate::DbError::InvalidData(format!( + "invalid channel_add_policy: {policy}" + ))); + } + let result = sqlx::query( + "UPDATE users SET channel_add_policy = ?1 WHERE community_id = ?2 AND pubkey = ?3", + ) + .bind(policy) + .bind(community.as_uuid().to_string()) + .bind(pubkey) + .execute(pool) + .await?; + if result.rows_affected() == 0 { + return Err(crate::DbError::NotFound( + "pubkey not found in users table".into(), + )); + } + Ok(()) +} + +pub(crate) async fn find_dm_by_participants( + pool: &SqlitePool, + community: CommunityId, + participant_hash: &[u8], +) -> Result> { + let row = sqlx::query("SELECT id, name, channel_type, visibility, description, canvas, created_by, created_at, updated_at, archived_at, deleted_at, nip29_group_id, topic_required, max_members, topic, topic_set_by, topic_set_at, purpose, purpose_set_by, purpose_set_at, ttl_seconds, ttl_deadline FROM channels WHERE community_id = ?1 AND participant_hash = ?2 AND channel_type = 'dm' AND deleted_at IS NULL LIMIT 1") + .bind(community.as_uuid().to_string()).bind(participant_hash).fetch_optional(pool).await?; + row.map(channel_record).transpose() +} + +pub(crate) async fn create_dm( + pool: &SqlitePool, + community: CommunityId, + participants: &[&[u8]], + created_by: &[u8], +) -> Result { + if !(2..=9).contains(&participants.len()) { + return Err(crate::DbError::InvalidData( + "DM requires 2-9 participants".into(), + )); + } + if participants.iter().any(|p| p.len() != 32) { + return Err(crate::DbError::InvalidData( + "DM participant pubkeys must be 32 bytes".into(), + )); + } + if !participants.contains(&created_by) { + return Err(crate::DbError::InvalidData( + "DM creator must be a participant".into(), + )); + } + let mut unique = participants.to_vec(); + unique.sort_unstable(); + unique.dedup(); + if unique.len() != participants.len() { + return Err(crate::DbError::InvalidData( + "DM participants must be unique".into(), + )); + } + let hash = crate::dm::compute_participant_hash(participants); + let mut tx = pool.begin().await?; + let query = "SELECT id, name, channel_type, visibility, description, canvas, created_by, created_at, updated_at, archived_at, deleted_at, nip29_group_id, topic_required, max_members, topic, topic_set_by, topic_set_at, purpose, purpose_set_by, purpose_set_at, ttl_seconds, ttl_deadline FROM channels WHERE community_id = ?1 AND participant_hash = ?2 AND channel_type = 'dm' AND deleted_at IS NULL LIMIT 1"; + if let Some(row) = sqlx::query(query) + .bind(community.as_uuid().to_string()) + .bind(hash.as_slice()) + .fetch_optional(&mut *tx) + .await? + { + tx.commit().await?; + return channel_record(row); + } + let id = Uuid::new_v4(); + let name = if participants.len() == 2 { + "DM".to_owned() + } else { + format!("Group DM ({})", participants.len()) + }; + sqlx::query("INSERT INTO channels (id, community_id, name, channel_type, visibility, participant_hash, created_by) VALUES (?1, ?2, ?3, 'dm', 'private', ?4, ?5)") + .bind(id.to_string()).bind(community.as_uuid().to_string()).bind(name).bind(hash.as_slice()).bind(created_by).execute(&mut *tx).await?; + for participant in participants { + sqlx::query("INSERT INTO channel_members (channel_id, pubkey, role, invited_by) VALUES (?1, ?2, 'member', ?3)") + .bind(id.to_string()).bind(*participant).bind(created_by).execute(&mut *tx).await?; + } + let row = sqlx::query("SELECT id, name, channel_type, visibility, description, canvas, created_by, created_at, updated_at, archived_at, deleted_at, nip29_group_id, topic_required, max_members, topic, topic_set_by, topic_set_at, purpose, purpose_set_by, purpose_set_at, ttl_seconds, ttl_deadline FROM channels WHERE community_id = ?1 AND participant_hash = ?2 AND channel_type = 'dm' AND deleted_at IS NULL LIMIT 1") + .bind(community.as_uuid().to_string()) + .bind(hash.as_slice()) + .fetch_one(&mut *tx) + .await?; + let record = channel_record(row)?; + tx.commit().await?; + Ok(record) +} + +pub(crate) async fn list_dms_for_user( + pool: &SqlitePool, + community: CommunityId, + pubkey: &[u8], + limit: u32, + cursor: Option, +) -> Result> { + let limit = limit.min(200) as i64; + let cursor_ts = if let Some(id) = cursor { + sqlx::query_scalar::<_, i64>( + "SELECT updated_at FROM channels WHERE id = ?1 AND community_id = ?2", + ) + .bind(id.to_string()) + .bind(community.as_uuid().to_string()) + .fetch_optional(pool) + .await? + } else { + None + }; + let rows = if let Some(ts) = cursor_ts { + sqlx::query("SELECT c.id, c.created_at, c.updated_at FROM channels c JOIN channel_members cm ON c.id = cm.channel_id AND cm.pubkey = ?2 AND cm.removed_at IS NULL AND cm.hidden_at IS NULL WHERE c.community_id = ?1 AND c.channel_type = 'dm' AND c.deleted_at IS NULL AND c.updated_at < ?3 ORDER BY c.updated_at DESC LIMIT ?4") + .bind(community.as_uuid().to_string()).bind(pubkey).bind(ts).bind(limit).fetch_all(pool).await? + } else { + sqlx::query("SELECT c.id, c.created_at, c.updated_at FROM channels c JOIN channel_members cm ON c.id = cm.channel_id AND cm.pubkey = ?2 AND cm.removed_at IS NULL AND cm.hidden_at IS NULL WHERE c.community_id = ?1 AND c.channel_type = 'dm' AND c.deleted_at IS NULL ORDER BY c.updated_at DESC LIMIT ?3") + .bind(community.as_uuid().to_string()).bind(pubkey).bind(limit).fetch_all(pool).await? + }; + let mut out = Vec::with_capacity(rows.len()); + for row in rows { + let id: String = row.try_get("id")?; + let channel_id = Uuid::parse_str(&id) + .map_err(|e| crate::DbError::InvalidData(format!("invalid SQLite channel id: {e}")))?; + let created_at = timestamp(row.try_get("created_at")?)?; + let updated_at = timestamp(row.try_get("updated_at")?)?; + let members = sqlx::query("SELECT cm.pubkey, cm.role, u.display_name FROM channel_members cm LEFT JOIN users u ON u.community_id = ?1 AND u.pubkey = cm.pubkey JOIN channels c ON c.id = cm.channel_id AND c.community_id = ?1 WHERE cm.channel_id = ?2 AND cm.removed_at IS NULL ORDER BY cm.joined_at ASC") + .bind(community.as_uuid().to_string()).bind(id).fetch_all(pool).await?; + let participants = members + .into_iter() + .map(|r| { + Ok(crate::dm::DmParticipant { + pubkey: r.try_get("pubkey")?, + display_name: r.try_get("display_name")?, + role: r.try_get("role")?, + }) + }) + .collect::>>()?; + out.push(crate::dm::DmRecord { + channel_id, + participants, + last_message_at: Some(updated_at), + created_at, + }); + } + Ok(out) +} + +pub(crate) async fn hide_dm( + pool: &SqlitePool, + community: CommunityId, + channel_id: Uuid, + pubkey: &[u8], +) -> Result<()> { + let result = sqlx::query("UPDATE channel_members SET hidden_at = unixepoch() WHERE channel_id = ?1 AND pubkey = ?2 AND removed_at IS NULL AND EXISTS (SELECT 1 FROM channels c WHERE c.id = channel_members.channel_id AND c.community_id = ?3)").bind(channel_id.to_string()).bind(pubkey).bind(community.as_uuid().to_string()).execute(pool).await?; + if result.rows_affected() == 0 { + return Err(crate::DbError::NotFound(format!( + "no active membership for channel {channel_id}" + ))); + } + Ok(()) +} + +pub(crate) async fn unhide_dm( + pool: &SqlitePool, + community: CommunityId, + channel_id: Uuid, + pubkey: &[u8], +) -> Result<()> { + sqlx::query("UPDATE channel_members SET hidden_at = NULL WHERE channel_id = ?1 AND pubkey = ?2 AND removed_at IS NULL AND EXISTS (SELECT 1 FROM channels c WHERE c.id = channel_members.channel_id AND c.community_id = ?3)").bind(channel_id.to_string()).bind(pubkey).bind(community.as_uuid().to_string()).execute(pool).await?; + Ok(()) +} + +pub(crate) async fn list_hidden_dms( + pool: &SqlitePool, + community: CommunityId, + pubkey: &[u8], +) -> Result> { + let rows = sqlx::query("SELECT cm.channel_id FROM channel_members cm JOIN channels c ON c.id = cm.channel_id AND c.community_id = ?1 WHERE cm.pubkey = ?2 AND cm.removed_at IS NULL AND cm.hidden_at IS NOT NULL AND c.channel_type = 'dm' AND c.deleted_at IS NULL ORDER BY cm.channel_id") + .bind(community.as_uuid().to_string()).bind(pubkey).fetch_all(pool).await?; + rows.into_iter() + .map(|r| { + let id: String = r.try_get("channel_id")?; + Uuid::parse_str(&id) + .map_err(|e| crate::DbError::InvalidData(format!("invalid SQLite channel id: {e}"))) + }) + .collect() +} + pub(crate) async fn open_dm( pool: &SqlitePool, community: CommunityId, @@ -1213,7 +1426,7 @@ mod tests { .fetch_one(&upgraded) .await .unwrap(), - 2 + 3 ); for (table, column) in [ ("channels", "participant_hash"), diff --git a/crates/buzz-relay/src/main.rs b/crates/buzz-relay/src/main.rs index fb5aacc15..7e1f064e2 100644 --- a/crates/buzz-relay/src/main.rs +++ b/crates/buzz-relay/src/main.rs @@ -1258,6 +1258,7 @@ async fn run_single_node(config: Config, tracer_init: telemetry::TracerInit) -> "local", buzz_media::S3AddressingStyle::Path, )?; + let search = SearchService::sqlite(db.sqlite_pool_clone()?); let replay: Arc = Arc::new(InProcessNip98ReplayGuard::new()); let (app_state, audit_shutdown) = AppState::new_with_backends( config, @@ -1265,7 +1266,7 @@ async fn run_single_node(config: Config, tracer_init: telemetry::TracerInit) -> db, redis_pool: None, pubsub, - search: SearchService::unsupported(), + search, media_storage: buzz_media::MediaStorage::filesystem(&media_root), git_store, nip98_replay: replay, diff --git a/crates/buzz-search/src/lib.rs b/crates/buzz-search/src/lib.rs index d24fde143..4010dd3bb 100644 --- a/crates/buzz-search/src/lib.rs +++ b/crates/buzz-search/src/lib.rs @@ -30,12 +30,13 @@ pub use buzz_core::CommunityId; pub use error::SearchError; pub use query::{search, ChannelScope, SearchHit, SearchMode, SearchQuery, SearchResult}; -use sqlx::PgPool; +use sqlx::{PgPool, SqlitePool}; /// Search backend selected at startup. #[derive(Debug, Clone)] enum SearchBackend { Postgres(PgPool), + SQLite(SqlitePool), Unsupported, } @@ -53,6 +54,13 @@ impl SearchService { } } + /// Build a search service over an existing SQLite pool with FTS5. + pub fn sqlite(pool: SqlitePool) -> Self { + Self { + backend: SearchBackend::SQLite(pool), + } + } + /// Build a backend that explicitly reports search as unsupported. pub fn unsupported() -> Self { Self { @@ -64,6 +72,7 @@ impl SearchService { pub async fn search(&self, query: &SearchQuery) -> Result { match &self.backend { SearchBackend::Postgres(pool) => query::search(pool, query).await, + SearchBackend::SQLite(pool) => query::search_sqlite(pool, query).await, SearchBackend::Unsupported => Err(SearchError::Unsupported), } } diff --git a/crates/buzz-search/src/query.rs b/crates/buzz-search/src/query.rs index d191353af..a1699a445 100644 --- a/crates/buzz-search/src/query.rs +++ b/crates/buzz-search/src/query.rs @@ -9,7 +9,7 @@ //! See conformance row 50. use buzz_core::CommunityId; -use sqlx::{PgPool, QueryBuilder, Row}; +use sqlx::{PgPool, QueryBuilder, Row, SqlitePool}; use uuid::Uuid; use crate::error::SearchError; @@ -336,6 +336,132 @@ pub async fn search(pool: &PgPool, query: &SearchQuery) -> Result Result { + let Some(search_text) = normalized_search_text(&query.q) else { + return Ok(SearchResult { + hits: Vec::new(), + page: query.page.clamp(1, PAGE_MAX), + }); + }; + let per_page = if query.per_page == 0 { + PER_PAGE_DEFAULT + } else { + query.per_page.clamp(1, PER_PAGE_MAX) + }; + let page = query.page.clamp(1, PAGE_MAX); + let offset = ((page - 1) as i64) * per_page as i64; + let terms: Vec = search_text + .split_whitespace() + .filter(|term| !term.is_empty()) + .map(|term| format!("\"{}\"", term.replace('"', "\"\""))) + .collect(); + if terms.is_empty() { + return Ok(SearchResult { hits: vec![], page }); + } + let expression = match query.mode { + SearchMode::FullText => terms.join(" AND "), + SearchMode::Prefix => terms + .iter() + .enumerate() + .map(|(index, term)| { + if index + 1 == terms.len() { + format!("{term}*") + } else { + term.clone() + } + }) + .collect::>() + .join(" AND "), + }; + let mut qb: QueryBuilder = QueryBuilder::new( + "SELECT e.id, e.kind, e.pubkey, e.channel_id, e.created_at AS created_at_s, bm25(events_fts) AS score FROM events_fts JOIN events e ON e.rowid = events_fts.rowid WHERE events_fts MATCH ", + ); + qb.push_bind(expression); + qb.push(" AND e.community_id = ") + .push_bind(query.community.as_uuid().to_string()); + match &query.channel_scope { + ChannelScope::Any => {} + ChannelScope::ChannelLessOnly => { + qb.push(" AND e.channel_id IS NULL"); + } + ChannelScope::Channels(ids) => { + qb.push(" AND e.channel_id IN ("); + let mut separated = qb.separated(", "); + for id in ids { + separated.push_bind(id.to_string()); + } + separated.push_unseparated(")"); + } + ChannelScope::ChannelsOrChannelLess(ids) => { + qb.push(" AND (e.channel_id IN ("); + let mut separated = qb.separated(", "); + for id in ids { + separated.push_bind(id.to_string()); + } + separated.push_unseparated(") OR e.channel_id IS NULL)"); + } + } + if let Some(kinds) = &query.kinds { + if !kinds.is_empty() { + qb.push(" AND e.kind IN ("); + let mut separated = qb.separated(", "); + for kind in kinds { + separated.push_bind(*kind); + } + separated.push_unseparated(")"); + } + } + if let Some(authors) = &query.authors { + if !authors.is_empty() { + qb.push(" AND e.pubkey IN ("); + let mut separated = qb.separated(", "); + for author in authors { + separated.push_bind(author); + } + separated.push_unseparated(")"); + } + } + if let Some(since) = query.since { + qb.push(" AND e.created_at >= ").push_bind(since); + } + if let Some(until) = query.until { + qb.push(" AND e.created_at <= ").push_bind(until); + } + qb.push(" ORDER BY score ASC, e.created_at DESC, e.id LIMIT ") + .push_bind(per_page as i64) + .push(" OFFSET ") + .push_bind(offset); + let rows = qb.build().fetch_all(pool).await?; + let mut hits = Vec::with_capacity(rows.len()); + for row in rows { + let id_bytes: Vec = row.try_get("id")?; + let pk_bytes: Vec = row.try_get("pubkey")?; + let id = id_bytes.try_into().map_err(|v: Vec| { + sqlx::Error::Decode(format!("event id column is {} bytes, expected 32", v.len()).into()) + })?; + let pubkey = pk_bytes.try_into().map_err(|v: Vec| { + sqlx::Error::Decode(format!("pubkey column is {} bytes, expected 32", v.len()).into()) + })?; + let channel: Option = row.try_get("channel_id")?; + hits.push(SearchHit { + event_id: id, + kind: row.try_get("kind")?, + pubkey, + channel_id: channel + .map(|value| Uuid::parse_str(&value)) + .transpose() + .map_err(|error| sqlx::Error::Decode(error.into()))?, + created_at: row.try_get("created_at_s")?, + rank: -(row.try_get::("score")? as f32), + }); + } + Ok(SearchResult { hits, page }) +} + #[cfg(test)] mod tests { use super::*; @@ -357,6 +483,52 @@ mod tests { ); } + #[tokio::test] + async fn sqlite_fts_is_tenant_scoped_prefix_capable_and_privacy_filtered() { + let pool = sqlx::sqlite::SqlitePoolOptions::new() + .max_connections(1) + .connect("sqlite::memory:") + .await + .unwrap(); + for statement in [ + "CREATE TABLE events (community_id TEXT NOT NULL, id BLOB NOT NULL, pubkey BLOB NOT NULL, created_at INTEGER NOT NULL, kind INTEGER NOT NULL, content TEXT NOT NULL, channel_id TEXT, deleted_at INTEGER)", + "CREATE VIRTUAL TABLE events_fts USING fts5(content, content='events', content_rowid='rowid', tokenize='unicode61')", + "CREATE TRIGGER events_fts_insert AFTER INSERT ON events WHEN new.kind NOT IN (1059, 30300, 30350, 30622, 44100, 44101, 44200) BEGIN INSERT INTO events_fts(rowid, content) VALUES (new.rowid, new.content); END", + ] { + sqlx::query(statement).execute(&pool).await.unwrap(); + } + let community = CommunityId::from_uuid(Uuid::new_v4()); + let foreign = CommunityId::from_uuid(Uuid::new_v4()); + let channel = Uuid::new_v4(); + for (cid, id, kind, content, channel_id) in [ + (community, vec![1; 32], 9, "project orchard", Some(channel)), + (community, vec![2; 32], 30300, "project private", None), + (foreign, vec![3; 32], 9, "project foreign", None), + ] { + sqlx::query("INSERT INTO events (community_id,id,pubkey,created_at,kind,content,channel_id) VALUES (?1,?2,?3,100,?4,?5,?6)") + .bind(cid.as_uuid().to_string()).bind(id).bind(vec![9; 32]).bind(kind).bind(content).bind(channel_id.map(|id| id.to_string())).execute(&pool).await.unwrap(); + } + let result = search_sqlite( + &pool, + &SearchQuery { + community, + q: "pro".into(), + channel_scope: ChannelScope::Channels(vec![channel]), + kinds: None, + authors: None, + since: None, + until: None, + page: 1, + per_page: 100, + mode: SearchMode::Prefix, + }, + ) + .await + .unwrap(); + assert_eq!(result.hits.len(), 1); + assert_eq!(result.hits[0].event_id, [1; 32]); + } + #[test] fn normalized_search_text_caps_length() { let long = "x".repeat(SEARCH_TEXT_MAX_CHARS + 10);