task_pilot/backend/src/api.rs

use axum::{
    Json, Router,
    extract::{DefaultBodyLimit, State},
    http::{HeaderValue, Method, header},
    middleware,
    response::IntoResponse,
    routing::any,
    routing::get,
};
use serde::Serialize;
use tower_http::cors::CorsLayer;
use utoipa::{OpenApi, ToSchema};

use crate::{
    cron::{self, CronPreviewRequest, CronPreviewResponse},
    directory_browser::{
        self, DirectoryEntry, DirectoryError, DirectoryListingResponse, ListDirectoriesQuery,
    },
    error::{ApiError, ApiJson, ApiQuery, ApiResult, ErrorResponse},
    python_env::{
        self, InterpreterKind, VenvProbeRequest, VenvProbeResponse, VenvProbeStatus, VenvSource,
    },
    runs::{
        model::{LogChunkResponse, TaskRunListResponse, TaskRunResponse},
        routes::{
            __path_cancel_run, __path_download_run_logs, __path_execution_events, __path_get_run,
            __path_get_run_logs, __path_list_task_runs, __path_run_task,
        },
    },
    security,
    state::AppState,
    tasks::{
        model::{
            ConcurrencyPolicy, CreateTaskRequest, HttpMethod, RetryPolicy, ScheduledTaskResponse,
            TaskAction, TaskActionType, TaskListResponse, TaskResponse, TaskStatsResponse,
            UpdateTaskRequest,
        },
        routes::{
            __path_create_task, __path_delete_task, __path_disable_task, __path_enable_task,
            __path_get_task, __path_list_tasks, __path_update_task,
        },
    },
};

#[derive(Debug, Serialize, ToSchema)]
pub struct HealthResponse {
    /// Current service status.
    pub status: String,
    /// Stable service identifier.
    pub service: String,
    /// Backend package version.
    pub version: String,
}

#[derive(Debug, Serialize, ToSchema)]
pub struct SessionResponse {
    /// Random per-process token required for mutating browser requests.
    pub token: String,
}

#[utoipa::path(
    get,
    path = "/api/v1/health",
    tag = "system",
    responses(
        (status = 200, description = "TaskPilot API is healthy", body = HealthResponse)
    )
)]
pub async fn health() -> Json<HealthResponse> {
    Json(HealthResponse {
        status: "ok".to_owned(),
        service: "task-pilot-backend".to_owned(),
        version: env!("CARGO_PKG_VERSION").to_owned(),
    })
}

#[utoipa::path(
    get,
    path = "/api/v1/session",
    tag = "system",
    responses(
        (status = 200, description = "Local API session token", body = SessionResponse)
    )
)]
pub async fn session(State(state): State<AppState>) -> Json<SessionResponse> {
    Json(SessionResponse {
        token: state.auth_token().to_owned(),
    })
}

#[utoipa::path(
    post,
    path = "/api/v1/cron/preview",
    tag = "system",
    request_body = CronPreviewRequest,
    responses(
        (status = 200, description = "Upcoming cron occurrences", body = CronPreviewResponse),
        (status = 400, description = "Invalid cron preview request", body = ErrorResponse)
    )
)]
pub async fn preview_cron(
    ApiJson(request): ApiJson<CronPreviewRequest>,
) -> ApiResult<Json<CronPreviewResponse>> {
    if request.count == 0 || request.count > 10 {
        return Err(ApiError::validation("count", "预览次数必须在 1 到 10 之间"));
    }
    let preview = cron::preview_in_timezone(
        &request.expression,
        &request.timezone,
        chrono::Utc::now(),
        request.count,
    )
    .map_err(|error| ApiError::validation("expression", error.to_string()))?;
    Ok(Json(preview))
}

#[utoipa::path(
    post,
    path = "/api/v1/python/venv",
    tag = "system",
    request_body = VenvProbeRequest,
    responses(
        (status = 200, description = "Detected Python virtual environment", body = VenvProbeResponse),
        (status = 400, description = "Invalid venv probe request", body = ErrorResponse)
    )
)]
pub async fn probe_python_venv(
    ApiJson(request): ApiJson<VenvProbeRequest>,
) -> ApiResult<Json<VenvProbeResponse>> {
    tokio::task::spawn_blocking(move || python_env::probe_venv(&request))
        .await
        .map(Json)
        .map_err(ApiError::internal)
}

#[utoipa::path(
    get,
    path = "/api/v1/system/directories",
    tag = "system",
    params(ListDirectoriesQuery),
    responses(
        (status = 200, description = "Subdirectories of the requested path, or the drive list", body = DirectoryListingResponse),
        (status = 400, description = "The path is not an existing absolute directory", body = ErrorResponse)
    )
)]
pub async fn list_directories(
    ApiQuery(query): ApiQuery<ListDirectoriesQuery>,
) -> ApiResult<Json<DirectoryListingResponse>> {
    tokio::task::spawn_blocking(move || directory_browser::list_directories(&query))
        .await
        .map_err(ApiError::internal)?
        .map(Json)
        .map_err(|error| match error {
            DirectoryError::RelativePath(message) => {
                ApiError::bad_request("DIRECTORY_PATH_INVALID", message)
            }
            DirectoryError::NotFound(message) => {
                ApiError::bad_request("DIRECTORY_NOT_FOUND", message)
            }
            DirectoryError::Unreadable(message) => {
                ApiError::bad_request("DIRECTORY_UNREADABLE", message)
            }
        })
}

