eba4fd1
use models::ai::ai_job;
use models::chat::webhook_deliveries;
use models::ci::workflow_run::{self, WorkflowRunState};
use models::import::job;
use models::search::index_job;
use models::system::system_tasks;
use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect};
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use utoipa::{IntoParams, ToSchema};
use uuid::Uuid;
use crate::errors::AdminError;
use crate::services::service::AdminService;
#[derive(Debug, Clone, Deserialize, IntoParams)]
pub struct ListJobLogsQuery {
pub source: 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 JobLogDto {
pub source: String,
pub id: Uuid,
pub state: String,
pub error: Option<String>,
pub created_at: OffsetDateTime,
pub updated_at: Option<OffsetDateTime>,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct JobLogPageDto {
pub items: Vec<JobLogDto>,
pub total: i64,
pub limit: u64,
pub offset: u64,
}
impl AdminService {
pub async fn list_job_logs(
&self,
query: ListJobLogsQuery,
) -> Result<JobLogPageDto, 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 logs: Vec<JobLogDto> = Vec::new();
let sources = match query.source.as_deref() {
Some("import") => vec!["import"],
Some("search_index") => vec!["search_index"],
Some("chat_webhook") => vec!["chat_webhook"],
Some("pipeline") => vec!["pipeline"],
Some("system_tasks") => vec!["system_tasks"],
Some("ai") => vec!["ai"],
_ => vec![
"import",
"search_index",
"chat_webhook",
"pipeline",
"system_tasks",
"ai",
],
};
for source in sources {
match source {
"import" => {
let mut select = job::Entity::find()
.filter(job::Column::LastError.is_not_null())
.order_by_desc(job::Column::UpdatedAt);
if let Some(from) = query.from {
select = select.filter(job::Column::CreatedAt.gte(from));
}
if let Some(to) = query.to {
select = select.filter(job::Column::CreatedAt.lte(to));
}
for row in select.limit(fetch_limit).all(self.db().read()).await? {
logs.push(JobLogDto {
source: "import".to_string(),
id: row.id,
state: row.state,
error: row.last_error,
created_at: row.created_at,
updated_at: Some(row.updated_at),
});
}
}
"search_index" => {
let mut select = index_job::Entity::find()
.filter(index_job::Column::LastError.is_not_null())
.order_by_desc(index_job::Column::UpdatedAt);
if let Some(from) = query.from {
select = select.filter(index_job::Column::CreatedAt.gte(from));
}
if let Some(to) = query.to {
select = select.filter(index_job::Column::CreatedAt.lte(to));
}
for row in select.limit(fetch_limit).all(self.db().read()).await? {
logs.push(JobLogDto {
source: "search_index".to_string(),
id: row.id,
state: row.state,
error: row.last_error,
created_at: row.created_at,
updated_at: Some(row.updated_at),
});
}
}
"chat_webhook" => {
let mut select = webhook_deliveries::Entity::find()
.filter(webhook_deliveries::Column::LastError.is_not_null())
.order_by_desc(webhook_deliveries::Column::UpdatedAt);
if let Some(from) = query.from {
select = select.filter(webhook_deliveries::Column::CreatedAt.gte(from));
}
if let Some(to) = query.to {
select = select.filter(webhook_deliveries::Column::CreatedAt.lte(to));
}
for row in select.limit(fetch_limit).all(self.db().read()).await? {
logs.push(JobLogDto {
source: "chat_webhook".to_string(),
id: row.id,
state: format!("{:?}", row.status).to_ascii_lowercase(),
error: row.last_error,
created_at: row.created_at,
updated_at: Some(row.updated_at),
});
}
}
"pipeline" => {
let mut select = workflow_run::Entity::find()
.filter(workflow_run::Column::State.eq(WorkflowRunState::Failure))
.order_by_desc(workflow_run::Column::UpdatedAt);
if let Some(from) = query.from {
select = select.filter(workflow_run::Column::CreatedAt.gte(from));
}
if let Some(to) = query.to {
select = select.filter(workflow_run::Column::CreatedAt.lte(to));
}
for row in select.limit(fetch_limit).all(self.db().read()).await? {
logs.push(JobLogDto {
source: "pipeline".to_string(),
id: row.id,
state: "failure".to_string(),
error: None,
created_at: row.created_at,
updated_at: Some(row.updated_at),
});
}
}
"ai" => {
let mut select = ai_job::Entity::find()
.filter(ai_job::Column::Error.is_not_null())
.order_by_desc(ai_job::Column::UpdatedAt);
if let Some(from) = query.from {
select = select.filter(ai_job::Column::CreatedAt.gte(from));
}
if let Some(to) = query.to {
select = select.filter(ai_job::Column::CreatedAt.lte(to));
}
for row in select.limit(fetch_limit).all(self.db().read()).await? {
logs.push(JobLogDto {
source: "ai".to_string(),
id: row.id,
state: format!("{:?}", row.status).to_ascii_lowercase(),
error: row.error,
created_at: row.created_at,
updated_at: row.updated_at,
});
}
}
"system_tasks" => {
let mut select = system_tasks::Entity::find()
.filter(system_tasks::Column::Error.is_not_null())
.order_by_desc(system_tasks::Column::CreatedAt);
if let Some(from) = query.from {
select = select.filter(system_tasks::Column::CreatedAt.gte(from));
}
if let Some(to) = query.to {
select = select.filter(system_tasks::Column::CreatedAt.lte(to));
}
for row in select.limit(fetch_limit).all(self.db().read()).await? {
logs.push(JobLogDto {
source: "system_tasks".to_string(),
id: row.id,
state: format!("{:?}", row.status).to_ascii_lowercase(),
error: row.error,
created_at: row.created_at,
updated_at: row.updated_at,
});
}
}
_ => {}
}
}
logs.sort_by_key(|log| std::cmp::Reverse(log.created_at));
let total = logs.len() as i64;
let items = logs
.into_iter()
.skip(offset as usize)
.take(limit as usize)
.collect();
Ok(JobLogPageDto {
items,
total,
limit,
offset,
})
}
}
impl AdminService {
pub async fn queue_failed_jobs(
&self,
name: &str,
limit: u64,
offset: u64,
) -> Result<JobLogPageDto, AdminError> {
let source = match name {
"search_index" => Some("search_index"),
"import" => Some("import"),
"chat_webhook" => Some("chat_webhook"),
"pipeline" => Some("pipeline"),
"system_tasks" => Some("system_tasks"),
"ai" => Some("ai"),
"git_sync" | "email" | "deletion" => None,
other => {
return Err(AdminError::BadRequest(format!("unknown queue: {other}")));
}
};
let Some(source) = source else {
return Ok(JobLogPageDto {
items: Vec::new(),
total: 0,
limit: limit.clamp(1, 100),
offset,
});
};
self.list_job_logs(ListJobLogsQuery {
source: Some(source.to_string()),
from: None,
to: None,
limit: Some(limit),
offset: Some(offset),
})
.await
}
}