yggdrasil/api/database/
tasks.rs

1#![allow(clippy::unused_unit, deprecated)]
2
3//! 备份/恢复的异步任务进度表（DashMap）。
4//!
5//! create_backup/restore_backup 立即返回 task_id，后台任务跑时通过
6//! `update` 更新进度，前端轮询 [`get_task_progress`][crate::api::database::tasks::get_task_progress]。
7//! 已完成超过 1 小时的任务惰性清理，避免内存累积。
8
9use crate::api::error::ApiError;
10use chrono::{DateTime, Utc};
11use leptos::prelude::*;
12use serde::{Deserialize, Serialize};
13
14// admin 鉴权仅 server 构建用到。
15#[cfg(feature = "server")]
16use crate::api::auth::get_current_admin_user;
17// DashMap / LazyLock 仅 server 构建持有任务进度表；WASM 端只序列化 TaskProgress。
18#[cfg(feature = "server")]
19use dashmap::DashMap;
20#[cfg(feature = "server")]
21use std::sync::LazyLock;
22
23#[derive(Clone, Serialize, Deserialize, Debug, PartialEq)]
24pub enum TaskKind {
25    Backup,
26    Restore,
27}
28
29#[derive(Clone, Serialize, Deserialize, Debug, PartialEq)]
30pub enum TaskStatus {
31    Running,
32    Done,
33    Failed,
34}
35
36#[derive(Clone, Serialize, Deserialize, Debug)]
37pub struct TaskProgress {
38    pub id: String,
39    pub kind: TaskKind,
40    pub stage: String,
41    pub percent: u8,
42    pub detail: Option<String>,
43    pub status: TaskStatus,
44    pub error: Option<String>,
45    #[cfg_attr(
46        target_arch = "wasm32",
47        serde(deserialize_with = "crate::models::rfc3339::deserialize")
48    )]
49    pub created_at: DateTime<Utc>,
50    /// 完成后的备份文件名（下载/恢复用）。
51    pub result_filename: Option<String>,
52}
53
54#[cfg(feature = "server")]
55static TASKS: LazyLock<DashMap<String, TaskProgress>> = LazyLock::new(DashMap::new);
56
57/// 注册新任务（初始 Running，0%）。
58///
59/// pub(crate)：定时备份调度任务（[`crate::tasks::backup`]）也需要注册，
60/// 让 admin 打开面板时能看到进行中的自动备份。
61#[cfg(feature = "server")]
62pub(crate) fn insert(id: String, kind: TaskKind) {
63    TASKS.insert(
64        id.clone(),
65        TaskProgress {
66            id,
67            kind,
68            stage: "排队中".to_string(),
69            percent: 0,
70            detail: None,
71            status: TaskStatus::Running,
72            error: None,
73            created_at: Utc::now(),
74            result_filename: None,
75        },
76    );
77}
78
79/// 更新任务进度（后台任务调用）。
80#[cfg(feature = "server")]
81pub(super) fn update(
82    id: &str,
83    stage: &str,
84    percent: u8,
85    status: TaskStatus,
86    detail: Option<String>,
87    error: Option<String>,
88    result_filename: Option<String>,
89) {
90    if let Some(mut p) = TASKS.get_mut(id) {
91        p.stage = stage.to_string();
92        p.percent = percent;
93        p.status = status;
94        p.detail = detail;
95        p.error = error;
96        p.result_filename = result_filename;
97    }
98}
99
100/// 惰性清理已完成超过 1 小时的任务（查询时顺便清）。
101#[cfg(feature = "server")]
102fn gc_old() {
103    let cutoff = Utc::now() - chrono::Duration::hours(1);
104    TASKS.retain(|_, p| {
105        !(matches!(p.status, TaskStatus::Done | TaskStatus::Failed) && p.created_at < cutoff)
106    });
107}
108
109/// 查询任务进度（轮询用）。
110#[server(GetTaskProgress, endpoint = "get_task_progress", input = leptos::server_fn::codec::Json)]
111pub async fn get_task_progress(task_id: String) -> Result<TaskProgress, ApiError> {
112    let _user = get_current_admin_user().await?;
113    #[cfg(feature = "server")]
114    {
115        gc_old();
116        TASKS
117            .get(&task_id)
118            .map(|p| p.clone())
119            .ok_or_else(|| crate::api::error::AppError::NotFound("任务不存在").into())
120    }
121    #[cfg(not(feature = "server"))]
122    {
123        Ok(TaskProgress {
124            id: task_id,
125            kind: TaskKind::Backup,
126            stage: String::new(),
127            percent: 0,
128            detail: None,
129            status: TaskStatus::Done,
130            error: None,
131            created_at: Utc::now(),
132            result_filename: None,
133        })
134    }
135}
yggdrasil/api/database/tasks.rs

yggdrasil/api/database/
tasks.rs
