use export::pagination::{Page, PageQuery};
use models::pull_request::{pull_request_event_line, pull_requests};
use models::users::user;
use sea_orm::{
    ActiveModelTrait, ActiveValue::Set, ColumnTrait, ConnectionTrait, EntityTrait, PaginatorTrait,
    QueryFilter, QueryOrder, QuerySelect,
};
use serde::{Deserialize, Serialize};
use space::services::spaces::UserDto;
use std::collections::HashMap;
use time::OffsetDateTime;
use uuid::Uuid;

use crate::errors::PullError;
use crate::services::service::PullService;

pub const EVENT_CREATED: &str = "created";
pub const EVENT_UPDATED: &str = "updated";
pub const EVENT_CLOSED: &str = "closed";
pub const EVENT_REOPENED: &str = "reopened";
pub const EVENT_MERGED: &str = "merged";
pub const EVENT_COMMENTED: &str = "commented";
pub const EVENT_REVIEW_SUBMITTED: &str = "review_submitted";
pub const EVENT_REVIEW_DISMISSED: &str = "review_dismissed";
pub const EVENT_REVIEW_REQUESTED: &str = "review_requested";
pub const EVENT_REVIEW_REQUEST_CANCELLED: &str = "review_request_cancelled";
pub const EVENT_HEAD_UPDATED: &str = "head_updated";
pub const EVENT_LABELED: &str = "labeled";
pub const EVENT_UNLABELED: &str = "unlabeled";
pub const EVENT_ASSIGNED: &str = "assigned";
pub const EVENT_UNASSIGNED: &str = "unassigned";
pub const EVENT_LOCKED: &str = "locked";
pub const EVENT_UNLOCKED: &str = "unlocked";
pub const EVENT_AUTO_MERGE_ENABLED: &str = "auto_merge_enabled";
pub const EVENT_AUTO_MERGE_DISABLED: &str = "auto_merge_disabled";

#[derive(Debug, Clone, Serialize, Deserialize, utoipa::ToSchema)]
pub struct PullEventDto {
    pub id: Uuid,
    pub pull_request_id: Uuid,
    pub event_type: String,
    pub actor: Option<UserDto>,
    pub commit_sha: Option<String>,
    pub content: Option<String>,
    pub created_at: OffsetDateTime,
}

pub type EventPage = Page<PullEventDto>;

pub async fn write_event(
    db: &impl ConnectionTrait,
    pull: &pull_requests::Model,
    actor: Option<Uuid>,
    event_type: &str,
    commit_sha: Option<String>,
    content: Option<String>,
) -> Result<(), sea_orm::DbErr> {
    let actor = match actor {
        Some(actor) => actor,
        None => pull.author_id,
    };
    let row = pull_request_event_line::ActiveModel {
        id: Set(Uuid::now_v7()),
        pull_request_id: Set(pull.id),
        user_id: Set(actor),
        event_type: Set(event_type.to_string()),
        commit_sha: Set(commit_sha),
        content: Set(content),
        created_at: Set(OffsetDateTime::now_utc()),
        ..Default::default()
    };
    row.insert(db).await?;
    Ok(())
}

impl PullService {
    pub async fn record_event(
        &self,
        pull: &pull_requests::Model,
        actor: Option<Uuid>,
        event_type: &str,
        commit_sha: Option<String>,
        content: Option<String>,
    ) -> Result<(), PullError> {
        write_event(
            self.db_write(),
            pull,
            actor,
            event_type,
            commit_sha,
            content,
        )
        .await?;
        self.publish_ci_event(pull, actor, event_type).await;
        self.publish_ai_event(pull, actor, event_type).await;
        Ok(())
    }

    /// Publishes the CI mirror for callers that already wrote their own row
    /// (for example inside a transaction).
    pub(crate) async fn publish_ci_event_for(
        &self,
        pull: &pull_requests::Model,
        actor: Option<Uuid>,
        event_type: &str,
    ) {
        self.publish_ci_event(pull, actor, event_type).await;
        self.publish_ai_event(pull, actor, event_type).await;
    }

