From dab682b5af28787cbaa1bfc71008393038ff1448 Mon Sep 17 00:00:00 2001 From: DanieCuevas <43822444+DanielCuevas1208@users.noreply.github.com> Date: Mon, 3 Aug 2026 13:36:00 -0700 Subject: [PATCH] feat: extend local first job queue --- README.md | 174 +++++--------- internal/cli/web.go | 33 +++ internal/web/assets/app.js | 296 +++++++++++++++++++++++ internal/web/assets/index.html | 70 ++++++ internal/web/assets/style.css | 419 +++++++++++++++++++++++++++++++++ internal/web/web.go | 111 +++++++++ internal/web/web_test.go | 296 +++++++++++++++++++++++ main.go | 3 + 8 files changed, 1292 insertions(+), 110 deletions(-) create mode 100644 internal/cli/web.go create mode 100644 internal/web/assets/app.js create mode 100644 internal/web/assets/index.html create mode 100644 internal/web/assets/style.css create mode 100644 internal/web/web.go create mode 100644 internal/web/web_test.go diff --git a/README.md b/README.md index 4daff53..c30514d 100644 --- a/README.md +++ b/README.md @@ -1,26 +1,29 @@ # Local-first Durable Job Queue -A small durable background queue built with Go and SQLite. +A small durable background job queue built with Go and SQLite. -The project shows leases, retries, idempotency, crash recovery, priority dispatch, priority aging, a dead-letter queue, Prometheus metrics, and an append-only event log. +It demonstrates leases, retries, idempotency keys, crash recovery, priority dispatch, priority aging, a dead-letter queue, Prometheus metrics, and an inspectable append-only event log. ## Value Use this project to study queue behavior without an external service. -Every lease, retry, recovery, and acknowledgement remains visible in SQLite. +Every lease, retry, recovery, and acknowledgement stays visible in SQLite. The demo injects repeatable faults, so failure paths are easy to inspect. +The web dashboard shows live queue state in a browser. + ## Architecture The queue separates durable state from worker execution. -- `internal/queue` owns the public queue API and SQLite store. +- `internal/queue` owns the queue API and the SQLite store. - `internal/worker` leases jobs and runs handlers. - `internal/fault` injects deterministic errors, panics, delays, and stalls. - `internal/cli` renders commands, snapshots, history, and the demo. - `internal/metrics` renders queue state in the Prometheus text format. +- `internal/web` serves a browser dashboard and a JSON API. - `internal/fixture` provides repeatable sample workloads. A job starts as `pending`. @@ -31,7 +34,7 @@ The worker acknowledges success or records failure. An expired lease returns to the queue during recovery. -Each transition appends an event with a timestamp and optional metadata. +Each transition appends one event with a timestamp and optional metadata. ## Setup @@ -60,13 +63,25 @@ Run the deterministic showcase with this command. go run . demo ``` -Load sample jobs with this command. +Load sample jobs with these commands. ```text go run . seed -db queue.db go run . inspect -db queue.db ``` +Serve the browser dashboard with this command. + +```text +go run . web -db queue.db -addr :8080 +``` + +Open `http://localhost:8080/` in a browser. + +The dashboard shows state counts, jobs, recent events, and per-job timelines. + +The same port serves Prometheus metrics at `/metrics`. + Inspect and requeue a dead-lettered job with these commands. ```text @@ -100,8 +115,6 @@ A job gains one priority point per interval it waits. Use `-aging 0` to disable aging. -Use `-metrics-addr` to serve Prometheus metrics beside the worker. - ### `inspect` ```text @@ -110,8 +123,6 @@ jobqueue inspect [-json] [-db ] The command prints state counts, recent events, and job details. -The JSON form supports scripts and other inspection tools. - ### `history` ```text @@ -128,10 +139,6 @@ jobqueue requeue [-db ] [-max-attempts ] [-payload ] The command returns a dead-lettered job to the pending state. -The job resets its attempt count and keeps its data. - -Use `-payload` to correct the job data before it runs again. - ### `seed` ```text @@ -150,10 +157,26 @@ The command serves queue state in the Prometheus text format. The default address is `:9090`. -Scrape the endpoint with a Prometheus server. - Use `-once` to print one snapshot and exit. +### `web` + +```text +jobqueue web [-addr ] [-db ] +``` + +The command serves a browser dashboard for queue inspection. + +It serves the dashboard, a JSON API, and the Prometheus endpoint on one port. + +The default address is `:8080`. + +The page refreshes every three seconds. + +Click a job row to open its event timeline. + +The API routes are `/api/snapshot` and `/api/jobs/`. + ### `demo` ```text @@ -174,9 +197,9 @@ An expired lease becomes recoverable. Ready jobs with higher priority values lease first. -A future job cannot bypass its `run_at` time, even when its priority is higher. +A future job cannot bypass its `run_at` time. -Equal priorities use readiness time, creation time, and job ID as deterministic tie breakers. +Equal priorities use readiness time, creation time, and job ID as tie breakers. ### Priority aging @@ -184,25 +207,13 @@ A pending job gains one priority point per aging interval it waits. The interval is a store setting; the default is 30 seconds. -The `work` command enables aging by default. - -Use `-aging 0` to disable it. - -An older low-priority job can overtake a fresher high-priority job. - -The store measures the wait from the job's readiness time. - -A scheduled job starts aging only when its `run_at` time passes. - Aging prevents a constant high-priority stream from starving other work. -The `demo` command shows a low-priority job winning after five intervals. - ### Retries A failed handler returns the job to `pending` while attempts remain. -The job enters the dead-letter queue after the attempt budget is exhausted. +The job enters the dead-letter queue after the budget runs out. ### Idempotency @@ -216,11 +227,7 @@ A job that exhausts its attempts enters the `dead_letter` state. The event log records one `dead_lettered` event per exhausted job. -Use the `requeue` command to return a dead-lettered job to `pending`. - -The job keeps its data unless the command supplies a new payload. - -A requeued job resets its attempt count and can fail again. +Use `requeue` to return the job to `pending`. ### Crash recovery @@ -228,8 +235,6 @@ Startup recovery finds leases past their deadlines. Recovery consumes an attempt and records a `recovered` event. -A recovered job with no attempts left enters the dead-letter queue. - ### Event log Every state change appends one event row. @@ -244,21 +249,17 @@ Each scrape computes a fresh snapshot from the SQLite store. The exporter reports four metric families. -`jobqueue_jobs` counts jobs by state. - -`jobqueue_jobs_by_kind` counts jobs by kind and state. - -`jobqueue_events_total` counts events by type. +Every known state and event type appears with an explicit zero. -`jobqueue_oldest_pending_seconds` reports the oldest pending job's age. +### Web dashboard -Every known state and event type appears with an explicit zero. +The `web` command serves a local dashboard in a browser. -The output order stays stable across scrapes. +The page reads the same SQLite store as every other command. -Use the `metrics` command for one snapshot or a live endpoint. +No build step or network service is required. -Use `work -metrics-addr` to serve the same endpoint beside a worker. +The assets are embedded in the binary. ### Scheduling @@ -309,15 +310,6 @@ Queue state ----------- completed: 6 -Recent events (32) ------------------ - [12:00:00] requeued attempts reset to 0/3 - [12:00:00] dead_lettered attempt 3/3 exhausted: disk full - -Jobs (6) --------- - kind=demo priority=0 state=completed attempts=0/3 - Metrics ------- # HELP jobqueue_jobs Number of jobs in each state. @@ -327,17 +319,13 @@ jobqueue_jobs{state="leased"} 0 jobqueue_jobs{state="completed"} 6 jobqueue_jobs{state="dead_letter"} 0 jobqueue_jobs{state="failed"} 0 -# HELP jobqueue_events_total Number of events per event type. -# TYPE jobqueue_events_total counter -jobqueue_events_total{type="enqueued"} 5 -jobqueue_events_total{type="retried"} 5 -jobqueue_events_total{type="dead_lettered"} 1 -jobqueue_events_total{type="requeued"} 1 ``` The demo uses generated job IDs and current timestamps. -The final counts depend on the scenario and run deadline. +Run the web dashboard on the demo database to inspect the same data in a browser. + +Use `-keep` to keep the demo database after the run. ## Verification @@ -376,7 +364,9 @@ A high-priority stream can delay lower-priority jobs until aging lifts them. The worker is one process and does not coordinate across hosts. -The project does not provide a web interface. +The dashboard is read-only. + +The project does not offer horizontal scaling yet. ## Roadmap @@ -386,59 +376,23 @@ The project does not provide a web interface. - [x] Priority aging to prevent starvation. - [x] Dead-letter queue with requeue of permanently failed jobs. - [x] Prometheus metrics for queue inspection. -- [ ] Web UI for queue inspection. +- [x] Browser dashboard for queue inspection. - [ ] Horizontal scaling with a shared SQLite file. ### Release notes -This release adds Prometheus metrics. - -The new `metrics` command serves the exposition format over HTTP. - -Use `-once` to print one snapshot instead. - -The `work` command can serve the same endpoint beside a worker. - -The demo prints the final metrics snapshot. - -Each scrape reads the SQLite store and reports current state. - -The previous release added priority aging to prevent starvation. - -A pending job gains one priority point per aging interval it waits. - -The default aging interval is 30 seconds. - -The `work` command enables aging by default. - -Use `-aging 0` to disable aging. - -The store measures the wait from the job's readiness time. - -A scheduled job starts aging only when its `run_at` time passes. - -The library keeps aging opt-in, so callers keep their exact ordering. - -The demo now shows a low-priority job overtaking a fresher one. - -The previous release added a dead-letter queue for jobs that exhaust their attempts. - -A job enters the `dead_letter` state after its attempt budget runs out. - -The event log records a `dead_lettered` event for each exhausted job. - -The new `requeue` command returns a dead-lettered job to `pending`. +This release adds a browser dashboard. -The command can supply a new payload and a new attempt budget. +The new `web` command serves a local inspection page. -The demo now shows the full dead-letter workflow. +It also serves the JSON API and the Prometheus endpoint on one port. -The previous release added durable priority dispatch. +Click a job row to see its full event timeline. -Jobs store an integer priority with a default of zero. +The dashboard refreshes every three seconds. -The lease query selects ready jobs by descending priority. +The previous release added Prometheus metrics. -The migration adds `priority` to existing databases before creating its indexes. +The previous release added a dead-letter queue with requeue. -That release also preserved sub-second schedule deadlines during SQLite writes. +The previous release added durable priority dispatch and priority aging. diff --git a/internal/cli/web.go b/internal/cli/web.go new file mode 100644 index 0000000..e9456f9 --- /dev/null +++ b/internal/cli/web.go @@ -0,0 +1,33 @@ +package cli + +import ( + "flag" + "fmt" + "log" + "net/http" + + "github.com/local-first-job-queue/internal/queue" + "github.com/local-first-job-queue/internal/web" +) + +// Web starts a local browser dashboard for queue inspection. The server serves +// the dashboard, a JSON API, and the Prometheus metrics endpoint on one port. +// The dashboard reads the same SQLite store as every other command, so it +// shows live state without any extra pipeline. +func Web(args []string) error { + fs := flag.NewFlagSet("web", flag.ExitOnError) + addr := fs.String("addr", ":8080", "listen address for the dashboard") + dbPath := fs.String("db", "queue.db", "database path") + fs.Parse(args) + + store, err := queue.NewSQLiteStore(*dbPath) + if err != nil { + return fmt.Errorf("open store: %w", err) + } + defer store.Close() + + log.Printf("web dashboard listening on %s (db=%s)", *addr, *dbPath) + log.Printf("open http://localhost%s/ in a browser", *addr) + log.Printf("scrape metrics with: curl -s %s/metrics", *addr) + return http.ListenAndServe(*addr, web.New(store)) +} diff --git a/internal/web/assets/app.js b/internal/web/assets/app.js new file mode 100644 index 0000000..6b00684 --- /dev/null +++ b/internal/web/assets/app.js @@ -0,0 +1,296 @@ +(function () { + "use strict"; + + var STATE_ORDER = ["pending", "leased", "completed", "dead_letter", "failed"]; + var STATE_LABELS = { + pending: "Pending", + leased: "Leased", + completed: "Completed", + dead_letter: "Dead letter", + failed: "Failed" + }; + var REFRESH_MS = 3000; + + var els = {}; + var jobsById = {}; + var currentFilter = "all"; + var paused = false; + + function $(id) { + return document.getElementById(id); + } + + function el(tag, className, text) { + var node = document.createElement(tag); + if (className) { + node.className = className; + } + if (text !== undefined && text !== null) { + node.textContent = text; + } + return node; + } + + function init() { + els.lastUpdated = $("last-updated"); + els.refreshState = $("refresh-state"); + els.pauseBtn = $("pause-btn"); + els.stats = $("stats"); + els.jobsBody = $("jobs-table-body"); + els.jobsEmpty = $("jobs-empty"); + els.events = $("events"); + els.filter = $("state-filter"); + els.drawer = $("drawer"); + els.drawerTitle = $("drawer-title"); + els.drawerBody = $("drawer-body"); + + els.pauseBtn.addEventListener("click", togglePause); + els.drawer.addEventListener("click", function (ev) { + if (ev.target === els.drawer) { + closeDrawer(); + } + }); + $("drawer-close").addEventListener("click", closeDrawer); + + renderFilter(); + load(); + window.setInterval(function () { + if (!paused) { + load(); + } + }, REFRESH_MS); + } + + function togglePause() { + paused = !paused; + els.pauseBtn.textContent = paused ? "Resume" : "Pause"; + if (!paused) { + load(); + } + } + + function markFresh() { + els.refreshState.className = "refresh-dot is-fresh"; + els.lastUpdated.textContent = "updated " + clockTime(); + } + + function markStale(message) { + els.refreshState.className = "refresh-dot is-stale"; + els.lastUpdated.textContent = "refresh failed: " + message; + } + + function clockTime() { + var d = new Date(); + var hh = String(d.getHours()).padStart(2, "0"); + var mm = String(d.getMinutes()).padStart(2, "0"); + var ss = String(d.getSeconds()).padStart(2, "0"); + return hh + ":" + mm + ":" + ss; + } + + function load() { + fetch("/api/snapshot") + .then(function (res) { + if (!res.ok) { + throw new Error("HTTP " + res.status); + } + return res.json(); + }) + .then(function (snap) { + jobsById = {}; + (snap.jobs || []).forEach(function (job) { + jobsById[job.id] = job; + }); + renderStats(snap.stats || {}); + renderJobs(); + renderEvents(snap.events || []); + markFresh(); + }) + .catch(function (err) { + markStale(err.message); + }); + } + + function renderStats(stats) { + els.stats.innerHTML = ""; + STATE_ORDER.forEach(function (state) { + var count = stats[state] || 0; + var card = el("div", "stat-card stat-" + state); + card.appendChild(el("span", "stat-count", String(count))); + card.appendChild(el("span", "stat-label", STATE_LABELS[state])); + els.stats.appendChild(card); + }); + } + + function renderFilter() { + els.filter.innerHTML = ""; + ["all"].concat(STATE_ORDER).forEach(function (name) { + var btn = el( + "button", + "filter-btn" + (name === currentFilter ? " is-active" : ""), + name === "all" ? "All" : STATE_LABELS[name] + ); + btn.type = "button"; + btn.dataset.state = name; + btn.addEventListener("click", function () { + currentFilter = name; + renderFilter(); + renderJobs(); + }); + els.filter.appendChild(btn); + }); + } + + function renderJobs() { + var jobs = []; + Object.keys(jobsById).forEach(function (key) { + jobs.push(jobsById[key]); + }); + jobs.sort(function (a, b) { + return String(b.created_at).localeCompare(String(a.created_at)); + }); + + var visible = + currentFilter === "all" + ? jobs + : jobs.filter(function (job) { + return job.state === currentFilter; + }); + + els.jobsBody.innerHTML = ""; + els.jobsEmpty.hidden = visible.length !== 0; + + visible.forEach(function (job) { + var tr = el("tr", "job-row"); + tr.tabIndex = 0; + tr.addEventListener("click", function () { + openJob(job.id); + }); + tr.addEventListener("keydown", function (ev) { + if (ev.key === "Enter" || ev.key === " ") { + openJob(job.id); + } + }); + + tr.appendChild(el("td", "job-id", shortID(job.id))); + tr.appendChild(el("td", "", job.kind)); + tr.appendChild(el("td", "", badgeFor(job.state))); + tr.appendChild(el("td", "num", String(job.priority))); + tr.appendChild(el("td", "num", job.retry_count + "/" + job.max_attempts)); + tr.appendChild(el("td", "", formatTime(job.created_at))); + tr.appendChild(el("td", "", job.run_at ? formatTime(job.run_at) : "now")); + + els.jobsBody.appendChild(tr); + }); + } + + function badgeFor(state) { + var label = STATE_LABELS[state] || state; + var node = el("span", "badge badge-" + (state || "pending"), label); + return node; + } + + function renderEvents(events) { + els.events.innerHTML = ""; + if (!events.length) { + els.events.appendChild(el("li", "empty", "No events yet.")); + return; + } + events.forEach(function (event) { + var li = el("li", ""); + li.appendChild(el("span", "ts", "[" + formatTime(event.timestamp) + "] ")); + li.appendChild(el("span", "event-type", event.event_type)); + if (event.metadata) { + li.appendChild(el("span", "event-meta", " " + event.metadata)); + } + els.events.appendChild(li); + }); + } + + function openJob(id) { + fetch("/api/jobs/" + encodeURIComponent(id)) + .then(function (res) { + if (!res.ok) { + throw new Error("HTTP " + res.status); + } + return res.json(); + }) + .then(function (data) { + renderJobDetail(data.job, data.events || []); + els.drawer.hidden = false; + }) + .catch(function (err) { + window.alert("Could not load job: " + err.message); + }); + } + + function closeDrawer() { + els.drawer.hidden = true; + els.drawerBody.innerHTML = ""; + } + + function renderJobDetail(job, events) { + els.drawerTitle.textContent = "Job " + shortID(job.id); + + var dl = el("dl", "detail-grid"); + addDetail(dl, "ID", job.id); + addDetail(dl, "Kind", job.kind); + addDetail(dl, "State", job.state); + addDetail(dl, "Priority", String(job.priority)); + addDetail(dl, "Attempts", job.retry_count + "/" + job.max_attempts); + if (job.idempotency_key) { + addDetail(dl, "Idempotency", job.idempotency_key); + } + addDetail(dl, "Created", formatTime(job.created_at)); + addDetail(dl, "Updated", formatTime(job.updated_at)); + if (job.leased_until) { + addDetail(dl, "Leased until", formatTime(job.leased_until)); + } + if (job.run_at) { + addDetail(dl, "Run at", formatTime(job.run_at)); + } + addDetail(dl, "Payload", job.payload); + + var section = el("div", "detail-section"); + section.appendChild(el("h3", "", "Event log (" + events.length + ")")); + var ol = el("ol", "timeline"); + events.forEach(function (event) { + var li = el("li", ""); + li.appendChild(el("span", "ts", formatTime(event.timestamp) + " ")); + li.appendChild(el("span", "event-type", event.event_type)); + if (event.metadata) { + li.appendChild(el("span", "event-meta", " " + event.metadata)); + } + ol.appendChild(li); + }); + section.appendChild(ol); + + els.drawerBody.innerHTML = ""; + els.drawerBody.appendChild(dl); + els.drawerBody.appendChild(section); + } + + function addDetail(dl, label, value) { + dl.appendChild(el("dt", "", label)); + dl.appendChild(el("dd", "", value === null || value === undefined ? "" : String(value))); + } + + function shortID(id) { + if (!id) { + return ""; + } + return id.length > 8 ? id.slice(0, 8) : id; + } + + function formatTime(value) { + if (!value) { + return ""; + } + var d = new Date(value); + if (isNaN(d.getTime())) { + return String(value); + } + return d.toLocaleString(); + } + + document.addEventListener("DOMContentLoaded", init); +})(); diff --git a/internal/web/assets/index.html b/internal/web/assets/index.html new file mode 100644 index 0000000..3185b40 --- /dev/null +++ b/internal/web/assets/index.html @@ -0,0 +1,70 @@ + + + + + + Local-first Job Queue + + + +
+
+

