use models::ai::ai_job;
use models::chat::webhook_deliveries::{self, WebhookDeliveryStatus};
use models::ci::workflow_run::{self, WorkflowRunState};
use models::import::job;
use models::search::index_job;
use models::system::system_tasks::{self, TaskStatus};
use models::users::user_deletion_requests::{self, UserDeletionRequestStatus};
use sea_orm::sea_query::Expr;
use sea_orm::{ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter};
use serde::{Deserialize, Serialize};
use utoipa::ToSchema;

use crate::errors::AdminError;
use crate::services::service::AdminService;
use crate::share::roles::{AdminActor, AdminRole};

/// 读取 Redis 列表长度作为 AI 队列积压深度。
async fn ai_depth(topic: &str, service: &AdminService) -> i64 {
    let mut conn = service.redis().connection_clone();
    let depth: Result<i64, redis::RedisError> = match &mut conn {
        export::RedisConnection::Standalone(connection) => {
            redis::cmd("LLEN").arg(topic).query_async(connection).await
        }
        export::RedisConnection::Cluster(connection) => {
            redis::cmd("LLEN").arg(topic).query_async(connection).await
        }
    };
    depth.unwrap_or_default()
}

pub const QUEUE_GIT_SYNC: &str = "git_sync";
pub const QUEUE_EMAIL: &str = "email";
pub const QUEUE_SEARCH_INDEX: &str = "search_index";
pub const QUEUE_IMPORT: &str = "import";
pub const QUEUE_CHAT_WEBHOOK: &str = "chat_webhook";
pub const QUEUE_PIPELINE: &str = "pipeline";
pub const QUEUE_DELETION: &str = "deletion";
pub const QUEUE_SYSTEM_TASKS: &str = "system_tasks";
pub const QUEUE_AI: &str = "ai";

#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct QueueSnapshotDto {
    pub name: String,
    pub kind: String,
    pub enabled: bool,
    pub depth: Option<i64>,
    pub queued: i64,
    pub running: i64,
    pub failed: i64,
    pub succeeded: i64,
    pub note: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct QueueActionResp {
    pub queue: String,
    pub affected: u64,
}

impl AdminService {
    async fn git_sync_depth(&self) -> Result<i64, AdminError> {
        let topic = self.config().git.sync.queue_topic.clone();
        let mut conn = self.redis().connection_clone();
        let depth: i64 = match &mut conn {
            export::RedisConnection::Standalone(connection) => {
                redis::cmd("LLEN")
                    .arg(&topic)
                    .query_async(connection)
                    .await?
            }
            export::RedisConnection::Cluster(connection) => {
                redis::cmd("LLEN")
                    .arg(&topic)
                    .query_async(connection)
                    .await?
            }
        };
        Ok(depth)
    }

