vector_core/

event_handler.rs

//! Event handler — gift wrap receive, unwrap, process, commit pipeline.
//!
//! Two-phase architecture:
//! - **Phase 1** (`prepare_event`): Parallel-safe — dedup, unwrap, process_rumor
//! - **Phase 2** (`commit_prepared_event`): Sequential — save DB, update STATE, emit
//!
//! Platform-specific behavior (notifications) handled by `InboundEventHandler` trait.

use nostr_sdk::prelude::*;

use crate::rumor::{RumorProcessingResult, RumorEvent, RumorContext, ConversationType, process_rumor};
use crate::types::Message;
use crate::state::WRAPPER_ID_CACHE;

/// Platform-specific callbacks for inbound event processing.
///
/// Same pattern as SendCallback/ProfileSyncHandler — trait with default no-ops.
/// Platforms implement only the hooks they need.
pub trait InboundEventHandler: Send + Sync {
    /// A DM text message was received and committed to STATE + DB.
    fn on_dm_received(&self, _chat_id: &str, _msg: &Message, _is_new: bool) {}

    /// A DM file attachment was received and committed to STATE + DB.
    fn on_file_received(&self, _chat_id: &str, _msg: &Message, _is_new: bool) {}

    /// A reaction was received and applied to a message.
    fn on_reaction_received(&self, _chat_id: &str, _msg: &Message) {}

    /// A previously-stored message was deleted by its sender (Layer 2
    /// cooperative hide via NIP-09 over NIP-17). Frontend drops the row.
    fn on_message_deleted(&self, _chat_id: &str, _message_id: &str) {}

    /// A Community invite was received over a gift wrap and the local user was
    /// joined (member-view Community persisted). Platform refreshes the Community
    /// subscription so messages start flowing, and surfaces the new Community in the UI.
    fn on_community_invite(&self, _community_id: &str) {}

    // --- Community realtime (Concord channel events; `chat_id` is the channel id hex) ---

    /// A new Community channel message was received, ingested into STATE, and persisted.
    fn on_community_message(&self, _chat_id: &str, _msg: &Message, _is_new: bool) {}

    /// A reaction or edit was applied to an existing Community message. `target_id` is the
    /// affected message; `msg` is the live-updated view.
    fn on_community_update(&self, _chat_id: &str, _target_id: &str, _msg: &Message) {}

    /// A Community message was removed (cooperative delete / moderation tombstone).
    fn on_community_removed(&self, _chat_id: &str, _target_id: &str) {}

    /// A join/leave presence announcement. `created_at` is the authenticated inner timestamp;
    /// `invited_by`/`invited_label` carry invite attribution when present.
    #[allow(clippy::too_many_arguments)]
    fn on_community_presence(
        &self,
        _chat_id: &str,
        _npub: &str,
        _joined: bool,
        _event_id: &str,
        _created_at: u64,
        _invited_by: Option<&str>,
        _invited_label: Option<&str>,
    ) {}

    /// A Community typing indicator (ephemeral). `until` is the unix-secs the typer stops being active.
    fn on_community_typing(&self, _chat_id: &str, _npub: &str, _until: u64) {}

    /// A WebXDC realtime peer signal. `node_addr` = `Some` advertises an Iroh node, `None` = peer-left.
    #[allow(clippy::too_many_arguments)]
    fn on_community_webxdc(
        &self,
        _chat_id: &str,
        _npub: &str,
        _topic_id: &str,
        _node_addr: Option<&str>,
        _event_id: &str,
        _created_at: u64,
    ) {}

    /// A WebXDC realtime peer signal in a DM. `contact` is the DM's npub; `node_addr` =
    /// `Some` advertises an Iroh node, `None` = peer-left. Clients without Iroh ignore it.
    #[allow(clippy::too_many_arguments)]
    fn on_webxdc_signal(
        &self,
        _contact: &str,
        _npub: &str,
        _topic_id: &str,
        _node_addr: Option<&str>,
        _event_id: &str,
        _created_at: u64,
    ) {}

    /// The local user was removed from a Community (kick / ban / a leave authored on another device).
    /// Local data is torn down (epoch keys retained); the platform surfaces it + refreshes subs.
    fn on_community_self_removed(&self, _community_id: &str) {}

    /// A Private channel is now readable — its key arrived, by rekey delivery or
    /// by a grant's key vend. `channel_id` is the 32-byte hex id.
    ///
    /// The inverse of the silent-mute failure: a bot granted access can act the
    /// moment it can actually read, rather than polling for a channel it has no
    /// way to know it is waiting on.
    ///
    /// `backfilled` counts the messages that predate the grant and were pulled
    /// into LOCAL STATE before this fired. They are history, not delivery: none
    /// of them reach [`on_message`](Self::on_message) — read them explicitly
    /// (`get_messages_before`) and decide what deserves acting on.
    fn on_channel_keyed(&self, _community_id: &str, _channel_id: &str, _backfilled: usize) {}

    /// The realtime Community subscription REGISTERED: healthy relays deliver
    /// live channel events from this moment (AUTH-gating relays join as their
    /// stream auth completes in the background). Fires once per
    /// [`listen`](crate::VectorCore::listen), after the startup refresh commits —
    /// even when `communities` is 0, so "subscribed to nothing" and "still
    /// connecting" are distinguishable. Community messages sent before this are
    /// only recoverable by a later sync; DMs sent since `listen` began still
    /// reach the handler.
    fn on_subscription_ready(&self, _communities: usize) {}

    /// A Community's control plane was refreshed in realtime (banlist/roles/metadata/mode change,
    /// or a re-founding followed). The platform re-reads display state.
    fn on_community_refreshed(&self, _community_id: &str) {}

    /// A Community was DISSOLVED by its owner (CORD-02 §9): sealed read-only, held keys still open
    /// history but nothing new is honored. The platform surfaces the grave.
    fn on_community_dissolved(&self, _community_id: &str) {}

    /// Bulk-sync persist intercept: return `true` to take ownership of persisting a committed
    /// DM/file message — the commit then SKIPS its per-message save AND its wrapper-ledger
    /// write (`wrapper` = the gift-wrap's `(id_bytes, created_at)`; the owner MUST commit it
    /// in the same transaction as the message row, or the message loses crash/drop
    /// recoverability). Streaming sync loops (see [`BatchingPersist`]) buffer here and drain
    /// many messages into one transaction. Default `false` keeps the per-message save.
    fn buffer_persist(&self, _chat_id: &str, _msg: &Message, _wrapper: Option<([u8; 32], u64)>) -> bool { false }
}

/// No-op handler for CLI/tests.
pub struct NoOpEventHandler;
impl InboundEventHandler for NoOpEventHandler {}

/// For catch-up syncs: forwards to `inner` only what is live rather than
/// history. Mini App peer signals always: a signal is live state, and a sync
/// that opens one ledgers its wrapper, so the live subscription will never
/// deliver it again. With [`from`](Self::from), also DMs and files sent since
/// that moment: sent while the sync ran, they can reach us through it alone.
pub struct LiveOnly<'a> {
    inner: &'a dyn InboundEventHandler,
    live_from_ms: Option<u64>,
}

impl<'a> LiveOnly<'a> {
    /// Signals only.
    pub fn signals(inner: &'a dyn InboundEventHandler) -> Self {
        LiveOnly { inner, live_from_ms: None }
    }

    /// Signals, and messages sent at or after `ms` (unix milliseconds).
    pub fn from(inner: &'a dyn InboundEventHandler, ms: u64) -> Self {
        LiveOnly { inner, live_from_ms: Some(ms) }
    }

    fn is_live(&self, msg: &Message) -> bool {
        self.live_from_ms.is_some_and(|from| msg.at >= from)
    }
}

