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).timeout(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    // `Client::fetch_metadata` is gone: fetch the newest kind-0 and parse it.
308    let fetch_result = client
309        .fetch_events(
310            Filter::new()
311                .author(profile_pubkey)
312                .kind(Kind::Metadata)
313                .limit(1),
314        )
315        .timeout(Duration::from_secs(15))
316        .await
317        .map(|events| {
318            events
319                .into_iter()
320                .max_by_key(|e| e.created_at)
321                .and_then(|e| Metadata::from_json(&e.content).ok())
322        });
323
324    // Abandon the fetch result if a swap happened during the await.
325    if !session.is_valid() { return false; }
326
327    match fetch_result {
328        Ok(meta) => {
329            if meta.is_some() {
330                let save_data = {
331                    let mut state = STATE.lock().await;
332                    let id = match state.interner.lookup(&npub) {
333                        Some(id) => id,
334                        None => return false,
335                    };
336                    let (changed, avatar_url, banner_url) = {
337                        let profile = match state.get_profile_mut_by_id(id) {
338                            Some(p) => p,
339                            None => return false,
340                        };
341                        profile.flags.set_mine(my_public_key == profile_pubkey);
342
343                        // Update status
344                        let status_changed = profile.status_title() != status_title.as_str()
345                            || profile.status_purpose() != status_purpose.as_str()
346                            || profile.status_url() != status_url.as_str();
347                        // Only touch the extras box when there's a real status to store or one
348                        // already exists to clear — never materialize an empty box on the common
349                        // status-less profile (that would make it larger than before the split).
350                        let has_status = !status_title.is_empty()
351                            || !status_purpose.is_empty() || !status_url.is_empty();
352                        if profile.extras.is_some() || has_status {
353                            let ex = profile.extras_mut();
354                            ex.status_title = status_title.into_boxed_str();
355                            ex.status_purpose = status_purpose.into_boxed_str();
356                            ex.status_url = status_url.into_boxed_str();
357                        }
358
359                        // Update metadata
360                        let metadata_changed = profile.from_metadata(meta.unwrap());
361
362                        // Update timestamp
363                        profile.last_updated = secs_to_compact(
364                            std::time::SystemTime::now()
365                                .duration_since(std::time::UNIX_EPOCH)
366                                .unwrap()
367                                .as_secs()
368                        );
369
370                        (status_changed || metadata_changed,
371                         profile.avatar.to_string(),
372                         profile.banner.to_string())
373                    };
374
375                    if changed {
376                        let slim = state.serialize_profile(id).unwrap();
377                        Some((slim, avatar_url, banner_url))
378                    } else {
379                        None
380                    }
381                };
382
383                if let Some((slim, avatar_url, banner_url)) = save_data {
384                    // Notify UI via EventEmitter
385                    emit_event("profile_update", &slim);
386                    // Platform-specific: DB persist + image caching
387                    handler.on_profile_fetched(&slim, &avatar_url, &banner_url);
388                }
389                true
390            } else {
391                // No metadata on relays — update timestamp so we don't keep retrying
392                let mut state = STATE.lock().await;
393                if let Some(profile) = state.get_profile_mut(&npub) {
394                    profile.last_updated = secs_to_compact(
395                        std::time::SystemTime::now()
396                            .duration_since(std::time::UNIX_EPOCH)
397                            .unwrap()
398                            .as_secs()
399                    );
400                }
401                true
402            }
403        }
404        Err(_) => false,
405    }
406}
407
408// ============================================================================
409// update_profile — publish metadata to relays
410// ============================================================================
411
412/// Update the current user's profile metadata and broadcast to relays.
413///
414/// Merges the provided fields with the existing profile (empty = keep existing).
415/// After successful broadcast, updates STATE and notifies via EventEmitter + handler.
416pub async fn update_profile(
417    name: String, avatar: String, banner: String, about: String,
418    handler: &dyn ProfileSyncHandler,
419) -> bool {
420    update_profile_inner(name, avatar, banner, about, false, handler).await
421}
422
423/// Publish the current user's profile and mark it as a bot (`bot: true` in the metadata). The SDK
424/// uses this so every bot it builds is tagged; human clients use [`update_profile`].
425pub async fn update_bot_profile(
426    name: String, avatar: String, banner: String, about: String,
427    handler: &dyn ProfileSyncHandler,
428) -> bool {
429    update_profile_inner(name, avatar, banner, about, true, handler).await
430}
431
432async fn update_profile_inner(
433    name: String, avatar: String, banner: String, about: String,
434    is_bot: bool,
435    handler: &dyn ProfileSyncHandler,
436) -> bool {
437    let client = match nostr_client() {
438        Some(c) => c,
439        None => return false,
440    };
441
442    let my_public_key = match my_public_key() {
443        Some(pk) => pk,
444        None => return false,
445    };
446
447    // Build metadata from current profile, then drop the lock before network I/O
448    let meta = {
449        let state = STATE.lock().await;
450        let npub = match my_public_key.to_bech32() {
451            Ok(n) => n,
452            Err(_) => return false,
453        };
454        // Start from the existing profile if we have one, else a blank profile so a first-time
455        // update (e.g. a freshly-created bot that has never published a kind-0) still works.
456        let profile = state.get_profile(&npub).cloned().unwrap_or_default();
457
458        // Merge: use new value if provided, else carry existing
459        let mut meta = Metadata::new().name(if name.is_empty() {
460            &*profile.name
461        } else {
462            name.as_str()
463        });
464
465        // Avatar
466        let avatar_url_str: &str = if avatar.is_empty() {
467            &profile.avatar
468        } else {
469            avatar.as_str()
470        };
471        if !avatar_url_str.is_empty() {
472            if let Ok(url) = Url::parse(avatar_url_str) {
473                meta = meta.picture(url);
474            }
475        }
476
477        // Banner
478        let banner_url_str: &str = if banner.is_empty() {
479            &profile.banner
480        } else {
481            banner.as_str()
482        };
483        if !banner_url_str.is_empty() {
484            if let Ok(url) = Url::parse(banner_url_str) {
485                meta = meta.banner(url);
486            }
487        }
488
489        // Carry forward display_name
490        if !profile.display_name.is_empty() {
491            meta = meta.display_name(&*profile.display_name);
492        }
493
494        // About
495        meta = meta.about(if about.is_empty() {
496            &*profile.about
497        } else {
498            about.as_str()
499        });
500
501        // Carry forward remaining fields
502        if !profile.website().is_empty() {
503            if let Ok(url) = Url::parse(profile.website()) {
504                meta = meta.website(url);
505            }
506        }
507        if !profile.nip05().is_empty() {
508            meta = meta.nip05(profile.nip05());
509        }
510        if !profile.lud06().is_empty() {
511            meta = meta.lud06(profile.lud06());
512        }
513        if !profile.lud16().is_empty() {
514            meta = meta.lud16(profile.lud16());
515        }
516
517        meta
518    }; // STATE lock dropped before network I/O
519
520    // SDK-built bots carry `bot: true` so clients can badge them; human clients never set it.
521    let meta = if is_bot { meta.custom_field("bot", true) } else { meta };
522
523    // Build and sign Kind 0 metadata event
524    let metadata_json = serde_json::to_string(&meta).unwrap();
525    let metadata_event = EventBuilder::new(Kind::Metadata, metadata_json)
526        .tag(Tag::custom("client", vec!["vector"]));
527
528    let Ok(event) = crate::sign_builder(metadata_event).await else {
529        return false;
530    };
531
532    // Broadcast — first-ACK so UI updates as soon as the fastest relay responds
533    match crate::inbox_relays::send_event_pool_first_ok(&client, &event).await {
534        Ok(_) => {
535            let npub = match my_public_key.to_bech32() {
536                Ok(n) => n,
537                Err(_) => return false,
538            };
539            let save_data = {
540                let mut state = STATE.lock().await;
541                // Apply the published metadata to our own profile, creating the entry if this
542                // identity had none yet (a freshly-created account is interned here on first set).
543                let mut profile = state.get_profile(&npub).cloned().unwrap_or_default();
544                profile.from_metadata(meta);
545                let (avatar_url, banner_url) = (profile.avatar.to_string(), profile.banner.to_string());
546                state.insert_or_replace_profile(&npub, profile);
547                let slim = match state.interner.lookup(&npub).and_then(|id| state.serialize_profile(id)) {
548                    Some(s) => s,
549                    None => return false,
550                };
551                (slim, avatar_url, banner_url)
552            };
553
554            let (slim, avatar_url, banner_url) = save_data;
555            emit_event("profile_update", &slim);
556            handler.on_profile_fetched(&slim, &avatar_url, &banner_url);
557            true
558        }
559        Err(e) => {
560            crate::log_warn!("[update_profile] relay broadcast failed: {e}");
561            false
562        }
563    }
564}
565
566// ============================================================================
567// update_status — publish status to relays
568// ============================================================================
569
570/// Update the current user's status (kind 30315) and broadcast to relays.
571///
572/// Status is ephemeral — updated in STATE + frontend but not persisted to DB.
573/// (Re-fetched from relays on next `load_profile` call.)
574pub async fn update_status(status: String) -> bool {
575    let client = match nostr_client() {
576        Some(c) => c,
577        None => return false,
578    };
579
580    let my_public_key = match my_public_key() {
581        Some(pk) => pk,
582        None => return false,
583    };
584
585    // Build and sign kind 30315 status event
586    let status_builder = EventBuilder::new(Kind::from_u16(30315), status.as_str())
587        .tag(Tag::custom("d", vec!["general"]));
588
589    let Ok(event) = crate::sign_builder(status_builder).await else {
590        return false;
591    };
592
593    match crate::inbox_relays::send_event_pool_first_ok(&client, &event).await {
594        Ok(_) => {
595            let mut state = STATE.lock().await;
596            let npub = match my_public_key.to_bech32() {
597                Ok(n) => n,
598                Err(_) => return false,
599            };
600            let id = match state.interner.lookup(&npub) {
601                Some(id) => id,
602                None => return false,
603            };
604            {
605                let profile = match state.get_profile_mut_by_id(id) {
606                    Some(p) => p,
607                    None => return false,
608                };
609                let ex = profile.extras_mut();
610                ex.status_purpose = "general".into();
611                ex.status_title = status.into_boxed_str();
612            }
613
614            let slim = state.serialize_profile(id).unwrap();
615            emit_event("profile_update", &slim);
616            true
617        }
618        Err(_) => false,
619    }
620}
621
622// ============================================================================
623// block / unblock / nickname / blocked list
624// ============================================================================
625
626/// Block a user by npub. DM events from blocked users are dropped after decryption.
627/// Group messages are stored but filtered in the UI.
628///
629/// Returns `false` if trying to block yourself or if the profile can't be found.
630pub async fn block_user(npub: String, handler: &dyn ProfileSyncHandler) -> bool {
631    // Prevent blocking yourself
632    if let Some(my_pk) = my_public_key() {
633        if my_pk.to_bech32().ok().as_deref() == Some(npub.as_str()) {
634            return false;
635        }
636    }
637
638    let mut state = STATE.lock().await;
639
640    // Create profile if it doesn't exist (can block someone with no prior contact)
641    if state.interner.lookup(&npub).is_none() {
642        state.insert_or_replace_profile(&npub, Profile::new());
643    }
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(true);
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/// Unblock a user by npub.
664pub async fn unblock_user(npub: String, handler: &dyn ProfileSyncHandler) -> bool {
665    let mut state = STATE.lock().await;
666
667    if let Some(id) = state.interner.lookup(&npub) {
668        {
669            let profile = match state.get_profile_mut_by_id(id) {
670                Some(p) => p,
671                None => return false,
672            };
673            profile.flags.set_blocked(false);
674        }
675        let slim = state.serialize_profile(id).unwrap();
676        drop(state);
677        emit_event("profile_update", &slim);
678        handler.on_profile_fetched(&slim, "", "");
679        true
680    } else {
681        false
682    }
683}
684
685/// Get all blocked profiles.
686pub async fn get_blocked_users() -> Vec<crate::SlimProfile> {
687    let state = STATE.lock().await;
688    state.profiles.iter()
689        .filter(|p| p.flags.is_blocked())
690        .filter_map(|p| state.serialize_profile(p.id))
691        .collect()
692}
693
694/// Set a nickname for a profile.
695pub async fn set_nickname(npub: String, nickname: String, handler: &dyn ProfileSyncHandler) -> bool {
696    let mut state = STATE.lock().await;
697
698    if let Some(id) = state.interner.lookup(&npub) {
699        {
700            let profile = match state.get_profile_mut_by_id(id) {
701                Some(p) => p,
702                None => return false,
703            };
704            profile.extras_mut().nickname = nickname.into_boxed_str();
705        }
706        let slim = state.serialize_profile(id).unwrap();
707        drop(state);
708        emit_event("profile_nick_changed", &serde_json::json!({
709            "profile_id": &npub,
710            "value": &slim.nickname
711        }));
712        handler.on_profile_fetched(&slim, "", "");
713        true
714    } else {
715        false
716    }
717}
718
719// ============================================================================
720// Background processor
721// ============================================================================
722
723/// Background processor that continuously drains the profile sync queue.
724///
725/// Spawned once at startup. Processes batches in priority order, calling
726/// `load_profile` for each entry with the provided handler.
727pub async fn start_profile_sync_processor(handler: Arc<dyn ProfileSyncHandler>) {
728    let mut last_own_profile_sync = Instant::now();
729    let own_profile_sync_interval = Duration::from_secs(5 * 60);
730
731    loop {
732        // Periodically queue our own profile to detect changes from other Nostr apps
733        if last_own_profile_sync.elapsed() >= own_profile_sync_interval {
734            let state = STATE.lock().await;
735            if let Some(own_profile) = state.profiles.iter().find(|p| p.flags.is_mine()) {
736                let npub = state.interner.resolve(own_profile.id).unwrap_or("").to_string();
737                drop(state);
738
739                let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
740                queue.add(npub, SyncPriority::Low, false);
741            }
742            last_own_profile_sync = Instant::now();
743        }
744
745        // Get next batch (lock scoped)
746        let (should_wait, batch) = {
747            let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
748
749            if queue.is_processing {
750                (true, vec![])
751            } else {
752                queue.is_processing = true;
753                let batch = queue.get_next_batch();
754                for entry in &batch {
755                    queue.mark_processing(&entry.npub);
756                }
757                (false, batch)
758            }
759        };
760
761        if should_wait {
762            tokio::time::sleep(Duration::from_secs(1)).await;
763            continue;
764        }
765
766        if batch.is_empty() {
767            {
768                let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
769                queue.is_processing = false;
770            }
771            tokio::time::sleep(Duration::from_secs(1)).await;
772            continue;
773        }
774
775        // Session captured per-batch so a swap aborts the loop before
776        // account A's queue work lands in account B's DB. The next
777        // outer-loop iteration picks up the new session's queue cleanly.
778        let batch_session = crate::state::SessionGuard::capture();
779
780        for entry in &batch {
781            if !batch_session.is_valid() {
782                break;
783            }
784            load_profile(entry.npub.clone(), handler.as_ref()).await;
785
786            {
787                let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
788                queue.mark_done(&entry.npub);
789            }
790
791            tokio::time::sleep(Duration::from_millis(100)).await;
792        }
793
794        // Release processing lock
795        {
796            let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
797            queue.is_processing = false;
798        }
799
800        tokio::time::sleep(Duration::from_millis(500)).await;
801    }
802}
803
804// ============================================================================
805// Public API
806// ============================================================================
807
808/// Queue a single profile for syncing.
809pub fn queue_profile_sync(npub: String, priority: SyncPriority, force_refresh: bool) {
810    let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
811    queue.add(npub, priority, force_refresh);
812}
813
814/// Queue all profiles for a chat.
815pub async fn queue_chat_profiles(chat_id: String, is_opening: bool) {
816    let state = STATE.lock().await;
817
818    let chat = match state.get_chat(&chat_id) {
819        Some(c) => c,
820        None => return,
821    };
822
823    let base_priority = if is_opening {
824        SyncPriority::High
825    } else {
826        SyncPriority::Medium
827    };
828
829    let mut profiles_to_queue = Vec::new();
830
831    for &handle in chat.participants() {
832        let member_npub = match state.interner.resolve(handle) {
833            Some(s) => s.to_string(),
834            None => continue,
835        };
836
837        let has_metadata = state.get_profile_by_id(handle)
838            .map(|p| {
839                let has_data = !p.name.is_empty() || !p.display_name.is_empty() || !p.avatar.is_empty();
840                let was_fetched = p.last_updated > 0;
841                has_data || was_fetched
842            })
843            .unwrap_or(false);
844
845        let priority = if !has_metadata {
846            SyncPriority::Critical
847        } else {
848            base_priority
849        };
850
851        profiles_to_queue.push((member_npub, priority));
852    }
853
854    drop(state);
855
856    let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
857    for (npub, priority) in profiles_to_queue {
858        queue.add(npub, priority, false);
859    }
860}
861
862/// Force immediate refresh of a profile (for user clicks).
863pub fn refresh_profile_now(npub: String) {
864    let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
865    queue.add(npub, SyncPriority::Critical, true);
866}
867
868/// Sync all profiles in the system.
869pub async fn sync_all_profiles() {
870    let state = STATE.lock().await;
871
872    let mut profiles_to_queue = Vec::new();
873
874    for profile in &state.profiles {
875        let npub = match state.interner.resolve(profile.id) {
876            Some(s) => s.to_string(),
877            None => continue,
878        };
879
880        let has_metadata = !profile.name.is_empty() || !profile.display_name.is_empty() || !profile.avatar.is_empty();
881        let was_fetched = profile.last_updated > 0;
882
883        let priority = if !has_metadata && !was_fetched {
884            SyncPriority::Critical
885        } else {
886            SyncPriority::Low
887        };
888
889        profiles_to_queue.push((npub, priority));
890    }
891
892    drop(state);
893
894    let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
895    for (npub, priority) in profiles_to_queue {
896        queue.add(npub, priority, false);
897    }
898}
899
900// ============================================================================
901// Tests
902// ============================================================================
903
904#[cfg(test)]
905mod tests {
906    use super::*;
907
908    #[test]
909    fn sync_priority_cache_windows() {
910        assert_eq!(SyncPriority::Critical.cache_window(), Duration::from_secs(0));
911        assert_eq!(SyncPriority::High.cache_window(), Duration::from_secs(300));
912        assert_eq!(SyncPriority::Medium.cache_window(), Duration::from_secs(1800));
913        assert_eq!(SyncPriority::Low.cache_window(), Duration::from_secs(86400));
914    }
915
916    #[test]
917    fn sync_priority_batch_sizes() {
918        assert_eq!(SyncPriority::Critical.batch_size(), 10);
919        assert_eq!(SyncPriority::High.batch_size(), 20);
920        assert_eq!(SyncPriority::Medium.batch_size(), 30);
921        assert_eq!(SyncPriority::Low.batch_size(), 50);
922    }
923
924    #[test]
925    fn queue_add_and_dedup() {
926        let mut queue = ProfileSyncQueue::new();
927
928        queue.add("npub1alice".to_string(), SyncPriority::Low, false);
929        queue.add("npub1alice".to_string(), SyncPriority::High, false);
930
931        // Should be in High queue only (deduped from Low)
932        assert!(queue.low_queue.is_empty());
933        assert_eq!(queue.high_queue.len(), 1);
934        assert_eq!(queue.high_queue[0].npub, "npub1alice");
935    }
936
937    #[test]
938    fn queue_skips_processing() {
939        let mut queue = ProfileSyncQueue::new();
940        queue.mark_processing("npub1bob");
941
942        queue.add("npub1bob".to_string(), SyncPriority::Critical, false);
943        assert!(queue.critical_queue.is_empty(), "should skip profiles being processed");
944    }
945
946    #[test]
947    fn queue_cache_window_skips() {
948        let mut queue = ProfileSyncQueue::new();
949
950        // Mark as recently fetched
951        queue.mark_done("npub1carol");
952
953        // Try to add with Low priority (24h cache window) — should skip
954        queue.add("npub1carol".to_string(), SyncPriority::Low, false);
955        assert!(queue.low_queue.is_empty(), "should skip within cache window");
956
957        // Force refresh should bypass cache
958        queue.add("npub1carol".to_string(), SyncPriority::Low, true);
959        assert_eq!(queue.low_queue.len(), 1, "force_refresh should bypass cache");
960    }
961
962    #[test]
963    fn queue_critical_skips_cache() {
964        let mut queue = ProfileSyncQueue::new();
965
966        // Critical has 0s cache window — always fetches
967        queue.mark_done("npub1dave");
968        queue.add("npub1dave".to_string(), SyncPriority::Critical, false);
969        assert_eq!(queue.critical_queue.len(), 1, "Critical should always fetch");
970    }
971
972    #[test]
973    fn get_next_batch_priority_order() {
974        let mut queue = ProfileSyncQueue::new();
975
976        // Add to Low and Critical queues
977        queue.low_queue.push_back(QueueEntry {
978            npub: "npub1low".to_string(),
979            added_at: Instant::now() - Duration::from_secs(600),
980        });
981        queue.critical_queue.push_back(QueueEntry {
982            npub: "npub1critical".to_string(),
983            added_at: Instant::now(),
984        });
985
986        let batch = queue.get_next_batch();
987        assert_eq!(batch.len(), 1);
988        assert_eq!(batch[0].npub, "npub1critical", "Critical should process before Low");
989    }
990
991    #[test]
992    fn get_next_batch_respects_delay() {
993        let mut queue = ProfileSyncQueue::new();
994
995        // Add a High priority entry just now (5s delay required)
996        queue.high_queue.push_back(QueueEntry {
997            npub: "npub1new".to_string(),
998            added_at: Instant::now(),
999        });
1000
1001        let batch = queue.get_next_batch();
1002        assert!(batch.is_empty(), "should not process before delay elapses");
1003    }
1004
1005    #[test]
1006    fn mark_done_updates_last_fetched() {
1007        let mut queue = ProfileSyncQueue::new();
1008        queue.mark_processing("npub1eve");
1009        assert!(queue.processing.contains("npub1eve"));
1010
1011        queue.mark_done("npub1eve");
1012        assert!(!queue.processing.contains("npub1eve"));
1013        assert!(queue.last_fetched.contains_key("npub1eve"));
1014    }
1015
1016    #[test]
1017    fn noop_handler_compiles() {
1018        let handler = NoOpProfileSyncHandler;
1019        let slim = crate::SlimProfile::default();
1020        handler.on_profile_fetched(&slim, "", "");
1021    }
1022}