diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 9dd4793d0db..fdb94bf5a20 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -5,6 +5,7 @@ //! Deduplication is application-layer: ON CONFLICT DO NOTHING. use chrono::{DateTime, Utc}; +use futures_util::TryStreamExt; use nostr::Event; use sqlx::{PgPool, QueryBuilder, Row}; use uuid::Uuid; @@ -238,7 +239,10 @@ fn huddle_started_content_links(content: &str, ephemeral_channel_id: Uuid) -> bo /// The creator constraint matters: a member of some unrelated channel can post /// their own kind:48100 event there, but they cannot sign as the creator of the /// target ephemeral channel. One set-based query replaces the liveness -/// endpoint's former session × parent lookup loop. Malformed historical start +/// endpoint's former session × parent lookup loop. Like the per-link checks, +/// SQL only narrows candidates to start events that mention the session ID, +/// and [`huddle_started_content_links`] decides: Postgres 15 cannot test +/// whether text is valid JSON without raising. Malformed historical start /// content is ignored rather than aborting the complete liveness snapshot. pub async fn huddle_started_links( pool: &PgPool, @@ -254,20 +258,16 @@ pub async fn huddle_started_links( crate::observability::WriterOperation::SubscriptionHistory, ) .await?; - let rows = sqlx::query( + let mut rows = sqlx::query( r#" - SELECT DISTINCT ON (backing.id) - backing.id AS session_id, + SELECT backing.id AS session_id, start.channel_id AS parent_channel_id, - backing.created_by + backing.created_by, + start.content FROM events start JOIN channels backing ON backing.community_id = start.community_id - AND backing.id::text = CASE - WHEN start.content IS JSON OBJECT - THEN (start.content::json ->> 'ephemeral_channel_id') - ELSE NULL - END + AND start.content ILIKE '%' || backing.id::text || '%' AND backing.deleted_at IS NULL WHERE start.deleted_at IS NULL AND start.community_id = $1 @@ -284,18 +284,29 @@ pub async fn huddle_started_links( .bind(KIND_HUDDLE_STARTED as i32) .bind(ephemeral_channel_ids) .bind(HUDDLE_LINK_CONTENT_MAX_BYTES) - .fetch_all(&mut *connection) - .await?; - - rows.into_iter() - .map(|row| { - Ok(( - row.try_get("session_id")?, + .fetch(&mut *connection); + + // Rows arrive newest first per session; keep each session's newest link. + // Streamed, so memory holds the links rather than every candidate start. + let mut links: Vec<(Uuid, Uuid, Vec)> = Vec::new(); + while let Some(row) = rows.try_next().await? { + let session_id: Uuid = row.try_get("session_id")?; + if links + .last() + .is_some_and(|(linked, _, _)| *linked == session_id) + { + continue; + } + let content: String = row.try_get("content")?; + if huddle_started_content_links(&content, session_id) { + links.push(( + session_id, row.try_get("parent_channel_id")?, row.try_get("created_by")?, - )) - }) - .collect() + )); + } + } + Ok(links) } /// Return whether a creator-signed huddle-start event links a parent channel @@ -3636,9 +3647,12 @@ mod postgres_tests { let session = make_test_channel(&pool, community_uuid, Some(60)).await; let creator = vec![7_u8; 32]; + // The newest start mentions the session but is not JSON, and must not + // hide the older valid link. for (index, content) in [ "not-json".to_owned(), serde_json::json!({ "ephemeral_channel_id": session }).to_string(), + format!("{{\"ephemeral_channel_id\":\"{session}\""), ] .into_iter() .enumerate()