use models::chat::presence_states::{self, PresenceStatus};
use models::space::space_members::{self, MemberStatus};
use models::users::user_preferences;
use redis::AsyncCommands;
use sea_orm::{
    ActiveModelTrait, ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, QuerySelect, Set,
};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use time::OffsetDateTime;
use utoipa::ToSchema;
use uuid::Uuid;

use crate::errors::ChatError;
use crate::services::service::ChatService;
use crate::share::lookup::{lookup_space_by_name, require_space_member};

#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct PresenceDto {
    pub user_id: i64,
    pub user: Option<space::services::spaces::UserDto>,
    pub status: PresenceStatus,
    pub activities: serde_json::Value,
    pub last_seen_at: Option<OffsetDateTime>,
    pub online: bool,
}

#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct SetPresenceCmd {
    pub status: PresenceStatus,
    #[serde(default)]
    pub activities: Option<serde_json::Value>,
}

fn presence_key(uid: i64) -> String {
    format!("chat:presence:{uid}")
}

#[derive(Debug, Serialize, Deserialize)]
struct PresencePayload {
    status: PresenceStatus,
    activities: serde_json::Value,
    at: i64,
}

fn encode_presence(status: PresenceStatus, activities: serde_json::Value) -> Option<String> {
    serde_json::to_string(&PresencePayload {
        status,
        activities,
        at: OffsetDateTime::now_utc().unix_timestamp(),
    })
    .map_err(|error| tracing::warn!(%error, "presence payload serialize failed"))
    .ok()
}

/// 隐藏在线状态时的对外形态：一律表现为离线，并抹掉最后在线时间与活动。
fn mask_presence(dto: &mut PresenceDto) {
    dto.status = PresenceStatus::Offline;
    dto.online = false;
    dto.activities = serde_json::json!([]);
    dto.last_seen_at = None;
}

impl ChatService {
    pub async fn set_presence(
        &self,
        caller: i64,
        cmd: SetPresenceCmd,
    ) -> Result<PresenceDto, ChatError> {
        let now = OffsetDateTime::now_utc();
        let activities = cmd.activities.unwrap_or(serde_json::json!([]));
        let payload = PresencePayload {
            status: cmd.status,
            activities: activities.clone(),
            at: now.unix_timestamp(),
        };
        let ttl = self.config().chat.presence_ttl_secs;

        let mut conn = self.state.redis.connection_clone();
        let key = presence_key(caller);
        let value = serde_json::to_string(&payload)?;
        let result: Result<(), redis::RedisError> = match &mut conn {
            export::RedisConnection::Standalone(c) => c.set_ex(&key, &value, ttl).await,
            export::RedisConnection::Cluster(c) => c.set_ex(&key, &value, ttl).await,
        };
        if let Err(error) = result {
            tracing::warn!(%error, "failed to write presence to redis");
        }

        let existing = presence_states::Entity::find_by_id(caller)
            .one(self.db().write())
            .await?;
        match existing {
            Some(row) => {
                let mut am: presence_states::ActiveModel = row.into();
                am.status = Set(cmd.status);
                am.activities = Set(activities.clone());
                am.last_seen_at = Set(Some(now));
                am.updated_at = Set(now);
                am.update(self.db().write()).await?;
            }
            None => {
                presence_states::ActiveModel {
                    user_id: Set(caller),
                    status: Set(cmd.status),
                    activities: Set(activities.clone()),
                    last_seen_at: Set(Some(now)),
                    updated_at: Set(now),
                }
                .insert(self.db().write())
                .await?;
            }
        }

        self.realtime
            .publish_user(
                caller,
                &serde_json::json!({"kind": "presence", "status": cmd.status}),
            )
            .await;

        Ok(PresenceDto {
            user_id: caller,
            user: None,
            status: cmd.status,
            activities,
            last_seen_at: Some(now),
            online: !matches!(cmd.status, PresenceStatus::Offline),
        })
    }

    /* ------------------------------------------------------------- 生命周期 */

    /// 聊天 WS 建立连接：用户上线。
    ///
    /// 已有键（手动 Away/Dnd/隐身）原样保留、只续期；没有键才写 Active 并
    /// 广播。多标签页/多实例共享同一份 Redis 状态：谁先连谁写，后来者续期。
    /// 生命周期 TTL 取两倍心跳，崩溃后由 TTL 兜底回离线。
    pub async fn mark_online(&self, uid: i64) {
        let ttl = self.config().chat.online_ttl_secs();
        let key = presence_key(uid);
        let Some(value) = encode_presence(PresenceStatus::Active, serde_json::json!([])) else {
            return;
        };
        match export::redis::set_nx_ex(&self.state.redis, &key, &value, ttl).await {
            Ok(true) => {
                self.realtime
                    .publish_user(
                        uid,
                        &serde_json::json!({"kind": "presence", "status": PresenceStatus::Active}),
                    )
                    .await;
            }
            Ok(false) => {
                if let Err(error) = export::redis::touch_expire(&self.state.redis, &key, ttl).await
                {
                    tracing::warn!(%error, "presence renew failed");
                }
            }
            Err(error) => tracing::warn!(%error, "presence mark online failed"),
        }
    }

