snerd_rust/
dashboard.rs

1use std::collections::{HashMap, VecDeque};
2use std::io::{BufRead, BufReader, Read, Write};
3use std::net::{TcpListener, TcpStream};
4use std::sync::{Arc, Mutex};
5use std::time::{SystemTime, UNIX_EPOCH};
6
7use serde_json::{json, Value};
8use tokio::sync::broadcast;
9
10use crate::queue::SnerdQueue;
11
12/// A single entry in the dashboard's Progress Stream.
13struct ProgressEvent {
14    ts: f64,
15    task_id: String,
16    data: String,
17}
18
19type ProgressRing = Arc<Mutex<VecDeque<ProgressEvent>>>;
20
21const RING_CAP: usize = 500;
22
23fn now_secs() -> f64 {
24    SystemTime::now()
25        .duration_since(UNIX_EPOCH)
26        .map(|d| d.as_secs_f64())
27        .unwrap_or(0.0)
28}
29
30/// Reads the append-only task log and returns the latest line per taskId
31/// (cron refires and retries append new lines over time).
32fn read_deduped_tasks(tasks_path: &str) -> HashMap<String, Value> {
33    let mut tasks_map: HashMap<String, Value> = HashMap::new();
34    let file = match std::fs::File::open(tasks_path) {
35        Ok(f) => f,
36        Err(_) => return tasks_map,
37    };
38    for line in BufReader::new(file).lines().map_while(Result::ok) {
39        if line.trim().is_empty() {
40            continue;
41        }
42        if let Ok(t) = serde_json::from_str::<Value>(&line) {
43            if let Some(tid) = t.get("taskId").and_then(|v| v.as_str()) {
44                tasks_map.insert(tid.to_string(), t);
45            }
46        }
47    }
48    tasks_map
49}
50
51fn has_job_error(t: &Value) -> bool {
52    // Tolerate the lowercase variant in case logs were written by older builds
53    t.get("LastJobError").is_some() || t.get("lastJobError").is_some()
54}
55
56/// Derives the UI status for a deduped task record.
57fn dashboard_status(t: &Value) -> &'static str {
58    let has_err = has_job_error(t);
59    let deleted = t
60        .get("deletedAt")
61        .map_or(false, |v| !v.is_null());
62    if deleted {
63        let retry_count = t.get("retryCount").and_then(|v| v.as_i64()).unwrap_or(0);
64        let max_retries = t.get("maxRetries").and_then(|v| v.as_i64()).unwrap_or(0);
65        if has_err && retry_count >= max_retries {
66            return "dead_letter";
67        } else if has_err {
68            return "failed";
69        }
70        return "completed";
71    }
72    if has_err {
73        return "failed";
74    }
75    if let Some(exec_at) = t.get("executeAt").and_then(|v| v.as_str()) {
76        if let Ok(et) = chrono::DateTime::parse_from_rfc3339(exec_at) {
77            if et <= chrono::Utc::now() {
78                return "active";
79            }
80        }
81    }
82    "queued"
83}
84
85fn stats_body(tasks_path: &str) -> String {
86    let tasks_map = read_deduped_tasks(tasks_path);
87    let enqueued = tasks_map.len();
88    let mut processed = 0usize;
89    let mut failed = 0usize;
90    for t in tasks_map.values() {
91        let deleted = t.get("deletedAt").map_or(false, |v| !v.is_null());
92        if deleted {
93            if has_job_error(t) {
94                failed += 1;
95            } else {
96                processed += 1;
97            }
98        }
99    }
100    format!(
101        "{{\"enqueued\":{},\"processed\":{},\"failed\":{}}}",
102        enqueued, processed, failed
103    )
104}
105
106fn tasks_body(tasks_path: &str) -> String {
107    let tasks_map = read_deduped_tasks(tasks_path);
108    let res: Vec<Value> = tasks_map
109        .values()
110        .map(|t| {
111            json!({
112                "id": t.get("taskId"),
113                "type": t.get("taskType"),
114                "status": dashboard_status(t),
115                "progress": 0,
116                "retryCount": t.get("retryCount").and_then(|v| v.as_i64()).unwrap_or(0),
117                "maxRetries": t.get("maxRetries").and_then(|v| v.as_i64()).unwrap_or(0),
118                "retryAfterTime": t.get("retryAfterTime").and_then(|v| v.as_str()).unwrap_or(""),
119                "cronExpression": t.get("cronExpression"),
120                "webhookUrl": t.get("webhookUrl"),
121                "maxExecutionSeconds": t.get("maxExecutionSeconds"),
122            })
123        })
124        .collect();
125    serde_json::to_string(&res).unwrap_or_else(|_| "[]".to_string())
126}
127
128fn progress_body(ring: &ProgressRing) -> String {
129    let events: Vec<Value> = {
130        let r = ring.lock().unwrap();
131        let skip = r.len().saturating_sub(100);
132        r.iter()
133            .skip(skip)
134            .map(|ev| {
135                json!({
136                    "ts": ev.ts,
137                    "task_id": ev.task_id,
138                    "data": ev.data,
139                })
140            })
141            .collect()
142    };
143    serde_json::to_string(&events).unwrap_or_else(|_| "[]".to_string())
144}
145
146fn handle_connection(mut stream: TcpStream, tasks_path: &str, ring: &ProgressRing) {
147    let mut buf = [0u8; 4096];
148    let n = match stream.read(&mut buf) {
149        Ok(n) if n > 0 => n,
150        _ => return,
151    };
152    let req = String::from_utf8_lossy(&buf[..n]);
153    let first_line = req.lines().next().unwrap_or("");
154    let mut parts = first_line.split_whitespace();
155    let method = parts.next().unwrap_or("");
156    let target = parts.next().unwrap_or("/");
157    let path = target.split('?').next().unwrap_or("/");
158
159    let (status, content_type, body) = if method != "GET" {
160        (
161            "405 Method Not Allowed",
162            "text/plain",
163            "Method Not Allowed".to_string(),
164        )
165    } else {
166        match path {
167            "/api/stats" => ("200 OK", "application/json", stats_body(tasks_path)),
168            "/api/tasks" => ("200 OK", "application/json", tasks_body(tasks_path)),
169            "/api/progress" => ("200 OK", "application/json", progress_body(ring)),
170            "/" => match std::fs::read_to_string("static/index.html") {
171                Ok(html) => ("200 OK", "text/html", html),
172                Err(_) => (
173                    "404 Not Found",
174                    "text/plain",
175                    "Dashboard UI not found: place the dashboard bundle at ./static/index.html"
176                        .to_string(),
177                ),
178            },
179            _ => ("404 Not Found", "text/plain", "Not Found".to_string()),
180        }
181    };
182
183    let resp = format!(
184        "HTTP/1.1 {}\r\nContent-Type: {}\r\nAccess-Control-Allow-Origin: *\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
185        status,
186        content_type,
187        body.len(),
188        body
189    );
190    let _ = stream.write_all(resp.as_bytes());
191}
192
193impl SnerdQueue {
194    /// Starts the built-in dashboard UI on the given port.
195    ///
196    /// The dashboard is a single-page React app (served from ./static/index.html
197    /// relative to the process working directory) that shows live queue stats,
198    /// a Recent Jobs table, and a real-time Progress Stream fed by yield_progress.
199    /// Updates are delivered via HTTP polling of the JSON API (/api/stats,
200    /// /api/tasks, /api/progress).
201    ///
202    /// The dashboard only serves the UI — jobs keep running whether or not it is open.
203    pub fn start_dashboard(&self, port: u16) {
204        let ring: ProgressRing = Arc::new(Mutex::new(VecDeque::with_capacity(RING_CAP)));
205
206        // Feed the progress stream from the queue's internal broadcast channel
207        let mut rx = self.subscribe_progress();
208        let feeder_ring = Arc::clone(&ring);
209        std::thread::spawn(move || {
210            let rt = match tokio::runtime::Builder::new_current_thread()
211                .enable_time()
212                .build()
213            {
214                Ok(rt) => rt,
215                Err(_) => return,
216            };
217            rt.block_on(async move {
218                loop {
219                    match rx.recv().await {
220                        Ok(msg) => {
221                            let mut r = feeder_ring.lock().unwrap();
222                            r.push_back(ProgressEvent {
223                                ts: now_secs(),
224                                task_id: msg.task_id,
225                                data: msg.data,
226                            });
227                            if r.len() > RING_CAP {
228                                r.pop_front();
229                            }
230                        }
231                        Err(broadcast::error::RecvError::Lagged(_)) => continue,
232                        Err(broadcast::error::RecvError::Closed) => break,
233                    }
234                }
235            });
236        });
237
238        let tasks_path = self.file_store.file_path().to_string_lossy().to_string();
239        let listener = match TcpListener::bind(format!("0.0.0.0:{}", port)) {
240            Ok(l) => l,
241            Err(e) => {
242                println!("[Snerd] Failed to start dashboard on port {}: {}", port, e);
243                return;
244            }
245        };
246
247        println!("[Snerd] Dashboard running on http://localhost:{}", port);
248        std::thread::spawn(move || {
249            for stream in listener.incoming() {
250                if let Ok(stream) = stream {
251                    let path = tasks_path.clone();
252                    let r = Arc::clone(&ring);
253                    std::thread::spawn(move || handle_connection(stream, &path, &r));
254                }
255            }
256        });
257    }
258}
snerd_rust/dashboard.rs

snerd_rust/
dashboard.rs
