use config::ai::AiConfig;
use export::{AppDatabase, AppRedis};
use models::ai::ai_job;
use models::ai::ai_job::{ActiveModel, AiJobStatus};
use models::git::git_tags;
use models::repo::{repo_release, repos};
use sea_orm::PaginatorTrait;
use sea_orm::{ActiveModelTrait, ColumnTrait, DatabaseConnection, EntityTrait, QueryFilter, Set};
use time::OffsetDateTime;
use uuid::Uuid;

use crate::sync::error::{SyncError, SyncResult};
use crate::sync::queue::SyncQueue;

/// 一次推送要不要为各能力投递任务：平台开关与仓库开关的交集。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct PushAiSwitches {
    review: bool,
    summary: bool,
    release_notes: bool,
}

impl PushAiSwitches {
    fn any(self) -> bool {
        self.review || self.summary || self.release_notes
    }
}

/// 三层准入的合并规则，单独成函数是为了能直接测：平台 `[ai]` 总开关 → 子能力
/// 开关 → 仓库列。任何一层关掉都不投递，因此「仓库开了但平台没开」不会产生调用。
fn push_ai_switches(cfg: &AiConfig, repo_review: bool, repo_summary: bool) -> PushAiSwitches {
    PushAiSwitches {
        review: cfg.review_on() && repo_review,
        summary: cfg.summary_on() && repo_summary,
        // release 文案没有仓库级开关：它跟着 tag 走，粒度由平台配置掌握。
        release_notes: cfg.release_notes_on(),
    }
}

pub async fn submit_ai_task(
    db: &DatabaseConnection,
    redis: &AppRedis,
    cfg: &AiConfig,
    task: &ci_protocol::ai::AgentTask,
) -> SyncResult<Uuid> {
    let job_id = task.job_id();
    let now = OffsetDateTime::now_utc();
    // 完整载荷随作业持久化：Redis 列表没有投递确认，管理端重试时必须
    // 能够原样重投，而不是只把数据库状态改回 pending。
    let payload = task
        .encode()
        .map_err(|e| SyncError::Internal(format!("encode ai task: {e}")))?;
    let request = serde_json::to_value(task)
        .map_err(|e| SyncError::Internal(format!("encode ai task: {e}")))?;
    ActiveModel {
        id: Set(job_id),
        repo_id: Set(task.repo_id()),
        kind: Set(kind_of(task)),
        target_id: Set(task.target_id()),
        status: Set(AiJobStatus::Pending),
        attempts: Set(0),
        model: Set(cfg.model.trim().to_string()),
        target_ref: Set(task.target_ref()),
        language: Set(task.language().unwrap_or_default().to_string()),
        request_json: Set(Some(request)),
        response_json: Set(None),
        prompt_tokens: Set(None),
        completion_tokens: Set(None),
        error: Set(None),
        created_at: Set(now),
        started_at: Set(None),
        finished_at: Set(None),
        updated_at: Set(None),
    }
    .insert(db)
    .await?;

    let queue = SyncQueue::new(redis.clone());
    if let Err(error) = queue.publish_raw(&cfg.queue_topic, None, &payload).await {
        mark_submit_failure(db, job_id, &error.to_string()).await;
        return Err(error);
    }
    tracing::debug!(job_id = %job_id, kind = task.kind_str(), "ai task submitted");
    Ok(job_id)
}

pub async fn submit_ai_task_logged(
    db: &DatabaseConnection,
    redis: &AppRedis,
    cfg: &AiConfig,
    task: &ci_protocol::ai::AgentTask,
) {
    match submit_ai_task(db, redis, cfg, task).await {
        Ok(job_id) => tracing::debug!(%job_id, kind = task.kind_str(), "ai task submitted"),
        Err(error) => {
            tracing::warn!(%error, kind = task.kind_str(), "ai task submission failed")
        }
    }
}

pub async fn trigger_push_ai_tasks(
    db: &AppDatabase,
    redis: &AppRedis,
    cfg: &AiConfig,
    repo_id: Uuid,
    actor_id: Option<Uuid>,
    refs: &[(String, String, String)],
) {
    let conn = db.write();
    // 仓库级开关是这一层的准入条件：平台开着、仓库不想用，就不该产生调用。
    // 读一次仓库行，两个开关都从它取，避免每条引用各查一次。
    let (review_on_push, summary_on_push) = match repos::Entity::find_by_id(repo_id).one(conn).await
    {
        Ok(Some(row)) => (row.ai_review_on_push, row.ai_summary_on_push),
        // 仓库行读不到（极少见：推送与同步竞态）时保持加列之前的行为：
        // 评审照旧按平台开关走，新增的自动总结不触发。丢掉一次自动评审比
        // 多发一次模型调用更难排查，所以这里偏向沿用旧行为。
        Ok(None) | Err(_) => (true, false),
    };
    let switches = push_ai_switches(cfg, review_on_push, summary_on_push);
    if !switches.any() {
        return;
    }
    // 一次推送产生的所有任务共用触发者语言。
    let language = crate::sync::ai_language::actor_language(conn, actor_id).await;
    if switches.review {
        for (name, _old, new) in refs {
            if !name.starts_with("refs/heads/") || is_zero_oid(new) {
                continue;
            }
            let task = ci_protocol::ai::AgentTask::CommitReview {
                job_id: Uuid::now_v7(),
                repo_id,
                commit_sha: new.clone(),
                language: language.clone(),
            };
            submit_ai_task_logged(conn, redis, cfg, &task).await;
        }
    }
    if switches.summary {
        for (name, _old, new) in refs {
            if !name.starts_with("refs/heads/") || is_zero_oid(new) {
                continue;
            }
            let task = ci_protocol::ai::AgentTask::RepoSummary {
                job_id: Uuid::now_v7(),
                repo_id,
                head_sha: new.clone(),
                base_sha: None,
                limit: None,
                language: language.clone(),
            };
            submit_ai_task_logged(conn, redis, cfg, &task).await;
        }
    }
    if switches.release_notes {
        for (name, _old, new) in refs {
            let Some(tag) = name.strip_prefix("refs/tags/") else {
                continue;
            };
            if is_zero_oid(new) {
                continue;
            }
            match ensure_release_draft(
                conn,
                redis,
                cfg,
                repo_id,
                actor_id,
                language.clone(),
                tag,
                new,
            )
            .await
            {
                Ok(Some(job_id)) => {
                    tracing::debug!(%job_id, tag, "release draft queued");
                }
                Ok(None) => {}
                Err(error) => tracing::warn!(%error, tag, "release draft creation failed"),
            }
        }
    }
}

/// 若该 tag 尚无 release，则建草稿并入队文案任务；已存在则返回 `None`。
#[allow(clippy::too_many_arguments)]
pub async fn ensure_release_draft(
    db: &DatabaseConnection,
    redis: &AppRedis,
    cfg: &AiConfig,
    repo_id: Uuid,
    actor_id: Option<Uuid>,
    language: Option<String>,
    tag_name: &str,
    target_sha: &str,
) -> SyncResult<Option<Uuid>> {
    let lower_tag = tag_name.to_lowercase();
    if repo_release::Entity::find()
        .filter(repo_release::Column::RepoId.eq(repo_id))
        .filter(repo_release::Column::LowerTagName.eq(lower_tag.clone()))
        .one(db)
        .await?
        .is_some()
    {
        return Ok(None);
    }
    let publisher = match actor_id {
        Some(uid) => uid,
        None => repo_publisher_fallback(db, repo_id).await?,
    };
    let now = OffsetDateTime::now_utc();
    let release_id = Uuid::now_v7();
    let previous = previous_tag_target(db, repo_id, tag_name, target_sha).await;
    let commit_count = count_commits_between(db, repo_id, previous.as_deref(), target_sha).await;
    repo_release::ActiveModel {
        id: Set(release_id),
        repo_id: Set(repo_id),
        // 草稿由系统创建，发布者取仓库所有者；用户发布时再改写为实际操作者。
        publisher_id: Set(publisher),
        publisher_name: Set(String::new()),
        tag_name: Set(tag_name.to_string()),
        lower_tag_name: Set(lower_tag),
        target: Set(target_sha.to_string()),
        title: Set(tag_name.to_string()),
        content: Set(None),
        sha: Set(target_sha.to_string()),
        num_commits: Set(commit_count),
        is_draft: Set(true),
        is_prerelease: Set(tag_name.contains('-')),
        is_tag: Set(true),
        created_at: Set(now),
        updated_at: Set(now),
    }
    .insert(db)
    .await?;

    let task = ci_protocol::ai::AgentTask::ReleaseNotes {
        job_id: Uuid::now_v7(),
        repo_id,
        release_id,
        tag_name: tag_name.to_string(),
        from_sha: previous,
        to_sha: target_sha.to_string(),
        language: language.or(resolve_language_for(db, actor_id).await),
    };
    let job_id = submit_ai_task(db, redis, cfg, &task).await?;
    Ok(Some(job_id))
}