    pub async fn queue_snapshots(&self) -> Result<Vec<QueueSnapshotDto>, AdminError> {
        let mut snapshots = Vec::new();

        let depth = self.git_sync_depth().await?;
        snapshots.push(QueueSnapshotDto {
            name: QUEUE_GIT_SYNC.to_string(),
            kind: "redis_list".to_string(),
            enabled: true,
            depth: Some(depth),
            queued: depth,
            running: 0,
            failed: 0,
            succeeded: 0,
            note: Some(format!(
                "topic={} concurrency={}",
                self.config().git.sync.queue_topic,
                self.config().git.sync.concurrency.max(1)
            )),
        });

        snapshots.push(QueueSnapshotDto {
            name: QUEUE_EMAIL.to_string(),
            kind: "redis_pubsub".to_string(),
            enabled: self.config().email_queue.enabled(),
            depth: None,
            queued: 0,
            running: 0,
            failed: 0,
            succeeded: 0,
            note: Some("redis pub/sub fan-out, no persistent backlog".to_string()),
        });

        let search_queued = index_job::Entity::find()
            .filter(index_job::Column::State.eq("queued"))
            .count(self.db().read())
            .await? as i64;
        let search_processing = index_job::Entity::find()
            .filter(index_job::Column::State.eq("processing"))
            .count(self.db().read())
            .await? as i64;
        let search_succeeded = index_job::Entity::find()
            .filter(index_job::Column::State.eq("succeeded"))
            .count(self.db().read())
            .await? as i64;
        let search_failed = index_job::Entity::find()
            .filter(index_job::Column::LastError.is_not_null())
            .count(self.db().read())
            .await? as i64;
        snapshots.push(QueueSnapshotDto {
            name: QUEUE_SEARCH_INDEX.to_string(),
            kind: "db".to_string(),
            enabled: self.config().search.enabled,
            depth: Some(search_queued + search_processing),
            queued: search_queued,
            running: search_processing,
            failed: search_failed,
            succeeded: search_succeeded,
            note: None,
        });

        let import_queued = job::Entity::find()
            .filter(job::Column::State.eq("queued"))
            .count(self.db().read())
            .await? as i64;
        let import_processing = job::Entity::find()
            .filter(job::Column::State.eq("processing"))
            .count(self.db().read())
            .await? as i64;
        let import_failed = job::Entity::find()
            .filter(job::Column::State.eq("failed"))
            .count(self.db().read())
            .await? as i64;
        let import_succeeded = job::Entity::find()
            .filter(job::Column::State.eq("succeeded"))
            .count(self.db().read())
            .await? as i64;
        snapshots.push(QueueSnapshotDto {
            name: QUEUE_IMPORT.to_string(),
            kind: "db".to_string(),
            enabled: self.config().import.enabled,
            depth: Some(import_queued + import_processing),
            queued: import_queued,
            running: import_processing,
            failed: import_failed,
            succeeded: import_succeeded,
            note: Some(format!(
                "max_concurrent_jobs={}",
                self.config().import.max_concurrent_jobs.max(1)
            )),
        });

        let webhook_pending = webhook_deliveries::Entity::find()
            .filter(webhook_deliveries::Column::Status.eq(WebhookDeliveryStatus::Pending))
            .count(self.db().read())
            .await? as i64;
        let webhook_running = webhook_deliveries::Entity::find()
            .filter(webhook_deliveries::Column::Status.eq(WebhookDeliveryStatus::Delivering))
            .count(self.db().read())
            .await? as i64;
        let webhook_failed = webhook_deliveries::Entity::find()
            .filter(webhook_deliveries::Column::Status.eq(WebhookDeliveryStatus::Failed))
            .count(self.db().read())
            .await? as i64;
        let webhook_done = webhook_deliveries::Entity::find()
            .filter(webhook_deliveries::Column::Status.eq(WebhookDeliveryStatus::Done))
            .count(self.db().read())
            .await? as i64;
        snapshots.push(QueueSnapshotDto {
            name: QUEUE_CHAT_WEBHOOK.to_string(),
            kind: "db".to_string(),
            enabled: true,
            depth: Some(webhook_pending + webhook_running),
            queued: webhook_pending,
            running: webhook_running,
            failed: webhook_failed,
            succeeded: webhook_done,
            note: None,
        });

        let pipeline_queued = workflow_run::Entity::find()
            .filter(workflow_run::Column::State.eq(WorkflowRunState::Queued))
            .count(self.db().read())
            .await? as i64;
        let pipeline_running = workflow_run::Entity::find()
            .filter(workflow_run::Column::State.eq(WorkflowRunState::Running))
            .count(self.db().read())
            .await? as i64;
        let pipeline_failed = workflow_run::Entity::find()
            .filter(workflow_run::Column::State.eq(WorkflowRunState::Failure))
            .count(self.db().read())
            .await? as i64;
        let pipeline_succeeded = workflow_run::Entity::find()
            .filter(workflow_run::Column::State.eq(WorkflowRunState::Success))
            .count(self.db().read())
            .await? as i64;
        snapshots.push(QueueSnapshotDto {
            name: QUEUE_PIPELINE.to_string(),
            kind: "db".to_string(),
            enabled: true,
            depth: Some(pipeline_queued + pipeline_running),
            queued: pipeline_queued,
            running: pipeline_running,
            failed: pipeline_failed,
            succeeded: pipeline_succeeded,
            note: Some("external runner driven".to_string()),
        });

        let deletion_pending = user_deletion_requests::Entity::find()
            .filter(user_deletion_requests::Column::Status.eq(UserDeletionRequestStatus::Pending))
            .count(self.db().read())
            .await? as i64;
        let deletion_scheduled = user_deletion_requests::Entity::find()
            .filter(user_deletion_requests::Column::Status.eq(UserDeletionRequestStatus::Scheduled))
            .count(self.db().read())
            .await? as i64;
        let deletion_completed = user_deletion_requests::Entity::find()
            .filter(user_deletion_requests::Column::Status.eq(UserDeletionRequestStatus::Completed))
            .count(self.db().read())
            .await? as i64;
        snapshots.push(QueueSnapshotDto {
            name: QUEUE_DELETION.to_string(),
            kind: "db".to_string(),
            enabled: true,
            depth: Some(deletion_pending + deletion_scheduled),
            queued: deletion_scheduled,
            running: deletion_pending,
            failed: 0,
            succeeded: deletion_completed,
            note: None,
        });

        let task_pending = system_tasks::Entity::find()
            .filter(system_tasks::Column::Status.eq(TaskStatus::Pending))
            .count(self.db().read())
            .await? as i64;
        let task_running = system_tasks::Entity::find()
            .filter(system_tasks::Column::Status.eq(TaskStatus::Running))
            .count(self.db().read())
            .await? as i64;
        let task_failed = system_tasks::Entity::find()
            .filter(system_tasks::Column::Status.eq(TaskStatus::Failed))
            .count(self.db().read())
            .await? as i64;
        let task_succeeded = system_tasks::Entity::find()
            .filter(system_tasks::Column::Status.eq(TaskStatus::Success))
            .count(self.db().read())
            .await? as i64;
        snapshots.push(QueueSnapshotDto {
            name: QUEUE_SYSTEM_TASKS.to_string(),
            kind: "db".to_string(),
            enabled: true,
            depth: Some(task_pending + task_running),
            queued: task_pending,
            running: task_running,
            failed: task_failed,
            succeeded: task_succeeded,
            note: Some("reserved scheduler table".to_string()),
        });

        let ai_pending = ai_job::Entity::find()
            .filter(ai_job::Column::Status.eq(models::ai::ai_job::AiJobStatus::Pending))
            .count(self.db().read())
            .await? as i64;
        let ai_running = ai_job::Entity::find()
            .filter(ai_job::Column::Status.eq(models::ai::ai_job::AiJobStatus::Running))
            .count(self.db().read())
            .await? as i64;
        let ai_succeeded = ai_job::Entity::find()
            .filter(ai_job::Column::Status.eq(models::ai::ai_job::AiJobStatus::Success))
            .count(self.db().read())
            .await? as i64;
        let ai_failed = ai_job::Entity::find()
            .filter(ai_job::Column::Status.eq(models::ai::ai_job::AiJobStatus::Failed))
            .count(self.db().read())
            .await? as i64;
        snapshots.push(QueueSnapshotDto {
            name: QUEUE_AI.to_string(),
            kind: "redis_list".to_string(),
            enabled: self.config().ai.enabled,
            depth: Some(ai_depth(&self.config().ai.queue_topic, self).await),
            queued: ai_pending,
            running: ai_running,
            failed: ai_failed,
            succeeded: ai_succeeded,
            note: Some(format!(
                "topic={} model={}",
                self.config().ai.queue_topic,
                self.config().ai.model
            )),
        });

        Ok(snapshots)
    }

