Enforce author-only event visibility in SQL

Push the authenticated reader into event query and count predicates before
pagination, including mixed-kind and kindless filters. Add the brownfield FTS
exclusion for private managed-agent aggregates without changing prior migration
checksums.

Co-authored-by: Carl <c7ebe626f000404285d3686e1dc74cc07cc60a9754a150041ba132e14bd3e2ec@buzz.block.builderlab.xyz>
Signed-off-by: Wes <wesbillman@users.noreply.github.com>

crates/buzz-db/src/event.rs

@@ -10,8 +10,8 @@ use sqlx::{PgPool, Postgres, QueryBuilder, Row, Transaction};
 use uuid::Uuid;
 
 use buzz_core::kind::{
-    event_kind_i32, is_ephemeral, is_parameterized_replaceable, KIND_AUTH, KIND_EVENT_REMINDER,
-    KIND_HUDDLE_STARTED, SHARED_GATED_KINDS,
+    event_kind_i32, is_ephemeral, is_parameterized_replaceable, AUTHOR_ONLY_KINDS, KIND_AUTH,
+    KIND_EVENT_REMINDER, KIND_HUDDLE_STARTED, SHARED_GATED_KINDS,
 };
 use buzz_core::{CommunityId, StoredEvent};
 
@@ -78,6 +78,11 @@ pub struct EventQuery {
     /// the COUNT fallback path, which needs to fetch all matching events for
     /// post-filter counting. When None, the default clamp applies.
     pub max_limit: Option<i64>,
+    /// Author-only visibility reader: when set, append an SQL visibility clause
+    /// for every kind in [`buzz_core::kind::AUTHOR_ONLY_KINDS`] before
+    /// ORDER/LIMIT. This prevents newer foreign private rows from starving the
+    /// authenticated owner's visible rows off a bounded page.
+    pub author_only_reader: Option<Vec<u8>>,
     /// Shared-gated visibility reader: when set, append an SQL visibility
     /// clause for every kind in [`SHARED_GATED_KINDS`] before ORDER/LIMIT so
     /// private events are excluded from the candidate page rather than
@@ -123,6 +128,7 @@ impl EventQuery {
             e_tags: None,
             channel_ids: None,
             max_limit: None,
+            author_only_reader: None,
             shared_gated_reader: None,
         }
     }
@@ -521,6 +527,21 @@ pub(crate) async fn query_events_on(
         }
     }
 
+    // Author-only visibility pushdown: exclude private rows not authored by
+    // the authenticated reader before ORDER/LIMIT. Post-filtering alone is not
+    // sufficient: a foreign owner can otherwise fill the candidate page and
+    // starve the reader's older rows.
+    if let Some(ref reader_bytes) = q.author_only_reader {
+        qb.push(format!(" AND ({col_prefix}kind NOT IN ("));
+        let mut sep = qb.separated(", ");
+        for kind in AUTHOR_ONLY_KINDS {
+            sep.push_bind(*kind as i32);
+        }
+        qb.push(format!(") OR {col_prefix}pubkey = "));
+        qb.push_bind(reader_bytes.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
@@ -761,6 +782,38 @@ pub(crate) async fn count_events_on(conn: &mut sqlx::PgConnection, q: &EventQuer
         }
     }
 
+    // Apply the same author-only SQL visibility gate used by query_events.
+    // COUNT must not leak foreign event existence, including for mixed-kind
+    // filters and kindless id lookups.
+    if let Some(ref reader_bytes) = q.author_only_reader {
+        qb.push(format!(" AND ({col_prefix}kind NOT IN ("));
+        let mut sep = qb.separated(", ");
+        for kind in AUTHOR_ONLY_KINDS {
+            sep.push_bind(*kind as i32);
+        }
+        qb.push(format!(") OR {col_prefix}pubkey = "));
+        qb.push_bind(reader_bytes.clone());
+        qb.push(")");
+    }
+
+    // Shared-gated visibility belongs in SQL for COUNT as well. Although relay
+    // callers currently use the per-event fallback for these kinds, keeping the
+    // query primitive complete prevents a future direct caller from leaking
+    // foreign unshared rows.
+    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(")");
+    }
+
     let row = qb.build().fetch_one(&mut *conn).await?;
     let cnt: i64 = row.try_get("cnt")?;
 

crates/buzz-db/src/migration.rs

@@ -561,7 +561,7 @@ mod tests {
         let mut migrations: Vec<_> = MIGRATOR.iter().collect();
         migrations.sort_by_key(|migration| migration.version);
 
-        assert_eq!(migrations.len(), 27);
+        assert_eq!(migrations.len(), 28);
         assert_eq!(migrations[0].version, 1);
         assert_eq!(&*migrations[0].description, "initial schema");
         assert!(migrations[0]
@@ -940,6 +940,16 @@ mod tests {
             desired_schema.contains("idx_channels_id_live"),
             "desired-state schema must carry the channel-id lookup index",
         );
+
+        // Private managed-agent ciphertext is author-only and must remain
+        // unsearchable on brownfield databases without changing 0001's sqlx
+        // checksum. The additive wrapper preserves the existing expression.
+        assert_eq!(migrations[27].version, 28);
+        let private_managed_agent = migrations[27].sql.as_str();
+        assert!(private_managed_agent.contains("kind = 30179"));
+        assert!(private_managed_agent.contains("pg_get_expr"));
+        assert!(private_managed_agent.contains("search_tsv"));
+        assert!(!migrations[0].sql.as_str().contains("30179"));
     }
 
     #[test]

crates/buzz-relay/src/handlers/req.rs

@@ -289,6 +289,12 @@ pub async fn handle_req(
             let mut params =
                 filter_to_query_params(filter, per_filter_channel, conn.tenant.community());
             apply_access_scope_to_query(&mut params, per_filter_channel, &accessible_channels);
+            // Private visibility pushdown must happen before ORDER/LIMIT. This
+            // applies even to mixed-kind and kindless filters; per-event checks
+            // remain defense-in-depth for id lookups and future call paths.
+            if filter_can_match_author_only_kinds(filter) {
+                params.author_only_reader = Some(pubkey_bytes.clone());
+            }
             // Shared-gated visibility pushdown: set reader bytes so query_events
             // appends the SQL visibility clause before ORDER/LIMIT, preventing
             // newer private events from starving older shared ones off the page.
@@ -752,12 +758,16 @@ async fn handle_search_req(
 /// Resolves accessible channels for the given pubkey and builds the query.
 pub async fn build_event_query_from_filter(
     filter: &Filter,
-    _pubkey_bytes: &[u8],
+    pubkey_bytes: &[u8],
     _state: &AppState,
     community: buzz_core::tenant::CommunityId,
 ) -> EventQuery {
     let channel_id = extract_channel_id_from_filter(filter);
-    filter_to_query_params(filter, channel_id, community)
+    let mut query = filter_to_query_params(filter, channel_id, community);
+    if filter_can_match_author_only_kinds(filter) {
+        query.author_only_reader = Some(pubkey_bytes.to_vec());
+    }
+    query
 }
 
 /// Maximum SQL candidate rows a non-pushable COUNT filter may inspect before

migrations/0028_private_managed_agent_foundation.sql

@@ -0,0 +1,27 @@
+-- NIP-PMA kind:30179 contains owner-encrypted runnable agent configuration and
+-- is author-only. Preserve every installation's current FTS expression while
+-- making existing and future private aggregate rows storage-level unsearchable.
+DO $$
+DECLARE
+    existing_expression TEXT;
+BEGIN
+    SELECT pg_get_expr(d.adbin, d.adrelid)
+      INTO existing_expression
+      FROM pg_attrdef d
+      JOIN pg_attribute a
+        ON a.attrelid = d.adrelid
+       AND a.attnum = d.adnum
+     WHERE d.adrelid = 'events'::regclass
+       AND a.attname = 'search_tsv';
+
+    IF existing_expression IS NULL THEN
+        RAISE EXCEPTION 'events.search_tsv generated expression not found';
+    END IF;
+
+    ALTER TABLE events DROP COLUMN search_tsv;
+    EXECUTE format(
+        'ALTER TABLE events ADD COLUMN search_tsv TSVECTOR GENERATED ALWAYS AS (CASE WHEN kind = 30179 THEN NULL::tsvector ELSE (%s) END) STORED',
+        existing_expression
+    );
+    CREATE INDEX idx_events_search_tsv ON events USING GIN (search_tsv);
+END $$;