impl InboundEventHandler for LiveOnly<'_> {
    fn on_community_webxdc(&self, chat_id: &str, npub: &str, topic_id: &str, node_addr: Option<&str>, event_id: &str, created_at: u64) {
        self.inner.on_community_webxdc(chat_id, npub, topic_id, node_addr, event_id, created_at)
    }

    fn on_webxdc_signal(&self, contact: &str, npub: &str, topic_id: &str, node_addr: Option<&str>, event_id: &str, created_at: u64) {
        self.inner.on_webxdc_signal(contact, npub, topic_id, node_addr, event_id, created_at)
    }

    fn on_dm_received(&self, chat_id: &str, msg: &Message, is_new: bool) {
        if self.is_live(msg) {
            self.inner.on_dm_received(chat_id, msg, is_new)
        }
    }

    fn on_file_received(&self, chat_id: &str, msg: &Message, is_new: bool) {
        if self.is_live(msg) {
            self.inner.on_file_received(chat_id, msg, is_new)
        }
    }
}

/// One deferred DM persist: the message, its chat, and its gift-wrap ledger entry — the
/// ledger row commits in the same flush transaction as the message row (see
/// `save_messages_batch_multi`), so a lost batch leaves the wrapper unledgered.
struct BufferedDm {
    chat_id: String,
    msg: Message,
    wrapper: Option<([u8; 32], u64)>,
}

/// Wraps any handler for a bulk-sync drain loop: every callback delegates to the inner
/// handler, but committed messages BUFFER here instead of saving one transaction each —
/// the loop calls [`BatchingPersist::flush`] periodically and at stream end to land them
/// in batched transactions (`save_messages_batch_multi`).
///
/// Deferral is recoverable because the wrapper ledger (the negentropy fingerprint set)
/// rides the flush transaction: a message lost to a crash, a stale-session drop, or a
/// failed flush leaves its wrapper unledgered, so the next reconciliation re-delivers it.
pub struct BatchingPersist<'a> {
    inner: &'a dyn InboundEventHandler,
    buf: std::sync::Mutex<Vec<BufferedDm>>,
    /// The account whose messages are in the buffer.
    ///
    /// Held rather than resolved at flush time: this outlives the calls that
    /// fill it, and a flush reached after a swap would otherwise write one
    /// account's inbox into another's. Holding it means the buffer always
    /// drains where it was filled, from any caller, bound or not.
    session: std::sync::Arc<crate::db::Session>,
}

impl<'a> BatchingPersist<'a> {
    pub fn new(inner: &'a dyn InboundEventHandler) -> Self {
        Self {
            inner,
            buf: std::sync::Mutex::new(Vec::new()),
            session: crate::db::current_session(),
        }
    }

    /// How many messages are waiting — the loop's flush-threshold probe.
    pub fn buffered(&self) -> usize {
        self.buf.lock().map(|b| b.len()).unwrap_or(0)
    }

    /// Drain the buffer into batched transactions (grouped by chat, arrival order kept),
    /// against the account that filled it.
    pub async fn flush(&self) -> usize {
        self.try_flush().await.unwrap_or(0)
    }

    /// [`Self::flush`], but a persist failure is distinguishable from "nothing
    /// to write" — callers that gate follow-on effects on the ledger actually
    /// holding the batch (reconcile-cursor births) need the difference: an
    /// advance over an unledgered batch skips those events forever.
    pub async fn try_flush(&self) -> Result<usize, String> {
        crate::db::with_session(self.session.clone(), async move {
            let mut drained: Vec<BufferedDm> = match self.buf.lock() {
                Ok(mut b) => b.drain(..).collect(),
                Err(_) => return Ok(0),
            };
            if drained.is_empty() {
                return Ok(0);
            }
            // A deletion may have landed (live subscription or this stream) while an entry sat
            // buffered: its delete_event no-ops on the not-yet-written row, so persisting the
            // buffered copy would resurrect a deleted message. Keyed on the POSITIVE deletion
            // tombstone, never STATE absence — the LRU evicts old messages from STATE, and
            // archive-synced history is exactly that tail (an evicted message must still
            // persist). A dropped entry's wrapper stays unledgered and re-delivers next sync,
            // where the DB dedup sees the (still-deleted) state cleanly.
            drained.retain(|e| !crate::state::was_message_deleted(&e.msg.id));
            if drained.is_empty() {
                return Ok(0);
            }
            // Group by chat preserving first-seen chat order + per-chat arrival order.
            let mut groups: Vec<crate::db::events::ChatBatch<'_>> = Vec::new();
            for e in &drained {
                match groups.iter_mut().find(|(c, _)| c == &e.chat_id) {
                    Some((_, v)) => v.push((&e.msg, e.wrapper)),
                    None => groups.push((e.chat_id.clone(), vec![(&e.msg, e.wrapper)])),
                }
            }
            match crate::db::events::save_messages_batch_multi(&groups).await {
                Ok(n) => Ok(n),
                Err(e) => {
                    crate::log_warn!("[Sync] batched persist failed ({} msgs): {}", drained.len(), e);
                    Err(e)
                }
            }
        })
        .await
    }
}

impl InboundEventHandler for BatchingPersist<'_> {
    fn buffer_persist(&self, chat_id: &str, msg: &Message, wrapper: Option<([u8; 32], u64)>) -> bool {
        match self.buf.lock() {
            Ok(mut b) => {
                b.push(BufferedDm { chat_id: chat_id.to_string(), msg: msg.clone(), wrapper });
                true
            }
            // Poisoned lock: fall back to the commit's own per-message save.
            Err(_) => false,
        }
    }

    fn on_dm_received(&self, chat_id: &str, msg: &Message, is_new: bool) {
        self.inner.on_dm_received(chat_id, msg, is_new)
    }
    fn on_file_received(&self, chat_id: &str, msg: &Message, is_new: bool) {
        self.inner.on_file_received(chat_id, msg, is_new)
    }
    fn on_reaction_received(&self, chat_id: &str, msg: &Message) {
        self.inner.on_reaction_received(chat_id, msg)
    }
    fn on_message_deleted(&self, chat_id: &str, message_id: &str) {
        // The deletion's delete_event no-ops when the target is still buffered here —
        // purge it (unledgered wrapper → re-delivers → dedups against the deleted state)
        // so the flush can't resurrect a deleted message.
        if let Ok(mut b) = self.buf.lock() {
            b.retain(|e| e.msg.id != message_id);
        }
        self.inner.on_message_deleted(chat_id, message_id)
    }
    fn on_community_invite(&self, community_id: &str) {
        self.inner.on_community_invite(community_id)
    }
    fn on_community_message(&self, chat_id: &str, msg: &Message, is_new: bool) {
        self.inner.on_community_message(chat_id, msg, is_new)
    }
    fn on_community_update(&self, chat_id: &str, target_id: &str, msg: &Message) {
        self.inner.on_community_update(chat_id, target_id, msg)
    }
    fn on_community_removed(&self, chat_id: &str, target_id: &str) {
        self.inner.on_community_removed(chat_id, target_id)
    }
    fn on_community_presence(
        &self,
        chat_id: &str,
        npub: &str,
        joined: bool,
        event_id: &str,
        created_at: u64,
        invited_by: Option<&str>,
        invited_label: Option<&str>,
    ) {
        self.inner.on_community_presence(chat_id, npub, joined, event_id, created_at, invited_by, invited_label)
    }
    fn on_community_typing(&self, chat_id: &str, npub: &str, until: u64) {
        self.inner.on_community_typing(chat_id, npub, until)
    }
    fn on_community_webxdc(
        &self,
        chat_id: &str,
        npub: &str,
        topic_id: &str,
        node_addr: Option<&str>,
        event_id: &str,
        created_at: u64,
    ) {
        self.inner.on_community_webxdc(chat_id, npub, topic_id, node_addr, event_id, created_at)
    }
    fn on_webxdc_signal(
        &self,
        contact: &str,
        npub: &str,
        topic_id: &str,
        node_addr: Option<&str>,
        event_id: &str,
        created_at: u64,
    ) {
        self.inner.on_webxdc_signal(contact, npub, topic_id, node_addr, event_id, created_at)
    }
    fn on_community_self_removed(&self, community_id: &str) {
        self.inner.on_community_self_removed(community_id)
    }
    fn on_community_refreshed(&self, community_id: &str) {
        self.inner.on_community_refreshed(community_id)
    }
    fn on_community_dissolved(&self, community_id: &str) {
        self.inner.on_community_dissolved(community_id)
    }
}

/// Result of Phase 1 (prepare_event) — everything needed for sequential commit.
#[allow(clippy::large_enum_variant)] // boxing a public variant's payload is an API break
pub enum PreparedEvent {
    /// Fully processed DM rumor — ready for state commit.
    Processed {
        result: RumorProcessingResult,
        contact: String,
        sender: PublicKey,
        is_mine: bool,
        wrapper_event_id: String,
        wrapper_event_id_bytes: [u8; 32],
        wrapper_created_at: u64,
        /// Time spent on ECDH + ChaCha20Poly1305 decryption (nanoseconds)
        unwrap_ns: u64,
        /// Time spent on rumor parsing (nanoseconds)
        parse_ns: u64,
    },
    /// Community invite bundle (kind 3304) — parked for explicit user consent.
    CommunityInvite {
        invite: crate::community::invite::CommunityInvite,
        /// Inviter's npub (bech32) — shown in the pending-invite UI.
        inviter: String,
        is_mine: bool,
        wrapper_event_id_bytes: [u8; 32],
        wrapper_created_at: u64,
        /// Inner rumor `created_at` (seconds) — the real send time. Unlike the outer
        /// wrapper, which NIP-59 backdates up to 2 days, this is honest; the tombstone
        /// supersession test needs it so a re-invite isn't misread as older than a decline.
        rumor_created_at: u64,
        /// Sender-declared NIP-40 expiry (unix secs); 0 = none, so permanent.
        expires_at: u64,
    },
    /// Concord v2 Direct Invite (inner kind 3313) — parked for explicit consent.
    /// Carries the raw bundle JSON (parked verbatim; the accept path re-parses it).
    CommunityInviteV2 {
        bundle_json: String,
        community_id: String,
        /// Inviter's npub (hex) — the proven seal signer.
        inviter: String,
        is_mine: bool,
        wrapper_event_id_bytes: [u8; 32],
        wrapper_created_at: u64,
        /// Inner rumor `created_at` (seconds) — the real send time (see the v1 variant).
        rumor_created_at: u64,
        /// Sender-declared NIP-40 expiry (unix secs); 0 = none, so permanent.
        expires_at: u64,
    },
    /// Duplicate event — just persist wrapper for negentropy.
    DedupSkip {
        wrapper_id_bytes: [u8; 32],
        wrapper_created_at: u64,
    },
    /// Error during unwrap/processing — persist wrapper for negentropy.
    ErrorSkip {
        wrapper_id_bytes: [u8; 32],
        wrapper_created_at: u64,
    },
}

/// Phase 1: Prepare an event for commit (parallel-safe, no state mutation).
///
/// Performs dedup check, gift wrap decryption, and rumor parsing.
/// Safe to call from multiple tokio tasks concurrently.
pub async fn prepare_event(
    event: Event,
    _client: &Client,
    my_public_key: PublicKey,
) -> PreparedEvent {
    if let Some(skip) = dedup_skip(&event).await {
        return skip;
    }
    let unwrap_start = web_time::Instant::now();
    let unwrapped = crate::signer::unwrap_gift_wrap(&event).await;
    finish_prepare(&event, unwrapped, unwrap_start.elapsed().as_nanos() as u64, my_public_key)
}

/// Wraps opened per vault read on bulk paths. It bounds how long the key stays
/// materialised: two ECDHs per wrap, about 1.5 ms per batch on a desktop core.
pub const UNWRAP_BATCH: usize = 16;

/// [`prepare_event`] for many events, one result per event in order. The wraps that pass
/// dedup are opened together, under one vault read for a local key.
pub async fn prepare_events(events: &[Event], my_public_key: PublicKey) -> Vec<PreparedEvent> {
    let mut out: Vec<Option<PreparedEvent>> = Vec::with_capacity(events.len());
    for event in events {
        out.push(dedup_skip(event).await);
    }
    let todo: Vec<&Event> = events.iter().zip(&out).filter(|(_, o)| o.is_none()).map(|(e, _)| e).collect();
    if !todo.is_empty() {
        let unwrap_start = web_time::Instant::now();
        let unwrapped = crate::signer::unwrap_gift_wraps(&todo).await;
        let unwrap_ns = unwrap_start.elapsed().as_nanos() as u64 / todo.len() as u64;
        let mut unwrapped = unwrapped.into_iter();
        for (event, slot) in events.iter().zip(out.iter_mut()) {
            if slot.is_none() {
                let u = unwrapped.next().unwrap_or_else(|| Err("missing unwrap result".into()));
                *slot = Some(finish_prepare(event, u, unwrap_ns, my_public_key));
            }
        }
    }
    out.into_iter().flatten().collect()
}

/// `Some(DedupSkip)` when this wrap has been seen before.
async fn dedup_skip(event: &Event) -> Option<PreparedEvent> {
    let wrapper_created_at = event.created_at.as_secs();
    let wrapper_event_id_bytes: [u8; 32] = event.id.to_bytes();
    let skip = || Some(PreparedEvent::DedupSkip { wrapper_id_bytes: wrapper_event_id_bytes, wrapper_created_at });

    // Dedup: in-memory cache first, then DB fallback
    {
        let cache = WRAPPER_ID_CACHE.lock().await;
        if cache.contains(&wrapper_event_id_bytes) {
            return skip();
        }
    }

    if let Ok(true) = crate::db::events::wrapper_event_exists(&event.id.to_hex()) {
        return skip();
    }

    // The persistent ledger too, not just the events table: a DELETED message has no
    // events row, so without this a wrap re-served after a restart (relays ignore
    // NIP-09 freely) re-processes cleanly and resurrects the message.
    if crate::db::wrappers::processed_wrapper_exists(&wrapper_event_id_bytes) {
        return skip();
    }
    None
}