    pub async fn requeue_failed(
        &self,
        actor: &AdminActor,
        queue: &str,
    ) -> Result<QueueActionResp, AdminError> {
        actor.require(AdminRole::Operator)?;
        let now = time::OffsetDateTime::now_utc();
        let affected = match queue {
            QUEUE_SEARCH_INDEX => {
                index_job::Entity::update_many()
                    .col_expr(index_job::Column::State, Expr::value("queued"))
                    .col_expr(index_job::Column::RunAfter, Expr::value(now))
                    .col_expr(
                        index_job::Column::LockedAt,
                        Expr::value(Option::<time::OffsetDateTime>::None),
                    )
                    .col_expr(index_job::Column::UpdatedAt, Expr::value(now))
                    .filter(index_job::Column::LastError.is_not_null())
                    .filter(index_job::Column::State.eq("queued"))
                    .exec(self.db().write())
                    .await?
                    .rows_affected
            }
            QUEUE_IMPORT => {
                job::Entity::update_many()
                    .col_expr(job::Column::State, Expr::value("queued"))
                    .col_expr(job::Column::RunAfter, Expr::value(now))
                    .col_expr(
                        job::Column::LockedAt,
                        Expr::value(Option::<time::OffsetDateTime>::None),
                    )
                    .col_expr(
                        job::Column::FinishedAt,
                        Expr::value(Option::<time::OffsetDateTime>::None),
                    )
                    .col_expr(job::Column::UpdatedAt, Expr::value(now))
                    .filter(job::Column::State.is_in(["failed", "canceled"]))
                    .exec(self.db().write())
                    .await?
                    .rows_affected
            }
            QUEUE_CHAT_WEBHOOK => {
                webhook_deliveries::Entity::update_many()
                    .col_expr(
                        webhook_deliveries::Column::Status,
                        Expr::value(WebhookDeliveryStatus::Pending),
                    )
                    .col_expr(webhook_deliveries::Column::NextRetryAt, Expr::value(now))
                    .col_expr(webhook_deliveries::Column::UpdatedAt, Expr::value(now))
                    .filter(webhook_deliveries::Column::Status.eq(WebhookDeliveryStatus::Failed))
                    .exec(self.db().write())
                    .await?
                    .rows_affected
            }
            QUEUE_PIPELINE => {
                workflow_run::Entity::update_many()
                    .col_expr(
                        workflow_run::Column::State,
                        Expr::value(WorkflowRunState::Queued),
                    )
                    .col_expr(
                        workflow_run::Column::StartedAt,
                        Expr::value(Option::<time::OffsetDateTime>::None),
                    )
                    .col_expr(
                        workflow_run::Column::FinishedAt,
                        Expr::value(Option::<time::OffsetDateTime>::None),
                    )
                    .col_expr(workflow_run::Column::UpdatedAt, Expr::value(now))
                    .filter(
                        workflow_run::Column::State
                            .is_in([WorkflowRunState::Failure, WorkflowRunState::Cancelled]),
                    )
                    .exec(self.db().write())
                    .await?
                    .rows_affected
            }
            QUEUE_DELETION => {
                user_deletion_requests::Entity::update_many()
                    .col_expr(
                        user_deletion_requests::Column::Status,
                        Expr::value(UserDeletionRequestStatus::Scheduled),
                    )
                    .col_expr(
                        user_deletion_requests::Column::ScheduledFor,
                        Expr::value(Some(now)),
                    )
                    .col_expr(
                        user_deletion_requests::Column::CancelledAt,
                        Expr::value(Option::<time::OffsetDateTime>::None),
                    )
                    .col_expr(user_deletion_requests::Column::UpdatedAt, Expr::value(now))
                    .filter(
                        user_deletion_requests::Column::Status
                            .eq(UserDeletionRequestStatus::Cancelled),
                    )
                    .exec(self.db().write())
                    .await?
                    .rows_affected
            }
            QUEUE_SYSTEM_TASKS => {
                system_tasks::Entity::update_many()
                    .col_expr(
                        system_tasks::Column::Status,
                        Expr::value(TaskStatus::Pending),
                    )
                    .col_expr(system_tasks::Column::NextRunAt, Expr::value(Some(now)))
                    .col_expr(
                        system_tasks::Column::Error,
                        Expr::value(Option::<String>::None),
                    )
                    .col_expr(
                        system_tasks::Column::StartedAt,
                        Expr::value(Option::<time::OffsetDateTime>::None),
                    )
                    .col_expr(
                        system_tasks::Column::FinishedAt,
                        Expr::value(Option::<time::OffsetDateTime>::None),
                    )
                    .col_expr(system_tasks::Column::UpdatedAt, Expr::value(now))
                    .filter(system_tasks::Column::Status.eq(TaskStatus::Failed))
                    .exec(self.db().write())
                    .await?
                    .rows_affected
            }
            QUEUE_AI => {
                // 批量重排与单条重试同源：只改状态不会让消费者看到任务，
                // 必须用持久化的载荷逐条重投；缺载荷的历史作业留失败。
                let rows = ai_job::Entity::find()
                    .filter(ai_job::Column::Status.eq(models::ai::ai_job::AiJobStatus::Failed))
                    .all(self.db().read())
                    .await?;
                let mut affected = 0u64;
                for row in rows {
                    if git::sync::retry_ai_job(
                        self.db().write(),
                        self.redis(),
                        &self.config().ai,
                        &row,
                    )
                    .await
                    .is_ok()
                    {
                        affected += 1;
                    }
                }
                affected
            }
            _ => {
                return Err(AdminError::BadRequest(format!(
                    "queue {queue} does not support requeue"
                )));
            }
        };
        Ok(QueueActionResp {
            queue: queue.to_string(),
            affected,
        })
    }

