use models::chat::messages::{self, FLAG_HAS_THREAD};
use models::chat::thread_members::{self, ThreadNotifyLevel};
use models::chat::threads;
use models::room::room_members;
use sea_orm::sea_query::Expr;
use sea_orm::{
    ActiveModelTrait, ColumnTrait, ConnectionTrait, EntityTrait, ExprTrait, PaginatorTrait,
    QueryFilter, QueryOrder, QuerySelect, Set, TransactionTrait,
};
use serde::{Deserialize, Serialize};
use space::services::dto::UserDto;
use time::OffsetDateTime;
use utoipa::{IntoParams, ToSchema};
use uuid::Uuid;

use crate::errors::ChatError;
use crate::services::service::ChatService;
use crate::share::realtime::RoomEvent;
use crate::share::{authz, lookup};

#[derive(Debug, Clone, Serialize, ToSchema)]
pub struct ThreadDto {
    pub id: Uuid,
    pub room_id: Uuid,
    pub root_message_id: Uuid,
    pub owner: Option<UserDto>,
    pub owner_id: i64,
    pub name: Option<String>,
    pub reply_count: i64,
    pub participant_count: i64,
    pub last_reply_id: Option<Uuid>,
    pub last_reply_at: Option<OffsetDateTime>,
    pub archived: bool,
    pub locked: bool,
    pub created_at: OffsetDateTime,
    pub updated_at: OffsetDateTime,
}

pub type ThreadPage = export::pagination::Page<ThreadDto>;

#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct CreateThreadCmd {
    #[serde(default)]
    pub name: Option<String>,
}

#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct UpdateThreadCmd {
    #[serde(default)]
    pub name: Option<String>,
    #[serde(default)]
    pub archived: Option<bool>,
    #[serde(default)]
    pub locked: Option<bool>,
}

#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct ThreadMemberCmd {
    #[serde(default)]
    pub notify_level: Option<ThreadNotifyLevel>,
}

#[derive(Debug, Clone, Deserialize, IntoParams, ToSchema)]
pub struct ListThreadQuery {
    pub archived: Option<bool>,
    pub limit: Option<u64>,
    pub offset: Option<u64>,
}

pub(crate) async fn maintain_thread_on_reply(
    txn: &impl ConnectionTrait,
    thread: &threads::Model,
    message_id: Uuid,
    author: i64,
    now: OffsetDateTime,
) -> Result<(), ChatError> {
    threads::Entity::update_many()
        .filter(threads::Column::Id.eq(thread.id))
        .col_expr(
            threads::Column::ReplyCount,
            Expr::col(threads::Column::ReplyCount).add(1),
        )
        .col_expr(threads::Column::LastReplyId, Expr::value(message_id))
        .col_expr(threads::Column::LastReplyAt, Expr::value(now))
        .col_expr(threads::Column::UpdatedAt, Expr::value(now))
        .exec(txn)
        .await?;

    let member = thread_members::Entity::find_by_id((thread.id, author))
        .one(txn)
        .await?;
    match member {
        Some(_) => {}
        None => {
            thread_members::ActiveModel {
                thread_id: Set(thread.id),
                user_id: Set(author),
                notify_level: Set(ThreadNotifyLevel::All),
                last_read_message_id: Set(None),
                last_read_seq: Set(0),
                created_at: Set(now),
                updated_at: Set(now),
            }
            .insert(txn)
            .await?;
            threads::Entity::update_many()
                .filter(threads::Column::Id.eq(thread.id))
                .col_expr(
                    threads::Column::ParticipantCount,
                    Expr::col(threads::Column::ParticipantCount).add(1),
                )
                .exec(txn)
                .await?;
        }
    }
    Ok(())
}

