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::{
    ActiveModelTrait, ColumnTrait, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder,
    QuerySelect, Set,
};
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use time::OffsetDateTime;
use utoipa::{IntoParams, ToSchema};
use uuid::Uuid;

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

pub const SOURCE_IMPORT: &str = "import";
pub const SOURCE_SEARCH_INDEX: &str = "search_index";
pub const SOURCE_CHAT_WEBHOOK: &str = "chat_webhook";
pub const SOURCE_PIPELINE: &str = "pipeline";
pub const SOURCE_DELETION: &str = "deletion";
pub const SOURCE_SYSTEM_TASKS: &str = "system_tasks";
pub const SOURCE_AI: &str = "ai";

#[derive(Debug, Clone, Deserialize, IntoParams)]
pub struct ListTasksQuery {
    pub source: Option<String>,
    pub state: Option<String>,
    pub q: Option<String>,
    pub from: Option<OffsetDateTime>,
    pub to: Option<OffsetDateTime>,
    pub limit: Option<u64>,
    pub offset: Option<u64>,
}

#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct TaskDto {
    pub id: Uuid,
    pub source: String,
    pub title: String,
    pub state: String,
    pub created_at: OffsetDateTime,
    pub updated_at: Option<OffsetDateTime>,
    pub finished_at: Option<OffsetDateTime>,
    pub last_error: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct TaskPageDto {
    pub items: Vec<TaskDto>,
    pub total: i64,
    pub limit: u64,
    pub offset: u64,
}

#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct TaskDetailDto {
    pub task: TaskDto,
    pub detail: Value,
}

#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct TaskActionResp {
    pub source: String,
    pub id: Uuid,
    pub state: String,
}

fn pipeline_state_str(state: WorkflowRunState) -> String {
    match state {
        WorkflowRunState::Queued => "queued",
        WorkflowRunState::Running => "running",
        WorkflowRunState::Success => "success",
        WorkflowRunState::Failure => "failure",
        WorkflowRunState::CancelRequested => "cancel_requested",
        WorkflowRunState::Cancelled => "cancelled",
    }
    .to_string()
}

fn deletion_state_str(state: UserDeletionRequestStatus) -> String {
    match state {
        UserDeletionRequestStatus::Pending => "pending",
        UserDeletionRequestStatus::Scheduled => "scheduled",
        UserDeletionRequestStatus::Completed => "completed",
        UserDeletionRequestStatus::Cancelled => "cancelled",
    }
    .to_string()
}

fn task_state_str(state: TaskStatus) -> String {
    match state {
        TaskStatus::Pending => "pending",
        TaskStatus::Running => "running",
        TaskStatus::Success => "success",
        TaskStatus::Failed => "failed",
    }
    .to_string()
}

fn webhook_state_str(state: WebhookDeliveryStatus) -> String {
    match state {
        WebhookDeliveryStatus::Pending => "pending",
        WebhookDeliveryStatus::Delivering => "delivering",
        WebhookDeliveryStatus::Done => "done",
        WebhookDeliveryStatus::Failed => "failed",
    }
    .to_string()
}

fn short_sha(sha: &str) -> String {
    sha.chars().take(8).collect()
}

fn parse_pipeline_state(value: &str) -> Option<WorkflowRunState> {
    match value {
        "queued" => Some(WorkflowRunState::Queued),
        "running" => Some(WorkflowRunState::Running),
        "success" => Some(WorkflowRunState::Success),
        "failure" => Some(WorkflowRunState::Failure),
        "cancel_requested" => Some(WorkflowRunState::CancelRequested),
        "cancelled" => Some(WorkflowRunState::Cancelled),
        _ => None,
    }
}

fn parse_deletion_state(value: &str) -> Option<UserDeletionRequestStatus> {
    match value {
        "pending" => Some(UserDeletionRequestStatus::Pending),
        "scheduled" => Some(UserDeletionRequestStatus::Scheduled),
        "completed" => Some(UserDeletionRequestStatus::Completed),
        "cancelled" => Some(UserDeletionRequestStatus::Cancelled),
        _ => None,
    }
}

fn ai_state_str(state: models::ai::ai_job::AiJobStatus) -> String {
    use models::ai::ai_job::AiJobStatus;
    match state {
        AiJobStatus::Pending => "pending",
        AiJobStatus::Running => "running",
        AiJobStatus::Success => "success",
        AiJobStatus::Failed => "failed",
    }
    .to_string()
}

fn parse_ai_state(value: &str) -> Option<models::ai::ai_job::AiJobStatus> {
    use models::ai::ai_job::AiJobStatus;
    match value {
        "pending" => Some(AiJobStatus::Pending),
        "running" => Some(AiJobStatus::Running),
        "success" => Some(AiJobStatus::Success),
        "failed" => Some(AiJobStatus::Failed),
        _ => None,
    }
}

fn parse_task_status(value: &str) -> Option<TaskStatus> {
    match value {
        "pending" => Some(TaskStatus::Pending),
        "running" => Some(TaskStatus::Running),
        "success" => Some(TaskStatus::Success),
        "failed" => Some(TaskStatus::Failed),
        _ => None,
    }
}

fn parse_webhook_state(value: &str) -> Option<WebhookDeliveryStatus> {
    match value {
        "pending" => Some(WebhookDeliveryStatus::Pending),
        "delivering" => Some(WebhookDeliveryStatus::Delivering),
        "done" => Some(WebhookDeliveryStatus::Done),
        "failed" => Some(WebhookDeliveryStatus::Failed),
        _ => None,
    }
}

