//! `ChatIdentity`: display name, avatar and flair for the people in a room.
//!
//! # Why there is a cache here at all
//!
//! `SessionUser` carries no `avatar_url`. It lives only on the `users` row, so
//! every rendered message needs a join the session cannot supply. Backlog
//! replay after a deploy asks for a whole window of messages at once, and every
//! client reconnects at the same moment, so the uncached shape is one query per
//! author per reconnecting client.
//!
//! The crate already batches the lookup (`ChatIdentity::attach` collects
//! distinct authors and calls `identify` once). This cache is the second half:
//! it makes the batch itself mostly free, because a room's authors are a small
//! set that barely changes between reconnects.
//!
//! # Bounding it
//!
//! Both units run under a hard 512M cgroup cap where an OOM restarts the whole
//! site, so an unbounded map keyed by user id is not acceptable: the key space
//! is every account that has ever spoken. Entries expire, and the map is
//! capped; past the cap the coldest entries go. Neither bound is tuning, both
//! are the memory budget.
//!
//! Invalidation is on profile update rather than only by expiry, so a rename
//! takes effect without waiting out the TTL.

use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};

use async_trait::async_trait;
use dashmap::DashMap;
use livechat::{ChatError, ChatIdentity, Identity, UserId};
use sqlx::PgPool;
use uuid::Uuid;

use super::host_error;

/// How long a cached identity is trusted.
///
/// Short, because the only cost of a miss is one row and the cost of a stale
/// hit is a user seeing their own rename not take. Profile updates invalidate
/// directly, so this is the backstop for changes that arrive another way.
const IDENTITY_TTL: Duration = Duration::from_mins(5);

/// Most identities held at once.
///
/// Sized for the realistic shape: a few busy rooms of a few hundred people
/// each. Past it the cache sheds rather than grows, because growing is what
/// trips the cgroup cap.
const IDENTITY_CAPACITY: usize = 5_000;

struct Cached {
    identity: Identity,
    fetched_at: Instant,
}

/// Process-wide identity cache. Cheap to clone; clones share one map.
#[derive(Clone)]
pub struct IdentityCache {
    entries: Arc<DashMap<Uuid, Cached>>,
}

impl IdentityCache {
    pub fn new() -> Self {
        Self {
            entries: Arc::new(DashMap::new()),
        }
    }

    /// Drop a user's cached identity, so the next render re-reads the row.
    ///
    /// Called from the profile-update path. Cheap and safe to call for a user
    /// who was never cached.
    pub fn invalidate(&self, user: Uuid) {
        self.entries.remove(&user);
    }

    fn get_fresh(&self, user: Uuid, now: Instant) -> Option<Identity> {
        let entry = self.entries.get(&user)?;
        (now.duration_since(entry.fetched_at) < IDENTITY_TTL).then(|| entry.identity.clone())
    }

    fn insert(&self, user: Uuid, identity: Identity, now: Instant) {
        // Shed before inserting, so the map cannot exceed the cap even briefly.
        // Expired entries first: they are free to drop and usually enough.
        if self.entries.len() >= IDENTITY_CAPACITY {
            self.entries
                .retain(|_, e| now.duration_since(e.fetched_at) < IDENTITY_TTL);
        }
        // Still full means every entry is live. Take the coldest rather than
        // refusing to cache, or a room busy enough to fill the map would be the
        // one room that never gets a cache.
        if self.entries.len() >= IDENTITY_CAPACITY
            && let Some(coldest) = self
                .entries
                .iter()
                .min_by_key(|e| e.fetched_at)
                .map(|e| *e.key())
        {
            self.entries.remove(&coldest);
        }

        self.entries.insert(
            user,
            Cached {
                identity,
                fetched_at: now,
            },
        );
    }

    pub fn len(&self) -> usize {
        self.entries.len()
    }

    pub fn is_empty(&self) -> bool {
        self.entries.is_empty()
    }
}

impl Default for IdentityCache {
    fn default() -> Self {
        Self::new()
    }
}

pub struct MtChatIdentity {
    db: PgPool,
    cache: IdentityCache,
}

impl MtChatIdentity {
    pub fn new(db: PgPool, cache: IdentityCache) -> Self {
        Self { db, cache }
    }
}