impl ChatService {
    async fn build_thread_dto(
        &self,
        rows: Vec<threads::Model>,
    ) -> Result<Vec<ThreadDto>, ChatError> {
        let owner_ids: Vec<i64> = rows.iter().map(|t| t.owner_id).collect();
        let user_map = self.load_user_briefs(&owner_ids).await?;
        Ok(rows
            .into_iter()
            .map(|t| ThreadDto {
                id: t.id,
                room_id: t.room_id,
                root_message_id: t.root_message_id,
                owner: user_map.get(&t.owner_id).cloned(),
                owner_id: t.owner_id,
                name: t.name,
                reply_count: t.reply_count,
                participant_count: t.participant_count,
                last_reply_id: t.last_reply_id,
                last_reply_at: t.last_reply_at,
                archived: t.archived,
                locked: t.locked,
                created_at: t.created_at,
                updated_at: t.updated_at,
            })
            .collect())
    }

    pub async fn create_thread(
        &self,
        caller: i64,
        room_id: Uuid,
        message_id: Uuid,
        cmd: CreateThreadCmd,
    ) -> Result<ThreadDto, ChatError> {
        let _ctx = lookup::require_room_readable(self, room_id, caller).await?;

        let root = messages::Entity::find_by_id(message_id)
            .filter(messages::Column::RoomId.eq(room_id))
            .one(self.db().read())
            .await?
            .filter(|m| m.deleted_at.is_none())
            .ok_or(ChatError::NotFound)?;

        let existing = threads::Entity::find()
            .filter(threads::Column::RootMessageId.eq(message_id))
            .one(self.db().read())
            .await?;
        if let Some(existing) = existing {
            let items = self.build_thread_dto(vec![existing]).await?;
            return Ok(items.into_iter().next().unwrap());
        }

        let now = OffsetDateTime::now_utc();
        let thread_id = Uuid::now_v7();
        let name = cmd
            .name
            .clone()
            .map(|n| n.trim().to_string())
            .filter(|n| !n.is_empty());
        if name
            .as_ref()
            .is_some_and(|n| n.chars().count() > self.config().chat.max_thread_name_len)
        {
            return Err(ChatError::BadRequest("thread name too long".to_string()));
        }

        self.db()
            .primary()
            .transaction::<_, (), ChatError>(move |txn| {
                Box::pin(async move {
                    threads::ActiveModel {
                        id: Set(thread_id),
                        room_id: Set(room_id),
                        root_message_id: Set(message_id),
                        owner_id: Set(caller),
                        name: Set(name),
                        reply_count: Set(0),
                        participant_count: Set(1),
                        last_reply_id: Set(None),
                        last_reply_at: Set(None),
                        archived: Set(false),
                        locked: Set(false),
                        auto_archive_minutes: Set(None),
                        created_at: Set(now),
                        updated_at: Set(now),
                    }
                    .insert(txn)
                    .await?;

                    thread_members::ActiveModel {
                        thread_id: Set(thread_id),
                        user_id: Set(caller),
                        notify_level: Set(ThreadNotifyLevel::All),
                        last_read_message_id: Set(None),
                        last_read_seq: Set(0),
                        created_at: Set(now),
                        updated_at: Set(now),
                    }
                    .insert(txn)
                    .await?;

                    let new_flags = root.flags | FLAG_HAS_THREAD;
                    messages::Entity::update_many()
                        .filter(messages::Column::Id.eq(message_id))
                        .col_expr(messages::Column::Flags, Expr::value(new_flags))
                        .exec(txn)
                        .await?;

                    Ok(())
                })
            })
            .await?;

        let thread = threads::Entity::find_by_id(thread_id)
            .one(self.db().read())
            .await?
            .ok_or(ChatError::Internal("thread vanished after insert".into()))?;
        let event = RoomEvent::ThreadUpdated { room_id, thread_id };
        self.realtime.publish_room(&event).await;

        let items = self.build_thread_dto(vec![thread]).await?;
        Ok(items.into_iter().next().unwrap())
    }

    pub async fn list_threads(
        &self,
        caller: i64,
        room_id: Uuid,
        q: ListThreadQuery,
    ) -> Result<ThreadPage, ChatError> {
        let _ctx = lookup::require_room_readable(self, room_id, caller).await?;
        let (limit, offset) = export::pagination::page_bounds(
            q.limit,
            q.offset,
            self.config().chat.default_pagination,
            self.config().chat.max_pagination,
        );

        let mut select = threads::Entity::find().filter(threads::Column::RoomId.eq(room_id));
        if let Some(archived) = q.archived {
            select = select.filter(threads::Column::Archived.eq(archived));
        }
        let total = select.clone().count(self.db().read()).await? as i64;
        let rows = select
            .order_by_desc(threads::Column::LastReplyAt)
            .order_by_desc(threads::Column::CreatedAt)
            .limit(limit)
            .offset(offset)
            .all(self.db().read())
            .await?;
        let items = self.build_thread_dto(rows).await?;
        Ok(ThreadPage {
            items,
            total,
            limit,
            offset,
        })
    }

