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, 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
308 .fetch_metadata(profile_pubkey, Duration::from_secs(15))
309 .await;
310
311 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 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 let metadata_changed = profile.from_metadata(meta.unwrap());
340
341 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 emit_event("profile_update", &slim);
365 handler.on_profile_fetched(&slim, &avatar_url, &banner_url);
367 }
368 true
369 } else {
370 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
387pub 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
402pub 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 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 let profile = state.get_profile(&npub).cloned().unwrap_or_default();
436
437 let mut meta = Metadata::new().name(if name.is_empty() {
439 &*profile.name
440 } else {
441 name.as_str()
442 });
443
444 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 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 if !profile.display_name.is_empty() {
470 meta = meta.display_name(&*profile.display_name);
471 }
472
473 meta = meta.about(if about.is_empty() {
475 &*profile.about
476 } else {
477 about.as_str()
478 });
479
480 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 }; let meta = if is_bot { meta.custom_field("bot", true) } else { meta };
501
502 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 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 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
545pub 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 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
600pub async fn block_user(npub: String, handler: &dyn ProfileSyncHandler) -> bool {
609 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 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
641pub 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
663pub 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
672pub 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
697pub 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 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 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 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 {
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
782pub 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
792pub 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
840pub fn refresh_profile_now(npub: String) {
842 let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
843 queue.add(npub, SyncPriority::Critical, true);
844}
845
846pub 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#[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 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 queue.mark_done("npub1carol");
930
931 queue.add("npub1carol".to_string(), SyncPriority::Low, false);
933 assert!(queue.low_queue.is_empty(), "should skip within cache window");
934
935 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 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 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 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}