vector_core/profile/

sync.rs

//! Profile sync — priority queue, background processor, and relay fetching.
//!
//! The sync queue batches profile fetches by priority (Critical → High → Medium → Low),
//! with cache windows to avoid hammering relays. The background processor
//! drains the queue and calls `load_profile` for each entry.
//!
//! Platform-specific work (DB persistence, image caching) is handled by the
//! `ProfileSyncHandler` trait — src-tauri provides `TauriProfileSyncHandler`,
//! CLI provides a no-op or logging implementation.

use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::{Arc, Mutex};
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;

// ============================================================================
// SyncPriority
// ============================================================================

/// Priority levels for profile syncing.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
pub enum SyncPriority {
    Critical,  // No metadata OR user clicked — fetch immediately
    High,      // Active chats — fetch soon
    Medium,    // Recent chats — fetch eventually
    Low,       // Old chats with metadata — passive refresh
}

impl SyncPriority {
    /// Cache window duration — how long before a profile can be re-fetched.
    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),
        }
    }

    /// Processing delay — how long after queuing before fetching.
    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),
        }
    }

    /// Maximum batch size for this priority.
    pub fn batch_size(&self) -> usize {
        match self {
            SyncPriority::Critical => 10,
            SyncPriority::High => 20,
            SyncPriority::Medium => 30,
            SyncPriority::Low => 50,
        }
    }
}

// ============================================================================
// QueueEntry
// ============================================================================

#[derive(Debug, Clone)]
pub(crate) struct QueueEntry {
    npub: String,
    added_at: Instant,
}

// ============================================================================
// ProfileSyncQueue
// ============================================================================

/// Profile sync queue manager with four priority lanes.
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 Default for ProfileSyncQueue {
    fn default() -> Self { Self::new() }
}

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,
        }
    }

    /// Add a profile to the sync queue.
    pub fn add(&mut self, npub: String, priority: SyncPriority, force_refresh: bool) {
        if self.processing.contains(&npub) {
            return;
        }

        // Check cache window (unless force_refresh)
        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);
    }

    /// Drop every queued + in-flight entry. Used by `reset_session()` so a
    /// post-reset processor doesn't keep fetching the prior account's contacts.
    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();
    }

    /// Get the next batch of profiles ready to process (highest priority first).
    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());
    }
}

// ============================================================================
// Global queue
// ============================================================================

/// Queued work for THIS account's contacts. The processor loop is
/// process-lifetime and services whichever queue the live session holds, so a
/// swap leaves the prior account's entries behind rather than fetching them.
struct ProfileSyncQueueKey;

fn profile_sync_queue() -> Arc<Mutex<ProfileSyncQueue>> {
    crate::db::current_session().scoped::<ProfileSyncQueueKey, _>()
}

// ============================================================================
// ProfileSyncHandler — platform-specific callbacks
// ============================================================================

/// Callback trait for platform-specific profile sync work.
///
/// The core `load_profile` handles relay fetching, STATE updates, and
/// EventEmitter notifications. This trait covers what differs per platform:
/// - **DB persistence** (SQLite upsert)
/// - **Image caching** (avatar/banner download + disk cache)
pub trait ProfileSyncHandler: Send + Sync {
    /// Called after a profile is fetched from relays and updated in STATE.
    /// `slim` is ready for DB persistence. `avatar_url`/`banner_url` are
    /// for image caching (may be empty).
    fn on_profile_fetched(&self, _slim: &crate::SlimProfile, _avatar_url: &str, _banner_url: &str) {}
}

/// No-op handler for CLI/tests.
pub struct NoOpProfileSyncHandler;
impl ProfileSyncHandler for NoOpProfileSyncHandler {}

// ============================================================================
// load_profile — core relay fetch + STATE update
// ============================================================================