Local-first Job Queue

+
+ + connecting + +
+
+
+ +
+
+ +
+
+
+

Jobs

+
+
+
+ + + + + + + + + + + + + +
IDKindStatePriorityAttemptsCreatedRun at
+ +
+
+ + +
+
+ + + + + + diff --git a/internal/web/assets/style.css b/internal/web/assets/style.css new file mode 100644 index 0000000..203bcd3 --- /dev/null +++ b/internal/web/assets/style.css @@ -0,0 +1,419 @@ +:root { + --bg: #f4f6f8; + --panel: #ffffff; + --border: #dbe1e8; + --text: #1c2333; + --text-muted: #5b6572; + --accent: #2f6fed; + --topbar: #1c2333; + --topbar-text: #e8edf5; + --font-mono: ui-monospace, SFMono-Regular, Menlo, Consolas, "Liberation Mono", monospace; + --font-sans: system-ui, -apple-system, "Segoe UI", Roboto, sans-serif; +} + +* { + box-sizing: border-box; +} + +body { + margin: 0; + background: var(--bg); + color: var(--text); + font-family: var(--font-sans); + line-height: 1.5; +} + +.topbar { + background: var(--topbar); + color: var(--topbar-text); +} + +.topbar-inner { + max-width: 1180px; + margin: 0 auto; + padding: 14px 20px; + display: flex; + align-items: center; + justify-content: space-between; + gap: 16px; +} + +.topbar-title { + margin: 0; + font-size: 18px; + font-weight: 600; + letter-spacing: 0.02em; +} + +.topbar-meta { + display: flex; + align-items: center; + gap: 10px; + color: var(--topbar-text); + font-size: 13px; +} + +.last-updated { + opacity: 0.85; +} + +.refresh-dot { + width: 9px; + height: 9px; + border-radius: 50%; + display: inline-block; +} + +.refresh-dot.is-fresh { + background: #4ade80; +} + +.refresh-dot.is-stale { + background: #f87171; +} + +.button { + border: 1px solid var(--border); + border-radius: 6px; + background: var(--panel); + color: var(--text); + font-family: var(--font-sans); + font-size: 13px; + cursor: pointer; + padding: 6px 12px; +} + +.button:hover { + border-color: var(--accent); + color: var(--accent); +} + +.button-small { + padding: 3px 10px; +} + +.layout { + max-width: 1180px; + margin: 0 auto; + padding: 20px; +} + +.stats { + display: grid; + grid-template-columns: repeat(5, 1fr); + gap: 12px; + margin-bottom: 20px; +} + +.stat-card { + background: var(--panel); + border: 1px solid var(--border); + border-radius: 8px; + padding: 14px 16px; + display: flex; + flex-direction: column; + gap: 2px; + border-left: 4px solid var(--border); +} + +.stat-count { + font-size: 26px; + font-weight: 700; + font-variant-numeric: tabular-nums; +} + +.stat-label { + font-size: 12px; + color: var(--text-muted); + text-transform: uppercase; + letter-spacing: 0.05em; +} + +.stat-pending { + border-left-color: #d97706; +} + +.stat-leased { + border-left-color: #2563eb; +} + +.stat-completed { + border-left-color: #16a34a; +} + +.stat-dead_letter { + border-left-color: #dc2626; +} + +.stat-failed { + border-left-color: #7c2d12; +} + +.grid { + display: grid; + grid-template-columns: 1fr 380px; + gap: 20px; + align-items: start; +} + +.panel { + background: var(--panel); + border: 1px solid var(--border); + border-radius: 8px; + overflow: hidden; +} + +.panel-head { + display: flex; + align-items: center; + justify-content: space-between; + gap: 12px; + flex-wrap: wrap; + padding: 12px 16px; + border-bottom: 1px solid var(--border); +} + +.panel-title { + margin: 0; + font-size: 15px; + font-weight: 600; +} + +.filter { + display: flex; + gap: 4px; + flex-wrap: wrap; +} + +.filter-btn { + border: 1px solid var(--border); + background: transparent; + border-radius: 14px; + padding: 2px 10px; + font-size: 12px; + color: var(--text-muted); + cursor: pointer; + font-family: var(--font-sans); +} + +.filter-btn:hover { + border-color: var(--accent); + color: var(--accent); +} + +.filter-btn.is-active { + background: var(--accent); + border-color: var(--accent); + color: #ffffff; +} + +.table-wrap { + overflow-x: auto; +} + +.jobs-table { + width: 100%; + border-collapse: collapse; + font-size: 13px; +} + +.jobs-table th, +.jobs-table td { + text-align: left; + padding: 8px 16px; + border-bottom: 1px solid var(--border); + white-space: nowrap; +} + +.jobs-table th { + background: #fafbfc; + color: var(--text-muted); + font-weight: 600; + font-size: 12px; + text-transform: uppercase; + letter-spacing: 0.04em; +} + +.jobs-table .num { + text-align: right; +} + +.jobs-table tbody tr { + cursor: pointer; +} + +.jobs-table tbody tr:hover { + background: #f2f6fc; +} + +.job-id { + font-family: var(--font-mono); + font-size: 12px; +} + +.badge { + display: inline-block; + border-radius: 12px; + padding: 1px 9px; + font-size: 12px; + font-weight: 600; +} + +.badge-pending { + background: #fef3c7; + color: #92400e; +} + +.badge-leased { + background: #dbeafe; + color: #1e40af; +} + +.badge-completed { + background: #dcfce7; + color: #166534; +} + +.badge-dead_letter { + background: #fee2e2; + color: #991b1b; +} + +.badge-failed { + background: #ffedd5; + color: #7c2d12; +} + +.empty { + margin: 0; + padding: 24px 16px; + color: var(--text-muted); + text-align: center; + font-size: 13px; +} + +.events { + list-style: none; + margin: 0; + padding: 0; + max-height: 520px; + overflow-y: auto; + font-family: var(--font-mono); + font-size: 12px; +} + +.events li { + padding: 8px 16px; + border-bottom: 1px solid var(--border); + overflow-wrap: anywhere; +} + +.event-type { + font-weight: 600; + color: var(--accent); +} + +.event-meta { + color: var(--text-muted); +} + +.drawer { + position: fixed; + inset: 0; + background: rgba(12, 16, 24, 0.45); + display: flex; + justify-content: flex-end; + z-index: 20; +} + +.drawer-panel { + width: min(480px, 100%); + background: var(--panel); + height: 100%; + overflow-y: auto; + box-shadow: -8px 0 24px rgba(12, 16, 24, 0.18); + animation: slide-in 0.18s ease-out; +} + +@keyframes slide-in { + from { + transform: translateX(24px); + opacity: 0; + } + to { + transform: translateX(0); + opacity: 1; + } +} + +.drawer-head { + display: flex; + align-items: center; + justify-content: space-between; + padding: 14px 20px; + border-bottom: 1px solid var(--border); + position: sticky; + top: 0; + background: var(--panel); +} + +.drawer-body { + padding: 16px 20px; +} + +.detail-grid { + display: grid; + grid-template-columns: 130px 1fr; + gap: 4px 14px; + font-size: 13px; +} + +.detail-grid dt { + color: var(--text-muted); +} + +.detail-grid dd { + margin: 0; + font-family: var(--font-mono); + overflow-wrap: anywhere; +} + +.detail-section { + margin-top: 18px; +} + +.detail-section h3 { + margin: 0 0 8px; + font-size: 13px; + text-transform: uppercase; + letter-spacing: 0.04em; + color: var(--text-muted); +} + +.timeline { + list-style: none; + margin: 0; + padding: 0; + font-family: var(--font-mono); + font-size: 12px; +} + +.timeline li { + padding: 6px 0; + border-bottom: 1px dashed var(--border); + overflow-wrap: anywhere; +} + +.timeline .ts { + color: var(--text-muted); +} + +@media (max-width: 860px) { + .stats { + grid-template-columns: repeat(2, 1fr); + } + + .grid { + grid-template-columns: 1fr; + } +} diff --git a/internal/web/web.go b/internal/web/web.go new file mode 100644 index 0000000..b38a093 --- /dev/null +++ b/internal/web/web.go @@ -0,0 +1,111 @@ +// Package web serves a local browser dashboard for queue inspection. The +// dashboard reads the shared SQLite store through the queue API and renders +// state counts, jobs, recent events, and per-job timelines. Assets are +// embedded in the binary, so the dashboard needs no build step or network. +package web + +import ( + "database/sql" + "embed" + "encoding/json" + "errors" + "fmt" + "io/fs" + "net/http" + + "github.com/local-first-job-queue/internal/metrics" + "github.com/local-first-job-queue/internal/queue" +) + +//go:embed assets +var assets embed.FS + +var indexHTML = readIndex() + +// assetRoot is the embedded filesystem rooted at the assets directory. It +// serves style.css and app.js at /assets/ after the prefix strip. +var assetRoot = mustSub() + +func mustSub() fs.FS { + root, err := fs.Sub(assets, "assets") + if err != nil { + panic("web: embedded assets missing: " + err.Error()) + } + return root +} + +func readIndex() []byte { + b, err := assets.ReadFile("assets/index.html") + if err != nil { + panic("web: embedded index.html missing: " + err.Error()) + } + return b +} + +// New returns an HTTP handler for the inspection dashboard. +// +// Routes: +// +// / the dashboard page +// /assets/* embedded styles and scripts +// /api/snapshot queue state as JSON +// /api/jobs/{id} one job and its event timeline +// /metrics the Prometheus exposition endpoint +func New(store *queue.SQLiteStore) http.Handler { + q := queue.NewQueue(store) + mux := http.NewServeMux() + + mux.HandleFunc("GET /api/snapshot", func(w http.ResponseWriter, r *http.Request) { + snap, err := q.Inspect() + if err != nil { + writeError(w, http.StatusInternalServerError, err) + return + } + writeJSON(w, snap) + }) + + mux.HandleFunc("GET /api/jobs/{id}", func(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + job, err := store.GetJob(id) + if errors.Is(err, sql.ErrNoRows) { + writeError(w, http.StatusNotFound, errors.New("job not found")) + return + } + if err != nil { + writeError(w, http.StatusInternalServerError, err) + return + } + events, err := store.GetJobEvents(id) + if err != nil { + writeError(w, http.StatusInternalServerError, err) + return + } + writeJSON(w, map[string]any{"job": job, "events": events}) + }) + + mux.Handle("GET /metrics", metrics.Handler(store)) + mux.Handle("GET /assets/", http.StripPrefix("/assets/", http.FileServer(http.FS(assetRoot)))) + mux.HandleFunc("GET /", func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/" { + http.NotFound(w, r) + return + } + w.Header().Set("Content-Type", "text/html; charset=utf-8") + _, _ = w.Write(indexHTML) + }) + + return mux +} + +func writeJSON(w http.ResponseWriter, v any) { + w.Header().Set("Content-Type", "application/json; charset=utf-8") + if err := json.NewEncoder(w).Encode(v); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + } +} + +func writeError(w http.ResponseWriter, status int, err error) { + w.Header().Set("Content-Type", "application/json; charset=utf-8") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(map[string]string{"error": fmt.Sprintf("%v", err)}) +} diff --git a/internal/web/web_test.go b/internal/web/web_test.go new file mode 100644 index 0000000..4870da1 --- /dev/null +++ b/internal/web/web_test.go @@ -0,0 +1,296 @@ +package web + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/local-first-job-queue/internal/queue" +) + +func newTestServer(t *testing.T) (*httptest.Server, *queue.SQLiteStore, *queue.Queue) { + t.Helper() + s, err := queue.NewSQLiteStore("file:web_" + t.Name() + "?mode=memory&cache=shared") + if err != nil { + t.Fatalf("new store: %v", err) + } + t.Cleanup(func() { s.Close() }) + ts := httptest.NewServer(New(s)) + t.Cleanup(ts.Close) + return ts, s, queue.NewQueue(s) +} + +func get(t *testing.T, url string) (*http.Response, string) { + t.Helper() + res, err := http.Get(url) + if err != nil { + t.Fatalf("get %s: %v", url, err) + } + defer res.Body.Close() + b, err := io.ReadAll(res.Body) + if err != nil { + t.Fatalf("read body: %v", err) + } + return res, string(b) +} + +// TestDashboardServesPage verifies that the root path returns the dashboard +// HTML and the correct content type. +func TestDashboardServesPage(t *testing.T) { + ts, _, _ := newTestServer(t) + + res, body := get(t, ts.URL+"/") + if res.StatusCode != http.StatusOK { + t.Fatalf("expected 200, got %d", res.StatusCode) + } + if ct := res.Header.Get("Content-Type"); !strings.Contains(ct, "text/html") { + t.Errorf("expected html content type, got %q", ct) + } + for _, want := range []string{ + "Local-first Job Queue", + "Recent events", + "/assets/app.js", + "/assets/style.css", + } { + if !strings.Contains(body, want) { + t.Errorf("dashboard missing %q", want) + } + } +} + +// TestUnknownPathReturns404 verifies that unmatched paths get a 404 instead of +// the dashboard page. +func TestUnknownPathReturns404(t *testing.T) { + ts, _, _ := newTestServer(t) + res, _ := get(t, ts.URL+"/does-not-exist") + if res.StatusCode != http.StatusNotFound { + t.Fatalf("expected 404, got %d", res.StatusCode) + } +} + +// TestAssetsServed verifies that the embedded stylesheet and script are served +// with the correct content types. +func TestAssetsServed(t *testing.T) { + ts, _, _ := newTestServer(t) + + for _, tc := range []struct { + path string + contentType string + marker string + }{ + {"/assets/style.css", "text/css", "--topbar"}, + {"/assets/app.js", "javascript", "REFRESH_MS"}, + } { + res, body := get(t, ts.URL+tc.path) + if res.StatusCode != http.StatusOK { + t.Errorf("%s: expected 200, got %d", tc.path, res.StatusCode) + } + if ct := res.Header.Get("Content-Type"); !strings.Contains(ct, tc.contentType) { + t.Errorf("%s: expected content type %q, got %q", tc.path, tc.contentType, ct) + } + if !strings.Contains(body, tc.marker) { + t.Errorf("%s: missing marker %q", tc.path, tc.marker) + } + } +} + +// TestMetricsMounted verifies that the dashboard server also exposes the +// Prometheus endpoint, so one port serves both the dashboard and metrics. +func TestMetricsMounted(t *testing.T) { + ts, _, _ := newTestServer(t) + res, body := get(t, ts.URL+"/metrics") + if res.StatusCode != http.StatusOK { + t.Fatalf("expected 200, got %d", res.StatusCode) + } + if !strings.Contains(body, "# TYPE jobqueue_jobs gauge") { + t.Errorf("expected metrics body, got %q", body) + } +} + +// TestAPISnapshotReflectsQueue verifies that the snapshot endpoint reports the +// jobs, event counts, and state statistics that exist in the store. +func TestAPISnapshotReflectsQueue(t *testing.T) { + ts, _, q := newTestServer(t) + ctx := context.Background() + + if _, err := q.Enqueue("email", `{"to":"a@example.com"}`); err != nil { + t.Fatalf("enqueue: %v", err) + } + flaky, err := q.Enqueue("report", `{}`, queue.WithMaxAttempts(1), queue.WithPriority(1)) + if err != nil { + t.Fatalf("enqueue flaky: %v", err) + } + leased, err := q.Lease(ctx, "report", time.Minute) + if err != nil || leased == nil || leased.ID != flaky.ID { + t.Fatalf("lease flaky: %v %v", leased, err) + } + if err := q.Fail(leased.ID, "boom"); err != nil { + t.Fatalf("fail: %v", err) + } + + res, body := get(t, ts.URL+"/api/snapshot") + if res.StatusCode != http.StatusOK { + t.Fatalf("expected 200, got %d", res.StatusCode) + } + if ct := res.Header.Get("Content-Type"); !strings.Contains(ct, "application/json") { + t.Errorf("expected json content type, got %q", ct) + } + + var snap queue.QueueSnapshot + if err := json.Unmarshal([]byte(body), &snap); err != nil { + t.Fatalf("decode snapshot: %v\n%s", err, body) + } + if len(snap.Jobs) != 2 { + t.Errorf("expected 2 jobs, got %d", len(snap.Jobs)) + } + if snap.Stats[queue.StatePending] != 1 { + t.Errorf("expected 1 pending, got %d", snap.Stats[queue.StatePending]) + } + if snap.Stats[queue.StateDeadLetter] != 1 { + t.Errorf("expected 1 dead_letter, got %d", snap.Stats[queue.StateDeadLetter]) + } + if len(snap.Events) != 4 { + t.Errorf("expected 4 events, got %d", len(snap.Events)) + } +} + +// TestAPIJobDetail verifies that the job detail endpoint returns the job and +// its full event timeline in order. +func TestAPIJobDetail(t *testing.T) { + ts, _, q := newTestServer(t) + ctx := context.Background() + + job, err := q.Enqueue("email", `{"to":"a@example.com"}`, queue.WithMaxAttempts(2)) + if err != nil { + t.Fatalf("enqueue: %v", err) + } + leased, err := q.Lease(ctx, "email", time.Minute) + if err != nil || leased == nil || leased.ID != job.ID { + t.Fatalf("lease: %v %v", leased, err) + } + if err := q.Fail(leased.ID, "rate limited"); err != nil { + t.Fatalf("fail: %v", err) + } + + res, body := get(t, ts.URL+"/api/jobs/"+job.ID) + if res.StatusCode != http.StatusOK { + t.Fatalf("expected 200, got %d: %s", res.StatusCode, body) + } + + var detail struct { + Job queue.Job `json:"job"` + Events []queue.Event `json:"events"` + } + if err := json.Unmarshal([]byte(body), &detail); err != nil { + t.Fatalf("decode detail: %v\n%s", err, body) + } + if detail.Job.ID != job.ID { + t.Errorf("expected job %s, got %s", job.ID, detail.Job.ID) + } + if detail.Job.State != queue.StatePending { + t.Errorf("expected pending after one retry, got %s", detail.Job.State) + } + wantTypes := []queue.EventType{ + queue.EventEnqueued, queue.EventLeased, queue.EventRetried, + } + if len(detail.Events) != len(wantTypes) { + t.Fatalf("expected %d events, got %d", len(wantTypes), len(detail.Events)) + } + for i, et := range wantTypes { + if detail.Events[i].EventType != et { + t.Errorf("event %d: expected %s, got %s", i, et, detail.Events[i].EventType) + } + } +} + +// TestAPIJobNotFound verifies that an unknown job id returns a 404 with a JSON +// error body. +func TestAPIJobNotFound(t *testing.T) { + ts, _, _ := newTestServer(t) + + res, body := get(t, ts.URL+"/api/jobs/nope") + if res.StatusCode != http.StatusNotFound { + t.Fatalf("expected 404, got %d", res.StatusCode) + } + var errBody map[string]string + if err := json.Unmarshal([]byte(body), &errBody); err != nil { + t.Fatalf("decode error body: %v\n%s", err, body) + } + if errBody["error"] == "" { + t.Errorf("expected error message, got %q", body) + } +} + +// TestAPIJobDetailTimelineIsAppendOnly verifies that the job detail endpoint +// returns events in insertion order. A job that crashes, exhausts its budget, +// and is requeued records every transition: enqueued, leased, dead_lettered, +// requeued. +func TestAPIJobDetailTimelineIsAppendOnly(t *testing.T) { + ts, s, q := newTestServer(t) + + job, err := q.Enqueue("report", `{}`, queue.WithMaxAttempts(1)) + if err != nil { + t.Fatalf("enqueue: %v", err) + } + // Simulate a worker crash: lease with an expired deadline, then recover. + // The single attempt budget forces the job into the dead-letter state. + if _, err := s.LeaseJobByID(job.ID, -time.Hour); err != nil { + t.Fatalf("orphan lease: %v", err) + } + if _, err := q.Recover(); err != nil { + t.Fatalf("recover: %v", err) + } + if _, err := q.Requeue(job.ID); err != nil { + t.Fatalf("requeue: %v", err) + } + + res, body := get(t, ts.URL+"/api/jobs/"+job.ID) + if res.StatusCode != http.StatusOK { + t.Fatalf("expected 200, got %d", res.StatusCode) + } + var detail struct { + Job queue.Job `json:"job"` + Events []queue.Event `json:"events"` + } + if err := json.Unmarshal([]byte(body), &detail); err != nil { + t.Fatalf("decode detail: %v\n%s", err, body) + } + wantTypes := []queue.EventType{ + queue.EventEnqueued, + queue.EventDeadLettered, + queue.EventRequeued, + } + if len(detail.Events) != len(wantTypes) { + t.Fatalf("expected %d events, got %d: %+v", len(wantTypes), len(detail.Events), detail.Events) + } + for i, et := range wantTypes { + if detail.Events[i].EventType != et { + t.Errorf("event %d: expected %s, got %s", i, et, detail.Events[i].EventType) + } + } + if detail.Job.State != queue.StatePending { + t.Errorf("expected pending after requeue, got %s", detail.Job.State) + } +} + +// TestAPISnapshotEmptyQueue verifies that the snapshot endpoint returns an +// empty result for a fresh store. +func TestAPISnapshotEmptyQueue(t *testing.T) { + ts, _, _ := newTestServer(t) + res, body := get(t, ts.URL+"/api/snapshot") + if res.StatusCode != http.StatusOK { + t.Fatalf("expected 200, got %d", res.StatusCode) + } + var snap queue.QueueSnapshot + if err := json.Unmarshal([]byte(body), &snap); err != nil { + t.Fatalf("decode snapshot: %v", err) + } + if len(snap.Jobs) != 0 || len(snap.Events) != 0 || len(snap.Stats) != 0 { + t.Errorf("expected empty snapshot, got %+v", snap) + } +} diff --git a/main.go b/main.go index 9119ad9..7989638 100644 --- a/main.go +++ b/main.go @@ -23,6 +23,7 @@ Commands: requeue Return a dead-lettered job to the queue seed Load bundled sample data metrics Expose queue state for Prometheus + web Serve a browser dashboard for queue inspection demo Run a self-contained scenario with fault injection Use -help for command flags.`) @@ -48,6 +49,8 @@ Use -help for command flags.`) err = cli.Seed(args) case "metrics": err = cli.Metrics(args) + case "web": + err = cli.Web(args) case "demo": err = cli.Demo(args) default: