initial commit

.env.example

@@ -0,0 +1 @@
+DATABASE_URL=sqlite://autotasker.db

.gitignore

@@ -0,0 +1,3 @@
+/target
+autotasker.db
+.env

Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "autotasker"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+serde_yaml = "0.9"
+chrono = { version = "0.4.26", features = ["serde"] }
+sqlx = { version = "0.7.4", features = ["sqlite", "runtime-tokio", "chrono"] }
+anyhow = "1.0.75"
+clap = "4.5.4"
+tera = "1.19.1"
+tokio = { version = "1.37.0", features = ["full"] }
+axum = { version = "0.7.5", features = ["json"] }
+minijinja = { version = "1.0.20", features = ["builtins"] }
+uuid = { version = "1.8.0", features = ["serde", "v4"] }
+fully_pub = "0.1.4"
+

README.md

@@ -0,0 +1,15 @@
+## features
+
+- List the jobs available
+- Run a background-process
+
+- actix
+- tera
+
+## vocabulary
+
+- Job
+- JobRun
+
+Using 
+https://docs.rs/sqlx-models/latest/sqlx_models/

TODO.md

@@ -0,0 +1,18 @@
+- [ ] Websocket return of logs
+- [ ] setup sqlite
+- [ ] setup task runs (or orders) in sqlite
+- [ ] store of task run logs in sqlite
+
+simple axum server
+simple templating route with axum
+
+Use minijinja for templating with django syntax https://lib.rs/crates/minijinja
+
+login OAuth2
+list tasks
+button to trigger job
+api to get logs of task run
+schedule tasks runs
+simple frontenv with htmlx
+
+- [  ] webhook

config.yaml

@@ -0,0 +1,17 @@
+tasks:
+  - id: do_magic_stuff
+    name: Do magic incantation
+    env:
+      PYTHONUNBUFFERED: "1"
+      SIMULATION_SPEED: 0.2
+    command:
+      - /usr/bin/python3
+      - /home/mbess/workspace/autotasker/examples/do_something_1.py
+    store_logs: true
+  - id: reindex_db
+    name: Reindex the whole database
+    env: {}
+    command:
+      - ls
+      - /etc/fstab
+

examples/do_something_1.py

@@ -0,0 +1,29 @@
+import random
+import os
+import string
+from time import sleep
+
+
+def main():
+    iterations = random.randint(10, 150)
+    speed = float(os.getenv("SIMULATION_SPEED") or 0.8)
+    print(f"Going for {iterations=}")
+    for i in range(iterations):
+        print(
+            str(i) + " " +
+            ''.join(
+                random.sample(
+                    " "+string.ascii_uppercase+string.ascii_lowercase+string.digits,
+                    int(random.uniform(10, 50))
+                )
+            )
+        )
+        sleep(
+            speed * (0.02 * random.expovariate(0.5) +
+                (random.uniform(0, 1) if random.uniform(0, 1) > 0.8 else 0) +
+                (random.uniform(0, 5) if random.uniform(0, 1) > 0.99 else 0))
+        )
+    print("Done, script is finished")
+if __name__ == "__main__":
+    main()
+

migrations/01_create_task_runs_table.sql

@@ -0,0 +1,11 @@
+CREATE TABLE IF NOT EXISTS task_runs (
+    id              TEXT PRIMARY KEY,
+    task_id         TEXT NOT NULL,
+    status          TEXT CHECK(status IN ('pending','running','failed','success'))  NOT NULL DEFAULT 'pending',
+    trigger_mode    TEXT CHECK(status IN ('manual','webhook','schedule'))  NOT NULL,
+    exit_code       INT,
+    logs            TEXT,
+    submitted_at    DATETIME,
+    started_at      DATETIME,
+    end_at          DATETIME
+)

src/controllers.rs

