vector_core/db/

events.rs

//! Event storage — save_event for the flat event architecture.

use crate::stored_event::{StoredEvent, event_kind};
use rusqlite::OptionalExtension;
use crate::crypto::maybe_encrypt;
use crate::types::{Message, Attachment, Reaction};

/// Save a StoredEvent to the events table.
///
/// Primary storage function for the flat event architecture.
/// Conditionally encrypts message/edit content based on user setting.
/// Uses INSERT OR REPLACE with COALESCE to preserve existing wrapper_event_id.
/// Conditionally encrypt an event's content per kind (messages/edits are encrypted at rest). Async,
/// so callers run it BEFORE opening a sync transaction.
async fn encrypt_event_content(event: &StoredEvent) -> String {
    if event.kind == event_kind::CHAT_MESSAGE
        || event.kind == event_kind::PRIVATE_DIRECT_MESSAGE
        || event.kind == event_kind::MESSAGE_EDIT
    {
        maybe_encrypt(event.content.clone()).await
    } else {
        event.content.clone()
    }
}

/// Upsert the event row onto the given connection or transaction (so it can commit atomically with
/// its attachment rows). `content` must already be encrypted (see `encrypt_event_content`).
///
/// UPSERT (not INSERT OR REPLACE) so a re-save (reaction/edit) UPDATES in place and PRESERVES the
/// rowid. get_messages_around's (created_at, received_at, rowid) cursor needs a stable final
/// tiebreak to page through same-timestamp bursts; INSERT OR REPLACE churns the rowid and drops rows.
fn insert_event_row(conn: &rusqlite::Connection, event: &StoredEvent, content: &str, tags_json: &str) -> Result<(), String> {
    // prepare_cached: this is the hottest write statement in the app — the cache lives on the
    // connection, so bulk-sync batches and every realtime save skip the SQL re-parse.
    let mut stmt = conn.prepare_cached(
        r#"
        INSERT INTO events (
            id, kind, chat_id, user_id, content, tags, reference_id,
            created_at, received_at, mine, pending, failed, wrapper_event_id, npub, preview_metadata
        ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15)
        ON CONFLICT(id) DO UPDATE SET
            kind = excluded.kind, chat_id = excluded.chat_id, user_id = excluded.user_id,
            content = excluded.content, tags = excluded.tags, reference_id = excluded.reference_id,
            created_at = excluded.created_at, received_at = excluded.received_at,
            mine = excluded.mine, pending = excluded.pending, failed = excluded.failed,
            wrapper_event_id = COALESCE(excluded.wrapper_event_id, events.wrapper_event_id),
            npub = excluded.npub, preview_metadata = excluded.preview_metadata
        "#,
    ).map_err(|e| format!("prepare save event: {}", e))?;
    stmt.execute(
        rusqlite::params![
            event.id, event.kind as i32, event.chat_id, event.user_id, content, tags_json,
            event.reference_id, event.created_at as i64, event.received_at as i64,
            event.mine as i32, event.pending as i32, event.failed as i32,
            event.wrapper_event_id, event.npub, event.preview_metadata,
        ],
    ).map_err(|e| format!("Failed to save event: {}", e))?;
    Ok(())
}

pub async fn save_event(event: &StoredEvent) -> Result<(), String> {
    let tags_json = serde_json::to_string(&event.tags).unwrap_or_else(|_| "[]".to_string());
    let content = encrypt_event_content(event).await;
    let conn = super::get_write_connection_guard_static()?;
    insert_event_row(&conn, event, &content, &tags_json)
}

/// Extract persisted `["bot", npub]` routing tags (the write side lives in
/// `message_to_stored_event`).
fn extract_bot_tags(tags: &[Vec<String>]) -> Vec<String> {
    tags.iter()
        .filter(|t| t.len() >= 2 && t[0] == "bot")
        .map(|t| t[1].clone())
        .collect()
}

/// Parse the NIP-40 `["expiration", <unix secs>]` tag, if present. Drives the
/// self-destruct countdown + purge for messages rehydrated from the DB.
fn extract_expiration_tag(tags: &[Vec<String>]) -> Option<u64> {
    tags.iter()
        .find(|t| t.len() >= 2 && t[0] == "expiration")
        .and_then(|t| t[1].parse::<u64>().ok())
}

/// Check if an event exists in the database.
pub fn event_exists(event_id: &str) -> Result<bool, String> {
    let conn = super::get_db_connection_guard_static()?;
    event_exists_on(&conn, event_id)
}

/// `event_exists` against a caller-held connection or transaction — an in-transaction check
/// sees the batch's own uncommitted rows, which the pooled read connection cannot.
fn event_exists_on(conn: &rusqlite::Connection, event_id: &str) -> Result<bool, String> {
    let mut stmt = conn.prepare_cached("SELECT EXISTS(SELECT 1 FROM events WHERE id = ?1)")
        .map_err(|e| format!("prepare event existence: {}", e))?;
    stmt.query_row(rusqlite::params![event_id], |row| row.get(0))
        .map_err(|e| format!("Failed to check event existence: {}", e))
}

/// Build the kind=7 StoredEvent for a reaction (shared by the single-save and batch paths).
fn reaction_to_stored_event(
    reaction: &Reaction,
    chat_id: i64,
    user_id: Option<i64>,
    mine: bool,
    wrapper_event_id: Option<String>,
) -> StoredEvent {
    // Persist the NIP-30 emoji tag alongside the `e` reference so the
    // image URL survives reload — pure `arrEmojiPacks` lookup would
    // fail when the user hasn't yet opened the picker or unsubscribed.
    let mut tags: Vec<Vec<String>> = vec![
        vec!["e".to_string(), reaction.reference_id.clone()],
    ];
    if let Some(url) = &reaction.emoji_url {
        if reaction.emoji.starts_with(':') && reaction.emoji.ends_with(':') && reaction.emoji.len() >= 3 {
            let shortcode = &reaction.emoji[1..reaction.emoji.len() - 1];
            if !shortcode.is_empty() && !url.is_empty() {
                tags.push(vec!["emoji".to_string(), shortcode.to_string(), url.clone()]);
            }
        }
    }
    StoredEvent {
        id: reaction.id.clone(),
        kind: event_kind::REACTION,
        chat_id,
        user_id,
        content: reaction.emoji.clone(),
        tags,
        reference_id: Some(reaction.reference_id.clone()),
        created_at: std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_secs()).unwrap_or(0),
        received_at: std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_millis() as u64).unwrap_or(0),
        mine,
        pending: false,
        failed: false,
        wrapper_event_id,
        npub: Some(reaction.author_id.clone()),
        preview_metadata: None,
    }
}

/// Save a reaction as a kind=7 event referencing the message.
pub async fn save_reaction_event(
    reaction: &Reaction,
    chat_id: i64,
    user_id: Option<i64>,
    mine: bool,
    wrapper_event_id: Option<String>,
) -> Result<(), String> {
    let event = reaction_to_stored_event(reaction, chat_id, user_id, mine, wrapper_event_id);
    save_event(&event).await
}

// ============================================================================
// save_message — Message → StoredEvent → DB
// ============================================================================

/// Save a single message to the database.
///
/// Converts Message to StoredEvent and saves via the flat event architecture.
/// Also saves reactions as separate kind=7 events.
pub async fn save_message(chat_id: &str, message: &Message) -> Result<(), String> {
    let chat_int_id = super::id_cache::get_or_create_chat_id(chat_id)?;

    let user_int_id = if let Some(ref npub_str) = message.npub {
        super::id_cache::get_or_create_user_id(npub_str)?
    } else {
        None
    };

    let event = message_to_stored_event(message, chat_int_id, user_int_id);

    // Commit the event row and its attachment rows (the dedicated table is the sole source of truth;
    // pre-migration events keep their legacy tag as an un-read fallback) in ONE transaction, so a
    // file message can never persist without its attachments — new events have no tag to fall back
    // on. Encrypt first: encryption is async and can't run inside the sync transaction.
    let tags_json = serde_json::to_string(&event.tags).unwrap_or_else(|_| "[]".to_string());
    let content = encrypt_event_content(&event).await;
    {
        let conn = super::get_write_connection_guard_static()?;
        let tx = conn.unchecked_transaction().map_err(|e| format!("save_message tx: {e}"))?;
        insert_event_row(&tx, &event, &content, &tags_json)?;
        super::attachments::insert_attachment_rows(&tx, &message.id, &message.attachments)?;
        tx.commit().map_err(|e| format!("save_message commit: {e}"))?;
    }

    // Save reactions as separate kind=7 events
    for reaction in &message.reactions {
        if !event_exists(&reaction.id)? {
            let user_id = super::id_cache::get_or_create_user_id(&reaction.author_id)?;
            let is_mine = super::get_current_account()
                .map(|npub| reaction.author_id == npub)
                .unwrap_or(false);
            save_reaction_event(reaction, chat_int_id, user_id, is_mine, None).await?;
        }
    }

    Ok(())
}

/// One fully-prepared batch row: the message, its encrypted event row, its prepared
/// reaction rows, and (DM stream only) the gift-wrap ledger entry that must commit in the
/// SAME transaction — everything phase 2 needs with zero async work and zero id_cache calls.
struct BatchRow<'a> {
    message: &'a Message,
    event: StoredEvent,
    content: String,
    tags_json: String,
    reactions: Vec<(StoredEvent, String)>,
    /// `(wrapper_id_bytes, wrapper_created_at)` — written to `processed_wrappers` only
    /// AFTER this row lands. The ledger is the negentropy fingerprint set: a wrapper
    /// ledgered before its row commits marks the message "have" forever if the row is lost.
    wrapper: Option<([u8; 32], u64)>,
}

/// Phase 1 of a batched save (async): resolve ids, build StoredEvents, encrypt contents.
/// ALL id_cache lookups happen here — get_or_create can write a fresh chat/user row, so it
/// must never run while phase 2 holds the write-connection guard.
async fn prepare_batch_rows<'a>(
    chat_id: &str,
    messages: &[(&'a Message, Option<([u8; 32], u64)>)],
    rows: &mut Vec<BatchRow<'a>>,
) -> Result<(), String> {
    let chat_int_id = super::id_cache::get_or_create_chat_id(chat_id)?;
    let my_npub = super::get_current_account();
    for (message, wrapper) in messages {
        let user_int_id = match &message.npub {
            Some(npub_str) => super::id_cache::get_or_create_user_id(npub_str)?,
            None => None,
        };
        let event = message_to_stored_event(message, chat_int_id, user_int_id);
        let tags_json = serde_json::to_string(&event.tags).unwrap_or_else(|_| "[]".to_string());
        let content = encrypt_event_content(&event).await;
        let mut reactions: Vec<(StoredEvent, String)> = Vec::with_capacity(message.reactions.len());
        for reaction in &message.reactions {
            let user_id = super::id_cache::get_or_create_user_id(&reaction.author_id)?;
            let is_mine = my_npub.as_deref().map(|n| reaction.author_id == n).unwrap_or(false);
            let rev = reaction_to_stored_event(reaction, chat_int_id, user_id, is_mine, None);
            let rtags = serde_json::to_string(&rev.tags).unwrap_or_else(|_| "[]".to_string());
            reactions.push((rev, rtags));
        }
        rows.push(BatchRow { message, event, content, tags_json, reactions, wrapper: *wrapper });
    }
    Ok(())
}

/// Phase 2 of a batched save (sync): ONE transaction for every event + attachment + reaction
/// + wrapper-ledger row. Insert order follows slice order, preserving the rowid tiebreak
/// that same-timestamp pagination depends on. A poison message SKIPS (logged) rather than
/// aborting the batch — one bad row must not lose the other 49. Returns how many messages
/// were written.
fn write_batch_rows(rows: &[BatchRow<'_>]) -> Result<usize, String> {
    let conn = super::get_write_connection_guard_static()?;
    let tx = conn.unchecked_transaction().map_err(|e| format!("batch tx: {e}"))?;
    let mut saved = 0usize;
    for row in rows {
        // Per-row savepoint = save_message's per-message atomicity inside the batch: a
        // failed event/attachment write unwinds THIS row completely — including partial
        // attachment upserts onto a pre-existing row, which a bare DELETE would destroy
        // (a re-saved old file message must keep its download record on a transient error).
        tx.execute_batch("SAVEPOINT batch_row").map_err(|e| format!("batch savepoint: {e}"))?;
        let row_written = insert_event_row(&tx, &row.event, &row.content, &row.tags_json)
            .and_then(|_| super::attachments::insert_attachment_rows(&tx, &row.message.id, &row.message.attachments));
        if let Err(e) = row_written {
            crate::log_warn!("[DB] batch skip {}: {}", &row.message.id[..8.min(row.message.id.len())], e);
            let _ = tx.execute_batch("ROLLBACK TO batch_row; RELEASE batch_row");
            continue;
        }
        saved += 1;
        for (rev, rtags) in &row.reactions {
            // Exists-check ON the tx so a reaction already inserted earlier in this batch dedups
            // (a fresh reaction row must not clobber one that arrived with a wrapper id).
            if event_exists_on(&tx, &rev.id).unwrap_or(true) {
                continue;
            }
            if let Err(e) = insert_event_row(&tx, rev, &rev.content, rtags) {
                crate::log_warn!("[DB] batch reaction {}: {}", &rev.id[..8.min(rev.id.len())], e);
            }
        }
        // Ledger the gift-wrap only now that its row is in the tx — commit lands both or neither.
        if let Some((wrapper_id, wrapper_created_at)) = &row.wrapper {
            let mut stmt = tx.prepare_cached(
                "INSERT OR IGNORE INTO processed_wrappers (wrapper_id, wrapper_created_at, transport) VALUES (?1, ?2, ?3)",
            ).map_err(|e| format!("prepare wrapper ledger: {e}"))?;
            if let Err(e) = stmt.execute(rusqlite::params![
                &wrapper_id[..], *wrapper_created_at as i64, super::wrappers::TRANSPORT_NIP17,
            ]) {
                crate::log_warn!("[DB] batch wrapper ledger {}: {}", &row.message.id[..8.min(row.message.id.len())], e);
            }
        }
        tx.execute_batch("RELEASE batch_row").map_err(|e| format!("batch release: {e}"))?;
    }
    tx.commit().map_err(|e| format!("batch commit: {e}"))?;
    Ok(saved)
}

/// Save many messages for one chat in a SINGLE transaction — the bulk-sync persist path
/// (community backfill pages, negentropy catch-up). One commit amortizes the per-transaction
/// WAL overhead across the whole page instead of paying it per message.
///
/// Structure mirrors `save_message` exactly: contents are encrypted FIRST (encryption is
/// async and can't run inside the sync transaction), then one transaction writes every event
/// row + its attachment rows + its reaction rows (kind-7 content is never encrypted at rest,
/// so reactions are tx-safe).
///
/// `session`: phase 1 awaits through encryption + id resolution, so a swap can land inside
/// it — when provided, the guard is re-checked between the phases and a stale batch is
/// dropped before it can write into the next account's DB.
pub async fn save_messages_batch(
    chat_id: &str,
    messages: &[&Message],
    session: Option<&crate::state::SessionGuard>,
) -> Result<usize, String> {
    if messages.is_empty() {
        return Ok(0);
    }
    let with_wrappers: Vec<(&Message, Option<([u8; 32], u64)>)> =
        messages.iter().map(|m| (*m, None)).collect();
    let mut rows = Vec::with_capacity(messages.len());
    prepare_batch_rows(chat_id, &with_wrappers, &mut rows).await?;
    if session.is_some_and(|s| !s.is_valid()) {
        return Ok(0);
    }
    write_batch_rows(&rows)
}

/// Multi-chat variant for the DM sync stream: gift-wrapped messages span many contacts, and
/// splitting per chat would give back most of the batching win. Each message may carry its
/// gift-wrap ledger entry, committed in the SAME transaction right after its row (see
/// `BatchRow::wrapper`). Groups keep their slice order; everything lands in ONE transaction.
pub async fn save_messages_batch_multi(
    groups: &[(String, Vec<(&Message, Option<([u8; 32], u64)>)>)],
    session: Option<&crate::state::SessionGuard>,
) -> Result<usize, String> {
    let total: usize = groups.iter().map(|(_, m)| m.len()).sum();
    if total == 0 {
        return Ok(0);
    }
    let mut rows = Vec::with_capacity(total);
    for (chat_id, messages) in groups {
        prepare_batch_rows(chat_id, messages, &mut rows).await?;
    }
    if session.is_some_and(|s| !s.is_valid()) {
        return Ok(0);
    }
    write_batch_rows(&rows)
}

/// Convert a Message to a StoredEvent.
fn message_to_stored_event(message: &Message, chat_id: i64, user_id: Option<i64>) -> StoredEvent {
    let kind = if !message.attachments.is_empty() {
        event_kind::FILE_ATTACHMENT
    } else {
        event_kind::PRIVATE_DIRECT_MESSAGE
    };

    let mut tags: Vec<Vec<String>> = Vec::new();

    // Millisecond precision tag
    let ms = message.at % 1000;
    if ms > 0 {
        tags.push(vec!["ms".to_string(), ms.to_string()]);
    }

    // Reply reference
    if !message.replied_to.is_empty() {
        tags.push(vec![
            "e".to_string(),
            message.replied_to.clone(),
            "".to_string(),
            "reply".to_string(),
        ]);
    }

    // Attachments are stored in the dedicated `attachments` table (see save_message), not a tag.

    // NIP-30 emoji tags — persist so reload from DB still renders the
    // custom emoji image instead of the literal `:shortcode:`.
    for et in &message.emoji_tags {
        tags.push(vec!["emoji".to_string(), et.shortcode.clone(), et.url.clone()]);
    }

    // Bot routing targets (npubs) — persist so the passive "ran /cmd with
    // Bot" render survives a reload.
    for npub in &message.addressed_bots {
        tags.push(vec!["bot".to_string(), npub.clone()]);
    }

    // NIP-40 self-destruct expiry — persist so the countdown + purge survive a
    // reload. Rides the same tags column as every other message-shaped tag.
    if let Some(exp) = message.expiration {
        tags.push(vec!["expiration".to_string(), exp.to_string()]);
    }

    let preview_metadata = message.preview_metadata.as_ref()
        .and_then(|m| serde_json::to_string(m).ok());

    StoredEvent {
        id: message.id.clone(),
        kind,
        chat_id,
        user_id,
        content: message.content.clone(),
        tags,
        reference_id: None,
        created_at: message.at / 1000,
        received_at: std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_millis() as u64)
            .unwrap_or(0),
        mine: message.mine,
        pending: message.pending,
        failed: message.failed,
        wrapper_event_id: message.wrapper_event_id.clone(),
        npub: message.npub.clone(),
        preview_metadata,
    }
}

/// Save a PIVX payment event, resolving chat_id from conversation identifier.
pub async fn save_pivx_payment_event(
    conversation_id: &str,
    mut event: StoredEvent,
) -> Result<(), String> {
    event.chat_id = super::id_cache::get_or_create_chat_id(conversation_id)?;
    save_event(&event).await
}

/// Save a system event (member joined/left/removed) with dedup.
/// Returns true if inserted, false if duplicate.
pub async fn save_system_event_by_id(
    event_id: &str,
    conversation_id: &str,
    event_type: crate::stored_event::SystemEventType,
    member_npub: &str,
    member_name: Option<&str>,
) -> Result<bool, String> {
    let now_secs = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs()).unwrap_or(0);
    save_system_event_at(event_id, conversation_id, event_type, member_npub, member_name, now_secs, None, None).await
}

