diff --git a/crates/buzz-db/src/event.rs b/crates/buzz-db/src/event.rs index 3a0b66e34..7e282712b 100644 --- a/crates/buzz-db/src/event.rs +++ b/crates/buzz-db/src/event.rs @@ -135,7 +135,9 @@ pub(crate) struct EventQueryParityFixture { pub second_id: Vec, pub third_id: Vec, pub first_author: Vec, + pub gated_reader: Vec, pub p_tag_hex: String, + pub e_tag_hex: String, pub channel_id: Uuid, } @@ -178,7 +180,7 @@ pub(crate) fn event_query_parity_vectors( fixture: &EventQueryParityFixture, ) -> Vec<(&'static str, EventQuery, Vec>, i64)> { let mut kinds = EventQuery::for_community(community_id); - kinds.kinds = Some(vec![1]); + kinds.kinds = Some(vec![30_175]); let mut author = EventQuery::for_community(community_id); author.authors = Some(vec![fixture.first_author.clone()]); @@ -201,6 +203,14 @@ pub(crate) fn event_query_parity_vectors( let mut cursor = EventQuery::for_community(community_id); cursor.until = Some(DateTime::from_timestamp(102, 0).expect("valid timestamp")); cursor.before_id = Some(fixture.third_id.clone()); + cursor.kinds = Some(vec![2, 30_023]); + cursor.channel_ids = Some(vec![fixture.channel_id]); + + let mut e_tag = EventQuery::for_community(community_id); + e_tag.e_tags = Some(vec![fixture.e_tag_hex.clone(), "missing".to_string()]); + + let mut gated = EventQuery::for_community(community_id); + gated.shared_gated_reader = Some(fixture.gated_reader.clone()); let mut limited = EventQuery::for_community(community_id); limited.limit = Some(2); @@ -234,11 +244,18 @@ pub(crate) fn event_query_parity_vectors( 2, ), ( - "cursor", + "cursor composition", cursor, vec![fixture.second_id.clone(), fixture.first_id.clone()], 2, ), + ("e-tag", e_tag, vec![fixture.first_id.clone()], 1), + ( + "gated-reader", + gated, + vec![fixture.second_id.clone(), fixture.first_id.clone()], + 2, + ), ( "limit", limited, @@ -452,244 +469,181 @@ pub async fn query_events(pool: &PgPool, q: &EventQuery) -> Result, + predicates: &[crate::query_plan::Predicate], + joined_mentions: bool, +) { + use crate::query_plan::Predicate; + + let prefix = if joined_mentions { "e." } else { "" }; + for predicate in predicates { + match predicate { + Predicate::MatchNone => { + qb.push(" AND FALSE"); + } + Predicate::Community(community) => { + qb.push(format!(" AND {prefix}community_id = ")) + .push_bind(community.as_uuid()); + if joined_mentions { + qb.push(" AND m.community_id = ") + .push_bind(community.as_uuid()); + } + } + Predicate::Channel(channel) => { + qb.push(format!(" AND {prefix}channel_id = ")) + .push_bind(*channel); + } + Predicate::GlobalOnly => { + qb.push(format!(" AND {prefix}channel_id IS NULL")); + } + Predicate::ChannelsOrGlobal(channels) => { + if channels.is_empty() { + qb.push(format!(" AND {prefix}channel_id IS NULL")); + } else { + qb.push(format!( + " AND ({prefix}channel_id IS NULL OR {prefix}channel_id IN (" + )); + let mut values = qb.separated(", "); + for channel in channels { + values.push_bind(*channel); + } + qb.push("))"); + } + } + Predicate::Kinds(kinds) => { + qb.push(format!(" AND {prefix}kind IN (")); + let mut values = qb.separated(", "); + for kind in kinds { + values.push_bind(*kind); + } + qb.push(")"); + } + Predicate::Author(author) => { + qb.push(format!(" AND {prefix}pubkey = ")) + .push_bind(author.clone()); + } + Predicate::Authors(authors) => { + qb.push(format!(" AND {prefix}pubkey IN (")); + let mut values = qb.separated(", "); + for author in authors { + values.push_bind(author.clone()); + } + qb.push(")"); + } + Predicate::Ids(ids) => { + qb.push(format!(" AND {prefix}id IN (")); + let mut values = qb.separated(", "); + for id in ids { + values.push_bind(id.clone()); + } + qb.push(")"); + } + Predicate::HasPTag(pubkey) => { + qb.push(" AND m.pubkey_hex = ").push_bind(pubkey.clone()); + } + Predicate::HasETag(event_ids) => { + qb.push(" AND ("); + for (index, event_id) in event_ids.iter().enumerate() { + if index > 0 { + qb.push(" OR "); + } + qb.push(format!("{prefix}tags @> ")) + .push_bind(serde_json::json!([["e", event_id]])); + } + qb.push(")"); + } + Predicate::HasDTag(value) => { + qb.push(format!(" AND {prefix}d_tag = ")) + .push_bind(value.clone()); + } + Predicate::HasAnyDTag(values) => { + qb.push(format!(" AND {prefix}d_tag IN (")); + let mut separated = qb.separated(", "); + for value in values { + separated.push_bind(value.clone()); + } + qb.push(")"); + } + Predicate::Since(since) => { + qb.push(format!(" AND {prefix}created_at >= ")) + .push_bind(*since); + } + Predicate::Until(until) => { + qb.push(format!(" AND {prefix}created_at <= ")) + .push_bind(*until); + } + Predicate::CursorBefore { until, id } => { + qb.push(format!(" AND ({prefix}created_at < ")) + .push_bind(*until) + .push(format!(" OR ({prefix}created_at = ")) + .push_bind(*until) + .push(format!(" AND {prefix}id > ")) + .push_bind(id.clone()) + .push("))"); + } + Predicate::GatedReader(reader) => { + qb.push(format!(" AND ({prefix}kind NOT IN (")); + let mut kinds = qb.separated(", "); + for kind in SHARED_GATED_KINDS { + kinds.push_bind(*kind as i32); + } + qb.push(format!(") OR {prefix}pubkey = ")) + .push_bind(reader.clone()) + .push(format!(" OR {prefix}tags @> ")) + .push_bind(serde_json::json!([["shared", "true"]])) + .push(")"); + } + Predicate::LiveOnly => { + qb.push(format!(" AND {prefix}deleted_at IS NULL")); + } + } + } +} + +fn event_query_builder( + select: &'static str, + plan: &crate::query_plan::QueryPlan, +) -> (QueryBuilder, bool) { + let joined_mentions = plan + .predicates + .iter() + .any(|predicate| matches!(predicate, crate::query_plan::Predicate::HasPTag(_))); + let mut qb = if joined_mentions { + QueryBuilder::new(format!("{select} FROM events e INNER JOIN event_mentions m ON e.community_id = m.community_id AND e.id = m.event_id WHERE TRUE")) + } else { + QueryBuilder::new(format!("{select} FROM events WHERE TRUE")) + }; + render_event_predicates(&mut qb, &plan.predicates, joined_mentions); + (qb, joined_mentions) +} + pub(crate) async fn query_events_on( conn: &mut sqlx::PgConnection, q: &EventQuery, ) -> Result> { - // Composite cursor requires both halves. - if q.before_id.is_some() && q.until.is_none() { - return Err(DbError::InvalidData( - "before_id requires until to be set".to_string(), - )); - } - - // global_only and channel_id are mutually exclusive. - if q.global_only && q.channel_id.is_some() { - return Err(DbError::InvalidData( - "global_only and channel_id are mutually exclusive".to_string(), - )); - } - - // Empty list means "match nothing" — return empty immediately. - if q.kinds.as_deref().is_some_and(|k| k.is_empty()) { - return Ok(vec![]); - } - if q.authors.as_deref().is_some_and(|a| a.is_empty()) { - return Ok(vec![]); - } - if q.ids.as_deref().is_some_and(|i| i.is_empty()) { - return Ok(vec![]); - } - if q.e_tags.as_deref().is_some_and(|e| e.is_empty()) { - return Ok(vec![]); - } - - let clamp = q.max_limit.unwrap_or(DEFAULT_MAX_PAGE_LIMIT); - let limit_val = q.limit.unwrap_or(100).min(clamp); - let offset_val = q.offset.unwrap_or(0); - - let mut qb: QueryBuilder = if let Some(ref p_hex) = q.p_tag_hex { - // Join against event_mentions for #p-filtered queries (indexed). - let mut b = QueryBuilder::new( - "SELECT e.id, e.pubkey, e.created_at, e.kind, e.tags, e.content, \ - e.sig, e.received_at, e.channel_id \ - FROM events e \ - INNER JOIN event_mentions m \ - ON e.community_id = m.community_id AND e.id = m.event_id \ - WHERE e.community_id = ", - ); - b.push_bind(q.community_id.as_uuid()); - b.push(" AND m.community_id = "); - b.push_bind(q.community_id.as_uuid()); - b.push(" AND e.deleted_at IS NULL AND m.pubkey_hex = "); - b.push_bind(p_hex.to_ascii_lowercase()); - b + let plan = crate::query_plan::plan(q)?; + let select = if plan + .predicates + .iter() + .any(|predicate| matches!(predicate, crate::query_plan::Predicate::HasPTag(_))) + { + "SELECT e.id, e.pubkey, e.created_at, e.kind, e.tags, e.content, e.sig, e.received_at, e.channel_id" } else { - let mut b = QueryBuilder::new( - "SELECT id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id \ - FROM events WHERE community_id = ", - ); - b.push_bind(q.community_id.as_uuid()); - b.push(" AND deleted_at IS NULL"); - b + "SELECT id, pubkey, created_at, kind, tags, content, sig, received_at, channel_id" }; - - // Use unqualified column names when no join, qualified when joined. - let col_prefix = if q.p_tag_hex.is_some() { "e." } else { "" }; - - if let Some(ch) = q.channel_id { - qb.push(format!(" AND {col_prefix}channel_id = ")) - .push_bind(ch); - } else if q.global_only { - qb.push(format!(" AND {col_prefix}channel_id IS NULL")); - } - - // Multi-channel IN pushdown: restrict to events in any of these channels - // OR global events (channel_id IS NULL). Used by NIP-45 COUNT to enforce - // channel access at the SQL level without fetching all rows. - // - // SECURITY: Some(empty vec) means "user has access to NO channels" — - // only global events (channel_id IS NULL) should be returned. - if let Some(ref ch_ids) = q.channel_ids { - if ch_ids.is_empty() { - // No channel access — only global (non-channel) events visible. - qb.push(format!(" AND {col_prefix}channel_id IS NULL")); - } else { - qb.push(format!( - " AND ({col_prefix}channel_id IS NULL OR {col_prefix}channel_id IN (" - )); - let mut sep = qb.separated(", "); - for ch in ch_ids { - sep.push_bind(*ch); - } - qb.push("))"); - } - } - - if let Some(ks) = q.kinds.as_deref().filter(|k| !k.is_empty()) { - qb.push(format!(" AND {col_prefix}kind IN (")); - let mut sep = qb.separated(", "); - for k in ks { - sep.push_bind(*k); - } - qb.push(")"); - } - - if let Some(ref pk) = q.pubkey { - qb.push(format!(" AND {col_prefix}pubkey = ")) - .push_bind(pk.clone()); - } - - // Multi-author IN pushdown (mutually exclusive with single pubkey in practice). - if let Some(ref authors) = q.authors { - if !authors.is_empty() { - qb.push(format!(" AND {col_prefix}pubkey IN (")); - let mut sep = qb.separated(", "); - for a in authors { - sep.push_bind(a.clone()); - } - qb.push(")"); - } - } - - // Multi-id IN pushdown. - if let Some(ref ids) = q.ids { - if !ids.is_empty() { - qb.push(format!(" AND {col_prefix}id IN (")); - let mut sep = qb.separated(", "); - for id in ids { - sep.push_bind(id.clone()); - } - qb.push(")"); - } - } - - // e-tag pushdown via JSONB containment: tags @> '[["e",""]]'. - // Multiple e-tags use OR (any match). Served by idx_events_tags_gin - // (GIN, jsonb_path_ops — migrations/0004): the channel-window aux closure - // fans this out once per retained row, which made unindexed containment - // the dominant scroll-back cost (~1.7s/page on staging). - if let Some(ref e_tags) = q.e_tags { - if !e_tags.is_empty() { - qb.push(" AND ("); - for (i, hex_id) in e_tags.iter().enumerate() { - if i > 0 { - qb.push(" OR "); - } - // Build the JSONB literal: [["e",""]] - let containment = serde_json::json!([["e", hex_id]]); - qb.push(format!("{col_prefix}tags @> ")); - qb.push_bind(containment); - } - qb.push(")"); - } - } - - if let Some(s) = q.since { - qb.push(format!(" AND {col_prefix}created_at >= ")) - .push_bind(s); - } - if let Some(u) = q.until { - if let Some(ref bid) = q.before_id { - // Composite keyset cursor for stable pagination. - // With ORDER BY created_at DESC, id ASC, "next page" means: - // created_at < cursor_ts OR (created_at = cursor_ts AND id > cursor_id) - qb.push(format!(" AND ({col_prefix}created_at < ")); - qb.push_bind(u); - qb.push(format!(" OR ({col_prefix}created_at = ")); - qb.push_bind(u); - qb.push(format!(" AND {col_prefix}id > ")); - qb.push_bind(bid.clone()); - qb.push("))"); - } else { - qb.push(format!(" AND {col_prefix}created_at <= ")) - .push_bind(u); - } - } - - if let Some(ref d) = q.d_tag { - qb.push(format!(" AND {col_prefix}d_tag = ")) - .push_bind(d.clone()); - } else if let Some(ref ds) = q.d_tags { - if !ds.is_empty() { - qb.push(format!(" AND {col_prefix}d_tag IN (")); - let mut sep = qb.separated(", "); - for d in ds { - sep.push_bind(d.clone()); - } - qb.push(")"); - } - } - - // Shared-gated visibility pushdown: exclude SHARED_GATED_KINDS events that - // are neither authored by the reader nor explicitly shared. Applied BEFORE - // ORDER/LIMIT so that a page of newer private events does not push visible - // shared ones off the end of the result set (the catalog query pattern). - // - // Clause: AND (kind NOT IN (30175, 30178) OR pubkey = $reader - // OR tags @> '[["shared","true"]]') - // - // The JSONB containment check is served by idx_events_tags_gin (migration - // 0004, jsonb_path_ops). `tags @> '[["shared","true"]]'` matches any array - // that contains exactly the sub-array — a two-element `["shared","true"]` - // tag passes; a tag-absent event does not. Because ingest requires exactly - // two elements for the shared tag (parts.len() == 2), no stored event can - // carry a three-element superset. - if let Some(ref reader_bytes) = q.shared_gated_reader { - let shared_containment = serde_json::json!([["shared", "true"]]); - qb.push(format!(" AND ({col_prefix}kind NOT IN (")); - let mut sep = qb.separated(", "); - for kind in SHARED_GATED_KINDS { - sep.push_bind(*kind as i32); - } - qb.push(format!(") OR {col_prefix}pubkey = ")); - qb.push_bind(reader_bytes.clone()); - qb.push(format!(" OR {col_prefix}tags @> ")); - qb.push_bind(shared_containment); - qb.push(")"); - } - - // Composite ordering for deterministic pagination across ALL callers of - // query_events (WebSocket REQ, REST endpoints, canvas, notes, etc.). - // The `id ASC` tiebreaker ensures stable results when events share the - // same second. No existing index covers this trailing column — Postgres - // sorts in memory, which is fine at current scale. If query performance - // degrades, add a composite index like `(pubkey, kind, created_at DESC, id ASC)`. + let (mut qb, joined_mentions) = event_query_builder(select, &plan); + let prefix = if joined_mentions { "e." } else { "" }; qb.push(format!( - " ORDER BY {col_prefix}created_at DESC, {col_prefix}id ASC LIMIT " - )); - qb.push_bind(limit_val); - qb.push(" OFFSET ").push_bind(offset_val); - + " ORDER BY {prefix}created_at DESC, {prefix}id ASC LIMIT " + )) + .push_bind(plan.modifiers.limit) + .push(" OFFSET ") + .push_bind(plan.modifiers.offset); let rows = qb.build().fetch_all(&mut *conn).await?; - - let mut out = Vec::with_capacity(rows.len()); - for row in rows { - if let Some(ev) = row_to_stored_event(row)? { - out.push(ev); - } - } - Ok(out) + rows.into_iter() + .filter_map(|row| row_to_stored_event(row).transpose()) + .collect() } pub(crate) fn row_to_stored_event(row: sqlx::postgres::PgRow) -> Result> { @@ -747,154 +701,10 @@ pub async fn count_events(pool: &PgPool, q: &EventQuery) -> Result { /// the count on the exact reader connection whose heartbeat observation /// proved its predicate. pub(crate) async fn count_events_on(conn: &mut sqlx::PgConnection, q: &EventQuery) -> Result { - // Empty list means "match nothing" — return 0 immediately. - if q.kinds.as_deref().is_some_and(|k| k.is_empty()) { - return Ok(0); - } - if q.authors.as_deref().is_some_and(|a| a.is_empty()) { - return Ok(0); - } - if q.ids.as_deref().is_some_and(|i| i.is_empty()) { - return Ok(0); - } - if q.e_tags.as_deref().is_some_and(|e| e.is_empty()) { - return Ok(0); - } - - let mut qb: QueryBuilder = if let Some(ref p_hex) = q.p_tag_hex { - let mut b = QueryBuilder::new( - "SELECT COUNT(*) as cnt FROM events e \ - INNER JOIN event_mentions m \ - ON e.community_id = m.community_id AND e.id = m.event_id \ - WHERE e.community_id = ", - ); - b.push_bind(q.community_id.as_uuid()); - b.push(" AND m.community_id = "); - b.push_bind(q.community_id.as_uuid()); - b.push(" AND e.deleted_at IS NULL AND m.pubkey_hex = "); - b.push_bind(p_hex.to_ascii_lowercase()); - b - } else { - let mut b = QueryBuilder::new("SELECT COUNT(*) as cnt FROM events WHERE community_id = "); - b.push_bind(q.community_id.as_uuid()); - b.push(" AND deleted_at IS NULL"); - b - }; - - let col_prefix = if q.p_tag_hex.is_some() { "e." } else { "" }; - - if let Some(ch) = q.channel_id { - qb.push(format!(" AND {col_prefix}channel_id = ")) - .push_bind(ch); - } else if q.global_only { - qb.push(format!(" AND {col_prefix}channel_id IS NULL")); - } - - // Multi-channel IN pushdown for COUNT: restrict to accessible channels + global. - // SECURITY: Some(empty vec) = no channel access → global events only. - if let Some(ref ch_ids) = q.channel_ids { - if ch_ids.is_empty() { - qb.push(format!(" AND {col_prefix}channel_id IS NULL")); - } else { - qb.push(format!( - " AND ({col_prefix}channel_id IS NULL OR {col_prefix}channel_id IN (" - )); - let mut sep = qb.separated(", "); - for ch in ch_ids { - sep.push_bind(*ch); - } - qb.push("))"); - } - } - - if let Some(ks) = q.kinds.as_deref().filter(|k| !k.is_empty()) { - qb.push(format!(" AND {col_prefix}kind IN (")); - let mut sep = qb.separated(", "); - for k in ks { - sep.push_bind(*k); - } - qb.push(")"); - } - - if let Some(ref pk) = q.pubkey { - qb.push(format!(" AND {col_prefix}pubkey = ")) - .push_bind(pk.clone()); - } - - if let Some(ref authors) = q.authors { - if !authors.is_empty() { - qb.push(format!(" AND {col_prefix}pubkey IN (")); - let mut sep = qb.separated(", "); - for a in authors { - sep.push_bind(a.clone()); - } - qb.push(")"); - } - } - - if let Some(ref ids) = q.ids { - if !ids.is_empty() { - qb.push(format!(" AND {col_prefix}id IN (")); - let mut sep = qb.separated(", "); - for id in ids { - sep.push_bind(id.clone()); - } - qb.push(")"); - } - } - - if let Some(ref e_tags) = q.e_tags { - if !e_tags.is_empty() { - qb.push(" AND ("); - for (i, hex_id) in e_tags.iter().enumerate() { - if i > 0 { - qb.push(" OR "); - } - let containment = serde_json::json!([["e", hex_id]]); - qb.push(format!("{col_prefix}tags @> ")); - qb.push_bind(containment); - } - qb.push(")"); - } - } - - if let Some(s) = q.since { - qb.push(format!(" AND {col_prefix}created_at >= ")) - .push_bind(s); - } - if let Some(u) = q.until { - if let Some(ref bid) = q.before_id { - qb.push(format!(" AND ({col_prefix}created_at < ")); - qb.push_bind(u); - qb.push(format!(" OR ({col_prefix}created_at = ")); - qb.push_bind(u); - qb.push(format!(" AND {col_prefix}id > ")); - qb.push_bind(bid.clone()); - qb.push("))"); - } else { - qb.push(format!(" AND {col_prefix}created_at <= ")) - .push_bind(u); - } - } - - if let Some(ref d) = q.d_tag { - qb.push(format!(" AND {col_prefix}d_tag = ")) - .push_bind(d.clone()); - } else if let Some(ref ds) = q.d_tags { - if !ds.is_empty() { - qb.push(format!(" AND {col_prefix}d_tag IN (")); - let mut sep = qb.separated(", "); - for d in ds { - sep.push_bind(d.clone()); - } - qb.push(")"); - } - } - + let plan = crate::query_plan::plan(q)?; + let (mut qb, _) = event_query_builder("SELECT COUNT(*) AS cnt", &plan); let row = qb.build().fetch_one(&mut *conn).await?; - let cnt: i64 = row.try_get("cnt")?; - - Ok(cnt) + Ok(row.try_get("cnt")?) } /// Soft-delete an event by setting `deleted_at = NOW()`. @@ -2018,10 +1828,12 @@ mod tests { let first_keys = Keys::generate(); let second_keys = Keys::generate(); let p_tag_hex = hex::encode(second_keys.public_key().to_bytes()); + let e_tag_hex = "ab".repeat(32); let first = EventBuilder::new(Kind::Custom(30_023), "first") .tags([ Tag::parse(["d", "alpha"]).unwrap(), Tag::parse(["p", &p_tag_hex.to_ascii_uppercase()]).unwrap(), + Tag::parse(["e", &e_tag_hex]).unwrap(), ]) .custom_created_at(Timestamp::from(100_u64)) .sign_with_keys(&first_keys) @@ -2030,7 +1842,7 @@ mod tests { .custom_created_at(Timestamp::from(101_u64)) .sign_with_keys(&second_keys) .unwrap(); - let third = EventBuilder::new(Kind::Custom(1), "third") + let third = EventBuilder::new(Kind::Custom(30_175), "third") .custom_created_at(Timestamp::from(102_u64)) .sign_with_keys(&first_keys) .unwrap(); @@ -2049,7 +1861,9 @@ mod tests { second_id: second.id.as_bytes().to_vec(), third_id: third.id.as_bytes().to_vec(), first_author: first.pubkey.to_bytes().to_vec(), + gated_reader: second.pubkey.to_bytes().to_vec(), p_tag_hex, + e_tag_hex, channel_id, }; for (name, query, expected_ids, expected_count) in diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 1a2fe3f87..e5eca466d 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -37,6 +37,7 @@ pub mod partition; pub mod product_feedback; /// Community-scoped push lease and durable wake-outbox persistence. pub mod push; +mod query_plan; /// Reaction persistence. pub mod reaction; /// Use-limited relay invite persistence (v2 opaque tokens). diff --git a/crates/buzz-db/src/query_plan.rs b/crates/buzz-db/src/query_plan.rs new file mode 100644 index 000000000..761e47bcf --- /dev/null +++ b/crates/buzz-db/src/query_plan.rs @@ -0,0 +1,149 @@ +//! Backend-neutral normalization for event queries. + +use chrono::{DateTime, Utc}; +use uuid::Uuid; + +use buzz_core::CommunityId; + +use crate::{DbError, EventQuery, Result, DEFAULT_MAX_PAGE_LIMIT}; + +/// A normalized event query, split into row predicates and result modifiers. +#[derive(Debug, Clone)] +pub(crate) struct QueryPlan { + pub(crate) predicates: Vec, + pub(crate) modifiers: Modifiers, +} + +/// Backend-neutral logical row conditions. +#[derive(Debug, Clone)] +pub(crate) enum Predicate { + MatchNone, + Community(CommunityId), + Channel(Uuid), + GlobalOnly, + ChannelsOrGlobal(Vec), + Kinds(Vec), + Author(Vec), + Authors(Vec>), + Ids(Vec>), + HasPTag(String), + HasETag(Vec), + HasDTag(String), + HasAnyDTag(Vec), + Since(DateTime), + Until(DateTime), + CursorBefore { until: DateTime, id: Vec }, + GatedReader(Vec), + LiveOnly, +} + +/// Ordering and pagination, deliberately excluded from count rendering. +#[derive(Debug, Clone, Copy)] +pub(crate) struct Modifiers { + pub(crate) limit: i64, + pub(crate) offset: i64, +} + +/// Validate and normalize every [`EventQuery`] field exactly once. +pub(crate) fn plan(query: &EventQuery) -> Result { + let EventQuery { + community_id, + channel_id, + kinds, + pubkey, + since, + until, + limit, + offset, + p_tag_hex, + d_tag, + d_tags, + before_id, + global_only, + authors, + ids, + e_tags, + channel_ids, + max_limit, + shared_gated_reader, + } = query; + + if before_id.is_some() && until.is_none() { + return Err(DbError::InvalidData( + "before_id requires until to be set".to_string(), + )); + } + if *global_only && channel_id.is_some() { + return Err(DbError::InvalidData( + "global_only and channel_id are mutually exclusive".to_string(), + )); + } + + let mut predicates = vec![Predicate::Community(*community_id), Predicate::LiveOnly]; + if kinds.as_ref().is_some_and(Vec::is_empty) + || authors.as_ref().is_some_and(Vec::is_empty) + || ids.as_ref().is_some_and(Vec::is_empty) + || e_tags.as_ref().is_some_and(Vec::is_empty) + { + predicates.push(Predicate::MatchNone); + } + + if let Some(channel_id) = channel_id { + predicates.push(Predicate::Channel(*channel_id)); + } else if *global_only { + predicates.push(Predicate::GlobalOnly); + } + if let Some(channel_ids) = channel_ids { + predicates.push(Predicate::ChannelsOrGlobal(channel_ids.clone())); + } + if let Some(kinds) = kinds.as_ref().filter(|values| !values.is_empty()) { + predicates.push(Predicate::Kinds(kinds.clone())); + } + if let Some(pubkey) = pubkey { + predicates.push(Predicate::Author(pubkey.clone())); + } + if let Some(authors) = authors.as_ref().filter(|values| !values.is_empty()) { + predicates.push(Predicate::Authors(authors.clone())); + } + if let Some(ids) = ids.as_ref().filter(|values| !values.is_empty()) { + predicates.push(Predicate::Ids(ids.clone())); + } + if let Some(p_tag_hex) = p_tag_hex { + predicates.push(Predicate::HasPTag(p_tag_hex.to_ascii_lowercase())); + } + if let Some(e_tags) = e_tags.as_ref().filter(|values| !values.is_empty()) { + predicates.push(Predicate::HasETag(e_tags.clone())); + } + if let Some(since) = since { + predicates.push(Predicate::Since(*since)); + } + if let Some(until) = until { + if let Some(id) = before_id { + predicates.push(Predicate::CursorBefore { + until: *until, + id: id.clone(), + }); + } else { + predicates.push(Predicate::Until(*until)); + } + } + // A singular d-tag takes precedence over the multi-value form. + if let Some(d_tag) = d_tag { + predicates.push(Predicate::HasDTag(d_tag.clone())); + } else if let Some(d_tags) = d_tags.as_ref().filter(|values| !values.is_empty()) { + predicates.push(Predicate::HasAnyDTag(d_tags.clone())); + } + if let Some(reader) = shared_gated_reader { + predicates.push(Predicate::GatedReader(reader.clone())); + } + + Ok(QueryPlan { + predicates, + modifiers: Modifiers { + limit: limit + .unwrap_or(100) + .min(max_limit.unwrap_or(DEFAULT_MAX_PAGE_LIMIT)), + offset: offset.unwrap_or(0), + }, + }) +} diff --git a/crates/buzz-db/src/sqlite.rs b/crates/buzz-db/src/sqlite.rs index a55f7e79e..ca93411ec 100644 --- a/crates/buzz-db/src/sqlite.rs +++ b/crates/buzz-db/src/sqlite.rs @@ -1165,144 +1165,147 @@ pub(crate) async fn get_events_by_ids( .collect() } +fn render_sqlite_event_predicates( + qb: &mut sqlx::QueryBuilder, + predicates: &[crate::query_plan::Predicate], +) { + use crate::query_plan::Predicate; + + for predicate in predicates { + match predicate { + Predicate::MatchNone => { + qb.push(" AND FALSE"); + } + Predicate::Community(community) => { + qb.push(" AND community_id = ") + .push_bind(community.as_uuid().to_string()); + } + Predicate::Channel(channel) => { + qb.push(" AND channel_id = ").push_bind(channel.to_string()); + } + Predicate::GlobalOnly => { + qb.push(" AND channel_id IS NULL"); + } + Predicate::ChannelsOrGlobal(channels) => { + if channels.is_empty() { + qb.push(" AND channel_id IS NULL"); + } else { + qb.push(" AND (channel_id IS NULL OR channel_id IN ("); + let mut values = qb.separated(", "); + for channel in channels { + values.push_bind(channel.to_string()); + } + qb.push("))"); + } + } + Predicate::Kinds(kinds) => { + qb.push(" AND kind IN ("); + let mut values = qb.separated(", "); + for kind in kinds { + values.push_bind(*kind); + } + qb.push(")"); + } + Predicate::Author(author) => { + qb.push(" AND pubkey = ").push_bind(author.clone()); + } + Predicate::Authors(authors) => { + qb.push(" AND pubkey IN ("); + let mut values = qb.separated(", "); + for author in authors { + values.push_bind(author.clone()); + } + qb.push(")"); + } + Predicate::Ids(ids) => { + qb.push(" AND id IN ("); + let mut values = qb.separated(", "); + for id in ids { + values.push_bind(id.clone()); + } + qb.push(")"); + } + Predicate::HasPTag(pubkey) => { + qb.push(" AND EXISTS (SELECT 1 FROM json_each(tags_json) tag WHERE json_extract(tag.value, '$[0]') = 'p' AND LOWER(json_extract(tag.value, '$[1]')) = ") + .push_bind(pubkey.clone()).push(")"); + } + Predicate::HasETag(event_ids) => { + qb.push(" AND EXISTS (SELECT 1 FROM json_each(tags_json) tag WHERE json_extract(tag.value, '$[0]') = 'e' AND json_extract(tag.value, '$[1]') IN ("); + let mut values = qb.separated(", "); + for id in event_ids { + values.push_bind(id.clone()); + } + qb.push("))"); + } + Predicate::HasDTag(value) => { + qb.push(" AND (SELECT json_extract(tag.value, '$[1]') FROM json_each(tags_json) tag WHERE json_extract(tag.value, '$[0]') = 'd' ORDER BY tag.key LIMIT 1) = ") + .push_bind(value.clone()); + } + Predicate::HasAnyDTag(values) => { + qb.push(" AND (SELECT json_extract(tag.value, '$[1]') FROM json_each(tags_json) tag WHERE json_extract(tag.value, '$[0]') = 'd' ORDER BY tag.key LIMIT 1) IN ("); + let mut separated = qb.separated(", "); + for value in values { + separated.push_bind(value.clone()); + } + qb.push(")"); + } + Predicate::Since(since) => { + qb.push(" AND created_at >= ").push_bind(since.timestamp()); + } + Predicate::Until(until) => { + qb.push(" AND created_at <= ").push_bind(until.timestamp()); + } + Predicate::CursorBefore { until, id } => { + qb.push(" AND (created_at < ") + .push_bind(until.timestamp()) + .push(" OR (created_at = ") + .push_bind(until.timestamp()) + .push(" AND id > ") + .push_bind(id.clone()) + .push("))"); + } + Predicate::GatedReader(reader) => { + qb.push(" AND (kind NOT IN ("); + let mut kinds = qb.separated(", "); + for kind in buzz_core::kind::SHARED_GATED_KINDS { + kinds.push_bind(*kind as i32); + } + qb.push(") OR pubkey = ").push_bind(reader.clone()) + .push(" OR EXISTS (SELECT 1 FROM json_each(tags_json) tag WHERE json_extract(tag.value, '$[0]') = 'shared' AND json_extract(tag.value, '$[1]') = 'true'))"); + } + Predicate::LiveOnly => {} + } + } +} + pub(crate) async fn query_events( pool: &SqlitePool, q: &crate::EventQuery, ) -> Result> { - if q.before_id.is_some() && q.until.is_none() { - return Err(crate::DbError::InvalidData( - "before_id requires until to be set".into(), - )); - } - if q.global_only && q.channel_id.is_some() { - return Err(crate::DbError::InvalidData( - "global_only and channel_id are mutually exclusive".into(), - )); - } - if q.kinds.as_ref().is_some_and(Vec::is_empty) - || q.authors.as_ref().is_some_and(Vec::is_empty) - || q.ids.as_ref().is_some_and(Vec::is_empty) - || q.e_tags.as_ref().is_some_and(Vec::is_empty) - { - return Ok(vec![]); - } - let rows = sqlx::query("SELECT event_json, received_at, channel_id FROM events WHERE community_id = ?1 ORDER BY created_at DESC, id ASC") - .bind(q.community_id.as_uuid().to_string()).fetch_all(pool).await?; - let mut events = Vec::new(); - for row in rows { - let stored = stored_event(row)?; - let event = &stored.event; - let created = event.created_at.as_secs() as i64; - let id = event.id.as_bytes().as_slice(); - let tags: Vec> = event - .tags - .iter() - .map(|tag| tag.as_slice().to_vec()) - .collect(); - let has_tag = |name: &str, value: &str| { - tags.iter().any(|tag| { - tag.first().is_some_and(|v| v == name) && tag.get(1).is_some_and(|v| v == value) - }) - }; - if q.channel_id.is_some_and(|ch| stored.channel_id != Some(ch)) - || (q.global_only && stored.channel_id.is_some()) - { - continue; - } - if q.channel_ids - .as_ref() - .is_some_and(|ids| stored.channel_id.is_some_and(|id| !ids.contains(&id))) - { - continue; - } - if q.kinds - .as_ref() - .is_some_and(|ks| !ks.contains(&(event.kind.as_u16() as i32))) - { - continue; - } - if q.pubkey - .as_ref() - .is_some_and(|pk| pk.as_slice() != event.pubkey.to_bytes().as_slice()) - { - continue; - } - if q.authors.as_ref().is_some_and(|authors| { - !authors - .iter() - .any(|pk| pk.as_slice() == event.pubkey.to_bytes().as_slice()) - }) { - continue; - } - if q.ids - .as_ref() - .is_some_and(|ids| !ids.iter().any(|candidate| candidate.as_slice() == id)) - { - continue; - } - if q.since.is_some_and(|since| created < since.timestamp()) - || q.until.is_some_and(|until| created > until.timestamp()) - { - continue; - } - if q.before_id.as_ref().is_some_and(|before| { - q.until.is_some_and(|until| created == until.timestamp()) && id <= before.as_slice() - }) { - continue; - } - if q.p_tag_hex.as_ref().is_some_and(|p| { - !tags.iter().any(|tag| { - tag.first().is_some_and(|name| name == "p") - && tag - .get(1) - .is_some_and(|value| value.eq_ignore_ascii_case(p)) - }) - }) { - continue; - } - if q.e_tags - .as_ref() - .is_some_and(|values| !values.iter().any(|value| has_tag("e", value))) - { - continue; - } - let d_tag = tags - .iter() - .find(|tag| tag.first().is_some_and(|v| v == "d")) - .and_then(|tag| tag.get(1)); - if q.d_tag.as_ref().is_some_and(|d| d_tag != Some(d)) { - continue; - } - if q.d_tags - .as_ref() - .is_some_and(|ds| !d_tag.is_some_and(|d| ds.contains(d))) - { - continue; - } - if q.shared_gated_reader.as_ref().is_some_and(|reader| { - buzz_core::kind::SHARED_GATED_KINDS.contains(&(event.kind.as_u16() as u32)) - && reader.as_slice() != event.pubkey.to_bytes().as_slice() - && !has_tag("shared", "true") - }) { - continue; - } - events.push(stored); - } - let offset = q.offset.unwrap_or(0).max(0) as usize; - let limit = q - .limit - .unwrap_or(100) - .min(q.max_limit.unwrap_or(crate::DEFAULT_MAX_PAGE_LIMIT)) - .max(0) as usize; - Ok(events.into_iter().skip(offset).take(limit).collect()) + let plan = crate::query_plan::plan(q)?; + let mut qb = sqlx::QueryBuilder::::new( + "SELECT event_json, received_at, channel_id FROM events WHERE TRUE", + ); + render_sqlite_event_predicates(&mut qb, &plan.predicates); + qb.push(" ORDER BY created_at DESC, id ASC LIMIT ") + .push_bind(plan.modifiers.limit.max(0)) + .push(" OFFSET ") + .push_bind(plan.modifiers.offset.max(0)); + qb.build() + .fetch_all(pool) + .await? + .into_iter() + .map(stored_event) + .collect() } pub(crate) async fn count_events(pool: &SqlitePool, q: &crate::EventQuery) -> Result { - let mut unpaged = q.clone(); - unpaged.offset = None; - unpaged.limit = Some(i64::MAX); - unpaged.max_limit = Some(i64::MAX); - Ok(query_events(pool, &unpaged).await?.len() as i64) + let plan = crate::query_plan::plan(q)?; + let mut qb = + sqlx::QueryBuilder::::new("SELECT COUNT(*) AS cnt FROM events WHERE TRUE"); + render_sqlite_event_predicates(&mut qb, &plan.predicates); + let row = qb.build().fetch_one(pool).await?; + Ok(row.try_get("cnt")?) } pub(crate) async fn huddle_started_link_exists( @@ -4594,10 +4597,12 @@ mod tests { let first_keys = Keys::generate(); let second_keys = Keys::generate(); let p_tag_hex = hex::encode(second_keys.public_key().to_bytes()); + let e_tag_hex = "ab".repeat(32); let first = EventBuilder::new(Kind::Custom(30_023), "first") .tags([ Tag::parse(["d", "alpha"]).unwrap(), Tag::parse(["p", &p_tag_hex.to_ascii_uppercase()]).unwrap(), + Tag::parse(["e", &e_tag_hex]).unwrap(), ]) .custom_created_at(Timestamp::from(100_u64)) .sign_with_keys(&first_keys) @@ -4606,7 +4611,7 @@ mod tests { .custom_created_at(Timestamp::from(101_u64)) .sign_with_keys(&second_keys) .unwrap(); - let third = EventBuilder::new(Kind::Custom(1), "third") + let third = EventBuilder::new(Kind::Custom(30_175), "third") .custom_created_at(Timestamp::from(102_u64)) .sign_with_keys(&first_keys) .unwrap(); @@ -4623,7 +4628,9 @@ mod tests { second_id: second.id.as_bytes().to_vec(), third_id: third.id.as_bytes().to_vec(), first_author: first.pubkey.to_bytes().to_vec(), + gated_reader: second.pubkey.to_bytes().to_vec(), p_tag_hex, + e_tag_hex, channel_id, }; for (name, query, expected_ids, expected_count) in