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::{
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,
})
}
}