1use 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#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
28pub enum SyncPriority {
29 Critical, High, Medium, Low, }
34
35impl SyncPriority {
36 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 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 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#[derive(Debug, Clone)]
72pub(crate) struct QueueEntry {
73 npub: String,
74 added_at: Instant,
75}
76
77pub 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 pub fn add(&mut self, npub: String, priority: SyncPriority, force_refresh: bool) {
107 if self.processing.contains(&npub) {
108 return;
109 }
110
111 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 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 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
192static PROFILE_SYNC_QUEUE: LazyLock<Arc<Mutex<ProfileSyncQueue>>> =
197 LazyLock::new(|| Arc::new(Mutex::new(ProfileSyncQueue::new())));
198
199pub fn clear_profile_sync_queue() {
203 if let Ok(mut q) = PROFILE_SYNC_QUEUE.lock() {
204 q.clear();
205 }
206}
207
208pub trait ProfileSyncHandler: Send + Sync {
219 fn on_profile_fetched(&self, _slim: &crate::SlimProfile, _avatar_url: &str, _banner_url: &str) {}
223}
224
225pub struct NoOpProfileSyncHandler;
227impl ProfileSyncHandler for NoOpProfileSyncHandler {}
228
229pub 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 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 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 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 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 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 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 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 let metadata_changed = profile.from_metadata(meta.unwrap());
361
362 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 emit_event("profile_update", &slim);
386 handler.on_profile_fetched(&slim, &avatar_url, &banner_url);
388 }
389 true
390 } else {
391 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
408pub 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
423pub 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 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 let profile = state.get_profile(&npub).cloned().unwrap_or_default();
457
458 let mut meta = Metadata::new().name(if name.is_empty() {
460 &*profile.name
461 } else {
462 name.as_str()
463 });
464
465 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 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 if !profile.display_name.is_empty() {
491 meta = meta.display_name(&*profile.display_name);
492 }
493
494 meta = meta.about(if about.is_empty() {
496 &*profile.about
497 } else {
498 about.as_str()
499 });
500
501 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 }; let meta = if is_bot { meta.custom_field("bot", true) } else { meta };
522
523 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 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 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
566pub 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 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
622pub async fn block_user(npub: String, handler: &dyn ProfileSyncHandler) -> bool {
631 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 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
663pub 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
685pub 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
694pub 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
719pub 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 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 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 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 {
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
804pub 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
814pub 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
862pub fn refresh_profile_now(npub: String) {
864 let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
865 queue.add(npub, SyncPriority::Critical, true);
866}
867
868pub 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#[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 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 queue.mark_done("npub1carol");
952
953 queue.add("npub1carol".to_string(), SyncPriority::Low, false);
955 assert!(queue.low_queue.is_empty(), "should skip within cache window");
956
957 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 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 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 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}