use export::pagination::Page;
use export::users::UserDto;
use git::facade::GitContext;
use models::issue::issue_relation_pull;
use models::pull_request::{
    pull_request_assignee, pull_request_comment, pull_request_event_line, pull_request_label,
    pull_request_review, pull_request_reviewer, pull_requests,
};
use sea_orm::ActiveValue::Set;
use sea_orm::{
    ActiveModelTrait, ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder,
    QuerySelect, TransactionTrait,
};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use time::OffsetDateTime;
use utoipa::{IntoParams, ToSchema};
use uuid::Uuid;

use crate::errors::PullError;
use crate::services::events::{
    EVENT_CLOSED, EVENT_CREATED, EVENT_MERGED, EVENT_REOPENED, EVENT_UPDATED, write_event,
};
use crate::services::service::PullService;

pub const MAX_TITLE_LEN: usize = 512;
pub const MAX_BODY_LEN: usize = 65_536;

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
#[serde(rename_all = "snake_case")]
pub enum PullState {
    Open,
    Closed,
    Merged,
}

#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct PullRefDto {
    pub label: String,
    pub branch: String,
    pub sha: String,
    pub repo_id: Option<Uuid>,
}

#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct PullDto {
    pub id: Uuid,
    pub repo_id: Uuid,
    pub number: i32,
    pub title: String,
    pub body: String,
    pub state: PullState,
    pub is_draft: bool,
    pub auto_merge: bool,
    pub locked: bool,
    pub head: PullRefDto,
    pub base: PullRefDto,
    pub author: Option<UserDto>,
    pub merger: Option<UserDto>,
    pub merge_method: models::pull_request::pull_requests::MergeMethod,
    pub additions: i32,
    pub deletions: i32,
    pub files_changed: i32,
    pub merged_commit_sha: Option<String>,
    pub squash_commit_sha: Option<String>,
    pub issue_id: Option<Uuid>,
    pub created_at: OffsetDateTime,
    pub updated_at: Option<OffsetDateTime>,
    pub closed_at: Option<OffsetDateTime>,
    pub merged_at: Option<OffsetDateTime>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub mergeable: Option<bool>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub mergeable_state: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub conflicts: Option<Vec<String>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub commits_ahead: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub commits_behind: Option<i64>,
}

pub type PullPage = Page<PullDto>;

#[derive(Debug, Clone, Default, Deserialize, IntoParams)]
pub struct ListPullQuery {
    /// open | closed | merged | all
    #[serde(default)]
    pub state: Option<String>,
    #[serde(default)]
    pub author: Option<String>,
    #[serde(default)]
    pub base: Option<String>,
    #[serde(default)]
    pub head: Option<String>,
    #[serde(default)]
    pub draft: Option<bool>,
    #[serde(default)]
    pub search: Option<String>,
    /// number | created | updated
    #[serde(default)]
    pub sort: Option<String>,
    /// asc | desc
    #[serde(default)]
    pub order: Option<String>,
    #[serde(default)]
    pub limit: Option<u64>,
    #[serde(default)]
    pub offset: Option<u64>,
}

#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct CreatePullCmd {
    pub title: String,
    #[serde(default)]
    pub body: Option<String>,
    pub head_branch: String,
    #[serde(default)]
    pub base_branch: Option<String>,
    #[serde(default)]
    pub draft: Option<bool>,
    #[serde(default)]
    pub issue_id: Option<Uuid>,
}

#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct UpdatePullCmd {
    #[serde(default)]
    pub title: Option<String>,
    #[serde(default, deserialize_with = "export::serde_patch::deserialize_opt_opt")]
    pub body: Option<Option<String>>,
    #[serde(default)]
    pub base_branch: Option<String>,
    #[serde(default)]
    pub draft: Option<bool>,
    #[serde(default, deserialize_with = "export::serde_patch::deserialize_opt_opt")]
    pub auto_merge: Option<Option<bool>>,
    #[serde(default, deserialize_with = "export::serde_patch::deserialize_opt_opt")]
    pub issue_id: Option<Option<Uuid>>,
}

