task_pilot/backend/src/tasks/service.rs

use std::{collections::BTreeMap, str::FromStr};

use axum::http::{HeaderName, HeaderValue};
use chrono::Utc;
use chrono_tz::Tz;
use croner::Cron;
use sqlx::SqlitePool;
use url::Url;

use crate::python_env::validate_configured_venv;
use crate::{error::ApiResult, tasks::model::*};

use super::repository::{self, ListFilters};

const MAX_TASK_ID_LEN: usize = 64;
const MAX_TASK_NAME_LEN: usize = 120;
const MAX_TIMEOUT_SECS: u64 = 86_400;
const MAX_ARGS: usize = 256;
const MAX_ARG_LEN: usize = 2_048;
const MAX_ENV_ENTRIES: usize = 128;
const MAX_ENV_KEY_LEN: usize = 128;
const MAX_ENV_VALUE_LEN: usize = 8_192;
const MAX_HTTP_BODY_LEN: usize = 1_048_576;
const DEFAULT_PAGE_SIZE: u32 = 20;
const MAX_PAGE_SIZE: u32 = 100;

pub async fn create(pool: &SqlitePool, mut request: CreateTaskRequest) -> ApiResult<TaskResponse> {
    request.name = request.name.trim().to_owned();
    request.cron = request.cron.trim().to_owned();
    request.timezone = request.timezone.trim().to_owned();
    normalize_action(&mut request.action);

    validate_task_id(&request.id)?;
    validate_common(
        &request.name,
        &request.cron,
        &request.timezone,
        request.timeout_secs,
        &request.retry_policy,
    )?;
    validate_action(&request.action)?;

    repository::create(pool, &request).await
}

pub async fn get(pool: &SqlitePool, id: &str) -> ApiResult<TaskResponse> {
    repository::get(pool, id)
        .await?
        .ok_or_else(|| crate::error::ApiError::not_found(format!("任务 {id} 不存在")))
}

pub async fn update(
    pool: &SqlitePool,
    id: &str,
    mut request: UpdateTaskRequest,
) -> ApiResult<TaskResponse> {
    validate_task_id(id)?;
    request.name = request.name.trim().to_owned();
    request.cron = request.cron.trim().to_owned();
    request.timezone = request.timezone.trim().to_owned();
    normalize_action(&mut request.action);

    validate_common(
        &request.name,
        &request.cron,
        &request.timezone,
        request.timeout_secs,
        &request.retry_policy,
    )?;
    validate_action(&request.action)?;

    repository::update(pool, id, &request).await
}

pub async fn delete(pool: &SqlitePool, id: &str) -> ApiResult<()> {
    if repository::delete(pool, id).await? {
        Ok(())
    } else {
        Err(crate::error::ApiError::not_found(format!(
            "任务 {id} 不存在"
        )))
    }
}

pub async fn set_enabled(pool: &SqlitePool, id: &str, enabled: bool) -> ApiResult<TaskResponse> {
    if !repository::set_enabled(pool, id, enabled).await? {
        return Err(crate::error::ApiError::not_found(format!(
            "任务 {id} 不存在"
        )));
    }
    get(pool, id).await
}

pub async fn list(pool: &SqlitePool, query: ListTasksQuery) -> ApiResult<TaskListResponse> {
    let page = query.page.unwrap_or(1).max(1);
    let page_size = query
        .page_size
        .unwrap_or(DEFAULT_PAGE_SIZE)
        .clamp(1, MAX_PAGE_SIZE);
    let filters = ListFilters {
        q: query
            .q
            .map(|value| value.trim().to_owned())
            .filter(|value| !value.is_empty()),
        enabled: query.enabled,
        action_type: query.action_type,
        page,
        page_size,
    };

    repository::list(pool, &filters).await
}

fn validate_task_id(id: &str) -> ApiResult<()> {
    if id.is_empty() || id.len() > MAX_TASK_ID_LEN {
        return Err(crate::error::ApiError::validation(
            "id",
            format!("任务标识长度必须在 1 到 {MAX_TASK_ID_LEN} 个字符之间"),
        ));
    }
    if !id
        .chars()
        .all(|character| character.is_ascii_alphanumeric() || matches!(character, '_' | '-'))
    {
        return Err(crate::error::ApiError::validation(
            "id",
            "任务标识只能包含 ASCII 字母、数字、下划线和连字符",
        ));
    }

    Ok(())
}