    /// 心跳续期：键在只续期；键已丢失（休眠唤醒、网络割接）时补写 Active
    /// ——连接还活着就应当在线。补写走 `SET NX`，不会覆盖用户此刻手动设置
    /// 的状态。
    pub async fn touch_presence(&self, uid: i64) {
        let ttl = self.config().chat.online_ttl_secs();
        let key = presence_key(uid);
        match export::redis::touch_expire(&self.state.redis, &key, ttl).await {
            Ok(true) => {}
            Ok(false) => {
                let Some(value) = encode_presence(PresenceStatus::Active, serde_json::json!([]))
                else {
                    return;
                };
                if let Err(error) =
                    export::redis::set_nx_ex(&self.state.redis, &key, &value, ttl).await
                {
                    tracing::warn!(%error, "presence re-mark failed");
                }
            }
            Err(error) => tracing::warn!(%error, "presence touch failed"),
        }
    }

    /// 最后一条聊天连接关闭：清 Redis 键并落库 `last_seen_at`。
    ///
    /// 键的消失就是"离线"的唯一真相；DB 行只保留最后一次状态与时间，供
    /// Redis 缺失时回退读取。
    pub async fn mark_offline(&self, uid: i64) {
        let now = OffsetDateTime::now_utc();
        let key = presence_key(uid);
        // 先读再删：DB 里留下的是用户最后处在的状态（Active/Away/Dnd/隐身）。
        let last = export::redis::get_string(&self.state.redis, &key)
            .await
            .ok()
            .flatten()
            .and_then(|value| serde_json::from_str::<PresencePayload>(&value).ok());
        if let Err(error) = export::redis::delete(&self.state.redis, &key).await {
            tracing::warn!(%error, "presence key delete failed");
        }
        let (status, activities) = match last {
            Some(payload) => (payload.status, payload.activities),
            None => (PresenceStatus::Offline, serde_json::json!([])),
        };
        match presence_states::Entity::find_by_id(uid)
            .one(self.db().write())
            .await
        {
            Ok(Some(row)) => {
                let mut am: presence_states::ActiveModel = row.into();
                am.status = Set(status);
                am.activities = Set(activities);
                am.last_seen_at = Set(Some(now));
                am.updated_at = Set(now);
                if let Err(error) = am.update(self.db().write()).await {
                    tracing::warn!(%error, "presence last_seen update failed");
                }
            }
            Ok(None) => {
                let row = presence_states::ActiveModel {
                    user_id: Set(uid),
                    status: Set(status),
                    activities: Set(activities),
                    last_seen_at: Set(Some(now)),
                    updated_at: Set(now),
                };
                if let Err(error) = row.insert(self.db().write()).await {
                    tracing::warn!(%error, "presence last_seen insert failed");
                }
            }
            Err(error) => tracing::warn!(%error, "presence row lookup failed"),
        }
        self.realtime
            .publish_user(
                uid,
                &serde_json::json!({"kind": "presence", "status": PresenceStatus::Offline}),
            )
            .await;
    }

    /// `show_online_status` 开关：没有偏好行（从未保存过设置）按公开处理。
    async fn show_online_map(&self, uids: &[i64]) -> Result<HashMap<i64, bool>, ChatError> {
        if uids.is_empty() {
            return Ok(HashMap::new());
        }
        let rows = user_preferences::Entity::find()
            .filter(user_preferences::Column::UserId.is_in(uids.to_vec()))
            .all(self.db().read())
            .await?;
        Ok(rows
            .into_iter()
            .map(|row| (row.user_id, row.show_online_status))
            .collect())
    }

    pub async fn get_presence(&self, uid: i64) -> Result<PresenceDto, ChatError> {
        let mut dto = self.get_presence_raw(uid).await?;
        let map = self.load_user_briefs(&[uid]).await?;
        dto.user = map.get(&uid).cloned();
        Ok(dto)
    }