impl PullService {
    pub(crate) fn dto(pull: &pull_requests::Model, users: &HashMap<i64, UserDto>) -> PullDto {
        PullDto {
            id: pull.id,
            repo_id: pull.repo_id,
            number: pull.autoincrement_id,
            title: pull.title.clone(),
            body: pull.body.clone(),
            state: {
                if pull.has_merged {
                    PullState::Merged
                } else if pull.closed_at.is_some() {
                    PullState::Closed
                } else {
                    PullState::Open
                }
            },
            is_draft: pull.is_draft,
            auto_merge: pull.auto_merge,
            locked: pull.is_locked,
            head: PullRefDto {
                label: pull
                    .head_label
                    .clone()
                    .unwrap_or_else(|| pull.head_branch.clone()),
                branch: pull.head_branch.clone(),
                sha: pull.head_sha.clone(),
                repo_id: pull.head_repo_id,
            },
            base: PullRefDto {
                label: pull.base_branch.clone(),
                branch: pull.base_branch.clone(),
                sha: pull.base_sha.clone(),
                repo_id: Some(pull.repo_id),
            },
            author: users.get(&pull.author_id).cloned(),
            merger: pull.merger_id.and_then(|id| users.get(&id).cloned()),
            merge_method: pull.merge_method,
            additions: pull.additions,
            deletions: pull.deletions,
            files_changed: pull.files_changed,
            merged_commit_sha: pull.merged_commit_sha.clone(),
            squash_commit_sha: pull.squash_commit_sha.clone(),
            issue_id: pull.issue_id,
            created_at: pull.created_at,
            updated_at: pull.updated_at,
            closed_at: pull.closed_at,
            merged_at: pull.merged_at,
            mergeable: None,
            mergeable_state: None,
            conflicts: None,
            commits_ahead: None,
            commits_behind: None,
        }
    }

    pub async fn list_pulls(&self, repo_id: Uuid, q: ListPullQuery) -> Result<PullPage, PullError> {
        let limit = q.limit.unwrap_or(50).clamp(1, 100);
        let offset = q.offset.unwrap_or(0);
        let state = q.state.clone().unwrap_or_else(|| "open".to_string());
        let mut select =
            pull_requests::Entity::find().filter(pull_requests::Column::RepoId.eq(repo_id));

        match state.as_str() {
            "all" => {}
            "open" => {
                select = select
                    .filter(pull_requests::Column::HasMerged.eq(false))
                    .filter(pull_requests::Column::ClosedAt.is_null());
            }
            "closed" => {
                select = select
                    .filter(pull_requests::Column::HasMerged.eq(false))
                    .filter(pull_requests::Column::ClosedAt.is_not_null());
            }
            "merged" => select = select.filter(pull_requests::Column::HasMerged.eq(true)),
            other => {
                return Err(PullError::BadRequest(format!("unknown state `{other}`")));
            }
        }
        if let Some(author) = q.author.as_deref().filter(|a| !a.trim().is_empty()) {
            let uid = self.actors.resolve(author).await?;
            select = select.filter(pull_requests::Column::AuthorId.eq(uid));
        }
        if let Some(base) = q.base.as_deref().filter(|b| !b.trim().is_empty()) {
            select = select.filter(pull_requests::Column::BaseBranch.eq(base));
        }
        if let Some(head) = q.head.as_deref().filter(|h| !h.trim().is_empty()) {
            select = select.filter(pull_requests::Column::HeadBranch.eq(head));
        }
        if let Some(draft) = q.draft {
            select = select.filter(pull_requests::Column::IsDraft.eq(draft));
        }
        if let Some(search) = q.search.as_deref().filter(|s| !s.trim().is_empty()) {
            select = select.filter(pull_requests::Column::Title.contains(search));
        }

        let total = select.clone().count(self.db_read()).await? as i64;
        let descending = q.order.as_deref().unwrap_or("desc") != "asc";
        select = match q.sort.as_deref().unwrap_or("created") {
            "number" => {
                if descending {
                    select.order_by_desc(pull_requests::Column::AutoincrementId)
                } else {
                    select.order_by_asc(pull_requests::Column::AutoincrementId)
                }
            }
            "updated" => {
                if descending {
                    select.order_by_desc(pull_requests::Column::UpdatedAt)
                } else {
                    select.order_by_asc(pull_requests::Column::UpdatedAt)
                }
            }
            _ => {
                if descending {
                    select.order_by_desc(pull_requests::Column::CreatedAt)
                } else {
                    select.order_by_asc(pull_requests::Column::CreatedAt)
                }
            }
        };
        let rows = select
            .limit(limit)
            .offset(offset)
            .all(self.db_read())
            .await?;
        let users = self.pull_users(&rows).await?;
        let items = rows.iter().map(|row| Self::dto(row, &users)).collect();
        Ok(Page {
            items,
            total,
            limit,
            offset,
        })
    }

