buzz/crates/buzz-db/src/store/thread.rs

//! Thread metadata persistence.
//!
//! Tracks parent/root relationships, depth, and reply counts for infinitely
//! nested threads. The `thread_metadata` table is populated when events are
//! ingested and updated as replies arrive or are deleted.

use buzz_core::StoredEvent;
use chrono::{DateTime, Utc};
use sqlx::{PgPool, Row};
use uuid::Uuid;

use crate::store::recent_pass::Range;

use buzz_datastore_tracing::datastore_span;

use buzz_core::CommunityId;

use crate::{
    error::Result, event::row_to_stored_event, route_proof::ChannelScoped, Db, ReadSession,
    ReadSessionInner, RouteDecision, RoutePredicate,
};

/// [`THREAD_PARTICIPANT_SCAN_LIMIT`] as a literal, so SQL can `concat!` it.
macro_rules! thread_participant_scan_limit {
    () => {
        1000
    };
}

/// How many of a thread's newest replies a participant summary reads.
///
/// Participants are the thread's most recently active distinct authors, so
/// the newest replies decide them. Aggregating the whole thread instead made
/// every summary cost O(thread), recomputed after every reply, so one very
/// large thread could saturate the writer. An author silent for this many
/// replies drops out of the facepile, which shows at most 10.
pub const THREAD_PARTICIPANT_SCAN_LIMIT: i64 = thread_participant_scan_limit!();

/// Correlated `ARRAY(...)` of up to 10 participant pubkeys, most recent
/// first, for the thread rooted at `root.event_id` in community `$1`.
///
/// Walks `idx_thread_metadata_window` newest-first and stops after
/// [`THREAD_PARTICIPANT_SCAN_LIMIT`] rows, so cost is bounded by the window,
/// not the thread. The one argument is an extra (possibly empty) predicate on
/// the window's `tm` rows. A macro so callers stay `&'static str` via
/// `concat!`.
macro_rules! recent_thread_participants_sql {
    ($extra:literal) => {
        concat!(
            r#"ARRAY(
                SELECT e.pubkey
                FROM (
                    SELECT tm.event_created_at, tm.event_id
                    FROM thread_metadata tm
                    WHERE tm.community_id = $1
                      AND tm.root_event_id = root.event_id "#,
            $extra,
            r#"
                    ORDER BY tm.event_created_at DESC, tm.event_id ASC
                    LIMIT "#,
            thread_participant_scan_limit!(),
            r#"
                ) recent
                JOIN events e
                    ON e.community_id = $1
                   AND e.created_at   = recent.event_created_at
                   AND e.id           = recent.event_id
                WHERE e.deleted_at IS NULL
                GROUP BY e.pubkey
                ORDER BY MAX(e.created_at) DESC, e.pubkey ASC
                LIMIT 10
            )"#
        )
    };
}

// -- Structs ------------------------------------------------------------------

/// A single reply within a thread, joined with event content.
#[derive(Debug, Clone)]
pub struct ThreadReply {
    /// The Nostr event ID of this reply.
    pub event_id: Vec<u8>,
    /// The event ID of the direct parent (one level up), if any.
    pub parent_event_id: Option<Vec<u8>>,
    /// The event ID of the thread root (top-level message), if any.
    pub root_event_id: Option<Vec<u8>>,
    /// The channel this reply belongs to.
    pub channel_id: Uuid,
    /// Compressed public key of the reply author.
    pub pubkey: Vec<u8>,
    /// Nostr event tags (JSON array), used to extract effective author.
    pub tags: serde_json::Value,
    /// Text content of the reply.
    pub content: String,
    /// Fully reconstructed event row for this reply.
    pub stored_event: StoredEvent,
    /// Nesting depth within the thread (root = 0, direct reply = 1, etc.).
    pub depth: i32,
    /// When the reply was created.
    pub created_at: DateTime<Utc>,
    /// Whether this reply is also broadcast to the channel timeline.
    pub broadcast: bool,
}

/// Aggregated thread statistics for a root message.
#[derive(Debug, Clone)]
pub struct ThreadSummary {
    /// Number of direct replies to the root message.
    pub reply_count: i32,
    /// Total number of replies at all nesting levels.
    pub descendant_count: i32,
    /// Timestamp of the most recent reply in the thread.
    pub last_reply_at: Option<DateTime<Utc>>,
    /// Up to 10 distinct reply authors, most recently active first, drawn
    /// from the newest [`THREAD_PARTICIPANT_SCAN_LIMIT`] replies. Deleted
    /// replies count toward that window but contribute no author.
    pub participants: Vec<Vec<u8>>,
}

/// One row of a channel window: the reconstructed signed event plus its
/// thread summary (populated when the row has thread activity).
#[derive(Debug, Clone)]
pub struct ChannelWindowRow {
    /// Fully reconstructed signed event for this row.
    pub stored_event: StoredEvent,
    /// Thread statistics for this row; `None` when it has no replies.
    pub thread_summary: Option<ThreadSummary>,
}

/// A page of top-level channel rows plus the server-side exhaustion fact.
#[derive(Debug, Clone)]
pub struct ChannelWindow {
    /// Retained rows in `(created_at DESC, id ASC)` order — at most `limit`.
    pub rows: Vec<ChannelWindowRow>,
    /// Whether more rows exist past the last retained row. Computed from an
    /// internal `limit + 1` probe; the sentinel row never leaves this module.
    pub has_more: bool,
    /// Composite keyset cursor `(created_at, id)` of the last retained row —
    /// the scan position, captured before event reconstruction so a page whose
    /// tail row fails to reconstruct still advances. `Some` iff `has_more`.
    pub next_cursor: Option<(DateTime<Utc>, Vec<u8>)>,
}

/// Raw thread_metadata row -- used when processing deletes or computing ancestry.
#[derive(Debug, Clone)]
pub struct ThreadMetadataRecord {
    /// The Nostr event ID this metadata row tracks.
    pub event_id: Vec<u8>,
    /// Partition key timestamp for the event.
    pub event_created_at: DateTime<Utc>,
    /// The channel this event belongs to.
    pub channel_id: Uuid,
    /// Event ID of the direct parent, if this is a reply.
    pub parent_event_id: Option<Vec<u8>>,
    /// Event ID of the thread root, if this is a nested reply.
    pub root_event_id: Option<Vec<u8>>,
    /// Nesting depth (root = 0).
    pub depth: i32,
    /// Number of direct replies to this event.
    pub reply_count: i32,
    /// Total number of descendants at all nesting levels.
    pub descendant_count: i32,
    /// Whether this event is broadcast to the channel timeline.
    pub broadcast: bool,
}

// -- Write operations ---------------------------------------------------------

/// Insert a row into `thread_metadata`.
///
/// If `parent_event_id` is `Some`, also increments the parent's reply count
/// and the root's descendant count (always, including when root == parent).
///
/// The INSERT and all counter UPDATEs are wrapped in a single transaction so a
/// crash between them cannot leave reply_count / descendant_count inconsistent
/// with the actual number of reply rows (F9).
#[allow(clippy::too_many_arguments)]
pub async fn insert_thread_metadata(
    pool: &PgPool,
    community_id: CommunityId,
    event_id: &[u8],
    event_created_at: DateTime<Utc>,
    channel_id: Uuid,
    parent_event_id: Option<&[u8]>,
    parent_event_created_at: Option<DateTime<Utc>>,
    root_event_id: Option<&[u8]>,
    root_event_created_at: Option<DateTime<Utc>>,
    depth: i32,
    broadcast: bool,
) -> Result<()> {
    let mut tx = crate::begin_community_write_transaction(
        pool,
        community_id,
        crate::observability::WriterOperation::EventWrite,
    )
    .await?;

    let result = sqlx::query(
        r#"
        INSERT INTO thread_metadata
            (community_id, event_created_at, event_id, channel_id,
             parent_event_id, parent_event_created_at,
             root_event_id, root_event_created_at,
             depth, broadcast)
        VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
        ON CONFLICT DO NOTHING
        "#,
    )
    .bind(community_id.as_uuid())
    .bind(event_created_at)
    .bind(event_id)
    .bind(channel_id)
    .bind(parent_event_id)
    .bind(parent_event_created_at)
    .bind(root_event_id)
    .bind(root_event_created_at)
    .bind(depth)
    .bind(broadcast)
    .execute(tx.conn())
    .await?;

    // Only bump reply counts if the row was actually inserted (not a duplicate).
    // ON CONFLICT DO NOTHING on a duplicate key returns rows_affected = 0.
    if result.rows_affected() > 0 {
        if let Some(pid) = parent_event_id {
            // Ensure the parent has a thread_metadata row so the UPDATE below
            // has something to hit. Root (depth=0) messages don't get a row on
            // first insert, so we create a stub here.
            let parent_ts = parent_event_created_at.unwrap_or(event_created_at);
            sqlx::query(
                r#"
                INSERT INTO thread_metadata
                    (community_id, event_created_at, event_id, channel_id,
                     parent_event_id, parent_event_created_at,
                     root_event_id, root_event_created_at,
                     depth, broadcast)
                VALUES ($1, $2, $3, $4, NULL, NULL, NULL, NULL, 0, false)
                ON CONFLICT DO NOTHING
                "#,
            )
            .bind(community_id.as_uuid())
            .bind(parent_ts)
            .bind(pid)
            .bind(channel_id)
            .execute(tx.conn())
            .await?;

            // Ensure the root also has a row (may differ from parent for nested replies).
            if let Some(root_id) = root_event_id {
                if root_id != pid {
                    let root_ts = root_event_created_at.unwrap_or(event_created_at);
                    sqlx::query(
                        r#"
                        INSERT INTO thread_metadata
                            (community_id, event_created_at, event_id, channel_id,
                             parent_event_id, parent_event_created_at,
                             root_event_id, root_event_created_at,
                             depth, broadcast)
                        VALUES ($1, $2, $3, $4, NULL, NULL, NULL, NULL, 0, false)
                        ON CONFLICT DO NOTHING
                        "#,
                    )
                    .bind(community_id.as_uuid())
                    .bind(root_ts)
                    .bind(root_id)
                    .bind(channel_id)
                    .execute(tx.conn())
                    .await?;
                }
            }

            // Increment parent's direct reply count and last_reply_at.
            sqlx::query(
                r#"
                UPDATE thread_metadata
                SET reply_count   = reply_count + 1,
                    last_reply_at = NOW()
                WHERE community_id = $1 AND event_id = $2
                "#,
            )
            .bind(community_id.as_uuid())
            .bind(pid)
            .execute(tx.conn())
            .await?;

            // Increment root's total descendant count.
            if let Some(root_id) = root_event_id {
                sqlx::query(
                    r#"
                    UPDATE thread_metadata
                    SET descendant_count = descendant_count + 1
                    WHERE community_id = $1 AND event_id = $2
                    "#,
                )
                .bind(community_id.as_uuid())
                .bind(root_id)
                .execute(tx.conn())
                .await?;
            }
        }
    }

    tx.commit().await?;

    Ok(())
}

// -- Read operations ----------------------------------------------------------

/// Fetch all replies under a root event, ordered chronologically.
///
/// - `depth_limit` -- if `Some(n)`, only returns replies at depth <= n.
/// - `cursor` -- keyset pagination cursor. The result order is
///   `(event_created_at ASC, event_id ASC)`; the cursor is the composite key of
///   the last row already seen and the query returns rows strictly after it. A
///   composite `(timestamp, event_id)` tiebreak is required because thread
///   replies routinely share a `created_at` second (bursty threads); a
///   timestamp-only cursor silently drops every tied reply past the page limit.
///   Wire encoding: `8-byte big-endian i64 seconds` followed by the raw
///   `event_id` bytes (32 for a standard Nostr id). A bare 8-byte cursor is
///   still accepted for back-compat and paginates on timestamp alone (unsafe
///   across same-second ties) -- prefer the composite form.
/// - `limit` -- maximum rows returned (caller should cap this).
pub async fn get_thread_replies(
    pool: &PgPool,
    community_id: CommunityId,
    root_event_id: &[u8],
    depth_limit: Option<u32>,
    limit: u32,
    cursor: Option<&[u8]>,
) -> Result<Vec<ThreadReply>> {
    let mut conn = crate::observability::acquire_writer(
        pool,
        crate::observability::WriterOperation::SubscriptionHistory,
    )
    .await?;
    get_thread_replies_on(
        &mut conn,
        community_id,
        root_event_id,
        depth_limit,
        limit,
        cursor,
    )
    .await
}