/// 管理端手动重试：按持久化的载荷原样重投队列。
///
/// 队列是 Redis 列表且没有投递确认：仅把 `ai_job.status` 改回 `pending`
/// 不会有任何消费者看到它，所以重试必须重新 `LPUSH` 原始载荷。载荷来自
/// 提交时的持久化副本；历史记录没有副本时按作业字段重建，重建失败才判失败。
pub async fn retry_ai_job(
    db: &DatabaseConnection,
    redis: &AppRedis,
    cfg: &AiConfig,
    row: &ai_job::Model,
) -> SyncResult<()> {
    // 优先使用提交时持久化的原始载荷；历史记录没有载荷时按作业字段重建，
    // 否则管理端对旧失败任务的重试永远无法生效。
    let task = match row.request_json.as_ref() {
        Some(value) => serde_json::from_value::<ci_protocol::ai::AgentTask>(value.clone())
            .map_err(|e| SyncError::Internal(format!("decode stored ai task: {e}")))?,
        None => match reconstruct_task(db, row).await {
            Some(task) => task,
            None => {
                let reason = "stored payload missing and task cannot be reconstructed".to_string();
                mark_retry_failure(db, row.id, &reason).await;
                return Err(SyncError::Internal(reason));
            }
        },
    };
    let payload = task
        .encode()
        .map_err(|e| SyncError::Internal(format!("encode ai task: {e}")))?;
    // 先复位再投递：消费者只能看到投递之后的作业，避免 running 被改回 pending。
    let now = OffsetDateTime::now_utc();
    if let Some(current) = ai_job::Entity::find_by_id(row.id).one(db).await? {
        let mut model: ActiveModel = current.into();
        model.status = Set(AiJobStatus::Pending);
        model.attempts = Set(0);
        model.error = Set(None);
        model.started_at = Set(None);
        model.finished_at = Set(None);
        model.updated_at = Set(Some(now));
        model.update(db).await?;
    }
    let queue = SyncQueue::new(redis.clone());
    if let Err(error) = queue.publish_raw(&cfg.queue_topic, None, &payload).await {
        mark_retry_failure(db, row.id, &error.to_string()).await;
        return Err(error);
    }
    tracing::info!(job_id = %row.id, kind = %row.kind.as_str(), "ai job requeued by admin");
    Ok(())
}

/// 历史作业没有 `request_json` 时按已存字段重建队列载荷。
///
/// 能唯一确定任务身份的信息都在 `ai_job` 上（kind / repo / target / ref /
/// language）；release 的区间起点需要回查实体，缺失时按整个历史生成。
async fn reconstruct_task(
    db: &DatabaseConnection,
    row: &ai_job::Model,
) -> Option<ci_protocol::ai::AgentTask> {
    use models::ai::ai_job::AiJobKind;
    let language = if row.language.trim().is_empty() {
        None
    } else {
        Some(row.language.clone())
    };
    let target_ref = row.target_ref.clone()?;
    match row.kind {
        AiJobKind::PullReview => Some(ci_protocol::ai::AgentTask::PullReview {
            job_id: row.id,
            repo_id: row.repo_id,
            pull_request_id: row.target_id?,
            head_sha: target_ref,
            language,
        }),
        AiJobKind::CommitReview => Some(ci_protocol::ai::AgentTask::CommitReview {
            job_id: row.id,
            repo_id: row.repo_id,
            commit_sha: target_ref,
            language,
        }),
        AiJobKind::RepoSummary => Some(ci_protocol::ai::AgentTask::RepoSummary {
            job_id: row.id,
            repo_id: row.repo_id,
            head_sha: target_ref,
            base_sha: None,
            limit: None,
            language,
        }),
        AiJobKind::ReleaseNotes => {
            let release = repo_release::Entity::find_by_id(row.target_id?)
                .one(db)
                .await
                .ok()??;
            Some(ci_protocol::ai::AgentTask::ReleaseNotes {
                job_id: row.id,
                repo_id: row.repo_id,
                release_id: release.id,
                tag_name: release.tag_name,
                from_sha: None,
                to_sha: release.sha,
                language,
            })
        }
    }
}

async fn mark_retry_failure(db: &DatabaseConnection, job_id: Uuid, error: &str) {
    let now = OffsetDateTime::now_utc();
    if let Ok(Some(row)) = ai_job::Entity::find_by_id(job_id).one(db).await {
        let mut model: ActiveModel = row.into();
        model.status = Set(AiJobStatus::Failed);
        model.error = Set(Some(error.to_string()));
        model.finished_at = Set(Some(now));
        model.updated_at = Set(Some(now));
        let _ = model.update(db).await;
    }
}

/// 草稿发布者的兜底：无推送者（如系统同步）时取空间所有者。
async fn repo_publisher_fallback(db: &DatabaseConnection, repo_id: Uuid) -> SyncResult<Uuid> {
    let repo = repos::Entity::find_by_id(repo_id)
        .one(db)
        .await?
        .ok_or_else(|| SyncError::Internal("repo not found for release".to_string()))?;
    let space = crate::share::access::space_for_dir(db, repo.dir_id)
        .await
        .map_err(|e| SyncError::Internal(format!("space lookup: {e}")))?
        .ok_or_else(|| SyncError::Internal("space not found for release".to_string()))?;
    Ok(space.owner_id)
}