    async fn get_presence_raw(&self, uid: i64) -> Result<PresenceDto, ChatError> {
        let mut conn = self.state.redis.connection_clone();
        let key = presence_key(uid);
        let value: Result<Option<String>, redis::RedisError> = match &mut conn {
            export::RedisConnection::Standalone(c) => c.get(&key).await,
            export::RedisConnection::Cluster(c) => c.get(&key).await,
        };
        if let Ok(Some(value)) = value
            && let Ok(payload) = serde_json::from_str::<PresencePayload>(&value)
        {
            return Ok(PresenceDto {
                user_id: uid,
                user: None,
                status: payload.status,
                activities: payload.activities,
                last_seen_at: Some(
                    OffsetDateTime::from_unix_timestamp(payload.at)
                        .unwrap_or_else(|_| OffsetDateTime::now_utc()),
                ),
                online: !matches!(payload.status, PresenceStatus::Offline),
            });
        }
        let row = presence_states::Entity::find_by_id(uid)
            .one(self.db().read())
            .await?;
        match row {
            Some(row) => Ok(PresenceDto {
                user_id: uid,
                user: None,
                status: PresenceStatus::Offline,
                activities: serde_json::json!([]),
                last_seen_at: row.last_seen_at,
                online: false,
            }),
            None => Ok(PresenceDto {
                user_id: uid,
                user: None,
                status: PresenceStatus::Offline,
                activities: serde_json::json!([]),
                last_seen_at: None,
                online: false,
            }),
        }
    }

    /// Presence of another user is only visible when both share an active
    /// space or a DM room; otherwise the requester gets NotFound. A target
    /// who turned off `show_online_status` always reads as offline.
    pub async fn get_presence_for(
        &self,
        caller: i64,
        target: i64,
    ) -> Result<PresenceDto, ChatError> {
        if caller == target || self.shares_room_with(caller, target).await? {
            let mut dto = self.get_presence(target).await?;
            if dto.user_id != caller {
                let visible = self
                    .show_online_map(&[target])
                    .await?
                    .get(&target)
                    .copied()
                    .unwrap_or(true);
                if !visible {
                    mask_presence(&mut dto);
                }
            }
            return Ok(dto);
        }
        Err(ChatError::NotFound)
    }

    async fn shares_room_with(&self, caller: i64, target: i64) -> Result<bool, ChatError> {
        use models::room::room_members;
        use models::space::space_members as sm;
        let my_spaces: Vec<Uuid> = sm::Entity::find()
            .filter(sm::Column::UserId.eq(caller))
            .select_only()
            .column(sm::Column::SpaceId)
            .into_tuple()
            .all(self.db().read())
            .await?;
        if !my_spaces.is_empty() {
            let shared = sm::Entity::find()
                .filter(sm::Column::UserId.eq(target))
                .filter(sm::Column::SpaceId.is_in(my_spaces))
                .count(self.db().read())
                .await?;
            if shared > 0 {
                return Ok(true);
            }
        }
        use models::room::rooms;
        use sea_orm::Condition;
        let my_rooms: Vec<Uuid> = room_members::Entity::find()
            .filter(room_members::Column::UserId.eq(caller))
            .filter(room_members::Column::LeftAt.is_null())
            .select_only()
            .column(room_members::Column::RoomId)
            .into_tuple()
            .all(self.db().read())
            .await?;
        if my_rooms.is_empty() {
            return Ok(false);
        }
        let shared = rooms::Entity::find()
            .filter(rooms::Column::Id.is_in(my_rooms.clone()))
            .filter(rooms::Column::SpaceId.is_null())
            .filter(
                Condition::any()
                    .add(rooms::Column::OwnerId.eq(caller))
                    .add(rooms::Column::OwnerId.eq(target)),
            )
            .count(self.db().read())
            .await?;
        if shared > 0 {
            return Ok(true);
        }
        let dm_peer = room_members::Entity::find()
            .filter(room_members::Column::UserId.eq(target))
            .filter(room_members::Column::RoomId.is_in(my_rooms))
            .filter(room_members::Column::LeftAt.is_null())
            .count(self.db().read())
            .await?;
        Ok(dm_peer > 0)
    }