    pub async fn create_pull(
        &self,
        space_name: String,
        repo_name: String,
        ctx: &GitContext,
        caller: i64,
        cmd: CreatePullCmd,
    ) -> Result<pull_requests::Model, PullError> {
        if !ctx.model.allow_pull_requests {
            return Err(PullError::BadRequest(
                "pull requests are disabled for this repository".to_string(),
            ));
        }
        let title = cmd.title.trim();
        if title.is_empty() || title.chars().count() > MAX_TITLE_LEN {
            return Err(PullError::BadRequest(format!(
                "title must be 1-{MAX_TITLE_LEN} characters"
            )));
        }
        let body = cmd.body.clone().unwrap_or_default();
        if body.chars().count() > MAX_BODY_LEN {
            return Err(PullError::BadRequest("body too large".to_string()));
        }
        self.actors.allow_pull(caller).await?;

        let base_branch = cmd
            .base_branch
            .clone()
            .unwrap_or_else(|| ctx.model.default_branch.clone());
        if base_branch == cmd.head_branch {
            return Err(PullError::BadRequest(
                "head and base branch must differ".to_string(),
            ));
        }
        if let Some(issue_id) = cmd.issue_id
            && !models::issue::issue_item::Entity::find_by_id(issue_id)
                .one(self.db_read())
                .await?
                .is_some()
        {
            return Err(PullError::BadRequest("issue not found".to_string()));
        }

        let plan = self
            .git
            .merge_plan(
                ctx,
                &format!("refs/heads/{base_branch}"),
                &format!("refs/heads/{}", cmd.head_branch),
            )
            .await
            .map_err(|e| match e {
                git::facade::GitError::NotFound => PullError::BadRequest(
                    "head or base branch not found in this repository".to_string(),
                ),
                other => PullError::from(other),
            })?;
        if plan.head_only == 0 {
            return Err(PullError::Conflict(
                "the head branch has no commits ahead of the base branch".to_string(),
            ));
        }
        let stats = self
            .diff_stats_for(&space_name, &repo_name, &plan, &plan.head_sha)
            .await?;

        let txn = self.database.write().begin().await?;
        self.git.lock_repository(&txn, ctx.model.id).await?;
        let last = pull_requests::Entity::find()
            .filter(pull_requests::Column::RepoId.eq(ctx.model.id))
            .order_by_desc(pull_requests::Column::AutoincrementId)
            .one(&txn)
            .await?;
        let number = last.map(|row| row.autoincrement_id + 1).unwrap_or(1);
        let now = OffsetDateTime::now_utc();
        let created = pull_requests::ActiveModel {
            id: Set(Uuid::now_v7()),
            repo_id: Set(ctx.model.id),
            issue_id: Set(cmd.issue_id),
            autoincrement_id: Set(number),
            author_id: Set(caller),
            title: Set(title.to_string()),
            body: Set(body),
            head_branch: Set(cmd.head_branch.clone()),
            head_sha: Set(plan.head_sha.clone()),
            head_repo_id: Set(None),
            head_label: Set(Some(cmd.head_branch.to_string())),
            base_branch: Set(base_branch.clone()),
            base_sha: Set(plan.base_sha.clone()),
            is_draft: Set(cmd.draft.unwrap_or(false)),
            merge_method: Set(models::pull_request::pull_requests::MergeMethod::Merge),
            has_merged: Set(false),
            merged_commit_sha: Set(None),
            squash_commit_sha: Set(None),
            auto_merge: Set(false),
            additions: Set(stats.0),
            deletions: Set(stats.1),
            files_changed: Set(stats.2),
            merger_id: Set(None),
            merged_at: Set(None),
            is_locked: Set(false),
            created_at: Set(now),
            updated_at: Set(None),
            closed_at: Set(None),
        };
        let pull = created.insert(&txn).await?;
        write_event(
            &txn,
            &pull,
            Some(caller),
            EVENT_CREATED,
            Some(plan.head_sha.clone()),
            None,
        )
        .await?;
        txn.commit().await?;
        self.recount_open(ctx.model.id).await?;
        Ok(pull)
    }