/// [`get_thread_replies`] on a specific session — the replica-routing path
/// runs the page on the exact reader connection whose heartbeat observation
/// proved coverage (the proof is connection-local; a different pooled
/// session may sit at a different replay position).
pub(crate) async fn get_thread_replies_on(
    conn: &mut sqlx::PgConnection,
    community_id: CommunityId,
    root_event_id: &[u8],
    depth_limit: Option<u32>,
    limit: u32,
    cursor: Option<&[u8]>,
) -> Result<Vec<ThreadReply>> {
    // Decode cursor bytes -> keyset (timestamp, optional event_id) for the
    // WHERE condition. Layout: 8-byte BE i64 seconds, then the raw event_id.
    // An 8-byte-only cursor is legacy timestamp-only paging (no tiebreak).
    let mut cursor_key: Option<(DateTime<Utc>, Option<Vec<u8>>)> = match cursor {
        Some(bytes) if bytes.len() >= 8 => {
            let secs = i64::from_be_bytes(bytes[..8].try_into().expect("length checked"));
            DateTime::from_timestamp(secs, 0).map(|ts| {
                let id = if bytes.len() > 8 {
                    Some(bytes[8..].to_vec())
                } else {
                    None
                };
                (ts, id)
            })
        }
        _ => None,
    };

    let mut replies = Vec::new();
    // Damaged rows must not cause unbounded work on a held writer connection.
    // Fail explicitly rather than exposing a partial page as authoritative EOF.
    const MAX_RAW_PAGES: usize = 64;
    for _ in 0..MAX_RAW_PAGES {
        let remaining = limit.saturating_sub(replies.len() as u32);
        if remaining == 0 {
            return Ok(replies);
        }
        // Build the query dynamically based on optional filters.
        // Track the next positional parameter index.
        let mut param_idx = 3u32; // $1 is community_id, $2 is root_event_id
        let mut sql = String::from(
            r#"
        SELECT
            tm.event_id,
            e.id,
            tm.parent_event_id,
            tm.root_event_id,
            tm.channel_id,
            e.pubkey,
            e.created_at AS created_at,
            e.tags,
            e.content,
            e.kind,
            e.sig,
            e.received_at,
            tm.depth,
            tm.event_created_at,
            tm.broadcast
        FROM thread_metadata tm
        JOIN events e
            ON e.community_id = tm.community_id
           AND e.created_at = tm.event_created_at
           AND e.id         = tm.event_id
        WHERE tm.community_id = $1
          AND tm.root_event_id = $2
          AND e.deleted_at IS NULL
        "#,
        );

        if depth_limit.is_some() {
            sql.push_str(&format!(" AND tm.depth <= ${param_idx}"));
            param_idx += 1;
        }
        match &cursor_key {
            Some((_, Some(_))) => {
                // Composite keyset: strict row comparison with an event_id tiebreak
                // so same-second replies paginate without gaps or duplicates.
                let ts_idx = param_idx;
                let id_idx = param_idx + 1;
                sql.push_str(&format!(
                    " AND (tm.event_created_at, tm.event_id) > (${ts_idx}, ${id_idx})"
                ));
                param_idx += 2;
            }
            Some((_, None)) => {
                // Legacy timestamp-only cursor (no tiebreak).
                sql.push_str(&format!(" AND tm.event_created_at > ${param_idx}"));
                param_idx += 1;
            }
            None => {}
        }

        sql.push_str(&format!(
            " ORDER BY tm.event_created_at ASC, tm.event_id ASC LIMIT ${param_idx}"
        ));

        let mut q = sqlx::query(sqlx::AssertSqlSafe(sql))
            .bind(community_id.as_uuid())
            .bind(root_event_id);

        if let Some(dl) = depth_limit {
            q = q.bind(dl as i32);
        }
        match &cursor_key {
            Some((ts, Some(id))) => {
                q = q.bind(*ts).bind(id.clone());
            }
            Some((ts, None)) => {
                q = q.bind(*ts);
            }
            None => {}
        }
        q = q.bind(remaining as i32);

        let rows = q.fetch_all(&mut *conn).await?;

        let raw_count = rows.len();
        if let Some(last) = rows.last() {
            cursor_key = Some((
                last.try_get("event_created_at")?,
                Some(last.try_get("event_id")?),
            ));
        }
        for row in rows {
            let event_id: Vec<u8> = row.try_get("event_id")?;
            let parent_event_id: Option<Vec<u8>> = row.try_get("parent_event_id")?;
            let root_event_id_col: Option<Vec<u8>> = row.try_get("root_event_id")?;
            let channel_id: Uuid = row.try_get("channel_id")?;
            let pubkey: Vec<u8> = row.try_get("pubkey")?;
            let tags: serde_json::Value = row.try_get("tags")?;
            let depth: i32 = row.try_get("depth")?;
            let created_at: DateTime<Utc> = row.try_get("event_created_at")?;
            let broadcast_val: bool = row.try_get("broadcast")?;

            // Skip rows that fail event reconstruction (e.g. corrupt signature)
            // while continuing past the raw page boundary to fill the requested
            // page. A short response must mean SQL EOF, not a skipped stored row.
            let stored_event = match row_to_stored_event(row)? {
                Some(se) => se,
                None => continue,
            };

            replies.push(ThreadReply {
                event_id,
                parent_event_id,
                root_event_id: root_event_id_col,
                channel_id,
                pubkey,
                tags,
                content: stored_event.event.content.clone(),
                stored_event,
                depth,
                created_at,
                broadcast: broadcast_val,
            });
        }

        if raw_count < remaining as usize || replies.len() == limit as usize {
            return Ok(replies);
        }
    }
    Err(crate::error::DbError::InvalidData(
        "thread page exceeded the corrupt-row refill limit".into(),
    ))
}

/// Fetch aggregated thread stats for a single event, plus up to 10 participant pubkeys.
pub async fn get_thread_summary(
    pool: &PgPool,
    community_id: CommunityId,
    event_id: &[u8],
) -> Result<Option<ThreadSummary>> {
    let mut connection = crate::observability::acquire_writer(
        pool,
        crate::observability::WriterOperation::EventWrite,
    )
    .await?;
    let row = sqlx::query(concat!(
        "SELECT root.reply_count, root.descendant_count, root.last_reply_at, ",
        recent_thread_participants_sql!(""),
        r#" AS participants
        FROM thread_metadata root
        WHERE root.community_id = $1 AND root.event_id = $2
        LIMIT 1
        "#,
    ))
    .bind(community_id.as_uuid())
    .bind(event_id)
    .fetch_optional(&mut *connection)
    .await?;

    row.map(|row| {
        Ok(ThreadSummary {
            reply_count: row.try_get("reply_count")?,
            descendant_count: row.try_get("descendant_count")?,
            last_reply_at: row.try_get("last_reply_at")?,
            participants: row.try_get("participants")?,
        })
    })
    .transpose()
}

/// Fetch one channel window: top-level rows (depth = 0, missing metadata, or
/// broadcast depth-1 replies) in `(created_at DESC, id ASC)` keyset order,
/// with thread summaries joined in, plus the server-side `has_more` fact.
///
/// `cursor` is the composite `(created_at, id)` of the last retained row from
/// the previous page — there is no timestamp-only fallback on this path (the
/// bridge rejects `until` without `before_id`). `None` = head of the channel.
///
/// `has_more` comes from an internal `limit + 1` probe evaluated after all
/// predicates (deletion, top-level, kinds). The sentinel row is dropped here
/// and never reaches the wire; callers must not re-derive exhaustion from row
/// counts (`rows < limit` proves nothing on an exact-multiple final page).
pub async fn get_channel_window(
    pool: &PgPool,
    community_id: CommunityId,
    channel_id: Uuid,
    limit: u32,
    cursor: Option<(DateTime<Utc>, Vec<u8>)>,
    kind_filter: Option<&[u32]>,
) -> Result<ChannelWindow> {
    let mut conn = crate::observability::acquire_writer(
        pool,
        crate::observability::WriterOperation::SubscriptionHistory,
    )
    .await?;
    get_channel_window_on(
        &mut conn,
        community_id,
        channel_id,
        limit,
        cursor,
        kind_filter,
        crate::event::ReadDeadline::PerRead,
    )
    .await
}