/// Like [`save_system_event_by_id`] but stamps `created_at` from the event's own authenticated timestamp
/// (clamped to not exceed local now, since the inner author sets it) so a HISTORICALLY-synced presence
/// (join/leave) sorts at the time it happened, not at ingest-time now. `received_at` stays local now.
pub async fn save_system_event_at(
    event_id: &str,
    conversation_id: &str,
    event_type: crate::stored_event::SystemEventType,
    member_npub: &str,
    member_name: Option<&str>,
    created_at_secs: u64,
    // Join attribution (public invites): who minted the link the member joined via, and its label.
    // Stored as queryable tags so per-link join counts fall out of a tag scan.
    invited_by: Option<&str>,
    invited_label: Option<&str>,
) -> Result<bool, String> {
    let chat_id = super::id_cache::get_or_create_chat_id(conversation_id)?;

    let now_secs = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs()).unwrap_or(0);
    // Author-set timestamp: clamp forward so a future-dated event can't jump ahead of real activity.
    let created_at = created_at_secs.min(now_secs);

    let display_name = member_name.unwrap_or(member_npub);
    let content = event_type.display_message(display_name);

    let mut tags: Vec<Vec<String>> = vec![
        vec!["d".to_string(), "system-event".to_string()],
        vec!["event-type".to_string(), event_type.as_u8().to_string()],
        vec!["member".to_string(), member_npub.to_string()],
    ];
    if let Some(by) = invited_by {
        tags.push(vec!["invited-by".to_string(), by.to_string()]);
        if let Some(l) = invited_label.filter(|l| !l.is_empty()) {
            tags.push(vec!["invited-label".to_string(), l.to_string()]);
        }
    }
    let tags_json = serde_json::to_string(&tags)
        .map_err(|e| format!("Failed to serialize tags: {}", e))?;

    let conn = super::get_write_connection_guard_static()?;
    let rows = conn.execute(
        r#"INSERT OR IGNORE INTO events (
            id, kind, chat_id, user_id, content, tags, reference_id,
            created_at, received_at, mine, pending, failed, wrapper_event_id, npub
        ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14)"#,
        rusqlite::params![
            event_id,
            event_kind::APPLICATION_SPECIFIC as i32,
            chat_id, None::<i64>, content, tags_json, None::<String>,
            created_at as i64, now_secs as i64,
            0, 0, 0, None::<String>, member_npub,
        ],
    ).map_err(|e| format!("Failed to save system event: {}", e))?;

    Ok(rows > 0)
}

/// Save a message edit as a kind=16 event referencing the original message.
pub async fn save_edit_event(
    edit_id: &str,
    message_id: &str,
    new_content: &str,
    emoji_tags: &[crate::types::EmojiTag],
    chat_id: i64,
    user_id: Option<i64>,
    npub: &str,
) -> Result<(), String> {
    let now = std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH).unwrap();

    // Carry NIP-30 emoji tags so a reload renders the edit's custom emoji image
    // (the reload fold reads the latest edit's tags, not the original message's).
    let mut tags = vec![
        vec!["e".to_string(), message_id.to_string(), "".to_string(), "edit".to_string()],
    ];
    for et in emoji_tags {
        tags.push(vec!["emoji".to_string(), et.shortcode.clone(), et.url.clone()]);
    }

    let event = StoredEvent {
        id: edit_id.to_string(),
        kind: event_kind::MESSAGE_EDIT,
        chat_id,
        user_id,
        content: new_content.to_string(),
        tags,
        reference_id: Some(message_id.to_string()),
        created_at: now.as_secs(),
        received_at: now.as_millis() as u64,
        mine: true,
        pending: false,
        failed: false,
        wrapper_event_id: None,
        npub: Some(npub.to_string()),
        preview_metadata: None,
    };

    save_event(&event).await
}

/// Delete an event from the events table by ID.
pub async fn delete_event(event_id: &str) -> Result<(), String> {
    let conn = super::get_write_connection_guard_static()?;
    // If this row is a chat's read marker, retreat it to the newest surviving event before it FIRST.
    // A deleted marker would leave `last_read` dangling and collapse the unread anchor (badge stuck
    // at 99+). The UPDATE fires only when this chat's marker is exactly the row being deleted.
    if let Ok(Some((chat_row, at))) = conn.query_row(
        "SELECT chat_id, created_at FROM events WHERE id = ?1",
        rusqlite::params![event_id],
        |r| Ok((r.get::<_, i64>(0)?, r.get::<_, i64>(1)?)),
    ).optional() {
        conn.execute(
            "UPDATE chats SET last_read = COALESCE(( \
                 SELECT id FROM events WHERE chat_id = ?1 AND id != ?2 AND created_at <= ?3 \
                 ORDER BY created_at DESC, id DESC LIMIT 1), '') \
             WHERE id = ?1 AND last_read = ?2",
            rusqlite::params![chat_row, event_id, at],
        ).map_err(|e| format!("read-marker retreat: {e}"))?;
    }
    conn.execute(
        "DELETE FROM events WHERE id = ?1",
        rusqlite::params![event_id],
    ).map_err(|e| format!("Failed to delete event: {}", e))?;
    Ok(())
}

/// The stored author (npub) of an event, or `None` if the row (or DB) is absent. Lets the
/// out-of-window moderation-hide path authorize against a paged-out message's real author.
pub fn event_author(event_id: &str) -> Result<Option<String>, String> {
    let conn = match super::get_db_connection_guard_static() {
        Ok(c) => c,
        Err(_) => return Ok(None),
    };
    conn.query_row(
        "SELECT npub FROM events WHERE id = ?1",
        rusqlite::params![event_id],
        |row| row.get::<_, Option<String>>(0),
    )
    .optional()
    .map(|o| o.flatten())
    .map_err(|e| format!("Failed to read event author: {}", e))
}

/// The owning chat identifier, `mine` flag, and stored author (npub) of an event, or
/// `None` if the row (or DB) is absent. Lets delete-affordance resolution give paged-out
/// rows the same verdict as resident ones — residency is a cache detail, not a verdict.
pub fn event_delete_context(event_id: &str) -> Result<Option<(String, bool, Option<String>)>, String> {
    let conn = match super::get_db_connection_guard_static() {
        Ok(c) => c,
        Err(_) => return Ok(None),
    };
    conn.query_row(
        "SELECT c.chat_identifier, e.mine, e.npub \
         FROM events e JOIN chats c ON c.id = e.chat_id \
         WHERE e.id = ?1",
        rusqlite::params![event_id],
        |row| {
            Ok((
                row.get::<_, String>(0)?,
                row.get::<_, i32>(1)? != 0,
                row.get::<_, Option<String>>(2)?,
            ))
        },
    )
    .optional()
    .map_err(|e| format!("Failed to read event delete context: {}", e))
}

/// Check if a message/event exists in the database. Returns false if DB unavailable.
pub fn message_exists_in_db(message_id: &str) -> Result<bool, String> {
    let conn = match super::get_db_connection_guard_static() {
        Ok(c) => c,
        Err(_) => return Ok(false),
    };
    conn.query_row(
        "SELECT EXISTS(SELECT 1 FROM events WHERE id = ?1)",
        rusqlite::params![message_id],
        |row| row.get(0),
    ).map_err(|e| format!("Failed to check event existence: {}", e))
}

/// Check if a wrapper (giftwrap) event ID exists. Returns false if DB unavailable.
pub fn wrapper_event_exists(wrapper_event_id: &str) -> Result<bool, String> {
    let conn = match super::get_db_connection_guard_static() {
        Ok(c) => c,
        Err(_) => return Ok(false),
    };
    conn.query_row(
        "SELECT EXISTS(SELECT 1 FROM events WHERE wrapper_event_id = ?1)",
        rusqlite::params![wrapper_event_id],
        |row| row.get(0),
    ).map_err(|e| format!("Failed to check wrapper event existence: {}", e))
}

/// Update the wrapper event ID for an existing event.
/// Returns true if updated, false if event already had a wrapper_id.
pub fn update_wrapper_event_id(event_id: &str, wrapper_event_id: &str) -> Result<bool, String> {
    let conn = match super::get_write_connection_guard_static() {
        Ok(c) => c,
        Err(_) => return Ok(false),
    };
    let rows = conn.execute(
        "UPDATE events SET wrapper_event_id = ?1 WHERE id = ?2 AND (wrapper_event_id IS NULL OR wrapper_event_id = '')",
        rusqlite::params![wrapper_event_id, event_id],
    ).map_err(|e| format!("Failed to update wrapper event ID: {}", e))?;
    Ok(rows > 0)
}

/// Get message count for a chat.
pub fn get_chat_message_count(chat_id: i64) -> Result<usize, String> {
    let conn = super::get_db_connection_guard_static()?;
    // Must count the SAME kinds get_message_views returns (community chat 9, DM 14, file 15). A
    // narrower set under-counts vs. the rows actually loaded, which latches the frontend cache's
    // `isFullyLoaded` flag true and wedges the local back-pager — community channels then never
    // page DB history past the first screen.
    let count: i64 = conn.query_row(
        &format!(
            "SELECT COUNT(*) FROM events WHERE chat_id = ?1 AND kind IN ({}, {}, {})",
            event_kind::CHAT_MESSAGE, event_kind::PRIVATE_DIRECT_MESSAGE, event_kind::FILE_ATTACHMENT
        ),
        rusqlite::params![chat_id],
        |row| row.get(0),
    ).map_err(|e| format!("Failed to count messages: {}", e))?;
    Ok(count as usize)
}