    /// Mirrors CI-relevant pull request activity onto the `gitdata:events`
    /// queue so the scheduler can evaluate `pull_request` workflow triggers.
    async fn publish_ci_event(
        &self,
        pull: &pull_requests::Model,
        actor: Option<Uuid>,
        event_type: &str,
    ) {
        let action = match event_type {
            EVENT_CREATED => ci_protocol::events::PullRequestAction::Opened,
            EVENT_HEAD_UPDATED => ci_protocol::events::PullRequestAction::Synchronize,
            EVENT_REOPENED => ci_protocol::events::PullRequestAction::Reopened,
            EVENT_MERGED => ci_protocol::events::PullRequestAction::Merged,
            EVENT_CLOSED => ci_protocol::events::PullRequestAction::Closed,
            _ => return,
        };
        if !self.config().ci.enabled {
            return;
        }
        let Some(space_id) = self.pull_space_id(pull.repo_id).await else {
            return;
        };
        let event =
            ci_protocol::events::CiEvent::PullRequest(ci_protocol::events::PullRequestEvent {
                id: Uuid::now_v7(),
                repo_id: pull.repo_id,
                space_id,
                actor_id: actor.or(Some(pull.author_id)),
                number: pull.autoincrement_id as i64,
                action,
                head_sha: pull.head_sha.clone(),
                head_ref: pull.head_branch.clone(),
                base_sha: pull.base_sha.clone(),
                base_ref: pull.base_branch.clone(),
            });
        let payload = match serde_json::to_vec(&event) {
            Ok(payload) => payload,
            Err(error) => {
                tracing::warn!(%error, "ci pull request event serialize failed");
                return;
            }
        };
        let queue = git::sync::queue::SyncQueue::new(self.state.redis.clone());
        if let Err(error) = queue
            .publish_raw(&self.config().ci.event_queue_topic, None, &payload)
            .await
        {
            tracing::warn!(%error, "ci pull request event publish failed");
        }
    }

    /// 打开/更新/重开 PR 时投递 AI 评审任务。
    ///
    /// 独立于 `ci.enabled`：只看 `[ai]` 总开关与评审子开关，且投递失败不影响
    /// 事件写入与接口返回。
    async fn publish_ai_event(
        &self,
        pull: &pull_requests::Model,
        actor: Option<Uuid>,
        event_type: &str,
    ) {
        let cfg = &self.config().ai;
        if !cfg.review_on() {
            return;
        }
        let trigger = matches!(
            event_type,
            EVENT_CREATED | EVENT_HEAD_UPDATED | EVENT_REOPENED
        );
        if !trigger {
            return;
        }
        let task = ci_protocol::ai::AgentTask::PullReview {
            job_id: Uuid::now_v7(),
            repo_id: pull.repo_id,
            pull_request_id: pull.id,
            head_sha: pull.head_sha.clone(),
            // 评审语言跟随触发者（推送/开 PR 的人）。
            language: git::sync::actor_language(self.db_read(), actor).await,
        };
        git::sync::ai_task::submit_ai_task_logged(self.db_write(), &self.state.redis, cfg, &task)
            .await;
    }

    async fn pull_space_id(&self, repo_id: Uuid) -> Option<Uuid> {
        let repo = models::repo::repos::Entity::find_by_id(repo_id)
            .one(self.db_read())
            .await
            .ok()??;
        git::share::access::space_for_dir(self.db_read(), repo.dir_id)
            .await
            .ok()?
            .map(|space| space.id)
    }

    pub async fn list_events(
        &self,
        pull_id: Uuid,
        page: PageQuery,
    ) -> Result<EventPage, PullError> {
        let (limit, offset) = page.page();
        let total = pull_request_event_line::Entity::find()
            .filter(pull_request_event_line::Column::PullRequestId.eq(pull_id))
            .count(self.db_read())
            .await? as i64;
        let rows = pull_request_event_line::Entity::find()
            .filter(pull_request_event_line::Column::PullRequestId.eq(pull_id))
            .order_by_asc(pull_request_event_line::Column::CreatedAt)
            .limit(limit)
            .offset(offset)
            .all(self.db_read())
            .await?;
        let actors = self.user_briefs(&unique_users(&rows)).await?;
        let items = rows
            .into_iter()
            .map(|row| PullEventDto {
                id: row.id,
                pull_request_id: row.pull_request_id,
                event_type: row.event_type,
                commit_sha: row.commit_sha,
                content: row.content,
                created_at: row.created_at,
                actor: actors.get(&row.user_id).cloned(),
            })
            .collect();
        Ok(Page {
            items,
            total,
            limit,
            offset,
        })
    }

    pub(crate) async fn user_briefs(
        &self,
        ids: &[Uuid],
    ) -> Result<HashMap<Uuid, UserDto>, PullError> {
        if ids.is_empty() {
            return Ok(HashMap::new());
        }
        let rows = user::Entity::find()
            .filter(user::Column::Uid.is_in(ids.iter().copied()))
            .all(self.db_read())
            .await?;
        Ok(rows
            .into_iter()
            .map(|row| (row.uid, UserDto::from(row)))
            .collect())
    }
}

fn unique_users(rows: &[pull_request_event_line::Model]) -> Vec<Uuid> {
    let mut ids: Vec<Uuid> = rows.iter().map(|row| row.user_id).collect();
    ids.sort();
    ids.dedup();
    ids
}