diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index d1b7874a3b..10078bde3e 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -82,15 +82,16 @@ pub struct EventQuery { /// Restrict results to events with any of these IDs (multi-id `IN` pushdown). pub ids: Option>>, /// Restrict results to events with an `e` tag referencing any of these event IDs (hex). - /// Uses JSONB containment (`tags @> ...`) against the `tags` column. + /// Matched positionally, with JSONB containment as the index prefilter. pub e_tags: Option>, /// Restrict artifact rows ([`ARTIFACT_KINDS`]) to those with a `d` tag - /// matching any of these values, via JSONB containment. Their `d_tag` + /// matching any of these values, matched positionally. Their `d_tag` /// column is NULL (not NIP-33), so this lets identity lookups match before /// SQL `LIMIT`. Rows of other kinds are left to the caller's post-filter. pub d_tag_values: Option>, - /// Restrict results to events with an exact custom tag pair. - /// Uses JSONB containment against `tags` before SQL `LIMIT`. + /// Restrict results to events with an exact custom tag pair: a tag whose + /// first element is the name and second the value, checked positionally + /// (JSONB containment is only the index prefilter) before SQL `LIMIT`. pub custom_tag: Option<(String, String)>, /// Restrict results to events that, for every `(name, values)` entry, /// have a tag whose name is `name` and whose first value is one of @@ -117,17 +118,12 @@ pub struct EventQuery { /// private events are excluded from the candidate page rather than /// discarded after it. /// - /// The clause is: `AND (kind NOT IN (...) OR pubkey = $reader OR tags @> ?)`, - /// where the `IN` list is [`SHARED_GATED_KINDS`] and `?` is the JSONB - /// literal `[["shared","true"]]`. The GIN index on `tags` (migration 0004, - /// jsonb_path_ops) makes the containment check fast. - /// - /// NOTE: `tags @> '[["shared","true"]]'` uses JSONB containment, which - /// matches any tag array that is a superset of `[["shared","true"]]` — it - /// would match `["shared","true","extra"]` too. The ingest `parts.len() == - /// 2` exact-shape check ensures such malformed tags are never stored, so the - /// SQL pushdown is sound. Keeping `event_visible_to_reader` as post-filter - /// defense-in-depth catches any residual mismatch. + /// The clause is: `AND (kind NOT IN (...) OR pubkey = $reader OR )`, + /// where the `IN` list is [`SHARED_GATED_KINDS`] and `` requires a + /// tag whose element 0 is `shared` and element 1 is `true`. Containment on + /// the GIN index (migration 0004, jsonb_path_ops) is the fast first pass; + /// containment alone would also admit decoys like `["x","shared","true"]`. + /// `event_visible_to_reader` remains the post-filter. pub shared_gated_reader: Option>, } @@ -729,14 +725,15 @@ fn build_query_events_sql(q: &EventQuery) -> QueryBuilder { } } - // e-tag pushdown via JSONB containment: tags @> '[["e",""]]'. - // Multiple e-tags use OR (any match). Served by idx_events_tags_gin + // e-tag pushdown: exact positional match with a JSONB containment + // prefilter, tags @> ANY('[["e",""]]', ...). 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() { - push_e_tag_filter(&mut qb, col_prefix, e_tags); + qb.push(" AND "); + push_exact_tag_match(&mut qb, col_prefix, "e", e_tags); } } @@ -745,9 +742,8 @@ fn build_query_events_sql(q: &EventQuery) -> QueryBuilder { } if let Some((ref name, ref value)) = q.custom_tag { - let containment = serde_json::json!([[name, value]]); - qb.push(format!(" AND {col_prefix}tags @> ")) - .push_bind(containment); + qb.push(" AND "); + push_exact_tag_match(&mut qb, col_prefix, name, std::slice::from_ref(value)); } push_tag_value_filters(&mut qb, col_prefix, &q.tag_values); @@ -794,16 +790,12 @@ fn build_query_events_sql(q: &EventQuery) -> QueryBuilder { // 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"]]') + // OR ) // - // 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. + // The shared match is positional, like the other tag pushdowns: plain + // containment would admit a private event carrying ["x","shared","true"], + // which then wastes a page slot before the post-filter drops it. 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 { @@ -811,8 +803,8 @@ fn build_query_events_sql(q: &EventQuery) -> QueryBuilder { } 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(" OR "); + push_exact_tag_match(&mut qb, col_prefix, "shared", &["true".to_owned()]); qb.push(")"); } @@ -863,17 +855,35 @@ async fn fetch_with_e_tag_deadline( Ok(rows) } -/// e-tag pushdown as one array-bound containment test instead of an N-way -/// `OR` chain, so planner cost does not scale with the number of referenced -/// ids (the thread aux hop sends one id per reply). -fn push_e_tag_filter(qb: &mut QueryBuilder, col_prefix: &str, e_tags: &[String]) { - let containments: Vec = e_tags +/// Match a tag whose first element is `name` and whose second is any of +/// `values` (NIP-01), as one parenthesized predicate. +/// +/// JSONB containment alone ignores element position and multiplicity: +/// `[["t","t"]]` is contained in `["t","other"]`, and `[["e",id]]` in +/// `["x","e",id]`. Over-matching before `LIMIT` lets a non-matching newer row +/// take the slot, so containment stays only as the GIN-indexed prefilter +/// (one array-bound test, so planner cost does not scale with `values`) and +/// the positional check makes the predicate exact. +fn push_exact_tag_match( + qb: &mut QueryBuilder, + col_prefix: &str, + name: &str, + values: &[String], +) { + let containments: Vec = values .iter() - .map(|hex_id| serde_json::json!([["e", hex_id]])) + .map(|value| serde_json::json!([[name, value]])) .collect(); - qb.push(format!(" AND {col_prefix}tags @> ANY(")) + qb.push(format!("({col_prefix}tags @> ANY(")) .push_bind(containments) - .push("::jsonb[])"); + .push(format!( + "::jsonb[]) AND EXISTS (SELECT 1 FROM jsonb_array_elements({col_prefix}tags) AS tag \ + WHERE tag->>0 = " + )) + .push_bind(name.to_owned()) + .push(" AND tag->>1 = ANY(") + .push_bind(values.to_vec()) + .push(")))"); } /// Whether `q` carries a [`EventQuery::tag_values`] match. Such a query runs @@ -884,16 +894,8 @@ fn has_tag_values(q: &EventQuery) -> bool { q.tag_values.iter().any(|(_, values)| !values.is_empty()) } -/// Two tests per [`EventQuery::tag_values`] entry, so each entry is an OR -/// over its values and entries are ANDed: -/// -/// - `tags @> ANY(...)` lets Postgres use the `tags` GIN index. On its own -/// it over-matches, because JSONB containment ignores order and repeats: -/// `[["t", "stream"]]` contains `[["t", "t"]]`, and `[["name", "t"]]` -/// contains `[["t", "name"]]`. Those false candidates could fill the page -/// before `LIMIT` and hide a real match. -/// - The `EXISTS` test then keeps only rows with a tag whose name is -/// exactly `name` and whose first value is one of `values`. +/// AND one [`push_exact_tag_match`] per non-empty [`EventQuery::tag_values`] +/// entry, so entries are ANDed and values within an entry are ORed. fn push_tag_value_filters( qb: &mut QueryBuilder, col_prefix: &str, @@ -903,26 +905,13 @@ fn push_tag_value_filters( if values.is_empty() { continue; } - let containments: Vec = values - .iter() - .map(|value| serde_json::json!([[name, value]])) - .collect(); - qb.push(format!(" AND {col_prefix}tags @> ANY(")) - .push_bind(containments) - .push("::jsonb[])"); - qb.push(format!( - " AND EXISTS (SELECT 1 FROM jsonb_array_elements({col_prefix}tags) AS tag \ - WHERE tag->>0 = " - )) - .push_bind(name.clone()) - .push(" AND tag->>1 = ANY(") - .push_bind(values.clone()) - .push("::text[]))"); + qb.push(" AND "); + push_exact_tag_match(qb, col_prefix, name, values); } } /// Match `#d` on artifact rows before `LIMIT` while leaving other kinds to the -/// caller's post-filter: `(kind NOT IN (artifact kinds) OR tags @> [["d", v]] ...)`. +/// caller's post-filter: `(kind NOT IN (artifact kinds) OR )`. fn push_artifact_d_tag_predicate( qb: &mut QueryBuilder, col_prefix: &str, @@ -933,12 +922,9 @@ fn push_artifact_d_tag_predicate( } let [revision, removal] = ARTIFACT_KINDS; qb.push(format!( - " AND ({col_prefix}kind NOT IN ({revision}, {removal})" + " AND ({col_prefix}kind NOT IN ({revision}, {removal}) OR " )); - for value in values { - qb.push(format!(" OR {col_prefix}tags @> ")); - qb.push_bind(serde_json::json!([["d", value]])); - } + push_exact_tag_match(qb, col_prefix, "d", values); qb.push(")"); } @@ -1111,7 +1097,8 @@ pub(crate) async fn count_events_on(conn: &mut sqlx::PgConnection, q: &EventQuer if let Some(ref e_tags) = q.e_tags { if !e_tags.is_empty() { - push_e_tag_filter(&mut qb, col_prefix, e_tags); + qb.push(" AND "); + push_exact_tag_match(&mut qb, col_prefix, "e", e_tags); } } @@ -1119,6 +1106,11 @@ pub(crate) async fn count_events_on(conn: &mut sqlx::PgConnection, q: &EventQuer push_artifact_d_tag_predicate(&mut qb, col_prefix, values); } + if let Some((ref name, ref value)) = q.custom_tag { + qb.push(" AND "); + push_exact_tag_match(&mut qb, col_prefix, name, std::slice::from_ref(value)); + } + push_tag_value_filters(&mut qb, col_prefix, &q.tag_values); if let Some(s) = q.since { @@ -2936,6 +2928,145 @@ mod postgres_tests { (inside, after) } + /// Every pushed-down tag filter matches by position: a newer decoy whose + /// tag holds the wanted name and value in the wrong slots must not take a + /// `limit: 1` page or be counted. + #[tokio::test] + #[ignore = "requires Postgres"] + async fn pushed_down_tag_filters_reject_positional_decoys() { + let pool = setup_pool().await; + let community = CommunityId::from_uuid(make_test_community(&pool).await); + let keys = Keys::generate(); + let now = nostr::Timestamp::now().as_secs(); + let target = "ab".repeat(32); + let insert = |kind: u16, tag: Vec<&str>, at: u64| { + let event = EventBuilder::new(Kind::Custom(kind), "") + .tags([Tag::parse(tag).unwrap()]) + .custom_created_at(nostr::Timestamp::from(at)) + .sign_with_keys(&keys) + .unwrap(); + let pool = pool.clone(); + async move { + insert_event(&pool, community, &event, None).await.unwrap(); + event.id.as_bytes().to_vec() + } + }; + // Removals: revisions also need an `artifact_heads` row to be read. + let [_, artifact_kind] = ARTIFACT_KINDS; + let artifact_kind = artifact_kind as u16; + + struct Case<'a> { + label: &'a str, + kind: u16, + real: Vec<&'a str>, + /// Inserted after `real`, so it wins a `limit: 1` read if matched. + decoy: Vec<&'a str>, + query: EventQuery, + } + let cases = [ + Case { + label: "custom tag value in name slot", + kind: 1, + real: vec!["t", "t"], + decoy: vec!["t", "other"], + query: { + let mut q = EventQuery::for_community(community); + q.custom_tag = Some(("t".into(), "t".into())); + q + }, + }, + Case { + label: "custom tag shifted right", + kind: 1, + real: vec!["t", "v"], + decoy: vec!["meta", "t", "v"], + query: { + let mut q = EventQuery::for_community(community); + q.custom_tag = Some(("t".into(), "v".into())); + q + }, + }, + Case { + label: "e tag shifted right", + kind: 7, + real: vec!["e", &target], + decoy: vec!["x", "e", &target], + query: { + let mut q = EventQuery::for_community(community); + q.e_tags = Some(vec![target.clone()]); + q + }, + }, + Case { + label: "artifact d tag shifted right", + kind: artifact_kind, + real: vec!["d", "a1"], + decoy: vec!["x", "d", "a1"], + query: { + let mut q = EventQuery::for_community(community); + q.d_tag_values = Some(vec!["a1".into()]); + q + }, + }, + ]; + for (offset, case) in (0u64..).step_by(10).zip(cases) { + let Case { + label, + kind, + real, + decoy, + query: mut q, + } = case; + let real_id = insert(kind, real, now - 1000 + offset).await; + insert(kind, decoy, now - 995 + offset).await; + q.kinds = Some(vec![i32::from(kind)]); + q.pubkey = Some(keys.public_key().to_bytes().to_vec()); + q.since = Some(DateTime::from_timestamp((now - 1000 + offset) as i64, 0).unwrap()); + q.until = Some(DateTime::from_timestamp((now - 995 + offset) as i64, 0).unwrap()); + q.limit = Some(1); + let rows = query_events(&pool, &q).await.unwrap(); + assert_eq!(rows.len(), 1, "{label}"); + assert_eq!(rows[0].event.id.as_bytes().to_vec(), real_id, "{label}"); + assert_eq!(count_events(&pool, &q).await.unwrap(), 1, "{label} count"); + } + } + + #[tokio::test] + #[ignore = "requires Postgres"] + async fn shared_gate_rejects_positional_decoy() { + let pool = setup_pool().await; + let community = CommunityId::from_uuid(make_test_community(&pool).await); + let (author, reader) = (Keys::generate(), Keys::generate()); + let kind = SHARED_GATED_KINDS[0]; + let now = nostr::Timestamp::now().as_secs(); + let mut ids = Vec::new(); + // Real shared event, then a newer private one whose shared marker is + // shifted right, so it would win a `limit: 1` read if admitted. + for (tag, at) in [ + (["shared", "true"].as_slice(), now - 10), + (&["x", "shared", "true"], now), + ] { + let event = EventBuilder::new(Kind::Custom(kind as u16), "") + .tags([ + Tag::parse(["d", &Uuid::new_v4().to_string()]).unwrap(), + Tag::parse(tag.iter().copied()).unwrap(), + ]) + .custom_created_at(nostr::Timestamp::from(at)) + .sign_with_keys(&author) + .unwrap(); + insert_event(&pool, community, &event, None).await.unwrap(); + ids.push(event.id.as_bytes().to_vec()); + } + let mut q = EventQuery::for_community(community); + q.kinds = Some(vec![kind as i32]); + q.pubkey = Some(author.public_key().to_bytes().to_vec()); + q.shared_gated_reader = Some(reader.public_key().to_bytes().to_vec()); + q.limit = Some(1); + let rows = query_events(&pool, &q).await.unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].event.id.as_bytes().to_vec(), ids[0]); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn e_tag_deadline_only_tightens_and_restores_timeout() { diff --git a/crates/buzz-relay/src/api/artifact_postgres_tests.rs b/crates/buzz-relay/src/api/artifact_postgres_tests.rs index b962d3f1c7..82d57dd2f1 100644 --- a/crates/buzz-relay/src/api/artifact_postgres_tests.rs +++ b/crates/buzz-relay/src/api/artifact_postgres_tests.rs @@ -144,25 +144,74 @@ impl Fixture { #[ignore = "requires Postgres"] async fn project_filter_survives_artifact_query_hook() { let f = Fixture::new().await; - let project = EventBuilder::new(Kind::Custom(30621), "") - .tags([ - Tag::parse(["d", "review-project"]).unwrap(), - Tag::parse(["buzz-channel", &f.home.to_string()]).unwrap(), - ]) - .sign_with_keys(&f.owner) - .unwrap(); - f.state - .db - .insert_event(f.community, &project, None) - .await - .unwrap(); - let filter = json!([{"kinds":[30621],"#buzz-channel":[f.home]}]); - let (status, body) = f.request("/query", filter.clone()).await; - assert!(status.is_success(), "{status}: {body}"); - assert_eq!(body.as_array().unwrap().len(), 1); - assert_eq!(body[0]["id"], project.id.to_hex()); - let (status, body) = f.request("/count", filter).await; - assert!(status.is_success(), "{status}: {body}"); + let home = f.home.to_string(); + // 30621 projects and 30617 repositories bind a channel via `buzz-channel`. + for kind in [30621u16, 30617] { + let event = |tag: &[&str], at: u64| { + EventBuilder::new(Kind::Custom(kind), "") + .tags([ + Tag::parse(["d", &Uuid::new_v4().to_string()]).unwrap(), + Tag::parse(tag.iter().copied()).unwrap(), + ]) + .custom_created_at(nostr::Timestamp::from(at)) + .sign_with_keys(&f.owner) + .unwrap() + }; + let now = nostr::Timestamp::now().as_secs(); + let bound = event(&["buzz-channel", &home], now - 10); + // Newer, unbound: `buzz-channel` and the wanted id sit in an unrelated tag. + let decoy = event(&["x", "buzz-channel", &home], now); + for e in [&bound, &decoy] { + f.state.db.insert_event(f.community, e, None).await.unwrap(); + } + let filter = json!([{"kinds":[kind],"#buzz-channel":[f.home],"limit":1}]); + let (status, body) = f.request("/query", filter.clone()).await; + assert!(status.is_success(), "{kind} {status}: {body}"); + assert_eq!(body.as_array().unwrap().len(), 1, "{kind}: {body}"); + assert_eq!(body[0]["id"], bound.id.to_hex(), "{kind}: {body}"); + let (status, body) = f.request("/count", filter).await; + assert!(status.is_success(), "{kind} {status}: {body}"); + assert_eq!(body["count"], 1, "{kind}: {body}"); + } +} + +#[tokio::test] +#[ignore = "requires Postgres"] +async fn count_honors_buzz_channel_filter() { + let f = Fixture::new().await; + // Two projects bound to `home`, one bound to `private`, all stored in + // `home` so the explicit `#h` scope alone would count all three. + for channel in [f.home, f.home, f.private] { + let event = EventBuilder::new(Kind::Custom(30621), "") + .tags([ + Tag::parse(["d", &Uuid::new_v4().to_string()]).unwrap(), + Tag::parse(["buzz-channel", &channel.to_string()]).unwrap(), + Tag::parse(["t", "proj"]).unwrap(), + ]) + .sign_with_keys(&f.owner) + .unwrap(); + f.state + .db + .insert_event(f.community, &event, Some(f.home)) + .await + .unwrap(); + } + for filter in [ + // Fast COUNT over the reader's accessible channels. + json!([{"kinds":[30621],"#buzz-channel":[f.home]}]), + // Fast COUNT on the explicit-channel path. + json!([{"kinds":[30621],"#h":[f.home],"#buzz-channel":[f.home]}]), + // `#t` is not fully pushable, forcing the per-event fallback on both paths. + json!([{"kinds":[30621],"#t":["proj"],"#buzz-channel":[f.home]}]), + json!([{"kinds":[30621],"#h":[f.home],"#t":["proj"],"#buzz-channel":[f.home]}]), + ] { + let (status, body) = f.request("/query", filter.clone()).await; + assert!(status.is_success(), "{filter} {status}: {body}"); + assert_eq!(body.as_array().unwrap().len(), 2, "{filter}: {body}"); + let (status, body) = f.request("/count", filter.clone()).await; + assert!(status.is_success(), "{filter} {status}: {body}"); + assert_eq!(body["count"], 2, "{filter}: {body}"); + } } #[tokio::test] diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 8f341341c1..64efc336de 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -2045,7 +2045,7 @@ async fn count_events_authed( .await?; let mut total: u64 = 0; - for filter in &filters { + for (raw, filter) in raw.iter().zip(&filters) { let needs_author_only_filtering = crate::handlers::req::filter_can_match_author_only_kinds(filter); // Same result-gated guard as the WS COUNT handler: force the per-event @@ -2092,6 +2092,9 @@ async fn count_events_authed( ch_id, &accessible_channels, ); + if let Some(channel) = extract_buzz_channel(raw) { + query.custom_tag = Some(("buzz-channel".into(), channel.into())); + } // Shared-gated visibility pushdown: same as REQ and /query paths, so // the fallback's query_events call doesn't over-fetch private rows. if needs_shared_gate_filtering { @@ -2161,6 +2164,9 @@ async fn count_events_authed( ) .await; query.channel_ids = Some(accessible_channels.to_vec()); + if let Some(channel) = extract_buzz_channel(raw) { + query.custom_tag = Some(("buzz-channel".into(), channel.into())); + } // Shared-gated visibility pushdown: pre-filter before ORDER/LIMIT on // the fallback query_events path. if needs_shared_gate_filtering { diff --git a/crates/buzz-relay/src/handlers/req.rs b/crates/buzz-relay/src/handlers/req.rs index df9fecd761..656cb23b0f 100644 --- a/crates/buzz-relay/src/handlers/req.rs +++ b/crates/buzz-relay/src/handlers/req.rs @@ -1097,7 +1097,7 @@ fn filter_to_query_params( } }); - // Push #e tag filter into SQL via JSONB containment. + // Push #e tag filter into SQL as an exact positional tag match. let e_tag_key = nostr::SingleLetterTag::lowercase(nostr::Alphabet::E); let e_tags = filter.generic_tags.get(&e_tag_key).and_then(|values| { if values.is_empty() {