feat: job queue html

src/bin/server.rs

@@ -158,7 +158,7 @@ fn html_routes(
             get(show_ingress_form).post(process_ingress_form),
         )
         .route("/queue", get(show_queue_tasks))
-        .route("/queue/:delivery_tag", post(delete_task))
+        .route("/queue/:delivery_tag", delete(delete_task))
         .route("/account", get(show_account_page))
         .route("/set-api-key", post(set_api_key))
         .route("/delete-account", delete(delete_account))

src/bin/worker.rs

@@ -1,11 +1,18 @@
 use std::sync::Arc;
 
 use futures::StreamExt;
+use surrealdb::Action;
 use tracing::{error, info};
 use tracing_subscriber::{fmt, prelude::*, EnvFilter};
 use zettle_db::{
-    ingress::{content_processor::ContentProcessor, jobqueue::JobQueue},
-    storage::db::SurrealDbClient,
+    ingress::{
+        content_processor::ContentProcessor,
+        jobqueue::{self, JobQueue, MAX_ATTEMPTS},
+    },
+    storage::{
+        db::{get_item, SurrealDbClient},
+        types::job::{Job, JobStatus},
+    },
     utils::config::get_config,
 };
 
@@ -54,12 +61,66 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         while let Some(notification) = job_stream.next().await {
             match notification {
                 Ok(notification) => {
-                    info!("Received new job: {}", notification.data.id);
-                    if let Err(e) = job_queue
-                        .process_job(notification.data, &content_processor)
-                        .await
-                    {
-                        error!("Error processing job: {}", e);
+                    info!("Received notification: {:?}", notification);
+
+                    match notification.action {
+                        Action::Create => {
+                            if let Err(e) = job_queue
+                                .process_job(notification.data, &content_processor)
+                                .await
+                            {
+                                error!("Error processing job: {}", e);
+                            }
+                        }
+                        Action::Update => {
+                            match notification.data.status {
+                                JobStatus::Completed
+                                | JobStatus::Error(_)
+                                | JobStatus::Cancelled => {
+                                    info!(
+                                        "Skipping already completed/error/cancelled job: {}",
+                                        notification.data.id
+                                    );
+                                    continue;
+                                }
+                                JobStatus::InProgress { attempts, .. } => {
+                                    // Only process if this is a retry after an error, not our own update
+                                    if let Ok(Some(current_job)) =
+                                        get_item::<Job>(&job_queue.db.client, &notification.data.id)
+                                            .await
+                                    {
+                                        match current_job.status {
+                                            JobStatus::Error(_) if attempts < MAX_ATTEMPTS => {
+                                                // This is a retry after an error
+                                                if let Err(e) = job_queue
+                                                    .process_job(current_job, &content_processor)
+                                                    .await
+                                                {
+                                                    error!("Error processing job retry: {}", e);
+                                                }
+                                            }
+                                            _ => {
+                                                info!(
+                                                    "Skipping in-progress update for job: {}",
+                                                    notification.data.id
+                                                );
+                                                continue;
+                                            }
+                                        }
+                                    }
+                                }
+                                JobStatus::Created => {
+                                    // Shouldn't happen with Update action, but process if it does
+                                    if let Err(e) = job_queue
+                                        .process_job(notification.data, &content_processor)
+                                        .await
+                                    {
+                                        error!("Error processing job: {}", e);
+                                    }
+                                }
+                            }
+                        }
+                        _ => {} // Ignore other actions
                     }
                 }
                 Err(e) => error!("Error in job notification: {}", e),

src/ingress/jobqueue.rs

@@ -9,7 +9,7 @@ use tracing::{error, info};
 use crate::{
     error::AppError,
     storage::{
-        db::{store_item, SurrealDbClient},
+        db::{delete_item, get_item, store_item, SurrealDbClient},
         types::{
             job::{Job, JobStatus},
             StoredObject,
@@ -20,10 +20,10 @@ use crate::{
 use super::{content_processor::ContentProcessor, types::ingress_object::IngressObject};
 
 pub struct JobQueue {
-    db: Arc<SurrealDbClient>,
+    pub db: Arc<SurrealDbClient>,
 }
 
-const MAX_ATTEMPTS: u32 = 3;
+pub const MAX_ATTEMPTS: u32 = 3;
 
 impl JobQueue {
     pub fn new(db: Arc<SurrealDbClient>) -> Self {
@@ -50,34 +50,22 @@ impl JobQueue {
     }
 
     pub async fn delete_job(&self, id: &str, user_id: &str) -> Result<(), AppError> {
-        // First, validate that the job exists and belongs to the user
-        let job: Option<Job> = self
-            .db
-            .query("SELECT * FROM job WHERE id = $id AND user_id = $user_id")
-            .bind(("id", id.to_string()))
-            .bind(("user_id", user_id.to_string()))
+        get_item::<Job>(&self.db.client, id)
             .await?
-            .take(0)?;
-
-        // If no job is found or it doesn't belong to the user, return Unauthorized
-        if job.is_none() {
-            error!("Unauthorized attempt to delete job {id} by user {user_id}");
-            return Err(AppError::Auth("Not authorized to delete this job".into()));
-        }
+            .filter(|job| job.user_id == user_id)
+            .ok_or_else(|| {
+                error!("Unauthorized attempt to delete job {id} by user {user_id}");
+                AppError::Auth("Not authorized to delete this job".into())
+            })?;
 
         info!("Deleting job {id} for user {user_id}");
-
-        // If validation passes, delete the job
-        let _deleted: Option<Job> = self
-            .db
-            .delete((Job::table_name(), id))
+        delete_item::<Job>(&self.db.client, id)
             .await
             .map_err(AppError::Database)?;
 
         Ok(())
     }
 
-    /// Update status for job
     pub async fn update_status(
         &self,
         id: &str,
@@ -89,13 +77,10 @@ impl JobQueue {
             .as_millis()
             .to_string();
 
-        let status_value =
-            serde_json::to_value(status).map_err(|e| AppError::LLMParsing(e.to_string()))?;
-
         let job: Option<Job> = self
             .db
             .update((Job::table_name(), id))
-            .patch(PatchOp::replace("/status", status_value))
+            .patch(PatchOp::replace("/status", status))
             .patch(PatchOp::replace("/updated_at", now))
             .await?;
 
@@ -109,14 +94,27 @@ impl JobQueue {
         self.db.select("job").live().await
     }
 
+    /// Get unfinished jobs, ie newly created and in progress up two times
     pub async fn get_unfinished_jobs(&self) -> Result<Vec<Job>, AppError> {
         let jobs: Vec<Job> = self
             .db
             .query(
-                "SELECT * FROM job WHERE status.Created = true OR (status.InProgress.attempts < $max_attempts) ORDER BY created_at ASC")
+                "SELECT * FROM type::table($table) 
+             WHERE 
+                status = 'Created' 
+                OR (
+                    status.InProgress != NONE 
+                    AND status.InProgress.attempts < $max_attempts
+                )
+             ORDER BY created_at ASC",
+            )
+            .bind(("table", Job::table_name()))
             .bind(("max_attempts", MAX_ATTEMPTS))
             .await?
             .take(0)?;
+
+        println!("Unfinished jobs found: {}", jobs.len());
+
         Ok(jobs)
     }
 

src/server/routes/html/ingress_tasks.rs

@@ -2,7 +2,10 @@ use crate::{
     error::{AppError, HtmlError},
     page_data,
     server::AppState,
-    storage::types::{job::Job, user::User},
+    storage::types::{
+        job::{Job, JobStatus},
+        user::User,
+    },
 };
 use axum::{
     extract::{Path, State},
@@ -11,10 +14,11 @@ use axum::{
 use axum_session_auth::AuthSession;
 use axum_session_surreal::SessionSurrealPool;
 use surrealdb::{engine::any::Any, Surreal};
+use tracing::info;
 
 use super::render_template;
 
-page_data!(ShowQueueTasks, "queue_tasks.html", {jobs: Vec<Job>});
+page_data!(ShowQueueTasks, "queue_tasks.html", {user : User,jobs: Vec<Job>});
 
 pub async fn show_queue_tasks(
     State(state): State<AppState>,
@@ -31,9 +35,16 @@ pub async fn show_queue_tasks(
         .await
         .map_err(|e| HtmlError::new(e, state.templates.clone()))?;
 
+    for job in &jobs {
+        match job.status {
+            JobStatus::Created => info!("Found a created job"),
+            _ => continue,
+        }
+    }
+
     let rendered = render_template(
         ShowQueueTasks::template_name(),
-        ShowQueueTasks { jobs },
+        ShowQueueTasks { jobs, user },
         state.templates.clone(),
     )
     .map_err(|e| HtmlError::new(AppError::from(e), state.templates.clone()))?;