/// Everything after the unwrap: classify the rumor and parse it.
fn finish_prepare(
    event: &Event,
    unwrapped: Result<UnwrappedGift, String>,
    unwrap_ns: u64,
    my_public_key: PublicKey,
) -> PreparedEvent {
    let wrapper_created_at = event.created_at.as_secs();
    let wrapper_event_id_bytes: [u8; 32] = event.id.to_bytes();
    let wrapper_event_id = event.id.to_hex();

    let (rumor, sender) = match unwrapped {
        Ok(UnwrappedGift { rumor, sender }) => (rumor, sender),
        Err(_) => return PreparedEvent::ErrorSkip {
            wrapper_id_bytes: wrapper_event_id_bytes, wrapper_created_at,
        },
    };

    // Inner rumor send time (seconds). The outer wrapper's `created_at` is NIP-59
    // backdated up to 2 days, so it can't order an invite against a decline tombstone.
    let rumor_created_at = rumor.created_at.as_secs();

    let is_mine = sender == my_public_key;
    let contact = if is_mine {
        rumor.tags.public_keys().next()
            .and_then(|pk| pk.to_bech32().ok())
            .unwrap_or_else(|| sender.to_bech32().unwrap_or_default())
    } else {
        sender.to_bech32().unwrap_or_default()
    };

    // Skip NIP-17 group messages (multiple p-tags) — Vector DMs are 1:1
    if rumor.tags.public_keys().count() > 1 {
        return PreparedEvent::ErrorSkip {
            wrapper_id_bytes: wrapper_event_id_bytes, wrapper_created_at,
        };
    }

    // Community invite (carrier) — a join, not a chat message. Recognized before
    // process_rumor so it never lands as an UnknownEvent in the DM thread.
    if rumor.kind == Kind::Custom(crate::stored_event::event_kind::COMMUNITY_INVITE_BUNDLE) {
        return match crate::community::invite::parse_invite_rumor(rumor.kind, &rumor.content) {
            Some(invite) => PreparedEvent::CommunityInvite {
                invite, inviter: contact.clone(), is_mine, wrapper_event_id_bytes, wrapper_created_at, rumor_created_at,
                expires_at: crate::community::invite::expiration_secs(&rumor.tags).unwrap_or(0),
            },
            None => PreparedEvent::ErrorSkip { wrapper_id_bytes: wrapper_event_id_bytes, wrapper_created_at },
        };
    }

    // Concord v2 Direct Invite (inner kind 3313) — the v2 join carrier. `from_bundle_json`
    // validates the owner commitment + bounds; we park the canonical re-serialized bundle.
    if rumor.kind == Kind::Custom(crate::community::v2::kind::DIRECT_INVITE) {
        return match crate::community::v2::invite::CommunityInvite::from_bundle_json(&rumor.content)
            .ok()
            .and_then(|b| serde_json::to_string(&b).ok().map(|j| (b.community_id, j)))
        {
            Some((community_id, bundle_json)) => PreparedEvent::CommunityInviteV2 {
                community_id,
                bundle_json,
                inviter: contact.clone(), // the seal signer's npub (bech32), like the v1 arm
                is_mine,
                wrapper_event_id_bytes,
                wrapper_created_at,
                rumor_created_at,
                expires_at: crate::community::invite::expiration_secs(&rumor.tags).unwrap_or(0),
            },
            None => PreparedEvent::ErrorSkip { wrapper_id_bytes: wrapper_event_id_bytes, wrapper_created_at },
        };
    }

    // Build RumorEvent for processing
    let Some(rumor_id) = rumor.id else {
        return PreparedEvent::ErrorSkip {
            wrapper_id_bytes: wrapper_event_id_bytes, wrapper_created_at,
        };
    };

    let rumor_event = RumorEvent {
        id: rumor_id,
        kind: rumor.kind,
        content: rumor.content,
        tags: rumor.tags,
        created_at: rumor.created_at,
        pubkey: rumor.pubkey,
    };
    let rumor_context = RumorContext {
        sender,
        is_mine,
        conversation_id: contact.clone(),
        conversation_type: ConversationType::DirectMessage,
    };

    let parse_start = web_time::Instant::now();
    let download_dir = crate::db::get_download_dir();
    match process_rumor(rumor_event, rumor_context, &download_dir) {
        Ok(result) => {
            let parse_ns = parse_start.elapsed().as_nanos() as u64;
            PreparedEvent::Processed {
                result, contact, sender, is_mine,
                wrapper_event_id, wrapper_event_id_bytes, wrapper_created_at,
                unwrap_ns, parse_ns,
            }
        }
        Err(e) => {
            log_warn!("[EventHandler] Failed to process rumor: {}", e);
            PreparedEvent::ErrorSkip {
                wrapper_id_bytes: wrapper_event_id_bytes, wrapper_created_at,
            }
        }
    }
}

// ============================================================================
// Phase 2: Commit — sequential state mutation, DB save, emit
// ============================================================================

/// Phase 2: commit a prepared event (sequential — not parallel-safe).
/// Saves to DB, updates STATE, emits to frontend, calls handler hooks.
/// Returns true if a new displayable message was committed.
///
/// Session-safety: captures the generation at the first line. If a swap
/// occurred between `prepare_event()` and here (e.g. long-running
/// negentropy fetch queued events for commit), bail before any STATE /
/// DB write. Centralized so individual spawn sites (sync.rs fetch_messages,
/// archive task, sync_dms, subscription_handler) don't have to wrap.
/// Direct Invites live 24 hours BY DESIGN, and the RECIPIENT enforces it: a
/// sender that omits the NIP-40 tag (older clients, other implementations)
/// must not mint a permanent invite, and an archive sync that resurrects a
/// months-old wrap must not park a ghost (a re-founded community's stale
/// invite has no held id for the exists-guard to match). The sender's
/// declared deadline is honored when EARLIER; the rumor-age lifetime is the
/// ceiling either way. One hour of slack absorbs sender clock skew.
pub const DIRECT_INVITE_LIFETIME_SECS: u64 = 24 * 3600 + 3600;

/// Park the Private-Channel keys a CATCH-UP bundle carries that we don't hold.
///
/// This is delivery, never authority: nothing is adopted here. The bundle's
/// self-certification is checked (a forged community binding never parks), the
/// keys are stashed, and [`judge_channel_key_vend`] rules on them against our own
/// fold after the next control follow. A vend that races its Grant therefore
/// parks quietly instead of being dropped, which is what makes an admin's grant
/// actually reach a member who is already in the community.
///
/// `inviter` is the seal author's npub (bech32) — recorded so the judge can
/// require an entitled vendor.
fn park_catch_up_channel_keys(
    community_id: &str,
    bundle_json: &str,
    inviter: &str,
) {
    use crate::community::v2::invite::CommunityInvite;
    let Ok(bundle) = CommunityInvite::from_bundle_json(bundle_json) else { return };
    // (1) Self-cert: the bundle cannot claim to be this community while naming a
    // different owner. Cheap, and it keeps garbage out of the park table.
    let (Some(cid), Some(owner), Some(salt)) = (
        crate::simd::hex::hex_to_bytes_32_checked(&bundle.community_id),
        crate::simd::hex::hex_to_bytes_32_checked(&bundle.owner),
        crate::simd::hex::hex_to_bytes_32_checked(&bundle.owner_salt),
    ) else {
        return;
    };
    if !crate::community::v2::derive::verify_community_id(&crate::community::CommunityId(cid), &owner, &salt) {
        log_warn!("[community] catch-up bundle failed self-certification — dropped");
        return;
    }
    let Ok(sender) = PublicKey::parse(inviter) else { return };
    let sender_hex = sender.to_hex();
    // The commit path's guard was captured before an await; re-check before the
    // writes below or a swap mid-commit parks account A's channel key into
    // account B's database.
    let held = crate::db::community::load_community_v2(&crate::community::CommunityId(cid)).ok().flatten();
    let mut parked_any = false;
    for grant in &bundle.channels {
        let Some(chan) = crate::simd::hex::hex_to_bytes_32_checked(&grant.id) else { continue };
        let Some(key) = crate::simd::hex::hex_to_bytes_32_checked(&grant.key) else { continue };
        // Only park what we LACK: a key at or below the held epoch is already
        // superseded, and a public channel's "key" is just the root.
        if let Some(c) = held.as_ref().and_then(|h| h.channel(&crate::community::ChannelId(chan))) {
            if !c.private || (c.key.is_some() && grant.epoch <= c.epoch.0) {
                continue;
            }
        }
        if let Err(e) = crate::db::community::park_channel_key(community_id, &grant.id, grant.epoch, &key, &sender_hex) {
            log_warn!("[community] parking a vended channel key failed: {e}");
            continue;
        }
        parked_any = true;
    }
    // Judge it now rather than whenever the next edition happens by. A vend that
    // lands AFTER its Grant folded changes no control state, so without a nudge
    // there is nothing left to trigger the re-judge.
    if parked_any {
        crate::community::v2::realtime::enqueue_follow(&crate::community::CommunityId(cid));
    }
}

fn expired_invite(expires_at: u64, rumor_created_at: u64) -> bool {
    let now = nostr_sdk::prelude::Timestamp::now().as_secs();
    if expires_at != 0 && expires_at <= now {
        return true;
    }
    rumor_created_at.saturating_add(DIRECT_INVITE_LIFETIME_SECS) <= now
}

#[cfg(test)]
mod revoked_reaction_tests {
    use super::*;

    #[tokio::test]
    async fn a_revoked_reaction_is_refused_under_any_wrapper() {
        let id = "7".repeat(64);
        let reaction = crate::types::Reaction {
            id: id.clone(),
            reference_id: "message".into(),
            author_id: "author".into(),
            emoji: "+".into(),
            emoji_url: None,
        };
        crate::state::note_message_deleted(&id);
        assert!(!commit_reaction(reaction, "contact", false, "any-wrapper", &NoOpEventHandler).await);
    }
}

#[cfg(test)]
mod invite_expiry_tests {
    use super::*;

    fn now() -> u64 {
        nostr_sdk::prelude::Timestamp::now().as_secs()
    }

    #[test]
    fn declared_deadline_is_honored() {
        assert!(expired_invite(now() - 10, now()));
        assert!(!expired_invite(now() + 3600, now()));
    }