/// Get PIVX payment events for a chat.
pub fn get_pivx_payments_for_chat(conversation_id: &str) -> Result<Vec<StoredEvent>, String> {
    let conn = super::get_db_connection_guard_static()?;
    let chat_id: i64 = conn.query_row(
        "SELECT id FROM chats WHERE chat_identifier = ?1",
        rusqlite::params![conversation_id], |row| row.get(0)
    ).map_err(|_| "Chat not found")?;

    let mut stmt = conn.prepare(
        "SELECT id, kind, chat_id, user_id, content, tags, reference_id, \
         created_at, received_at, mine, pending, failed, wrapper_event_id, npub \
         FROM events WHERE chat_id = ?1 AND kind = ?2 ORDER BY created_at ASC, received_at ASC"
    ).map_err(|e| format!("Failed to prepare: {}", e))?;

    let rows = stmt.query_map(
        rusqlite::params![chat_id, event_kind::APPLICATION_SPECIFIC as i32],
        |row| {
            let tags_json: String = row.get(5)?;
            let tags: Vec<Vec<String>> = serde_json::from_str(&tags_json).unwrap_or_default();
            Ok(StoredEvent {
                id: row.get(0)?, kind: row.get::<_, i32>(1)? as u16,
                chat_id: row.get(2)?, user_id: row.get(3)?, content: row.get(4)?,
                tags, reference_id: row.get(6)?,
                created_at: row.get::<_, i64>(7)? as u64, received_at: row.get::<_, i64>(8)? as u64,
                mine: row.get::<_, i32>(9)? != 0, pending: row.get::<_, i32>(10)? != 0,
                failed: row.get::<_, i32>(11)? != 0, wrapper_event_id: row.get(12)?,
                npub: row.get(13)?, preview_metadata: None,
            })
        }
    ).map_err(|e| format!("Failed to query: {}", e))?;

    let mut payments = Vec::new();
    for row in rows {
        let event = row.map_err(|e| format!("Failed to read event: {}", e))?;
        if event.tags.iter().any(|t| t.len() >= 2 && t[0] == "d" && t[1] == "pivx-payment") {
            payments.push(event);
        }
    }
    Ok(payments)
}

/// Get system events (member joined/left) for a chat.
pub fn get_system_events_for_chat(conversation_id: &str) -> Result<Vec<StoredEvent>, String> {
    let conn = super::get_db_connection_guard_static()?;
    let chat_id: i64 = conn.query_row(
        "SELECT id FROM chats WHERE chat_identifier = ?1",
        rusqlite::params![conversation_id], |row| row.get(0)
    ).map_err(|_| "Chat not found")?;

    let mut stmt = conn.prepare(
        "SELECT id, kind, chat_id, user_id, content, tags, reference_id, \
         created_at, received_at, mine, pending, failed, wrapper_event_id, npub \
         FROM events WHERE chat_id = ?1 AND kind = ?2 ORDER BY created_at ASC, received_at ASC"
    ).map_err(|e| format!("Failed to prepare: {}", e))?;

    let rows = stmt.query_map(
        rusqlite::params![chat_id, event_kind::APPLICATION_SPECIFIC as i32],
        |row| {
            let tags_json: String = row.get(5)?;
            let tags: Vec<Vec<String>> = serde_json::from_str(&tags_json).unwrap_or_default();
            Ok(StoredEvent {
                id: row.get(0)?, kind: row.get::<_, i32>(1)? as u16,
                chat_id: row.get(2)?, user_id: row.get(3)?, content: row.get(4)?,
                tags, reference_id: row.get(6)?,
                created_at: row.get::<_, i64>(7)? as u64, received_at: row.get::<_, i64>(8)? as u64,
                mine: row.get::<_, i32>(9)? != 0, pending: row.get::<_, i32>(10)? != 0,
                failed: row.get::<_, i32>(11)? != 0, wrapper_event_id: row.get(12)?,
                npub: row.get(13)?, preview_metadata: None,
            })
        }
    ).map_err(|e| format!("Failed to query: {}", e))?;

    let mut events = Vec::new();
    for row in rows {
        let event = row.map_err(|e| format!("Failed to read event: {}", e))?;
        if event.tags.iter().any(|t| t.len() >= 2 && t[0] == "d" && t[1] == "system-event") {
            events.push(event);
        }
    }
    Ok(events)
}

// ============================================================================
// Event Read Operations
// ============================================================================

/// Helper to parse a SQLite row into a StoredEvent.
fn parse_event_row(row: &rusqlite::Row) -> rusqlite::Result<StoredEvent> {
    let tags_json: String = row.get(5)?;
    let tags: Vec<Vec<String>> = serde_json::from_str(&tags_json).unwrap_or_default();

    Ok(StoredEvent {
        id: row.get(0)?,
        kind: row.get::<_, i32>(1)? as u16,
        chat_id: row.get(2)?,
        user_id: row.get(3)?,
        content: row.get(4)?,
        tags,
        reference_id: row.get(6)?,
        created_at: row.get::<_, i64>(7)? as u64,
        received_at: row.get::<_, i64>(8)? as u64,
        mine: row.get::<_, i32>(9)? != 0,
        pending: row.get::<_, i32>(10)? != 0,
        failed: row.get::<_, i32>(11)? != 0,
        wrapper_event_id: row.get(12)?,
        npub: row.get(13)?,
        preview_metadata: row.get(14)?,
    })
}

/// Get events for a chat with pagination, optionally filtered by kind.
/// Message/edit content is decrypted via maybe_decrypt.
pub async fn get_events(
    chat_id: i64,
    kinds: Option<&[u16]>,
    limit: usize,
    offset: usize,
) -> Result<Vec<StoredEvent>, String> {
    let events: Vec<StoredEvent> = {
        let conn = super::get_db_connection_guard_static()?;

        if let Some(k) = kinds {
            let kind_placeholders: String = (0..k.len())
                .map(|i| format!("?{}", i + 2))
                .collect::<Vec<_>>()
                .join(",");
            let limit_param = k.len() + 2;
            let offset_param = k.len() + 3;

            let sql = format!(
                "SELECT id, kind, chat_id, user_id, content, tags, reference_id, \
                 created_at, received_at, mine, pending, failed, wrapper_event_id, npub, preview_metadata \
                 FROM events WHERE chat_id = ?1 AND kind IN ({}) \
                 ORDER BY created_at DESC, received_at DESC \
                 LIMIT ?{} OFFSET ?{}",
                kind_placeholders, limit_param, offset_param
            );

            let mut stmt = conn.prepare(&sql)
                .map_err(|e| format!("Failed to prepare events query: {}", e))?;

            match k.len() {
                1 => {
                    let rows = stmt.query_map(
                        rusqlite::params![chat_id, k[0] as i32, limit as i64, offset as i64],
                        parse_event_row
                    ).map_err(|e| format!("Failed to query events: {}", e))?;
                    rows.filter_map(|r| r.ok()).collect()
                },
                2 => {
                    let rows = stmt.query_map(
                        rusqlite::params![chat_id, k[0] as i32, k[1] as i32, limit as i64, offset as i64],
                        parse_event_row
                    ).map_err(|e| format!("Failed to query events: {}", e))?;
                    rows.filter_map(|r| r.ok()).collect()
                },
                3 => {
                    let rows = stmt.query_map(
                        rusqlite::params![chat_id, k[0] as i32, k[1] as i32, k[2] as i32, limit as i64, offset as i64],
                        parse_event_row
                    ).map_err(|e| format!("Failed to query events: {}", e))?;
                    rows.filter_map(|r| r.ok()).collect()
                },
                _ => return Err("Unsupported number of kinds".to_string()),
            }
        } else {
            let mut stmt = conn.prepare(
                "SELECT id, kind, chat_id, user_id, content, tags, reference_id, \
                 created_at, received_at, mine, pending, failed, wrapper_event_id, npub, preview_metadata \
                 FROM events WHERE chat_id = ?1 \
                 ORDER BY created_at DESC, received_at DESC \
                 LIMIT ?2 OFFSET ?3"
            ).map_err(|e| format!("Failed to prepare events query: {}", e))?;

            let rows = stmt.query_map(
                rusqlite::params![chat_id, limit as i64, offset as i64],
                parse_event_row
            ).map_err(|e| format!("Failed to query events: {}", e))?;
            rows.filter_map(|r| r.ok()).collect()
        }
    };

    // Decrypt message content
    let mut decrypted = Vec::with_capacity(events.len());
    for mut event in events {
        if event.kind == event_kind::CHAT_MESSAGE || event.kind == event_kind::PRIVATE_DIRECT_MESSAGE {
            event.content = crate::crypto::maybe_decrypt(event.content).await
                .unwrap_or_else(|_| "[Decryption failed]".to_string());
        }
        decrypted.push(event);
    }

    Ok(decrypted)
}

/// Get events that reference specific message IDs (reactions, edits).
pub async fn get_related_events(
    reference_ids: &[String],
) -> Result<Vec<StoredEvent>, String> {
    if reference_ids.is_empty() {
        return Ok(Vec::new());
    }

    let conn = super::get_db_connection_guard_static()?;

    let placeholders: String = reference_ids.iter().map(|_| "?").collect::<Vec<_>>().join(",");
    let sql = format!(
        "SELECT id, kind, chat_id, user_id, content, tags, reference_id, \
         created_at, received_at, mine, pending, failed, wrapper_event_id, npub, preview_metadata \
         FROM events WHERE reference_id IN ({}) \
         ORDER BY created_at ASC, received_at ASC",
        placeholders
    );

    let mut stmt = conn.prepare(&sql)
        .map_err(|e| format!("Failed to prepare related events query: {}", e))?;

    let params: Vec<&dyn rusqlite::ToSql> = reference_ids.iter()
        .map(|s| s as &dyn rusqlite::ToSql)
        .collect();

    let events: Vec<StoredEvent> = stmt.query_map(params.as_slice(), parse_event_row)
        .map_err(|e| format!("Failed to query related events: {}", e))?
        .filter_map(|r| r.ok())
        .collect();

    Ok(events)
}

/// Context data for a replied-to message.
pub struct ReplyContext {
    pub content: String,
    pub npub: Option<String>,
    pub has_attachment: bool,
    /// Extension of the attachment, when the replied-to message is a file, so
    /// the reply quote can label the type even when the target is off-screen.
    pub extension: Option<String>,
}

/// Fetch reply context for a list of message IDs.
pub async fn get_reply_contexts(
    message_ids: &[String],
) -> Result<std::collections::HashMap<String, ReplyContext>, String> {
    use std::collections::HashMap;

    if message_ids.is_empty() {
        return Ok(HashMap::new());
    }

    let (events, edits): (Vec<(String, i32, String, Option<String>, Option<String>)>, Vec<(String, String)>) = {
        let conn = super::get_db_connection_guard_static()?;

        let placeholders: String = (0..message_ids.len())
            .map(|i| format!("?{}", i + 1))
            .collect::<Vec<_>>()
            .join(",");

        // Query original messages (tags carry the file-type/name for attachment quotes)
        let sql = format!(
            "SELECT id, kind, content, npub, tags FROM events WHERE id IN ({})",
            placeholders
        );
        let mut stmt = conn.prepare(&sql)
            .map_err(|e| format!("Failed to prepare reply context query: {}", e))?;

        let params: Vec<&str> = message_ids.iter().map(|s| s.as_str()).collect();
        let params_dyn: Vec<&dyn rusqlite::ToSql> = params.iter().map(|s| s as &dyn rusqlite::ToSql).collect();

        let rows = stmt.query_map(params_dyn.as_slice(), |row| {
            Ok((row.get::<_, String>(0)?, row.get::<_, i32>(1)?,
                row.get::<_, String>(2)?, row.get::<_, Option<String>>(3)?,
                row.get::<_, Option<String>>(4)?))
        }).map_err(|e| format!("Failed to query reply contexts: {}", e))?;
        let events_result: Vec<_> = rows.filter_map(|r| r.ok()).collect();
        drop(stmt);

        // Query latest edits
        let edit_sql = format!(
            "SELECT reference_id, content FROM events \
             WHERE kind = {} AND reference_id IN ({}) \
             ORDER BY created_at DESC, received_at DESC",
            event_kind::MESSAGE_EDIT, placeholders
        );
        let mut edit_stmt = conn.prepare(&edit_sql)
            .map_err(|e| format!("Failed to prepare edit query: {}", e))?;
        let edit_rows = edit_stmt.query_map(params_dyn.as_slice(), |row| {
            Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?))
        }).map_err(|e| format!("Failed to query edits: {}", e))?;
        let edits_result: Vec<_> = edit_rows.filter_map(|r| r.ok()).collect();

        (events_result, edits_result)
    };

    // Build latest edit map (first = most recent since ordered DESC)
    let mut latest_edits: HashMap<String, String> = HashMap::new();
    for (ref_id, content) in edits {
        latest_edits.entry(ref_id).or_insert(content);
    }

    // Batch the file replies' attachments (one query, not one per reply) for the extension below.
    let file_ids: Vec<String> = events.iter()
        .filter(|(_, kind, _, _, _)| *kind == event_kind::FILE_ATTACHMENT as i32)
        .map(|(id, _, _, _, _)| id.clone())
        .collect();
    let atts_by_event = super::attachments::get_attachments_for_events(&file_ids).unwrap_or_default();

    // Decrypt and build contexts
    let mut contexts = HashMap::new();
    for (id, kind, original_content, npub, tags) in events {
        let has_attachment = kind == event_kind::FILE_ATTACHMENT as i32;
        let content_to_decrypt = latest_edits.get(&id).cloned().unwrap_or(original_content);

        let decrypted_content = if kind == event_kind::CHAT_MESSAGE as i32
            || kind == event_kind::PRIVATE_DIRECT_MESSAGE as i32
        {
            crate::crypto::maybe_decrypt(content_to_decrypt).await
                .unwrap_or_else(|_| "[Decryption failed]".to_string())
        } else {
            String::new()
        };

        // The first attachment's extension lets the quote show the file type. From the table, with a
        // legacy-tag fallback for an un-backfilled pre-migration row.
        let extension = if has_attachment {
            atts_by_event.get(&id)
                .and_then(|atts| atts.first())
                .map(|a| a.extension.to_lowercase())
                .filter(|e| !e.is_empty())
                .or_else(|| tags.as_deref()
                    .and_then(|t| serde_json::from_str::<Vec<Vec<String>>>(t).ok())
                    .and_then(|parsed| parsed.into_iter()
                        .find(|t| t.first().map(|k| k == "attachments").unwrap_or(false))
                        .and_then(|t| t.into_iter().nth(1)))
                    .and_then(|json| serde_json::from_str::<Vec<serde_json::Value>>(&json).ok())
                    .and_then(|atts| atts.into_iter().next())
                    .and_then(|a| a.get("extension").and_then(|e| e.as_str()).map(str::to_lowercase))
                    .filter(|e| !e.is_empty()))
        } else {
            None
        };

        contexts.insert(id, ReplyContext { content: decrypted_content, npub, has_attachment, extension });
    }

    Ok(contexts)
}

/// Populate reply context for a PAGE of messages in one query — the sync/back-page
/// counterpart to [`populate_reply_context`]. Every path that emits a message straight to
/// the UI must resolve the quote first: the frontend renders the emitted payload, and a
/// reply whose parent lies outside the rendered window has no other source for it (a
/// parent inside the window is resolved from memory instead). Messages whose parent isn't
/// persisted yet are left untouched.
pub async fn populate_reply_contexts(messages: Vec<&mut Message>) -> Result<(), String> {
    let ids: Vec<String> = messages
        .iter()
        .filter(|m| !m.replied_to.is_empty())
        .map(|m| m.replied_to.clone())
        .collect();
    if ids.is_empty() {
        return Ok(());
    }
    let contexts = get_reply_contexts(&ids).await?;
    for message in messages {
        if let Some(ctx) = contexts.get(&message.replied_to) {
            message.replied_to_content = Some(ctx.content.clone());
            message.replied_to_npub = ctx.npub.clone();
            message.replied_to_has_attachment = Some(ctx.has_attachment);
            message.replied_to_attachment_extension = ctx.extension.clone();
        }
    }
    Ok(())
}

/// Populate reply context for a single message.
/// Used for real-time messages that don't go through get_message_views.
pub async fn populate_reply_context(message: &mut Message) -> Result<(), String> {
    if message.replied_to.is_empty() {
        return Ok(());
    }

    let contexts = get_reply_contexts(&[message.replied_to.clone()]).await?;

    if let Some(ctx) = contexts.get(&message.replied_to) {
        message.replied_to_content = Some(ctx.content.clone());
        message.replied_to_npub = ctx.npub.clone();
        message.replied_to_has_attachment = Some(ctx.has_attachment);
        message.replied_to_attachment_extension = ctx.extension.clone();
    }

    Ok(())
}

/// Whether `event_id` is one of our own messages (`mine = 1`). A reply to our
/// own message is an implicit ping, so notifications treat it like a direct
/// @mention (breaks through a muted channel). Missing row → not ours → false.
pub fn is_own_event(event_id: &str) -> bool {
    let Ok(conn) = super::get_db_connection_guard_static() else {
        return false;
    };
    conn.query_row(
        "SELECT mine FROM events WHERE id = ?1",
        [event_id],
        |row| row.get::<_, i64>(0),
    )
    .map(|mine| mine == 1)
    .unwrap_or(false)
}

// ============================================================================
// Message Views — compose full Messages from events + reactions + edits
// ============================================================================

/// Extract a single tag value from raw tags JSON without full allocation.
fn extract_tag_from_json(tags_json: &str, key: &str) -> Option<String> {
    if tags_json.len() <= 2 { return None; }
    let pattern = format!("[\"{}\"", key);
    if !tags_json.contains(&pattern) { return None; }
    let tags: Vec<Vec<String>> = serde_json::from_str(tags_json).ok()?;
    tags.into_iter()
        .find(|tag| tag.first().map(|s| s.as_str()) == Some(key))
        .and_then(|tag| tag.into_iter().nth(1))
}


/// A stored reaction author written as 64-char hex (an early v2 ingest) reads
/// back as the npub the frontend contract expects — self-heals old rows with no
/// migration; a bech32 or unknown value passes through untouched.
fn normalize_reaction_author(author: String) -> String {
    if author.len() == 64 && author.bytes().all(|b| b.is_ascii_hexdigit()) {
        if let Ok(pk) = nostr_sdk::prelude::PublicKey::from_hex(&author) {
            use nostr_sdk::prelude::ToBech32;
            let Ok(npub) = pk.to_bech32();
            return npub;
        }
    }
    author
}
/// Extract the NIP-30 `["emoji", shortcode, url]` URL from a stored
/// reaction's tags. The reaction's content must be `:shortcode:` form
/// and the matching tag's shortcode must agree — otherwise we get the
/// URL of a stray emoji tag that doesn't actually represent the
/// reaction's chosen emoji.
fn extract_reaction_emoji_url(tags: &[Vec<String>], content: &str) -> Option<String> {
    if !content.starts_with(':') || !content.ends_with(':') || content.len() < 3 {
        return None;
    }
    let sc = &content[1..content.len() - 1];
    tags.iter().find_map(|t| {
        if t.len() >= 3 && t[0] == "emoji" && t[1] == sc {
            Some(t[2].clone())
        } else {
            None
        }
    })
}

/// Extract a NIP-10 reply reference ("e" tag with "reply" marker at position 3).
fn extract_reply_tag_from_json(tags_json: &str) -> Option<String> {
    if tags_json.len() <= 2 { return None; }
    if !tags_json.contains("[\"e\"") { return None; }
    let tags: Vec<Vec<String>> = serde_json::from_str(tags_json).ok()?;
    tags.into_iter()
        .find(|tag| {
            tag.first().map(|s| s.as_str()) == Some("e")
                && tag.get(3).map(|s| s.as_str()) == Some("reply")
        })
        .and_then(|tag| tag.into_iter().nth(1))
}

/// Get message events with reactions, edits, and attachments composed.
///
/// This is the main "get messages" function. Queries events, fetches related
/// reactions/edits, parses attachments, applies edits, resolves reply context.
pub async fn get_message_views(
    chat_id: i64,
    limit: usize,
    offset: usize,
) -> Result<Vec<Message>, String> {
    // Step 1: Get message events (kind 9, 14, 15)
    let message_kinds = [event_kind::CHAT_MESSAGE, event_kind::PRIVATE_DIRECT_MESSAGE, event_kind::FILE_ATTACHMENT];
    let message_events = get_events(chat_id, Some(&message_kinds), limit, offset).await?;

    compose_message_views(message_events).await
}

/// Compose Message views from already-fetched message events (kind 9/14/15):
/// fetch related reactions/edits, parse attachments, apply edits, resolve reply
/// context. Shared by `get_message_views` (offset pager) and `get_messages_around`
/// (anchored window). Input order is preserved in the output.
async fn compose_message_views(message_events: Vec<StoredEvent>) -> Result<Vec<Message>, String> {
    use std::collections::HashMap;

    if message_events.is_empty() {
        return Ok(Vec::new());
    }

    // Step 2: Get related events (reactions, edits)
    let message_ids: Vec<String> = message_events.iter().map(|e| e.id.clone()).collect();
    let related_events = get_related_events(&message_ids).await?;

    let mut reactions_by_msg: HashMap<String, Vec<Reaction>> = HashMap::new();
    let mut edits_by_msg: HashMap<String, Vec<(u64, String, Vec<crate::types::EmojiTag>)>> = HashMap::new();

    for event in related_events {
        if let Some(ref_id) = &event.reference_id {
            match event.kind {
                k if k == event_kind::REACTION => {
                    let emoji_url = extract_reaction_emoji_url(&event.tags, &event.content);
                    reactions_by_msg.entry(ref_id.clone()).or_default().push(Reaction {
                        id: event.id.clone(),
                        reference_id: ref_id.clone(),
                        author_id: normalize_reaction_author(event.npub.clone().unwrap_or_default()),
                        emoji: event.content.clone(),
                        emoji_url,
                    });
                }
                k if k == event_kind::MESSAGE_EDIT => {
                    let decrypted = crate::crypto::maybe_decrypt(event.content.clone()).await
                        .unwrap_or_else(|_| event.content.clone());
                    let edit_emoji = crate::types::EmojiTag::extract_from_stored(&event.tags);
                    edits_by_msg.entry(ref_id.clone()).or_default().push((event.created_at * 1000, decrypted, edit_emoji));
                }
                _ => {}
            }
        }
    }

    for edits in edits_by_msg.values_mut() {
        edits.sort_by_key(|(ts, _, _)| *ts);
    }

    // Step 3: Attachments from the dedicated table (batched), with a legacy-tag fallback for any
    // file event not represented in the table (an un-backfilled pre-migration row).
    let attach_ids: Vec<String> = message_events.iter()
        .filter(|e| e.kind == event_kind::FILE_ATTACHMENT || e.kind == event_kind::CHAT_MESSAGE)
        .map(|e| e.id.clone())
        .collect();
    let mut attachments_by_msg = super::attachments::get_attachments_for_events(&attach_ids)
        .unwrap_or_default();
    for event in &message_events {
        if event.kind != event_kind::FILE_ATTACHMENT && event.kind != event_kind::CHAT_MESSAGE {
            continue;
        }
        if attachments_by_msg.contains_key(&event.id) {
            continue;
        }
        if let Some(json) = event.get_tag("attachments") {
            if let Ok(atts) = serde_json::from_str::<Vec<Attachment>>(json) {
                if !atts.is_empty() {
                    attachments_by_msg.insert(event.id.clone(), atts);
                }
            }
        }
    }

    // Step 4: Compose Message structs
    let mut messages = Vec::with_capacity(message_events.len());
    for event in message_events {
        let replied_to = event.get_reply_reference().unwrap_or("").to_string();
        let at = event.timestamp_ms();
        let reactions = reactions_by_msg.remove(&event.id).unwrap_or_default();
        let attachments = attachments_by_msg.remove(&event.id).unwrap_or_default();

        let original_content = if event.kind == event_kind::FILE_ATTACHMENT {
            String::new()
        } else {
            event.content.clone()
        };

        // Edits carry their own emoji tags; the newest edit's tags win so the
        // displayed (latest) content renders its custom emoji, not the original's.
        let original_emoji = crate::types::EmojiTag::extract_from_stored(&event.tags);
        let (content, edited, edit_history, emoji_tags) = if let Some(edits) = edits_by_msg.remove(&event.id) {
            let mut history = Vec::with_capacity(edits.len() + 1);
            history.push(crate::types::EditEntry { content: original_content.clone(), edited_at: at });
            for (ts, c, _) in &edits {
                history.push(crate::types::EditEntry { content: c.clone(), edited_at: *ts });
            }
            let (latest, latest_emoji) = edits.last()
                .map(|(_, c, e)| (c.clone(), e.clone()))
                .unwrap_or_else(|| (original_content.clone(), original_emoji.clone()));
            (latest, true, Some(history), latest_emoji)
        } else {
            (original_content, false, None, original_emoji)
        };

        let preview_metadata = event.preview_metadata
            .and_then(|json| serde_json::from_str(&json).ok());

        let addressed_bots = extract_bot_tags(&event.tags);
        let expiration = extract_expiration_tag(&event.tags);
        messages.push(Message {
            expiration,
            id: event.id, content, replied_to,
            replied_to_content: None, replied_to_npub: None, replied_to_has_attachment: None,
            replied_to_attachment_extension: None,
            preview_metadata, attachments, reactions, at,
            pending: event.pending, failed: event.failed, mine: event.mine,
            npub: event.npub, wrapper_event_id: event.wrapper_event_id,
            edited, edit_history,
            emoji_tags,
            addressed_bots,
        });
    }

    // Rows whose NIP-40 expiry passed while out of STATE (app closed, chat
    // unopened) must never reach a renderer — strip them here, at the single
    // DB→Message chokepoint, and purge their remnants in the background.
    crate::self_destruct::strip_expired(&mut messages);

    // Step 5: Reply context
    let reply_ids: Vec<String> = messages.iter()
        .filter(|m| !m.replied_to.is_empty())
        .map(|m| m.replied_to.clone())
        .collect();

    if !reply_ids.is_empty() {
        let contexts = get_reply_contexts(&reply_ids).await?;
        for msg in &mut messages {
            if let Some(ctx) = contexts.get(&msg.replied_to) {
                msg.replied_to_content = Some(ctx.content.clone());
                msg.replied_to_npub = ctx.npub.clone();
                msg.replied_to_has_attachment = Some(ctx.has_attachment);
                msg.replied_to_attachment_extension = ctx.extension.clone();
            }
        }
    }

    Ok(messages)
}

/// Anchored (random-access) message window: load `before` messages up to and
/// including the anchor, plus `after` messages strictly newer than it. O(window)
/// regardless of how deep the anchor sits in the chat — unlike the offset pager,
/// which is O(depth) to reach a far-back message.
///
/// Returns ASC by `created_at` (oldest first), composed with reactions/edits/
/// attachments. Errs if the anchor id isn't in the DB so the caller can fall back.
pub async fn get_messages_around(
    chat_id: i64,
    anchor_id: &str,
    before: usize,
    after: usize,
) -> Result<Vec<Message>, String> {
    let message_kinds = [event_kind::CHAT_MESSAGE, event_kind::PRIVATE_DIRECT_MESSAGE, event_kind::FILE_ATTACHMENT];

    let message_events: Vec<StoredEvent> = {
        let conn = super::get_db_connection_guard_static()?;

        // Resolve the anchor's FULL sort key (created_at, received_at, rowid). Paging by created_at
        // alone wedges on a wall of equal timestamps (a message burst): the query keeps returning the
        // same newest-N of the cluster, so back-paging stalls before reaching older history. The
        // (received_at, rowid) tiebreak — rowid being the unique final key — gives a strict total
        // order, so every page steps strictly past the previous, through any same-timestamp cluster.
        let (anchor_at, anchor_rt, anchor_rowid): (i64, i64, i64) = conn.query_row(
            "SELECT created_at, received_at, rowid FROM events WHERE id = ?1",
            rusqlite::params![anchor_id],
            |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
        ).map_err(|e| format!("Anchor message not found: {}", e))?;

        // Kinds occupy ?2..?4; then ?5 created_at, ?6 received_at, ?7 rowid, ?8 limit.
        let kind_placeholders: String = (0..message_kinds.len())
            .map(|i| format!("?{}", i + 2))
            .collect::<Vec<_>>()
            .join(",");
        let cols = "id, kind, chat_id, user_id, content, tags, reference_id, \
                    created_at, received_at, mine, pending, failed, wrapper_event_id, npub, preview_metadata";

        // Older incl. anchor: strict key <= anchor key; newest-first then reverse to ASC.
        let older_sql = format!(
            "SELECT {} FROM events WHERE chat_id = ?1 AND kind IN ({}) \
             AND (created_at < ?5 OR (created_at = ?5 AND (received_at < ?6 \
                  OR (received_at = ?6 AND rowid <= ?7)))) \
             ORDER BY created_at DESC, received_at DESC, rowid DESC LIMIT ?8",
            cols, kind_placeholders
        );
        let mut older_stmt = conn.prepare(&older_sql)
            .map_err(|e| format!("Failed to prepare older window query: {}", e))?;
        let older_rows = older_stmt.query_map(
            rusqlite::params![
                chat_id,
                message_kinds[0] as i32, message_kinds[1] as i32, message_kinds[2] as i32,
                anchor_at, anchor_rt, anchor_rowid, before as i64
            ],
            parse_event_row,
        ).map_err(|e| format!("Failed to query older window: {}", e))?;
        let mut older: Vec<StoredEvent> = older_rows.filter_map(|r| r.ok()).collect();
        older.reverse(); // DESC -> ASC

        // Newer: strictly after the anchor key.
        let newer_sql = format!(
            "SELECT {} FROM events WHERE chat_id = ?1 AND kind IN ({}) \
             AND (created_at > ?5 OR (created_at = ?5 AND (received_at > ?6 \
                  OR (received_at = ?6 AND rowid > ?7)))) \
             ORDER BY created_at ASC, received_at ASC, rowid ASC LIMIT ?8",
            cols, kind_placeholders
        );
        let mut newer_stmt = conn.prepare(&newer_sql)
            .map_err(|e| format!("Failed to prepare newer window query: {}", e))?;
        let newer_rows = newer_stmt.query_map(
            rusqlite::params![
                chat_id,
                message_kinds[0] as i32, message_kinds[1] as i32, message_kinds[2] as i32,
                anchor_at, anchor_rt, anchor_rowid, after as i64
            ],
            parse_event_row,
        ).map_err(|e| format!("Failed to query newer window: {}", e))?;
        let newer: Vec<StoredEvent> = newer_rows.filter_map(|r| r.ok()).collect();

        older.into_iter().chain(newer).collect()
    };

    // Decrypt message content (mirror get_events).
    let mut decrypted = Vec::with_capacity(message_events.len());
    for mut event in message_events {
        if event.kind == event_kind::CHAT_MESSAGE || event.kind == event_kind::PRIVATE_DIRECT_MESSAGE {
            event.content = crate::crypto::maybe_decrypt(event.content).await
                .unwrap_or_else(|_| "[Decryption failed]".to_string());
        }
        decrypted.push(event);
    }

    compose_message_views(decrypted).await
}

