Port queue page

src/server/web/pages/queue.rs

@@ -0,0 +1,198 @@
+use std::{
+    collections::HashMap,
+    sync::{Arc, Mutex},
+};
+
+use askama::Template;
+use axum::{extract::State, response::IntoResponse};
+use futures::TryStreamExt;
+use sqlx::SqlitePool;
+
+use crate::{
+    config::Config,
+    server::{
+        util,
+        web::{
+            base::{Base, Link, Tab},
+            link::{LinkCommit, LinkRunShort, LinkWorker},
+            paths::{
+                PathAdminQueueDecrease, PathAdminQueueDelete, PathAdminQueueIncrease, PathQueue,
+                PathQueueInner,
+            },
+            r#static::QUEUE_JS,
+        },
+        workers::{WorkerInfo, Workers},
+    },
+    shared::WorkerStatus,
+    somehow,
+};
+
+enum Status {
+    Idle,
+    Busy,
+    Working(LinkRunShort),
+}
+
+struct Worker {
+    link: LinkWorker,
+    status: Status,
+}
+
+struct Task {
+    link_delete: Link,
+    link_increase: Link,
+    link_decrease: Link,
+    hash: String,
+    commit: LinkCommit,
+    since: String,
+    priority: i64,
+    workers: Vec<LinkWorker>,
+    odd: bool,
+}
+
+fn sorted_workers(workers: &Mutex<Workers>) -> Vec<(String, WorkerInfo)> {
+    let mut workers = workers
+        .lock()
+        .unwrap()
+        .clean()
+        .get_all()
+        .into_iter()
+        .collect::<Vec<_>>();
+    workers.sort_unstable_by(|(a, _), (b, _)| a.cmp(b));
+    workers
+}
+
+async fn get_workers(
+    db: &SqlitePool,
+    workers: &[(String, WorkerInfo)],
+    base: &Base,
+) -> somehow::Result<Vec<Worker>> {
+    let mut result = vec![];
+    for (name, info) in workers {
+        let status = match &info.status {
+            WorkerStatus::Idle => Status::Idle,
+            WorkerStatus::Busy => Status::Busy,
+            WorkerStatus::Working(unfinished) => {
+                let message = sqlx::query_scalar!(
+                    "SELECT message FROM commits WHERE hash = ?",
+                    unfinished.run.hash
+                )
+                .fetch_one(db)
+                .await?;
+                Status::Working(LinkRunShort::new(
+                    base,
+                    unfinished.run.id.clone(),
+                    &unfinished.run.hash,
+                    &message,
+                ))
+            }
+        };
+
+        result.push(Worker {
+            link: LinkWorker::new(base, name.clone()),
+            status,
+        })
+    }
+    Ok(result)
+}
+
+async fn get_queue_data(
+    db: &SqlitePool,
+    workers: &[(String, WorkerInfo)],
+    base: &Base,
+) -> somehow::Result<Vec<Task>> {
+    // Group workers by commit hash
+    let mut workers_by_commit: HashMap<String, Vec<LinkWorker>> = HashMap::new();
+    for (name, info) in workers {
+        if let WorkerStatus::Working(unfinished) = &info.status {
+            workers_by_commit
+                .entry(unfinished.run.hash.clone())
+                .or_default()
+                .push(LinkWorker::new(base, name.clone()));
+        }
+    }
+
+    let mut tasks = sqlx::query!(
+        "\
+        SELECT \
+            hash, \
+            message, \
+            reachable, \
+            date AS \"date: time::OffsetDateTime\", \
+            priority \
+        FROM queue \
+        JOIN commits USING (hash) \
+        ORDER BY priority DESC, unixepoch(date) DESC, hash ASC \
+        "
+    )
+    .fetch(db)
+    .map_ok(|r| Task {
+        link_delete: base.link(PathAdminQueueDelete {}),
+        link_increase: base.link(PathAdminQueueIncrease {}),
+        link_decrease: base.link(PathAdminQueueDecrease {}),
+        hash: r.hash.clone(),
+        workers: workers_by_commit.remove(&r.hash).unwrap_or_default(),
+        commit: LinkCommit::new(base, r.hash, &r.message, r.reachable),
+        since: util::format_delta_from_now(r.date),
+        priority: r.priority,
+        odd: false,
+    })
+    .try_collect::<Vec<_>>()
+    .await?;
+
+    let mut last_priority = None;
+    let mut odd = false;
+    for task in tasks.iter_mut().rev() {
+        if last_priority.is_some() && last_priority != Some(task.priority) {
+            odd = !odd;
+        }
+        task.odd = odd;
+        last_priority = Some(task.priority);
+    }
+
+    Ok(tasks)
+}
+
+#[derive(Template)]
+#[template(path = "pages/queue_inner.html")]
+struct QueueInnerTemplate {
+    workers: Vec<Worker>,
+    tasks: Vec<Task>,
+}
+
+pub async fn get_queue_inner(
+    _path: PathQueueInner,
+    State(config): State<&'static Config>,
+    State(db): State<SqlitePool>,
+    State(workers): State<Arc<Mutex<Workers>>>,
+) -> somehow::Result<impl IntoResponse> {
+    let base = Base::new(config, Tab::Queue);
+    let sorted_workers = sorted_workers(&workers);
+    let workers = get_workers(&db, &sorted_workers, &base).await?;
+    let tasks = get_queue_data(&db, &sorted_workers, &base).await?;
+    Ok(QueueInnerTemplate { workers, tasks })
+}
+#[derive(Template)]
+#[template(path = "pages/queue.html")]
+struct QueueTemplate {
+    link_queue_js: Link,
+    base: Base,
+    inner: QueueInnerTemplate,
+}
+
+pub async fn get_queue(
+    _path: PathQueue,
+    State(config): State<&'static Config>,
+    State(db): State<SqlitePool>,
+    State(workers): State<Arc<Mutex<Workers>>>,
+) -> somehow::Result<impl IntoResponse> {
+    let base = Base::new(config, Tab::Queue);
+    let sorted_workers = sorted_workers(&workers);
+    let workers = get_workers(&db, &sorted_workers, &base).await?;
+    let tasks = get_queue_data(&db, &sorted_workers, &base).await?;
+    Ok(QueueTemplate {
+        link_queue_js: base.link(QUEUE_JS),
+        base,
+        inner: QueueInnerTemplate { workers, tasks },
+    })
+}