    #[test]
    fn tagless_invites_die_at_the_recipient_lifetime() {
        // Fresh, no tag: parks.
        assert!(!expired_invite(0, now() - 3600));
        // A day-and-slack old, no tag: never parks — the class the archive
        // recovery resurrected (months-old invite to a re-founded community).
        assert!(expired_invite(0, now() - DIRECT_INVITE_LIFETIME_SECS));
        assert!(expired_invite(0, now() - 90 * 24 * 3600));
    }

    #[test]
    fn lifetime_caps_a_generous_declared_deadline() {
        // Sender promised a week — the recipient's 24h ceiling still wins.
        assert!(expired_invite(now() + 7 * 24 * 3600, now() - DIRECT_INVITE_LIFETIME_SECS));
    }
}

pub async fn commit_prepared_event(
    prepared: PreparedEvent,
    is_new: bool,
    handler: &dyn InboundEventHandler,
) -> bool {
    let session = crate::db::current_session();
    if !session.is_live() {
        return false;
    }
    match prepared {
        PreparedEvent::Processed { result, contact, sender, is_mine, wrapper_event_id, wrapper_event_id_bytes, wrapper_created_at, .. } => {
            // Cache wrapper for session dedup
            {
                let mut cache = WRAPPER_ID_CACHE.lock().await;
                cache.insert(wrapper_event_id_bytes);
            }

            // Blocked check — drop content from blocked contacts (wrapper still ledgered so
            // the dropped content never re-syncs)
            if !is_mine {
                let blocked = {
                    let state = crate::state::STATE.lock().await;
                    state.get_profile(&contact).is_some_and(|p| p.flags.is_blocked())
                };
                if blocked {
                    let _ = crate::db::wrappers::save_processed_wrapper(&wrapper_event_id_bytes, wrapper_created_at, crate::db::wrappers::TRANSPORT_NIP17);
                    return false;
                }
            }

            // Persist for cross-session dedup + negentropy — EXCEPT message rumors: the
            // ledger IS the negentropy fingerprint set, and a batching handler may defer the
            // message's save, so its wrapper must never be ledgered before its row lands (a
            // ledgered-but-unpersisted message reads as "have" and is never re-delivered).
            // Message wrappers ledger inside commit_dm_message / the batch-flush transaction.
            if !matches!(result, RumorProcessingResult::TextMessage(_) | RumorProcessingResult::FileAttachment(_)) {
                let _ = crate::db::wrappers::save_processed_wrapper(&wrapper_event_id_bytes, wrapper_created_at, crate::db::wrappers::TRANSPORT_NIP17);
            }

            match result {
                RumorProcessingResult::TextMessage(mut msg) => {
                    msg.wrapper_event_id = Some(wrapper_event_id.clone());
                    commit_dm_message(msg, &contact, is_mine, is_new, &wrapper_event_id, wrapper_event_id_bytes, wrapper_created_at, handler, false).await
                }
                RumorProcessingResult::FileAttachment(mut msg) => {
                    msg.wrapper_event_id = Some(wrapper_event_id.clone());
                    // If the sender's client (e.g. 0xChat) didn't ship `size` in
                    // the imeta tag, probe the URL via Content-Length so the
                    // frontend's auto-download gate has accurate metadata to
                    // decide on.
                    //
                    // Skip for self-echoes (is_mine): we just uploaded these
                    // files, the local Attachment.size is authoritative, and
                    // probing our own blossom URL right after upload is a
                    // correlation-fingerprint privacy regression.
                    //
                    // Each probe is bounded by a 3s outer timeout so a slow or
                    // dead server can't stall the inbound rumor pipeline. If
                    // the probe times out, we ship size=0 and the frontend
                    // falls back to a manual "Click to Download" affordance.
                    if !is_mine {
                        for att in &mut msg.attachments {
                            if att.size == 0
                                && (att.url.starts_with("https://") || att.url.starts_with("http://"))
                            {
                                if let Ok(Some(size)) = crate::rt::time::timeout(
                                    std::time::Duration::from_secs(3),
                                    crate::net::get_remote_file_size(&att.url),
                                ).await {
                                    att.size = size;
                                }
                            }
                        }
                    }
                    // A claim whose bytes are already on disk verifies at ARRIVAL,
                    // not at tap: the box renders downloaded from its first paint and
                    // no bytes are re-fetched. The ledger probe is an indexed lookup;
                    // the content hash only runs when a plausible candidate exists.
                    crate::db::attachments::verify_message_attachments(&mut msg, None).await;
                    commit_dm_message(msg, &contact, is_mine, is_new, &wrapper_event_id, wrapper_event_id_bytes, wrapper_created_at, handler, true).await
                }
                RumorProcessingResult::Reaction(reaction) => {
                    commit_reaction(reaction, &contact, is_mine, &wrapper_event_id, handler).await
                }
                RumorProcessingResult::Edit { message_id, new_content, edited_at, emoji_tags, mut event } => {
                    commit_edit(&mut event, &contact, &message_id, &new_content, edited_at, emoji_tags, &wrapper_event_id).await
                }
                RumorProcessingResult::TypingIndicator { profile_id, until } => {
                    let active_typers = {
                        let mut state = crate::state::STATE.lock().await;
                        state.update_typing_and_get_active(&contact, &profile_id, until)
                    };
                    crate::traits::emit_event("typing-update", &serde_json::json!({
                        "conversation_id": contact,
                        "typers": active_typers,
                    }));
                    false
                }
                RumorProcessingResult::PivxPayment { gift_code, amount_piv, address, message_id, mut event } => {
                    if crate::db::events::event_exists(&event.id).unwrap_or(false) {
                        return false;
                    }
                    event.wrapper_event_id = Some(wrapper_event_id.clone());
                    let ts = event.created_at;
                    let _ = crate::db::events::save_pivx_payment_event(&contact, event).await;
                    crate::traits::emit_event("pivx_payment_received", &serde_json::json!({
                        "conversation_id": contact,
                        "gift_code": gift_code, "amount_piv": amount_piv,
                        "address": address, "message_id": message_id,
                        "sender": sender.to_hex(), "is_mine": is_mine,
                        "at": ts * 1000,
                    }));
                    true
                }
                RumorProcessingResult::UnknownEvent(mut event) => {
                    event.wrapper_event_id = Some(wrapper_event_id.clone());
                    // Store unknown events for forward compatibility
                    if let Ok(chat_id) = crate::db::id_cache::get_or_create_chat_id(&contact) {
                        event.chat_id = chat_id;
                    }
                    let _ = crate::db::events::save_event(&event).await;
                    false
                }
                RumorProcessingResult::LeaveRequest { .. } => false,
                RumorProcessingResult::WebxdcPeerAdvertisement { event_id, topic_id, node_addr, sender_npub, created_at } => {
                    handler.on_webxdc_signal(&contact, &sender_npub, &topic_id, Some(&node_addr), &event_id, created_at);
                    false
                }
                RumorProcessingResult::WebxdcPeerLeft { event_id, topic_id, sender_npub, created_at } => {
                    handler.on_webxdc_signal(&contact, &sender_npub, &topic_id, None, &event_id, created_at);
                    false
                }
                #[cfg(feature = "calls")]
                RumorProcessingResult::CallSignal { call_id, signal, node_addr, sender_npub, created_at, video, media, .. } => {
                    // Only the DM's other party may signal; a group member cannot ring us through a channel.
                    if contact == sender_npub {
                        crate::calls::session::on_signal(&sender_npub, &call_id, &signal, node_addr.as_deref(), created_at, video.as_deref(), media.as_deref()).await;
                    }
                    false
                }
                #[cfg(not(feature = "calls"))]
                RumorProcessingResult::CallSignal { .. } => false,
                RumorProcessingResult::WallpaperChanged {
                    sender_npub, created_at, url, decryption_key, decryption_nonce,
                    plaintext_hash, mime, blur, dim, event_id,
                } => {
                    let _ = crate::wallpaper::apply_received_wallpaper(
                        &contact, &sender_npub, created_at, &url,
                        &decryption_key, &decryption_nonce,
                        plaintext_hash.as_deref(), mime.as_deref(),
                        blur, dim,
                        &event_id,
                    ).await;
                    // System event is saved inside apply_received_wallpaper.
                    // Return true so the caller treats this as a stored event.
                    true
                }
                RumorProcessingResult::DeletionRequest { target_event_id } => {
                    // A deletion targets a message OR a reaction (both event ids,
                    // never colliding). Try the message path; if it's not a known
                    // message, treat it as a reaction revocation.
                    if commit_deletion(&target_event_id, &contact, &sender, handler).await {
                        true
                    } else {
                        commit_reaction_deletion(&target_event_id, &sender).await
                    }
                }
                RumorProcessingResult::Ignored => false,
            }
        }
        PreparedEvent::CommunityInvite { invite, inviter, is_mine, wrapper_event_id_bytes, wrapper_created_at, rumor_created_at, expires_at } => {
            // Negentropy bookkeeping regardless of outcome (the outer wrapper id is
            // attacker-controlled, so it can't be the join-idempotency key — see below).
            {
                let mut cache = WRAPPER_ID_CACHE.lock().await;
                cache.insert(wrapper_event_id_bytes);
            }
            let _ = crate::db::wrappers::save_processed_wrapper(&wrapper_event_id_bytes, wrapper_created_at, crate::db::wrappers::TRANSPORT_NIP17);

            // Never park our own echoed invite.
            if is_mine {
                return false;
            }

            // Past the sender's deadline OR past the recipient-enforced 24h
            // lifetime (a catch-up sync of a stale wrap, a relay that ignores
            // expiry, a sender that never set the tag): never park it.
            if expired_invite(expires_at, rumor_created_at) {
                return false;
            }

            // Cap-check before touching the DB (a hostile bundle can declare an
            // unbounded channel/relay list).
            if let Err(e) = invite.validate() {
                log_warn!("[community] invite rejected: {}", e);
                return false;
            }

            // Idempotency on the INNER identity (community_id), NOT the wrapper id: a
            // replayed bundle re-wrapped under fresh ephemeral keys must not re-notify
            // or churn. If we already hold this Community, or already have it parked,
            // drop silently.
            let community_id = invite.community_id.clone();
            // PUBLIC input: a signature-valid invite can still carry a malformed id, so decode through
            // the SIMD-validated path (rejects non-hex / wrong length in-register).
            let already_held = crate::community::CommunityId(
                match crate::simd::hex::hex_to_bytes_32_checked(&community_id) {
                    Some(b) => b,
                    None => { log_warn!("[community] invite has malformed id"); return false; }
                },
            );
            if crate::db::community::community_exists(&already_held).unwrap_or(false) {
                return false;
            }
            if crate::db::community::pending_invite_exists(&community_id).unwrap_or(false) {
                return false;
            }

            // Supersession: a decline/leave tombstone suppresses any invite no newer than the
            // decision (so the un-deletable 3304 can't re-nag, and a sibling's decline propagated via
            // the synced list silences this device too). A STRICTLY-newer invite falls through and
            // parks — a deliberate re-invite resurfaces. Ordered on the inner rumor time, not the
            // NIP-59-backdated wrapper (which would make a fresh re-invite look older than the decline).
            if crate::community::list::tombstone_suppresses(&community_id, rumor_created_at) {
                return false;
            }

            // Park for explicit consent — do NOT join, subscribe, or dial the bundle's
            // relays here. The user accepts via the command layer.
            let bundle_json = match invite.to_json() {
                Ok(j) => j,
                Err(e) => { log_warn!("[community] invite re-serialize failed: {}", e); return false; }
            };
            match crate::db::community::save_pending_invite(&community_id, &bundle_json, &inviter, expires_at as i64) {
                Ok(true) => {
                    handler.on_community_invite(&community_id);
                    // Warm the community's first page in the background so a subsequent Accept opens
                    // populated instead of paying the join sync. RAM-only + best-effort; promotion on
                    // Join re-validates freshness. std::sync::Arc<crate::db::Session>'d so a mid-flight swap is a no-op.
                    let invite_warm = invite.clone();
                    let bg = crate::db::current_session();
                    crate::rt::spawn(async move {
                        if !bg.is_live() {
                            return;
                        }
                        crate::community::service::preload_community(&invite_warm).await;
                    });
                }
                Ok(false) => {} // raced — already parked
                Err(e) => log_warn!("[community] invite park failed: {}", e),
            }
            false
        }
        PreparedEvent::CommunityInviteV2 { bundle_json, community_id, inviter, is_mine, wrapper_event_id_bytes, wrapper_created_at, rumor_created_at, expires_at } => {
            {
                let mut cache = WRAPPER_ID_CACHE.lock().await;
                cache.insert(wrapper_event_id_bytes);
            }
            let _ = crate::db::wrappers::save_processed_wrapper(&wrapper_event_id_bytes, wrapper_created_at, crate::db::wrappers::TRANSPORT_NIP17);

            // Never park our own echoed invite.
            if is_mine {
                return false;
            }
            // Past the sender's deadline or the 24h lifetime — see the v1 arm.
            if expired_invite(expires_at, rumor_created_at) {
                return false;
            }
            // Idempotency on the INNER community_id (already validated hex), not the
            // attacker-controlled wrapper id: already held, or already parked → drop.
            let held = match crate::simd::hex::hex_to_bytes_32_checked(&community_id) {
                Some(b) => crate::community::CommunityId(b),
                None => return false,
            };
            // Already a member? Then this is not an invitation but a CATCH-UP: a
            // grant's Private-Channel key vend (CORD-03 "delivered on grant"), or
            // an admin healing us forward. Park the keys we lack for the judge to
            // rule on after the next control fold — dropping it here is why a
            // granted member (or bot) stayed silently keyless.
            if crate::db::community::community_exists(&held).unwrap_or(false) {
                park_catch_up_channel_keys(&community_id, &bundle_json, &inviter);
                return false;
            }
            if crate::db::community::pending_invite_exists(&community_id).unwrap_or(false) {
                return false;
            }
            // Supersession: a decline/leave tombstone (protocol-agnostic, keyed on
            // community_id) suppresses a re-wrapped invite no newer than the decision,
            // so a declined/left community can't be re-nagged by a fresh ephemeral wrap.
            // Ordered on the inner rumor time, not the NIP-59-backdated wrapper.
            if crate::community::list::tombstone_suppresses(&community_id, rumor_created_at) {
                return false;
            }
            // Park for explicit consent — do NOT join/subscribe here. Accept via the command layer.
            match crate::db::community::save_pending_invite(&community_id, &bundle_json, &inviter, expires_at as i64) {
                Ok(true) => handler.on_community_invite(&community_id),
                Ok(false) => {} // raced — already parked
                Err(e) => log_warn!("[community] v2 invite park failed: {}", e),
            }
            false
        }
        PreparedEvent::DedupSkip { wrapper_id_bytes, wrapper_created_at } => {
            // Persist wrapper timestamp for negentropy backfill (skip no-op writes).
            // Guarded: a cache-hit skip can name a wrapper whose message is still sitting in
            // a batch buffer (deferred ledger) — inserting it here would mark the message
            // "have" before its row exists. Only touch the ledger when the wrapper is
            // already ledgered (timestamp backfill) or its row is verifiably persisted
            // (legacy pre-ledger events, whose wrapper lives only on the events row).
            if wrapper_created_at > 0 {
                if crate::db::wrappers::processed_wrapper_exists(&wrapper_id_bytes) {
                    let _ = crate::db::wrappers::update_wrapper_timestamp(&wrapper_id_bytes, wrapper_created_at);
                } else {
                    let wrapper_hex = crate::simd::hex::bytes_to_hex_32(&wrapper_id_bytes);
                    if crate::db::events::wrapper_event_exists(&wrapper_hex).unwrap_or(false) {
                        let _ = crate::db::wrappers::save_processed_wrapper(&wrapper_id_bytes, wrapper_created_at, crate::db::wrappers::TRANSPORT_NIP17);
                    }
                }
            }
            false
        }
        PreparedEvent::ErrorSkip { wrapper_id_bytes, wrapper_created_at } => {
            let _ = crate::db::wrappers::save_processed_wrapper(&wrapper_id_bytes, wrapper_created_at, crate::db::wrappers::TRANSPORT_NIP17);
            false
        }
    }
}