/// Get the last message for ALL chats in a single batch query.
/// Optimized for app startup (chat list sidebar).
pub async fn get_all_chats_last_messages() -> Result<std::collections::HashMap<String, Vec<Message>>, String> {
    use std::collections::HashMap;

    // Step 1: Query last message per chat via correlated subquery
    let message_events: Vec<(String, StoredEvent, String)> = {
        let conn = super::get_db_connection_guard_static()?;
        let mut stmt = conn.prepare(
            "SELECT c.chat_identifier, \
             e.id, e.kind, e.chat_id, e.user_id, e.content, e.tags, e.reference_id, \
             e.created_at, e.received_at, e.mine, e.pending, e.failed, e.wrapper_event_id, e.npub, e.preview_metadata \
             FROM chats c JOIN events e ON e.rowid = ( \
                 SELECT e2.rowid FROM events e2 WHERE e2.chat_id = c.id \
                 AND e2.kind IN (?1, ?2, ?3) \
                 ORDER BY e2.created_at DESC, e2.received_at DESC LIMIT 1) \
             WHERE c.chat_type != 1"
        ).map_err(|e| format!("Failed to prepare: {}", e))?;

        let rows = stmt.query_map(
            rusqlite::params![
                event_kind::CHAT_MESSAGE as i32,
                event_kind::PRIVATE_DIRECT_MESSAGE as i32,
                event_kind::FILE_ATTACHMENT as i32
            ],
            |row| {
                let chat_id: String = row.get(0)?;
                let tags_json: String = row.get(6)?;
                let event = StoredEvent {
                    id: row.get(1)?, kind: row.get::<_, i32>(2)? as u16,
                    chat_id: row.get(3)?, user_id: row.get(4)?, content: row.get(5)?,
                    tags: Vec::new(), // Deferred — parsed on-demand
                    reference_id: row.get(7)?,
                    created_at: row.get::<_, i64>(8)? as u64, received_at: row.get::<_, i64>(9)? as u64,
                    mine: row.get::<_, i32>(10)? != 0, pending: row.get::<_, i32>(11)? != 0,
                    failed: row.get::<_, i32>(12)? != 0, wrapper_event_id: row.get(13)?,
                    npub: row.get(14)?, preview_metadata: row.get(15)?,
                };
                Ok((chat_id, event, tags_json))
            }
        ).map_err(|e| format!("Failed to query: {}", e))?;
        rows.filter_map(|r| r.ok()).collect()
    };

    if message_events.is_empty() {
        return Ok(HashMap::new());
    }

    // Step 2: Related events (reactions, edits)
    let message_ids: Vec<String> = message_events.iter().map(|(_, e, _)| e.id.clone()).collect();
    let related_events = get_related_events(&message_ids).await?;

    let mut reactions_by_msg: HashMap<String, Vec<Reaction>> = HashMap::new();
    let mut edits_by_msg: HashMap<String, Vec<(u64, String, Vec<crate::types::EmojiTag>)>> = HashMap::new();

    for event in related_events {
        if let Some(ref_id) = &event.reference_id {
            match event.kind {
                k if k == event_kind::REACTION => {
                    let emoji_url = extract_reaction_emoji_url(&event.tags, &event.content);
                    reactions_by_msg.entry(ref_id.clone()).or_default().push(Reaction {
                        id: event.id.clone(), reference_id: ref_id.clone(),
                        author_id: normalize_reaction_author(event.npub.clone().unwrap_or_default()),
                        emoji: event.content.clone(),
                        emoji_url,
                    });
                }
                k if k == event_kind::MESSAGE_EDIT => {
                    let decrypted = crate::crypto::maybe_decrypt(event.content.clone()).await
                        .unwrap_or_else(|_| event.content.clone());
                    let edit_emoji = crate::types::EmojiTag::extract_from_stored(&event.tags);
                    edits_by_msg.entry(ref_id.clone()).or_default().push((event.created_at * 1000, decrypted, edit_emoji));
                }
                _ => {}
            }
        }
    }
    for edits in edits_by_msg.values_mut() {
        edits.sort_by_key(|(ts, _, _)| *ts);
    }

    // Step 3: Attachments from the dedicated table (batched), with a legacy-tag fallback.
    let attach_ids: Vec<String> = message_events.iter()
        .filter(|(_, e, _)| e.kind == event_kind::FILE_ATTACHMENT || e.kind == event_kind::CHAT_MESSAGE)
        .map(|(_, e, _)| e.id.clone())
        .collect();
    let mut attachments_by_msg = super::attachments::get_attachments_for_events(&attach_ids)
        .unwrap_or_default();
    for (_, event, tags_json) in &message_events {
        if event.kind != event_kind::FILE_ATTACHMENT && event.kind != event_kind::CHAT_MESSAGE {
            continue;
        }
        if attachments_by_msg.contains_key(&event.id) {
            continue;
        }
        if let Some(val) = extract_tag_from_json(tags_json, "attachments") {
            if let Ok(atts) = serde_json::from_str::<Vec<Attachment>>(&val) {
                if !atts.is_empty() {
                    attachments_by_msg.insert(event.id.clone(), atts);
                }
            }
        }
    }

    // Step 4: Compose Messages grouped by chat_identifier
    let mut result: HashMap<String, Vec<Message>> = HashMap::new();

    for (chat_identifier, event, tags_json) in message_events {
        let reactions = reactions_by_msg.remove(&event.id).unwrap_or_default();
        let attachments = attachments_by_msg.remove(&event.id).unwrap_or_default();
        let replied_to = extract_reply_tag_from_json(&tags_json).unwrap_or_default();

        // Decrypt content
        let original_content = if event.kind == event_kind::CHAT_MESSAGE
            || event.kind == event_kind::PRIVATE_DIRECT_MESSAGE
        {
            crate::crypto::maybe_decrypt(event.content.clone()).await
                .unwrap_or_else(|_| "[Decryption failed]".to_string())
        } else {
            String::new()
        };

        let stored_tags = serde_json::from_str::<Vec<Vec<String>>>(&tags_json).unwrap_or_default();
        let original_emoji = crate::types::EmojiTag::extract_from_stored(&stored_tags);
        let addressed_bots = extract_bot_tags(&stored_tags);
        let expiration = extract_expiration_tag(&stored_tags);
        // Newest edit's emoji tags win so the latest content renders correctly.
        let (content, edited, edit_history, emoji_tags) = if let Some(edits) = edits_by_msg.remove(&event.id) {
            let (latest, latest_emoji) = edits.last()
                .map(|(_, c, e)| (c.clone(), e.clone()))
                .unwrap_or_else(|| (original_content.clone(), original_emoji.clone()));
            let history: Vec<crate::types::EditEntry> = std::iter::once(crate::types::EditEntry {
                content: original_content, edited_at: event.created_at * 1000,
            }).chain(edits.into_iter().map(|(ts, c, _)| crate::types::EditEntry { content: c, edited_at: ts }))
            .collect();
            (latest, true, Some(history), latest_emoji)
        } else {
            (original_content, false, None, original_emoji)
        };

        let preview_metadata = event.preview_metadata
            .and_then(|json| serde_json::from_str(&json).ok());

        result.entry(chat_identifier).or_default().push(Message {
            expiration,
            id: event.id, content, replied_to,
            replied_to_content: None, replied_to_npub: None, replied_to_has_attachment: None,
            replied_to_attachment_extension: None,
            preview_metadata, attachments, reactions, at: event.created_at * 1000,
            pending: event.pending, failed: event.failed, mine: event.mine,
            npub: event.npub, wrapper_event_id: event.wrapper_event_id,
            edited, edit_history,
            emoji_tags,
            addressed_bots,
        });
    }

    // Step 5: Reply context — the openChat pre-paint renders this boot last-message
    // synchronously (before the richer get_message_views load lands), so without
    // context here a reply shows its quote only on the second open.
    let reply_ids: Vec<String> = result.values()
        .flatten()
        .filter(|m| !m.replied_to.is_empty())
        .map(|m| m.replied_to.clone())
        .collect();

    if !reply_ids.is_empty() {
        let contexts = get_reply_contexts(&reply_ids).await?;
        for msg in result.values_mut().flatten() {
            if let Some(ctx) = contexts.get(&msg.replied_to) {
                msg.replied_to_content = Some(ctx.content.clone());
                msg.replied_to_npub = ctx.npub.clone();
                msg.replied_to_has_attachment = Some(ctx.has_attachment);
                msg.replied_to_attachment_extension = ctx.extension.clone();
            }
        }
    }

    // An expired self-destruct as a chat's last message must not flash in the
    // chat list — strip + background-purge; the preview shows empty until the
    // next boot resolves the prior message, same as a mid-session sweep.
    for msgs in result.values_mut() {
        crate::self_destruct::strip_expired(msgs);
    }

    Ok(result)
}

/// Per-chat unread count, computed straight from the DB so it's correct even when only the last
/// message is in RAM (the boot state). Mirrors the in-memory walk-back exactly: unread = non-mine
/// messages newer than the most recent "anchor" (our own message OR the `last_read` marker,
/// whichever is latest). A never-read chat (empty `last_read`, no own message) counts all its
/// non-mine messages. Returns `chat_identifier → count`; chats with 0 unread are omitted.
/// Muted/blocked filtering is left to the caller (it lives in RAM state, cheaply).
pub async fn unread_counts() -> Result<std::collections::HashMap<String, u32>, String> {
    let conn = super::get_db_connection_guard_static()?;
    // Anchor computed once per chat in the CTE so the count scan doesn't re-derive it per row. The
    // anchor filter rides the LEFT JOIN's ON clause so a never-read chat still yields a row (anchor
    // 0 via COALESCE) and counts all its messages; in WHERE it would drop those chats. The
    // `last_read` anchor is kind-agnostic: a "read to here" marker can land on a system event (kind
    // 30078) and must still cut the count by its timestamp. Only the own-message anchor is kind-filtered.
    let mut stmt = conn
        .prepare(
            "WITH anchors AS ( \
                SELECT c.id AS chat_id, c.chat_identifier AS chat_identifier, \
                       COALESCE(MAX(e.created_at), 0) AS anchor_ts \
                FROM chats c \
                LEFT JOIN events e ON e.chat_id = c.id \
                  AND ((e.mine = 1 AND e.kind IN (?1, ?2, ?3)) OR e.id = c.last_read) \
                GROUP BY c.id \
             ) \
             SELECT a.chat_identifier, COUNT(*) AS unread \
             FROM events e JOIN anchors a ON a.chat_id = e.chat_id \
             WHERE e.kind IN (?1, ?2, ?3) AND e.mine = 0 AND e.created_at > a.anchor_ts \
             GROUP BY a.chat_identifier",
        )
        .map_err(|e| format!("prepare unread_counts: {e}"))?;
    let rows = stmt
        .query_map(
            rusqlite::params![
                event_kind::CHAT_MESSAGE as i32,
                event_kind::PRIVATE_DIRECT_MESSAGE as i32,
                event_kind::FILE_ATTACHMENT as i32
            ],
            |row| Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)? as u32)),
        )
        .map_err(|e| format!("query unread_counts: {e}"))?;
    let mut out = std::collections::HashMap::new();
    for r in rows.flatten() {
        out.insert(r.0, r.1);
    }
    Ok(out)
}

/// Unread count for a SINGLE chat, same semantics as [`unread_counts`]. The RAM cache calls this to
/// reconcile one chat (open / delete / retreat) without recomputing every chat's count.
pub async fn unread_count_for_chat(chat_identifier: &str) -> Result<u32, String> {
    let conn = super::get_db_connection_guard_static()?;
    let count: i64 = conn
        .query_row(
            "SELECT COUNT(*) FROM events e JOIN chats c ON e.chat_id = c.id \
             WHERE c.chat_identifier = ?4 AND e.kind IN (?1, ?2, ?3) AND e.mine = 0 \
               AND e.created_at > COALESCE(( \
                     SELECT MAX(e2.created_at) FROM events e2 \
                     WHERE e2.chat_id = c.id \
                       AND ((e2.mine = 1 AND e2.kind IN (?1, ?2, ?3)) OR e2.id = c.last_read)), 0)",
            rusqlite::params![
                event_kind::CHAT_MESSAGE as i32,
                event_kind::PRIVATE_DIRECT_MESSAGE as i32,
                event_kind::FILE_ATTACHMENT as i32,
                chat_identifier
            ],
            |row| row.get(0),
        )
        .map_err(|e| format!("query unread_count_for_chat: {e}"))?;
    Ok(count as u32)
}

/// What [`compute_unread_anchor`] decided a chat's read marker should become to surface its newest
/// contact message as unread. Computed from the full DB history (RAM may hold only a preview
/// message for an unopened community).
#[derive(Debug, PartialEq)]
pub enum UnreadMark {
    /// Nothing to surface: no contact message, or a strictly-newer own message (we spoke last).
    NoOp,
    /// Reset to the never-read anchor: the target is the chat's earliest message.
    Clear,
    /// Retreat `last_read` to this event id (the newest message in a strictly earlier second).
    Anchor(String),
}