    pub async fn list_my_threads(&self, caller: i64) -> Result<Vec<ThreadDto>, ChatError> {
        let memberships = thread_members::Entity::find()
            .filter(thread_members::Column::UserId.eq(caller))
            .order_by_desc(thread_members::Column::UpdatedAt)
            .limit(100)
            .all(self.db().read())
            .await?;
        let thread_ids: Vec<Uuid> = memberships.into_iter().map(|m| m.thread_id).collect();
        if thread_ids.is_empty() {
            return Ok(Vec::new());
        }
        let rows = threads::Entity::find()
            .filter(threads::Column::Id.is_in(thread_ids))
            .all(self.db().read())
            .await?;
        self.build_thread_dto(rows).await
    }

    pub async fn get_thread(&self, caller: i64, thread_id: Uuid) -> Result<ThreadDto, ChatError> {
        let thread = threads::Entity::find_by_id(thread_id)
            .one(self.db().read())
            .await?
            .ok_or(ChatError::NotFound)?;
        let _ctx = lookup::require_room_readable(self, thread.room_id, caller).await?;
        let items = self.build_thread_dto(vec![thread]).await?;
        Ok(items.into_iter().next().unwrap())
    }

    pub async fn update_thread(
        &self,
        caller: i64,
        thread_id: Uuid,
        cmd: UpdateThreadCmd,
    ) -> Result<ThreadDto, ChatError> {
        let thread = threads::Entity::find_by_id(thread_id)
            .one(self.db().read())
            .await?
            .ok_or(ChatError::NotFound)?;
        let ctx = lookup::require_room_readable(self, thread.room_id, caller).await?;

        let is_manager = match &ctx.space {
            Some(space) => authz::can_manage_space(self, space, caller).await?,
            None => false,
        };
        if thread.owner_id != caller && !is_manager {
            return Err(ChatError::Forbidden);
        }

        let now = OffsetDateTime::now_utc();
        let mut am: threads::ActiveModel = thread.clone().into();
        if let Some(name) = cmd.name {
            let name = name.trim().to_string();
            if name.chars().count() > self.config().chat.max_thread_name_len {
                return Err(ChatError::BadRequest("thread name too long".to_string()));
            }
            am.name = Set(if name.is_empty() { None } else { Some(name) });
        }
        if let Some(archived) = cmd.archived {
            am.archived = Set(archived);
        }
        if let Some(locked) = cmd.locked {
            am.locked = Set(locked);
        }
        am.updated_at = Set(now);
        am.update(self.db().write()).await?;

        let event = RoomEvent::ThreadUpdated {
            room_id: thread.room_id,
            thread_id,
        };
        self.realtime.publish_room(&event).await;

        let updated = threads::Entity::find_by_id(thread_id)
            .one(self.db().read())
            .await?
            .ok_or(ChatError::NotFound)?;
        let items = self.build_thread_dto(vec![updated]).await?;
        Ok(items.into_iter().next().unwrap())
    }