/// Commit a DM text or file message (shared logic for both).
///
/// Owns this message's wrapper-ledger write (`processed_wrappers` = the negentropy
/// fingerprint set): the wrapper is ledgered only once the message row is durably handled —
/// immediately after a successful save, inside the batch-flush transaction when a handler
/// defers, or right away when the message is a known duplicate. An unledgered wrapper is
/// re-delivered by the next reconciliation, which is what makes a dropped batch recoverable.
#[allow(clippy::too_many_arguments)]
async fn commit_dm_message(
    mut msg: Message,
    contact: &str,
    _is_mine: bool,
    is_new: bool,
    wrapper_event_id: &str,
    wrapper_event_id_bytes: [u8; 32],
    wrapper_created_at: u64,
    handler: &dyn InboundEventHandler,
    is_file: bool,
) -> bool {
    let ledger_wrapper = || {
        let _ = crate::db::wrappers::save_processed_wrapper(
            &wrapper_event_id_bytes, wrapper_created_at, crate::db::wrappers::TRANSPORT_NIP17,
        );
    };
    // A tombstoned (user-deleted) message must never re-commit, whatever wrap
    // carried it — a retry wrap this device never ledgered can still deliver
    // the deleted rumor. Ledger the wrap so it stops re-delivering.
    if crate::state::was_message_deleted(&msg.id) {
        ledger_wrapper();
        return false;
    }

    // Dedup: check if message already in DB
    if let Ok(true) = crate::db::events::message_exists_in_db(&msg.id) {
        // Already in DB — try to backfill wrapper_event_id
        if let Ok(updated) = crate::db::events::update_wrapper_event_id(&msg.id, wrapper_event_id) {
            if !updated {
                let mut cache = WRAPPER_ID_CACHE.lock().await;
                cache.insert(wrapper_event_id_bytes);
            }
        }
        ledger_wrapper();
        return false;
    }

    // Populate reply context
    if !msg.replied_to.is_empty() {
        let _ = crate::db::events::populate_reply_context(&mut msg).await;
    }

    // Add to STATE (+ clear typing indicator for file senders)
    let added = {
        let mut state = crate::state::STATE.lock().await;
        let added = state.add_message_to_participant(contact, &msg);
        if is_file && added {
            state.update_typing_and_get_active(contact, contact, 0);
        }
        added
    };

    // A banned author's message is stored but never SHOWN: the queries that build a
    // page already drop it, and painting it live was the one way it still reached the
    // screen. Gated at emission rather than at persist so an unban restores their
    // history — the row is still there, the filter simply stops matching.
    let hidden_author = msg
        .npub
        .as_deref()
        .and_then(|author| {
            let community = crate::db::community::community_id_for_channel(contact).ok().flatten()?;
            nostr_sdk::prelude::PublicKey::parse(author)
                .ok()
                .map(|pk| crate::db::community::is_author_banned(&community, &pk))
        })
        .unwrap_or(false);

    if added && !hidden_author {
        // Emit to frontend
        crate::traits::emit_message_new(contact, &msg);

        // Platform callback (notifications, badge, etc.)
        if is_file {
            handler.on_file_received(contact, &msg, is_new);
        } else {
            handler.on_dm_received(contact, &msg, is_new);
        }

        // Save to DB — unless a bulk-sync handler owns batched persistence (the handler then
        // also owns the wrapper-ledger write, inside its flush transaction). On the immediate
        // path the wrapper ledgers only after a successful save: a failed save left unledgered
        // re-delivers on the next reconciliation instead of being lost.
        if !handler.buffer_persist(contact, &msg, Some((wrapper_event_id_bytes, wrapper_created_at)))
            && crate::db::events::save_message(contact, &msg).await.is_ok()
        {
            ledger_wrapper();
        }
    } else {
        // STATE-level duplicate: a same-session twin owns the row; this wrapper carried
        // nothing new, so ledger it now (parity with the old eager ledger).
        ledger_wrapper();
    }

    added
}