    /// `handle` carries the resolved repository, pull row, caller and admin bit
    /// so the signature does not repeat four positionally-similar arguments.
    pub async fn update_pull(
        &self,
        space_name: String,
        repo_name: String,
        handle: &crate::share::lookup::PullHandle,
        cmd: UpdatePullCmd,
    ) -> Result<pull_requests::Model, PullError> {
        let crate::share::lookup::PullHandle {
            ctx, pull, admin, ..
        } = handle;
        let caller = handle.caller.ok_or(PullError::Unauthorized)?;
        let admin = *admin;
        if pull.is_locked && !admin {
            return Err(PullError::Forbidden);
        }
        let mut model: pull_requests::ActiveModel = pull.clone().into();
        let mut changed = false;
        if let Some(title) = cmd.title.as_deref() {
            let title = title.trim();
            if title.is_empty() || title.chars().count() > MAX_TITLE_LEN {
                return Err(PullError::BadRequest("invalid title".to_string()));
            }
            model.title = Set(title.to_string());
            changed = true;
        }
        if let Some(body) = cmd.body {
            if body
                .as_deref()
                .is_some_and(|b| b.chars().count() > MAX_BODY_LEN)
            {
                return Err(PullError::BadRequest("body too large".to_string()));
            }
            model.body = Set(body.unwrap_or_default());
            changed = true;
        }
        if let Some(draft) = cmd.draft {
            model.is_draft = Set(draft);
            changed = true;
        }
        if let Some(auto) = cmd.auto_merge {
            if auto.is_some_and(|a| a) {
                crate::share::authz::require_open(pull)?;
            }
            model.auto_merge = Set(auto.unwrap_or(false));
            changed = true;
        }
        if let Some(issue_id) = cmd.issue_id {
            if let Some(id) = issue_id
                && !models::issue::issue_item::Entity::find_by_id(id)
                    .one(self.db_read())
                    .await?
                    .is_some()
            {
                return Err(PullError::BadRequest("issue not found".to_string()));
            }
            model.issue_id = Set(issue_id);
            changed = true;
        }
        if let Some(base_branch) = cmd.base_branch.as_deref() {
            if base_branch == pull.head_branch {
                return Err(PullError::BadRequest(
                    "base branch must differ from head branch".to_string(),
                ));
            }
            let plan = self
                .git
                .merge_plan(ctx, &format!("refs/heads/{base_branch}"), &pull.head_sha)
                .await
                .map_err(|_| {
                    PullError::BadRequest(format!("base branch `{base_branch}` not found"))
                })?;
            model.base_branch = Set(base_branch.to_string());
            model.base_sha = Set(plan.base_sha);
            changed = true;
        }
        if !changed {
            return Ok(pull.clone());
        }
        model.updated_at = Set(Some(OffsetDateTime::now_utc()));
        let this = self.clone();
        let updated = self
            .db_write()
            .transaction::<_, _, PullError>(move |tx| {
                Box::pin(async move {
                    let updated = model.update(tx).await?;
                    this.record_event_with(tx, &updated, Some(caller), EVENT_UPDATED, None, None)
                        .await?;
                    Ok(updated)
                })
            })
            .await?;
        let _ = (space_name, repo_name);
        Ok(updated)
    }

    pub async fn set_state(
        &self,
        pull: &pull_requests::Model,
        caller: i64,
        closed: bool,
    ) -> Result<pull_requests::Model, PullError> {
        if pull.has_merged {
            return Err(PullError::Conflict(
                "a merged pull request cannot change state".to_string(),
            ));
        }
        if closed && pull.closed_at.is_some() {
            return Ok(pull.clone());
        }
        if !closed && pull.closed_at.is_none() {
            return Ok(pull.clone());
        }
        let mut model: pull_requests::ActiveModel = pull.clone().into();
        model.closed_at = Set(if closed {
            Some(OffsetDateTime::now_utc())
        } else {
            None
        });
        model.updated_at = Set(Some(OffsetDateTime::now_utc()));
        let this = self.clone();
        let updated = self
            .db_write()
            .transaction::<_, _, PullError>(move |tx| {
                Box::pin(async move {
                    let updated = model.update(tx).await?;
                    this.record_event_with(
                        tx,
                        &updated,
                        Some(caller),
                        if closed { EVENT_CLOSED } else { EVENT_REOPENED },
                        None,
                        None,
                    )
                    .await?;
                    Ok(updated)
                })
            })
            .await?;
        self.recount_open(updated.repo_id).await?;
        Ok(updated)
    }

    pub async fn delete_pull(
        &self,
        pull: &pull_requests::Model,
        caller: i64,
    ) -> Result<(), PullError> {
        let _ = caller;
        let id = pull.id;
        let txn = self.database.write().begin().await?;
        pull_request_comment::Entity::delete_many()
            .filter(pull_request_comment::Column::PullRequestId.eq(id))
            .exec(&txn)
            .await?;
        pull_request_review::Entity::delete_many()
            .filter(pull_request_review::Column::PullRequestId.eq(id))
            .exec(&txn)
            .await?;
        pull_request_reviewer::Entity::delete_many()
            .filter(pull_request_reviewer::Column::PullRequestId.eq(id))
            .exec(&txn)
            .await?;
        pull_request_label::Entity::delete_many()
            .filter(pull_request_label::Column::PullRequestId.eq(id))
            .exec(&txn)
            .await?;
        pull_request_assignee::Entity::delete_many()
            .filter(pull_request_assignee::Column::PullRequestId.eq(id))
            .exec(&txn)
            .await?;
        pull_request_event_line::Entity::delete_many()
            .filter(pull_request_event_line::Column::PullRequestId.eq(id))
            .exec(&txn)
            .await?;
        issue_relation_pull::Entity::delete_many()
            .filter(issue_relation_pull::Column::PullRequestId.eq(id))
            .exec(&txn)
            .await?;
        pull_requests::Entity::delete_many()
            .filter(pull_requests::Column::Id.eq(id))
            .exec(&txn)
            .await?;
        txn.commit().await?;
        self.recount_open(pull.repo_id).await?;
        Ok(())
    }