impl AdminService {
    fn requested_sources(&self, source: &Option<String>) -> Vec<&'static str> {
        match source.as_deref() {
            Some(SOURCE_IMPORT) => vec![SOURCE_IMPORT],
            Some(SOURCE_SEARCH_INDEX) => vec![SOURCE_SEARCH_INDEX],
            Some(SOURCE_CHAT_WEBHOOK) => vec![SOURCE_CHAT_WEBHOOK],
            Some(SOURCE_PIPELINE) => vec![SOURCE_PIPELINE],
            Some(SOURCE_DELETION) => vec![SOURCE_DELETION],
            Some(SOURCE_SYSTEM_TASKS) => vec![SOURCE_SYSTEM_TASKS],
            Some(SOURCE_AI) => vec![SOURCE_AI],
            _ => vec![
                SOURCE_IMPORT,
                SOURCE_SEARCH_INDEX,
                SOURCE_CHAT_WEBHOOK,
                SOURCE_PIPELINE,
                SOURCE_DELETION,
                SOURCE_SYSTEM_TASKS,
                SOURCE_AI,
            ],
        }
    }

    pub async fn list_tasks(&self, query: ListTasksQuery) -> Result<TaskPageDto, AdminError> {
        let limit = query.limit.unwrap_or(50).clamp(1, 100);
        let offset = query.offset.unwrap_or(0);
        let fetch_limit = (offset + limit).clamp(1, 500);
        let mut tasks: Vec<TaskDto> = Vec::new();
        for source in self.requested_sources(&query.source) {
            tasks.extend(
                self.fetch_source(
                    source,
                    fetch_limit,
                    query.state.as_deref(),
                    query.from,
                    query.to,
                )
                .await?,
            );
        }
        if let Some(needle) = query
            .q
            .as_deref()
            .map(str::trim)
            .filter(|value| !value.is_empty())
        {
            let needle = needle.to_ascii_lowercase();
            tasks.retain(|task| {
                task.title.to_ascii_lowercase().contains(&needle)
                    || task.id.to_string().contains(&needle)
                    || task
                        .last_error
                        .as_deref()
                        .is_some_and(|error| error.to_ascii_lowercase().contains(&needle))
            });
        }
        tasks.sort_by_key(|task| std::cmp::Reverse(task.created_at));
        let total = tasks.len() as i64;
        let items = tasks
            .into_iter()
            .skip(offset as usize)
            .take(limit as usize)
            .collect();
        Ok(TaskPageDto {
            items,
            total,
            limit,
            offset,
        })
    }
    async fn fetch_source(
        &self,
        source: &str,
        fetch_limit: u64,
        state: Option<&str>,
        from: Option<OffsetDateTime>,
        to: Option<OffsetDateTime>,
    ) -> Result<Vec<TaskDto>, AdminError> {
        let mut tasks = Vec::new();
        match source {
            SOURCE_IMPORT => {
                let mut select = job::Entity::find().order_by_desc(job::Column::CreatedAt);
                if let Some(state) = state {
                    select = select.filter(job::Column::State.eq(state));
                }
                if let Some(from) = from {
                    select = select.filter(job::Column::CreatedAt.gte(from));
                }
                if let Some(to) = to {
                    select = select.filter(job::Column::CreatedAt.lte(to));
                }
                for row in select.limit(fetch_limit).all(self.db().read()).await? {
                    tasks.push(TaskDto {
                        id: row.id,
                        source: SOURCE_IMPORT.to_string(),
                        title: format!("import {}", row.repo_name),
                        state: row.state,
                        created_at: row.created_at,
                        updated_at: Some(row.updated_at),
                        finished_at: row.finished_at,
                        last_error: row.last_error,
                    });
                }
            }
            SOURCE_SEARCH_INDEX => {
                let mut select =
                    index_job::Entity::find().order_by_desc(index_job::Column::CreatedAt);
                if let Some(state) = state {
                    select = select.filter(index_job::Column::State.eq(state));
                }
                if let Some(from) = from {
                    select = select.filter(index_job::Column::CreatedAt.gte(from));
                }
                if let Some(to) = to {
                    select = select.filter(index_job::Column::CreatedAt.lte(to));
                }
                for row in select.limit(fetch_limit).all(self.db().read()).await? {
                    tasks.push(TaskDto {
                        id: row.id,
                        source: SOURCE_SEARCH_INDEX.to_string(),
                        title: format!(
                            "index {} {}",
                            row.kind,
                            row.target_id
                                .map(|value| value.to_string())
                                .unwrap_or_else(|| "-".to_string())
                        ),
                        state: row.state,
                        created_at: row.created_at,
                        updated_at: Some(row.updated_at),
                        finished_at: None,
                        last_error: row.last_error,
                    });
                }
            }
            SOURCE_CHAT_WEBHOOK => {
                let mut select = webhook_deliveries::Entity::find()
                    .order_by_desc(webhook_deliveries::Column::CreatedAt);
                if let Some(state) = state {
                    select = select.filter(webhook_deliveries::Column::Status.eq(
                        parse_webhook_state(state).ok_or_else(|| {
                            AdminError::BadRequest("invalid webhook state".to_string())
                        })?,
                    ));
                }
                if let Some(from) = from {
                    select = select.filter(webhook_deliveries::Column::CreatedAt.gte(from));
                }
                if let Some(to) = to {
                    select = select.filter(webhook_deliveries::Column::CreatedAt.lte(to));
                }
                for row in select.limit(fetch_limit).all(self.db().read()).await? {
                    tasks.push(TaskDto {
                        id: row.id,
                        source: SOURCE_CHAT_WEBHOOK.to_string(),
                        title: format!("{} {}", row.event_type, row.webhook_id),
                        state: webhook_state_str(row.status),
                        created_at: row.created_at,
                        updated_at: Some(row.updated_at),
                        finished_at: None,
                        last_error: row.last_error,
                    });
                }
            }
            SOURCE_PIPELINE => {
                let mut select =
                    workflow_run::Entity::find().order_by_desc(workflow_run::Column::CreatedAt);
                if let Some(state) = state {
                    select = select.filter(workflow_run::Column::State.eq(
                        parse_pipeline_state(state).ok_or_else(|| {
                            AdminError::BadRequest("invalid pipeline state".to_string())
                        })?,
                    ));
                }
                if let Some(from) = from {
                    select = select.filter(workflow_run::Column::CreatedAt.gte(from));
                }
                if let Some(to) = to {
                    select = select.filter(workflow_run::Column::CreatedAt.lte(to));
                }
                for row in select.limit(fetch_limit).all(self.db().read()).await? {
                    tasks.push(TaskDto {
                        id: row.id,
                        source: SOURCE_PIPELINE.to_string(),
                        title: format!(
                            "run {}@{}",
                            short_sha(&row.commit_sha),
                            row.branch.clone().unwrap_or_else(|| row.git_ref.clone())
                        ),
                        state: pipeline_state_str(row.state),
                        created_at: row.created_at,
                        updated_at: Some(row.updated_at),
                        finished_at: row.finished_at,
                        last_error: None,
                    });
                }
            }
            SOURCE_DELETION => {
                let mut select = user_deletion_requests::Entity::find()
                    .order_by_desc(user_deletion_requests::Column::CreatedAt);
                if let Some(state) = state {
                    select = select.filter(user_deletion_requests::Column::Status.eq(
                        parse_deletion_state(state).ok_or_else(|| {
                            AdminError::BadRequest("invalid deletion state".to_string())
                        })?,
                    ));
                }
                if let Some(from) = from {
                    select = select.filter(user_deletion_requests::Column::CreatedAt.gte(from));
                }
                if let Some(to) = to {
                    select = select.filter(user_deletion_requests::Column::CreatedAt.lte(to));
                }
                for row in select.limit(fetch_limit).all(self.db().read()).await? {
                    tasks.push(TaskDto {
                        id: row.id,
                        source: SOURCE_DELETION.to_string(),
                        title: format!("deletion request {}", row.user_id),
                        state: deletion_state_str(row.status),
                        created_at: row.created_at,
                        updated_at: Some(row.updated_at),
                        finished_at: row.completed_at,
                        last_error: None,
                    });
                }
            }
            SOURCE_SYSTEM_TASKS => {
                let mut select =
                    system_tasks::Entity::find().order_by_desc(system_tasks::Column::CreatedAt);
                if let Some(state) = state {
                    select = select.filter(system_tasks::Column::Status.eq(
                        parse_task_status(state).ok_or_else(|| {
                            AdminError::BadRequest("invalid task state".to_string())
                        })?,
                    ));
                }
                if let Some(from) = from {
                    select = select.filter(system_tasks::Column::CreatedAt.gte(from));
                }
                if let Some(to) = to {
                    select = select.filter(system_tasks::Column::CreatedAt.lte(to));
                }
                for row in select.limit(fetch_limit).all(self.db().read()).await? {
                    tasks.push(TaskDto {
                        id: row.id,
                        source: SOURCE_SYSTEM_TASKS.to_string(),
                        title: row.task_type,
                        state: task_state_str(row.status),
                        created_at: row.created_at,
                        updated_at: row.updated_at,
                        finished_at: row.finished_at,
                        last_error: row.error,
                    });
                }
            }
            SOURCE_AI => {
                let mut select = ai_job::Entity::find().order_by_desc(ai_job::Column::CreatedAt);
                if let Some(state) = state {
                    select = select.filter(ai_job::Column::Status.eq(
                        parse_ai_state(state).ok_or_else(|| {
                            AdminError::BadRequest("invalid ai job state".to_string())
                        })?,
                    ));
                }
                if let Some(from) = from {
                    select = select.filter(ai_job::Column::CreatedAt.gte(from));
                }
                if let Some(to) = to {
                    select = select.filter(ai_job::Column::CreatedAt.lte(to));
                }
                for row in select.limit(fetch_limit).all(self.db().read()).await? {
                    tasks.push(TaskDto {
                        id: row.id,
                        source: SOURCE_AI.to_string(),
                        title: format!(
                            "{} {}",
                            row.kind.as_str(),
                            row.target_ref.unwrap_or_default()
                        ),
                        state: ai_state_str(row.status),
                        created_at: row.created_at,
                        updated_at: row.updated_at,
                        finished_at: row.finished_at,
                        last_error: row.error,
                    });
                }
            }
            _ => {}
        }
        Ok(tasks)
    }
    pub async fn get_task(&self, source: &str, id: Uuid) -> Result<TaskDetailDto, AdminError> {
        let (task, detail) = match source {
            SOURCE_IMPORT => {
                let row = job::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                let detail = json!({
                    "provider": row.provider,
                    "source_url": row.source_url,
                    "clone_url": row.clone_url,
                    "space_id": row.space_id,
                    "repo_id": row.repo_id,
                    "repo_name": row.repo_name,
                    "state": row.state,
                    "stage": row.stage,
                    "attempts": row.attempts,
                    "stats": row.stats,
                    "options": row.options,
                    "run_after": row.run_after,
                    "locked_at": row.locked_at,
                    "started_at": row.started_at,
                    "finished_at": row.finished_at,
                    "token": "[redacted]",
                });
                let task = TaskDto {
                    id: row.id,
                    source: SOURCE_IMPORT.to_string(),
                    title: format!("import {}", row.repo_name),
                    state: row.state,
                    created_at: row.created_at,
                    updated_at: Some(row.updated_at),
                    finished_at: row.finished_at,
                    last_error: row.last_error,
                };
                (task, detail)
            }
            SOURCE_SEARCH_INDEX => {
                let row = index_job::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                let detail = json!({
                    "kind": row.kind,
                    "target_id": row.target_id,
                    "repo_id": row.repo_id,
                    "attempts": row.attempts,
                    "run_after": row.run_after,
                    "locked_at": row.locked_at,
                });
                let task = TaskDto {
                    id: row.id,
                    source: SOURCE_SEARCH_INDEX.to_string(),
                    title: format!("index {}", row.kind),
                    state: row.state,
                    created_at: row.created_at,
                    updated_at: Some(row.updated_at),
                    finished_at: None,
                    last_error: row.last_error,
                };
                (task, detail)
            }
            SOURCE_CHAT_WEBHOOK => {
                let row = webhook_deliveries::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                let detail = json!({
                    "webhook_id": row.webhook_id,
                    "room_id": row.room_id,
                    "event_type": row.event_type,
                    "attempts": row.attempts,
                    "next_retry_at": row.next_retry_at,
                    "payload": "[redacted]",
                });
                let task = TaskDto {
                    id: row.id,
                    source: SOURCE_CHAT_WEBHOOK.to_string(),
                    title: format!("{} {}", row.event_type, row.webhook_id),
                    state: webhook_state_str(row.status),
                    created_at: row.created_at,
                    updated_at: Some(row.updated_at),
                    finished_at: None,
                    last_error: row.last_error,
                };
                (task, detail)
            }
            SOURCE_PIPELINE => {
                let row = workflow_run::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                let detail = json!({
                    "repo_id": row.repo_id,
                    "trigger_user_id": row.trigger_user_id,
                    "commit_sha": row.commit_sha,
                    "branch": row.branch,
                    "started_at": row.started_at,
                });
                let task = TaskDto {
                    id: row.id,
                    source: SOURCE_PIPELINE.to_string(),
                    title: format!(
                        "run {}@{}",
                        short_sha(&row.commit_sha),
                        row.branch.clone().unwrap_or_else(|| row.git_ref.clone())
                    ),
                    state: pipeline_state_str(row.state),
                    created_at: row.created_at,
                    updated_at: Some(row.updated_at),
                    finished_at: row.finished_at,
                    last_error: None,
                };
                (task, detail)
            }
            SOURCE_DELETION => {
                let row = user_deletion_requests::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                let detail = json!({
                    "user_id": row.user_id,
                    "reason": row.reason,
                    "scheduled_for": row.scheduled_for,
                    "completed_at": row.completed_at,
                    "cancelled_at": row.cancelled_at,
                });
                let task = TaskDto {
                    id: row.id,
                    source: SOURCE_DELETION.to_string(),
                    title: format!("deletion request {}", row.user_id),
                    state: deletion_state_str(row.status),
                    created_at: row.created_at,
                    updated_at: Some(row.updated_at),
                    finished_at: row.completed_at,
                    last_error: None,
                };
                (task, detail)
            }
            SOURCE_SYSTEM_TASKS => {
                let row = system_tasks::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                let detail = json!({
                    "task_type": row.task_type,
                    "payload": row.payload,
                    "retry_count": row.retry_count,
                    "next_run_at": row.next_run_at,
                    "started_at": row.started_at,
                });
                let task = TaskDto {
                    id: row.id,
                    source: SOURCE_SYSTEM_TASKS.to_string(),
                    title: row.task_type,
                    state: task_state_str(row.status),
                    created_at: row.created_at,
                    updated_at: row.updated_at,
                    finished_at: row.finished_at,
                    last_error: row.error,
                };
                (task, detail)
            }
            SOURCE_AI => {
                let row = ai_job::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                let detail = json!({
                    "kind": row.kind,
                    "repo_id": row.repo_id,
                    "target_id": row.target_id,
                    "target_ref": row.target_ref,
                    "model": row.model,
                    "attempts": row.attempts,
                    "prompt_tokens": row.prompt_tokens,
                    "completion_tokens": row.completion_tokens,
                    "error": row.error,
                    "started_at": row.started_at,
                });
                let task = TaskDto {
                    id: row.id,
                    source: SOURCE_AI.to_string(),
                    title: format!(
                        "{} {}",
                        row.kind.as_str(),
                        row.target_ref.unwrap_or_default()
                    ),
                    state: ai_state_str(row.status),
                    created_at: row.created_at,
                    updated_at: row.updated_at,
                    finished_at: row.finished_at,
                    last_error: row.error,
                };
                (task, detail)
            }
            _ => return Err(AdminError::BadRequest("unknown task source".to_string())),
        };
        Ok(TaskDetailDto { task, detail })
    }
    pub async fn retry_task(
        &self,
        actor: &AdminActor,
        source: &str,
        id: Uuid,
    ) -> Result<TaskActionResp, AdminError> {
        actor.require(AdminRole::Operator)?;
        let now = OffsetDateTime::now_utc();
        let state = match source {
            SOURCE_IMPORT => {
                let row = job::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                if !matches!(row.state.as_str(), "failed" | "canceled") {
                    return Err(AdminError::Conflict(
                        "only failed or canceled imports can be retried".to_string(),
                    ));
                }
                let mut am: job::ActiveModel = row.into();
                am.state = Set("queued".to_string());
                am.run_after = Set(now);
                am.locked_at = Set(None);
                am.finished_at = Set(None);
                am.updated_at = Set(now);
                am.update(self.db().write()).await?;
                "queued".to_string()
            }
            SOURCE_SEARCH_INDEX => {
                let row = index_job::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                let mut am: index_job::ActiveModel = row.into();
                am.state = Set("queued".to_string());
                am.run_after = Set(now);
                am.locked_at = Set(None);
                am.updated_at = Set(now);
                am.update(self.db().write()).await?;
                "queued".to_string()
            }
            SOURCE_CHAT_WEBHOOK => {
                let row = webhook_deliveries::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                if row.status != WebhookDeliveryStatus::Failed {
                    return Err(AdminError::Conflict(
                        "only failed webhook deliveries can be retried".to_string(),
                    ));
                }
                let mut am: webhook_deliveries::ActiveModel = row.into();
                am.status = Set(WebhookDeliveryStatus::Pending);
                am.next_retry_at = Set(now);
                am.updated_at = Set(now);
                am.update(self.db().write()).await?;
                "pending".to_string()
            }
            SOURCE_PIPELINE => {
                let row = workflow_run::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                if !matches!(
                    row.state,
                    WorkflowRunState::Failure | WorkflowRunState::Cancelled
                ) {
                    return Err(AdminError::Conflict(
                        "only failed or cancelled pipeline runs can be retried".to_string(),
                    ));
                }
                let mut am: workflow_run::ActiveModel = row.into();
                am.state = Set(WorkflowRunState::Queued);
                am.started_at = Set(None);
                am.finished_at = Set(None);
                am.updated_at = Set(now);
                am.update(self.db().write()).await?;
                "queued".to_string()
            }
            SOURCE_DELETION => {
                let row = user_deletion_requests::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                if row.status != UserDeletionRequestStatus::Cancelled {
                    return Err(AdminError::Conflict(
                        "only cancelled deletion requests can be retried".to_string(),
                    ));
                }
                let mut am: user_deletion_requests::ActiveModel = row.into();
                am.status = Set(UserDeletionRequestStatus::Scheduled);
                am.scheduled_for = Set(Some(now));
                am.cancelled_at = Set(None);
                am.updated_at = Set(now);
                am.update(self.db().write()).await?;
                "scheduled".to_string()
            }
            SOURCE_SYSTEM_TASKS => {
                let row = system_tasks::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                if row.status != TaskStatus::Failed {
                    return Err(AdminError::Conflict(
                        "only failed system tasks can be retried".to_string(),
                    ));
                }
                let mut am: system_tasks::ActiveModel = row.into();
                am.status = Set(TaskStatus::Pending);
                am.error = Set(None);
                am.next_run_at = Set(Some(now));
                am.started_at = Set(None);
                am.finished_at = Set(None);
                am.updated_at = Set(Some(now));
                am.update(self.db().write()).await?;
                "pending".to_string()
            }
            SOURCE_AI => {
                let row = ai_job::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                if row.status != models::ai::ai_job::AiJobStatus::Failed {
                    return Err(AdminError::Conflict(
                        "only failed ai jobs can be retried".to_string(),
                    ));
                }
                // 仅改数据库状态不会有消费者看到任务：必须按持久化载荷原样重投。
                git::sync::retry_ai_job(self.db().write(), self.redis(), &self.config().ai, &row)
                    .await
                    .map_err(|error| AdminError::Internal(format!("requeue ai job: {error}")))?;
                "pending".to_string()
            }
            _ => return Err(AdminError::BadRequest("unknown task source".to_string())),
        };
        Ok(TaskActionResp {
            source: source.to_string(),
            id,
            state,
        })
    }

    pub async fn cancel_task(
        &self,
        actor: &AdminActor,
        source: &str,
        id: Uuid,
    ) -> Result<TaskActionResp, AdminError> {
        actor.require(AdminRole::Operator)?;
        let now = OffsetDateTime::now_utc();
        let state = match source {
            SOURCE_IMPORT => {
                let row = job::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                if row.state == "processing" {
                    return Err(AdminError::Conflict(
                        "import is already processing and cannot be cancelled".to_string(),
                    ));
                }
                if row.state != "queued" {
                    return Err(AdminError::Conflict(
                        "only queued imports can be cancelled".to_string(),
                    ));
                }
                let mut am: job::ActiveModel = row.into();
                am.state = Set("canceled".to_string());
                am.finished_at = Set(Some(now));
                am.updated_at = Set(now);
                am.update(self.db().write()).await?;
                "canceled".to_string()
            }
            SOURCE_SEARCH_INDEX => {
                let row = index_job::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                if !matches!(row.state.as_str(), "queued" | "processing") {
                    return Err(AdminError::Conflict(
                        "only queued or processing index jobs can be cancelled".to_string(),
                    ));
                }
                let mut am: index_job::ActiveModel = row.into();
                am.state = Set("cancelled".to_string());
                am.locked_at = Set(None);
                am.updated_at = Set(now);
                am.update(self.db().write()).await?;
                "cancelled".to_string()
            }
            SOURCE_CHAT_WEBHOOK => {
                let row = webhook_deliveries::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                if !matches!(
                    row.status,
                    WebhookDeliveryStatus::Pending | WebhookDeliveryStatus::Delivering
                ) {
                    return Err(AdminError::Conflict(
                        "only pending webhook deliveries can be cancelled".to_string(),
                    ));
                }
                let mut am: webhook_deliveries::ActiveModel = row.into();
                am.status = Set(WebhookDeliveryStatus::Failed);
                am.last_error = Set(Some("cancelled by admin".to_string()));
                am.updated_at = Set(now);
                am.update(self.db().write()).await?;
                "failed".to_string()
            }
            SOURCE_PIPELINE => {
                let row = workflow_run::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                if !matches!(
                    row.state,
                    WorkflowRunState::Queued | WorkflowRunState::Running
                ) {
                    return Err(AdminError::Conflict(
                        "only queued or running pipeline runs can be cancelled".to_string(),
                    ));
                }
                let mut am: workflow_run::ActiveModel = row.into();
                am.state = Set(WorkflowRunState::CancelRequested);
                am.updated_at = Set(now);
                am.update(self.db().write()).await?;
                "cancel_requested".to_string()
            }
            SOURCE_DELETION => {
                let row = user_deletion_requests::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                if !matches!(
                    row.status,
                    UserDeletionRequestStatus::Pending | UserDeletionRequestStatus::Scheduled
                ) {
                    return Err(AdminError::Conflict(
                        "only pending deletion requests can be cancelled".to_string(),
                    ));
                }
                let mut am: user_deletion_requests::ActiveModel = row.into();
                am.status = Set(UserDeletionRequestStatus::Cancelled);
                am.cancelled_at = Set(Some(now));
                am.updated_at = Set(now);
                am.update(self.db().write()).await?;
                "cancelled".to_string()
            }
            SOURCE_SYSTEM_TASKS => {
                let row = system_tasks::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                if !matches!(row.status, TaskStatus::Pending | TaskStatus::Running) {
                    return Err(AdminError::Conflict(
                        "only pending or running tasks can be cancelled".to_string(),
                    ));
                }
                let mut am: system_tasks::ActiveModel = row.into();
                am.status = Set(TaskStatus::Failed);
                am.error = Set(Some("cancelled by admin".to_string()));
                am.finished_at = Set(Some(now));
                am.updated_at = Set(Some(now));
                am.update(self.db().write()).await?;
                "failed".to_string()
            }
            SOURCE_AI => {
                let row = ai_job::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                if !matches!(
                    row.status,
                    models::ai::ai_job::AiJobStatus::Pending
                        | models::ai::ai_job::AiJobStatus::Running
                ) {
                    return Err(AdminError::Conflict(
                        "only pending or running ai jobs can be cancelled".to_string(),
                    ));
                }
                let mut am: ai_job::ActiveModel = row.into();
                am.status = Set(models::ai::ai_job::AiJobStatus::Failed);
                am.error = Set(Some("cancelled by admin".to_string()));
                am.finished_at = Set(Some(now));
                am.updated_at = Set(Some(now));
                am.update(self.db().write()).await?;
                "failed".to_string()
            }
            _ => return Err(AdminError::BadRequest("unknown task source".to_string())),
        };
        Ok(TaskActionResp {
            source: source.to_string(),
            id,
            state,
        })
    }
    pub async fn delete_task(
        &self,
        actor: &AdminActor,
        source: &str,
        id: Uuid,
    ) -> Result<(), AdminError> {
        actor.require(AdminRole::SuperAdmin)?;
        let terminal = match source {
            SOURCE_IMPORT => {
                let row = job::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                matches!(row.state.as_str(), "succeeded" | "failed" | "canceled")
            }
            SOURCE_SEARCH_INDEX => {
                let row = index_job::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                matches!(row.state.as_str(), "succeeded" | "cancelled")
            }
            SOURCE_CHAT_WEBHOOK => {
                let row = webhook_deliveries::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                matches!(
                    row.status,
                    WebhookDeliveryStatus::Done | WebhookDeliveryStatus::Failed
                )
            }
            SOURCE_PIPELINE => {
                let row = workflow_run::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                matches!(
                    row.state,
                    WorkflowRunState::Success
                        | WorkflowRunState::Failure
                        | WorkflowRunState::Cancelled
                )
            }
            SOURCE_DELETION => {
                let row = user_deletion_requests::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                matches!(
                    row.status,
                    UserDeletionRequestStatus::Completed | UserDeletionRequestStatus::Cancelled
                )
            }
            SOURCE_SYSTEM_TASKS => {
                let row = system_tasks::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                matches!(row.status, TaskStatus::Success | TaskStatus::Failed)
            }
            SOURCE_AI => {
                let row = ai_job::Entity::find_by_id(id)
                    .one(self.db().read())
                    .await?
                    .ok_or(AdminError::NotFound)?;
                matches!(
                    row.status,
                    models::ai::ai_job::AiJobStatus::Success
                        | models::ai::ai_job::AiJobStatus::Failed
                )
            }
            _ => return Err(AdminError::BadRequest("unknown task source".to_string())),
        };
        if !terminal {
            return Err(AdminError::Conflict(
                "only finished tasks can be deleted".to_string(),
            ));
        }
        match source {
            SOURCE_IMPORT => {
                job::Entity::delete_by_id(id)
                    .exec(self.db().write())
                    .await?;
            }
            SOURCE_SEARCH_INDEX => {
                index_job::Entity::delete_by_id(id)
                    .exec(self.db().write())
                    .await?;
            }
            SOURCE_CHAT_WEBHOOK => {
                webhook_deliveries::Entity::delete_by_id(id)
                    .exec(self.db().write())
                    .await?;
            }
            SOURCE_PIPELINE => {
                workflow_run::Entity::delete_by_id(id)
                    .exec(self.db().write())
                    .await?;
            }
            SOURCE_DELETION => {
                user_deletion_requests::Entity::delete_by_id(id)
                    .exec(self.db().write())
                    .await?;
            }
            SOURCE_SYSTEM_TASKS => {
                system_tasks::Entity::delete_by_id(id)
                    .exec(self.db().write())
                    .await?;
            }
            SOURCE_AI => {
                ai_job::Entity::delete_by_id(id)
                    .exec(self.db().write())
                    .await?;
            }
            _ => {}
        }
        Ok(())
    }
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct TaskSourceStatDto {
    pub source: String,
    pub queued: i64,
    pub running: i64,
    pub succeeded: i64,
    pub failed: i64,
}