/// Commit a reaction event.
async fn commit_reaction(
    reaction: crate::types::Reaction,
    contact: &str,
    is_mine: bool,
    wrapper_event_id: &str,
    handler: &dyn InboundEventHandler,
) -> bool {
    // A revoked reaction can come back under a new wrapper id; its own id is the stable key.
    if crate::state::was_message_deleted(&reaction.id) {
        return false;
    }

    // Add to STATE
    let msg_for_emit = {
        let mut state = crate::state::STATE.lock().await;
        if let Some((chat_id, was_added)) = state.add_reaction_to_message(&reaction.reference_id, reaction.clone()) {
            if was_added {
                state.find_message(&reaction.reference_id)
                    .map(|(_, msg)| (chat_id, msg))
            } else { None }
        } else { None }
    };

    if let Some((chat_id, mut msg)) = msg_for_emit {
        crate::traits::emit_message_update(&chat_id, &reaction.reference_id, &mut msg).await;
        let _ = crate::db::events::save_message(&chat_id, &msg).await;
        handler.on_reaction_received(&chat_id, &msg);
    }

    // Always save reaction event with wrapper for dedup
    if let Ok(chat_id) = crate::db::id_cache::get_chat_id_by_identifier(contact) {
        let _ = crate::db::events::save_reaction_event(
            &reaction, chat_id, None, is_mine, Some(wrapper_event_id.to_string())
        ).await;
    }

    true
}

/// Commit a message edit.
async fn commit_edit(
    event: &mut crate::stored_event::StoredEvent,
    contact: &str,
    message_id: &str,
    new_content: &str,
    edited_at: u64,
    emoji_tags: Vec<crate::types::EmojiTag>,
    wrapper_event_id: &str,
) -> bool {
    if crate::db::events::event_exists(&event.id).unwrap_or(false) {
        return false;
    }
    if let Ok(chat_id) = crate::db::id_cache::get_chat_id_by_identifier(contact) {
        event.chat_id = chat_id;
    }
    event.wrapper_event_id = Some(wrapper_event_id.to_string());
    let _ = crate::db::events::save_event(event).await;

    let msg_for_emit = {
        let mut state = crate::state::STATE.lock().await;
        state.update_message_in_chat(contact, message_id, |msg| {
            msg.apply_edit(new_content.to_string(), edited_at, emoji_tags.clone());
        })
    };
    if let Some(mut msg) = msg_for_emit {
        crate::traits::emit_message_update(contact, message_id, &mut msg).await;
    }
    true
}

