use std::collections::HashSet;

use models::chat::{messages, read_states, room_seq};
use models::room::rooms;
use models::users::user_recent_mentions;
use sea_orm::{ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter};
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use utoipa::{IntoParams, ToSchema};
use uuid::Uuid;

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

#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct ReadStateDto {
    pub room_id: Uuid,
    pub last_read_message_id: Option<Uuid>,
    pub last_read_seq: i64,
    pub last_viewed_at: Option<OffsetDateTime>,
    pub tip_seq: i64,
    pub unread_count: i64,
    pub unread_mentions: i64,
}

#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct SpaceReadStates {
    pub items: Vec<ReadStateDto>,
}

#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct MarkReadCmd {
    #[serde(default)]
    pub message_id: Option<Uuid>,
    #[serde(default)]
    pub seq: Option<i64>,
}

#[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)]
pub struct ThreadReadCmd {
    pub thread_id: Uuid,
}

impl ChatService {
    pub async fn mark_read(
        &self,
        caller: i64,
        room_id: Uuid,
        cmd: MarkReadCmd,
    ) -> Result<ReadStateDto, ChatError> {
        let _ctx = require_room_readable(self, room_id, caller).await?;

        let (message_id, seq) = match (cmd.message_id, cmd.seq) {
            (Some(message_id), _) => {
                let msg = messages::Entity::find_by_id(message_id)
                    .filter(messages::Column::RoomId.eq(room_id))
                    .one(self.database.read())
                    .await?
                    .ok_or(ChatError::NotFound)?;
                (Some(message_id), msg.seq)
            }
            (None, Some(seq)) => {
                let tip = room_seq::Entity::find_by_id(room_id)
                    .one(self.database.read())
                    .await?
                    .map(|s| s.seq)
                    .unwrap_or(0);
                (None, seq.clamp(0, tip))
            }
            (None, None) => {
                let tip = room_seq::Entity::find_by_id(room_id)
                    .one(self.database.read())
                    .await?
                    .map(|s| s.seq)
                    .unwrap_or(0);
                let last_msg = rooms::Entity::find_by_id(room_id)
                    .one(self.database.read())
                    .await?
                    .and_then(|r| r.last_message_id);
                (last_msg, tip)
            }
        };

        let changed =
            notify::advance_read_state(self.database.write(), caller, room_id, message_id, seq)
                .await?;
        if changed {
            self.audience
                .realtime
                .publish_user(
                    caller,
                    &serde_json::json!({
                        "kind": "read_marker",
                        "room_id": room_id,
                        "last_read_message_id": message_id,
                        "last_read_seq": seq,
                    }),
                )
                .await;
        }
        self.get_read_state(caller, room_id).await
    }

    pub async fn get_read_state(
        &self,
        caller: i64,
        room_id: Uuid,
    ) -> Result<ReadStateDto, ChatError> {
        require_room_readable(self, room_id, caller).await?;
        let tip = room_seq::Entity::find_by_id(room_id)
            .one(self.database.read())
            .await?
            .map(|s| s.seq)
            .unwrap_or(0);
        let state = read_states::Entity::find_by_id((caller, room_id))
            .one(self.database.read())
            .await?;
        let last_viewed_at = state.as_ref().and_then(|s| s.last_viewed_at);
        let unread_mentions = match last_viewed_at {
            Some(at) => {
                user_recent_mentions::Entity::find()
                    .filter(user_recent_mentions::Column::UserId.eq(caller))
                    .filter(user_recent_mentions::Column::ChannelId.eq(room_id))
                    .filter(user_recent_mentions::Column::CreatedAt.gt(at))
                    .count(self.database.read())
                    .await? as i64
            }
            None => {
                user_recent_mentions::Entity::find()
                    .filter(user_recent_mentions::Column::UserId.eq(caller))
                    .filter(user_recent_mentions::Column::ChannelId.eq(room_id))
                    .count(self.database.read())
                    .await? as i64
            }
        };
        let last_read_seq = state.as_ref().map(|s| s.last_read_seq).unwrap_or(0);
        Ok(ReadStateDto {
            room_id,
            last_read_message_id: state.as_ref().and_then(|s| s.last_read_message_id),
            last_read_seq,
            last_viewed_at,
            tip_seq: tip,
            unread_count: (tip - last_read_seq).max(0),
            unread_mentions,
        })
    }

    pub async fn list_space_read_states(
        &self,
        caller: i64,
        space_name: &str,
    ) -> Result<SpaceReadStates, ChatError> {
        let space = lookup_space_by_name(self, space_name).await?;
        require_space_member(self, &space, caller).await?;

        let space_rooms = rooms::Entity::find()
            .filter(rooms::Column::SpaceId.eq(space.id()))
            .filter(rooms::Column::IsActive.eq(true))
            .all(self.database.read())
            .await?;
        let room_ids: Vec<Uuid> = space_rooms.iter().map(|room| room.id).collect();
        let my_rooms: HashSet<Uuid> = models::room::room_members::Entity::find()
            .filter(models::room::room_members::Column::UserId.eq(caller))
            .filter(models::room::room_members::Column::RoomId.is_in(room_ids))
            .filter(models::room::room_members::Column::LeftAt.is_null())
            .all(self.database.read())
            .await?
            .into_iter()
            .map(|row| row.room_id)
            .collect();
        let mut items = Vec::new();
        for room in space_rooms {
            let is_visible =
                !room.is_private || room.owner_id == caller || my_rooms.contains(&room.id);
            if !is_visible {
                continue;
            }
            items.push(self.get_read_state(caller, room.id).await?);
        }
        Ok(SpaceReadStates { items })
    }

    pub async fn mark_thread_read(
        &self,
        caller: i64,
        thread_id: Uuid,
        message_id: Option<Uuid>,
    ) -> Result<(), ChatError> {
        let thread = models::chat::threads::Entity::find_by_id(thread_id)
            .one(self.database.read())
            .await?
            .ok_or(ChatError::NotFound)?;
        let _ctx = require_room_readable(self, thread.room_id, caller).await?;
        let (mid, seq) = match message_id {
            Some(message_id) => {
                let msg = messages::Entity::find_by_id(message_id)
                    .one(self.database.read())
                    .await?
                    .filter(|m| m.thread_id == Some(thread_id))
                    .ok_or(ChatError::NotFound)?;
                (message_id, msg.seq)
            }
            None => match thread.last_reply_id {
                Some(last) => {
                    let msg = messages::Entity::find_by_id(last)
                        .one(self.database.read())
                        .await?
                        .ok_or(ChatError::NotFound)?;
                    (last, msg.seq)
                }
                None => return Ok(()),
            },
        };
        self.upsert_thread_read(caller, thread_id, mid, seq).await
    }
}