/// [`get_channel_window`] on a specific session — the replica-routing path
/// runs the page (and its participants batch) on the exact reader connection
/// whose heartbeat observation proved coverage (the proof is
/// connection-local; a different pooled session may sit at a different
/// replay position). The page runs under the event read deadline, shared by
/// its passes (see `ReadTransaction`).
pub(crate) async fn get_channel_window_on(
    conn: &mut sqlx::PgConnection,
    community_id: CommunityId,
    channel_id: Uuid,
    limit: u32,
    cursor: Option<(DateTime<Utc>, Vec<u8>)>,
    kind_filter: Option<&[u32]>,
    deadline: crate::event::ReadDeadline,
) -> Result<ChannelWindow> {
    let mut param_idx = 3u32; // $1 is community_id, $2 is channel_id
    let mut sql = String::from(
        r#"
        SELECT
            e.id,
            e.pubkey,
            e.created_at,
            e.kind,
            e.tags,
            e.content,
            e.sig,
            e.received_at,
            e.channel_id,
            tm.reply_count,
            tm.descendant_count,
            tm.last_reply_at
        FROM events e
        LEFT JOIN thread_metadata tm
            ON tm.community_id = e.community_id
           AND tm.event_created_at = e.created_at
           AND tm.event_id         = e.id
        WHERE e.community_id = $1
          AND e.channel_id = $2
          AND e.deleted_at IS NULL
          AND (
                tm.depth IS NULL
             OR tm.depth = 0
             OR (tm.depth = 1 AND tm.broadcast = true)
          )
        "#,
    );

    if cursor.is_some() {
        // Composite keyset: with ORDER BY created_at DESC, id ASC, the page
        // after (ts, id) is created_at < ts OR (created_at = ts AND id > id).
        let ts_idx = param_idx;
        let id_idx = param_idx + 1;
        sql.push_str(&format!(
            " AND (e.created_at < ${ts_idx} OR (e.created_at = ${ts_idx} AND e.id > ${id_idx}))"
        ));
        param_idx += 2;
    }

    if let Some(kinds) = kind_filter {
        if !kinds.is_empty() {
            let list = kinds
                .iter()
                .map(|k| k.to_string())
                .collect::<Vec<_>>()
                .join(",");
            sql.push_str(&format!(" AND e.kind IN ({list})"));
        }
    }

    // Recent-first (see `recent_pass`): the newest partitions answer a window
    // that fills there (with its +1 probe row); older rows are read only for
    // what is still missing.
    let kinds: Option<Vec<i32>> =
        kind_filter.map(|ks| ks.iter().filter_map(|&k| i32::try_from(k).ok()).collect());
    let split = crate::store::recent_pass::split_for(
        Utc::now(),
        kinds.as_deref(),
        None,
        cursor.as_ref().map(|(ts, _)| *ts),
    );
    // The +1 probe row is the server-internal has_more evidence.
    let probe_limit = i64::from(limit) + 1;
    let pass_query = |range: Range, pass_limit: i64| {
        let mut pass_sql = sql.clone();
        if let Some(bound) = range.predicate("e.created_at") {
            pass_sql.push_str(&bound);
        }
        pass_sql.push_str(&format!(
            " ORDER BY e.created_at DESC, e.id ASC LIMIT ${param_idx}"
        ));
        let mut q = sqlx::query(sqlx::AssertSqlSafe(pass_sql))
            .bind(community_id.as_uuid())
            .bind(channel_id);
        if let Some((ts, id)) = &cursor {
            q = q.bind(*ts).bind(id.clone());
        }
        q.bind(pass_limit)
    };
    let mut db_rows = match split {
        None => {
            crate::event::fetch_with_read_deadline(
                &mut *conn,
                pass_query(Range::All, probe_limit),
                deadline,
            )
            .await?
        }
        Some(split) => {
            // One snapshot and one deadline for both passes (see
            // `ReadTransaction`).
            let mut read = crate::event::ReadTransaction::begin(conn, deadline).await?;
            let mut rows = Vec::new();
            let mut pass = Some((Range::Recent(split), probe_limit));
            while let Some((range, pass_limit)) = pass {
                rows.extend(read.fetch(pass_query(range, pass_limit)).await?);
                pass = crate::store::recent_pass::next(range, rows.len(), probe_limit);
            }
            read.finish().await?;
            rows
        }
    };

    let has_more = db_rows.len() > limit as usize;
    db_rows.truncate(limit as usize);

    // Scan position of this page: the (created_at, id) of the last retained
    // raw row, captured before reconstruction so skip-and-continue rows can't
    // stall the cursor. Only meaningful when more rows exist past it.
    let next_cursor = if has_more {
        match db_rows.last() {
            Some(row) => Some((
                row.try_get::<DateTime<Utc>, _>("created_at")?,
                row.try_get::<Vec<u8>, _>("id")?,
            )),
            None => None,
        }
    } else {
        None
    };

    let mut rows = Vec::with_capacity(db_rows.len());
    for row in db_rows {
        let reply_count: Option<i32> = row.try_get("reply_count")?;
        let descendant_count: Option<i32> = row.try_get("descendant_count")?;
        let last_reply_at: Option<DateTime<Utc>> = row.try_get("last_reply_at")?;

        // Skip rows that fail event reconstruction rather than failing the
        // window, matching get_thread_replies' skip-and-continue semantics.
        let stored_event = match row_to_stored_event(row)? {
            Some(se) => se,
            None => continue,
        };

        let thread_summary = match reply_count {
            Some(rc) if rc > 0 => Some(ThreadSummary {
                reply_count: rc,
                descendant_count: descendant_count.unwrap_or(0),
                last_reply_at,
                participants: Vec::new(), // batch-filled below
            }),
            _ => None,
        };

        rows.push(ChannelWindowRow {
            stored_event,
            thread_summary,
        });
    }

    // Batch participants for every row with thread activity — one query for
    // the whole window instead of a per-root fan-out, each root bounded the
    // same way as get_thread_summary.
    let roots: Vec<Vec<u8>> = rows
        .iter()
        .filter(|r| r.thread_summary.is_some())
        .map(|r| r.stored_event.event.id.as_bytes().to_vec())
        .collect();
    if !roots.is_empty() {
        let participant_rows = sqlx::query(concat!(
            "SELECT root.event_id, ",
            recent_thread_participants_sql!(""),
            r#" AS participants
            FROM unnest($2::bytea[]) AS root(event_id)
            "#,
        ))
        .bind(community_id.as_uuid())
        .bind(&roots)
        .fetch_all(&mut *conn)
        .await?;

        let mut by_root: std::collections::HashMap<Vec<u8>, Vec<Vec<u8>>> =
            std::collections::HashMap::new();
        for row in participant_rows {
            by_root.insert(row.try_get("event_id")?, row.try_get("participants")?);
        }
        for row in &mut rows {
            if let Some(summary) = &mut row.thread_summary {
                if let Some(p) = by_root.remove(row.stored_event.event.id.as_bytes().as_slice()) {
                    summary.participants = p;
                }
            }
        }
    }

    Ok(ChannelWindow {
        rows,
        has_more,
        next_cursor,
    })
}

/// Look up a single thread_metadata row by event_id.
///
/// Used when processing soft-deletes to find the parent/root so reply counts
/// can be decremented.
pub async fn get_thread_metadata_by_event(
    pool: &PgPool,
    community_id: CommunityId,
    event_id: &[u8],
) -> Result<Option<ThreadMetadataRecord>> {
    let mut connection = crate::observability::acquire_writer(
        pool,
        crate::observability::WriterOperation::EventWrite,
    )
    .await?;
    let row = sqlx::query(
        r#"
        SELECT
            event_id,
            event_created_at,
            channel_id,
            parent_event_id,
            root_event_id,
            depth,
            reply_count,
            descendant_count,
            broadcast
        FROM thread_metadata
        WHERE community_id = $1 AND event_id = $2
        LIMIT 1
        "#,
    )
    .bind(community_id.as_uuid())
    .bind(event_id)
    .fetch_optional(&mut *connection)
    .await?;

    let row = match row {
        Some(r) => r,
        None => return Ok(None),
    };

    let event_id_col: Vec<u8> = row.try_get("event_id")?;
    let event_created_at: DateTime<Utc> = row.try_get("event_created_at")?;
    let channel_id: Uuid = row.try_get("channel_id")?;
    let parent_event_id: Option<Vec<u8>> = row.try_get("parent_event_id")?;
    let root_event_id: Option<Vec<u8>> = row.try_get("root_event_id")?;
    let depth: i32 = row.try_get("depth")?;
    let reply_count: i32 = row.try_get("reply_count")?;
    let descendant_count: i32 = row.try_get("descendant_count")?;
    let broadcast_val: bool = row.try_get("broadcast")?;

    Ok(Some(ThreadMetadataRecord {
        event_id: event_id_col,
        event_created_at,
        channel_id,
        parent_event_id,
        root_event_id,
        depth,
        reply_count,
        descendant_count,
        broadcast: broadcast_val,
    }))
}

// -- Db API -------------------------------------------------------------------

impl Db {
    /// Insert thread metadata.
    #[allow(clippy::too_many_arguments)]
    #[datastore_span(name = "insert_thread_metadata", system = "postgresql")]
    pub async fn insert_thread_metadata(
        &self,
        community_id: CommunityId,
        event_id: &[u8],
        event_created_at: DateTime<Utc>,
        channel_id: Uuid,
        parent_event_id: Option<&[u8]>,
        parent_event_created_at: Option<DateTime<Utc>>,
        root_event_id: Option<&[u8]>,
        root_event_created_at: Option<DateTime<Utc>>,
        depth: i32,
        broadcast: bool,
    ) -> Result<()> {
        crate::thread::insert_thread_metadata(
            &self.pool,
            community_id,
            event_id,
            event_created_at,
            channel_id,
            parent_event_id,
            parent_event_created_at,
            root_event_id,
            root_event_created_at,
            depth,
            broadcast,
        )
        .await
    }

    /// Fetch replies under a root event.
    ///
    /// Routing mirrors [`Db::get_channel_window_with_session`]: a head
    /// fetch (`cursor: None`) is Predicate A (bounded staleness, gated by
    /// the default-off head budget); cursor pages are Predicate B
    /// (completeness). Thread pagination walks **forward** from oldest to
    /// newest, so a cursor carries no upper bound — instead the served page
    /// is post-verified against the wall the serving session proved:
    ///
    /// - an under-`limit` page is a candidate terminal page — the client
    ///   treats it as EOF, so it is re-run on the writer to keep the EOF
    ///   decision authoritative (a lagged replica could truncate the tail);
    /// - a full page whose newest row exceeds the proved fence wall could
    ///   straddle a row the session has not replayed (commit order is not
    ///   `created_at` order), so it is also re-run on the writer. Only a
    ///   full page that sits entirely at or below the proved wall is served
    ///   from the replica.
    ///
    /// A head fetch routed under Predicate A skips the re-run: bounded
    /// staleness (missing at most the freshest budget-window of replies) is
    /// exactly the semantic the head gate accepts.
    #[datastore_span(name = "get_thread_replies", system = "postgresql")]
    pub async fn get_thread_replies(
        &self,
        community_id: CommunityId,
        root_event_id: &[u8],
        depth_limit: Option<u32>,
        limit: u32,
        cursor: Option<&[u8]>,
    ) -> Result<Vec<crate::thread::ThreadReply>> {
        let (path, predicate): (&'static str, RoutePredicate) = match cursor {
            Some(_) => (
                "thread_cursor",
                RoutePredicate::CoveredPostVerified {
                    proof: ChannelScoped::from_thread_metadata_join(),
                },
            ),
            None => ("thread_head", RoutePredicate::Bounded),
        };
        if let RouteDecision::Replica(mut tx, entry, reason) = self
            .route_read(
                path,
                predicate,
                crate::observability::ReaderOperation::SubscriptionHistory,
            )
            .await?
        {
            match crate::thread::get_thread_replies_on(
                &mut tx,
                community_id,
                root_event_id,
                depth_limit,
                limit,
                cursor,
            )
            .await
            {
                Ok(replies) => {
                    if cursor.is_none() {
                        // Predicate A: bounded-stale head page, served as proved.
                        Self::record_route(path, "replica", reason);
                        return Ok(replies);
                    }
                    let full = replies.len() >= limit as usize;
                    let below_fence = replies
                        .last()
                        .is_some_and(|tail| tail.created_at <= entry.fence_wall);
                    if full && below_fence {
                        Self::record_route(path, "replica", reason);
                        return Ok(replies);
                    }
                    // Candidate terminal page, or page reaching above the
                    // proved wall — verify against the writer. Recorded as
                    // the request's ONLY route event: the replica leg was
                    // discarded, so counting it would overstate offload.
                    Self::record_route("thread_eof", "writer", "stale");
                }
                // Surface a cancel instead of re-running it on the writer; see
                // `DbError::is_statement_cancelled` for the trade-off.
                Err(e) if e.is_statement_cancelled() => {
                    Self::record_route(path, "replica", "statement_cancelled");
                    return Err(e);
                }
                Err(e) => {
                    // Mid-request replica failure (e.g. a hot-standby
                    // recovery conflict) fails closed to the writer.
                    tracing::warn!(
                        error = %e,
                        path,
                        "replica thread query failed; re-running on writer"
                    );
                    Self::record_route(path, "writer", "replica_error");
                }
            }
        }
        crate::thread::get_thread_replies(
            &self.pool,
            community_id,
            root_event_id,
            depth_limit,
            limit,
            cursor,
        )
        .await
    }

    /// Fetch aggregated thread stats.
    #[datastore_span(name = "get_thread_summary", system = "postgresql")]
    pub async fn get_thread_summary(
        &self,
        community_id: CommunityId,
        event_id: &[u8],
    ) -> Result<Option<crate::thread::ThreadSummary>> {
        crate::thread::get_thread_summary(&self.pool, community_id, event_id).await
    }

    /// Resolve channel-scoped reply IDs, including tombstones, to current root
    /// summaries on the writer. Returns metadata only, never target payloads.
    #[datastore_span(name = "resolve_thread_root_summaries", system = "postgresql")]
    pub async fn resolve_thread_root_summaries(
        &self,
        community_id: CommunityId,
        channel_id: Uuid,
        target_ids: &[Vec<u8>],
    ) -> Result<Vec<(Vec<u8>, ThreadSummary)>> {
        let mut connection = crate::observability::acquire_writer(
            &self.pool,
            crate::observability::WriterOperation::SubscriptionHistory,
        )
        .await?;
        let rows = sqlx::query(concat!(
            "SELECT root.event_id, root.reply_count, root.descendant_count, root.last_reply_at, ",
            recent_thread_participants_sql!("AND tm.channel_id = root.channel_id"),
            r#" AS participants
            FROM thread_metadata root
            WHERE root.community_id = $1 AND root.channel_id = $2
                AND root.event_id IN (
                    SELECT COALESCE(target.root_event_id, target.parent_event_id)
                    FROM thread_metadata target
                    WHERE target.community_id = $1 AND target.channel_id = $2
                        AND target.event_id = ANY($3) AND target.parent_event_id IS NOT NULL
                )
        "#,
        ))
        .bind(community_id.as_uuid())
        .bind(channel_id)
        .bind(target_ids)
        .fetch_all(&mut *connection)
        .await?;
        rows.into_iter()
            .map(|row| {
                Ok((
                    row.try_get("event_id")?,
                    ThreadSummary {
                        reply_count: row.try_get("reply_count")?,
                        descendant_count: row.try_get("descendant_count")?,
                        last_reply_at: row.try_get("last_reply_at")?,
                        participants: row.try_get("participants")?,
                    },
                ))
            })
            .collect()
    }

