From 593f085e3c052d8a3832279fe4755bc7bc0a1128 Mon Sep 17 00:00:00 2001 From: Tee8z Date: Fri, 23 Oct 2026 10:52:23 -0400 Subject: [PATCH 1/2] fix(db): make the huddle liveness link query run on Postgres 15 huddle_started_links tested start content with `IS JSON OBJECT`, which needs Postgres 16. Prod runs 15, so every liveness batch failed with "syntax error at or near JSON". SQL now only narrows candidates to start events that mention the session ID, and huddle_started_content_links decides, as the per-link checks already do. Each session keeps its newest valid link, so a newer malformed start cannot hide an older valid one. --- crates/buzz-db/src/store/event.rs | 45 ++++++++++++++++++++----------- 1 file changed, 29 insertions(+), 16 deletions(-) diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 9dd4793d0db..dd222e8fb7c 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -238,7 +238,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, @@ -256,18 +259,14 @@ pub async fn huddle_started_links( .await?; let 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 @@ -287,15 +286,26 @@ pub async fn huddle_started_links( .fetch_all(&mut *connection) .await?; - rows.into_iter() - .map(|row| { - Ok(( - row.try_get("session_id")?, + // Rows arrive newest first per session; keep each session's newest link. + let mut links: Vec<(Uuid, Uuid, Vec)> = Vec::new(); + for row in rows { + 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 +3646,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() From 130ce1c8480c208b3c5531dbeb305ee38ec371b6 Mon Sep 17 00:00:00 2001 From: Tee8z Date: Fri, 23 Oct 2026 11:10:01 -0400 Subject: [PATCH 2/2] fix(db): stream huddle link candidates instead of loading them all --- crates/buzz-db/src/store/event.rs | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index dd222e8fb7c..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; @@ -257,7 +258,7 @@ pub async fn huddle_started_links( crate::observability::WriterOperation::SubscriptionHistory, ) .await?; - let rows = sqlx::query( + let mut rows = sqlx::query( r#" SELECT backing.id AS session_id, start.channel_id AS parent_channel_id, @@ -283,12 +284,12 @@ 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?; + .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(); - for row in rows { + while let Some(row) = rows.try_next().await? { let session_id: Uuid = row.try_get("session_id")?; if links .last()