/// Fetch a profile's metadata and status from relays, update STATE, and
/// notify via EventEmitter + handler callback.
///
/// Returns `true` if the fetch succeeded (even if nothing changed).
pub async fn load_profile(npub: String, handler: &dyn ProfileSyncHandler) -> bool {
    let client = match nostr_client() {
        Some(c) => c,
        None => return false,
    };

    // Session captured for the whole load_profile lifecycle. Relay
    // fetches can sleep multi-second; we re-check before writing back
    // to STATE / DB so a mid-fetch swap doesn't land account A's
    // profile in account B's storage.

    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,
    };

    // Grab old status (or create profile if missing)
    let (old_status_title, old_status_purpose, old_status_url): (String, String, String);
    let old_status_emoji_tags: Vec<crate::types::EmojiTag>;
    {
        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();
                old_status_emoji_tags = p.status_emoji_tags().to_vec();
            }
            None => {
                state.insert_or_replace_profile(&npub, Profile::new());
                old_status_title = String::new();
                old_status_purpose = String::new();
                old_status_url = String::new();
                old_status_emoji_tags = Vec::new();
            }
        }
    }

    // Fetch status (kind 30315) from relays
    let status_filter = Filter::new()
        .author(profile_pubkey)
        .kind(Kind::from_u16(30315))
        .limit(1);

    let (status_title, status_purpose, status_url, status_emoji_tags) = match client
        .fetch_events(status_filter).timeout(Duration::from_secs(15))
        .await
    {
        Ok(res) => {
            if !res.is_empty() {
                let status_event = res.first().unwrap();
                (
                    clamp_status(status_event.content.clone()),
                    status_event.tags.first()
                        .and_then(|t| t.content())
                        .unwrap_or_default()
                        .to_string(),
                    String::new(),
                    crate::types::EmojiTag::extract_from_tags(status_event.tags.iter()),
                )
            } else {
                (old_status_title, old_status_purpose, old_status_url, old_status_emoji_tags)
            }
        }
        Err(_) => (old_status_title, old_status_purpose, old_status_url, old_status_emoji_tags),
    };

    // Fetch metadata from relays
    // `Client::fetch_metadata` is gone: fetch the newest kind-0 and parse it.
    let fetch_result = client
        .fetch_events(
            Filter::new()
                .author(profile_pubkey)
                .kind(Kind::Metadata)
                .limit(1),
        )
        .timeout(Duration::from_secs(15))
        .await
        .map(|events| {
            events
                .into_iter()
                .max_by_key(|e| e.created_at)
                .and_then(|e| Metadata::from_json(&e.content).ok())
        });


    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);

                        // Update status
                        let status_changed = profile.status_title() != status_title.as_str()
                            || profile.status_purpose() != status_purpose.as_str()
                            || profile.status_url() != status_url.as_str()
                            || profile.status_emoji_tags() != status_emoji_tags.as_slice();
                        // Only touch the extras box when there's a real status to store or one
                        // already exists to clear — never materialize an empty box on the common
                        // status-less profile (that would make it larger than before the split).
                        let has_status = !status_title.is_empty()
                            || !status_purpose.is_empty() || !status_url.is_empty();
                        if profile.extras.is_some() || has_status {
                            let ex = profile.extras_mut();
                            ex.status_title = status_title.into_boxed_str();
                            ex.status_purpose = status_purpose.into_boxed_str();
                            ex.status_url = status_url.into_boxed_str();
                            ex.status_emoji_tags = status_emoji_tags.into_boxed_slice();
                        }

                        // Update metadata
                        let metadata_changed = profile.from_metadata(meta.unwrap());

                        // Update timestamp
                        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 {
                    // Notify UI via EventEmitter
                    emit_event("profile_update", &slim);
                    // Platform-specific: DB persist + image caching
                    handler.on_profile_fetched(&slim, &avatar_url, &banner_url);
                }
                true
            } else {
                // No metadata on relays — update timestamp so we don't keep retrying
                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,
    }
}

// ============================================================================
// update_profile — publish metadata to relays
// ============================================================================

/// Update the current user's profile metadata and broadcast to relays.
///
/// Merges the provided fields with the existing profile (empty = keep existing).
/// After successful broadcast, updates STATE and notifies via EventEmitter + handler.
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
}