    /// One channel window: top-level rows + summaries + server `has_more`.
    ///
    /// Convenience wrapper over [`Db::get_channel_window_with_session`] for
    /// callers with no follow-up queries; the serving session is released.
    pub async fn get_channel_window(
        &self,
        community_id: CommunityId,
        channel_id: Uuid,
        limit: u32,
        cursor: Option<(DateTime<Utc>, Vec<u8>)>,
        kind_filter: Option<&[u32]>,
    ) -> Result<crate::thread::ChannelWindow> {
        self.get_channel_window_with_session(community_id, channel_id, limit, cursor, kind_filter)
            .await
            .map(|(window, _session)| window)
    }

    /// [`Db::get_channel_window`], additionally returning the session that
    /// served the page so request-scoped follow-ups (the aux closure) run on
    /// the same proved connection.
    ///
    /// Routing:
    ///
    /// - **Cursor page** (Predicate B — completeness): scrolls *backward*
    ///   into history bounded above by the cursor timestamp (`created_at <
    ///   ts`, or `= ts` with the id tiebreak), so it may be served by a
    ///   replica session when one is configured AND that session **proves**
    ///   coverage of the cursor timestamp: the heartbeat token/epoch is
    ///   observed on the exact connection that will serve the page and
    ///   resolved against the fence's retained ring ([`crate::replica_fence`]).
    /// - **Head fetch** (Predicate A — bounded staleness): served by a
    ///   proved replica session only when the head gate is configured
    ///   ([`crate::DbConfig::replica_read_max_age_ms`], default off) and the
    ///   proved entry is within the budget. This trades a bounded staleness
    ///   window (budget plus probe cadence) on the GET leg for writer
    ///   offload. NOTE: enabling the budget also breaks read-your-own-writes
    ///   on the GET leg; the client-side WS `since`-overlap union intended
    ///   to cover fresh events has NOT shipped yet — do not enable
    ///   `BUZZ_REPLICA_HEAD_MAX_AGE_SECS` until it has, proven by a
    ///   post-then-immediately-refetch test.
    ///
    /// Every replica failure except a cancelled statement (57014, which the
    /// writer would only repeat) fails closed to the writer and is recorded
    /// in `buzz_db_route_decision`. A busy reader rejects the read as
    /// retryable instead (see [`Db::route_read`]).
    #[datastore_span(name = "get_channel_window", system = "postgresql")]
    pub async fn get_channel_window_with_session(
        &self,
        community_id: CommunityId,
        channel_id: Uuid,
        limit: u32,
        cursor: Option<(DateTime<Utc>, Vec<u8>)>,
        kind_filter: Option<&[u32]>,
    ) -> Result<(crate::thread::ChannelWindow, ReadSession)> {
        let path: &'static str = if cursor.is_some() {
            "channel_cursor"
        } else {
            "channel_head"
        };
        match self
            .route_read(
                path,
                RoutePredicate::from_channel_cursor(channel_id, &cursor),
                crate::observability::ReaderOperation::SubscriptionHistory,
            )
            .await?
        {
            RouteDecision::Replica(mut tx, _entry, reason) => {
                match crate::thread::get_channel_window_on(
                    &mut tx,
                    community_id,
                    channel_id,
                    limit,
                    cursor.clone(),
                    kind_filter,
                    crate::event::ReadDeadline::Session,
                )
                .await
                {
                    Ok(window) => {
                        Self::record_route(path, "replica", reason);
                        return Ok((
                            window,
                            ReadSession {
                                inner: ReadSessionInner::Replica {
                                    tx,
                                    writer: self.pool.clone(),
                                },
                            },
                        ));
                    }
                    // Surface a cancel instead of re-running it on the writer; see
                    // `DbError::is_statement_cancelled` for the trade-off.
                    Err(e) if e.is_statement_cancelled() => {
                        Self::record_route(path, "replica", "statement_cancelled");
                        return Err(e);
                    }
                    Err(e) => {
                        // A mid-request replica failure (e.g. a hot-standby
                        // recovery conflict cancelling the held snapshot)
                        // fails closed to the writer: a stale-but-served
                        // page, never an error the writer could have
                        // answered. Dropping `tx` rolls the reader
                        // transaction back.
                        tracing::warn!(
                            error = %e,
                            path,
                            "replica window query failed; re-running on writer"
                        );
                        Self::record_route(path, "writer", "replica_error");
                    }
                }
            }
            RouteDecision::Writer => {}
        }
        let window = crate::thread::get_channel_window(
            &self.pool,
            community_id,
            channel_id,
            limit,
            cursor,
            kind_filter,
        )
        .await?;
        Ok((
            window,
            ReadSession {
                inner: ReadSessionInner::Writer(self.pool.clone()),
            },
        ))
    }

    /// Look up a single thread_metadata row by event_id.
    #[datastore_span(name = "get_thread_metadata_by_event", system = "postgresql")]
    pub async fn get_thread_metadata_by_event(
        &self,
        community_id: CommunityId,
        event_id: &[u8],
    ) -> Result<Option<crate::thread::ThreadMetadataRecord>> {
        crate::thread::get_thread_metadata_by_event(&self.pool, community_id, event_id).await
    }
}

#[cfg(test)]
mod postgres_tests {
    use super::*;
    use crate::{
        channel::{ChannelType, ChannelVisibility},
        event::{insert_event_with_thread_metadata, ReadDeadline, ThreadMetadataParams},
    };
    use nostr::{EventBuilder, Keys, Kind};

    const TEST_DB_URL: &str = "postgres://buzz:buzz_dev@localhost:5432/buzz"; // sadscan:disable np.postgres.1

    async fn setup_pool() -> PgPool {
        let database_url = std::env::var("BUZZ_TEST_DATABASE_URL")
            .or_else(|_| std::env::var("DATABASE_URL"))
            .unwrap_or_else(|_| TEST_DB_URL.to_owned());

        PgPool::connect(&database_url)
            .await
            .expect("connect to test DB")
    }

    fn make_stream_event(keys: &Keys, content: &str) -> nostr::Event {
        EventBuilder::new(Kind::Custom(9), content)
            .sign_with_keys(keys)
            .expect("sign event")
    }

    fn event_created_at(event: &nostr::Event) -> DateTime<Utc> {
        DateTime::from_timestamp(event.created_at.as_secs() as i64, 0)
            .expect("event timestamp is valid")
    }

    async fn make_test_community(pool: &PgPool) -> Uuid {
        let id = Uuid::new_v4();
        let host = format!("thread-test-{}.example", id.simple());
        sqlx::query("INSERT INTO communities (id, host) VALUES ($1, $2)")
            .bind(id)
            .bind(host)
            .execute(pool)
            .await
            .expect("insert test community");
        id
    }

    async fn create_test_channel(
        pool: &PgPool,
        name: &str,
        channel_type: ChannelType,
        visibility: ChannelVisibility,
        description: Option<&str>,
        created_by: &[u8],
        ttl_seconds: Option<i32>,
    ) -> crate::error::Result<(crate::channel::ChannelRecord, buzz_core::CommunityId)> {
        let id = Uuid::new_v4();
        let community_id = make_test_community(pool).await;

        sqlx::query(
            r#"
            INSERT INTO channels
                (id, community_id, name, channel_type, visibility, description, created_by, ttl_seconds, ttl_deadline)
            VALUES
                ($1, $2, $3, $4::channel_type, $5::channel_visibility, $6, $7, $8,
                 CASE WHEN $8 IS NOT NULL THEN NOW() + ($8 || ' seconds')::interval ELSE NULL END)
            "#,
        )
        .bind(id)
        .bind(community_id)
        .bind(name)
        .bind(channel_type.as_str())
        .bind(visibility.as_str())
        .bind(description)
        .bind(created_by)
        .bind(ttl_seconds)
        .execute(pool)
        .await
        .expect("insert test channel");

        sqlx::query(
            r#"
            INSERT INTO channel_members (community_id, channel_id, pubkey, role, invited_by)
            VALUES ($1, $2, $3, 'owner', $4)
            "#,
        )
        .bind(community_id)
        .bind(id)
        .bind(created_by)
        .bind(created_by)
        .execute(pool)
        .await
        .expect("insert owner membership");

        crate::channel::get_channel(pool, buzz_core::CommunityId::from_uuid(community_id), id)
            .await
            .map(|channel| (channel, buzz_core::CommunityId::from_uuid(community_id)))
    }

    #[tokio::test]
    #[ignore = "requires Postgres"]
    async fn resolve_deleted_roots_preserves_community_scope_for_colliding_ids() {
        let pool = setup_pool().await;
        let db = Db::from_pool(pool.clone());
        let author = Keys::generate();
        let channel = Uuid::new_v4();
        let communities = [
            CommunityId::from_uuid(make_test_community(&pool).await),
            CommunityId::from_uuid(make_test_community(&pool).await),
        ];
        let root = make_stream_event(&author, "shared root identity");
        let reply = make_stream_event(&author, "shared reply identity");
        for community in communities {
            crate::channel::create_channel_with_id(
                &pool,
                community,
                channel,
                &format!("root-scope-{channel}"),
                ChannelType::Stream,
                ChannelVisibility::Open,
                None,
                author.public_key().to_bytes().as_slice(),
                None,
                &[],
            )
            .await
            .expect("channel");
            for (event, parent) in [(&root, None), (&reply, Some(&root))] {
                insert_event_with_thread_metadata(
                    &pool,
                    community,
                    event,
                    Some(channel),
                    Some(ThreadMetadataParams {
                        event_id: event.id.as_bytes(),
                        event_created_at: event_created_at(event),
                        channel_id: channel,
                        parent_event_id: parent.map(|parent| parent.id.as_bytes().as_slice()),
                        parent_event_created_at: parent.map(event_created_at),
                        root_event_id: parent.map(|parent| parent.id.as_bytes().as_slice()),
                        root_event_created_at: parent.map(event_created_at),
                        depth: i32::from(parent.is_some()),
                        broadcast: false,
                    }),
                )
                .await
                .expect("thread event");
            }
        }
        db.soft_delete_event_and_update_thread(
            communities[0],
            reply.id.as_bytes(),
            Some(root.id.as_bytes()),
            Some(root.id.as_bytes()),
        )
        .await
        .expect("delete only A");
        for (community, expected) in [(communities[0], 0), (communities[1], 1)] {
            let summaries = db
                .resolve_thread_root_summaries(community, channel, &[reply.id.as_bytes().to_vec()])
                .await
                .expect("metadata query");
            assert_eq!(summaries.len(), 1);
            assert_eq!(summaries[0].0, root.id.as_bytes());
            assert_eq!(summaries[0].1.descendant_count, expected);
        }
        assert!(db
            .resolve_thread_root_summaries(
                communities[0],
                Uuid::new_v4(),
                &[reply.id.as_bytes().to_vec()]
            )
            .await
            .expect("wrong channel")
            .is_empty());
    }

