bizi-server 0.5.1

Task runner server for the bizi project
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108

use axum::routing::post;
use axum::{Router, routing::get};
use sea_orm::DatabaseConnection;
use std::{
    collections::{HashMap, HashSet},
    sync::Arc,
};
use tokio::sync::{Mutex, broadcast, oneshot};
use utoipa::OpenApi;
use utoipa_swagger_ui::SwaggerUi;

use crate::api::error::ErrorResponse;
use crate::api::tasks::{
    CancelTaskRequest, CancelTaskResponse, CancelTaskResponseBody, GetTaskRunLogsRequest,
    GetTaskRunLogsResponse, GetTaskRunLogsResponseBody, GetTaskRunResponse, GetTaskRunResponseBody,
    ListTaskRunsRequest, ListTaskRunsResponse, ListTaskRunsResponseBody, ListTasksRequest,
    ListTasksResponse, ListTasksResponseBody, RestartTaskRequest, RestartTaskResponse,
    RestartTaskResponseBody, StartTaskRequest, StartTaskResponse, StartTaskResponseBody,
    TaskRunLogLine, TaskRunTreeNode, cancel_task, get_task_run, get_task_run_logs, list_task_runs,
    list_tasks, restart_task, run_task,
};
use crate::config::Task;
use crate::db::entities::task_run::TaskRunStatus;

pub mod error;
pub mod tasks;

#[derive(Clone)]
pub struct AppState {
    pub db: DatabaseConnection,
    pub task_events: broadcast::Sender<tasks::TaskRunStatusChangedEvent>,
    pub task_log_events: broadcast::Sender<tasks::TaskRunLogLine>,
    pub running_processes: Arc<Mutex<HashMap<String, RunningProcessEntry>>>,
    pub run_include_tasks: Arc<Mutex<HashMap<String, HashSet<String>>>>,
}

pub struct RunningProcessEntry {
    pub execution_id: String,
    pub cancel_tx: oneshot::Sender<()>,
}

pub fn create_app_state(db: DatabaseConnection) -> AppState {
    let (task_events, _) = broadcast::channel(256);
    let (task_log_events, _) = broadcast::channel(2048);
    AppState {
        db,
        task_events,
        task_log_events,
        running_processes: Arc::new(Mutex::new(HashMap::new())),
        run_include_tasks: Arc::new(Mutex::new(HashMap::new())),
    }
}

pub fn create_router(state: AppState) -> Router {
    tasks::spawn_task_completion_listener(state.clone());

    Router::new()
        .route("/api/tasks", get(list_tasks))
        .route("/api/tasks/runs", get(list_task_runs))
        .route("/api/tasks/:run_id", get(get_task_run))
        .route("/api/tasks/:run_id/logs", get(get_task_run_logs))
        .route("/api/tasks/run", post(run_task))
        .route("/api/tasks/cancel", post(cancel_task))
        .route("/api/tasks/restart", post(restart_task))
        .merge(SwaggerUi::new("/swagger-ui").url("/openapi.json", ApiDoc::openapi()))
        .with_state(state)
}

#[derive(OpenApi)]
#[openapi(
    paths(
        tasks::list_tasks,
        tasks::list_task_runs,
        tasks::get_task_run,
        tasks::get_task_run_logs,
        tasks::run_task,
        tasks::cancel_task,
        tasks::restart_task
    ),
    components(schemas(
        ListTasksRequest,
        ListTaskRunsRequest,
        ListTasksResponse,
        ListTasksResponseBody,
        ListTaskRunsResponse,
        ListTaskRunsResponseBody,
        GetTaskRunResponse,
        GetTaskRunResponseBody,
        GetTaskRunLogsResponse,
        GetTaskRunLogsResponseBody,
        GetTaskRunLogsRequest,
        TaskRunLogLine,
        TaskRunTreeNode,
        ErrorResponse,
        Task,
        StartTaskRequest,
        StartTaskResponse,
        StartTaskResponseBody,
        CancelTaskRequest,
        CancelTaskResponse,
        CancelTaskResponseBody,
        RestartTaskRequest,
        RestartTaskResponse,
        RestartTaskResponseBody,
        TaskRunStatus,
    ))
)]
pub struct ApiDoc;