robinpath_modules/modules/

queue_mod.rs

use robinpath::{RobinPath, Value};
use std::sync::{Arc, Mutex};
use std::collections::HashMap;

struct Queue {
    name: String,
    jobs: HashMap<String, Job>,
    dead_letter: Vec<Job>,
    max_dead_letter: usize,
    paused: bool,
    processed: u64,
    failed: u64,
    next_id: u64,
}

#[derive(Clone)]
struct Job {
    id: String,
    data: Value,
    priority: i64,
    status: String,
    attempts: u32,
    max_attempts: u32,
    created_at: u64,
    scheduled_for: u64,
    started_at: Option<u64>,
    completed_at: Option<u64>,
    failed_at: Option<u64>,
    error: Option<String>,
    result: Option<Value>,
}

fn now_ms() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis() as u64
}

pub fn register(rp: &mut RobinPath) {
    let state = Arc::new(Mutex::new(HashMap::<String, Queue>::new()));

    // queue.create name options? → {name, paused, maxDeadLetter}
    let s = state.clone();
    rp.register_builtin("queue.create", move |args, _| {
        let name = args.first().map(|v| v.to_display_string()).unwrap_or_else(|| "default".to_string());
        let opts = args.get(1).cloned().unwrap_or(Value::Null);
        let max_dl = if let Value::Object(obj) = &opts {
            obj.get("maxDeadLetter").map(|v| v.to_number() as usize).unwrap_or(1000)
        } else {
            1000
        };
        let mut queues = s.lock().unwrap();
        let q = queues.entry(name.clone()).or_insert_with(|| Queue {
            name: name.clone(),
            jobs: HashMap::new(),
            dead_letter: Vec::new(),
            max_dead_letter: max_dl,
            paused: false,
            processed: 0,
            failed: 0,
            next_id: 1,
        });
        q.max_dead_letter = max_dl;
        let mut obj = indexmap::IndexMap::new();
        obj.insert("name".to_string(), Value::String(q.name.clone()));
        obj.insert("paused".to_string(), Value::Bool(q.paused));
        obj.insert("maxDeadLetter".to_string(), Value::Number(q.max_dead_letter as f64));
        Ok(Value::Object(obj))
    });

    // queue.push queueName data options? → {id, status, priority}
    let s = state.clone();
    rp.register_builtin("queue.push", move |args, _| {
        let queue_name = args.first().map(|v| v.to_display_string()).unwrap_or_else(|| "default".to_string());
        let data = args.get(1).cloned().unwrap_or(Value::Null);
        let opts = args.get(2).cloned().unwrap_or(Value::Null);
        let (priority, delay_ms, max_attempts) = if let Value::Object(obj) = &opts {
            (
                obj.get("priority").map(|v| v.to_number() as i64).unwrap_or(0),
                obj.get("delay").map(|v| v.to_number() as u64).unwrap_or(0),
                obj.get("maxAttempts").map(|v| v.to_number() as u32).unwrap_or(3),
            )
        } else {
            (0, 0, 3)
        };
        let now = now_ms();
        let mut queues = s.lock().unwrap();
        let q = queues.entry(queue_name).or_insert_with(|| Queue {
            name: "default".to_string(),
            jobs: HashMap::new(),
            dead_letter: Vec::new(),
            max_dead_letter: 1000,
            paused: false,
            processed: 0,
            failed: 0,
            next_id: 1,
        });
        let id = format!("job_{}", q.next_id);
        q.next_id += 1;
        let status = if delay_ms > 0 { "delayed" } else { "pending" };
        let job = Job {
            id: id.clone(),
            data,
            priority,
            status: status.to_string(),
            attempts: 0,
            max_attempts,
            created_at: now,
            scheduled_for: now + delay_ms,
            started_at: None,
            completed_at: None,
            failed_at: None,
            error: None,
            result: None,
        };
        q.jobs.insert(id.clone(), job);
        let mut obj = indexmap::IndexMap::new();
        obj.insert("id".to_string(), Value::String(id));
        obj.insert("status".to_string(), Value::String(status.to_string()));
        obj.insert("priority".to_string(), Value::Number(priority as f64));
        Ok(Value::Object(obj))
    });

    // queue.pop queueName → job or null
    let s = state.clone();
    rp.register_builtin("queue.pop", move |args, _| {
        let queue_name = args.first().map(|v| v.to_display_string()).unwrap_or_else(|| "default".to_string());
        let mut queues = s.lock().unwrap();
        let q = match queues.get_mut(&queue_name) {
            Some(q) => q,
            None => return Ok(Value::Null),
        };
        if q.paused {
            return Ok(Value::Null);
        }
        let now = now_ms();
        // Promote delayed jobs
        for job in q.jobs.values_mut() {
            if job.status == "delayed" && job.scheduled_for <= now {
                job.status = "pending".to_string();
            }
        }
        // Find best pending job (highest priority, oldest first)
        let best_id = q.jobs.values()
            .filter(|j| j.status == "pending")
            .max_by(|a, b| {
                a.priority.cmp(&b.priority)
                    .then_with(|| b.created_at.cmp(&a.created_at))
            })
            .map(|j| j.id.clone());

        if let Some(id) = best_id {
            let job = q.jobs.get_mut(&id).unwrap();
            job.status = "active".to_string();
            job.attempts += 1;
            job.started_at = Some(now);
            let mut obj = indexmap::IndexMap::new();
            obj.insert("id".to_string(), Value::String(job.id.clone()));
            obj.insert("data".to_string(), job.data.clone());
            obj.insert("priority".to_string(), Value::Number(job.priority as f64));
            obj.insert("attempts".to_string(), Value::Number(job.attempts as f64));
            obj.insert("maxAttempts".to_string(), Value::Number(job.max_attempts as f64));
            Ok(Value::Object(obj))
        } else {
            Ok(Value::Null)
        }
    });

    // queue.complete queueName jobId result? → {id, status, duration}
    let s = state.clone();
    rp.register_builtin("queue.complete", move |args, _| {
        let queue_name = args.first().map(|v| v.to_display_string()).unwrap_or_else(|| "default".to_string());
        let job_id = args.get(1).map(|v| v.to_display_string()).unwrap_or_default();
        let result = args.get(2).cloned();
        let mut queues = s.lock().unwrap();
        let q = queues.get_mut(&queue_name)
            .ok_or_else(|| format!("Queue \"{}\" not found", queue_name))?;
        let job = q.jobs.get_mut(&job_id)
            .ok_or_else(|| format!("Job \"{}\" not found in queue \"{}\"", job_id, queue_name))?;
        let now = now_ms();
        job.status = "completed".to_string();
        job.completed_at = Some(now);
        job.result = result;
        q.processed += 1;
        let duration = now - job.started_at.unwrap_or(job.created_at);
        let mut obj = indexmap::IndexMap::new();
        obj.insert("id".to_string(), Value::String(job_id));
        obj.insert("status".to_string(), Value::String("completed".to_string()));
        obj.insert("duration".to_string(), Value::Number(duration as f64));
        Ok(Value::Object(obj))
    });

    // queue.fail queueName jobId error → {id, status, attempts}
    let s = state.clone();
    rp.register_builtin("queue.fail", move |args, _| {
        let queue_name = args.first().map(|v| v.to_display_string()).unwrap_or_else(|| "default".to_string());
        let job_id = args.get(1).map(|v| v.to_display_string()).unwrap_or_default();
        let error = args.get(2).map(|v| v.to_display_string()).unwrap_or_else(|| "Unknown error".to_string());
        let mut queues = s.lock().unwrap();
        let q = queues.get_mut(&queue_name)
            .ok_or_else(|| format!("Queue \"{}\" not found", queue_name))?;
        let job = q.jobs.get_mut(&job_id)
            .ok_or_else(|| format!("Job \"{}\" not found in queue \"{}\"", job_id, queue_name))?;
        job.error = Some(error.clone());
        job.failed_at = Some(now_ms());
        if job.attempts < job.max_attempts {
            job.status = "pending".to_string();
            let mut obj = indexmap::IndexMap::new();
            obj.insert("id".to_string(), Value::String(job_id));
            obj.insert("status".to_string(), Value::String("retry".to_string()));
            obj.insert("attempts".to_string(), Value::Number(job.attempts as f64));
            obj.insert("maxAttempts".to_string(), Value::Number(job.max_attempts as f64));
            Ok(Value::Object(obj))
        } else {
            let mut dead_job = job.clone();
            dead_job.status = "failed".to_string();
            q.failed += 1;
            q.dead_letter.push(dead_job);
            q.jobs.remove(&job_id);
            if q.dead_letter.len() > q.max_dead_letter {
                q.dead_letter.remove(0);
            }
            let mut obj = indexmap::IndexMap::new();
            obj.insert("id".to_string(), Value::String(job_id));
            obj.insert("status".to_string(), Value::String("dead-letter".to_string()));
            obj.insert("attempts".to_string(), Value::Number(q.dead_letter.last().map(|j| j.attempts).unwrap_or(0) as f64));
            obj.insert("error".to_string(), Value::String(error));
            Ok(Value::Object(obj))
        }
    });

    // queue.retry queueName jobId → {id, status, source}
    let s = state.clone();
    rp.register_builtin("queue.retry", move |args, _| {
        let queue_name = args.first().map(|v| v.to_display_string()).unwrap_or_else(|| "default".to_string());
        let job_id = args.get(1).map(|v| v.to_display_string()).unwrap_or_default();
        let mut queues = s.lock().unwrap();
        let q = queues.get_mut(&queue_name)
            .ok_or_else(|| format!("Queue \"{}\" not found", queue_name))?;
        // Check dead letter first
        if let Some(idx) = q.dead_letter.iter().position(|j| j.id == job_id) {
            let mut job = q.dead_letter.remove(idx);
            job.status = "pending".to_string();
            job.attempts = 0;
            job.error = None;
            job.failed_at = None;
            q.jobs.insert(job.id.clone(), job);
            let mut obj = indexmap::IndexMap::new();
            obj.insert("id".to_string(), Value::String(job_id));
            obj.insert("status".to_string(), Value::String("pending".to_string()));
            obj.insert("source".to_string(), Value::String("dead-letter".to_string()));
            return Ok(Value::Object(obj));
        }
        let job = q.jobs.get_mut(&job_id)
            .ok_or_else(|| format!("Job \"{}\" not found", job_id))?;
        job.status = "pending".to_string();
        job.error = None;
        let mut obj = indexmap::IndexMap::new();
        obj.insert("id".to_string(), Value::String(job_id));
        obj.insert("status".to_string(), Value::String("pending".to_string()));
        obj.insert("source".to_string(), Value::String("queue".to_string()));
        Ok(Value::Object(obj))
    });

    // queue.remove queueName jobId → bool
    let s = state.clone();
    rp.register_builtin("queue.remove", move |args, _| {
        let queue_name = args.first().map(|v| v.to_display_string()).unwrap_or_else(|| "default".to_string());
        let job_id = args.get(1).map(|v| v.to_display_string()).unwrap_or_default();
        let mut queues = s.lock().unwrap();
        if let Some(q) = queues.get_mut(&queue_name) {
            Ok(Value::Bool(q.jobs.remove(&job_id).is_some()))
        } else {
            Ok(Value::Bool(false))
        }
    });

    // queue.size queueName status? → number
    let s = state.clone();
    rp.register_builtin("queue.size", move |args, _| {
        let queue_name = args.first().map(|v| v.to_display_string()).unwrap_or_else(|| "default".to_string());
        let status_filter = args.get(1).map(|v| v.to_display_string());
        let mut queues = s.lock().unwrap();
        if let Some(q) = queues.get_mut(&queue_name) {
            let now = now_ms();
            for job in q.jobs.values_mut() {
                if job.status == "delayed" && job.scheduled_for <= now {
                    job.status = "pending".to_string();
                }
            }
            let count = if let Some(status) = status_filter {
                q.jobs.values().filter(|j| j.status == status).count()
            } else {
                q.jobs.len()
            };
            Ok(Value::Number(count as f64))
        } else {
            Ok(Value::Number(0.0))
        }
    });

    // queue.status queueName → {name, paused, total, pending, active, ...}
    let s = state.clone();
    rp.register_builtin("queue.status", move |args, _| {
        let queue_name = args.first().map(|v| v.to_display_string()).unwrap_or_else(|| "default".to_string());
        let mut queues = s.lock().unwrap();
        if let Some(q) = queues.get_mut(&queue_name) {
            let now = now_ms();
            let mut counts: HashMap<&str, u64> = HashMap::new();
            for job in q.jobs.values_mut() {
                if job.status == "delayed" && job.scheduled_for <= now {
                    job.status = "pending".to_string();
                }
                *counts.entry(match job.status.as_str() {
                    "pending" => "pending",
                    "active" => "active",
                    "completed" => "completed",
                    "failed" => "failed",
                    "delayed" => "delayed",
                    _ => "pending",
                }).or_insert(0) += 1;
            }
            let mut obj = indexmap::IndexMap::new();
            obj.insert("name".to_string(), Value::String(q.name.clone()));
            obj.insert("paused".to_string(), Value::Bool(q.paused));
            obj.insert("total".to_string(), Value::Number(q.jobs.len() as f64));
            obj.insert("pending".to_string(), Value::Number(*counts.get("pending").unwrap_or(&0) as f64));
            obj.insert("active".to_string(), Value::Number(*counts.get("active").unwrap_or(&0) as f64));
            obj.insert("completed".to_string(), Value::Number(*counts.get("completed").unwrap_or(&0) as f64));
            obj.insert("failed".to_string(), Value::Number(*counts.get("failed").unwrap_or(&0) as f64));
            obj.insert("delayed".to_string(), Value::Number(*counts.get("delayed").unwrap_or(&0) as f64));
            obj.insert("deadLetter".to_string(), Value::Number(q.dead_letter.len() as f64));
            obj.insert("totalProcessed".to_string(), Value::Number(q.processed as f64));
            obj.insert("totalFailed".to_string(), Value::Number(q.failed as f64));
            Ok(Value::Object(obj))
        } else {
            let mut obj = indexmap::IndexMap::new();
            obj.insert("name".to_string(), Value::String(queue_name));
            obj.insert("total".to_string(), Value::Number(0.0));
            Ok(Value::Object(obj))
        }
    });

    // queue.pause queueName → {name, paused: true}
    let s = state.clone();
    rp.register_builtin("queue.pause", move |args, _| {
        let queue_name = args.first().map(|v| v.to_display_string()).unwrap_or_else(|| "default".to_string());
        let mut queues = s.lock().unwrap();
        if let Some(q) = queues.get_mut(&queue_name) {
            q.paused = true;
        }
        let mut obj = indexmap::IndexMap::new();
        obj.insert("name".to_string(), Value::String(queue_name));
        obj.insert("paused".to_string(), Value::Bool(true));
        Ok(Value::Object(obj))
    });

    // queue.resume queueName → {name, paused: false}
    let s = state.clone();
    rp.register_builtin("queue.resume", move |args, _| {
        let queue_name = args.first().map(|v| v.to_display_string()).unwrap_or_else(|| "default".to_string());
        let mut queues = s.lock().unwrap();
        if let Some(q) = queues.get_mut(&queue_name) {
            q.paused = false;
        }
        let mut obj = indexmap::IndexMap::new();
        obj.insert("name".to_string(), Value::String(queue_name));
        obj.insert("paused".to_string(), Value::Bool(false));
        Ok(Value::Object(obj))
    });

    // queue.clear queueName → {cleared: number}
    let s = state.clone();
    rp.register_builtin("queue.clear", move |args, _| {
        let queue_name = args.first().map(|v| v.to_display_string()).unwrap_or_else(|| "default".to_string());
        let mut queues = s.lock().unwrap();
        let count = if let Some(q) = queues.get_mut(&queue_name) {
            let c = q.jobs.len();
            q.jobs.clear();
            c
        } else {
            0
        };
        let mut obj = indexmap::IndexMap::new();
        obj.insert("cleared".to_string(), Value::Number(count as f64));
        Ok(Value::Object(obj))
    });

    // queue.deadLetter queueName limit? → array of failed jobs
    let s = state.clone();
    rp.register_builtin("queue.deadLetter", move |args, _| {
        let queue_name = args.first().map(|v| v.to_display_string()).unwrap_or_else(|| "default".to_string());
        let limit = args.get(1).map(|v| v.to_number() as usize).unwrap_or(50);
        let queues = s.lock().unwrap();
        if let Some(q) = queues.get(&queue_name) {
            let start = if q.dead_letter.len() > limit { q.dead_letter.len() - limit } else { 0 };
            let items: Vec<Value> = q.dead_letter[start..].iter().map(|j| {
                let mut obj = indexmap::IndexMap::new();
                obj.insert("id".to_string(), Value::String(j.id.clone()));
                obj.insert("data".to_string(), j.data.clone());
                obj.insert("error".to_string(), Value::String(j.error.clone().unwrap_or_default()));
                obj.insert("attempts".to_string(), Value::Number(j.attempts as f64));
                Ok(Value::Object(obj))
            }).collect::<Result<Vec<_>, String>>()?;
            Ok(Value::Array(items))
        } else {
            Ok(Value::Array(Vec::new()))
        }
    });

    // queue.getJob queueName jobId → job object or null
    let s = state.clone();
    rp.register_builtin("queue.getJob", move |args, _| {
        let queue_name = args.first().map(|v| v.to_display_string()).unwrap_or_else(|| "default".to_string());
        let job_id = args.get(1).map(|v| v.to_display_string()).unwrap_or_default();
        let queues = s.lock().unwrap();
        if let Some(q) = queues.get(&queue_name) {
            if let Some(job) = q.jobs.get(&job_id) {
                let mut obj = indexmap::IndexMap::new();
                obj.insert("id".to_string(), Value::String(job.id.clone()));
                obj.insert("data".to_string(), job.data.clone());
                obj.insert("status".to_string(), Value::String(job.status.clone()));
                obj.insert("priority".to_string(), Value::Number(job.priority as f64));
                obj.insert("attempts".to_string(), Value::Number(job.attempts as f64));
                obj.insert("maxAttempts".to_string(), Value::Number(job.max_attempts as f64));
                if let Some(err) = &job.error {
                    obj.insert("error".to_string(), Value::String(err.clone()));
                }
                return Ok(Value::Object(obj));
            }
        }
        Ok(Value::Null)
    });

    // queue.destroy queueName → bool
    let s = state.clone();
    rp.register_builtin("queue.destroy", move |args, _| {
        let queue_name = args.first().map(|v| v.to_display_string()).unwrap_or_else(|| "default".to_string());
        let mut queues = s.lock().unwrap();
        Ok(Value::Bool(queues.remove(&queue_name).is_some()))
    });
}