/// 入队时的语言解析：调用方已解析过就直接用，避免重复查用户表。
async fn resolve_language_for(db: &DatabaseConnection, actor_id: Option<Uuid>) -> Option<String> {
    crate::sync::ai_language::actor_language(db, actor_id).await
}

/// 新标签的 changelog 区间起点。
///
/// 用「该标签之前最近一个已登记的标签」作为起点。同一个推送里可能一次带上
/// 多个标签，因此先按标签指向的提交在历史中的先后挑出真正的前驱，而不是
/// 依赖 `created_at` 的写入顺序。
async fn previous_tag_target(
    db: &DatabaseConnection,
    repo_id: Uuid,
    tag_name: &str,
    target_sha: &str,
) -> Option<String> {
    let rows = git_tags::Entity::find()
        .filter(git_tags::Column::RepoId.eq(repo_id))
        .filter(git_tags::Column::Name.ne(tag_name.to_string()))
        .all(db)
        .await
        .ok()?;
    // 排除指向同一个提交的别名标签，避免生成空区间。
    rows.into_iter()
        .find(|row| row.commit_sha != target_sha)
        .map(|row| row.commit_sha)
}

async fn count_commits_between(
    db: &DatabaseConnection,
    repo_id: Uuid,
    from: Option<&str>,
    to: &str,
) -> i32 {
    use models::git::git_commits;
    let mut select = git_commits::Entity::find().filter(git_commits::Column::RepoId.eq(repo_id));
    if let Some(from) = from {
        select = select.filter(git_commits::Column::Sha.ne(from.to_string()));
    }
    let _ = to;
    select
        .count(db)
        .await
        .map(|count| count as i32)
        .unwrap_or(0)
}

fn is_zero_oid(oid: &str) -> bool {
    oid.is_empty() || oid.chars().all(|c| c == '0')
}

async fn mark_submit_failure(db: &DatabaseConnection, job_id: Uuid, error: &str) {
    use models::ai::ai_job;
    use sea_orm::EntityTrait;
    let now = OffsetDateTime::now_utc();
    if let Ok(Some(row)) = ai_job::Entity::find_by_id(job_id).one(db).await {
        let mut model: ai_job::ActiveModel = row.into();
        model.status = Set(AiJobStatus::Failed);
        model.error = Set(Some(error.to_string()));
        model.finished_at = Set(Some(now));
        model.updated_at = Set(Some(now));
        let _ = model.update(db).await;
    }
}

fn kind_of(task: &ci_protocol::ai::AgentTask) -> models::ai::ai_job::AiJobKind {
    use ci_protocol::ai::AgentTask;
    use models::ai::ai_job::AiJobKind;
    match task {
        AgentTask::PullReview { .. } => AiJobKind::PullReview,
        AgentTask::CommitReview { .. } => AiJobKind::CommitReview,
        AgentTask::RepoSummary { .. } => AiJobKind::RepoSummary,
        AgentTask::ReleaseNotes { .. } => AiJobKind::ReleaseNotes,
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn cfg(enabled: bool, review: bool, summary: bool) -> AiConfig {
        AiConfig {
            enabled,
            review_enabled: review,
            summary_enabled: summary,
            ..Default::default()
        }
    }

    /// 仓库开关是「与」而不是「或」：仓库打开但平台没开时不能产生调用，否则
    /// 关闭平台 AI 的部署仍会往外发请求。这也是设置页要提示「开了也不生效」的原因。
    #[test]
    fn platform_switch_gates_repo_switch() {
        let off = push_ai_switches(&cfg(false, true, true), true, true);
        assert!(!off.review && !off.summary, "平台关闭时仓库开关不得放行");

        let on = push_ai_switches(&cfg(true, true, true), true, true);
        assert!(on.review && on.summary);
    }

    /// 仓库关掉某项时，平台开着也不能投递——这是该功能的全部意义。
    #[test]
    fn repo_switch_gates_each_capability() {
        let both_off = push_ai_switches(&cfg(true, true, true), false, false);
        assert!(!both_off.review && !both_off.summary);

        // 只开评审、不开自动总结：两个开关必须互不影响。
        let review_only = push_ai_switches(&cfg(true, true, true), true, false);
        assert!(review_only.review && !review_only.summary);

        let summary_only = push_ai_switches(&cfg(true, true, true), false, true);
        assert!(!summary_only.review && summary_only.summary);
    }

    /// 全关时不该继续往下走（`any()` 决定是否提前返回）。
    #[test]
    fn any_is_false_when_everything_is_off() {
        let mut off = cfg(true, false, false);
        off.release_notes_enabled = false;
        assert!(!push_ai_switches(&off, false, false).any());
        // release 文案没有仓库级开关，平台开着就仍应投递。
        let mut with_release = cfg(true, false, false);
        with_release.release_notes_enabled = true;
        assert!(push_ai_switches(&with_release, false, false).any());
    }
}