fix(db): make the huddle liveness link query run on Postgres 15

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<u8>)> = 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()