    pub async fn get_presences(&self, uids: &[i64]) -> Result<Vec<PresenceDto>, ChatError> {
        if uids.is_empty() {
            return Ok(Vec::new());
        }
        let mut conn = self.state.redis.connection_clone();
        let keys: Vec<String> = uids.iter().map(|uid| presence_key(*uid)).collect();
        let values: Vec<Option<String>> = match &mut conn {
            export::RedisConnection::Standalone(c) => {
                redis::cmd("MGET")
                    .arg(keys.clone())
                    .query_async::<Vec<Option<String>>>(c)
                    .await?
            }
            export::RedisConnection::Cluster(c) => {
                // 集群里每个键可能落在不同槽位，MGET 不成立，只能逐个 GET。
                let mut values = Vec::with_capacity(keys.len());
                for key in &keys {
                    values.push(redis::cmd("GET").arg(key).query_async(c).await?);
                }
                values
            }
        };
        let mut by_uid: HashMap<i64, PresencePayload> = HashMap::new();
        for (uid, value) in uids.iter().zip(values) {
            if let Some(value) = value
                && let Ok(payload) = serde_json::from_str::<PresencePayload>(&value)
            {
                by_uid.insert(*uid, payload);
            }
        }
        let missing: Vec<i64> = uids
            .iter()
            .filter(|u| !by_uid.contains_key(u))
            .copied()
            .collect();
        let rows = if missing.is_empty() {
            Vec::new()
        } else {
            presence_states::Entity::find()
                .filter(presence_states::Column::UserId.is_in(missing))
                .all(self.db().read())
                .await?
        };
        let mut result = Vec::with_capacity(uids.len());
        for uid in uids {
            if let Some(payload) = by_uid.get(uid) {
                result.push(PresenceDto {
                    user_id: *uid,
                    user: None,
                    status: payload.status,
                    activities: payload.activities.clone(),
                    last_seen_at: Some(
                        OffsetDateTime::from_unix_timestamp(payload.at)
                            .unwrap_or_else(|_| OffsetDateTime::now_utc()),
                    ),
                    online: !matches!(payload.status, PresenceStatus::Offline),
                });
                continue;
            }
            let row = rows.iter().find(|r| r.user_id == *uid);
            result.push(PresenceDto {
                user_id: *uid,
                user: None,
                status: PresenceStatus::Offline,
                activities: serde_json::json!([]),
                last_seen_at: row.and_then(|r| r.last_seen_at),
                online: false,
            });
        }
        let map = self.load_user_briefs(uids).await?;
        for dto in result.iter_mut() {
            dto.user = map.get(&dto.user_id).cloned();
        }
        Ok(result)
    }

    pub async fn list_space_presences(
        &self,
        caller: i64,
        space_name: &str,
    ) -> Result<Vec<PresenceDto>, ChatError> {
        let space = lookup_space_by_name(self, space_name).await?;
        require_space_member(self, &space, caller).await?;
        let members = space_members::Entity::find()
            .filter(space_members::Column::SpaceId.eq(space.id))
            .filter(space_members::Column::Status.eq(MemberStatus::Active))
            .all(self.db().read())
            .await?;
        let mut uids: Vec<i64> = members.into_iter().map(|m| m.user_id).collect();
        if !uids.contains(&space.owner_id) {
            uids.push(space.owner_id);
        }
        let mut presences = self.get_presences(&uids).await?;
        // 隐私：关掉"展示在线状态"的成员对他人一律表现为离线，本人照常。
        let visible = self.show_online_map(&uids).await?;
        for dto in presences.iter_mut() {
            if dto.user_id != caller && !visible.get(&dto.user_id).copied().unwrap_or(true) {
                mask_presence(dto);
            }
        }
        Ok(presences)
    }
}

impl ChatService {
    /// 聊天连接建立：置在线并把这条连接登记进在线占位集合。
    ///
    /// 占位集合与 `presence` 键是两套语义：`chat:ws:{uid}` 回答"有多少条
    /// 连接还活着"（推送 worker 用它决定是否跳过系统通知），`chat:presence:{uid}`
    /// 回答"用户此刻的状态是什么"。同一连接同时续期两者。
    pub async fn connection_open(&self, uid: i64, session: &str) -> Result<(), ChatError> {
        self.mark_online(uid).await;
        let ttl = self.config().chat.online_ttl_secs();
        export::redis::zset_touch(
            &self.state.redis,
            &export::push::online_key(uid),
            session,
            ttl,
        )
        .await?;
        Ok(())
    }

    /// 心跳续期：两套状态一起续，避免"有连接但状态过期"的中间态。
    pub async fn connection_touch(&self, uid: i64, session: &str) -> Result<(), ChatError> {
        self.touch_presence(uid).await;
        let ttl = self.config().chat.online_ttl_secs();
        export::redis::zset_touch(
            &self.state.redis,
            &export::push::online_key(uid),
            session,
            ttl,
        )
        .await?;
        Ok(())
    }

    /// 连接关闭：注销该连接占位；只有最后一条连接消失才算离线。
    ///
    /// 多标签页场景下先关的那一页不能把用户判成离线——这是把
    /// `mark_offline` 放在 `zset_release` 返回值之后的原因。
    pub async fn connection_close(&self, uid: i64, session: &str) -> Result<(), ChatError> {
        let live =
            export::redis::zset_release(&self.state.redis, &export::push::online_key(uid), session)
                .await?;
        if live == 0 {
            self.mark_offline(uid).await;
        }
        Ok(())
    }
}