    pub async fn set_thread_membership(
        &self,
        caller: i64,
        thread_id: Uuid,
        cmd: ThreadMemberCmd,
    ) -> Result<(), ChatError> {
        let thread = threads::Entity::find_by_id(thread_id)
            .one(self.db().read())
            .await?
            .ok_or(ChatError::NotFound)?;
        let _ctx = lookup::require_room_readable(self, thread.room_id, caller).await?;

        let now = OffsetDateTime::now_utc();
        let notify_level = cmd.notify_level.unwrap_or(ThreadNotifyLevel::All);
        self.db()
            .primary()
            .transaction::<_, (), ChatError>(move |txn| {
                Box::pin(async move {
                    let member = thread_members::Entity::find_by_id((thread_id, caller))
                        .one(txn)
                        .await?;
                    match member {
                        Some(row) => {
                            let mut am: thread_members::ActiveModel = row.into();
                            am.notify_level = Set(notify_level);
                            am.updated_at = Set(now);
                            am.update(txn).await?;
                        }
                        None => {
                            thread_members::ActiveModel {
                                thread_id: Set(thread_id),
                                user_id: Set(caller),
                                notify_level: Set(notify_level),
                                last_read_message_id: Set(None),
                                last_read_seq: Set(0),
                                created_at: Set(now),
                                updated_at: Set(now),
                            }
                            .insert(txn)
                            .await?;
                            threads::Entity::update_many()
                                .filter(threads::Column::Id.eq(thread_id))
                                .col_expr(
                                    threads::Column::ParticipantCount,
                                    Expr::col(threads::Column::ParticipantCount).add(1),
                                )
                                .exec(txn)
                                .await?;
                        }
                    }
                    Ok(())
                })
            })
            .await?;
        Ok(())
    }

    pub async fn leave_thread(&self, caller: i64, thread_id: Uuid) -> Result<(), ChatError> {
        let thread = threads::Entity::find_by_id(thread_id)
            .one(self.db().read())
            .await?
            .ok_or(ChatError::NotFound)?;
        let _ctx = lookup::require_room_readable(self, thread.room_id, caller).await?;

        self.db()
            .primary()
            .transaction::<_, (), ChatError>(move |txn| {
                Box::pin(async move {
                    let deleted = thread_members::Entity::delete_by_id((thread_id, caller))
                        .exec(txn)
                        .await?;
                    if deleted.rows_affected > 0 {
                        threads::Entity::update_many()
                            .filter(threads::Column::Id.eq(thread_id))
                            .col_expr(
                                threads::Column::ParticipantCount,
                                Expr::col(threads::Column::ParticipantCount).sub(1),
                            )
                            .exec(txn)
                            .await?;
                    }
                    Ok(())
                })
            })
            .await?;
        Ok(())
    }

    pub(crate) async fn upsert_thread_read(
        &self,
        caller: i64,
        thread_id: Uuid,
        message_id: Uuid,
        seq: i64,
    ) -> Result<(), ChatError> {
        let now = OffsetDateTime::now_utc();
        let member = thread_members::Entity::find_by_id((thread_id, caller))
            .one(self.db().write())
            .await?;
        match member {
            Some(row) if row.last_read_seq < seq => {
                let mut am: thread_members::ActiveModel = row.into();
                am.last_read_message_id = Set(Some(message_id));
                am.last_read_seq = Set(seq);
                am.updated_at = Set(now);
                am.update(self.db().write()).await?;
            }
            Some(_) => {}
            None => {
                let am = thread_members::ActiveModel {
                    thread_id: Set(thread_id),
                    user_id: Set(caller),
                    notify_level: Set(ThreadNotifyLevel::All),
                    last_read_message_id: Set(Some(message_id)),
                    last_read_seq: Set(seq),
                    created_at: Set(now),
                    updated_at: Set(now),
                };
                match am.insert(self.db().write()).await {
                    Ok(_) => {
                        threads::Entity::update_many()
                            .filter(threads::Column::Id.eq(thread_id))
                            .col_expr(
                                threads::Column::ParticipantCount,
                                Expr::col(threads::Column::ParticipantCount).add(1),
                            )
                            .exec(self.db().write())
                            .await?;
                    }
                    Err(err) => {
                        let msg = err.to_string().to_lowercase();
                        if !(msg.contains("unique") || msg.contains("duplicate")) {
                            return Err(ChatError::from(err));
                        }
                    }
                }
            }
        }
        Ok(())
    }

    pub(crate) async fn room_member_count(&self, room_id: Uuid) -> Result<u64, ChatError> {
        Ok(room_members::Entity::find()
            .filter(room_members::Column::RoomId.eq(room_id))
            .filter(room_members::Column::LeftAt.is_null())
            .count(self.db().read())
            .await?)
    }
}