/// Decide how to mark `chat_identifier` unread. Anchors on the newest message strictly before the
/// newest contact message's second — the count query compares whole seconds with a strict `>`, so a
/// same-second anchor would leave the target on the boundary and it would read as caught-up.
pub async fn compute_unread_anchor(chat_identifier: &str) -> Result<UnreadMark, String> {
    let conn = super::get_db_connection_guard_static()?;
    let (k0, k1, k2) = (
        event_kind::CHAT_MESSAGE as i32,
        event_kind::PRIVATE_DIRECT_MESSAGE as i32,
        event_kind::FILE_ATTACHMENT as i32,
    );
    // Newest non-mine message second (the target) and newest overall (to detect we spoke last).
    let (target_ts, newest_ts): (Option<i64>, Option<i64>) = conn
        .query_row(
            "SELECT MAX(CASE WHEN e.mine = 0 THEN e.created_at END), MAX(e.created_at) \
             FROM events e JOIN chats c ON e.chat_id = c.id \
             WHERE c.chat_identifier = ?1 AND e.kind IN (?2, ?3, ?4)",
            rusqlite::params![chat_identifier, k0, k1, k2],
            |row| Ok((row.get(0)?, row.get(1)?)),
        )
        .map_err(|e| format!("unread anchor target: {e}"))?;

    let target_ts = match target_ts {
        Some(t) => t,
        None => return Ok(UnreadMark::NoOp), // no contact message to surface
    };
    if newest_ts.map_or(false, |n| n > target_ts) {
        return Ok(UnreadMark::NoOp); // a strictly-newer own message → we spoke last
    }

    let anchor_id: Option<String> = conn
        .query_row(
            "SELECT e.id FROM events e JOIN chats c ON e.chat_id = c.id \
             WHERE c.chat_identifier = ?1 AND e.kind IN (?2, ?3, ?4) AND e.created_at < ?5 \
             ORDER BY e.created_at DESC LIMIT 1",
            rusqlite::params![chat_identifier, k0, k1, k2, target_ts],
            |row| row.get(0),
        )
        .optional()
        .map_err(|e| format!("unread anchor prev: {e}"))?;

    Ok(match anchor_id {
        Some(id) => UnreadMark::Anchor(id),
        None => UnreadMark::Clear,
    })
}

/// Drain a sync loop's pending message batch into one transaction. The shared flush for the
/// segment-flush pattern: bulk loops COLLECT message saves and call this at delete barriers +
/// loop end (a batched save committing AFTER a delete it originally preceded would resurrect
/// the deleted row — flushing first preserves wire order). Session-guarded HERE so every bulk
/// path gets the same swap-safety: on a stale session the batch is DROPPED, never written
/// into the next account's DB (the caller's loop is about to bail anyway).
pub async fn flush_message_batch(
    chat_id: &str,
    pending: &mut Vec<&Message>,
    session: &crate::state::SessionGuard,
) {
    if pending.is_empty() {
        return;
    }
    if !session.is_valid() {
        pending.clear();
        return;
    }
    if let Err(e) = save_messages_batch(chat_id, pending, Some(session)).await {
        crate::log_warn!("[DB] batch flush failed for {}: {}", chat_id, e);
    }
    pending.clear();
}