    #[tokio::test]
    #[ignore = "requires Postgres"]
    async fn get_thread_metadata_by_event_is_scoped_when_event_id_collides_across_communities() {
        let pool = setup_pool().await;
        let author = Keys::generate();
        let channel_id = Uuid::new_v4();
        let community_a = make_test_community(&pool).await;
        let community_b = make_test_community(&pool).await;
        let community_a = buzz_core::CommunityId::from_uuid(community_a);
        let community_b = buzz_core::CommunityId::from_uuid(community_b);

        crate::channel::create_channel_with_id(
            &pool,
            community_a,
            channel_id,
            &format!("thread-collision-a-{channel_id}"),
            ChannelType::Stream,
            ChannelVisibility::Open,
            None,
            author.public_key().to_bytes().as_slice(),
            None,
            &[],
        )
        .await
        .expect("create community A channel");
        crate::channel::create_channel_with_id(
            &pool,
            community_b,
            channel_id,
            &format!("thread-collision-b-{channel_id}"),
            ChannelType::Stream,
            ChannelVisibility::Open,
            None,
            author.public_key().to_bytes().as_slice(),
            None,
            &[],
        )
        .await
        .expect("create community B channel");

        let event = make_stream_event(&author, "same id in both communities");
        let created_at = event_created_at(&event);
        insert_event_with_thread_metadata(
            &pool,
            community_a,
            &event,
            Some(channel_id),
            Some(ThreadMetadataParams {
                event_id: event.id.as_bytes(),
                event_created_at: created_at,
                channel_id,
                parent_event_id: None,
                parent_event_created_at: None,
                root_event_id: None,
                root_event_created_at: None,
                depth: 0,
                broadcast: true,
            }),
        )
        .await
        .expect("insert community A metadata");
        insert_event_with_thread_metadata(
            &pool,
            community_b,
            &event,
            Some(channel_id),
            Some(ThreadMetadataParams {
                event_id: event.id.as_bytes(),
                event_created_at: created_at,
                channel_id,
                parent_event_id: None,
                parent_event_created_at: None,
                root_event_id: None,
                root_event_created_at: None,
                depth: 3,
                broadcast: false,
            }),
        )
        .await
        .expect("insert community B metadata");

        let a = get_thread_metadata_by_event(&pool, community_a, event.id.as_bytes())
            .await
            .expect("lookup community A metadata")
            .expect("community A metadata exists");
        let b = get_thread_metadata_by_event(&pool, community_b, event.id.as_bytes())
            .await
            .expect("lookup community B metadata")
            .expect("community B metadata exists");

        assert_eq!(a.depth, 0);
        assert!(a.broadcast);
        assert_eq!(b.depth, 3);
        assert!(!b.broadcast);
    }

    #[tokio::test]
    #[ignore = "requires Postgres"]
    async fn get_thread_replies_reconstructs_stored_events() {
        let pool = setup_pool().await;
        let author = Keys::generate();
        let (channel, community) = create_test_channel(
            &pool,
            &format!("thread-replies-{}", Uuid::new_v4()),
            ChannelType::Stream,
            ChannelVisibility::Open,
            None,
            author.public_key().to_bytes().as_slice(),
            None,
        )
        .await
        .expect("create channel");

        let root = make_stream_event(&author, "root");
        let root_created_at = event_created_at(&root);
        insert_event_with_thread_metadata(&pool, community, &root, Some(channel.id), None)
            .await
            .expect("insert root event");

        let reply = make_stream_event(&author, "reply");
        let reply_created_at = event_created_at(&reply);
        let reply_id = reply.id.to_hex();
        insert_event_with_thread_metadata(
            &pool,
            community,
            &reply,
            Some(channel.id),
            Some(ThreadMetadataParams {
                event_id: reply.id.as_bytes(),
                event_created_at: reply_created_at,
                channel_id: channel.id,
                parent_event_id: Some(root.id.as_bytes()),
                parent_event_created_at: Some(root_created_at),
                root_event_id: Some(root.id.as_bytes()),
                root_event_created_at: Some(root_created_at),
                depth: 1,
                broadcast: false,
            }),
        )
        .await
        .expect("insert reply event and metadata");

        let replies = get_thread_replies(&pool, community, root.id.as_bytes(), Some(10), 10, None)
            .await
            .expect("fetch thread replies");

        assert_eq!(replies.len(), 1);
        assert_eq!(replies[0].stored_event.event.id.to_hex(), reply_id);
        assert_eq!(replies[0].stored_event.event.content, "reply");
        assert_eq!(replies[0].stored_event.channel_id, Some(channel.id));
        assert_eq!(replies[0].depth, 1);
    }

    /// Replies that share the same `created_at` second (bursty threads are the
    /// common case) must paginate without gaps or duplicates. Before the
    /// composite `(event_created_at, event_id)` keyset, a timestamp-only cursor
    /// advanced past the whole tied second after one page, silently dropping
    /// every tied reply beyond the first page's limit — the "missed messages"
    /// bug this read-path work exists to fix. This pins the tiebreak.
    #[tokio::test]
    #[ignore = "requires Postgres"]
    async fn get_thread_replies_pages_same_second_ties_without_loss() {
        use nostr::Timestamp;

        let pool = setup_pool().await;
        let author = Keys::generate();
        let (channel, community) = create_test_channel(
            &pool,
            &format!("thread-ties-{}", Uuid::new_v4()),
            ChannelType::Stream,
            ChannelVisibility::Open,
            None,
            author.public_key().to_bytes().as_slice(),
            None,
        )
        .await
        .expect("create channel");

        let root = make_stream_event(&author, "root");
        let root_created_at = event_created_at(&root);
        insert_event_with_thread_metadata(&pool, community, &root, Some(channel.id), None)
            .await
            .expect("insert root event");

        // Pin every reply to the SAME second so pagination must lean entirely on
        // the event_id tiebreak. Distinct content keeps the ids distinct.
        let tie_secs: u64 = root.created_at.as_secs() + 1;
        let tie_ts = DateTime::from_timestamp(tie_secs as i64, 0).expect("valid timestamp");
        let reply_count = 5usize;
        let mut expected_ids = Vec::with_capacity(reply_count);
        for i in 0..reply_count {
            let reply = EventBuilder::new(Kind::Custom(9), format!("tie-{i}"))
                .custom_created_at(Timestamp::from(tie_secs))
                .sign_with_keys(&author)
                .expect("sign tied reply");
            expected_ids.push(reply.id.as_bytes().to_vec());
            insert_event_with_thread_metadata(
                &pool,
                community,
                &reply,
                Some(channel.id),
                Some(ThreadMetadataParams {
                    event_id: reply.id.as_bytes(),
                    event_created_at: tie_ts,
                    channel_id: channel.id,
                    parent_event_id: Some(root.id.as_bytes()),
                    parent_event_created_at: Some(root_created_at),
                    root_event_id: Some(root.id.as_bytes()),
                    root_event_created_at: Some(root_created_at),
                    depth: 1,
                    broadcast: false,
                }),
            )
            .await
            .expect("insert tied reply");
        }

        // Page with limit=2 across 5 same-second replies. Build the next cursor
        // from the last row's (created_at seconds, event_id) — the exact keyset
        // the bridge derives transparently for the client.
        let page_limit = 2u32;
        let mut collected: Vec<Vec<u8>> = Vec::new();
        let mut cursor: Option<Vec<u8>> = None;
        loop {
            let page = get_thread_replies(
                &pool,
                community,
                root.id.as_bytes(),
                Some(10),
                page_limit,
                cursor.as_deref(),
            )
            .await
            .expect("fetch page");
            if page.is_empty() {
                break;
            }
            let last = page.last().expect("non-empty page");
            let mut next = last.created_at.timestamp().to_be_bytes().to_vec();
            next.extend_from_slice(&last.event_id);
            cursor = Some(next);
            let full = page.len() as u32 == page_limit;
            for reply in page {
                collected.push(reply.event_id);
            }
            if !full {
                break;
            }
        }

        // No gaps, no duplicates: the paged union equals the full tied set.
        assert_eq!(
            collected.len(),
            reply_count,
            "paged {} replies, expected all {}",
            collected.len(),
            reply_count
        );
        let mut unique = collected.clone();
        unique.sort();
        unique.dedup();
        assert_eq!(unique.len(), reply_count, "paging produced duplicates");
        let mut expected_sorted = expected_ids.clone();
        expected_sorted.sort();
        assert_eq!(unique, expected_sorted, "paged set != inserted tied set");
    }

    /// Nested replies (depth >= 2) must be reachable in a subtree read. Every
    /// existing thread test uses `parent == root` (depth-1 direct replies), so
    /// the depth>1 path — where `root_event_id != parent_event_id` and the root
    /// stub is created by a nested reply arriving before the root has a
    /// metadata row — was never exercised. `get_thread_replies` advertises
    /// depth-64 subtree reads, so pin that a grandchild reply is returned and
    /// its depth is recorded. This also exercises the root-stub INSERT branch
    /// (`root_id != pid`) end-to-end through the production insert path.
    #[tokio::test]
    #[ignore = "requires Postgres"]
    async fn get_thread_replies_reaches_nested_depth_two_replies() {
        let pool = setup_pool().await;
        let author = Keys::generate();
        let (channel, community) = create_test_channel(
            &pool,
            &format!("thread-nested-{}", Uuid::new_v4()),
            ChannelType::Stream,
            ChannelVisibility::Open,
            None,
            author.public_key().to_bytes().as_slice(),
            None,
        )
        .await
        .expect("create channel");

        // Root (no metadata row on first insert — a depth-0 message).
        let root = make_stream_event(&author, "root");
        let root_created_at = event_created_at(&root);
        insert_event_with_thread_metadata(&pool, community, &root, Some(channel.id), None)
            .await
            .expect("insert root event");

        // Depth-1 direct reply to the root (parent == root).
        let child = make_stream_event(&author, "child");
        let child_created_at = event_created_at(&child);
        insert_event_with_thread_metadata(
            &pool,
            community,
            &child,
            Some(channel.id),
            Some(ThreadMetadataParams {
                event_id: child.id.as_bytes(),
                event_created_at: child_created_at,
                channel_id: channel.id,
                parent_event_id: Some(root.id.as_bytes()),
                parent_event_created_at: Some(root_created_at),
                root_event_id: Some(root.id.as_bytes()),
                root_event_created_at: Some(root_created_at),
                depth: 1,
                broadcast: false,
            }),
        )
        .await
        .expect("insert depth-1 child");

        // Depth-2 grandchild: parent is the child, root is the root. This is the
        // `root_id != parent_id` case that fires the nested root-stub branch.
        let grandchild = make_stream_event(&author, "grandchild");
        let grandchild_created_at = event_created_at(&grandchild);
        insert_event_with_thread_metadata(
            &pool,
            community,
            &grandchild,
            Some(channel.id),
            Some(ThreadMetadataParams {
                event_id: grandchild.id.as_bytes(),
                event_created_at: grandchild_created_at,
                channel_id: channel.id,
                parent_event_id: Some(child.id.as_bytes()),
                parent_event_created_at: Some(child_created_at),
                root_event_id: Some(root.id.as_bytes()),
                root_event_created_at: Some(root_created_at),
                depth: 2,
                broadcast: false,
            }),
        )
        .await
        .expect("insert depth-2 grandchild");

        // Read the whole subtree under the root.
        let replies = get_thread_replies(&pool, community, root.id.as_bytes(), Some(64), 100, None)
            .await
            .expect("fetch subtree");

        let by_id: std::collections::HashMap<Vec<u8>, i32> = replies
            .iter()
            .map(|r| (r.event_id.clone(), r.depth))
            .collect();
        assert_eq!(
            replies.len(),
            2,
            "both the child and grandchild must be reached"
        );
        assert_eq!(
            by_id.get(child.id.as_bytes().as_slice()),
            Some(&1),
            "depth-1 child must be present at depth 1"
        );
        assert_eq!(
            by_id.get(grandchild.id.as_bytes().as_slice()),
            Some(&2),
            "depth-2 grandchild must be reached (subtree read must not stop at depth 1)"
        );
    }

