use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::{Arc, Mutex, LazyLock};
use std::time::{Duration, Instant};
use nostr_sdk::prelude::*;
use crate::compact::secs_to_compact;
use crate::profile::Profile;
use crate::state::{nostr_client, my_public_key, STATE};
use crate::traits::emit_event;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
pub enum SyncPriority {
Critical, High, Medium, Low, }
impl SyncPriority {
pub fn cache_window(&self) -> Duration {
match self {
SyncPriority::Critical => Duration::from_secs(0),
SyncPriority::High => Duration::from_secs(5 * 60),
SyncPriority::Medium => Duration::from_secs(30 * 60),
SyncPriority::Low => Duration::from_secs(24 * 60 * 60),
}
}
pub fn processing_delay(&self) -> Duration {
match self {
SyncPriority::Critical => Duration::from_secs(0),
SyncPriority::High => Duration::from_secs(5),
SyncPriority::Medium => Duration::from_secs(30),
SyncPriority::Low => Duration::from_secs(5 * 60),
}
}
pub fn batch_size(&self) -> usize {
match self {
SyncPriority::Critical => 10,
SyncPriority::High => 20,
SyncPriority::Medium => 30,
SyncPriority::Low => 50,
}
}
}
#[derive(Debug, Clone)]
pub(crate) struct QueueEntry {
npub: String,
added_at: Instant,
}
pub struct ProfileSyncQueue {
critical_queue: VecDeque<QueueEntry>,
high_queue: VecDeque<QueueEntry>,
medium_queue: VecDeque<QueueEntry>,
low_queue: VecDeque<QueueEntry>,
processing: HashSet<String>,
last_fetched: HashMap<String, Instant>,
is_processing: bool,
}
impl ProfileSyncQueue {
pub fn new() -> Self {
Self {
critical_queue: VecDeque::new(),
high_queue: VecDeque::new(),
medium_queue: VecDeque::new(),
low_queue: VecDeque::new(),
processing: HashSet::new(),
last_fetched: HashMap::new(),
is_processing: false,
}
}
pub fn add(&mut self, npub: String, priority: SyncPriority, force_refresh: bool) {
if self.processing.contains(&npub) {
return;
}
if !force_refresh {
if let Some(last_fetch) = self.last_fetched.get(&npub) {
if last_fetch.elapsed() < priority.cache_window() {
return;
}
}
}
self.remove_from_all_queues(&npub);
let entry = QueueEntry { npub, added_at: Instant::now() };
match priority {
SyncPriority::Critical => self.critical_queue.push_back(entry),
SyncPriority::High => self.high_queue.push_back(entry),
SyncPriority::Medium => self.medium_queue.push_back(entry),
SyncPriority::Low => self.low_queue.push_back(entry),
}
}
fn remove_from_all_queues(&mut self, npub: &str) {
self.critical_queue.retain(|e| e.npub != npub);
self.high_queue.retain(|e| e.npub != npub);
self.medium_queue.retain(|e| e.npub != npub);
self.low_queue.retain(|e| e.npub != npub);
}
pub fn clear(&mut self) {
self.critical_queue.clear();
self.high_queue.clear();
self.medium_queue.clear();
self.low_queue.clear();
self.processing.clear();
self.last_fetched.clear();
}
pub(crate) fn get_next_batch(&mut self) -> Vec<QueueEntry> {
let mut batch = Vec::new();
let (queue, priority) = if !self.critical_queue.is_empty() {
(&mut self.critical_queue, SyncPriority::Critical)
} else if !self.high_queue.is_empty() {
(&mut self.high_queue, SyncPriority::High)
} else if !self.medium_queue.is_empty() {
(&mut self.medium_queue, SyncPriority::Medium)
} else if !self.low_queue.is_empty() {
(&mut self.low_queue, SyncPriority::Low)
} else {
return batch;
};
let batch_size = priority.batch_size();
let processing_delay = priority.processing_delay();
while batch.len() < batch_size && !queue.is_empty() {
if let Some(entry) = queue.front() {
if entry.added_at.elapsed() >= processing_delay {
let entry = queue.pop_front().unwrap();
batch.push(entry);
} else {
break;
}
}
}
batch
}
pub fn mark_processing(&mut self, npub: &str) {
self.processing.insert(npub.to_string());
}
pub fn mark_done(&mut self, npub: &str) {
self.processing.remove(npub);
self.last_fetched.insert(npub.to_string(), Instant::now());
}
}
static PROFILE_SYNC_QUEUE: LazyLock<Arc<Mutex<ProfileSyncQueue>>> =
LazyLock::new(|| Arc::new(Mutex::new(ProfileSyncQueue::new())));
pub fn clear_profile_sync_queue() {
if let Ok(mut q) = PROFILE_SYNC_QUEUE.lock() {
q.clear();
}
}
pub trait ProfileSyncHandler: Send + Sync {
fn on_profile_fetched(&self, _slim: &crate::SlimProfile, _avatar_url: &str, _banner_url: &str) {}
}
pub struct NoOpProfileSyncHandler;
impl ProfileSyncHandler for NoOpProfileSyncHandler {}
pub async fn load_profile(npub: String, handler: &dyn ProfileSyncHandler) -> bool {
let client = match nostr_client() {
Some(c) => c,
None => return false,
};
let session = crate::state::SessionGuard::capture();
let profile_pubkey = match PublicKey::from_bech32(npub.as_str()) {
Ok(pk) => pk,
Err(_) => return false,
};
let my_public_key = match my_public_key() {
Some(pk) => pk,
None => return false,
};
let (old_status_title, old_status_purpose, old_status_url): (String, String, String);
{
let mut state = STATE.lock().await;
match state.get_profile(&npub) {
Some(p) => {
old_status_title = p.status_title.to_string();
old_status_purpose = p.status_purpose.to_string();
old_status_url = p.status_url.to_string();
}
None => {
state.insert_or_replace_profile(&npub, Profile::new());
old_status_title = String::new();
old_status_purpose = String::new();
old_status_url = String::new();
}
}
}
let status_filter = Filter::new()
.author(profile_pubkey)
.kind(Kind::from_u16(30315))
.limit(1);
let (status_title, status_purpose, status_url) = match client
.fetch_events(status_filter, Duration::from_secs(15))
.await
{
Ok(res) => {
if !res.is_empty() {
let status_event = res.first().unwrap();
(
status_event.content.clone(),
status_event.tags.first()
.and_then(|t| t.content())
.unwrap_or_default()
.to_string(),
String::new(),
)
} else {
(old_status_title, old_status_purpose, old_status_url)
}
}
Err(_) => (old_status_title, old_status_purpose, old_status_url),
};
let fetch_result = client
.fetch_metadata(profile_pubkey, Duration::from_secs(15))
.await;
if !session.is_valid() { return false; }
match fetch_result {
Ok(meta) => {
if meta.is_some() {
let save_data = {
let mut state = STATE.lock().await;
let id = match state.interner.lookup(&npub) {
Some(id) => id,
None => return false,
};
let (changed, avatar_url, banner_url) = {
let profile = match state.get_profile_mut_by_id(id) {
Some(p) => p,
None => return false,
};
profile.flags.set_mine(my_public_key == profile_pubkey);
let status_changed = *profile.status_title != *status_title
|| *profile.status_purpose != *status_purpose
|| *profile.status_url != *status_url;
profile.status_title = status_title.into_boxed_str();
profile.status_purpose = status_purpose.into_boxed_str();
profile.status_url = status_url.into_boxed_str();
let metadata_changed = profile.from_metadata(meta.unwrap());
profile.last_updated = secs_to_compact(
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs()
);
(status_changed || metadata_changed,
profile.avatar.to_string(),
profile.banner.to_string())
};
if changed {
let slim = state.serialize_profile(id).unwrap();
Some((slim, avatar_url, banner_url))
} else {
None
}
};
if let Some((slim, avatar_url, banner_url)) = save_data {
emit_event("profile_update", &slim);
handler.on_profile_fetched(&slim, &avatar_url, &banner_url);
}
true
} else {
let mut state = STATE.lock().await;
if let Some(profile) = state.get_profile_mut(&npub) {
profile.last_updated = secs_to_compact(
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs()
);
}
true
}
}
Err(_) => false,
}
}
pub async fn update_profile(
name: String, avatar: String, banner: String, about: String,
handler: &dyn ProfileSyncHandler,
) -> bool {
update_profile_inner(name, avatar, banner, about, false, handler).await
}
pub async fn update_bot_profile(
name: String, avatar: String, banner: String, about: String,
handler: &dyn ProfileSyncHandler,
) -> bool {
update_profile_inner(name, avatar, banner, about, true, handler).await
}
async fn update_profile_inner(
name: String, avatar: String, banner: String, about: String,
is_bot: bool,
handler: &dyn ProfileSyncHandler,
) -> bool {
let client = match nostr_client() {
Some(c) => c,
None => return false,
};
let my_public_key = match my_public_key() {
Some(pk) => pk,
None => return false,
};
let meta = {
let state = STATE.lock().await;
let npub = match my_public_key.to_bech32() {
Ok(n) => n,
Err(_) => return false,
};
let profile = state.get_profile(&npub).cloned().unwrap_or_default();
let mut meta = Metadata::new().name(if name.is_empty() {
&*profile.name
} else {
name.as_str()
});
let avatar_url_str: &str = if avatar.is_empty() {
&profile.avatar
} else {
avatar.as_str()
};
if !avatar_url_str.is_empty() {
if let Ok(url) = Url::parse(avatar_url_str) {
meta = meta.picture(url);
}
}
let banner_url_str: &str = if banner.is_empty() {
&profile.banner
} else {
banner.as_str()
};
if !banner_url_str.is_empty() {
if let Ok(url) = Url::parse(banner_url_str) {
meta = meta.banner(url);
}
}
if !profile.display_name.is_empty() {
meta = meta.display_name(&*profile.display_name);
}
meta = meta.about(if about.is_empty() {
&*profile.about
} else {
about.as_str()
});
if !profile.website.is_empty() {
if let Ok(url) = Url::parse(&*profile.website) {
meta = meta.website(url);
}
}
if !profile.nip05.is_empty() {
meta = meta.nip05(&*profile.nip05);
}
if !profile.lud06.is_empty() {
meta = meta.lud06(&*profile.lud06);
}
if !profile.lud16.is_empty() {
meta = meta.lud16(&*profile.lud16);
}
meta
};
let meta = if is_bot { meta.custom_field("bot", true) } else { meta };
let metadata_json = serde_json::to_string(&meta).unwrap();
let metadata_event = EventBuilder::new(Kind::Metadata, metadata_json)
.tag(Tag::custom(TagKind::Custom(String::from("client").into()), vec!["vector"]));
let Ok(event) = client.sign_event_builder(metadata_event).await else {
return false;
};
match crate::inbox_relays::send_event_pool_first_ok(&client, &event).await {
Ok(_) => {
let npub = match my_public_key.to_bech32() {
Ok(n) => n,
Err(_) => return false,
};
let save_data = {
let mut state = STATE.lock().await;
let mut profile = state.get_profile(&npub).cloned().unwrap_or_default();
profile.from_metadata(meta);
let (avatar_url, banner_url) = (profile.avatar.to_string(), profile.banner.to_string());
state.insert_or_replace_profile(&npub, profile);
let slim = match state.interner.lookup(&npub).and_then(|id| state.serialize_profile(id)) {
Some(s) => s,
None => return false,
};
(slim, avatar_url, banner_url)
};
let (slim, avatar_url, banner_url) = save_data;
emit_event("profile_update", &slim);
handler.on_profile_fetched(&slim, &avatar_url, &banner_url);
true
}
Err(e) => {
crate::log_warn!("[update_profile] relay broadcast failed: {e}");
false
}
}
}
pub async fn update_status(status: String) -> bool {
let client = match nostr_client() {
Some(c) => c,
None => return false,
};
let my_public_key = match my_public_key() {
Some(pk) => pk,
None => return false,
};
let status_builder = EventBuilder::new(Kind::from_u16(30315), status.as_str())
.tag(Tag::custom(TagKind::d(), vec!["general"]));
let Ok(event) = client.sign_event_builder(status_builder).await else {
return false;
};
match crate::inbox_relays::send_event_pool_first_ok(&client, &event).await {
Ok(_) => {
let mut state = STATE.lock().await;
let npub = match my_public_key.to_bech32() {
Ok(n) => n,
Err(_) => return false,
};
let id = match state.interner.lookup(&npub) {
Some(id) => id,
None => return false,
};
{
let profile = match state.get_profile_mut_by_id(id) {
Some(p) => p,
None => return false,
};
profile.status_purpose = "general".into();
profile.status_title = status.into_boxed_str();
}
let slim = state.serialize_profile(id).unwrap();
emit_event("profile_update", &slim);
true
}
Err(_) => false,
}
}
pub async fn block_user(npub: String, handler: &dyn ProfileSyncHandler) -> bool {
if let Some(my_pk) = my_public_key() {
if my_pk.to_bech32().ok().as_deref() == Some(npub.as_str()) {
return false;
}
}
let mut state = STATE.lock().await;
if state.interner.lookup(&npub).is_none() {
state.insert_or_replace_profile(&npub, Profile::new());
}
if let Some(id) = state.interner.lookup(&npub) {
{
let profile = match state.get_profile_mut_by_id(id) {
Some(p) => p,
None => return false,
};
profile.flags.set_blocked(true);
}
let slim = state.serialize_profile(id).unwrap();
drop(state);
emit_event("profile_update", &slim);
handler.on_profile_fetched(&slim, "", "");
true
} else {
false
}
}
pub async fn unblock_user(npub: String, handler: &dyn ProfileSyncHandler) -> bool {
let mut state = STATE.lock().await;
if let Some(id) = state.interner.lookup(&npub) {
{
let profile = match state.get_profile_mut_by_id(id) {
Some(p) => p,
None => return false,
};
profile.flags.set_blocked(false);
}
let slim = state.serialize_profile(id).unwrap();
drop(state);
emit_event("profile_update", &slim);
handler.on_profile_fetched(&slim, "", "");
true
} else {
false
}
}
pub async fn get_blocked_users() -> Vec<crate::SlimProfile> {
let state = STATE.lock().await;
state.profiles.iter()
.filter(|p| p.flags.is_blocked())
.filter_map(|p| state.serialize_profile(p.id))
.collect()
}
pub async fn set_nickname(npub: String, nickname: String, handler: &dyn ProfileSyncHandler) -> bool {
let mut state = STATE.lock().await;
if let Some(id) = state.interner.lookup(&npub) {
{
let profile = match state.get_profile_mut_by_id(id) {
Some(p) => p,
None => return false,
};
profile.nickname = nickname.into_boxed_str();
}
let slim = state.serialize_profile(id).unwrap();
drop(state);
emit_event("profile_nick_changed", &serde_json::json!({
"profile_id": &npub,
"value": &slim.nickname
}));
handler.on_profile_fetched(&slim, "", "");
true
} else {
false
}
}
pub async fn start_profile_sync_processor(handler: Arc<dyn ProfileSyncHandler>) {
let mut last_own_profile_sync = Instant::now();
let own_profile_sync_interval = Duration::from_secs(5 * 60);
loop {
if last_own_profile_sync.elapsed() >= own_profile_sync_interval {
let state = STATE.lock().await;
if let Some(own_profile) = state.profiles.iter().find(|p| p.flags.is_mine()) {
let npub = state.interner.resolve(own_profile.id).unwrap_or("").to_string();
drop(state);
let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
queue.add(npub, SyncPriority::Low, false);
}
last_own_profile_sync = Instant::now();
}
let (should_wait, batch) = {
let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
if queue.is_processing {
(true, vec![])
} else {
queue.is_processing = true;
let batch = queue.get_next_batch();
for entry in &batch {
queue.mark_processing(&entry.npub);
}
(false, batch)
}
};
if should_wait {
tokio::time::sleep(Duration::from_secs(1)).await;
continue;
}
if batch.is_empty() {
{
let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
queue.is_processing = false;
}
tokio::time::sleep(Duration::from_secs(1)).await;
continue;
}
let batch_session = crate::state::SessionGuard::capture();
for entry in &batch {
if !batch_session.is_valid() {
break;
}
load_profile(entry.npub.clone(), handler.as_ref()).await;
{
let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
queue.mark_done(&entry.npub);
}
tokio::time::sleep(Duration::from_millis(100)).await;
}
{
let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
queue.is_processing = false;
}
tokio::time::sleep(Duration::from_millis(500)).await;
}
}
pub fn queue_profile_sync(npub: String, priority: SyncPriority, force_refresh: bool) {
let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
queue.add(npub, priority, force_refresh);
}
pub async fn queue_chat_profiles(chat_id: String, is_opening: bool) {
let state = STATE.lock().await;
let chat = match state.get_chat(&chat_id) {
Some(c) => c,
None => return,
};
let base_priority = if is_opening {
SyncPriority::High
} else {
SyncPriority::Medium
};
let mut profiles_to_queue = Vec::new();
for &handle in chat.participants() {
let member_npub = match state.interner.resolve(handle) {
Some(s) => s.to_string(),
None => continue,
};
let has_metadata = state.get_profile_by_id(handle)
.map(|p| {
let has_data = !p.name.is_empty() || !p.display_name.is_empty() || !p.avatar.is_empty();
let was_fetched = p.last_updated > 0;
has_data || was_fetched
})
.unwrap_or(false);
let priority = if !has_metadata {
SyncPriority::Critical
} else {
base_priority
};
profiles_to_queue.push((member_npub, priority));
}
drop(state);
let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
for (npub, priority) in profiles_to_queue {
queue.add(npub, priority, false);
}
}
pub fn refresh_profile_now(npub: String) {
let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
queue.add(npub, SyncPriority::Critical, true);
}
pub async fn sync_all_profiles() {
let state = STATE.lock().await;
let mut profiles_to_queue = Vec::new();
for profile in &state.profiles {
let npub = match state.interner.resolve(profile.id) {
Some(s) => s.to_string(),
None => continue,
};
let has_metadata = !profile.name.is_empty() || !profile.display_name.is_empty() || !profile.avatar.is_empty();
let was_fetched = profile.last_updated > 0;
let priority = if !has_metadata && !was_fetched {
SyncPriority::Critical
} else {
SyncPriority::Low
};
profiles_to_queue.push((npub, priority));
}
drop(state);
let mut queue = PROFILE_SYNC_QUEUE.lock().unwrap();
for (npub, priority) in profiles_to_queue {
queue.add(npub, priority, false);
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn sync_priority_cache_windows() {
assert_eq!(SyncPriority::Critical.cache_window(), Duration::from_secs(0));
assert_eq!(SyncPriority::High.cache_window(), Duration::from_secs(300));
assert_eq!(SyncPriority::Medium.cache_window(), Duration::from_secs(1800));
assert_eq!(SyncPriority::Low.cache_window(), Duration::from_secs(86400));
}
#[test]
fn sync_priority_batch_sizes() {
assert_eq!(SyncPriority::Critical.batch_size(), 10);
assert_eq!(SyncPriority::High.batch_size(), 20);
assert_eq!(SyncPriority::Medium.batch_size(), 30);
assert_eq!(SyncPriority::Low.batch_size(), 50);
}
#[test]
fn queue_add_and_dedup() {
let mut queue = ProfileSyncQueue::new();
queue.add("npub1alice".to_string(), SyncPriority::Low, false);
queue.add("npub1alice".to_string(), SyncPriority::High, false);
assert!(queue.low_queue.is_empty());
assert_eq!(queue.high_queue.len(), 1);
assert_eq!(queue.high_queue[0].npub, "npub1alice");
}
#[test]
fn queue_skips_processing() {
let mut queue = ProfileSyncQueue::new();
queue.mark_processing("npub1bob");
queue.add("npub1bob".to_string(), SyncPriority::Critical, false);
assert!(queue.critical_queue.is_empty(), "should skip profiles being processed");
}
#[test]
fn queue_cache_window_skips() {
let mut queue = ProfileSyncQueue::new();
queue.mark_done("npub1carol");
queue.add("npub1carol".to_string(), SyncPriority::Low, false);
assert!(queue.low_queue.is_empty(), "should skip within cache window");
queue.add("npub1carol".to_string(), SyncPriority::Low, true);
assert_eq!(queue.low_queue.len(), 1, "force_refresh should bypass cache");
}
#[test]
fn queue_critical_skips_cache() {
let mut queue = ProfileSyncQueue::new();
queue.mark_done("npub1dave");
queue.add("npub1dave".to_string(), SyncPriority::Critical, false);
assert_eq!(queue.critical_queue.len(), 1, "Critical should always fetch");
}
#[test]
fn get_next_batch_priority_order() {
let mut queue = ProfileSyncQueue::new();
queue.low_queue.push_back(QueueEntry {
npub: "npub1low".to_string(),
added_at: Instant::now() - Duration::from_secs(600),
});
queue.critical_queue.push_back(QueueEntry {
npub: "npub1critical".to_string(),
added_at: Instant::now(),
});
let batch = queue.get_next_batch();
assert_eq!(batch.len(), 1);
assert_eq!(batch[0].npub, "npub1critical", "Critical should process before Low");
}
#[test]
fn get_next_batch_respects_delay() {
let mut queue = ProfileSyncQueue::new();
queue.high_queue.push_back(QueueEntry {
npub: "npub1new".to_string(),
added_at: Instant::now(),
});
let batch = queue.get_next_batch();
assert!(batch.is_empty(), "should not process before delay elapses");
}
#[test]
fn mark_done_updates_last_fetched() {
let mut queue = ProfileSyncQueue::new();
queue.mark_processing("npub1eve");
assert!(queue.processing.contains("npub1eve"));
queue.mark_done("npub1eve");
assert!(!queue.processing.contains("npub1eve"));
assert!(queue.last_fetched.contains_key("npub1eve"));
}
#[test]
fn noop_handler_compiles() {
let handler = NoOpProfileSyncHandler;
let slim = crate::SlimProfile::default();
handler.on_profile_fetched(&slim, "", "");
}
}