563bb99
use models::admin::admin_audit_log;
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, Condition, EntityTrait, PaginatorTrait, QueryFilter, QueryOrder, QuerySelect,
Select,
};
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 ListAuditQuery {
pub admin_id: Option<Uuid>,
pub admin_username: Option<String>,
pub action: Option<String>,
pub target_type: Option<String>,
pub target_id: 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 AuditLogDto {
pub id: Uuid,
pub admin_id: Option<Uuid>,
pub admin_username: String,
pub action: String,
pub target_type: Option<String>,
pub target_id: Option<String>,
pub detail: serde_json::Value,
pub ip: Option<String>,
pub user_agent: Option<String>,
pub request_id: Option<String>,
pub created_at: OffsetDateTime,
}
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct AuditLogPageDto {
pub items: Vec<AuditLogDto>,
pub total: i64,
pub limit: u64,
pub offset: u64,
}
#[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_audit_logs(
&self,
query: ListAuditQuery,
) -> Result<AuditLogPageDto, AdminError> {
let limit = query.limit.unwrap_or(50).clamp(1, 100);
let offset = query.offset.unwrap_or(0);
let select = self.audit_select(&query).await?;
let total = select.clone().count(self.db().read()).await? as i64;
let rows = select
.limit(limit)
.offset(offset)
.all(self.db().read())
.await?;
let items = rows
.into_iter()
.map(|row| AuditLogDto {
id: row.id,
admin_id: row.admin_id,
admin_username: row.admin_username,
action: row.action,
target_type: row.target_type,
target_id: row.target_id,
detail: row.detail,
ip: row.ip,
user_agent: row.user_agent,
request_id: row.request_id,
created_at: row.created_at,
})
.collect();
Ok(AuditLogPageDto {
items,
total,
limit,
offset,
})
}
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(|left, right| right.created_at.cmp(&left.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 {
async fn audit_select(
&self,
query: &ListAuditQuery,
) -> Result<Select<admin_audit_log::Entity>, AdminError> {
let mut select = admin_audit_log::Entity::find();
if let Some(admin_id) = query.admin_id {
select = select.filter(admin_audit_log::Column::AdminId.eq(admin_id));
}
if let Some(username) = query
.admin_username
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
{
select = select
.filter(admin_audit_log::Column::AdminUsername.eq(username.to_ascii_lowercase()));
}
if let Some(action) = query
.action
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
{
select = select
.filter(admin_audit_log::Column::Action.contains(action.to_ascii_lowercase()));
}
if let Some(target_type) = query
.target_type
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
{
select = select.filter(admin_audit_log::Column::TargetType.eq(target_type));
}
if let Some(target_id) = query
.target_id
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
{
select = select.filter(admin_audit_log::Column::TargetId.eq(target_id));
}
if let Some(q) = query
.q
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
{
let needle = q.to_ascii_lowercase();
select = select.filter(
Condition::any()
.add(admin_audit_log::Column::AdminUsername.contains(needle.clone()))
.add(admin_audit_log::Column::Action.contains(needle.clone()))
.add(admin_audit_log::Column::TargetId.contains(q))
.add(sea_orm::sea_query::Expr::cust_with_values(
"detail::text ILIKE $1",
[format!("%{needle}%")],
)),
);
}
if let Some(from) = query.from {
select = select.filter(admin_audit_log::Column::CreatedAt.gte(from));
}
if let Some(to) = query.to {
select = select.filter(admin_audit_log::Column::CreatedAt.lte(to));
}
Ok(select.order_by_desc(admin_audit_log::Column::CreatedAt))
}
pub async fn export_audit_logs(
&self,
query: ListAuditQuery,
) -> Result<Vec<AuditLogDto>, AdminError> {
let rows = self
.audit_select(&query)
.await?
.limit(10000)
.all(self.db().read())
.await?;
Ok(rows
.into_iter()
.map(|row| AuditLogDto {
id: row.id,
admin_id: row.admin_id,
admin_username: row.admin_username,
action: row.action,
target_type: row.target_type,
target_id: row.target_id,
detail: row.detail,
ip: row.ip,
user_agent: row.user_agent,
request_id: row.request_id,
created_at: row.created_at,
})
.collect())
}
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
}
}