    /// Direct guard on [`insert_thread_metadata`]'s nested root-stub INSERT. The
    /// column list once omitted `community_id` while still binding it, scrambling
    /// every placeholder (the bind for `community_id` landed on `event_created_at`
    /// etc.), so any nested reply (`root_id != parent_id`) whose root lacked a
    /// metadata row failed the whole insert. The production ingest path uses
    /// `insert_event_with_thread_metadata` (event.rs), whose copy was already
    /// correct, which is why no test caught this. Pin the standalone function so
    /// the scrambled SQL can't return.
    #[tokio::test]
    #[ignore = "requires Postgres"]
    async fn insert_thread_metadata_nested_reply_creates_root_stub() {
        let pool = setup_pool().await;
        let author = Keys::generate();
        let (channel, community) = create_test_channel(
            &pool,
            &format!("thread-stub-{}", Uuid::new_v4()),
            ChannelType::Stream,
            ChannelVisibility::Open,
            None,
            author.public_key().to_bytes().as_slice(),
            None,
        )
        .await
        .expect("create channel");

        // Insert the events themselves (no thread metadata yet) so the FK/row
        // exists; then drive `insert_thread_metadata` directly for a depth-2
        // reply whose root has no metadata row — the root-stub branch.
        let root = make_stream_event(&author, "root");
        let child = make_stream_event(&author, "child");
        let grandchild = make_stream_event(&author, "grandchild");
        for ev in [&root, &child, &grandchild] {
            insert_event_with_thread_metadata(&pool, community, ev, Some(channel.id), None)
                .await
                .expect("insert event");
        }

        // Depth-2 reply where root_id != parent_id. Before the fix this errored
        // inside the transaction (UUID bound to a TIMESTAMPTZ placeholder).
        insert_thread_metadata(
            &pool,
            community,
            grandchild.id.as_bytes(),
            event_created_at(&grandchild),
            channel.id,
            Some(child.id.as_bytes()),
            Some(event_created_at(&child)),
            Some(root.id.as_bytes()),
            Some(event_created_at(&root)),
            2,
            false,
        )
        .await
        .expect("nested insert must succeed and create the root stub");

        // The root stub must now exist and be readable.
        let replies = get_thread_replies(&pool, community, root.id.as_bytes(), Some(64), 100, None)
            .await
            .expect("fetch subtree");
        assert!(
            replies
                .iter()
                .any(|r| r.event_id == grandchild.id.as_bytes()),
            "grandchild must be reachable under the root after the nested insert"
        );
    }

    /// A reply whose stored row can no longer be reconstructed into a
    /// `nostr::Event` (e.g. corrupt signature from out-of-band storage damage)
    /// must be skipped, with the rest of the thread still returned — not
    /// surfaced as a query error that takes down the whole thread read.
    #[tokio::test]
    #[ignore = "requires Postgres"]
    async fn get_thread_replies_skips_unreconstructable_row() {
        let pool = setup_pool().await;
        let author = Keys::generate();
        let (channel, community) = create_test_channel(
            &pool,
            &format!("thread-replies-corrupt-{}", Uuid::new_v4()),
            ChannelType::Stream,
            ChannelVisibility::Open,
            None,
            author.public_key().to_bytes().as_slice(),
            None,
        )
        .await
        .expect("create channel");

        let root = make_stream_event(&author, "root");
        let root_created_at = event_created_at(&root);
        insert_event_with_thread_metadata(&pool, community, &root, Some(channel.id), None)
            .await
            .expect("insert root event");

        // Two replies under the same root: one stays valid, one we corrupt.
        let good = EventBuilder::new(Kind::Custom(9), "good")
            .custom_created_at(nostr::Timestamp::from(root.created_at.as_secs() + 2))
            .sign_with_keys(&author)
            .expect("good reply");
        let good_id = good.id.to_hex();
        let good_created_at = event_created_at(&good);
        insert_event_with_thread_metadata(
            &pool,
            community,
            &good,
            Some(channel.id),
            Some(ThreadMetadataParams {
                event_id: good.id.as_bytes(),
                event_created_at: good_created_at,
                channel_id: channel.id,
                parent_event_id: Some(root.id.as_bytes()),
                parent_event_created_at: Some(root_created_at),
                root_event_id: Some(root.id.as_bytes()),
                root_event_created_at: Some(root_created_at),
                depth: 1,
                broadcast: false,
            }),
        )
        .await
        .expect("insert good reply");

        let bad = EventBuilder::new(Kind::Custom(9), "bad")
            .custom_created_at(nostr::Timestamp::from(root.created_at.as_secs() + 1))
            .sign_with_keys(&author)
            .expect("bad reply");
        let bad_created_at = event_created_at(&bad);
        insert_event_with_thread_metadata(
            &pool,
            community,
            &bad,
            Some(channel.id),
            Some(ThreadMetadataParams {
                event_id: bad.id.as_bytes(),
                event_created_at: bad_created_at,
                channel_id: channel.id,
                parent_event_id: Some(root.id.as_bytes()),
                parent_event_created_at: Some(root_created_at),
                root_event_id: Some(root.id.as_bytes()),
                root_event_created_at: Some(root_created_at),
                depth: 1,
                broadcast: false,
            }),
        )
        .await
        .expect("insert bad reply");

        // Corrupt the bad reply's signature in place: truncating the 64-byte
        // sig makes `row_to_stored_event` fail to deserialize the event and
        // return `Ok(None)` — the case the skip-and-continue must handle.
        let rows_changed = sqlx::query("UPDATE events SET sig = $1 WHERE id = $2")
            .bind(vec![0u8; 32])
            .bind(bad.id.as_bytes())
            .execute(&pool)
            .await
            .expect("corrupt bad reply sig")
            .rows_affected();
        assert_eq!(rows_changed, 1, "expected to corrupt exactly one row");

        let replies = get_thread_replies(&pool, community, root.id.as_bytes(), Some(10), 10, None)
            .await
            .expect("fetch thread replies must succeed despite a corrupt row");

        // The corrupt reply is skipped; the valid one survives. The whole
        // query does NOT 500 on a single unreconstructable row.
        assert_eq!(replies.len(), 1);
        assert_eq!(replies[0].stored_event.event.id.to_hex(), good_id);
        assert_eq!(replies[0].stored_event.event.content, "good");

        let tail = EventBuilder::new(Kind::Custom(9), "tail")
            .custom_created_at(nostr::Timestamp::from(root.created_at.as_secs() + 3))
            .sign_with_keys(&author)
            .expect("tail reply");
        insert_event_with_thread_metadata(
            &pool,
            community,
            &tail,
            Some(channel.id),
            Some(ThreadMetadataParams {
                event_id: tail.id.as_bytes(),
                event_created_at: event_created_at(&tail),
                channel_id: channel.id,
                parent_event_id: Some(root.id.as_bytes()),
                parent_event_created_at: Some(root_created_at),
                root_event_id: Some(root.id.as_bytes()),
                root_event_created_at: Some(root_created_at),
                depth: 1,
                broadcast: false,
            }),
        )
        .await
        .expect("insert tail");
        let mut origin = (-1_i64).to_be_bytes().to_vec();
        origin.extend_from_slice(&[0; 32]);
        let page = get_thread_replies(
            &pool,
            community,
            root.id.as_bytes(),
            Some(10),
            2,
            Some(&origin),
        )
        .await
        .expect("fill page past corrupt first row");
        assert_eq!(
            page.iter()
                .map(|reply| reply.stored_event.event.id)
                .collect::<Vec<_>>(),
            [good.id, tail.id]
        );
        let first = get_thread_replies(
            &pool,
            community,
            root.id.as_bytes(),
            Some(10),
            1,
            Some(&origin),
        )
        .await
        .expect("page entirely corrupt must not signal EOF");
        assert_eq!(first[0].stored_event.event.id, good.id);
        let mut after = (tail.created_at.as_secs() as i64).to_be_bytes().to_vec();
        after.extend_from_slice(tail.id.as_bytes());
        assert!(get_thread_replies(
            &pool,
            community,
            root.id.as_bytes(),
            Some(10),
            2,
            Some(&after)
        )
        .await
        .expect("true EOF")
        .is_empty());
    }

    #[tokio::test]
    #[ignore = "requires Postgres"]
    async fn get_thread_replies_bounds_corrupt_row_refills() {
        let pool = setup_pool().await;
        let author = Keys::generate();
        let (channel, community) = create_test_channel(
            &pool,
            &format!("corrupt-bound-{}", Uuid::new_v4()),
            ChannelType::Stream,
            ChannelVisibility::Open,
            None,
            author.public_key().to_bytes().as_slice(),
            None,
        )
        .await
        .expect("channel");
        let root = make_stream_event(&author, "root");
        insert_event_with_thread_metadata(&pool, community, &root, Some(channel.id), None)
            .await
            .expect("root");
        let mut corrupt_ids = Vec::new();
        for index in 0..65 {
            let reply = EventBuilder::new(Kind::Custom(9), format!("reply-{index}"))
                .custom_created_at(nostr::Timestamp::from(
                    root.created_at.as_secs() + index + 1,
                ))
                .sign_with_keys(&author)
                .expect("reply");
            insert_event_with_thread_metadata(
                &pool,
                community,
                &reply,
                Some(channel.id),
                Some(ThreadMetadataParams {
                    event_id: reply.id.as_bytes(),
                    event_created_at: event_created_at(&reply),
                    channel_id: channel.id,
                    parent_event_id: Some(root.id.as_bytes()),
                    parent_event_created_at: Some(event_created_at(&root)),
                    root_event_id: Some(root.id.as_bytes()),
                    root_event_created_at: Some(event_created_at(&root)),
                    depth: 1,
                    broadcast: false,
                }),
            )
            .await
            .expect("reply metadata");
            if index < 64 {
                corrupt_ids.push(reply.id.as_bytes().to_vec());
            }
        }
        sqlx::query("UPDATE events SET sig = $1 WHERE community_id = $2 AND id = ANY($3)")
            .bind(vec![0u8; 32])
            .bind(community.as_uuid())
            .bind(&corrupt_ids)
            .execute(&pool)
            .await
            .expect("damage first 64 rows");
        let error = get_thread_replies(&pool, community, root.id.as_bytes(), Some(10), 1, None)
            .await
            .expect_err("refill cap must fail, never return partial EOF or scan to the tail");
        assert!(
            matches!(error, crate::error::DbError::InvalidData(ref message)
            if message.contains("corrupt-row refill limit"))
        );
        // A larger page needs fewer than 64 raw reads and still finds the tail.
        let replies = get_thread_replies(&pool, community, root.id.as_bytes(), Some(10), 2, None)
            .await
            .expect("bounded skip remains supported");
        assert_eq!(replies.len(), 1);
        assert_eq!(replies[0].stored_event.event.content, "reply-64");
        // The final permitted raw page may be valid; accept it at the cap.
        sqlx::query("UPDATE events SET deleted_at = NOW() WHERE community_id = $1 AND id = $2")
            .bind(community.as_uuid())
            .bind(&corrupt_ids[0])
            .execute(&pool)
            .await
            .expect("hide one corrupt row");
        let boundary = get_thread_replies(&pool, community, root.id.as_bytes(), Some(10), 1, None)
            .await
            .expect("valid final refill page must succeed");
        assert_eq!(boundary[0].stored_event.event.content, "reply-64");
    }

    /// Insert one top-level event (root metadata, broadcast) into a channel.
    async fn insert_root(
        pool: &PgPool,
        community: CommunityId,
        channel_id: Uuid,
        event: &nostr::Event,
    ) {
        insert_event_with_thread_metadata(
            pool,
            community,
            event,
            Some(channel_id),
            Some(ThreadMetadataParams {
                event_id: event.id.as_bytes(),
                event_created_at: event_created_at(event),
                channel_id,
                parent_event_id: None,
                parent_event_created_at: None,
                root_event_id: None,
                root_event_created_at: None,
                depth: 0,
                broadcast: true,
            }),
        )
        .await
        .expect("insert top-level event");
    }

    /// Insert a depth-1 reply under `root`, with the given broadcast flag.
    async fn insert_reply(
        pool: &PgPool,
        community: CommunityId,
        channel_id: Uuid,
        root: &nostr::Event,
        reply: &nostr::Event,
        broadcast: bool,
    ) {
        insert_event_with_thread_metadata(
            pool,
            community,
            reply,
            Some(channel_id),
            Some(ThreadMetadataParams {
                event_id: reply.id.as_bytes(),
                event_created_at: event_created_at(reply),
                channel_id,
                parent_event_id: Some(root.id.as_bytes()),
                parent_event_created_at: Some(event_created_at(root)),
                root_event_id: Some(root.id.as_bytes()),
                root_event_created_at: Some(event_created_at(root)),
                depth: 1,
                broadcast,
            }),
        )
        .await
        .expect("insert reply event");
    }

    /// The window's top-level predicate: roots (depth 0), events with no
    /// thread metadata at all (pre-metadata legacy rows), and broadcast
    /// depth-1 replies are rows; ordinary replies never are. This is the
    /// SQL-level guarantee that replaces the client-side "filter out replies,
    /// splice back islands" machinery.
    #[tokio::test]
    #[ignore = "requires Postgres"]
    async fn channel_window_top_level_predicate() {
        let pool = setup_pool().await;
        let author = Keys::generate();
        let (channel, community) = create_test_channel(
            &pool,
            &format!("window-predicate-{}", Uuid::new_v4()),
            ChannelType::Stream,
            ChannelVisibility::Open,
            None,
            author.public_key().to_bytes().as_slice(),
            None,
        )
        .await
        .expect("create channel");

        let root = make_stream_event(&author, "root");
        insert_root(&pool, community, channel.id, &root).await;

        // No thread metadata at all — the legacy-ingest shape. Top-level.
        let bare = make_stream_event(&author, "bare");
        insert_event_with_thread_metadata(&pool, community, &bare, Some(channel.id), None)
            .await
            .expect("insert bare event");

        let broadcast_reply = make_stream_event(&author, "broadcast reply");
        insert_reply(&pool, community, channel.id, &root, &broadcast_reply, true).await;

        let quiet_reply = make_stream_event(&author, "quiet reply");
        insert_reply(&pool, community, channel.id, &root, &quiet_reply, false).await;

        let window = get_channel_window(&pool, community, channel.id, 50, None, None)
            .await
            .expect("fetch window");

        let ids: Vec<String> = window
            .rows
            .iter()
            .map(|r| r.stored_event.event.id.to_hex())
            .collect();
        assert!(ids.contains(&root.id.to_hex()), "root is a row");
        assert!(
            ids.contains(&bare.id.to_hex()),
            "metadata-less event is a row"
        );
        assert!(
            ids.contains(&broadcast_reply.id.to_hex()),
            "broadcast depth-1 reply is a row"
        );
        assert!(
            !ids.contains(&quiet_reply.id.to_hex()),
            "ordinary reply must never be a channel row"
        );
        assert!(!window.has_more);
        assert!(window.next_cursor.is_none());
    }

    /// Same-second top-level rows must paginate by the composite
    /// `(created_at, id)` keyset without loss or duplication, chaining each
    /// page from the server-issued `next_cursor` — the exact loop the GUI
    /// runs. A timestamp-only cursor loses every tied row past the first
    /// page; this pins the tiebreak on the window path.
    #[tokio::test]
    #[ignore = "requires Postgres"]
    async fn channel_window_pages_same_second_ties_without_loss() {
        use nostr::Timestamp;

        let pool = setup_pool().await;
        let author = Keys::generate();
        let (channel, community) = create_test_channel(
            &pool,
            &format!("window-ties-{}", Uuid::new_v4()),
            ChannelType::Stream,
            ChannelVisibility::Open,
            None,
            author.public_key().to_bytes().as_slice(),
            None,
        )
        .await
        .expect("create channel");

        let tie_secs = nostr::Timestamp::now().as_secs();
        let row_count = 5usize;
        let mut expected_ids = Vec::with_capacity(row_count);
        for i in 0..row_count {
            let event = EventBuilder::new(Kind::Custom(9), format!("tie-{i}"))
                .custom_created_at(Timestamp::from(tie_secs))
                .sign_with_keys(&author)
                .expect("sign tied event");
            expected_ids.push(event.id.as_bytes().to_vec());
            insert_root(&pool, community, channel.id, &event).await;
        }

        let mut collected: Vec<Vec<u8>> = Vec::new();
        let mut cursor: Option<(DateTime<Utc>, Vec<u8>)> = None;
        loop {
            let window = get_channel_window(&pool, community, channel.id, 2, cursor, None)
                .await
                .expect("fetch window page");
            for row in &window.rows {
                collected.push(row.stored_event.event.id.as_bytes().to_vec());
            }
            if !window.has_more {
                break;
            }
            cursor = Some(window.next_cursor.expect("has_more implies next_cursor"));
        }

        assert_eq!(collected.len(), row_count, "paged set lost or grew rows");
        let mut unique = collected.clone();
        unique.sort();
        unique.dedup();
        assert_eq!(unique.len(), row_count, "paging produced duplicates");
        let mut expected_sorted = expected_ids.clone();
        expected_sorted.sort();
        assert_eq!(unique, expected_sorted, "paged set != inserted tied set");
    }

    /// The exact-multiple final page: when the channel's row count is an
    /// exact multiple of the page limit, the last full page must report
    /// `has_more = false` (from the limit+1 probe) even though it contains
    /// exactly `limit` rows — `rows < limit` proves nothing, and `rows ==
    /// limit` must not imply more. Frozen ruling from the contract thread.
    #[tokio::test]
    #[ignore = "requires Postgres"]
    async fn channel_window_exact_multiple_final_page_reports_exhausted() {
        let pool = setup_pool().await;
        let author = Keys::generate();
        let (channel, community) = create_test_channel(
            &pool,
            &format!("window-exact-{}", Uuid::new_v4()),
            ChannelType::Stream,
            ChannelVisibility::Open,
            None,
            author.public_key().to_bytes().as_slice(),
            None,
        )
        .await
        .expect("create channel");

        // Exactly 4 rows, page limit 2 → two full pages.
        for i in 0..4 {
            let event = make_stream_event(&author, &format!("row-{i}"));
            insert_root(&pool, community, channel.id, &event).await;
        }

        let page1 = get_channel_window(&pool, community, channel.id, 2, None, None)
            .await
            .expect("fetch page 1");
        assert_eq!(page1.rows.len(), 2);
        assert!(page1.has_more, "two more rows exist past page 1");
        let cursor = page1.next_cursor.expect("has_more implies next_cursor");

        let page2 = get_channel_window(&pool, community, channel.id, 2, Some(cursor), None)
            .await
            .expect("fetch page 2");
        assert_eq!(page2.rows.len(), 2, "final page is exactly full");
        assert!(
            !page2.has_more,
            "exact-multiple final page must report exhausted despite rows == limit"
        );
        assert!(page2.next_cursor.is_none(), "no cursor past the last row");
    }

    /// The window is split at the floor: a recent pass that fills the page
    /// and its +1 probe row answers alone; otherwise the older pass reads the
    /// missing rows, so `has_more` still sees rows before the floor.
    #[tokio::test]
    #[ignore = "requires Postgres"]
    async fn channel_window_split_matches_the_unsplit_read() {
        use nostr::Timestamp;

        let pool = setup_pool().await;
        let author = Keys::generate();
        let (channel, community) = create_test_channel(
            &pool,
            &format!("window-recent-{}", Uuid::new_v4()),
            ChannelType::Stream,
            ChannelVisibility::Open,
            None,
            author.public_key().to_bytes().as_slice(),
            None,
        )
        .await
        .expect("create channel");
        let now = Utc::now().timestamp() as u64;
        let floor = crate::store::recent_pass::floor_at(Utc::now()).timestamp() as u64;
        // Two rows inside the recent window and two before it, newest first.
        let mut newest_first = Vec::new();
        for at in [now - 10, now - 20, floor - 86_400, floor - 2 * 86_400] {
            let event = EventBuilder::new(Kind::Custom(9), format!("at {at}"))
                .custom_created_at(Timestamp::from(at))
                .sign_with_keys(&author)
                .expect("sign");
            insert_root(&pool, community, channel.id, &event).await;
            newest_first.push(event.id);
        }

        let mut conn = pool.acquire().await.expect("acquire");
        sqlx::Connection::clear_cached_statements(&mut *conn)
            .await
            .expect("clear prepared statements");
        let prepared = |like: &'static str| {
            sqlx::query_scalar::<_, i64>(
                "SELECT count(*) FROM pg_prepared_statements WHERE statement LIKE $1",
            )
            .bind(like)
        };
        let recent = "%FROM events e%created_at >= '%::timestamptz%";
        let older = "%FROM events e%created_at < '%::timestamptz%";
        let all = "%FROM events e%";
        let ids = |w: &ChannelWindow| {
            w.rows
                .iter()
                .map(|r| r.stored_event.event.id)
                .collect::<Vec<_>>()
        };

        // Limit 1: two recent rows fill the page and its probe row.
        let window = get_channel_window_on(
            &mut conn,
            community,
            channel.id,
            1,
            None,
            None,
            ReadDeadline::PerRead,
        )
        .await
        .expect("recent window");
        assert_eq!(ids(&window), newest_first[..1]);
        assert!(window.has_more);
        let recent_count = prepared(recent).fetch_one(&mut *conn).await.expect("count");
        let all_count = prepared(all).fetch_one(&mut *conn).await.expect("count");
        assert_eq!((recent_count, all_count), (1, 1), "recent pass alone");

        // Limit 2: the recent pass has no probe row, so the older pass reads
        // it and `has_more` sees the rows before the floor.
        let window = get_channel_window_on(
            &mut conn,
            community,
            channel.id,
            2,
            None,
            None,
            ReadDeadline::PerRead,
        )
        .await
        .expect("split window");
        assert_eq!(ids(&window), newest_first[..2]);
        assert!(window.has_more, "older rows lie before the floor");
        let older_count = prepared(older).fetch_one(&mut *conn).await.expect("count");
        let all_count = prepared(all).fetch_one(&mut *conn).await.expect("count");
        assert_eq!((older_count, all_count), (1, 2), "older pass ran");

        // The whole channel, then a cursor page below the floor.
        let window = get_channel_window_on(
            &mut conn,
            community,
            channel.id,
            4,
            None,
            None,
            ReadDeadline::PerRead,
        )
        .await
        .expect("whole window");
        assert_eq!(ids(&window), newest_first);
        assert!(!window.has_more);
        let cursor = get_channel_window_on(
            &mut conn,
            community,
            channel.id,
            2,
            None,
            None,
            ReadDeadline::PerRead,
        )
        .await
        .expect("first page")
        .next_cursor
        .expect("cursor");
        let window = get_channel_window_on(
            &mut conn,
            community,
            channel.id,
            2,
            Some(cursor),
            None,
            ReadDeadline::PerRead,
        )
        .await
        .expect("older page");
        assert_eq!(ids(&window), newest_first[2..]);
        assert!(!window.has_more);

