eba4fd1
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,
})
}
}