    pub async fn purge_queue(
        &self,
        actor: &AdminActor,
        queue: &str,
    ) -> Result<QueueActionResp, AdminError> {
        actor.require(AdminRole::SuperAdmin)?;
        let affected = match queue {
            QUEUE_GIT_SYNC => {
                let topic = self.config().git.sync.queue_topic.clone();
                let mut conn = self.redis().connection_clone();
                let deleted: i64 = match &mut conn {
                    export::RedisConnection::Standalone(connection) => {
                        redis::cmd("DEL")
                            .arg(&topic)
                            .query_async(connection)
                            .await?
                    }
                    export::RedisConnection::Cluster(connection) => {
                        redis::cmd("DEL")
                            .arg(&topic)
                            .query_async(connection)
                            .await?
                    }
                };
                deleted.max(0) as u64
            }
            QUEUE_SEARCH_INDEX => {
                index_job::Entity::delete_many()
                    .filter(index_job::Column::State.is_in(["succeeded", "cancelled"]))
                    .exec(self.db().write())
                    .await?
                    .rows_affected
            }
            QUEUE_IMPORT => {
                job::Entity::delete_many()
                    .filter(job::Column::State.is_in(["succeeded", "failed", "canceled"]))
                    .exec(self.db().write())
                    .await?
                    .rows_affected
            }
            QUEUE_CHAT_WEBHOOK => {
                webhook_deliveries::Entity::delete_many()
                    .filter(
                        webhook_deliveries::Column::Status
                            .is_in([WebhookDeliveryStatus::Done, WebhookDeliveryStatus::Failed]),
                    )
                    .exec(self.db().write())
                    .await?
                    .rows_affected
            }
            QUEUE_PIPELINE => {
                workflow_run::Entity::delete_many()
                    .filter(workflow_run::Column::State.is_in([
                        WorkflowRunState::Success,
                        WorkflowRunState::Failure,
                        WorkflowRunState::Cancelled,
                    ]))
                    .exec(self.db().write())
                    .await?
                    .rows_affected
            }
            QUEUE_DELETION => {
                user_deletion_requests::Entity::delete_many()
                    .filter(user_deletion_requests::Column::Status.is_in([
                        UserDeletionRequestStatus::Completed,
                        UserDeletionRequestStatus::Cancelled,
                    ]))
                    .exec(self.db().write())
                    .await?
                    .rows_affected
            }
            QUEUE_SYSTEM_TASKS => {
                system_tasks::Entity::delete_many()
                    .filter(
                        system_tasks::Column::Status
                            .is_in([TaskStatus::Success, TaskStatus::Failed]),
                    )
                    .exec(self.db().write())
                    .await?
                    .rows_affected
            }
            QUEUE_AI => {
                ai_job::Entity::delete_many()
                    .filter(ai_job::Column::Status.is_in([
                        models::ai::ai_job::AiJobStatus::Success,
                        models::ai::ai_job::AiJobStatus::Failed,
                    ]))
                    .exec(self.db().write())
                    .await?
                    .rows_affected
            }
            _ => {
                return Err(AdminError::BadRequest(format!(
                    "queue {queue} does not support purge"
                )));
            }
        };
        Ok(QueueActionResp {
            queue: queue.to_string(),
            affected,
        })
    }
}