#[async_trait]
impl ChatIdentity for MtChatIdentity {
    async fn identify(&self, users: &[UserId]) -> Result<HashMap<UserId, Identity>, ChatError> {
        let now = Instant::now();
        let mut resolved = HashMap::with_capacity(users.len());
        let mut misses = Vec::new();

        for id in users {
            match self.cache.get_fresh(id.0, now) {
                Some(identity) => {
                    resolved.insert(*id, identity);
                }
                None => misses.push(id.0),
            }
        }

        if misses.is_empty() {
            return Ok(resolved);
        }

        let rows = mt_db::queries::chat_identities(&self.db, &misses)
            .await
            .map_err(host_error)?;

        for row in rows {
            let identity = Identity {
                // Display name falls back to the username, which is NOT NULL.
                // A blank name in a chat room is worse than an ugly one.
                display_name: row.display_name.unwrap_or(row.username),
                avatar_url: row.avatar_url,
                // Role flair is deliberately absent: it is per-community, and
                // this cache is process-wide. Caching a role here would show a
                // moderator's badge in a community they moderate nothing in.
                flair: None,
            };
            self.cache.insert(row.id, identity.clone(), now);
            resolved.insert(UserId(row.id), identity);
        }

        // Anyone still missing could not be resolved: a deleted account. Left
        // out of the map rather than errored, so their messages stay readable
        // instead of the room failing to render.
        Ok(resolved)
    }
}

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

    fn identity(name: &str) -> Identity {
        Identity {
            display_name: name.to_owned(),
            avatar_url: None,
            flair: None,
        }
    }

    #[test]
    fn a_fresh_entry_is_returned_and_a_stale_one_is_not() {
        let cache = IdentityCache::new();
        let user = Uuid::new_v4();
        let t0 = Instant::now();

        cache.insert(user, identity("alice"), t0);

        assert_eq!(
            cache.get_fresh(user, t0).map(|i| i.display_name),
            Some("alice".to_owned())
        );
        assert!(
            cache.get_fresh(user, t0 + IDENTITY_TTL).is_none(),
            "the TTL boundary is exclusive"
        );
    }

    #[test]
    fn invalidate_makes_a_rename_visible_without_waiting_out_the_ttl() {
        let cache = IdentityCache::new();
        let user = Uuid::new_v4();
        let t0 = Instant::now();

        cache.insert(user, identity("old name"), t0);
        cache.invalidate(user);

        assert!(cache.get_fresh(user, t0).is_none());
    }

    #[test]
    fn invalidating_an_uncached_user_is_harmless() {
        IdentityCache::new().invalidate(Uuid::new_v4());
    }

    #[test]
    fn expired_entries_are_shed_before_the_cap_bites() {
        let cache = IdentityCache::new();
        let t0 = Instant::now();

        for _ in 0..IDENTITY_CAPACITY {
            cache.insert(Uuid::new_v4(), identity("old"), t0);
        }
        assert_eq!(cache.len(), IDENTITY_CAPACITY);

        // Long enough later that everything already held is expired.
        cache.insert(Uuid::new_v4(), identity("new"), t0 + IDENTITY_TTL * 2);

        assert_eq!(cache.len(), 1, "the expired generation went in one sweep");
    }

    #[test]
    fn a_full_cache_of_live_entries_still_accepts_a_new_one() {
        // The room busy enough to fill the map must not be the one room that
        // never gets a cache.
        let cache = IdentityCache::new();
        let t0 = Instant::now();

        for i in 0..IDENTITY_CAPACITY {
            // Staggered so there is a distinct coldest entry to evict.
            cache.insert(
                Uuid::new_v4(),
                identity("live"),
                t0 + Duration::from_millis(i as u64),
            );
        }

        let newcomer = Uuid::new_v4();
        cache.insert(newcomer, identity("newcomer"), t0 + Duration::from_secs(1));

        assert!(cache.len() <= IDENTITY_CAPACITY, "the cap holds");
        assert!(
            cache
                .get_fresh(newcomer, t0 + Duration::from_secs(1))
                .is_some(),
            "the new entry is the one kept"
        );
    }
}