#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct TaskStatsDto {
    pub sources: Vec<TaskSourceStatDto>,
    pub total: i64,
    pub generated_at: OffsetDateTime,
}

#[derive(Debug, Clone, Deserialize, ToSchema)]
pub struct BulkTaskCmd {
    pub action: String,
    pub ids: Vec<Uuid>,
}

#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct BulkTaskFailure {
    pub id: Uuid,
    pub error: String,
}

#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct BulkTaskResp {
    pub source: String,
    pub action: String,
    pub succeeded: Vec<Uuid>,
    pub failed: Vec<BulkTaskFailure>,
}

impl AdminService {
    pub async fn task_stats(&self) -> Result<TaskStatsDto, AdminError> {
        let mut sources: Vec<TaskSourceStatDto> = Vec::new();
        for source in self.requested_sources(&None) {
            let stat = match source {
                SOURCE_IMPORT => TaskSourceStatDto {
                    source: source.to_string(),
                    queued: job::Entity::find()
                        .filter(job::Column::State.eq("queued"))
                        .count(self.db().read())
                        .await? as i64,
                    running: job::Entity::find()
                        .filter(job::Column::State.eq("processing"))
                        .count(self.db().read())
                        .await? as i64,
                    succeeded: job::Entity::find()
                        .filter(job::Column::State.eq("succeeded"))
                        .count(self.db().read())
                        .await? as i64,
                    failed: job::Entity::find()
                        .filter(job::Column::State.eq("failed"))
                        .count(self.db().read())
                        .await? as i64,
                },
                SOURCE_SEARCH_INDEX => TaskSourceStatDto {
                    source: source.to_string(),
                    queued: index_job::Entity::find()
                        .filter(index_job::Column::State.eq("queued"))
                        .count(self.db().read())
                        .await? as i64,
                    running: index_job::Entity::find()
                        .filter(index_job::Column::State.eq("processing"))
                        .count(self.db().read())
                        .await? as i64,
                    succeeded: index_job::Entity::find()
                        .filter(index_job::Column::State.eq("succeeded"))
                        .count(self.db().read())
                        .await? as i64,
                    failed: index_job::Entity::find()
                        .filter(index_job::Column::LastError.is_not_null())
                        .count(self.db().read())
                        .await? as i64,
                },
                SOURCE_CHAT_WEBHOOK => TaskSourceStatDto {
                    source: source.to_string(),
                    queued: webhook_deliveries::Entity::find()
                        .filter(
                            webhook_deliveries::Column::Status.eq(WebhookDeliveryStatus::Pending),
                        )
                        .count(self.db().read())
                        .await? as i64,
                    running: webhook_deliveries::Entity::find()
                        .filter(
                            webhook_deliveries::Column::Status
                                .eq(WebhookDeliveryStatus::Delivering),
                        )
                        .count(self.db().read())
                        .await? as i64,
                    succeeded: webhook_deliveries::Entity::find()
                        .filter(webhook_deliveries::Column::Status.eq(WebhookDeliveryStatus::Done))
                        .count(self.db().read())
                        .await? as i64,
                    failed: webhook_deliveries::Entity::find()
                        .filter(
                            webhook_deliveries::Column::Status.eq(WebhookDeliveryStatus::Failed),
                        )
                        .count(self.db().read())
                        .await? as i64,
                },
                SOURCE_PIPELINE => TaskSourceStatDto {
                    source: source.to_string(),
                    queued: workflow_run::Entity::find()
                        .filter(workflow_run::Column::State.eq(WorkflowRunState::Queued))
                        .count(self.db().read())
                        .await? as i64,
                    running: workflow_run::Entity::find()
                        .filter(workflow_run::Column::State.eq(WorkflowRunState::Running))
                        .count(self.db().read())
                        .await? as i64,
                    succeeded: workflow_run::Entity::find()
                        .filter(workflow_run::Column::State.eq(WorkflowRunState::Success))
                        .count(self.db().read())
                        .await? as i64,
                    failed: workflow_run::Entity::find()
                        .filter(workflow_run::Column::State.eq(WorkflowRunState::Failure))
                        .count(self.db().read())
                        .await? as i64,
                },
                SOURCE_DELETION => TaskSourceStatDto {
                    source: source.to_string(),
                    queued: user_deletion_requests::Entity::find()
                        .filter(user_deletion_requests::Column::Status.is_in([
                            UserDeletionRequestStatus::Pending,
                            UserDeletionRequestStatus::Scheduled,
                        ]))
                        .count(self.db().read())
                        .await? as i64,
                    running: 0,
                    succeeded: user_deletion_requests::Entity::find()
                        .filter(
                            user_deletion_requests::Column::Status
                                .eq(UserDeletionRequestStatus::Completed),
                        )
                        .count(self.db().read())
                        .await? as i64,
                    failed: 0,
                },
                SOURCE_SYSTEM_TASKS => TaskSourceStatDto {
                    source: source.to_string(),
                    queued: system_tasks::Entity::find()
                        .filter(system_tasks::Column::Status.eq(TaskStatus::Pending))
                        .count(self.db().read())
                        .await? as i64,
                    running: system_tasks::Entity::find()
                        .filter(system_tasks::Column::Status.eq(TaskStatus::Running))
                        .count(self.db().read())
                        .await? as i64,
                    succeeded: system_tasks::Entity::find()
                        .filter(system_tasks::Column::Status.eq(TaskStatus::Success))
                        .count(self.db().read())
                        .await? as i64,
                    failed: system_tasks::Entity::find()
                        .filter(system_tasks::Column::Status.eq(TaskStatus::Failed))
                        .count(self.db().read())
                        .await? as i64,
                },
                SOURCE_AI => {
                    use models::ai::ai_job::AiJobStatus;
                    TaskSourceStatDto {
                        source: source.to_string(),
                        queued: ai_job::Entity::find()
                            .filter(ai_job::Column::Status.eq(AiJobStatus::Pending))
                            .count(self.db().read())
                            .await? as i64,
                        running: ai_job::Entity::find()
                            .filter(ai_job::Column::Status.eq(AiJobStatus::Running))
                            .count(self.db().read())
                            .await? as i64,
                        succeeded: ai_job::Entity::find()
                            .filter(ai_job::Column::Status.eq(AiJobStatus::Success))
                            .count(self.db().read())
                            .await? as i64,
                        failed: ai_job::Entity::find()
                            .filter(ai_job::Column::Status.eq(AiJobStatus::Failed))
                            .count(self.db().read())
                            .await? as i64,
                    }
                }
                // `&str` 常量模式不构成穷尽匹配；requested_sources 只会产出已知来源。
                _ => continue,
            };
            sources.push(stat);
        }
        let total = sources
            .iter()
            .map(|stat| stat.queued + stat.running + stat.succeeded + stat.failed)
            .sum();
        Ok(TaskStatsDto {
            sources,
            total,
            generated_at: OffsetDateTime::now_utc(),
        })
    }