/// Publish the current user's profile and mark it as a bot (`bot: true` in the metadata). The SDK
/// uses this so every bot it builds is tagged; human clients use [`update_profile`].
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,
    };

    // Build metadata from current profile, then drop the lock before network I/O
    let meta = {
        let state = STATE.lock().await;
        let npub = match my_public_key.to_bech32() {
            Ok(n) => n,
            Err(_) => return false,
        };
        // Start from the existing profile if we have one, else a blank profile so a first-time
        // update (e.g. a freshly-created bot that has never published a kind-0) still works.
        let profile = state.get_profile(&npub).cloned().unwrap_or_default();

        // Merge: use new value if provided, else carry existing
        let mut meta = Metadata::new().name(if name.is_empty() {
            &*profile.name
        } else {
            name.as_str()
        });

        // Avatar
        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);
            }
        }

        // Banner
        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);
            }
        }

        // Carry forward display_name
        if !profile.display_name.is_empty() {
            meta = meta.display_name(&*profile.display_name);
        }

        // About
        meta = meta.about(if about.is_empty() {
            &*profile.about
        } else {
            about.as_str()
        });

        // Carry forward remaining fields
        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
    }; // STATE lock dropped before network I/O

    // SDK-built bots carry `bot: true` so clients can badge them; human clients never set it.
    let meta = if is_bot { meta.custom_field("bot", true) } else { meta };

    // Build and sign Kind 0 metadata event
    let metadata_json = serde_json::to_string(&meta).unwrap();
    let metadata_event = EventBuilder::new(Kind::Metadata, metadata_json)
        .tag(Tag::custom("client", vec!["vector"]));

    let Ok(event) = crate::sign_builder(metadata_event).await else {
        return false;
    };

    // Broadcast — first-ACK so UI updates as soon as the fastest relay responds
    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;
                // Apply the published metadata to our own profile, creating the entry if this
                // identity had none yet (a freshly-created account is interned here on first set).
                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
        }
    }
}

// ============================================================================
// update_status — publish status to relays
// ============================================================================

/// Update the current user's status (kind 30315) and broadcast to relays.
///
/// Status length cap in Unicode scalar characters, enforced on BOTH sides:
/// our own publishes and every stored inbound status. Characters, not bytes —
/// a byte cap would chop emoji-heavy statuses to a third of what text gets.
pub const STATUS_MAX_CHARS: usize = 120;

/// Truncate a status to [`STATUS_MAX_CHARS`] on a character boundary.
fn clamp_status(s: String) -> String {
    if s.chars().count() <= STATUS_MAX_CHARS {
        s
    } else {
        s.chars().take(STATUS_MAX_CHARS).collect()
    }
}

/// Status is ephemeral — updated in STATE + frontend but not persisted to DB.
/// (Re-fetched from relays on next `load_profile` call.)
pub async fn update_status(status: String) -> bool {
    let status = clamp_status(status);
    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,
    };

    // Build and sign kind 30315 status event. `:shortcode:`s from the user's
    // equipped packs ride along as NIP-30 tags so other clients render them.
    let emoji_tags = crate::emoji_packs::resolve_outbound_emoji_tags(&status);
    let mut status_builder = EventBuilder::new(Kind::from_u16(30315), status.as_str())
        .tag(Tag::custom("d", vec!["general"]));
    for et in &emoji_tags {
        status_builder = status_builder.tag(Tag::custom("emoji", [et.shortcode.clone(), et.url.clone()]));
    }

    let Ok(event) = crate::sign_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,
                };
                let ex = profile.extras_mut();
                ex.status_purpose = "general".into();
                ex.status_title = status.into_boxed_str();
                ex.status_emoji_tags = emoji_tags.into_boxed_slice();
            }

            let slim = state.serialize_profile(id).unwrap();
            // Persist NOW: without this the new status (and its emoji tags)
            // survives a reboot only if a self-profile sync happens to run
            // before the app closes.
            let _ = crate::db::profiles::set_profile(&slim);
            emit_event("profile_update", &slim);
            true
        }
        Err(_) => false,
    }
}

// ============================================================================
// block / unblock / nickname / blocked list
// ============================================================================

/// Block a user by npub. DM events from blocked users are dropped after decryption.
/// Group messages are stored but filtered in the UI.
///
/// Returns `false` if trying to block yourself or if the profile can't be found.
pub async fn block_user(npub: String, handler: &dyn ProfileSyncHandler) -> bool {
    // Prevent blocking yourself
    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;

    // Create profile if it doesn't exist (can block someone with no prior contact)
    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
    }
}

/// Unblock a user by npub.
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
    }
}