/// Commit a NIP-09 cooperative deletion request.
///
/// Authorization: only the original message's author can delete it
/// (matches NIP-09's `event.pubkey == deletion.pubkey` rule applied to
/// the inner rumor). For DMs, that means either the sender is `MY` (we
/// deleted from another device) or the sender is the chat counterpart
/// who originally sent that message. Anyone else's deletion is silently
/// ignored.
///
/// On success: drops the message from in-memory STATE, removes the row
/// from the events table, and emits `message_removed` so the frontend
/// can fade the row out — same code path as failed-message cleanup.
async fn commit_deletion(
    target_event_id: &str,
    contact: &str,
    sender: &PublicKey,
    handler: &dyn InboundEventHandler,
) -> bool {
    // Look up the original. If not present locally there's nothing to
    // delete — the deletion notice arrived before the original (rare),
    // or we never had it. Either way, no-op.
    //
    // KNOWN LIMITATION: late-binding deletions are not handled. If
    // the deletion arrives BEFORE the original (cold sync, out-of-order
    // relay delivery), we drop the deletion silently here, and when the
    // original arrives later it shows up unhidden. A future enhancement
    // would persist a `pending_deletions` table keyed by target id and
    // apply queued deletions when the target is committed in
    // commit_dm_message. The common case (deletion arrives after the
    // original) works correctly today.
    //
    // For DM rumors the `npub` field is intentionally empty: the chat
    // is between two parties, so the author is implicit from `mine`
    // (me if true, chat counterpart if false). We derive the original
    // author from that, since the rumor pubkey isn't stored.
    let (mine, chat_id) = {
        let state = crate::state::STATE.lock().await;
        match state.find_message(target_event_id) {
            Some((chat, msg)) => (msg.mine, chat.id.clone()),
            None => return false,
        }
    };

    // Authorization: deletion sender must match the original author.
    // For DMs:
    //   - mine == true:  original author == us (MY_PUBLIC_KEY).
    //                    Authorized if the deletion sender is also us
    //                    (i.e. came in via our own self-wrap from
    //                    another device, multi-device sync).
    //   - mine == false: original author == chat counterpart. Chat id
    //                    for a DM is the counterpart's npub, so we
    //                    parse it and compare against the deletion
    //                    sender.
    let authorized = if mine {
        match crate::state::my_public_key() {
            Some(my_pk) => *sender == my_pk,
            None => false,
        }
    } else {
        match nostr_sdk::prelude::PublicKey::from_bech32(&chat_id) {
            Ok(counterpart) => sender == &counterpart,
            Err(_) => false, // chat id wasn't an npub (shouldn't happen for DMs)
        }
    };
    if !authorized {
        eprintln!(
            "[NIP-17 cooperative-delete] unauthorized: sender {} not the author of target {} (mine={}, chat={})",
            sender.to_hex(), target_event_id, mine, chat_id
        );
        return false;
    }

    // Drop from in-memory state.
    let removed = {
        let mut state = crate::state::STATE.lock().await;
        state.remove_message(target_event_id)
    };
    let removed_msg = match removed {
        Some((_chat_id, msg)) => msg,
        None => return false,
    };
    // Tombstone BEFORE the row delete: the target may sit unflushed in a sync batch buffer
    // (delete_event below would no-op) — the flush consults this and drops it. The durable
    // row keeps the refusal across restarts: the sender's NIP-09 may never land on every
    // relay, and this side must not resurrect what it already agreed to drop.
    crate::state::note_message_deleted(target_event_id);
    if let Err(e) = crate::db::events::add_message_tombstone(target_event_id) {
        crate::log_warn!("[NIP-17 cooperative-delete] tombstone write failed: {}", e);
    }

    // Nuke any cached attachment files for this message — sender asked
    // for the message to disappear, and a downloaded file the receiver
    // never moved out of Vector's cache should go with it.
    //
    // Refcount filter: drop attachments still referenced by sibling
    // messages so we don't yank a cached file from messages that
    // still need it (Vector dedupes by SHA-256, so the same file
    // can back multiple messages). User-managed paths are also left
    // alone (canonicalize + starts_with check).
    let unique = crate::deletion::filter_unreferenced_attachments(
        target_event_id,
        removed_msg.attachments,
    ).await;
    crate::deletion::delete_cached_attachment_files_pub(&unique);

    // Drop from the events table.
    if let Err(e) = crate::db::events::delete_event(target_event_id).await {
        eprintln!(
            "[NIP-17 cooperative-delete] DB delete failed for {}: {}",
            target_event_id, e
        );
    }

    // Tell the frontend to fade the row out. Reuses the existing
    // message_removed event handled in main.js, so no new wiring.
    crate::traits::emit_event(
        "message_removed",
        &serde_json::json!({
            "id": target_event_id,
            "chat_id": &chat_id,
            "reason": "deleted-by-sender",
        }),
    );

    handler.on_message_deleted(&chat_id, target_event_id);
    let _ = contact;
    true
}

/// Apply a cooperative reaction revocation (NIP-09 k=7) from the reaction's
/// author. Removes the reaction from its parent message and drops the kind-7
/// row, then live-refreshes the parent's chips. Returns false if we don't hold
/// the reaction or the sender isn't its author.
async fn commit_reaction_deletion(target_reaction_id: &str, sender: &PublicKey) -> bool {
    let found = {
        let state = crate::state::STATE.lock().await;
        state.find_reaction(target_reaction_id)
    };
    let (chat_id, message_id, author_npub, _is_community) = match found {
        Some(v) => v,
        None => return false,
    };

    // Authorization: only the reaction's own author may revoke it.
    let authorized = nostr_sdk::prelude::PublicKey::parse(&author_npub)
        .map(|pk| pk == *sender)
        .unwrap_or(false);
    if !authorized {
        eprintln!(
            "[reaction-delete] unauthorized: sender {} is not the author of reaction {}",
            sender.to_hex(), target_reaction_id
        );
        return false;
    }

    crate::state::note_message_deleted(target_reaction_id);
    if let Err(e) = crate::db::events::add_message_tombstone(target_reaction_id) {
        crate::log_warn!("[reaction-delete] tombstone write failed: {}", e);
    }

    let updated = {
        let mut state = crate::state::STATE.lock().await;
        state.remove_reaction_from_message(&message_id, target_reaction_id)
    };
    let mut message = match updated {
        Some((_cid, msg)) => msg,
        None => return false,
    };

    // save_message is additive for reactions, so the kind-7 row must be
    // dropped explicitly or it resurrects on reload.
    if let Err(e) = crate::db::events::delete_event(target_reaction_id).await {
        eprintln!("[reaction-delete] DB delete failed for {}: {}", target_reaction_id, e);
    }

    crate::traits::emit_message_update(&chat_id, &message_id, &mut message).await;
    true
}

// ============================================================================
// Convenience: single-call event processing
// ============================================================================

/// Process a single event through the full pipeline (prepare + commit).
///
/// Gets client and public key from globals. For callers that manage their
/// own notification loop but want the full vector-core processing pipeline.
pub async fn process_event(
    event: Event,
    is_new: bool,
    handler: &dyn InboundEventHandler,
) -> std::result::Result<bool, String> {
    let client = crate::state::nostr_client()
        .ok_or_else(|| "Nostr client not initialized".to_string())?;
    let my_pk = crate::state::my_public_key()
        .ok_or_else(|| "Public key not initialized".to_string())?;
    let prepared = prepare_event(event, &client, my_pk).await;
    Ok(commit_prepared_event(prepared, is_new, handler).await)
}

#[cfg(test)]
mod signals_only_tests {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};

    #[derive(Default)]
    struct Counts {
        signals: AtomicUsize,
        messages: AtomicUsize,
    }

    impl InboundEventHandler for Counts {
        fn on_webxdc_signal(&self, _: &str, _: &str, _: &str, _: Option<&str>, _: &str, _: u64) {
            self.signals.fetch_add(1, Ordering::Relaxed);
        }
        fn on_community_webxdc(&self, _: &str, _: &str, _: &str, _: Option<&str>, _: &str, _: u64) {
            self.signals.fetch_add(1, Ordering::Relaxed);
        }
        fn on_dm_received(&self, _: &str, _: &Message, _: bool) {
            self.messages.fetch_add(1, Ordering::Relaxed);
        }
        fn on_file_received(&self, _: &str, _: &Message, _: bool) {
            self.messages.fetch_add(1, Ordering::Relaxed);
        }
    }

    #[test]
    fn a_catch_up_sync_passes_on_signals_and_nothing_else() {
        let inner = Counts::default();
        let sync = LiveOnly::signals(&inner);
        sync.on_webxdc_signal("npub1a", "npub1a", "T", Some("addr"), "e1", 1);
        sync.on_community_webxdc("chan", "npub1a", "T", None, "e2", 1);
        sync.on_dm_received("npub1a", &Message { at: u64::MAX, ..Default::default() }, true);
        assert_eq!(inner.signals.load(Ordering::Relaxed), 2);
        assert_eq!(inner.messages.load(Ordering::Relaxed), 0, "messages from a sync are history");
    }

    #[test]
    fn a_startup_sync_passes_on_messages_sent_since_listening_began() {
        let inner = Counts::default();
        let sync = LiveOnly::from(&inner, 5_000);
        sync.on_dm_received("npub1a", &Message { at: 4_999, ..Default::default() }, true);
        assert_eq!(inner.messages.load(Ordering::Relaxed), 0, "before: history");
        sync.on_dm_received("npub1a", &Message { at: 5_000, ..Default::default() }, true);
        sync.on_file_received("npub1a", &Message { at: 6_000, ..Default::default() }, true);
        assert_eq!(inner.messages.load(Ordering::Relaxed), 2, "since: live");
    }
}