fn validate_common(
    name: &str,
    cron_expression: &str,
    timezone: &str,
    timeout_secs: u64,
    retry_policy: &RetryPolicy,
) -> ApiResult<()> {
    if name.is_empty() || name.chars().count() > MAX_TASK_NAME_LEN {
        return Err(crate::error::ApiError::validation(
            "name",
            format!("任务名称长度必须在 1 到 {MAX_TASK_NAME_LEN} 个字符之间"),
        ));
    }
    if timeout_secs == 0 || timeout_secs > MAX_TIMEOUT_SECS {
        return Err(crate::error::ApiError::validation(
            "timeout_secs",
            format!("超时时间必须在 1 到 {MAX_TIMEOUT_SECS} 秒之间"),
        ));
    }
    if retry_policy.max_attempts > 10 {
        return Err(crate::error::ApiError::validation(
            "retry_policy.max_attempts",
            "最大重试次数不能超过 10",
        ));
    }
    if retry_policy.delay_secs > 3_600 {
        return Err(crate::error::ApiError::validation(
            "retry_policy.delay_secs",
            "重试间隔不能超过 3600 秒",
        ));
    }

    Tz::from_str(timezone).map_err(|_| {
        crate::error::ApiError::validation("timezone", "时区必须是有效的 IANA 时区名称")
    })?;

    let cron = Cron::from_str(cron_expression).map_err(|error| {
        tracing::debug!(error = %error, "invalid cron expression");
        let message =
            crate::cron::CronPreviewError::InvalidExpression(error.to_string()).user_message();
        crate::error::ApiError::validation("cron", message)
    })?;
    cron.find_next_occurrence(&Utc::now(), false)
        .map_err(|error| {
            tracing::debug!(error = %error, "cron has no future occurrence");
            crate::error::ApiError::validation("cron", "Cron 表达式没有可计算的未来执行时间")
        })?;

    Ok(())
}

fn normalize_action(action: &mut TaskAction) {
    if let TaskAction::Script {
        venv, interpreter, ..
    } = action
    {
        if interpreter.trim().is_empty() {
            *interpreter = "python.exe".to_owned();
        }
        if venv.as_deref().is_some_and(|value| value.trim().is_empty()) {
            *venv = None;
        }
    }
}

fn validate_action(action: &TaskAction) -> ApiResult<()> {
    match action {
        TaskAction::Script {
            working_dir,
            venv,
            interpreter,
            script,
            args,
            env,
        } => {
            validate_required("action.working_dir", working_dir)?;
            if let Some(venv) = venv {
                validate_required("action.venv", venv)?;
                validate_configured_venv(working_dir, venv).map_err(|error| {
                    crate::error::ApiError::validation(
                        "action.venv",
                        format!("虚拟环境无效: {error}"),
                    )
                })?;
            }
            validate_required("action.interpreter", interpreter)?;
            validate_required("action.script", script)?;
            validate_args(args)?;
            validate_env(env)?;
        }
        TaskAction::Http {
            url,
            headers,
            body,
            expected_status,
            ..
        } => {
            let parsed_url = Url::parse(url).map_err(|error| {
                crate::error::ApiError::validation("action.url", format!("URL 无效: {error}"))
            })?;
            if !matches!(parsed_url.scheme(), "http" | "https") || parsed_url.host_str().is_none() {
                return Err(crate::error::ApiError::validation(
                    "action.url",
                    "URL 必须使用 http 或 https 协议并包含主机名",
                ));
            }
            validate_headers(headers)?;
            if body
                .as_ref()
                .is_some_and(|value| value.len() > MAX_HTTP_BODY_LEN)
            {
                return Err(crate::error::ApiError::validation(
                    "action.body",
                    "HTTP Body 不能超过 1 MiB",
                ));
            }
            if expected_status
                .iter()
                .any(|status| !(100..=599).contains(status))
            {
                return Err(crate::error::ApiError::validation(
                    "action.expected_status",
                    "期望状态码必须在 100 到 599 之间",
                ));
            }
        }
        TaskAction::Process {
            executable,
            working_dir,
            args,
            env,
        } => {
            validate_required("action.executable", executable)?;
            if let Some(working_dir) = working_dir {
                validate_required("action.working_dir", working_dir)?;
            }
            validate_args(args)?;
            validate_env(env)?;
        }
    }

    Ok(())
}