/// Get all blocked profiles.
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()
}

/// Set a nickname for a profile.
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.extras_mut().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
    }
}

// ============================================================================
// Background processor
// ============================================================================

/// Background processor that continuously drains the profile sync queue.
///
/// Spawned once at startup. Processes batches in priority order, calling
/// `load_profile` for each entry with the provided handler.
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 {
        // Periodically queue our own profile to detect changes from other Nostr apps
        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 owner = profile_sync_queue();
                let mut queue = owner.lock().unwrap();
                queue.add(npub, SyncPriority::Low, false);
            }
            last_own_profile_sync = Instant::now();
        }

        // Get next batch (lock scoped)
        let (should_wait, batch) = {
            let owner = profile_sync_queue();
            let mut queue = owner.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 owner = profile_sync_queue();
                let mut queue = owner.lock().unwrap();
                queue.is_processing = false;
            }
            tokio::time::sleep(Duration::from_secs(1)).await;
            continue;
        }

        // Session captured per-batch so a swap aborts the loop before
        // account A's queue work lands in account B's DB. The next
        // outer-loop iteration picks up the new session's queue cleanly.

        for entry in &batch {
            load_profile(entry.npub.clone(), handler.as_ref()).await;

            {
                let owner = profile_sync_queue();
                let mut queue = owner.lock().unwrap();
                queue.mark_done(&entry.npub);
            }

            tokio::time::sleep(Duration::from_millis(100)).await;
        }

        // Release processing lock
        {
            let owner = profile_sync_queue();
            let mut queue = owner.lock().unwrap();
            queue.is_processing = false;
        }

        tokio::time::sleep(Duration::from_millis(500)).await;
    }
}

// ============================================================================
// Public API
// ============================================================================

/// Queue a single profile for syncing.
pub fn queue_profile_sync(npub: String, priority: SyncPriority, force_refresh: bool) {
    let owner = profile_sync_queue();
    let mut queue = owner.lock().unwrap();
    queue.add(npub, priority, force_refresh);
}

/// Queue all profiles for a chat.
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 owner = profile_sync_queue();
    let mut queue = owner.lock().unwrap();
    for (npub, priority) in profiles_to_queue {
        queue.add(npub, priority, false);
    }
}

/// Force immediate refresh of a profile (for user clicks).
pub fn refresh_profile_now(npub: String) {
    let owner = profile_sync_queue();
    let mut queue = owner.lock().unwrap();
    queue.add(npub, SyncPriority::Critical, true);
}

/// Sync all profiles in the system.
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 owner = profile_sync_queue();
    let mut queue = owner.lock().unwrap();
    for (npub, priority) in profiles_to_queue {
        queue.add(npub, priority, false);
    }
}

// ============================================================================
// Tests
// ============================================================================

#[cfg(test)]
mod status_clamp_tests {
    use super::*;

    #[test]
    fn a_status_clamps_at_120_characters_not_bytes() {
        assert_eq!(clamp_status("hi".to_string()), "hi");
        let exact: String = "a".repeat(STATUS_MAX_CHARS);
        assert_eq!(clamp_status(exact.clone()), exact, "at the cap is untouched");
        let long = "b".repeat(10_000);
        assert_eq!(clamp_status(long).chars().count(), STATUS_MAX_CHARS);
        // Characters, not bytes: 120 four-byte emoji survive whole.
        let emoji: String = "\u{1F980}".repeat(STATUS_MAX_CHARS);
        let clamped = clamp_status(format!("{emoji}overflow"));
        assert_eq!(clamped.chars().count(), STATUS_MAX_CHARS);
        assert_eq!(clamped, emoji, "truncation lands on a character boundary");
    }
}

#[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);

        // Should be in High queue only (deduped from Low)
        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();

        // Mark as recently fetched
        queue.mark_done("npub1carol");

        // Try to add with Low priority (24h cache window) — should skip
        queue.add("npub1carol".to_string(), SyncPriority::Low, false);
        assert!(queue.low_queue.is_empty(), "should skip within cache window");

        // Force refresh should bypass cache
        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();

        // Critical has 0s cache window — always fetches
        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();

        // Add to Low and Critical queues
        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();

        // Add a High priority entry just now (5s delay required)
        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, "", "");
    }
}