        // A failed older pass fails the window; it never reports the recent
        // rows as the whole channel. Another session locks the partition of a
        // root from before every dated monthly, which only the older pass
        // plans (the recent rows' partition can hold rows below the floor too).
        let ancient = EventBuilder::new(Kind::Custom(9), "ancient")
            .custom_created_at(Timestamp::from(1_748_736_000)) // 2025-06-01
            .sign_with_keys(&author)
            .expect("sign");
        insert_root(&pool, community, channel.id, &ancient).await;
        let partition_of = |id: nostr::EventId| {
            let pool = pool.clone();
            async move {
                sqlx::query_scalar::<_, String>(
                    "SELECT tableoid::regclass::text FROM events WHERE id = $1",
                )
                .bind(id.as_bytes().to_vec())
                .fetch_one(&pool)
                .await
                .expect("partition")
            }
        };
        let old_partition = partition_of(ancient.id).await;
        assert_ne!(old_partition, partition_of(newest_first[0]).await);
        let mut locker = pool.begin().await.expect("begin locker");
        sqlx::query(sqlx::AssertSqlSafe(format!(
            "LOCK TABLE {old_partition} IN ACCESS EXCLUSIVE MODE"
        )))
        .execute(&mut *locker)
        .await
        .expect("lock old partition");
        sqlx::query("SET lock_timeout = '200ms'")
            .execute(&mut *conn)
            .await
            .expect("set lock_timeout");
        let window = get_channel_window_on(
            &mut conn,
            community,
            channel.id,
            1,
            None,
            None,
            ReadDeadline::PerRead,
        )
        .await
        .expect("a full recent window never touches the old partition");
        assert_eq!(ids(&window), newest_first[..1]);
        let failed = get_channel_window_on(
            &mut conn,
            community,
            channel.id,
            2,
            None,
            None,
            ReadDeadline::PerRead,
        )
        .await;
        assert!(
            failed.is_err(),
            "a failed older pass must fail the window, got {:?}",
            failed.map(|w| (w.rows.len(), w.has_more))
        );
        locker.rollback().await.expect("release lock");
        sqlx::query("RESET lock_timeout")
            .execute(&mut *conn)
            .await
            .expect("reset lock_timeout");
    }

    /// The window's passes share one snapshot: a replacement that commits
    /// between them neither drops the row it replaces nor shows the row it
    /// adds.
    #[tokio::test]
    #[ignore = "requires Postgres"]
    async fn channel_window_passes_share_one_snapshot() {
        let pool = setup_pool().await;
        let author = Keys::generate();
        let (channel, community) = create_test_channel(
            &pool,
            &format!("window-snapshot-{}", Uuid::new_v4()),
            ChannelType::Stream,
            ChannelVisibility::Open,
            None,
            author.public_key().to_bytes().as_slice(),
            None,
        )
        .await
        .expect("create channel");
        let channel_id = channel.id;
        let (window, expected) = crate::test_support::replace_between_passes(
            &pool,
            community,
            channel_id,
            move |mut conn| async move {
                get_channel_window_on(
                    &mut conn,
                    community,
                    channel_id,
                    10,
                    None,
                    None,
                    ReadDeadline::PerRead,
                )
                .await
                .expect("split window")
                .rows
                .iter()
                .map(|row| row.stored_event.event.id)
                .collect::<Vec<_>>()
            },
        )
        .await;
        assert_eq!(window, expected);
    }

    /// A split window's passes share one read deadline. Another session holds
    /// the newest partition for a while, so the recent pass spends part of
    /// the deadline waiting, and holds the oldest partition, which only the
    /// older pass plans, for good. The window must be cancelled once the whole
    /// deadline has passed, not a full deadline after its older pass starts.
    #[tokio::test]
    #[ignore = "requires Postgres"]
    async fn channel_window_passes_share_one_deadline() {
        window_cancelled_within_the_read_deadline(Some(std::time::Duration::from_secs(8))).await;
    }

    /// A writer window's first pass already runs under the read deadline, not
    /// the writer session's longer timeout: with the newest partition held
    /// for good, the window is cancelled in its recent pass.
    #[tokio::test]
    #[ignore = "requires Postgres"]
    async fn writer_window_first_pass_runs_under_the_read_deadline() {
        window_cancelled_within_the_read_deadline(None).await;
    }

    /// Read a split window through the production writer entry, whose own
    /// session timeout is longer than the read deadline, while other sessions
    /// lock the newest partition (released after `hold`, or never) and the
    /// oldest one, which only the older pass plans (never released). The
    /// window must be cancelled within the read deadline of its start.
    async fn window_cancelled_within_the_read_deadline(hold: Option<std::time::Duration>) {
        let pool = setup_pool().await;
        let author = Keys::generate();
        let (channel, community) = create_test_channel(
            &pool,
            &format!("window-deadline-{}", Uuid::new_v4()),
            ChannelType::Stream,
            ChannelVisibility::Open,
            None,
            author.public_key().to_bytes().as_slice(),
            None,
        )
        .await
        .expect("create channel");
        let db = crate::Db::from_pool(pool.clone());
        let mut partitions = Vec::new();
        for at in [Utc::now().timestamp() as u64 - 10, 1_748_736_000] {
            let event = EventBuilder::new(Kind::Custom(9), format!("at {at}"))
                .custom_created_at(nostr::Timestamp::from(at))
                .sign_with_keys(&author)
                .expect("sign");
            db.insert_event(community, &event, Some(channel.id))
                .await
                .expect("insert");
            partitions.push(
                sqlx::query_scalar::<_, String>(
                    "SELECT tableoid::regclass::text FROM events WHERE id = $1",
                )
                .bind(event.id.as_bytes().to_vec())
                .fetch_one(&pool)
                .await
                .expect("partition"),
            );
        }
        let [newest, oldest] = <[String; 2]>::try_from(partitions).expect("two partitions");
        assert_ne!(newest, oldest);
        let lock = |partition: String| {
            let pool = pool.clone();
            async move {
                let mut tx = pool.begin().await.expect("begin locker");
                sqlx::query(sqlx::AssertSqlSafe(format!(
                    "LOCK TABLE {partition} IN ACCESS EXCLUSIVE MODE"
                )))
                .execute(&mut *tx)
                .await
                .expect("lock");
                tx
            }
        };
        let mut newest_lock = lock(newest.clone()).await;
        let oldest_lock = lock(oldest).await;

        let started = std::time::Instant::now();
        let window_pool = pool.clone();
        let mut task = tokio::spawn(async move {
            get_channel_window(&window_pool, community, channel.id, 10, None, None).await
        });
        let waiting = tokio::time::timeout(std::time::Duration::from_secs(10), async {
            loop {
                let blocked: bool = sqlx::query_scalar(
                    // `pg_locks` spans every database, and the per-test
                    // databases are clones that share relation OIDs.
                    "SELECT EXISTS(SELECT 1 FROM pg_locks \
                     WHERE relation = $1::text::regclass AND NOT granted \
                     AND database = (SELECT oid FROM pg_database WHERE datname = current_database()))",
                )
                .bind(&newest)
                .fetch_one(&mut *newest_lock)
                .await
                .expect("inspect the recent pass");
                if blocked {
                    break;
                }
                tokio::task::yield_now().await;
            }
        })
        .await;
        assert!(waiting.is_ok(), "the recent pass never waited on the lock");
        // Spend part of the deadline in the recent pass, then let it finish.
        if let Some(hold) = hold {
            tokio::time::sleep(hold).await;
            newest_lock
                .rollback()
                .await
                .expect("release the newest partition");
        }

        let margin = std::time::Duration::from_secs(4);
        let result =
            tokio::time::timeout(crate::event::READ_DEADLINE + margin * 4, &mut task).await;
        let elapsed = started.elapsed();
        drop(oldest_lock);
        let Ok(result) = result else {
            task.abort();
            panic!("the window was never cancelled");
        };
        let err = result
            .expect("window task")
            .expect_err("the window must be cancelled");
        assert!(
            err.is_statement_cancelled(),
            "cancelled by the deadline: {err}"
        );
        assert!(
            elapsed < crate::event::READ_DEADLINE + margin,
            "the window got more than one read deadline; took {elapsed:?}"
        );
    }

    /// Rows with replies carry a thread summary (counts + batched
    /// participants); rows without replies carry none. This is the join that
    /// lets the GUI render thread affordances without a per-root fan-out.
    #[tokio::test]
    #[ignore = "requires Postgres"]
    async fn channel_window_joins_thread_summaries_with_participants() {
        let pool = setup_pool().await;
        let author = Keys::generate();
        let replier = Keys::generate();
        let (channel, community) = create_test_channel(
            &pool,
            &format!("window-summaries-{}", Uuid::new_v4()),
            ChannelType::Stream,
            ChannelVisibility::Open,
            None,
            author.public_key().to_bytes().as_slice(),
            None,
        )
        .await
        .expect("create channel");

        let discussed = make_stream_event(&author, "discussed");
        insert_root(&pool, community, channel.id, &discussed).await;
        let quiet = make_stream_event(&author, "quiet");
        insert_root(&pool, community, channel.id, &quiet).await;

        for i in 0..2 {
            let reply = make_stream_event(&replier, &format!("reply-{i}"));
            insert_reply(&pool, community, channel.id, &discussed, &reply, false).await;
        }

        let window = get_channel_window(&pool, community, channel.id, 50, None, None)
            .await
            .expect("fetch window");

        let discussed_row = window
            .rows
            .iter()
            .find(|r| r.stored_event.event.id == discussed.id)
            .expect("discussed row present");
        let summary = discussed_row
            .thread_summary
            .as_ref()
            .expect("replied row has a summary");
        assert_eq!(summary.reply_count, 2);
        assert!(
            summary
                .participants
                .contains(&replier.public_key().to_bytes().to_vec()),
            "batched participants include the replier"
        );

        let quiet_row = window
            .rows
            .iter()
            .find(|r| r.stored_event.event.id == quiet.id)
            .expect("quiet row present");
        assert!(
            quiet_row.thread_summary.is_none(),
            "reply-less row carries no summary"
        );
    }

    /// Every participant reader (live summary, channel window, deleted-reply
    /// root resolution) aggregates only the newest
    /// [`THREAD_PARTICIPANT_SCAN_LIMIT`] replies, most recent author first,
    /// skipping deleted replies. An author who last posted before that window
    /// drops out; a whole-thread aggregate would still return them.
    #[tokio::test]
    #[ignore = "requires Postgres"]
    async fn participant_summaries_read_only_the_newest_replies() {
        let pool = setup_pool().await;
        let db = Db::from_pool(pool.clone());
        let early = Keys::generate();
        let steady = Keys::generate();
        let deleted_author = Keys::generate();
        let latest = Keys::generate();
        let (channel, community) = create_test_channel(
            &pool,
            &format!("participant-window-{}", Uuid::new_v4()),
            ChannelType::Stream,
            ChannelVisibility::Open,
            None,
            steady.public_key().to_bytes().as_slice(),
            None,
        )
        .await
        .expect("create channel");

        let root = make_stream_event(&steady, "root");
        insert_root(&pool, community, channel.id, &root).await;
        let base = root.created_at.as_secs();
        let reply_at = |keys: &Keys, offset: u64| {
            EventBuilder::new(Kind::Custom(9), format!("reply-{offset}"))
                .custom_created_at(nostr::Timestamp::from(base + offset))
                .sign_with_keys(keys)
                .expect("sign reply")
        };

        // `early` posts first, then `steady` fills the whole window after it.
        let early_reply = reply_at(&early, 1);
        insert_reply(&pool, community, channel.id, &root, &early_reply, false).await;
        let window = THREAD_PARTICIPANT_SCAN_LIMIT as u64;
        for offset in 2..window {
            let reply = reply_at(&steady, offset);
            insert_reply(&pool, community, channel.id, &root, &reply, false).await;
        }
        let deleted_reply = reply_at(&deleted_author, window);
        insert_reply(&pool, community, channel.id, &root, &deleted_reply, false).await;
        let latest_reply = reply_at(&latest, window + 1);
        insert_reply(&pool, community, channel.id, &root, &latest_reply, false).await;
        db.soft_delete_event_and_update_thread(
            community,
            deleted_reply.id.as_bytes(),
            Some(root.id.as_bytes()),
            Some(root.id.as_bytes()),
        )
        .await
        .expect("delete reply");

        let expected = vec![
            latest.public_key().to_bytes().to_vec(),
            steady.public_key().to_bytes().to_vec(),
        ];

        let summary = get_thread_summary(&pool, community, root.id.as_bytes())
            .await
            .expect("thread summary")
            .expect("root has thread metadata");
        assert_eq!(summary.participants, expected, "live summary");

        let window = get_channel_window(&pool, community, channel.id, 50, None, None)
            .await
            .expect("fetch window");
        let row = window
            .rows
            .iter()
            .find(|r| r.stored_event.event.id == root.id)
            .expect("root row present");
        assert_eq!(
            row.thread_summary
                .as_ref()
                .expect("root has a summary")
                .participants,
            expected,
            "channel window"
        );

        let resolved = db
            .resolve_thread_root_summaries(
                community,
                channel.id,
                &[deleted_reply.id.as_bytes().to_vec()],
            )
            .await
            .expect("resolve root");
        assert_eq!(resolved.len(), 1);
        assert_eq!(resolved[0].1.participants, expected, "resolved root");
    }
}