use chrono::{DateTime, Utc}; use sqlx::{FromRow, QueryBuilder, Sqlite, SqlitePool}; use crate::{ error::{ApiError, ApiResult}, tasks::model::{ CreateTaskRequest, TaskAction, TaskActionType, TaskListResponse, TaskResponse, UpdateTaskRequest, }, }; const TASK_COLUMNS: &str = " id, name, description, cron, timezone, enabled, timeout_secs, action_type, action_json, created_at, updated_at "; #[derive(Debug, Clone, Copy, Default)] pub struct TaskCounts { pub total: u64, pub enabled: u64, pub script: u64, pub http: u64, pub process: u64, } /// 单条 SQL 统计全部指标,供仪表盘使用。 pub async fn counts(pool: &SqlitePool) -> ApiResult { let (total, enabled, script, http, process): (i64, i64, i64, i64, i64) = sqlx::query_as( "SELECT COUNT(*), COALESCE(SUM(enabled), 0), COALESCE(SUM(action_type = 'script'), 0), COALESCE(SUM(action_type = 'http'), 0), COALESCE(SUM(action_type = 'process'), 0) FROM tasks", ) .fetch_one(pool) .await?; Ok(TaskCounts { total: u64::try_from(total).unwrap_or_default(), enabled: u64::try_from(enabled).unwrap_or_default(), script: u64::try_from(script).unwrap_or_default(), http: u64::try_from(http).unwrap_or_default(), process: u64::try_from(process).unwrap_or_default(), }) } pub struct ListFilters { pub q: Option, pub enabled: Option, pub action_type: Option, pub page: u32, pub page_size: u32, } #[derive(Debug, FromRow)] struct TaskRow { id: String, name: String, description: String, cron: String, timezone: String, enabled: bool, timeout_secs: i64, action_type: String, action_json: String, created_at: DateTime, updated_at: DateTime, } impl TryFrom for TaskResponse { type Error = ApiError; fn try_from(row: TaskRow) -> Result { let action: TaskAction = serde_json::from_str(&row.action_json)?; if row.action_type != action.action_type().as_str() { return Err(ApiError::internal(format!( "task {} action type does not match action payload", row.id ))); } Ok(Self { id: row.id, name: row.name, description: row.description, cron: row.cron, timezone: row.timezone, enabled: row.enabled, timeout_secs: u64::try_from(row.timeout_secs).map_err(ApiError::internal)?, action, created_at: row.created_at, updated_at: row.updated_at, }) } } pub async fn create(pool: &SqlitePool, request: &CreateTaskRequest) -> ApiResult { let now = Utc::now(); let action_json = serde_json::to_string(&request.action)?; let result = sqlx::query( "INSERT INTO tasks ( id, name, description, cron, timezone, enabled, timeout_secs, action_type, action_json, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", ) .bind(&request.id) .bind(&request.name) .bind(&request.description) .bind(&request.cron) .bind(&request.timezone) .bind(request.enabled) .bind(request.timeout_secs as i64) .bind(request.action.action_type().as_str()) .bind(action_json) .bind(now) .bind(now) .execute(pool) .await; if let Err(error) = result { if is_unique_violation(&error) { return Err(ApiError::conflict(format!( "任务标识 {} 已存在", request.id ))); } return Err(error.into()); } Ok(TaskResponse { id: request.id.clone(), name: request.name.clone(), description: request.description.clone(), cron: request.cron.clone(), timezone: request.timezone.clone(), enabled: request.enabled, timeout_secs: request.timeout_secs, action: request.action.clone(), created_at: now, updated_at: now, }) } pub async fn get(pool: &SqlitePool, id: &str) -> ApiResult> { let row = sqlx::query_as::<_, TaskRow>( "SELECT id, name, description, cron, timezone, enabled, timeout_secs, action_type, action_json, created_at, updated_at FROM tasks WHERE id = ?", ) .bind(id) .fetch_optional(pool) .await?; row.map(TaskResponse::try_from).transpose() } pub async fn update( pool: &SqlitePool, id: &str, request: &UpdateTaskRequest, ) -> ApiResult { let now = Utc::now(); let action_json = serde_json::to_string(&request.action)?; let result = sqlx::query( "UPDATE tasks SET name = ?, description = ?, cron = ?, timezone = ?, enabled = ?, timeout_secs = ?, action_type = ?, action_json = ?, updated_at = ? WHERE id = ?", ) .bind(&request.name) .bind(&request.description) .bind(&request.cron) .bind(&request.timezone) .bind(request.enabled) .bind(request.timeout_secs as i64) .bind(request.action.action_type().as_str()) .bind(action_json) .bind(now) .bind(id) .execute(pool) .await?; if result.rows_affected() == 0 { return Err(ApiError::not_found(format!("任务 {id} 不存在"))); } get(pool, id) .await? .ok_or_else(|| ApiError::not_found(format!("任务 {id} 不存在"))) } pub async fn delete(pool: &SqlitePool, id: &str) -> ApiResult { let result = sqlx::query("DELETE FROM tasks WHERE id = ?") .bind(id) .execute(pool) .await?; Ok(result.rows_affected() > 0) } pub async fn set_enabled(pool: &SqlitePool, id: &str, enabled: bool) -> ApiResult { let result = sqlx::query("UPDATE tasks SET enabled = ?, updated_at = ? WHERE id = ?") .bind(enabled) .bind(Utc::now()) .bind(id) .execute(pool) .await?; Ok(result.rows_affected() > 0) } pub async fn list(pool: &SqlitePool, filters: &ListFilters) -> ApiResult { let mut count_builder = QueryBuilder::::new("SELECT COUNT(*) FROM tasks"); apply_filters(&mut count_builder, filters); let total: i64 = count_builder.build_query_scalar().fetch_one(pool).await?; let offset = i64::from(filters.page.saturating_sub(1)) * i64::from(filters.page_size); let mut list_builder = QueryBuilder::::new(format!("SELECT {TASK_COLUMNS} FROM tasks")); apply_filters(&mut list_builder, filters); list_builder .push(" ORDER BY updated_at DESC, id ASC LIMIT ") .push_bind(i64::from(filters.page_size)) .push(" OFFSET ") .push_bind(offset); let rows = list_builder .build_query_as::() .fetch_all(pool) .await?; let items = rows .into_iter() .map(TaskResponse::try_from) .collect::>>()?; Ok(TaskListResponse { items, page: filters.page, page_size: filters.page_size, total: u64::try_from(total).map_err(ApiError::internal)?, }) } /// 启动时恢复调度用的列表:所有启用任务都参与调度,不再只限脚本。 pub async fn list_enabled_tasks(pool: &SqlitePool) -> ApiResult> { let rows = sqlx::query_as::<_, TaskRow>( "SELECT id, name, description, cron, timezone, enabled, timeout_secs, action_type, action_json, created_at, updated_at FROM tasks WHERE enabled = 1 ORDER BY updated_at DESC, id ASC", ) .fetch_all(pool) .await?; rows.into_iter() .map(TaskResponse::try_from) .collect::>>() } fn apply_filters(builder: &mut QueryBuilder, filters: &ListFilters) { builder.push(" WHERE 1 = 1"); if let Some(search) = &filters.q { let pattern = format!( "%{}%", search .replace('\\', "\\\\") .replace('%', "\\%") .replace('_', "\\_") ); builder .push(" AND (id LIKE ") .push_bind(pattern.clone()) .push(" ESCAPE '\\' OR name LIKE ") .push_bind(pattern) .push(" ESCAPE '\\')"); } if let Some(enabled) = filters.enabled { builder.push(" AND enabled = ").push_bind(enabled); } if let Some(action_type) = filters.action_type { builder .push(" AND action_type = ") .push_bind(action_type.as_str()); } } fn is_unique_violation(error: &sqlx::Error) -> bool { matches!(error, sqlx::Error::Database(database_error) if database_error.is_unique_violation()) }