Skip to main content

vector_core/profile/
sync.rs

1//! Profile sync — priority queue, background processor, and relay fetching.
2//!
3//! The sync queue batches profile fetches by priority (Critical → High → Medium → Low),
4//! with cache windows to avoid hammering relays. The background processor
5//! drains the queue and calls `load_profile` for each entry.
6//!
7//! Platform-specific work (DB persistence, image caching) is handled by the
8//! `ProfileSyncHandler` trait — src-tauri provides `TauriProfileSyncHandler`,
9//! CLI provides a no-op or logging implementation.
10
11use std::collections::{HashMap, HashSet, VecDeque};
12use std::sync::{Arc, Mutex, LazyLock};
13use std::time::{Duration, Instant};
14
15use nostr_sdk::prelude::*;
16
17use crate::compact::secs_to_compact;
18use crate::profile::Profile;
19use crate::state::{nostr_client, my_public_key, STATE};
20use crate::traits::emit_event;
21
22// ============================================================================
23// SyncPriority
24// ============================================================================
25
26/// Priority levels for profile syncing.
27#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
28pub enum SyncPriority {
29    Critical,  // No metadata OR user clicked — fetch immediately
30    High,      // Active chats — fetch soon
31    Medium,    // Recent chats — fetch eventually
32    Low,       // Old chats with metadata — passive refresh
33}
34
35impl SyncPriority {
36    /// Cache window duration — how long before a profile can be re-fetched.
37    pub fn cache_window(&self) -> Duration {
38        match self {
39            SyncPriority::Critical => Duration::from_secs(0),
40            SyncPriority::High => Duration::from_secs(5 * 60),
41            SyncPriority::Medium => Duration::from_secs(30 * 60),
42            SyncPriority::Low => Duration::from_secs(24 * 60 * 60),
43        }
44    }
45
46    /// Processing delay — how long after queuing before fetching.
47    pub fn processing_delay(&self) -> Duration {
48        match self {
49            SyncPriority::Critical => Duration::from_secs(0),
50            SyncPriority::High => Duration::from_secs(5),
51            SyncPriority::Medium => Duration::from_secs(30),
52            SyncPriority::Low => Duration::from_secs(5 * 60),
53        }
54    }
55
56    /// Maximum batch size for this priority.
57    pub fn batch_size(&self) -> usize {
58        match self {
59            SyncPriority::Critical => 10,
60            SyncPriority::High => 20,
61            SyncPriority::Medium => 30,
62            SyncPriority::Low => 50,
63        }
64    }
65}
66
67// ============================================================================
68// QueueEntry
69// ============================================================================
70
71#[derive(Debug, Clone)]
72pub(crate) struct QueueEntry {
73    npub: String,
74    added_at: Instant,
75}
76
77// ============================================================================
78// ProfileSyncQueue
79// ============================================================================
80
81/// Profile sync queue manager with four priority lanes.
82pub struct ProfileSyncQueue {
83    critical_queue: VecDeque<QueueEntry>,
84    high_queue: VecDeque<QueueEntry>,
85    medium_queue: VecDeque<QueueEntry>,
86    low_queue: VecDeque<QueueEntry>,
87    processing: HashSet<String>,
88    last_fetched: HashMap<String, Instant>,
89    is_processing: bool,
90}
91
92impl ProfileSyncQueue {
93    pub fn new() -> Self {
94        Self {
95            critical_queue: VecDeque::new(),
96            high_queue: VecDeque::new(),
97            medium_queue: VecDeque::new(),
98            low_queue: VecDeque::new(),
99            processing: HashSet::new(),
100            last_fetched: HashMap::new(),
101            is_processing: false,
102        }
103    }
104
105    /// Add a profile to the sync queue.
106    pub fn add(&mut self, npub: String, priority: SyncPriority, force_refresh: bool) {
107        if self.processing.contains(&npub) {
108            return;
109        }
110
111        // Check cache window (unless force_refresh)
112        if !force_refresh {
113            if let Some(last_fetch) = self.last_fetched.get(&npub) {
114                if last_fetch.elapsed() < priority.cache_window() {
115                    return;
116                }
117            }
118        }
119
120        self.remove_from_all_queues(&npub);
121
122        let entry = QueueEntry { npub, added_at: Instant::now() };
123        match priority {
124            SyncPriority::Critical => self.critical_queue.push_back(entry),
125            SyncPriority::High => self.high_queue.push_back(entry),
126            SyncPriority::Medium => self.medium_queue.push_back(entry),
127            SyncPriority::Low => self.low_queue.push_back(entry),
128        }
129    }
130
131    fn remove_from_all_queues(&mut self, npub: &str) {
132        self.critical_queue.retain(|e| e.npub != npub);
133        self.high_queue.retain(|e| e.npub != npub);
134        self.medium_queue.retain(|e| e.npub != npub);
135        self.low_queue.retain(|e| e.npub != npub);
136    }
137
138    /// Drop every queued + in-flight entry. Used by `reset_session()` so a
139    /// post-reset processor doesn't keep fetching the prior account's contacts.
140    pub fn clear(&mut self) {
141        self.critical_queue.clear();
142        self.high_queue.clear();
143        self.medium_queue.clear();
144        self.low_queue.clear();
145        self.processing.clear();
146        self.last_fetched.clear();
147    }
148
149    /// Get the next batch of profiles ready to process (highest priority first).
150    pub(crate) fn get_next_batch(&mut self) -> Vec<QueueEntry> {
151        let mut batch = Vec::new();
152
153        let (queue, priority) = if !self.critical_queue.is_empty() {
154            (&mut self.critical_queue, SyncPriority::Critical)
155        } else if !self.high_queue.is_empty() {
156            (&mut self.high_queue, SyncPriority::High)
157        } else if !self.medium_queue.is_empty() {
158            (&mut self.medium_queue, SyncPriority::Medium)
159        } else if !self.low_queue.is_empty() {
160            (&mut self.low_queue, SyncPriority::Low)
161        } else {
162            return batch;
163        };
164
165        let batch_size = priority.batch_size();
166        let processing_delay = priority.processing_delay();
167
168        while batch.len() < batch_size && !queue.is_empty() {
169            if let Some(entry) = queue.front() {
170                if entry.added_at.elapsed() >= processing_delay {
171                    let entry = queue.pop_front().unwrap();
172                    batch.push(entry);
173                } else {
174                    break;
175                }
176            }
177        }
178
179        batch
180    }
181
182    pub fn mark_processing(&mut self, npub: &str) {
183        self.processing.insert(npub.to_string());
184    }
185
186    pub fn mark_done(&mut self, npub: &str) {
187        self.processing.remove(npub);
188        self.last_fetched.insert(npub.to_string(), Instant::now());
189    }
190}
191
192// ============================================================================
193// Global queue
194// ============================================================================
195
196static PROFILE_SYNC_QUEUE: LazyLock<Arc<Mutex<ProfileSyncQueue>>> =
197    LazyLock::new(|| Arc::new(Mutex::new(ProfileSyncQueue::new())));
198
199/// Drop every queued profile-sync entry. Called by `reset_session()` so the
200/// long-lived `start_profile_sync_processor` loop doesn't service the prior
201/// account's contacts after an inline session swap.
202pub fn clear_profile_sync_queue() {
203    if let Ok(mut q) = PROFILE_SYNC_QUEUE.lock() {
204        q.clear();
205    }
206}
207
208// ============================================================================
209// ProfileSyncHandler — platform-specific callbacks
210// ============================================================================
211
212/// Callback trait for platform-specific profile sync work.
213///
214/// The core `load_profile` handles relay fetching, STATE updates, and
215/// EventEmitter notifications. This trait covers what differs per platform:
216/// - **DB persistence** (SQLite upsert)
217/// - **Image caching** (avatar/banner download + disk cache)
218pub trait ProfileSyncHandler: Send + Sync {
219    /// Called after a profile is fetched from relays and updated in STATE.
220    /// `slim` is ready for DB persistence. `avatar_url`/`banner_url` are
221    /// for image caching (may be empty).
222    fn on_profile_fetched(&self, _slim: &crate::SlimProfile, _avatar_url: &str, _banner_url: &str) {}
223}
224
225/// No-op handler for CLI/tests.
226pub struct NoOpProfileSyncHandler;
227impl ProfileSyncHandler for NoOpProfileSyncHandler {}
228
229// ============================================================================
230// load_profile — core relay fetch + STATE update
231// ============================================================================
232
233/// Fetch a profile's metadata and status from relays, update STATE, and
234/// notify via EventEmitter + handler callback.
235///
236/// Returns `true` if the fetch succeeded (even if nothing changed).
237pub async fn load_profile(npub: String, handler: &dyn ProfileSyncHandler) -> bool {
238    let client = match nostr_client() {
239        Some(c) => c,
240        None => return false,
241    };
242
243    // Session captured for the whole load_profile lifecycle. Relay
244    // fetches can sleep multi-second; we re-check before writing back
245    // to STATE / DB so a mid-fetch swap doesn't land account A's
246    // profile in account B's storage.
247    let session = crate::state::SessionGuard::capture();
248
249    let profile_pubkey = match PublicKey::from_bech32(npub.as_str()) {
250        Ok(pk) => pk,
251        Err(_) => return false,
252    };
253
254    let my_public_key = match my_public_key() {
255        Some(pk) => pk,
256        None => return false,
257    };
258
259    // Grab old status (or create profile if missing)
260    let (old_status_title, old_status_purpose, old_status_url): (String, String, String);
261    {
262        let mut state = STATE.lock().await;
263        match state.get_profile(&npub) {
264            Some(p) => {
265                old_status_title = p.status_title.to_string();
266                old_status_purpose = p.status_purpose.to_string();
267                old_status_url = p.status_url.to_string();
268            }
269            None => {
270                state.insert_or_replace_profile(&npub, Profile::new());
271                old_status_title = String::new();
272                old_status_purpose = String::new();
273                old_status_url = String::new();
274            }
275        }
276    }
277
278    // Fetch status (kind 30315) from relays
279    let status_filter = Filter::new()
280        .author(profile_pubkey)
281        .kind(Kind::from_u16(30315))
282        .limit(1);
283
284    let (status_title, status_purpose, status_url) = match client
285        .fetch_events(status_filter, Duration::from_secs(15))
286        .await
287    {
288        Ok(res) => {
289            if !res.is_empty() {
290                let status_event = res.first().unwrap();
291                (
292                    status_event.content.clone(),
293                    status_event.tags.first()
294                        .and_then(|t| t.content())
295                        .unwrap_or_default()
296                        .to_string(),
297                    String::new(),
298                )
299            } else {
300                (old_status_title, old_status_purpose, old_status_url)
301            }
302        }
303        Err(_) => (old_status_title, old_status_purpose, old_status_url),
304    };
305
306    // Fetch metadata from relays
307    let fetch_result = client
308        .fetch_metadata(profile_pubkey, Duration::from_secs(15))
309        .await;
310
311    // Abandon the fetch result if a swap happened during the await.
312    if !session.is_valid() { return false; }
313
314    match fetch_result {
315        Ok(meta) => {
316            if meta.is_some() {
317                let save_data = {
318                    let mut state = STATE.lock().await;
319                    let id = match state.interner.lookup(&npub) {
320                        Some(id) => id,
321                        None => return false,
322                    };
323                    let (changed, avatar_url, banner_url) = {
324                        let profile = match state.get_profile_mut_by_id(id) {
325                            Some(p) => p,
326                            None => return false,
327                        };
328                        profile.flags.set_mine(my_public_key == profile_pubkey);
329
330                        // Update status
331                        let status_changed = *profile.status_title != *status_title
332                            || *profile.status_purpose != *status_purpose
333                            || *profile.status_url != *status_url;
334                        profile.status_title = status_title.into_boxed_str();
335                        profile.status_purpose = status_purpose.into_boxed_str();
336                        profile.status_url = status_url.into_boxed_str();
337
338                        // Update metadata
339                        let metadata_changed = profile.from_metadata(meta.unwrap());
340
341                        // Update timestamp
342                        profile.last_updated = secs_to_compact(
343                            std::time::SystemTime::now()
344                                .duration_since(std::time::UNIX_EPOCH)
345                                .unwrap()
346                                .as_secs()
347                        );
348
349                        (status_changed || metadata_changed,
350                         profile.avatar.to_string(),
351                         profile.banner.to_string())
352                    };
353
354                    if changed {
355                        let slim = state.serialize_profile(id).unwrap();
356                        Some((slim, avatar_url, banner_url))
357                    } else {
358                        None
359                    }
360                };
361
362                if let Some((slim, avatar_url, banner_url)) = save_data {
363                    // Notify UI via EventEmitter
364                    emit_event("profile_update", &slim);
365                    // Platform-specific: DB persist + image caching
366                    handler.on_profile_fetched(&slim, &avatar_url, &banner_url);
367                }
368                true
369            } else {
370                // No metadata on relays — update timestamp so we don't keep retrying
371                let mut state = STATE.lock().await;
372                if let Some(profile) = state.get_profile_mut(&npub) {
373                    profile.last_updated = secs_to_compact(
374                        std::time::SystemTime::now()
375                            .duration_since(std::time::UNIX_EPOCH)
376                            .unwrap()
377                            .as_secs()
378                    );
379                }
380                true
381            }
382        }
383        Err(_) => false,
384    }
385}
386
387// ============================================================================
388// update_profile — publish metadata to relays
389// ============================================================================
390
391/// Update the current user's profile metadata and broadcast to relays.
392///
393/// Merges the provided fields with the existing profile (empty = keep existing).
394/// After successful broadcast, updates STATE and notifies via EventEmitter + handler.
395pub async fn update_profile(
396    name: String, avatar: String, banner: String, about: String,
397    handler: &dyn ProfileSyncHandler,
398) -> bool {
399    update_profile_inner(name, avatar, banner, about, false, handler).await
400}
401
402/// Publish the current user's profile and mark it as a bot (`bot: true` in the metadata). The SDK
403/// uses this so every bot it builds is tagged; human clients use [`update_profile`].
404pub async fn update_bot_profile(
405    name: String, avatar: String, banner: String, about: String,
406    handler: &dyn ProfileSyncHandler,
407) -> bool {
408    update_profile_inner(name, avatar, banner, about, true, handler).await
409}
410
411async fn update_profile_inner(
412    name: String, avatar: String, banner: String, about: String,
413    is_bot: bool,
414    handler: &dyn ProfileSyncHandler,
415) -> bool {
416    let client = match nostr_client() {
417        Some(c) => c,
418        None => return false,
419    };
420
421    let my_public_key = match my_public_key() {
422        Some(pk) => pk,
423        None => return false,
424    };
425
426    // Build metadata from current profile, then drop the lock before network I/O
427    let meta = {
428        let state = STATE.lock().await;
429        let npub = match my_public_key.to_bech32() {
430            Ok(n) => n,
431            Err(_) => return false,
432        };
433        // Start from the existing profile if we have one, else a blank profile so a first-time
434        // update (e.g. a freshly-created bot that has never published a kind-0) still works.
435        let profile = state.get_profile(&npub).cloned().unwrap_or_default();
436
437        // Merge: use new value if provided, else carry existing
438        let mut meta = Metadata::new().name(if name.is_empty() {
439            &*profile.name
440        } else {
441            name.as_str()
442        });
443
444        // Avatar
445        let avatar_url_str: &str = if avatar.is_empty() {
446            &profile.avatar
447        } else {
448            avatar.as_str()
449        };
450        if !avatar_url_str.is_empty() {
451            if let Ok(url) = Url::parse(avatar_url_str) {
452                meta = meta.picture(url);
453            }
454        }
455
456        // Banner
457        let banner_url_str: &str = if banner.is_empty() {
458            &profile.banner
459        } else {
460            banner.as_str()
461        };
462        if !banner_url_str.is_empty() {
463            if let Ok(url) = Url::parse(banner_url_str) {
464                meta = meta.banner(url);
465            }
466        }
467
468        // Carry forward display_name
469        if !profile.display_name.is_empty() {
470            meta = meta.display_name(&*profile.display_name);
471        }
472
473        // About
474        meta = meta.about(if about.is_empty() {
475            &*profile.about
476        } else {
477            about.as_str()
478        });
479
480        // Carry forward remaining fields
481        if !profile.website.is_empty() {
482            if let Ok(url) = Url::parse(&*profile.website) {
483                meta = meta.website(url);
484            }
485        }
486        if !profile.nip05.is_empty() {
487            meta = meta.nip05(&*profile.nip05);
488        }
489        if !profile.lud06.is_empty() {
490            meta = meta.lud06(&*profile.lud06);
491        }
492        if !profile.lud16.is_empty() {
493            meta = meta.lud16(&*profile.lud16);
494        }
495
496        meta
497    }; // STATE lock dropped before network I/O
498
499    // SDK-built bots carry `bot: true` so clients can badge them; human clients never set it.
500    let meta = if is_bot { meta.custom_field("bot", true) } else { meta };
501
502    // Build and sign Kind 0 metadata event
503    let metadata_json = serde_json::to_string(&meta).unwrap();
504    let metadata_event = EventBuilder::new(Kind::Metadata, metadata_json)
505        .tag(Tag::custom(TagKind::Custom(String::from("client").into()), vec!["vector"]));
506
507    let Ok(event) = client.sign_event_builder(metadata_event).await else {
508        return false;
509    };
510
511    // Broadcast — first-ACK so UI updates as soon as the fastest relay responds
512    match crate::inbox_relays::send_event_pool_first_ok(&client, &event).await {
513        Ok(_) => {
514            let npub = match my_public_key.to_bech32() {
515                Ok(n) => n,
516                Err(_) => return false,
517            };
518            let save_data = {
519                let mut state = STATE.lock().await;
520                // Apply the published metadata to our own profile, creating the entry if this
521                // identity had none yet (a freshly-created account is interned here on first set).
522                let mut profile = state.get_profile(&npub).cloned().unwrap_or_default();
523                profile.from_metadata(meta);
524                let (avatar_url, banner_url) = (profile.avatar.to_string(), profile.banner.to_string());
525                state.insert_or_replace_profile(&npub, profile);
526                let slim = match state.interner.lookup(&npub).and_then(|id| state.serialize_profile(id)) {
527                    Some(s) => s,
528                    None => return false,
529                };
530                (slim, avatar_url, banner_url)
531            };
532
533            let (slim, avatar_url, banner_url) = save_data;
534            emit_event("profile_update", &slim);
535            handler.on_profile_fetched(&slim, &avatar_url, &banner_url);
536            true
537        }
538        Err(e) => {
539            crate::log_warn!("[update_profile] relay broadcast failed: {e}");
540            false
541        }
542    }
543}
544
545// ============================================================================
546// update_status — publish status to relays
547// ============================================================================
548
549/// Update the current user's status (kind 30315) and broadcast to relays.
550///
551/// Status is ephemeral — updated in STATE + frontend but not persisted to DB.
552/// (Re-fetched from relays on next `load_profile` call.)
553pub async fn update_status(status: String) -> bool {
554    let client = match nostr_client() {
555        Some(c) => c,
556        None => return false,
557    };
558
559    let my_public_key = match my_public_key() {
560        Some(pk) => pk,
561        None => return false,
562    };
563
564    // Build and sign kind 30315 status event
565    let status_builder = EventBuilder::new(Kind::from_u16(30315), status.as_str())
566        .tag(Tag::custom(TagKind::d(), vec!["general"]));
567
568    let Ok(event) = client.sign_event_builder(status_builder).await else {
569        return false;
570    };
571
572    match crate::inbox_relays::send_event_pool_first_ok(&client, &event).await {
573        Ok(_) => {
574            let mut state = STATE.lock().await;
575            let npub = match my_public_key.to_bech32() {
576                Ok(n) => n,
577                Err(_) => return false,
578            };
579            let id = match state.interner.lookup(&npub) {
580                Some(id) => id,
581                None => return false,
582            };
583            {
584                let profile = match state.get_profile_mut_by_id(id) {
585                    Some(p) => p,
586                    None => return false,
587                };
588                profile.status_purpose = "general".into();
589                profile.status_title = status.into_boxed_str();
590            }
591
592            let slim = state.serialize_profile(id).unwrap();
593            emit_event("profile_update", &slim);
594            true
595        }
596        Err(_) => false,
597    }
598}
599
600// ============================================================================
601// block / unblock / nickname / blocked list
602// ============================================================================
603
604/// Block a user by npub. DM events from blocked users are dropped after decryption.
605/// Group messages are stored but filtered in the UI.
606///
607/// Returns `false` if trying to block yourself or if the profile can't be found.
608pub async fn block_user(npub: String, handler: &dyn ProfileSyncHandler) -> bool {
609    // Prevent blocking yourself
610    if let Some(my_pk) = my_public_key() {
611        if my_pk.to_bech32().ok().as_deref() == Some(npub.as_str()) {
612            return false;
613        }
614    }
615
616    let mut state = STATE.lock().await;
617
618    // Create profile if it doesn't exist (can block someone with no prior contact)
619    if state.interner.lookup(&npub).is_none() {
620        state.insert_or_replace_profile(&npub, Profile::new());
621    }
622
623    if let Some(id) = state.interner.lookup(&npub) {
624        {
625            let profile = match state.get_profile_mut_by_id(id) {
626                Some(p) => p,
627                None => return false,
628            };
629            profile.flags.set_blocked(true);
630        }
631        let slim = state.serialize_profile(id).unwrap();
632        drop(state);
633        emit_event("profile_update", &slim);
634        handler.on_profile_fetched(&slim, "", "");
635        true
636    } else {
637        false
638    }
639}
640
641/// Unblock a user by npub.
642pub async fn unblock_user(npub: String, handler: &dyn ProfileSyncHandler) -> bool {
643    let mut state = STATE.lock().await;
644
645    if let Some(id) = state.interner.lookup(&npub) {
646        {
647            let profile = match state.get_profile_mut_by_id(id) {
648                Some(p) => p,
649                None => return false,
650            };
651            profile.flags.set_blocked(false);
652        }
653        let slim = state.serialize_profile(id).unwrap();
654        drop(state);
655        emit_event("profile_update", &slim);
656        handler.on_profile_fetched(&slim, "", "");
657        true
658    } else {
659        false
660    }
661}
662
663/// Get all blocked profiles.
664pub async fn get_blocked_users() -> Vec<crate::SlimProfile> {
665    let state = STATE.lock().await;
666    state.profiles.iter()
667        .filter(|p| p.flags.is_blocked())
668        .filter_map(|p| state.serialize_profile(p.id))
669        .collect()
670}
671
672/// Set a nickname for a profile.
673pub async fn set_nickname(npub: String, nickname: String, handler: &dyn ProfileSyncHandler) -> bool {
674    let mut state = STATE.lock().await;
675
676    if let Some(id) = state.interner.lookup(&npub) {
677        {
678            let profile = match state.get_profile_mut_by_id(id) {
679                Some(p) => p,
680                None => return false,
681            };
682            profile.nickname = nickname.into_boxed_str();
683        }
684        let slim = state.serialize_profile(id).unwrap();
685        drop(state);
686        emit_event("profile_nick_changed", &serde_json::json!({
687            "profile_id": &npub,
688            "value": &slim.nickname
689        }));
690        handler.on_profile_fetched(&slim, "", "");
691        true
692    } else {
693        false
694    }
695}
696
697// ============================================================================
698// Background processor
699// ============================================================================
700
701/// Background processor that continuously drains the profile sync queue.
702///
703/// Spawned once at startup. Processes batches in priority order, calling
704/// `load_profile` for each entry with the provided handler.
705pub async fn start_profile_sync_processor(handler: Arc<dyn ProfileSyncHandler>) {
706    let mut last_own_profile_sync = Instant::now();
707    let own_profile_sync_interval = Duration::from_secs(5 * 60);
708
709    loop {
710        // Periodically queue our own profile to detect changes from other Nostr apps
711        if last_own_profile_sync.elapsed() >= own_profile_sync_interval {
712            let state = STATE.lock().await;
713            if let Some(own_profile) = state.profiles.iter().find(|p| p.flags.is_mine()) {
714                let npub = state.interner.resolve(own_profile.id).unwrap_or("").to_string();
715                drop(state);
716
717                let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
718                queue.add(npub, SyncPriority::Low, false);
719            }
720            last_own_profile_sync = Instant::now();
721        }
722
723        // Get next batch (lock scoped)
724        let (should_wait, batch) = {
725            let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
726
727            if queue.is_processing {
728                (true, vec![])
729            } else {
730                queue.is_processing = true;
731                let batch = queue.get_next_batch();
732                for entry in &batch {
733                    queue.mark_processing(&entry.npub);
734                }
735                (false, batch)
736            }
737        };
738
739        if should_wait {
740            tokio::time::sleep(Duration::from_secs(1)).await;
741            continue;
742        }
743
744        if batch.is_empty() {
745            {
746                let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
747                queue.is_processing = false;
748            }
749            tokio::time::sleep(Duration::from_secs(1)).await;
750            continue;
751        }
752
753        // Session captured per-batch so a swap aborts the loop before
754        // account A's queue work lands in account B's DB. The next
755        // outer-loop iteration picks up the new session's queue cleanly.
756        let batch_session = crate::state::SessionGuard::capture();
757
758        for entry in &batch {
759            if !batch_session.is_valid() {
760                break;
761            }
762            load_profile(entry.npub.clone(), handler.as_ref()).await;
763
764            {
765                let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
766                queue.mark_done(&entry.npub);
767            }
768
769            tokio::time::sleep(Duration::from_millis(100)).await;
770        }
771
772        // Release processing lock
773        {
774            let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
775            queue.is_processing = false;
776        }
777
778        tokio::time::sleep(Duration::from_millis(500)).await;
779    }
780}
781
782// ============================================================================
783// Public API
784// ============================================================================
785
786/// Queue a single profile for syncing.
787pub fn queue_profile_sync(npub: String, priority: SyncPriority, force_refresh: bool) {
788    let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
789    queue.add(npub, priority, force_refresh);
790}
791
792/// Queue all profiles for a chat.
793pub async fn queue_chat_profiles(chat_id: String, is_opening: bool) {
794    let state = STATE.lock().await;
795
796    let chat = match state.get_chat(&chat_id) {
797        Some(c) => c,
798        None => return,
799    };
800
801    let base_priority = if is_opening {
802        SyncPriority::High
803    } else {
804        SyncPriority::Medium
805    };
806
807    let mut profiles_to_queue = Vec::new();
808
809    for &handle in chat.participants() {
810        let member_npub = match state.interner.resolve(handle) {
811            Some(s) => s.to_string(),
812            None => continue,
813        };
814
815        let has_metadata = state.get_profile_by_id(handle)
816            .map(|p| {
817                let has_data = !p.name.is_empty() || !p.display_name.is_empty() || !p.avatar.is_empty();
818                let was_fetched = p.last_updated > 0;
819                has_data || was_fetched
820            })
821            .unwrap_or(false);
822
823        let priority = if !has_metadata {
824            SyncPriority::Critical
825        } else {
826            base_priority
827        };
828
829        profiles_to_queue.push((member_npub, priority));
830    }
831
832    drop(state);
833
834    let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
835    for (npub, priority) in profiles_to_queue {
836        queue.add(npub, priority, false);
837    }
838}
839
840/// Force immediate refresh of a profile (for user clicks).
841pub fn refresh_profile_now(npub: String) {
842    let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
843    queue.add(npub, SyncPriority::Critical, true);
844}
845
846/// Sync all profiles in the system.
847pub async fn sync_all_profiles() {
848    let state = STATE.lock().await;
849
850    let mut profiles_to_queue = Vec::new();
851
852    for profile in &state.profiles {
853        let npub = match state.interner.resolve(profile.id) {
854            Some(s) => s.to_string(),
855            None => continue,
856        };
857
858        let has_metadata = !profile.name.is_empty() || !profile.display_name.is_empty() || !profile.avatar.is_empty();
859        let was_fetched = profile.last_updated > 0;
860
861        let priority = if !has_metadata && !was_fetched {
862            SyncPriority::Critical
863        } else {
864            SyncPriority::Low
865        };
866
867        profiles_to_queue.push((npub, priority));
868    }
869
870    drop(state);
871
872    let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
873    for (npub, priority) in profiles_to_queue {
874        queue.add(npub, priority, false);
875    }
876}
877
878// ============================================================================
879// Tests
880// ============================================================================
881
882#[cfg(test)]
883mod tests {
884    use super::*;
885
886    #[test]
887    fn sync_priority_cache_windows() {
888        assert_eq!(SyncPriority::Critical.cache_window(), Duration::from_secs(0));
889        assert_eq!(SyncPriority::High.cache_window(), Duration::from_secs(300));
890        assert_eq!(SyncPriority::Medium.cache_window(), Duration::from_secs(1800));
891        assert_eq!(SyncPriority::Low.cache_window(), Duration::from_secs(86400));
892    }
893
894    #[test]
895    fn sync_priority_batch_sizes() {
896        assert_eq!(SyncPriority::Critical.batch_size(), 10);
897        assert_eq!(SyncPriority::High.batch_size(), 20);
898        assert_eq!(SyncPriority::Medium.batch_size(), 30);
899        assert_eq!(SyncPriority::Low.batch_size(), 50);
900    }
901
902    #[test]
903    fn queue_add_and_dedup() {
904        let mut queue = ProfileSyncQueue::new();
905
906        queue.add("npub1alice".to_string(), SyncPriority::Low, false);
907        queue.add("npub1alice".to_string(), SyncPriority::High, false);
908
909        // Should be in High queue only (deduped from Low)
910        assert!(queue.low_queue.is_empty());
911        assert_eq!(queue.high_queue.len(), 1);
912        assert_eq!(queue.high_queue[0].npub, "npub1alice");
913    }
914
915    #[test]
916    fn queue_skips_processing() {
917        let mut queue = ProfileSyncQueue::new();
918        queue.mark_processing("npub1bob");
919
920        queue.add("npub1bob".to_string(), SyncPriority::Critical, false);
921        assert!(queue.critical_queue.is_empty(), "should skip profiles being processed");
922    }
923
924    #[test]
925    fn queue_cache_window_skips() {
926        let mut queue = ProfileSyncQueue::new();
927
928        // Mark as recently fetched
929        queue.mark_done("npub1carol");
930
931        // Try to add with Low priority (24h cache window) — should skip
932        queue.add("npub1carol".to_string(), SyncPriority::Low, false);
933        assert!(queue.low_queue.is_empty(), "should skip within cache window");
934
935        // Force refresh should bypass cache
936        queue.add("npub1carol".to_string(), SyncPriority::Low, true);
937        assert_eq!(queue.low_queue.len(), 1, "force_refresh should bypass cache");
938    }
939
940    #[test]
941    fn queue_critical_skips_cache() {
942        let mut queue = ProfileSyncQueue::new();
943
944        // Critical has 0s cache window — always fetches
945        queue.mark_done("npub1dave");
946        queue.add("npub1dave".to_string(), SyncPriority::Critical, false);
947        assert_eq!(queue.critical_queue.len(), 1, "Critical should always fetch");
948    }
949
950    #[test]
951    fn get_next_batch_priority_order() {
952        let mut queue = ProfileSyncQueue::new();
953
954        // Add to Low and Critical queues
955        queue.low_queue.push_back(QueueEntry {
956            npub: "npub1low".to_string(),
957            added_at: Instant::now() - Duration::from_secs(600),
958        });
959        queue.critical_queue.push_back(QueueEntry {
960            npub: "npub1critical".to_string(),
961            added_at: Instant::now(),
962        });
963
964        let batch = queue.get_next_batch();
965        assert_eq!(batch.len(), 1);
966        assert_eq!(batch[0].npub, "npub1critical", "Critical should process before Low");
967    }
968
969    #[test]
970    fn get_next_batch_respects_delay() {
971        let mut queue = ProfileSyncQueue::new();
972
973        // Add a High priority entry just now (5s delay required)
974        queue.high_queue.push_back(QueueEntry {
975            npub: "npub1new".to_string(),
976            added_at: Instant::now(),
977        });
978
979        let batch = queue.get_next_batch();
980        assert!(batch.is_empty(), "should not process before delay elapses");
981    }
982
983    #[test]
984    fn mark_done_updates_last_fetched() {
985        let mut queue = ProfileSyncQueue::new();
986        queue.mark_processing("npub1eve");
987        assert!(queue.processing.contains("npub1eve"));
988
989        queue.mark_done("npub1eve");
990        assert!(!queue.processing.contains("npub1eve"));
991        assert!(queue.last_fetched.contains_key("npub1eve"));
992    }
993
994    #[test]
995    fn noop_handler_compiles() {
996        let handler = NoOpProfileSyncHandler;
997        let slim = crate::SlimProfile::default();
998        handler.on_profile_fetched(&slim, "", "");
999    }
1000}