#[utoipa::path(
    get,
    path = "/api/v1/stats",
    tag = "tasks",
    responses((status = 200, description = "Dashboard counters", body = TaskStatsResponse))
)]
pub async fn task_stats(State(state): State<AppState>) -> ApiResult<Json<TaskStatsResponse>> {
    let counts = crate::tasks::repository::counts(&state.pool).await?;
    Ok(Json(TaskStatsResponse {
        total: counts.total,
        enabled: counts.enabled,
        running: u64::try_from(state.executor.running_count()).unwrap_or_default(),
        script: counts.script,
        http: counts.http,
        process: counts.process,
    }))
}

#[utoipa::path(
    get,
    path = "/api/v1/schedule",
    tag = "tasks",
    responses(
        (status = 200, description = "Next trigger time per enabled task", body = Vec<ScheduledTaskResponse>)
    )
)]
pub async fn list_schedule(
    State(state): State<AppState>,
) -> ApiResult<Json<Vec<ScheduledTaskResponse>>> {
    let tasks = crate::tasks::repository::list_enabled_tasks(&state.pool).await?;
    Ok(Json(
        tasks
            .into_iter()
            .map(|task| ScheduledTaskResponse {
                next_run_at: state.scheduler.next_run_at(&task.id),
                task_id: task.id,
            })
            .collect(),
    ))
}

#[derive(OpenApi)]
#[openapi(
    info(
        title = "TaskPilot API",
        version = env!("CARGO_PKG_VERSION"),
        description = "Local task scheduling and process management API."
    ),
    paths(
        health,
        session,
        preview_cron,
        probe_python_venv,
        list_directories,
        task_stats,
        list_schedule,
        list_tasks,
        create_task,
        get_task,
        update_task,
        delete_task,
        enable_task,
        disable_task,
        run_task,
        list_task_runs,
        get_run,
        get_run_logs,
        cancel_run,
        download_run_logs,
        execution_events
    ),
    components(schemas(
        HealthResponse,
        SessionResponse,
        CronPreviewRequest,
        CronPreviewResponse,
        VenvProbeRequest,
        VenvProbeResponse,
        VenvProbeStatus,
        VenvSource,
        InterpreterKind,
        DirectoryEntry,
        DirectoryListingResponse,
        TaskStatsResponse,
        ScheduledTaskResponse,
        ErrorResponse,
        TaskActionType,
        HttpMethod,
        ConcurrencyPolicy,
        RetryPolicy,
        TaskAction,
        CreateTaskRequest,
        UpdateTaskRequest,
        TaskResponse,
        TaskListResponse,
        TaskRunResponse,
        TaskRunListResponse,
        LogChunkResponse
    ))
)]
pub struct ApiDoc;

pub fn openapi() -> utoipa::openapi::OpenApi {
    ApiDoc::openapi()
}

async fn openapi_json() -> Json<utoipa::openapi::OpenApi> {
    Json(openapi())
}

pub fn router(state: AppState) -> Router {
    let allowed_origins = [
        HeaderValue::from_static("http://127.0.0.1:5173"),
        HeaderValue::from_static("http://localhost:5173"),
    ];
    let cors = CorsLayer::new()
        .allow_origin(allowed_origins)
        .allow_methods([
            Method::GET,
            Method::POST,
            Method::PUT,
            Method::DELETE,
            Method::OPTIONS,
        ])
        .allow_headers([header::ACCEPT, header::CONTENT_TYPE]);

    let api = Router::new()
        .route("/api/v1/health", get(health))
        .route("/api/v1/session", get(session))
        .route("/api/v1/cron/preview", axum::routing::post(preview_cron))
        .route(
            "/api/v1/python/venv",
            axum::routing::post(probe_python_venv),
        )
        .route("/api/v1/system/directories", get(list_directories))
        .route("/api/v1/stats", get(task_stats))
        .route("/api/v1/schedule", get(list_schedule))
        .route("/api/v1/openapi.json", get(openapi_json))
        .merge(crate::tasks::routes::router())
        .merge(crate::runs::routes::router())
        .route("/api/{*path}", any(api_not_found))
        // 本地工具也不需要接受超大请求体；任务载荷与日志上传都用不到 2 MiB 以上。
        .layer(DefaultBodyLimit::max(2 * 1024 * 1024))
        .layer(cors)
        .route_layer(middleware::from_fn_with_state(
            state.clone(),
            security::api_guard,
        ));

    api.with_state(state)
}

async fn api_not_found() -> impl IntoResponse {
    ApiError::route_not_found("API 路由不存在")
}