/// Batch save messages for a chat — one transaction for the whole slice.
pub async fn save_chat_messages(chat_id: &str, messages: &[Message]) -> Result<(), String> {
    if messages.is_empty() {
        return Ok(());
    }
    let refs: Vec<&Message> = messages.iter().collect();
    save_messages_batch(chat_id, &refs, None).await.map(|_| ())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::stored_event::SystemEventType;

    static TEST_COUNTER: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(71000);

    fn make_test_npub(n: u32) -> String {
        const BECH32: &[u8] = b"qpzry9x8gf2tvdw0s3jn54khce6mua7l";
        let mut payload = vec![b'q'; 58];
        let mut x = n as u64;
        let mut i = 58;
        while x > 0 && i > 0 {
            i -= 1;
            payload[i] = BECH32[(x as usize) % 32];
            x /= 32;
        }
        format!("npub1{}", std::str::from_utf8(&payload).unwrap())
    }

    fn init_test_db() -> (tempfile::TempDir, std::sync::MutexGuard<'static, ()>) {
        let guard = crate::db::DB_TEST_GUARD.lock().unwrap_or_else(|e| e.into_inner());
        crate::db::close_database();
        // Each test rebinds to a fresh per-account DB; the row-id caches are per-account, so a stale
        // entry (e.g. a shared author npub) would point into the prior test's DB and FK-fail the insert.
        crate::db::clear_id_caches();
        let tmp = tempfile::tempdir().unwrap();
        let n = TEST_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
        let account = make_test_npub(n);
        std::fs::create_dir_all(tmp.path().join(&account)).unwrap();
        crate::db::set_app_data_dir(tmp.path().to_path_buf());
        crate::db::set_current_account(account.clone()).unwrap();
        crate::db::init_database(&account).unwrap();
        (tmp, guard)
    }

    fn now_secs() -> u64 {
        std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs()
    }

    /// A page of synced replies must come out of the batch resolver carrying their quotes.
    /// The sync/promote paths emit straight to the UI, and the renderer falls back to the
    /// in-memory parent only when the parent is inside the rendered window — so a reply to
    /// an OLDER message renders with no quote at all unless the backend resolved it here.
    /// Reopening the chat re-reads through `get_message_views` (which resolves), which is
    /// why the bug looked like "replies lose their context until you reopen".
    #[tokio::test]
    async fn batch_resolver_fills_quotes_for_a_synced_page() {
        let (_tmp, _guard) = init_test_db();
        let chat = "channel_reply_batch";
        let author = make_test_npub(90_001);

        // An older, already-persisted parent (the case with no in-memory fallback).
        let mut parent = Message::default();
        parent.id = "parent_evt".to_string();
        parent.content = "the original message".to_string();
        parent.npub = Some(author.clone());
        parent.at = now_secs() - 5_000;
        save_message(chat, &parent).await.unwrap();

        // The freshly-synced page: a reply to it, plus an unrelated message.
        let mut reply = Message::default();
        reply.id = "reply_evt".to_string();
        reply.content = "replying now".to_string();
        reply.replied_to = "parent_evt".to_string();
        reply.at = now_secs();
        let mut plain = Message::default();
        plain.id = "plain_evt".to_string();
        plain.content = "unrelated".to_string();
        plain.at = now_secs();
        // A reply whose parent this device has never seen must stay empty, not fabricate a quote.
        let mut orphan = Message::default();
        orphan.id = "orphan_evt".to_string();
        orphan.replied_to = "never_seen_evt".to_string();
        orphan.at = now_secs();

        populate_reply_contexts(vec![&mut reply, &mut plain, &mut orphan]).await.unwrap();

        assert_eq!(
            reply.replied_to_content.as_deref(),
            Some("the original message"),
            "the synced reply carries its parent's content"
        );
        assert_eq!(reply.replied_to_npub.as_deref(), Some(author.as_str()), "and its parent's author");
        assert!(plain.replied_to_content.is_none(), "a non-reply is untouched");
        assert!(orphan.replied_to_content.is_none(), "an unresolvable parent leaves the quote empty");
    }

    /// An empty page (or one with no replies at all) must not query or error — the sync path
    /// calls this for every page, most of which carry no replies.
    #[tokio::test]
    async fn batch_resolver_is_a_noop_without_replies() {
        let (_tmp, _guard) = init_test_db();
        populate_reply_contexts(vec![]).await.unwrap();
        let mut plain = Message::default();
        plain.id = "solo".to_string();
        populate_reply_contexts(vec![&mut plain]).await.unwrap();
        assert!(plain.replied_to_content.is_none());
    }

    // C-H2: a presence (join/leave) persisted from HISTORY must keep its authenticated timestamp so it
    // sorts where it happened, not at ingest-time "now"; a future-dated one is clamped so it can't jump
    // ahead of real activity.
    #[tokio::test]
    async fn system_event_stamps_authenticated_time_clamped_to_now() {
        let (_tmp, _guard) = init_test_db();
        let chat = "channel_ch2_timestamp";
        let before = now_secs();
        let past = before - 100_000;

        save_system_event_at("ev_past", chat, SystemEventType::MemberJoined, "npubX", None, past, None, None).await.unwrap();
        save_system_event_at("ev_future", chat, SystemEventType::MemberJoined, "npubX", None, before + 100_000, None, None).await.unwrap();
        let after = now_secs();

        let evs = get_system_events_for_chat(chat).unwrap();
        let past_ev = evs.iter().find(|e| e.id == "ev_past").expect("past event saved");
        assert_eq!(past_ev.created_at, past, "historical join keeps its real (authenticated) timestamp");

        let fut_ev = evs.iter().find(|e| e.id == "ev_future").expect("future event saved");
        assert!(fut_ev.created_at >= before && fut_ev.created_at <= after,
            "future-dated event clamped to local now ({} not in {}..={})", fut_ev.created_at, before, after);
    }

    // Delete-affordance resolution must work on paged-out rows: the events table is the
    // fallback source for (chat, mine, author) when a message isn't STATE-resident.
    #[tokio::test]
    async fn event_delete_context_resolves_from_db() {
        let (_tmp, _guard) = init_test_db();
        let chat = "npub1contactdc";

        let mine_msg = Message { id: "dc_mine".into(), content: "x".into(), at: 1_000, mine: true, ..Default::default() };
        let theirs = Message {
            id: "dc_theirs".into(), content: "y".into(), at: 2_000, mine: false,
            npub: Some("npub1sender".to_string()),
            ..Default::default()
        };
        save_message(chat, &mine_msg).await.unwrap();
        save_message(chat, &theirs).await.unwrap();

        let (chat_id, mine, _author) = event_delete_context("dc_mine").unwrap().expect("own row resolves");
        assert_eq!(chat_id, chat);
        assert!(mine);

        let (chat_id, mine, author) = event_delete_context("dc_theirs").unwrap().expect("contact row resolves");
        assert_eq!(chat_id, chat);
        assert!(!mine);
        assert_eq!(author.as_deref(), Some("npub1sender"));

        assert!(event_delete_context("dc_absent").unwrap().is_none(), "unknown id is None, not an error");
    }

    // The reported bug: unread must accumulate across a restart (when only the last message per
    // chat is in RAM). The DB cutoff count must mirror the in-memory walk-back exactly.
    #[tokio::test]
    async fn unread_counts_match_walk_back_semantics() {
        let (_tmp, _guard) = init_test_db();
        let chat = "npub1contactdm";
        // `at` is ms (created_at = at/1000); use distinct seconds.
        let mk = |id: &str, secs: u64, mine: bool| Message {
            id: id.into(), content: "x".into(), at: secs * 1000, mine,
            npub: (!mine).then(|| "npub1sender".to_string()),
            ..Default::default()
        };
        let unread = || async { unread_counts().await.unwrap().get(chat).copied().unwrap_or(0) };

        // 6 contact messages, never opened/read, no own reply → all 6 unread.
        for i in 0..6u64 {
            save_message(chat, &mk(&format!("m{i}"), 1000 + i, false)).await.unwrap();
        }
        assert_eq!(unread().await, 6, "never-read backlog counts all 6");

        // 2 more arrive → 8, NOT replaced by 2 (the exact reported symptom).
        save_message(chat, &mk("m6", 2000, false)).await.unwrap();
        save_message(chat, &mk("m7", 2001, false)).await.unwrap();
        assert_eq!(unread().await, 8, "6 backlog + 2 new = 8");

        // Our own reply clears it (walk-back stops at the newest mine).
        save_message(chat, &mk("mine", 2002, true)).await.unwrap();
        assert_eq!(unread().await, 0, "own message = read up to here");

        // A contact message after our send is unread again.
        save_message(chat, &mk("m8", 2003, false)).await.unwrap();
        assert_eq!(unread().await, 1, "one new after our send");

        // last_read marker advances the cutoff just like an own message.
        {
            let conn = crate::db::get_write_connection_guard_static().unwrap();
            conn.execute(
                "UPDATE chats SET last_read = ?1 WHERE chat_identifier = ?2",
                rusqlite::params!["m8", chat],
            ).unwrap();
        }
        assert_eq!(unread().await, 0, "last_read=m8 clears all");
        save_message(chat, &mk("m9", 2004, false)).await.unwrap();
        assert_eq!(unread().await, 1, "one arrival after last_read");
    }

    // The single-chat reconcile query must agree with the full map for every state the cache
    // reconciles from (never-read, own-reply cutoff, last_read marker).
    #[tokio::test]
    async fn unread_count_for_chat_matches_the_map() {
        let (_tmp, _guard) = init_test_db();
        let chat = "npub1reconcile";
        let mk = |id: &str, secs: u64, mine: bool| Message {
            id: id.into(), content: "x".into(), at: secs * 1000, mine,
            npub: (!mine).then(|| "npub1sender".to_string()),
            ..Default::default()
        };
        let agree = || async {
            let map = unread_counts().await.unwrap().get(chat).copied().unwrap_or(0);
            let one = unread_count_for_chat(chat).await.unwrap();
            assert_eq!(map, one, "single-chat query diverged from the map");
            one
        };

        for i in 0..4u64 { save_message(chat, &mk(&format!("m{i}"), 1000 + i, false)).await.unwrap(); }
        assert_eq!(agree().await, 4);
        save_message(chat, &mk("mine", 1010, true)).await.unwrap();
        assert_eq!(agree().await, 0);
        save_message(chat, &mk("after", 1011, false)).await.unwrap();
        assert_eq!(agree().await, 1);
        {
            let conn = crate::db::get_write_connection_guard_static().unwrap();
            conn.execute("UPDATE chats SET last_read = ?1 WHERE chat_identifier = ?2",
                rusqlite::params!["after", chat]).unwrap();
        }
        assert_eq!(agree().await, 0);
        // A chat with nothing at all reconciles to 0 (no row in the map).
        assert_eq!(unread_count_for_chat("npub1nonexistent").await.unwrap(), 0);
    }

    #[tokio::test]
    async fn attachments_table_round_trip_dedup_and_cascade() {
        let (_tmp, _guard) = init_test_db();
        // downloaded:false mirrors a freshly-received attachment (Attachment::default is downloaded:true).
        let att = |id: &str, name: &str| Attachment {
            id: id.into(), url: format!("https://blossom/{id}"), name: name.into(),
            extension: "png".into(), size: 42, downloaded: false, ..Default::default()
        };
        let msg = |mid: &str, secs: u64, atts: Vec<Attachment>| Message {
            id: mid.into(), content: String::new(), at: secs * 1000, mine: false,
            npub: Some("npub1sender".into()), attachments: atts, ..Default::default()
        };

        // Save a message with two attachments → both rows, order preserved by att_index.
        save_message("npub1att", &msg("m1", 1000, vec![att("hashA", "a.png"), att("hashB", "b.png")])).await.unwrap();
        let got = crate::db::attachments::get_attachments_for_event("m1").unwrap();
        assert_eq!(got.len(), 2);
        assert_eq!((got[0].id.as_str(), got[1].id.as_str()), ("hashA", "hashB"), "att_index order");
        assert_eq!(got[0].name, "a.png");
        assert_eq!(got[0].size, 42);
        assert!(!got[0].downloaded);

        // Single-row download flip (no read-modify-write of a blob).
        crate::db::attachments::set_attachment_downloaded("m1", "hashA", true, "/tmp/a.png").unwrap();
        let got = crate::db::attachments::get_attachments_for_event("m1").unwrap();
        assert!(got[0].downloaded && got[0].path == "/tmp/a.png");
        assert!(!got[1].downloaded, "sibling attachment untouched");

        // Dedup: a second message shares hashA → backfill-by-hash marks it (indexed, no LIKE scan).
        save_message("npub1att", &msg("m2", 1001, vec![att("hashA", "a-again.png")])).await.unwrap();
        let affected = crate::db::attachments::backfill_downloaded_by_hash("hashA", "/tmp/a.png", "m1").unwrap();
        assert_eq!(affected, vec!["m2".to_string()]);
        assert!(crate::db::attachments::get_attachments_for_event("m2").unwrap()[0].downloaded);

        // The write funnel dual-populates the legacy tag, so get_message_views (still tag-backed
        // during shadow-populate) composes the attachments onto the Message.
        let chat_int = crate::db::id_cache::get_or_create_chat_id("npub1att").unwrap();
        let views = get_message_views(chat_int, 10, 0).await.unwrap();
        let m1 = views.iter().find(|m| m.id == "m1").unwrap();
        assert_eq!(m1.attachments.len(), 2);

        // Cascade: deleting the event removes its attachment rows.
        delete_event("m1").await.unwrap();
        assert!(crate::db::attachments::get_attachments_for_event("m1").unwrap().is_empty(), "ON DELETE CASCADE");
    }

    // The download persist path: a re-save must never DOWNGRADE download state (relay re-delivery),
    // but a re-save carrying a completed download (+ the nonce→content-hash id rewrite) must persist.
    #[tokio::test]
    async fn attachment_download_state_is_monotonic_across_resaves() {
        let (_tmp, _guard) = init_test_db();
        let att = |id: &str, downloaded: bool, path: &str| Attachment {
            id: id.into(), url: "u".into(), name: "f.png".into(), extension: "png".into(),
            size: 1, downloaded, path: path.into(), ..Default::default()
        };
        let msg = |atts: Vec<Attachment>| Message {
            id: "dl1".into(), content: String::new(), at: 1_000_000, mine: false,
            npub: Some("npub1s".into()), attachments: atts, ..Default::default()
        };

        // Receive (not downloaded), then the user downloads → single-row flip.
        save_message("npub1dl", &msg(vec![att("nonceid", false, "")])).await.unwrap();
        crate::db::attachments::set_attachment_downloaded("dl1", "nonceid", true, "/tmp/f.png").unwrap();

        // Relay re-delivery (downloaded=false) must NOT reset the download.
        save_message("npub1dl", &msg(vec![att("nonceid", false, "")])).await.unwrap();
        let got = crate::db::attachments::get_attachments_for_event("dl1").unwrap();
        assert!(got[0].downloaded && got[0].path == "/tmp/f.png", "re-delivery preserves the download");

        // Post-download re-save: id rewritten nonce→content-hash, downloaded=true persists in one pass.
        save_message("npub1dl", &msg(vec![att("contenthash", true, "/tmp/f.png")])).await.unwrap();
        let got = crate::db::attachments::get_attachments_for_event("dl1").unwrap();
        assert_eq!(got[0].id, "contenthash", "hash rewritten nonce→content");
        assert!(got[0].downloaded && got[0].path == "/tmp/f.png");

        // A re-delivery AFTER the rewrite must keep the content-hash key (not revert to the nonce),
        // so the hash-keyed download/backfill/clear helpers still resolve the row.
        save_message("npub1dl", &msg(vec![att("nonceid", false, "")])).await.unwrap();
        let got = crate::db::attachments::get_attachments_for_event("dl1").unwrap();
        assert_eq!(got[0].id, "contenthash", "content-hash key survives a later re-delivery");
        assert!(got[0].downloaded && got[0].path == "/tmp/f.png");
    }

    // An un-backfilled pre-migration event (attachments only in the legacy tag, no table row) still
    // renders via the read fallback.
    #[tokio::test]
    async fn attachments_fall_back_to_legacy_tag_when_table_empty() {
        let (_tmp, _guard) = init_test_db();
        let a = Attachment {
            id: "tagonly".into(), url: "u".into(), name: "old.png".into(), extension: "png".into(),
            size: 7, downloaded: false, ..Default::default()
        };
        save_message("npub1old", &Message {
            id: "old1".into(), content: String::new(), at: 2_000_000, mine: false,
            npub: Some("npub1s".into()), attachments: vec![a.clone()], ..Default::default()
        }).await.unwrap();

        // Simulate the pre-migration shape: drop the table rows, write the legacy tag onto the event.
        {
            let conn = crate::db::get_write_connection_guard_static().unwrap();
            conn.execute("DELETE FROM attachments WHERE event_id='old1'", []).unwrap();
            let inner = serde_json::to_string(&vec![a]).unwrap();
            let tags = serde_json::to_string(&vec![vec!["attachments".to_string(), inner]]).unwrap();
            conn.execute("UPDATE events SET tags=?1 WHERE id='old1'", rusqlite::params![tags]).unwrap();
        }
        assert!(crate::db::attachments::get_attachments_for_event("old1").unwrap().is_empty(), "table empty");

        let chat_int = crate::db::id_cache::get_or_create_chat_id("npub1old").unwrap();
        let views = get_message_views(chat_int, 10, 0).await.unwrap();
        let old = views.iter().find(|m| m.id == "old1").unwrap();
        assert_eq!(old.attachments.len(), 1, "attachment served from the legacy-tag fallback");
        assert_eq!(old.attachments[0].name, "old.png");
    }

    // Migration 75's strip logic: remove the legacy attachments tag ONLY from events that are
    // provably backfilled (have a table row); un-backfilled events keep their tag as the fallback.
    #[tokio::test]
    async fn attachment_tag_strip_is_gated_on_backfill() {
        let (_tmp, _guard) = init_test_db();
        let a = Attachment { id: "h1".into(), extension: "png".into(), name: "f.png".into(), downloaded: false, ..Default::default() };
        let mk = |id: &str, secs: u64| Message {
            id: id.into(), content: String::new(), at: secs * 1000, mine: false,
            npub: Some("npub1s".into()), attachments: vec![a.clone()], ..Default::default()
        };
        save_message("npub1s", &mk("bf", 1000)).await.unwrap();
        save_message("npub1s", &mk("unbf", 2000)).await.unwrap();

        {
            let conn = crate::db::get_write_connection_guard_static().unwrap();
            let inner = serde_json::to_string(&vec![a.clone()]).unwrap();
            let with_tag = |ms: &str| serde_json::to_string(&vec![
                vec!["ms".to_string(), ms.to_string()],
                vec!["attachments".to_string(), inner.clone()],
            ]).unwrap();
            // Both events carry a legacy tag; only `unbf` loses its table rows (un-backfilled).
            conn.execute("UPDATE events SET tags=?1 WHERE id='bf'", rusqlite::params![with_tag("5")]).unwrap();
            conn.execute("UPDATE events SET tags=?1 WHERE id='unbf'", rusqlite::params![with_tag("6")]).unwrap();
            conn.execute("DELETE FROM attachments WHERE event_id='unbf'", []).unwrap();

            // Replicate migration 75's gated strip.
            let events: Vec<(String, String)> = {
                let mut stmt = conn.prepare("SELECT id, tags FROM events WHERE tags LIKE '%attachments%' AND id IN (SELECT DISTINCT event_id FROM attachments)").unwrap();
                let m = stmt.query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?))).unwrap();
                m.flatten().collect()
            };
            for (id, tj) in events {
                let mut tags: Vec<Vec<String>> = serde_json::from_str(&tj).unwrap();
                tags.retain(|t| t.first().map(|s| s.as_str()) != Some("attachments"));
                conn.execute("UPDATE events SET tags=?1 WHERE id=?2",
                    rusqlite::params![serde_json::to_string(&tags).unwrap(), id]).unwrap();
            }

            let bf: String = conn.query_row("SELECT tags FROM events WHERE id='bf'", [], |r| r.get(0)).unwrap();
            assert!(!bf.contains("attachments"), "backfilled event's attachments tag stripped");
            assert!(bf.contains("\"ms\""), "sibling ms tag survives the strip");
            let unbf: String = conn.query_row("SELECT tags FROM events WHERE id='unbf'", [], |r| r.get(0)).unwrap();
            assert!(unbf.contains("attachments"), "un-backfilled event keeps its tag (no table row)");
        }

        // The un-backfilled event still renders via the read fallback after the strip.
        let chat_int = crate::db::id_cache::get_or_create_chat_id("npub1s").unwrap();
        let views = get_message_views(chat_int, 10, 0).await.unwrap();
        assert_eq!(views.iter().find(|m| m.id == "unbf").unwrap().attachments.len(), 1, "fallback still renders unbf");
    }

    // Mark-as-unread anchors from the FULL DB history (a community row often holds only a preview
    // message in RAM). The anchor lands strictly before the target's second so the count query's
    // strict `>` still counts the newest contact message. Covers the community repro + edge cases.
    #[tokio::test]
    async fn compute_unread_anchor_covers_the_cases() {
        let (_tmp, _guard) = init_test_db();
        let mk = |id: &str, secs: u64, mine: bool| Message {
            id: id.into(), content: "x".into(), at: secs * 1000, mine,
            npub: (!mine).then(|| "npub1sender".to_string()),
            ..Default::default()
        };
        let unread = |chat: &'static str| async move {
            unread_counts().await.unwrap().get(chat).copied().unwrap_or(0)
        };
        let set_lr = |chat: &str, lr: &str| {
            let conn = crate::db::get_write_connection_guard_static().unwrap();
            conn.execute("UPDATE chats SET last_read = ?1 WHERE chat_identifier = ?2",
                rusqlite::params![lr, chat]).unwrap();
        };

        // (A) The community repro: we spoke long ago, they kept talking. Anchor = second-newest
        // contact message → exactly one unread, whatever the RAM cache held.
        let a = "npub1anchorA";
        save_message(a, &mk("a_mine", 1000, true)).await.unwrap();
        for i in 0..8u64 { save_message(a, &mk(&format!("a{i}"), 2000 + i, false)).await.unwrap(); }
        assert_eq!(compute_unread_anchor(a).await.unwrap(), UnreadMark::Anchor("a6".into()));
        set_lr(a, "a6");
        assert_eq!(unread(a).await, 1, "A: newest contact message is the sole unread");

        // (B) We spoke last → NoOp (no phantom badge, no snap-back jiggle).
        let b = "npub1anchorB";
        save_message(b, &mk("b0", 2000, false)).await.unwrap();
        save_message(b, &mk("b_mine", 2001, true)).await.unwrap();
        assert_eq!(compute_unread_anchor(b).await.unwrap(), UnreadMark::NoOp);

        // (C) Same-second tail: the two newest share a second. The anchor must skip to a strictly
        // earlier second, so both same-second messages surface instead of snapping back to read.
        let c = "npub1anchorC";
        save_message(c, &mk("c0", 3000, false)).await.unwrap();
        save_message(c, &mk("c1", 3005, false)).await.unwrap();
        save_message(c, &mk("c2", 3005, false)).await.unwrap();
        assert_eq!(compute_unread_anchor(c).await.unwrap(), UnreadMark::Anchor("c0".into()));
        set_lr(c, "c0");
        assert_eq!(unread(c).await, 2, "C: same-second tail both count");

        // (D) The newest contact message is the chat's first → Clear (never-read) → it still counts.
        let d = "npub1anchorD";
        save_message(d, &mk("d0", 4000, false)).await.unwrap();
        assert_eq!(compute_unread_anchor(d).await.unwrap(), UnreadMark::Clear);
        set_lr(d, "");
        assert_eq!(unread(d).await, 1, "D: lone contact message surfaces");

        // (E) No contact message at all (only our own) → NoOp.
        let e = "npub1anchorE";
        save_message(e, &mk("e_mine", 5000, true)).await.unwrap();
        assert_eq!(compute_unread_anchor(e).await.unwrap(), UnreadMark::NoOp);
    }

    // Regression: a "read to here" marker that lands on a system event (kind 30078, not a counted
    // kind, e.g. the windowed jump-reveal path marking off the raw tail) must still clear unread.
    // The anchor keys off the marker row's time whatever its kind, so it can't wedge at 99+.
    #[tokio::test]
    async fn unread_clears_when_last_read_is_a_system_event() {
        let (_tmp, _guard) = init_test_db();
        let chat = "npub1sysevtdm";
        let mk = |id: &str, secs: u64| Message {
            id: id.into(), content: "x".into(), at: secs * 1000, mine: false,
            npub: Some("npub1sender".to_string()), ..Default::default()
        };
        let unread = || async { unread_counts().await.unwrap().get(chat).copied().unwrap_or(0) };

        // 5 contact messages, never read → all unread. No own message, so the ONLY viable anchor
        // is last_read (this is the case that used to stick at a permanent count).
        for i in 0..5u64 {
            save_message(chat, &mk(&format!("m{i}"), 1000 + i)).await.unwrap();
        }
        assert_eq!(unread().await, 5, "never-read backlog");

        // A system event is the newest row (a join notification, after every contact message).
        save_system_event_at("sysev", chat, SystemEventType::MemberJoined, "npubX", None, 2000, None, None).await.unwrap();

        // last_read pinned to that system event (kind 30078) — the reported bad state.
        {
            let conn = crate::db::get_write_connection_guard_static().unwrap();
            conn.execute(
                "UPDATE chats SET last_read = ?1 WHERE chat_identifier = ?2",
                rusqlite::params!["sysev", chat],
            ).unwrap();
        }
        assert_eq!(unread().await, 0, "read marker on a system event still clears the badge");
    }

    // Deleting a message adjusts unread correctly: removing an UNREAD message drops the count by
    // one, removing the read MARKER retreats it to the prior message (never collapses to 99+), and
    // removing the last read message clears the marker without over-counting.
    #[tokio::test]
    async fn deleting_a_message_adjusts_unread_without_wedging() {
        let (_tmp, _guard) = init_test_db();
        let chat = "npub1delunread";
        let mk = |id: &str, secs: u64| Message {
            id: id.into(), content: "x".into(), at: secs * 1000, mine: false,
            npub: Some("npub1sender".to_string()), ..Default::default()
        };
        let unread = || async { unread_counts().await.unwrap().get(chat).copied().unwrap_or(0) };
        let set_marker = |id: &str| {
            let conn = crate::db::get_write_connection_guard_static().unwrap();
            conn.execute("UPDATE chats SET last_read = ?1 WHERE chat_identifier = ?2",
                rusqlite::params![id, chat]).unwrap();
        };
        let marker = || -> String {
            let conn = crate::db::get_db_connection_guard_static().unwrap();
            conn.query_row("SELECT last_read FROM chats WHERE chat_identifier = ?1",
                rusqlite::params![chat], |r| r.get::<_, String>(0)).unwrap()
        };

        // m0..m5, read up to m1 → m2,m3,m4,m5 unread.
        for i in 0..6u64 { save_message(chat, &mk(&format!("m{i}"), 1000 + i)).await.unwrap(); }
        set_marker("m1");
        assert_eq!(unread().await, 4, "m2..m5 unread");

        // Delete an UNREAD mid-block message → badge drops by one, marker untouched.
        delete_event("m3").await.unwrap();
        assert_eq!(unread().await, 3, "one unread deleted → badge minus one");
        assert_eq!(marker(), "m1", "deleting an unread message leaves the marker alone");

        // Delete the read MARKER → retreats to the prior surviving message (m0), unread unchanged.
        delete_event("m1").await.unwrap();
        assert_eq!(marker(), "m0", "marker retreats to the newest survivor before it");
        assert_eq!(unread().await, 3, "retreat keeps the count, no collapse to 99+");

        // Delete the last surviving read message → marker clears, still exactly the unread block.
        delete_event("m0").await.unwrap();
        assert_eq!(marker(), "", "no earlier survivor → marker clears");
        assert_eq!(unread().await, 3, "cleared marker counts only the true unread survivors");
    }

    // Edits are event-sourced for BOTH transports: a MESSAGE_EDIT event folds into the target's
    // history on reload (latest content + revisions + the edit's own emoji). Community used to
    // overwrite the row and lose all of this — this locks in the unified fold.
    #[tokio::test]
    async fn edit_event_folds_into_history_on_reload() {
        let (_tmp, _guard) = init_test_db();
        let chat = "channel_edit_fold";
        save_message(chat, &Message {
            id: "orig1".into(), content: "original".into(), at: 5_000_000,
            npub: Some("npub1author".into()), ..Default::default()
        }).await.unwrap();

        let cid = crate::db::id_cache::get_chat_id_by_identifier(chat).unwrap();
        let emoji = vec![crate::types::EmojiTag { shortcode: "wave".into(), url: "u/wave".into() }];
        save_edit_event("edit1", "orig1", "edited :wave:", &emoji, cid, None, "npub1author").await.unwrap();

        let m = get_message_views(cid, 50, 0).await.unwrap()
            .into_iter().find(|m| m.id == "orig1").expect("message reloaded");
        assert!(m.edited, "folded edit sets the edited flag");
        let h = m.edit_history.as_ref().expect("history reconstructed from the edit event");
        assert_eq!(h.len(), 2, "original + one edit");
        assert_eq!(h[0].content, "original");
        assert_eq!(h[1].content, "edited :wave:");
        assert_eq!(m.content, "edited :wave:", "latest revision is the displayed content");
        assert_eq!(m.emoji_tags.len(), 1, "the edit's own emoji folds onto the message");
        assert_eq!(m.emoji_tags[0].shortcode, "wave");
    }

    // The bulk-sync persist path: one transaction must land events + attachments + reactions
    // with the same shape save_message produces, and slice order must set rowid order (the
    // same-timestamp pagination tiebreak).
    #[tokio::test]
    async fn batched_save_matches_single_save_shape() {
        let (_tmp, _guard) = init_test_db();
        let chat = "channel_batch1";
        let att = crate::types::Attachment {
            id: "atthash1".into(), extension: "png".into(), name: "a.png".into(),
            url: "https://x/att".into(), downloaded: false, ..Default::default()
        };
        let reaction = Reaction {
            id: "react_b1".into(), reference_id: "b1".into(),
            author_id: "npub1reactor".into(), emoji: "👍".into(), emoji_url: None,
        };
        // Same `at` second across the batch — rowid is the only orderer.
        let msgs: Vec<Message> = (0..5u64).map(|i| Message {
            id: format!("b{i}"), content: format!("c{i}"), at: 7_000_000,
            npub: Some("npub1sender".into()),
            attachments: if i == 2 { vec![att.clone()] } else { Vec::new() },
            reactions: if i == 1 { vec![reaction.clone()] } else { Vec::new() },
            ..Default::default()
        }).collect();
        let refs: Vec<&Message> = msgs.iter().collect();

        let saved = save_messages_batch(chat, &refs, None).await.unwrap();
        assert_eq!(saved, 5, "every message written");

        for i in 0..5u64 {
            assert!(event_exists(&format!("b{i}")).unwrap(), "b{i} row exists");
        }
        assert!(event_exists("react_b1").unwrap(), "reaction landed as its own kind-7 row");
        let atts = crate::db::attachments::get_attachments_for_event("b2").unwrap();
        assert_eq!(atts.len(), 1, "attachment row committed with its event");
        assert_eq!(atts[0].id, "atthash1");

        // rowid order == slice order despite identical timestamps.
        let conn = crate::db::get_db_connection_guard_static().unwrap();
        let ids: Vec<String> = conn
            .prepare("SELECT id FROM events WHERE id IN ('b0','b1','b2','b3','b4') ORDER BY rowid")
            .unwrap()
            .query_map([], |r| r.get(0)).unwrap()
            .flatten().collect();
        assert_eq!(ids, vec!["b0", "b1", "b2", "b3", "b4"], "insert order preserves the rowid tiebreak");
    }

    // A batched re-save must keep save_message's upsert semantics: wrapper_event_id is
    // COALESCE-preserved, and a duplicate reaction in the batch never doubles its row.
    #[tokio::test]
    async fn batched_resave_preserves_wrapper_and_dedups_reactions() {
        let (_tmp, _guard) = init_test_db();
        let chat = "channel_batch2";
        let reaction = Reaction {
            id: "react_rs".into(), reference_id: "rs1".into(),
            author_id: "npub1reactor".into(), emoji: "🔥".into(), emoji_url: None,
        };
        let mut msg = Message {
            id: "rs1".into(), content: "hello".into(), at: 8_000_000,
            npub: Some("npub1sender".into()),
            wrapper_event_id: Some("wrap_original".into()),
            reactions: vec![reaction],
            ..Default::default()
        };
        save_message(chat, &msg).await.unwrap();

        // Re-delivery re-save without a wrapper id, reaction still attached.
        msg.wrapper_event_id = None;
        let saved = save_messages_batch(chat, &[&msg], None).await.unwrap();
        assert_eq!(saved, 1);

        let conn = crate::db::get_db_connection_guard_static().unwrap();
        let wrapper: Option<String> = conn.query_row(
            "SELECT wrapper_event_id FROM events WHERE id = 'rs1'", [], |r| r.get(0),
        ).unwrap();
        assert_eq!(wrapper.as_deref(), Some("wrap_original"), "COALESCE keeps the known wrapper");
        let reaction_rows: i64 = conn.query_row(
            "SELECT COUNT(*) FROM events WHERE id = 'react_rs'", [], |r| r.get(0),
        ).unwrap();
        assert_eq!(reaction_rows, 1, "reaction row not duplicated by the re-save");
    }

    // The DM stream's multi-chat flush: one call, one transaction, rows land under their own
    // chats with their gift-wrap ledger entries; a flush against a stale session drops the
    // buffer AND leaves the wrappers unledgered (that's what makes the drop recoverable).
    #[tokio::test]
    async fn batching_persist_flushes_multi_chat_and_drops_on_stale_session() {
        let (_tmp, _guard) = init_test_db();
        let handler = crate::event_handler::NoOpEventHandler;
        let batcher = crate::event_handler::BatchingPersist::new(&handler);

        let mk = |id: &str, npub: &str| Message {
            id: id.into(), content: "x".into(), at: 9_000_000,
            npub: Some(npub.into()), ..Default::default()
        };
        // Mirror the commit path: buffering only ever happens AFTER the STATE add (the
        // flush drops anything not STATE-resident as deletion protection).
        let seed = |chat: &str, m: &Message| {
            let m = m.clone();
            let chat = chat.to_string();
            async move {
                let mut st = crate::state::STATE.lock().await;
                st.add_message_to_participant(&chat, &m);
            }
        };
        let wrap_a1 = ([0xA1u8; 32], 111u64);
        let wrap_b1 = ([0xB1u8; 32], 222u64);
        let a1 = mk("bp_a1", "npub1chata");
        let b1 = mk("bp_b1", "npub1chatb");
        let a2 = mk("bp_a2", "npub1chata");
        seed("npub1chata", &a1).await;
        seed("npub1chatb", &b1).await;
        seed("npub1chata", &a2).await;

        use crate::event_handler::InboundEventHandler;
        assert!(batcher.buffer_persist("npub1chata", &a1, Some(wrap_a1)), "batcher owns the persist");
        assert!(batcher.buffer_persist("npub1chatb", &b1, Some(wrap_b1)));
        assert!(batcher.buffer_persist("npub1chata", &a2, None));
        assert_eq!(batcher.buffered(), 3);

        let ledgered = |bytes: [u8; 32]| {
            let id = nostr_sdk::prelude::EventId::from_byte_array(bytes);
            crate::db::wrappers::load_negentropy_items().unwrap().iter().any(|(e, _)| *e == id)
        };
        assert!(!ledgered(wrap_a1.0), "wrapper unledgered while its message sits buffered");

        let session = crate::state::SessionGuard::capture();
        assert_eq!(batcher.flush(&session).await, 3, "all buffered messages written");
        assert_eq!(batcher.buffered(), 0);
        assert!(event_exists("bp_a1").unwrap() && event_exists("bp_b1").unwrap() && event_exists("bp_a2").unwrap());
        assert!(ledgered(wrap_a1.0) && ledgered(wrap_b1.0), "wrappers ledgered with the flush");
        let a = crate::db::id_cache::get_chat_id_by_identifier("npub1chata").unwrap();
        let b = crate::db::id_cache::get_chat_id_by_identifier("npub1chatb").unwrap();
        assert_ne!(a, b, "rows grouped under their own chats");

        // Stale session: buffered messages are dropped, never written — and the wrapper
        // stays out of the negentropy fingerprint set, so the message re-delivers.
        let stale = mk("bp_stale", "npub1chata");
        seed("npub1chata", &stale).await;
        let wrap_stale = ([0x5Eu8; 32], 333u64);
        batcher.buffer_persist("npub1chata", &stale, Some(wrap_stale));
        crate::state::bump_session_generation();
        assert_eq!(batcher.flush(&session).await, 0, "stale flush writes nothing");
        assert!(!event_exists("bp_stale").unwrap(), "stale message never reached the DB");
        assert!(!ledgered(wrap_stale.0), "dropped message's wrapper NOT ledgered — negentropy will re-deliver it");
        assert_eq!(batcher.buffered(), 0, "stale buffer drained, not retried into the next account");
    }

    // A deletion landing while its target sits buffered must not resurrect the message:
    // the same-task path purges via on_message_deleted, and the cross-task path (live
    // subscription, different handler) is caught by the flush's deletion-tombstone filter.
    // The filter keys on the POSITIVE tombstone, never STATE absence — an LRU-evicted (but
    // not deleted) message MUST still persist and ledger, or archive sync could silently
    // drop the exact history it exists to persist.
    #[tokio::test]
    async fn buffered_message_deleted_before_flush_never_persists() {
        let (_tmp, _guard) = init_test_db();
        let handler = crate::event_handler::NoOpEventHandler;
        let batcher = crate::event_handler::BatchingPersist::new(&handler);
        use crate::event_handler::InboundEventHandler;

        let chat = "npub1delchat";
        let mk = |id: &str| Message {
            id: id.into(), content: "x".into(), at: 9_500_000,
            npub: Some(chat.into()), ..Default::default()
        };
        let ledgered = |bytes: [u8; 32]| {
            let id = nostr_sdk::prelude::EventId::from_byte_array(bytes);
            crate::db::wrappers::load_negentropy_items().unwrap().iter().any(|(e, _)| *e == id)
        };

        // Same-task deletion (sync stream): commit_deletion fires on_message_deleted on
        // the batching handler → the buffered entry purges immediately.
        let m1 = mk("del_sametask");
        {
            let mut st = crate::state::STATE.lock().await;
            st.add_message_to_participant(chat, &m1);
        }
        batcher.buffer_persist(chat, &m1, Some(([0xD1u8; 32], 444)));
        {
            let mut st = crate::state::STATE.lock().await;
            st.remove_message("del_sametask");
        }
        batcher.on_message_deleted(chat, "del_sametask");
        assert_eq!(batcher.buffered(), 0, "deletion purges the buffered target");

        // Cross-task deletion (live subscription, plain handler): no purge call — the
        // deletion tombstone (recorded by commit_deletion) drops it at flush time.
        let m2 = mk("del_crosstask");
        {
            let mut st = crate::state::STATE.lock().await;
            st.add_message_to_participant(chat, &m2);
        }
        batcher.buffer_persist(chat, &m2, Some(([0xD2u8; 32], 555)));
        crate::state::note_message_deleted("del_crosstask");

        // LRU eviction is NOT deletion: gone from STATE, no tombstone → must persist.
        let m3 = mk("evicted_ok");
        let wrap_evicted = ([0xE0u8; 32], 666u64);
        {
            let mut st = crate::state::STATE.lock().await;
            st.add_message_to_participant(chat, &m3);
            st.remove_message("evicted_ok");
        }
        batcher.buffer_persist(chat, &m3, Some(wrap_evicted));

        let session = crate::state::SessionGuard::capture();
        assert_eq!(batcher.flush(&session).await, 1, "tombstoned target dropped, evicted message written");
        assert!(!event_exists("del_sametask").unwrap(), "purged message never persisted");
        assert!(!event_exists("del_crosstask").unwrap(), "tombstoned message never persisted");
        assert!(event_exists("evicted_ok").unwrap(), "evicted-but-not-deleted message persisted");
        assert!(ledgered(wrap_evicted.0), "evicted message's wrapper ledgered with it");
    }
}