    pub async fn bulk_tasks(
        &self,
        actor: &AdminActor,
        source: &str,
        cmd: BulkTaskCmd,
    ) -> Result<BulkTaskResp, AdminError> {
        actor.require(AdminRole::Operator)?;
        if ![
            SOURCE_IMPORT,
            SOURCE_SEARCH_INDEX,
            SOURCE_CHAT_WEBHOOK,
            SOURCE_PIPELINE,
            SOURCE_DELETION,
            SOURCE_SYSTEM_TASKS,
            SOURCE_AI,
        ]
        .contains(&source)
        {
            return Err(AdminError::BadRequest(format!(
                "unknown task source: {source}"
            )));
        }
        let action = cmd.action.trim().to_ascii_lowercase();
        match action.as_str() {
            "retry" | "cancel" => {}
            other => {
                return Err(AdminError::BadRequest(format!(
                    "unsupported bulk action: {other}"
                )));
            }
        }
        if cmd.ids.is_empty() {
            return Err(AdminError::BadRequest("ids must not be empty".to_string()));
        }
        if cmd.ids.len() > 100 {
            return Err(AdminError::BadRequest(
                "at most 100 ids per request".to_string(),
            ));
        }
        let mut succeeded: Vec<Uuid> = Vec::new();
        let mut failed: Vec<BulkTaskFailure> = Vec::new();
        for id in cmd.ids {
            let outcome = if action == "retry" {
                self.retry_task(actor, source, id).await.map(|_| ())
            } else {
                self.cancel_task(actor, source, id).await.map(|_| ())
            };
            match outcome {
                Ok(()) => succeeded.push(id),
                Err(AdminError::NotFound) => failed.push(BulkTaskFailure {
                    id,
                    error: "not found".to_string(),
                }),
                Err(error) => failed.push(BulkTaskFailure {
                    id,
                    error: error.to_string(),
                }),
            }
        }
        Ok(BulkTaskResp {
            source: source.to_string(),
            action,
            succeeded,
            failed,
        })
    }
}