fn validate_required(field: &str, value: &str) -> ApiResult<()> {
    if value.trim().is_empty() {
        Err(crate::error::ApiError::validation(field, "该字段不能为空"))
    } else {
        Ok(())
    }
}

fn validate_args(args: &[String]) -> ApiResult<()> {
    if args.len() > MAX_ARGS {
        return Err(crate::error::ApiError::validation(
            "action.args",
            format!("参数数量不能超过 {MAX_ARGS} 个"),
        ));
    }
    if args.iter().any(|argument| argument.len() > MAX_ARG_LEN) {
        return Err(crate::error::ApiError::validation(
            "action.args",
            format!("单个参数不能超过 {MAX_ARG_LEN} 个字符"),
        ));
    }

    Ok(())
}

fn validate_env(env: &BTreeMap<String, String>) -> ApiResult<()> {
    if env.len() > MAX_ENV_ENTRIES {
        return Err(crate::error::ApiError::validation(
            "action.env",
            format!("环境变量数量不能超过 {MAX_ENV_ENTRIES} 个"),
        ));
    }

    for (key, value) in env {
        if key.is_empty()
            || key.len() > MAX_ENV_KEY_LEN
            || key.contains(['=', '\0'])
            || value.len() > MAX_ENV_VALUE_LEN
            || value.contains('\0')
        {
            return Err(crate::error::ApiError::validation(
                "action.env",
                "环境变量名称或值无效",
            ));
        }
    }

    Ok(())
}

fn validate_headers(headers: &BTreeMap<String, String>) -> ApiResult<()> {
    for (name, value) in headers {
        HeaderName::from_bytes(name.as_bytes()).map_err(|_| {
            crate::error::ApiError::validation("action.headers", "HTTP Header 名称无效")
        })?;
        HeaderValue::from_str(value).map_err(|_| {
            crate::error::ApiError::validation("action.headers", "HTTP Header 值无效")
        })?;
    }

    Ok(())
}

#[cfg(test)]
mod tests {
    use axum::response::IntoResponse;

    use super::*;

    fn script_action() -> TaskAction {
        TaskAction::Script {
            working_dir: "D:\\Projects\\sync".to_owned(),
            venv: None,
            interpreter: "python.exe".to_owned(),
            script: "main.py".to_owned(),
            args: vec!["--mode".to_owned(), "full".to_owned()],
            env: BTreeMap::new(),
        }
    }

    #[test]
    fn accepts_valid_task_fields() {
        validate_task_id("db_backup_daily").expect("task id should be valid");
        validate_common(
            "每日数据库备份",
            "0 0 2 * * *",
            "Asia/Shanghai",
            300,
            &RetryPolicy::default(),
        )
        .expect("task fields should be valid");
        validate_action(&script_action()).expect("script action should be valid");
    }

    #[test]
    fn rejects_invalid_cron() {
        let error = validate_common(
            "Invalid cron",
            "*/0 * * * * *",
            "Asia/Shanghai",
            300,
            &RetryPolicy::default(),
        )
        .expect_err("invalid cron should fail");

        assert_eq!(error.into_response().status(), 400);
    }

    #[test]
    fn rejects_empty_script_path() {
        let action = TaskAction::Script {
            working_dir: "D:\\Projects\\sync".to_owned(),
            venv: None,
            interpreter: "python.exe".to_owned(),
            script: "  ".to_owned(),
            args: Vec::new(),
            env: BTreeMap::new(),
        };

        validate_action(&action).expect_err("empty script path should fail");
    }
}