vector_core/
event_handler.rs

1//! Event handler — gift wrap receive, unwrap, process, commit pipeline.
2//!
3//! Two-phase architecture:
4//! - **Phase 1** (`prepare_event`): Parallel-safe — dedup, unwrap, process_rumor
5//! - **Phase 2** (`commit_prepared_event`): Sequential — save DB, update STATE, emit
6//!
7//! Platform-specific behavior (notifications) handled by `InboundEventHandler` trait.
8
9use nostr_sdk::prelude::*;
10
11use crate::rumor::{RumorProcessingResult, RumorEvent, RumorContext, ConversationType, process_rumor};
12use crate::types::Message;
13use crate::state::WRAPPER_ID_CACHE;
14
15/// Platform-specific callbacks for inbound event processing.
16///
17/// Same pattern as SendCallback/ProfileSyncHandler — trait with default no-ops.
18/// Platforms implement only the hooks they need.
19pub trait InboundEventHandler: Send + Sync {
20    /// A DM text message was received and committed to STATE + DB.
21    fn on_dm_received(&self, _chat_id: &str, _msg: &Message, _is_new: bool) {}
22
23    /// A DM file attachment was received and committed to STATE + DB.
24    fn on_file_received(&self, _chat_id: &str, _msg: &Message, _is_new: bool) {}
25
26    /// A reaction was received and applied to a message.
27    fn on_reaction_received(&self, _chat_id: &str, _msg: &Message) {}
28
29    /// A previously-stored message was deleted by its sender (Layer 2
30    /// cooperative hide via NIP-09 over NIP-17). Frontend drops the row.
31    fn on_message_deleted(&self, _chat_id: &str, _message_id: &str) {}
32
33    /// A Community invite was received over a gift wrap and the local user was
34    /// joined (member-view Community persisted). Platform refreshes the Community
35    /// subscription so messages start flowing, and surfaces the new Community in the UI.
36    fn on_community_invite(&self, _community_id: &str) {}
37
38    // --- Community realtime (Concord channel events; `chat_id` is the channel id hex) ---
39
40    /// A new Community channel message was received, ingested into STATE, and persisted.
41    fn on_community_message(&self, _chat_id: &str, _msg: &Message, _is_new: bool) {}
42
43    /// A reaction or edit was applied to an existing Community message. `target_id` is the
44    /// affected message; `msg` is the live-updated view.
45    fn on_community_update(&self, _chat_id: &str, _target_id: &str, _msg: &Message) {}
46
47    /// A Community message was removed (cooperative delete / moderation tombstone).
48    fn on_community_removed(&self, _chat_id: &str, _target_id: &str) {}
49
50    /// A join/leave presence announcement. `created_at` is the authenticated inner timestamp;
51    /// `invited_by`/`invited_label` carry invite attribution when present.
52    #[allow(clippy::too_many_arguments)]
53    fn on_community_presence(
54        &self,
55        _chat_id: &str,
56        _npub: &str,
57        _joined: bool,
58        _event_id: &str,
59        _created_at: u64,
60        _invited_by: Option<&str>,
61        _invited_label: Option<&str>,
62    ) {}
63
64    /// A Community typing indicator (ephemeral). `until` is the unix-secs the typer stops being active.
65    fn on_community_typing(&self, _chat_id: &str, _npub: &str, _until: u64) {}
66
67    /// A WebXDC realtime peer signal. `node_addr` = `Some` advertises an Iroh node, `None` = peer-left.
68    #[allow(clippy::too_many_arguments)]
69    fn on_community_webxdc(
70        &self,
71        _chat_id: &str,
72        _npub: &str,
73        _topic_id: &str,
74        _node_addr: Option<&str>,
75        _event_id: &str,
76        _created_at: u64,
77    ) {}
78
79    /// The local user was removed from a Community (kick / ban / a leave authored on another device).
80    /// Local data is torn down (epoch keys retained); the platform surfaces it + refreshes subs.
81    fn on_community_self_removed(&self, _community_id: &str) {}
82
83    /// A Community's control plane was refreshed in realtime (banlist/roles/metadata/mode change,
84    /// or a re-founding followed). The platform re-reads display state.
85    fn on_community_refreshed(&self, _community_id: &str) {}
86}
87
88/// No-op handler for CLI/tests.
89pub struct NoOpEventHandler;
90impl InboundEventHandler for NoOpEventHandler {}
91
92/// Result of Phase 1 (prepare_event) — everything needed for sequential commit.
93pub enum PreparedEvent {
94    /// Fully processed DM rumor — ready for state commit.
95    Processed {
96        result: RumorProcessingResult,
97        contact: String,
98        sender: PublicKey,
99        is_mine: bool,
100        wrapper_event_id: String,
101        wrapper_event_id_bytes: [u8; 32],
102        wrapper_created_at: u64,
103        /// Time spent on ECDH + ChaCha20Poly1305 decryption (nanoseconds)
104        unwrap_ns: u64,
105        /// Time spent on rumor parsing (nanoseconds)
106        parse_ns: u64,
107    },
108    /// Community invite bundle (kind 3304) — parked for explicit user consent.
109    CommunityInvite {
110        invite: crate::community::invite::CommunityInvite,
111        /// Inviter's npub (bech32) — shown in the pending-invite UI.
112        inviter: String,
113        is_mine: bool,
114        wrapper_event_id_bytes: [u8; 32],
115        wrapper_created_at: u64,
116    },
117    /// Duplicate event — just persist wrapper for negentropy.
118    DedupSkip {
119        wrapper_id_bytes: [u8; 32],
120        wrapper_created_at: u64,
121    },
122    /// Error during unwrap/processing — persist wrapper for negentropy.
123    ErrorSkip {
124        wrapper_id_bytes: [u8; 32],
125        wrapper_created_at: u64,
126    },
127}
128
129/// Phase 1: Prepare an event for commit (parallel-safe, no state mutation).
130///
131/// Performs dedup check, gift wrap decryption, and rumor parsing.
132/// Safe to call from multiple tokio tasks concurrently.
133pub async fn prepare_event(
134    event: Event,
135    client: &Client,
136    my_public_key: PublicKey,
137) -> PreparedEvent {
138    let wrapper_created_at = event.created_at.as_secs();
139    let wrapper_event_id_bytes: [u8; 32] = event.id.to_bytes();
140    let wrapper_event_id = event.id.to_hex();
141
142    // Dedup: in-memory cache first, then DB fallback
143    {
144        let cache = WRAPPER_ID_CACHE.lock().await;
145        if cache.contains(&wrapper_event_id_bytes) {
146            return PreparedEvent::DedupSkip { wrapper_id_bytes: wrapper_event_id_bytes, wrapper_created_at };
147        }
148    }
149
150    if let Ok(true) = crate::db::events::wrapper_event_exists(&wrapper_event_id) {
151        return PreparedEvent::DedupSkip { wrapper_id_bytes: wrapper_event_id_bytes, wrapper_created_at };
152    }
153
154    // Unwrap gift wrap (CPU-bound ECDH + ChaCha20Poly1305)
155    let unwrap_start = std::time::Instant::now();
156    let (rumor, sender) = match client.unwrap_gift_wrap(&event).await {
157        Ok(UnwrappedGift { rumor, sender }) => (rumor, sender),
158        Err(_) => return PreparedEvent::ErrorSkip {
159            wrapper_id_bytes: wrapper_event_id_bytes, wrapper_created_at,
160        },
161    };
162
163    let unwrap_ns = unwrap_start.elapsed().as_nanos() as u64;
164
165    let is_mine = sender == my_public_key;
166    let contact = if is_mine {
167        rumor.tags.public_keys().next()
168            .and_then(|pk| pk.to_bech32().ok())
169            .unwrap_or_else(|| sender.to_bech32().unwrap_or_default())
170    } else {
171        sender.to_bech32().unwrap_or_default()
172    };
173
174    // Skip NIP-17 group messages (multiple p-tags) — Vector DMs are 1:1
175    if rumor.tags.public_keys().count() > 1 {
176        return PreparedEvent::ErrorSkip {
177            wrapper_id_bytes: wrapper_event_id_bytes, wrapper_created_at,
178        };
179    }
180
181    // Community invite (carrier) — a join, not a chat message. Recognized before
182    // process_rumor so it never lands as an UnknownEvent in the DM thread.
183    if rumor.kind == Kind::Custom(crate::stored_event::event_kind::COMMUNITY_INVITE_BUNDLE) {
184        return match crate::community::invite::parse_invite_rumor(rumor.kind, &rumor.content) {
185            Some(invite) => PreparedEvent::CommunityInvite {
186                invite, inviter: contact.clone(), is_mine, wrapper_event_id_bytes, wrapper_created_at,
187            },
188            None => PreparedEvent::ErrorSkip { wrapper_id_bytes: wrapper_event_id_bytes, wrapper_created_at },
189        };
190    }
191
192    // Build RumorEvent for processing
193    let Some(rumor_id) = rumor.id else {
194        return PreparedEvent::ErrorSkip {
195            wrapper_id_bytes: wrapper_event_id_bytes, wrapper_created_at,
196        };
197    };
198
199    let rumor_event = RumorEvent {
200        id: rumor_id,
201        kind: rumor.kind,
202        content: rumor.content,
203        tags: rumor.tags,
204        created_at: rumor.created_at,
205        pubkey: rumor.pubkey,
206    };
207    let rumor_context = RumorContext {
208        sender,
209        is_mine,
210        conversation_id: contact.clone(),
211        conversation_type: ConversationType::DirectMessage,
212    };
213
214    let parse_start = std::time::Instant::now();
215    let download_dir = crate::db::get_download_dir();
216    match process_rumor(rumor_event, rumor_context, &download_dir) {
217        Ok(result) => {
218            let parse_ns = parse_start.elapsed().as_nanos() as u64;
219            PreparedEvent::Processed {
220                result, contact, sender, is_mine,
221                wrapper_event_id, wrapper_event_id_bytes, wrapper_created_at,
222                unwrap_ns, parse_ns,
223            }
224        }
225        Err(e) => {
226            log_warn!("[EventHandler] Failed to process rumor: {}", e);
227            PreparedEvent::ErrorSkip {
228                wrapper_id_bytes: wrapper_event_id_bytes, wrapper_created_at,
229            }
230        }
231    }
232}
233
234// ============================================================================
235// Phase 2: Commit — sequential state mutation, DB save, emit
236// ============================================================================
237
238/// Phase 2: commit a prepared event (sequential — not parallel-safe).
239/// Saves to DB, updates STATE, emits to frontend, calls handler hooks.
240/// Returns true if a new displayable message was committed.
241///
242/// Session-safety: captures the generation at the first line. If a swap
243/// occurred between `prepare_event()` and here (e.g. long-running
244/// negentropy fetch queued events for commit), bail before any STATE /
245/// DB write. Centralized so individual spawn sites (sync.rs fetch_messages,
246/// archive task, sync_dms, subscription_handler) don't have to wrap.
247pub async fn commit_prepared_event(
248    prepared: PreparedEvent,
249    is_new: bool,
250    handler: &dyn InboundEventHandler,
251) -> bool {
252    let session = crate::state::SessionGuard::capture();
253    if !session.is_valid() {
254        return false;
255    }
256    match prepared {
257        PreparedEvent::Processed { result, contact, sender, is_mine, wrapper_event_id, wrapper_event_id_bytes, wrapper_created_at, .. } => {
258            // Cache wrapper for session dedup
259            {
260                let mut cache = WRAPPER_ID_CACHE.lock().await;
261                cache.insert(wrapper_event_id_bytes);
262            }
263            // Persist for cross-session dedup + negentropy
264            let _ = crate::db::wrappers::save_processed_wrapper(&wrapper_event_id_bytes, wrapper_created_at, crate::db::wrappers::TRANSPORT_NIP17);
265
266            // Blocked check — drop content from blocked contacts (wrapper still persisted for negentropy)
267            if !is_mine {
268                let state = crate::state::STATE.lock().await;
269                if state.get_profile(&contact).map_or(false, |p| p.flags.is_blocked()) {
270                    return false;
271                }
272            }
273
274            match result {
275                RumorProcessingResult::TextMessage(mut msg) => {
276                    msg.wrapper_event_id = Some(wrapper_event_id.clone());
277                    commit_dm_message(msg, &contact, is_mine, is_new, &wrapper_event_id, wrapper_event_id_bytes, handler, false).await
278                }
279                RumorProcessingResult::FileAttachment(mut msg) => {
280                    msg.wrapper_event_id = Some(wrapper_event_id.clone());
281                    // If the sender's client (e.g. 0xChat) didn't ship `size` in
282                    // the imeta tag, probe the URL via Content-Length so the
283                    // frontend's auto-download gate has accurate metadata to
284                    // decide on.
285                    //
286                    // Skip for self-echoes (is_mine): we just uploaded these
287                    // files, the local Attachment.size is authoritative, and
288                    // probing our own blossom URL right after upload is a
289                    // correlation-fingerprint privacy regression.
290                    //
291                    // Each probe is bounded by a 3s outer timeout so a slow or
292                    // dead server can't stall the inbound rumor pipeline. If
293                    // the probe times out, we ship size=0 and the frontend
294                    // falls back to a manual "Click to Download" affordance.
295                    if !is_mine {
296                        for att in &mut msg.attachments {
297                            if att.size == 0
298                                && (att.url.starts_with("https://") || att.url.starts_with("http://"))
299                            {
300                                if let Ok(Some(size)) = tokio::time::timeout(
301                                    std::time::Duration::from_secs(3),
302                                    crate::net::get_remote_file_size(&att.url),
303                                ).await {
304                                    att.size = size;
305                                }
306                            }
307                        }
308                    }
309                    commit_dm_message(msg, &contact, is_mine, is_new, &wrapper_event_id, wrapper_event_id_bytes, handler, true).await
310                }
311                RumorProcessingResult::Reaction(reaction) => {
312                    commit_reaction(reaction, &contact, is_mine, &wrapper_event_id, handler).await
313                }
314                RumorProcessingResult::Edit { message_id, new_content, edited_at, emoji_tags, mut event } => {
315                    commit_edit(&mut event, &contact, &message_id, &new_content, edited_at, emoji_tags, &wrapper_event_id).await
316                }
317                RumorProcessingResult::TypingIndicator { profile_id, until } => {
318                    let active_typers = {
319                        let mut state = crate::state::STATE.lock().await;
320                        state.update_typing_and_get_active(&contact, &profile_id, until)
321                    };
322                    crate::traits::emit_event("typing-update", &serde_json::json!({
323                        "conversation_id": contact,
324                        "typers": active_typers,
325                    }));
326                    false
327                }
328                RumorProcessingResult::PivxPayment { gift_code, amount_piv, address, message_id, mut event } => {
329                    if crate::db::events::event_exists(&event.id).unwrap_or(false) {
330                        return false;
331                    }
332                    event.wrapper_event_id = Some(wrapper_event_id.clone());
333                    let ts = event.created_at;
334                    let _ = crate::db::events::save_pivx_payment_event(&contact, event).await;
335                    crate::traits::emit_event("pivx_payment_received", &serde_json::json!({
336                        "conversation_id": contact,
337                        "gift_code": gift_code, "amount_piv": amount_piv,
338                        "address": address, "message_id": message_id,
339                        "sender": sender.to_hex(), "is_mine": is_mine,
340                        "at": ts * 1000,
341                    }));
342                    true
343                }
344                RumorProcessingResult::UnknownEvent(mut event) => {
345                    event.wrapper_event_id = Some(wrapper_event_id.clone());
346                    // Store unknown events for forward compatibility
347                    if let Ok(chat_id) = crate::db::id_cache::get_or_create_chat_id(&contact) {
348                        event.chat_id = chat_id;
349                    }
350                    let _ = crate::db::events::save_event(&event).await;
351                    false
352                }
353                RumorProcessingResult::LeaveRequest { .. } => false,
354                RumorProcessingResult::WebxdcPeerAdvertisement { .. } |
355                RumorProcessingResult::WebxdcPeerLeft { .. } => {
356                    // WebXDC is platform-specific — handled by src-tauri directly
357                    false
358                }
359                RumorProcessingResult::WallpaperChanged {
360                    sender_npub, created_at, url, decryption_key, decryption_nonce,
361                    plaintext_hash, mime, blur, dim, event_id,
362                } => {
363                    let _ = crate::wallpaper::apply_received_wallpaper(
364                        &contact, &sender_npub, created_at, &url,
365                        &decryption_key, &decryption_nonce,
366                        plaintext_hash.as_deref(), mime.as_deref(),
367                        blur, dim,
368                        &event_id,
369                    ).await;
370                    // System event is saved inside apply_received_wallpaper.
371                    // Return true so the caller treats this as a stored event.
372                    true
373                }
374                RumorProcessingResult::DeletionRequest { target_event_id } => {
375                    commit_deletion(&target_event_id, &contact, &sender, handler).await
376                }
377                RumorProcessingResult::Ignored => false,
378            }
379        }
380        PreparedEvent::CommunityInvite { invite, inviter, is_mine, wrapper_event_id_bytes, wrapper_created_at } => {
381            // Negentropy bookkeeping regardless of outcome (the outer wrapper id is
382            // attacker-controlled, so it can't be the join-idempotency key — see below).
383            {
384                let mut cache = WRAPPER_ID_CACHE.lock().await;
385                cache.insert(wrapper_event_id_bytes);
386            }
387            let _ = crate::db::wrappers::save_processed_wrapper(&wrapper_event_id_bytes, wrapper_created_at, crate::db::wrappers::TRANSPORT_NIP17);
388
389            // Never park our own echoed invite.
390            if is_mine {
391                return false;
392            }
393
394            // Cap-check before touching the DB (a hostile bundle can declare an
395            // unbounded channel/relay list).
396            if let Err(e) = invite.validate() {
397                log_warn!("[community] invite rejected: {}", e);
398                return false;
399            }
400
401            // Idempotency on the INNER identity (community_id), NOT the wrapper id: a
402            // replayed bundle re-wrapped under fresh ephemeral keys must not re-notify
403            // or churn. If we already hold this Community, or already have it parked,
404            // drop silently.
405            let community_id = invite.community_id.clone();
406            // PUBLIC input: a signature-valid invite can still carry a malformed id, so decode through
407            // the SIMD-validated path (rejects non-hex / wrong length in-register).
408            let already_held = crate::community::CommunityId(
409                match crate::simd::hex::hex_to_bytes_32_checked(&community_id) {
410                    Some(b) => b,
411                    None => { log_warn!("[community] invite has malformed id"); return false; }
412                },
413            );
414            if crate::db::community::community_exists(&already_held).unwrap_or(false) {
415                return false;
416            }
417            if crate::db::community::pending_invite_exists(&community_id).unwrap_or(false) {
418                return false;
419            }
420
421            // Supersession: a decline/leave tombstone suppresses any invite no newer than the
422            // decision (so the un-deletable 3304 can't re-nag, and a sibling's decline propagated via
423            // the synced list silences this device too). A STRICTLY-newer invite falls through and
424            // parks — a deliberate re-invite resurfaces. `wrapper_created_at` is outer-send seconds.
425            if crate::community::list::tombstone_suppresses(&community_id, wrapper_created_at) {
426                return false;
427            }
428
429            // Park for explicit consent — do NOT join, subscribe, or dial the bundle's
430            // relays here. The user accepts via the command layer.
431            let bundle_json = match invite.to_json() {
432                Ok(j) => j,
433                Err(e) => { log_warn!("[community] invite re-serialize failed: {}", e); return false; }
434            };
435            match crate::db::community::save_pending_invite(&community_id, &bundle_json, &inviter) {
436                Ok(true) => {
437                    handler.on_community_invite(&community_id);
438                    // Warm the community's first page in the background so a subsequent Accept opens
439                    // populated instead of paying the join sync. RAM-only + best-effort; promotion on
440                    // Join re-validates freshness. SessionGuard'd so a mid-flight swap is a no-op.
441                    let invite_warm = invite.clone();
442                    let bg = crate::state::SessionGuard::capture();
443                    tokio::spawn(async move {
444                        if !bg.is_valid() {
445                            return;
446                        }
447                        crate::community::service::preload_community(&invite_warm).await;
448                    });
449                }
450                Ok(false) => {} // raced — already parked
451                Err(e) => log_warn!("[community] invite park failed: {}", e),
452            }
453            false
454        }
455        PreparedEvent::DedupSkip { wrapper_id_bytes, wrapper_created_at } => {
456            // Persist wrapper timestamp for negentropy backfill (skip no-op writes)
457            if wrapper_created_at > 0 {
458                let _ = crate::db::wrappers::update_wrapper_timestamp(&wrapper_id_bytes, wrapper_created_at);
459            }
460            false
461        }
462        PreparedEvent::ErrorSkip { wrapper_id_bytes, wrapper_created_at } => {
463            let _ = crate::db::wrappers::save_processed_wrapper(&wrapper_id_bytes, wrapper_created_at, crate::db::wrappers::TRANSPORT_NIP17);
464            false
465        }
466    }
467}
468
469/// Commit a DM text or file message (shared logic for both).
470async fn commit_dm_message(
471    mut msg: Message,
472    contact: &str,
473    _is_mine: bool,
474    is_new: bool,
475    wrapper_event_id: &str,
476    wrapper_event_id_bytes: [u8; 32],
477    handler: &dyn InboundEventHandler,
478    is_file: bool,
479) -> bool {
480    // Dedup: check if message already in DB
481    if let Ok(true) = crate::db::events::message_exists_in_db(&msg.id) {
482        // Already in DB — try to backfill wrapper_event_id
483        if let Ok(updated) = crate::db::events::update_wrapper_event_id(&msg.id, wrapper_event_id) {
484            if !updated {
485                let mut cache = WRAPPER_ID_CACHE.lock().await;
486                cache.insert(wrapper_event_id_bytes);
487            }
488        }
489        return false;
490    }
491
492    // Populate reply context
493    if !msg.replied_to.is_empty() {
494        let _ = crate::db::events::populate_reply_context(&mut msg).await;
495    }
496
497    // Add to STATE (+ clear typing indicator for file senders)
498    let added = {
499        let mut state = crate::state::STATE.lock().await;
500        let added = state.add_message_to_participant(contact, msg.clone());
501        if is_file && added {
502            state.update_typing_and_get_active(contact, contact, 0);
503        }
504        added
505    };
506
507    if added {
508        // Emit to frontend
509        crate::traits::emit_event("message_new", &serde_json::json!({
510            "message": &msg,
511            "chat_id": contact
512        }));
513
514        // Platform callback (notifications, badge, etc.)
515        if is_file {
516            handler.on_file_received(contact, &msg, is_new);
517        } else {
518            handler.on_dm_received(contact, &msg, is_new);
519        }
520
521        // Save to DB
522        let _ = crate::db::events::save_message(contact, &msg).await;
523    }
524
525    added
526}
527
528/// Commit a reaction event.
529async fn commit_reaction(
530    reaction: crate::types::Reaction,
531    contact: &str,
532    is_mine: bool,
533    wrapper_event_id: &str,
534    handler: &dyn InboundEventHandler,
535) -> bool {
536    // Add to STATE
537    let msg_for_emit = {
538        let mut state = crate::state::STATE.lock().await;
539        if let Some((chat_id, was_added)) = state.add_reaction_to_message(&reaction.reference_id, reaction.clone()) {
540            if was_added {
541                state.find_message(&reaction.reference_id)
542                    .map(|(_, msg)| (chat_id, msg))
543            } else { None }
544        } else { None }
545    };
546
547    if let Some((chat_id, msg)) = msg_for_emit {
548        crate::traits::emit_event("message_update", &serde_json::json!({
549            "old_id": &reaction.reference_id,
550            "message": &msg,
551            "chat_id": &chat_id
552        }));
553        let _ = crate::db::events::save_message(&chat_id, &msg).await;
554        handler.on_reaction_received(&chat_id, &msg);
555    }
556
557    // Always save reaction event with wrapper for dedup
558    if let Ok(chat_id) = crate::db::id_cache::get_chat_id_by_identifier(contact) {
559        let _ = crate::db::events::save_reaction_event(
560            &reaction, chat_id, None, is_mine, Some(wrapper_event_id.to_string())
561        ).await;
562    }
563
564    true
565}
566
567/// Commit a message edit.
568async fn commit_edit(
569    event: &mut crate::stored_event::StoredEvent,
570    contact: &str,
571    message_id: &str,
572    new_content: &str,
573    edited_at: u64,
574    emoji_tags: Vec<crate::types::EmojiTag>,
575    wrapper_event_id: &str,
576) -> bool {
577    if crate::db::events::event_exists(&event.id).unwrap_or(false) {
578        return false;
579    }
580    if let Ok(chat_id) = crate::db::id_cache::get_chat_id_by_identifier(contact) {
581        event.chat_id = chat_id;
582    }
583    event.wrapper_event_id = Some(wrapper_event_id.to_string());
584    let _ = crate::db::events::save_event(event).await;
585
586    let msg_for_emit = {
587        let mut state = crate::state::STATE.lock().await;
588        state.update_message_in_chat(contact, message_id, |msg| {
589            msg.apply_edit(new_content.to_string(), edited_at, emoji_tags.clone());
590        })
591    };
592    if let Some(msg) = msg_for_emit {
593        crate::traits::emit_event("message_update", &serde_json::json!({
594            "old_id": message_id,
595            "message": msg,
596            "chat_id": contact
597        }));
598    }
599    true
600}
601
602/// Commit a NIP-09 cooperative deletion request.
603///
604/// Authorization: only the original message's author can delete it
605/// (matches NIP-09's `event.pubkey == deletion.pubkey` rule applied to
606/// the inner rumor). For DMs, that means either the sender is `MY` (we
607/// deleted from another device) or the sender is the chat counterpart
608/// who originally sent that message. Anyone else's deletion is silently
609/// ignored.
610///
611/// On success: drops the message from in-memory STATE, removes the row
612/// from the events table, and emits `message_removed` so the frontend
613/// can fade the row out — same code path as failed-message cleanup.
614async fn commit_deletion(
615    target_event_id: &str,
616    contact: &str,
617    sender: &PublicKey,
618    handler: &dyn InboundEventHandler,
619) -> bool {
620    // Look up the original. If not present locally there's nothing to
621    // delete — the deletion notice arrived before the original (rare),
622    // or we never had it. Either way, no-op.
623    //
624    // KNOWN LIMITATION: late-binding deletions are not handled. If
625    // the deletion arrives BEFORE the original (cold sync, out-of-order
626    // relay delivery), we drop the deletion silently here, and when the
627    // original arrives later it shows up unhidden. A future enhancement
628    // would persist a `pending_deletions` table keyed by target id and
629    // apply queued deletions when the target is committed in
630    // commit_dm_message. The common case (deletion arrives after the
631    // original) works correctly today.
632    //
633    // For DM rumors the `npub` field is intentionally empty: the chat
634    // is between two parties, so the author is implicit from `mine`
635    // (me if true, chat counterpart if false). We derive the original
636    // author from that, since the rumor pubkey isn't stored.
637    let (mine, chat_id) = {
638        let state = crate::state::STATE.lock().await;
639        match state.find_message(target_event_id) {
640            Some((chat, msg)) => (msg.mine, chat.id.clone()),
641            None => return false,
642        }
643    };
644
645    // Authorization: deletion sender must match the original author.
646    // For DMs:
647    //   - mine == true:  original author == us (MY_PUBLIC_KEY).
648    //                    Authorized if the deletion sender is also us
649    //                    (i.e. came in via our own self-wrap from
650    //                    another device, multi-device sync).
651    //   - mine == false: original author == chat counterpart. Chat id
652    //                    for a DM is the counterpart's npub, so we
653    //                    parse it and compare against the deletion
654    //                    sender.
655    let authorized = if mine {
656        match crate::state::my_public_key() {
657            Some(my_pk) => *sender == my_pk,
658            None => false,
659        }
660    } else {
661        match nostr_sdk::PublicKey::from_bech32(&chat_id) {
662            Ok(counterpart) => sender == &counterpart,
663            Err(_) => false, // chat id wasn't an npub (shouldn't happen for DMs)
664        }
665    };
666    if !authorized {
667        eprintln!(
668            "[NIP-17 cooperative-delete] unauthorized: sender {} not the author of target {} (mine={}, chat={})",
669            sender.to_hex(), target_event_id, mine, chat_id
670        );
671        return false;
672    }
673
674    // Drop from in-memory state.
675    let removed = {
676        let mut state = crate::state::STATE.lock().await;
677        state.remove_message(target_event_id)
678    };
679    let removed_msg = match removed {
680        Some((_chat_id, msg)) => msg,
681        None => return false,
682    };
683
684    // Nuke any cached attachment files for this message — sender asked
685    // for the message to disappear, and a downloaded file the receiver
686    // never moved out of Vector's cache should go with it.
687    //
688    // Refcount filter: drop attachments still referenced by sibling
689    // messages so we don't yank a cached file from messages that
690    // still need it (Vector dedupes by SHA-256, so the same file
691    // can back multiple messages). User-managed paths are also left
692    // alone (canonicalize + starts_with check).
693    let unique = crate::deletion::filter_unreferenced_attachments(
694        target_event_id,
695        removed_msg.attachments,
696    ).await;
697    crate::deletion::delete_cached_attachment_files_pub(&unique);
698
699    // Drop from the events table.
700    if let Err(e) = crate::db::events::delete_event(target_event_id).await {
701        eprintln!(
702            "[NIP-17 cooperative-delete] DB delete failed for {}: {}",
703            target_event_id, e
704        );
705    }
706
707    // Tell the frontend to fade the row out. Reuses the existing
708    // message_removed event handled in main.js, so no new wiring.
709    crate::traits::emit_event(
710        "message_removed",
711        &serde_json::json!({
712            "id": target_event_id,
713            "chat_id": &chat_id,
714            "reason": "deleted-by-sender",
715        }),
716    );
717
718    handler.on_message_deleted(&chat_id, target_event_id);
719    let _ = contact;
720    true
721}
722
723// ============================================================================
724// Convenience: single-call event processing
725// ============================================================================
726
727/// Process a single event through the full pipeline (prepare + commit).
728///
729/// Gets client and public key from globals. For callers that manage their
730/// own notification loop but want the full vector-core processing pipeline.
731pub async fn process_event(
732    event: Event,
733    is_new: bool,
734    handler: &dyn InboundEventHandler,
735) -> std::result::Result<bool, String> {
736    let client = crate::state::nostr_client()
737        .ok_or_else(|| "Nostr client not initialized".to_string())?;
738    let my_pk = crate::state::my_public_key()
739        .ok_or_else(|| "Public key not initialized".to_string())?;
740    let prepared = prepare_event(event, &client, my_pk).await;
741    Ok(commit_prepared_event(prepared, is_new, handler).await)
742}
vector_core/event_handler.rs

vector_core/
event_handler.rs