@@ -0,0 +1,97 @@
+use crate::models::{Config, Task};
+use anyhow::{anyhow, Context, Result};
+use axum::extract::{Path as ExtractPath, State};
+use axum::http::{Response, StatusCode};
+use axum::Json;
+use axum::{response::Html, routing::get, Router};
+use minijinja::render;
+use serde::{Deserialize, Serialize};
+use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
+use sqlx::{ConnectOptions, Connection, Pool, Sqlite};
+use std::collections::HashMap;
+use std::fs;
+use std::process::Stdio;
+use std::str::FromStr;
+use std::sync::Arc;
+use tokio::io::{AsyncBufReadExt, BufReader};
+use tokio::process::{Child, Command};
+use tokio::sync::mpsc::{self, Receiver, Sender};
+use uuid::Uuid;
+
+use crate::models::ExecutorOrder;
+use crate::AppState;
+
+pub async fn home() -> Html<String> {
+    Html(render!(
+        include_str!("./templates/home.html"),
+        first_name => "John Doe"
+    ))
+}
+
+pub async fn list_tasks(State(app_state): State<AppState>) -> Html<String> {
+    let tasks: Vec<Task> = app_state.config.tasks;
+    Html(render!(
+        include_str!("./templates/list_tasks.html"),
+        tasks => tasks
+    ))
+}
+
+pub async fn trigger_task(
+    State(app_state): State<AppState>,
+    ExtractPath(task_id): ExtractPath<String>,
+) -> (StatusCode, Html<String>) {
+    println!("Run task {}", task_id);
+    let tasks: Vec<Task> = app_state.config.tasks;
+    let task = match tasks.iter().find(|t| t.id == task_id) {
+        Some(t) => t,
+        None => {
+            return (
+                StatusCode::NOT_FOUND,
+                Html(r##"<b style="color: red;">404 Error</b>"##.to_string()),
+            )
+        }
+    };
+    app_state
+        .executor_tx
+        .send(ExecutorOrder {
+            id: Uuid::new_v4(),
+            task: task.clone(),
+        })
+        .await
+        .unwrap();
+
+    (
+        StatusCode::OK,
+        Html(render!(
+            include_str!("./templates/run_task.html"),
+            tasks => tasks
+        )),
+    )
+}
+
+pub async fn handle_webhook(
+    State(app_state): State<AppState>,
+    ExtractPath(token): ExtractPath<String>,
+) -> (StatusCode, Json<String>) {
+    println!("Webhook token {}", token);
+
+    (StatusCode::OK, axum::Json("WebHook handle".to_string()))
+}
+
+pub async fn list_task_runs(
+    State(app_state): State<AppState>,
+    ExtractPath(task_id): ExtractPath<String>,
+) -> Html<String> {
+    Html(render!(include_str!("./templates/list_task_runs.html")))
+}
+
+pub async fn get_task_run(
+    State(app_state): State<AppState>,
+    ExtractPath(task_id): ExtractPath<String>,
+    ExtractPath(run_id): ExtractPath<String>,
+) -> Html<String> {
+    Html(render!(
+        include_str!("./templates/task_run_details.html"),
+        run => false
+    ))
+}

src/main.rs

@@ -0,0 +1,136 @@
+use anyhow::{anyhow, Context, Result};
+use axum::extract::{Path as ExtractPath, State};
+use axum::http::{Response, StatusCode};
+use axum::routing::get;
+use axum::{Json, Router};
+use minijinja::render;
+use models::{Config, Task};
+use serde::{Deserialize, Serialize};
+use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
+use sqlx::{ConnectOptions, Connection, Pool, Sqlite};
+use std::collections::HashMap;
+use std::fs;
+use std::process::Stdio;
+use std::str::FromStr;
+use std::sync::Arc;
+use tokio::io::{AsyncBufReadExt, BufReader};
+use tokio::process::{Child, Command};
+use tokio::sync::mpsc::{self, Receiver, Sender};
+use uuid::Uuid;
+
+mod controllers;
+mod models;
+
+#[derive(Debug, Clone)]
+pub struct AppState {
+    config: Config,
+    db: Pool<Sqlite>,
+    executor_tx: Arc<Sender<ExecutorOrder>>,
+}
+
+fn get_config() -> Result<Config> {
+    let inp_def_yaml = fs::read_to_string("./config.yaml")
+        .expect("Should have been able to read the the config file");
+
+    return serde_yaml::from_str(&inp_def_yaml)
+        .map_err(|e| anyhow!("Failed to parse config, {:?}", e));
+}
+
+use crate::models::ExecutorOrder;
+
+async fn run_task(order: ExecutorOrder) {
+    let executable = match order.task.command.iter().next() {
+        Some(v) => v,
+        None => return,
+    };
+    println!("Start processing of order {}", order.id);
+    let mut cmd = Command::new(executable);
+
+    cmd.args(order.task.command.iter().skip(1).collect::<Vec<&String>>())
+        .stdout(Stdio::piped());
+    for (key, val) in order.task.env.iter() {
+        cmd.env(key, val);
+    }
+
+    let mut child = cmd.spawn().expect("failed to execute process");
+
+    let stdout = child.stdout.take().unwrap();
+
+    tokio::spawn(async move {
+        let status = child
+            .wait()
+            .await
+            .expect("child process encountered an error");
+
+        println!("child status was: {}", status);
+    });
+
+    let mut lines = BufReader::with_capacity(16, stdout).lines();
+    while let Some(line) = lines.next_line().await.unwrap() {
+        println!("{}: {}", order.id, line);
+    }
+    println!("End of task")
+}
+
+async fn run_executor(mut rx: Receiver<ExecutorOrder>) {
+    println!("Executor started");
+    while let Some(order) = rx.recv().await {
+        println!("Got Order: {:?}", order);
+        tokio::spawn(async { run_task(order).await });
+    }
+    println!("Executor stopped");
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    let pool = prepare_database().await.context("Prepare db")?;
+
+    // start executor daemon
+    let (tx, rx) = mpsc::channel::<ExecutorOrder>(32);
+
+    let executor_handle = tokio::spawn(async { run_executor(rx).await });
+
+    let config: Config = get_config().expect("Cannot get config");
+    let state = AppState {
+        config,
+        db: pool,
+        executor_tx: Arc::new(tx),
+    };
+
+    // build our application with a single route
+    let app = Router::new()
+        .route("/", get(controllers::home))
+        .route("/tasks", get(controllers::list_tasks))
+        .route("/tasks/:task_id/trigger", get(controllers::trigger_task))
+        .route("/tasks/:task_id/runs", get(controllers::list_task_runs))
+        .route(
+            "/tasks/:task_id/runs/:run_id",
+            get(controllers::get_task_run),
+        )
+        .route("/webhooks/:token", get(controllers::handle_webhook))
+        .with_state(state);
+
+    let listener = tokio::net::TcpListener::bind("0.0.0.0:8085").await.unwrap();
+    axum::serve(listener, app).await?;
+
+    Ok(())
+}
+
+async fn prepare_database() -> Result<Pool<Sqlite>> {
+    // create database if it does not exist
+    let conn = SqliteConnectOptions::from_str("sqlite:autotasker.db")?
+        .create_if_missing(true)
+        .connect()
+        .await?;
+    let _ = conn.close().await;
+
+    let pool = SqlitePoolOptions::new()
+        .max_connections(50)
+        .connect("sqlite:autotasker.db")
+        .await
+        .context("could not connect to database_url")?;
+
+    sqlx::migrate!().run(&pool).await?;
+
+    Ok(pool)
+}

src/models.rs

@@ -0,0 +1,61 @@
+use std::collections::HashMap;
+
+use chrono::{DateTime, Utc};
+use fully_pub::fully_pub;
+use serde::{Deserialize, Serialize};
+use uuid::Uuid;
+
+#[derive(sqlx::Type, Debug, Serialize, Deserialize)]
+#[fully_pub]
+#[sqlx(rename_all = "lowercase")]
+enum TriggerMode {
+    Manual,
+    Webhook,
+    Schedule,
+}
+
+#[derive(sqlx::Type, Debug, Serialize, Deserialize)]
+#[fully_pub]
+#[sqlx(rename_all = "lowercase")]
+enum TaskStatus {
+    Pending,
+    Running,
+    Failed,
+    Success,
+}
+
+#[derive(sqlx::FromRow, Deserialize, Serialize)]
+#[fully_pub]
+struct TaskRun {
+    id: String,
+    task_id: String,
+    status: TaskStatus,
+    trigger_mode: TriggerMode,
+    exit_code: u32,
+    logs: String,
+    submitted_at: DateTime<Utc>,
+    started_at: DateTime<Utc>,
+    end_at: DateTime<Utc>,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[fully_pub]
+struct Task {
+    id: String,
+    name: String,
+    env: HashMap<String, String>,
+    command: Vec<String>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[fully_pub]
+struct ExecutorOrder {
+    id: Uuid,
+    task: Task,
+}
+
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+#[fully_pub]
+struct Config {
+    tasks: Vec<Task>,
+}

src/templates/home.html

@@ -0,0 +1 @@
+<h1>Hello, {{ first_name }}</h1>

src/templates/list_tasks.html

@@ -0,0 +1,5 @@
+<ul>
+{% for task in tasks %}
+    <li>{{ task.name }} <a href="/tasks/{{ task.id }}/run">Run task</a></li>
+{% endfor %}
+</ul>

src/templates/run_task.html

@@ -0,0 +1 @@
+Task triggered!