    pub(crate) async fn pull_users(
        &self,
        rows: &[pull_requests::Model],
    ) -> Result<HashMap<i64, UserDto>, PullError> {
        let mut ids: Vec<i64> = rows
            .iter()
            .flat_map(|row| [Some(row.author_id), row.merger_id])
            .flatten()
            .collect();
        ids.sort();
        ids.dedup();
        if ids.is_empty() {
            return Ok(HashMap::new());
        }
        self.actors.briefs(&ids).await
    }

    /// (additions, deletions, files_changed)
    pub(crate) async fn diff_stats_for(
        &self,
        space_name: &str,
        repo_name: &str,
        plan: &git::facade::MergePlanDto,
        head: &str,
    ) -> Result<(i32, i32, i32), PullError> {
        let from = plan
            .merge_base
            .clone()
            .unwrap_or_else(|| plan.base_sha.clone());
        let repo_ctx = self
            .git
            .context(space_name.to_string(), repo_name.to_string())
            .await?;
        let stat = self
            .git
            .diff_short_stat(&repo_ctx, from, head.to_string())
            .await?;
        Ok((
            stat.additions as i32,
            stat.deletions as i32,
            stat.files_changed as i32,
        ))
    }

    pub async fn apply_merge_outcome(
        &self,
        pull: &pull_requests::Model,
        caller: i64,
        merged_sha: String,
        squash_sha: Option<String>,
        method: models::pull_request::pull_requests::MergeMethod,
    ) -> Result<pull_requests::Model, PullError> {
        let mut model: pull_requests::ActiveModel = pull.clone().into();
        model.has_merged = Set(true);
        model.merger_id = Set(Some(caller));
        model.merged_at = Set(Some(OffsetDateTime::now_utc()));
        model.closed_at = Set(Some(OffsetDateTime::now_utc()));
        model.merge_method = Set(method);
        model.squash_commit_sha = Set(squash_sha);
        model.merged_commit_sha = Set(Some(merged_sha.clone()));
        model.auto_merge = Set(false);
        model.updated_at = Set(Some(OffsetDateTime::now_utc()));
        let this = self.clone();
        let updated = self
            .db_write()
            .transaction::<_, _, PullError>(move |tx| {
                Box::pin(async move {
                    let updated = model.update(tx).await?;
                    this.record_event_with(
                        tx,
                        &updated,
                        Some(caller),
                        EVENT_MERGED,
                        Some(merged_sha),
                        None,
                    )
                    .await?;
                    Ok(updated)
                })
            })
            .await?;
        self.recount_open(updated.repo_id).await?;
        Ok(updated)
    }
}

impl PullService {
    pub async fn set_lock(
        &self,
        pull: &pull_requests::Model,
        caller: i64,
        locked: bool,
    ) -> Result<pull_requests::Model, PullError> {
        if pull.is_locked == locked {
            return Ok(pull.clone());
        }
        let mut model: pull_requests::ActiveModel = pull.clone().into();
        model.is_locked = Set(locked);
        model.updated_at = Set(Some(OffsetDateTime::now_utc()));
        let this = self.clone();
        let updated = self
            .db_write()
            .transaction::<_, _, PullError>(move |tx| {
                Box::pin(async move {
                    let updated = model.update(tx).await?;
                    this.record_event_with(
                        tx,
                        &updated,
                        Some(caller),
                        if locked {
                            crate::services::events::EVENT_LOCKED
                        } else {
                            crate::services::events::EVENT_UNLOCKED
                        },
                        None,
                        None,
                    )
                    .await?;
                    Ok(updated)
                })
            })
            .await?;
        Ok(updated)
    }
}

#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct LinkIssueCmd {
    pub space_name: String,
    pub repo_name: String,
    /// PR 编号或 UUID。
    pub pull_ref: String,
}