diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..50b893b --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Daniel Cuevas + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md index 4daff53..9cd841f 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,9 @@ A small durable background 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. +The project shows leases, retries, idempotency, crash recovery, priority dispatch, priority aging, a dead-letter queue, and an append-only event log. + +It also exposes Prometheus metrics and a read-only web dashboard. ## Value @@ -21,6 +23,7 @@ The queue separates durable state from worker execution. - `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/dashboard` serves a read-only web interface and JSON endpoints. - `internal/fixture` provides repeatable sample workloads. A job starts as `pending`. @@ -102,6 +105,14 @@ Use `-aging 0` to disable aging. Use `-metrics-addr` to serve Prometheus metrics beside the worker. +Serve the inspection dashboard with this command. + +```text +jobqueue web -db queue.db +``` + +Open `http://localhost:8080` in a browser. + ### `inspect` ```text @@ -154,6 +165,38 @@ Scrape the endpoint with a Prometheus server. Use `-once` to print one snapshot and exit. +### `web` + +```text +jobqueue web [-addr ] [-refresh ] [-db ] +``` + +The command serves a read-only inspection dashboard. + +The dashboard renders state counts, jobs, and recent events in one page. + +The page refreshes in place, so it needs no reload. + +Use `-addr` to change the listen address. + +The default address is `:8080`. + +Use `-refresh` to set the auto-refresh interval. + +The default interval is two seconds. + +The page links to the Prometheus endpoint at `/metrics`. + +The command never writes to the queue. + +JSON endpoints expose the same state to scripts. + +`GET /api/overview` returns the full snapshot. + +`GET /api/jobs` returns every job. + +`GET /api/jobs/` returns one job and its event timeline. + ### `demo` ```text @@ -260,6 +303,22 @@ Use the `metrics` command for one snapshot or a live endpoint. Use `work -metrics-addr` to serve the same endpoint beside a worker. +### Web dashboard + +The `web` command serves a read-only dashboard in a browser. + +The page shows state counts, jobs, recent events, and per-kind totals. + +The page refreshes in place every two seconds. + +Each refresh reads the SQLite store and renders a fresh snapshot. + +The dashboard is read-only and never modifies the queue. + +JSON endpoints back the page and support scripts. + +The `/metrics` endpoint works beside the dashboard. + ### Scheduling A scheduled job stores its earliest lease time in `run_at`. @@ -376,7 +435,7 @@ 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 serves the queue read-only, so it does not run jobs. ## Roadmap @@ -386,12 +445,24 @@ 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] Web dashboard for queue inspection. - [ ] Horizontal scaling with a shared SQLite file. ### Release notes -This release adds Prometheus metrics. +This release adds a read-only web dashboard. + +The new `web` command serves one HTML page and JSON endpoints. + +The page shows state counts, jobs, recent events, and per-kind totals. + +It refreshes in place, so it needs no reload. + +Scripts can read the same state from `/api/overview`. + +The dashboard never writes to the queue. + +The previous release added Prometheus metrics. The new `metrics` command serves the exposition format over HTTP. diff --git a/cmd/web/main.go b/cmd/web/main.go new file mode 100644 index 0000000..7444ced --- /dev/null +++ b/cmd/web/main.go @@ -0,0 +1,16 @@ +package main + +import ( + "log" + "os" + + "github.com/local-first-job-queue/internal/cli" +) + +func main() { + log.SetFlags(0) + if err := cli.Web(os.Args[1:]); err != nil { + log.Fatal(err) + os.Exit(1) + } +} diff --git a/internal/cli/demo.go b/internal/cli/demo.go index 7048f28..baed7fb 100644 --- a/internal/cli/demo.go +++ b/internal/cli/demo.go @@ -208,6 +208,7 @@ func Demo(args []string) error { fmt.Printf("\ninspect again with: jobqueue inspect -db %q\n", path) fmt.Printf("inspect a job with: jobqueue history -db %q\n", path) fmt.Printf("requeue a dead letter with: jobqueue requeue -db %q\n", path) + fmt.Printf("open the dashboard with: jobqueue web -db %q\n", path) return nil } diff --git a/internal/cli/web.go b/internal/cli/web.go new file mode 100644 index 0000000..e95dcde --- /dev/null +++ b/internal/cli/web.go @@ -0,0 +1,35 @@ +package cli + +import ( + "flag" + "fmt" + "log" + "net/http" + + "github.com/local-first-job-queue/internal/dashboard" + "github.com/local-first-job-queue/internal/queue" +) + +// Web serves the read-only inspection dashboard. The dashboard renders the +// queue state as one HTML page and exposes the same state as JSON endpoints. +// It never writes to the queue, so it is safe to leave running beside a worker. +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") + refresh := fs.Duration("refresh", dashboard.DefaultRefreshInterval, "client auto-refresh interval") + fs.Parse(args) + + store, err := queue.NewSQLiteStore(*dbPath) + if err != nil { + return fmt.Errorf("open store: %w", err) + } + defer store.Close() + + log.Printf("dashboard listening on %s (db=%s)", *addr, *dbPath) + log.Printf("open http://localhost%s", *addr) + return http.ListenAndServe(*addr, dashboard.Handler(store, + dashboard.WithDBPath(*dbPath), + dashboard.WithRefreshInterval(*refresh), + )) +} diff --git a/internal/dashboard/template.go b/internal/dashboard/template.go new file mode 100644 index 0000000..4e3136c --- /dev/null +++ b/internal/dashboard/template.go @@ -0,0 +1,288 @@ +package dashboard + +import ( + "fmt" + "html/template" + "time" + + "github.com/local-first-job-queue/internal/queue" +) + +// embeddedPage is the dashboard HTML. The page is self-contained: CSS and +// JavaScript live inline, so the binary serves it without external assets. The +// server renders the page from a fresh snapshot, and the script keeps it fresh +// by polling the JSON endpoints. +const embeddedPage = ` + + + + +Local-first Durable Job Queue + + + +
+

Local-first Durable Job Queue

+ {{if .DBPath}}db: {{.DBPath}}{{else}}db: (temp){{end}} + refresh: {{printf "%.0f" (seconds .Refresh)}}s + connecting +
+ +
+
pending
{{statecount .Overview "pending"}}
+
leased
{{statecount .Overview "leased"}}
+
completed
{{statecount .Overview "completed"}}
+
dead letter
{{statecount .Overview "dead_letter"}}
+
failed
{{statecount .Overview "failed"}}
+
oldest pending
{{age .Overview.OldestPending}}
+
events logged
{{.Overview.TotalEvents}}
+
+ +
+
+
+

Jobs ({{.Overview.TotalJobs}})

+ + + + {{range .Overview.Jobs}} + + + + + + + + + + {{else}} + + {{end}} + +
idkindprioritystateattemptscreatedrun at
{{shortid .ID}}{{.Kind}}{{.Priority}}{{.State}}{{.RetryCount}}/{{.MaxAttempts}}{{timefmt .CreatedAt}}{{optionaltime .RunAt}}
No jobs yet.
+
+ +
+

Per kind

+ {{if .Overview.ByKind}} + + + + {{range .Overview.ByKind}} + + {{end}} + +
kindstatecount
{{.Kind}}{{.State}}{{.Count}}
+ {{else}} + No jobs. + {{end}} +
+
+ +
+
+

Recent events

+ + + + {{range .Overview.Events}} + + + + + + + {{else}} + + {{end}} + +
timejobtypedetails
{{timefmt .Timestamp}}{{shortid .JobID}}{{.EventType}}{{ptrstr .Metadata}}
No events yet.
+
+
+
+ +
+ Read-only dashboard. Scrape /metrics for Prometheus output. + Use the CLI for writes: enqueue, work, inspect, history, requeue. +
+ + + + +` + +// templateFuncs are helpers used by the dashboard page. Keeping the helpers +// here keeps the embedded template free of presentation logic. +var templateFuncs = template.FuncMap{ + "shortid": func(id string) string { + if len(id) > 8 { + return id[:8] + } + return id + }, + "timefmt": func(t time.Time) string { + return t.UTC().Format("2006-01-02 15:04:05") + }, + "optionaltime": func(t *time.Time) string { + if t == nil { + return "" + } + return t.UTC().Format("2006-01-02 15:04:05") + }, + "ptrstr": func(s *string) string { + if s == nil { + return "" + } + return *s + }, + "age": func(f *float64) string { + if f == nil { + return "-" + } + return fmt.Sprintf("%.1fs", *f) + }, + "seconds": func(d time.Duration) float64 { + return d.Seconds() + }, + "statecount": func(ov Overview, state string) int { + return ov.Stats[queue.JobState(state)] + }, +} + +var dashboardTemplate = template.Must(template.New("dashboard").Funcs(templateFuncs).Parse(embeddedPage)) diff --git a/internal/dashboard/viewer.go b/internal/dashboard/viewer.go new file mode 100644 index 0000000..c890671 --- /dev/null +++ b/internal/dashboard/viewer.go @@ -0,0 +1,252 @@ +// Package dashboard serves a read-only web interface for the queue. The +// interface renders queue state as one HTML page and exposes the same state as +// JSON endpoints for live refresh. An operator can watch leases, retries, and +// the append-only event log without opening the SQLite file. +package dashboard + +import ( + "database/sql" + "encoding/json" + "errors" + "fmt" + "log" + "net/http" + "time" + + "github.com/local-first-job-queue/internal/metrics" + "github.com/local-first-job-queue/internal/queue" +) + +// MaxJobs limits the jobs rendered by one page refresh. The endpoint still +// reads the full store; the cap only bounds the payload. +const MaxJobs = 500 + +// MaxEvents limits the events rendered by one page refresh. The event log is +// append-only, so a long-lived queue can grow far beyond this window. +const MaxEvents = 200 + +// DefaultRefreshInterval is the client-side auto-refresh interval. +const DefaultRefreshInterval = 2 * time.Second + +// Option configures a Viewer. +type Option func(*Viewer) + +// WithNow overrides the clock used for age calculations. Tests use it to make +// the dashboard output deterministic. +func WithNow(fn func() time.Time) Option { + return func(v *Viewer) { + v.now = fn + } +} + +// WithRefreshInterval sets the client-side auto-refresh interval. The default +// is two seconds. +func WithRefreshInterval(d time.Duration) Option { + return func(v *Viewer) { + v.refresh = d + } +} + +// WithDBPath labels the page with the database file it inspects. The path is +// for display only; the store opens the database itself. +func WithDBPath(path string) Option { + return func(v *Viewer) { + v.dbPath = path + } +} + +// Viewer reads queue state from a store and renders it for the web. Each +// request computes a fresh snapshot, so no counters are kept between requests. +type Viewer struct { + store *queue.SQLiteStore + now func() time.Time + refresh time.Duration + dbPath string +} + +// NewViewer returns a Viewer that reads from store. +func NewViewer(store *queue.SQLiteStore, opts ...Option) *Viewer { + v := &Viewer{ + store: store, + now: time.Now, + refresh: DefaultRefreshInterval, + } + for _, o := range opts { + o(v) + } + return v +} + +// Overview is the aggregate state rendered by the dashboard. One method call +// computes it from the store, so the page always reflects the current state. +type Overview struct { + Stats map[queue.JobState]int `json:"stats"` + ByKind []queue.KindStateCount `json:"by_kind"` + EventCounts []queue.EventTypeCount `json:"event_counts"` + Jobs []queue.Job `json:"jobs"` + Events []queue.Event `json:"events"` + OldestPending *float64 `json:"oldest_pending_seconds,omitempty"` + TotalJobs int `json:"total_jobs"` + TotalEvents int `json:"total_events"` + GeneratedAt time.Time `json:"generated_at"` +} + +// Overview computes a fresh snapshot of the queue. The result is deterministic +// for a fixed store and clock: states and event types keep a stable order, and +// jobs sort by creation time with the newest first. +func (v *Viewer) Overview() (Overview, error) { + stats, err := v.store.GetQueueStats() + if err != nil { + return Overview{}, fmt.Errorf("queue stats: %w", err) + } + byKind, err := v.store.GetStateKindCounts() + if err != nil { + return Overview{}, fmt.Errorf("kind counts: %w", err) + } + eventCounts, err := v.store.GetEventTypeCounts() + if err != nil { + return Overview{}, fmt.Errorf("event counts: %w", err) + } + jobs, err := v.store.GetAllJobs() + if err != nil { + return Overview{}, fmt.Errorf("jobs: %w", err) + } + if len(jobs) > MaxJobs { + jobs = jobs[:MaxJobs] + } + events, err := v.store.GetAllEvents() + if err != nil { + return Overview{}, fmt.Errorf("events: %w", err) + } + if len(events) > MaxEvents { + events = events[:MaxEvents] + } + + ov := Overview{ + Stats: stats, + ByKind: byKind, + EventCounts: eventCounts, + Jobs: jobs, + Events: events, + GeneratedAt: v.now().UTC(), + } + for _, s := range stats { + ov.TotalJobs += s + } + for _, e := range eventCounts { + ov.TotalEvents += e.Count + } + if ready, ok, err := v.store.GetOldestPendingReadyTime(); err != nil { + return Overview{}, fmt.Errorf("oldest pending: %w", err) + } else if ok { + age := v.now().Sub(ready).Seconds() + ov.OldestPending = &age + } + return ov, nil +} + +// Handler returns an HTTP handler that serves the dashboard. The routes are: +// +// GET / the HTML dashboard +// GET /api/overview the queue snapshot as JSON +// GET /api/jobs the full job list as JSON +// GET /api/jobs/{id} one job and its event timeline as JSON +// GET /metrics the Prometheus exposition format +// +// The dashboard is read-only. No route modifies the queue. +func Handler(store *queue.SQLiteStore, opts ...Option) http.Handler { + v := NewViewer(store, opts...) + mux := http.NewServeMux() + mux.HandleFunc("GET /{$}", v.page) + mux.HandleFunc("GET /api/overview", v.apiOverview) + mux.HandleFunc("/api/overview", methodNotAllowed) + mux.HandleFunc("GET /api/jobs", v.apiJobs) + mux.HandleFunc("/api/jobs", methodNotAllowed) + mux.HandleFunc("GET /api/jobs/{id}", v.apiJob) + mux.HandleFunc("/api/jobs/{id}", methodNotAllowed) + mux.Handle("GET /metrics", metrics.Handler(store)) + mux.HandleFunc("/metrics", methodNotAllowed) + mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { + http.NotFound(w, r) + }) + return mux +} + +// methodNotAllowed answers every request that targets a known route with a +// method the route does not accept. GET-only routes stay read-only. +func methodNotAllowed(w http.ResponseWriter, r *http.Request) { + writeError(w, http.StatusMethodNotAllowed, fmt.Errorf("method %s not allowed", r.Method)) +} + +// page renders the dashboard HTML. The page is server-rendered from a fresh +// snapshot so it shows data even before the client-side script runs. +func (v *Viewer) page(w http.ResponseWriter, r *http.Request) { + ov, err := v.Overview() + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "text/html; charset=utf-8") + if err := dashboardTemplate.Execute(w, pageData{ + Overview: ov, + DBPath: v.dbPath, + Refresh: v.refresh, + }); err != nil { + log.Printf("dashboard render: %v", err) + } +} + +type pageData struct { + Overview Overview + DBPath string + Refresh time.Duration +} + +func (v *Viewer) apiOverview(w http.ResponseWriter, r *http.Request) { + ov, err := v.Overview() + if err != nil { + writeError(w, http.StatusInternalServerError, err) + return + } + writeJSON(w, http.StatusOK, ov) +} + +func (v *Viewer) apiJobs(w http.ResponseWriter, r *http.Request) { + jobs, err := v.store.GetAllJobs() + if err != nil { + writeError(w, http.StatusInternalServerError, err) + return + } + writeJSON(w, http.StatusOK, map[string]any{"jobs": jobs, "total": len(jobs)}) +} + +// apiJob serves one job and its event timeline. A job that does not exist +// produces a JSON 404, so scripts can tell a missing ID from a broken store. +func (v *Viewer) apiJob(w http.ResponseWriter, r *http.Request) { + id := r.PathValue("id") + job, err := v.store.GetJob(id) + if errors.Is(err, sql.ErrNoRows) { + writeError(w, http.StatusNotFound, fmt.Errorf("job %s not found", id)) + return + } + if err != nil { + writeError(w, http.StatusInternalServerError, err) + return + } + events, err := v.store.GetJobEvents(id) + if err != nil { + writeError(w, http.StatusInternalServerError, err) + return + } + writeJSON(w, http.StatusOK, map[string]any{"job": job, "events": events}) +} + +func writeJSON(w http.ResponseWriter, status int, v any) { + w.Header().Set("Content-Type", "application/json; charset=utf-8") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(v) +} + +func writeError(w http.ResponseWriter, status int, err error) { + writeJSON(w, status, map[string]string{"error": err.Error()}) +} diff --git a/internal/dashboard/viewer_test.go b/internal/dashboard/viewer_test.go new file mode 100644 index 0000000..0cd8017 --- /dev/null +++ b/internal/dashboard/viewer_test.go @@ -0,0 +1,306 @@ +package dashboard + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/local-first-job-queue/internal/queue" +) + +func newTestStore(t *testing.T) (*queue.SQLiteStore, *queue.Queue) { + t.Helper() + s, err := queue.NewSQLiteStore("file:dashboard_" + t.Name() + "?mode=memory&cache=shared") + if err != nil { + t.Fatalf("new store: %v", err) + } + t.Cleanup(func() { s.Close() }) + return s, queue.NewQueue(s) +} + +// TestOverviewEmptyQueue verifies that a fresh store yields an overview with +// explicit zero counts and no oldest-pending value. +func TestOverviewEmptyQueue(t *testing.T) { + s, _ := newTestStore(t) + v := NewViewer(s) + + ov, err := v.Overview() + if err != nil { + t.Fatalf("overview: %v", err) + } + if ov.TotalJobs != 0 || ov.TotalEvents != 0 { + t.Errorf("expected empty totals, got jobs=%d events=%d", ov.TotalJobs, ov.TotalEvents) + } + if ov.OldestPending != nil { + t.Errorf("expected no oldest pending, got %v", *ov.OldestPending) + } + for _, state := range []queue.JobState{queue.StatePending, queue.StateLeased, queue.StateCompleted, queue.StateDeadLetter} { + if ov.Stats[state] != 0 { + t.Errorf("expected zero %s, got %d", state, ov.Stats[state]) + } + } + if len(ov.Jobs) != 0 || len(ov.Events) != 0 { + t.Errorf("expected no jobs or events, got %d jobs %d events", len(ov.Jobs), len(ov.Events)) + } +} + +// TestOverviewReflectsWorkload verifies that the overview aggregates the store +// state. One completed email job and one dead-lettered report job produce the +// expected counts, kind rows, and event totals. +func TestOverviewReflectsWorkload(t *testing.T) { + s, q := newTestStore(t) + ctx := context.Background() + + if _, err := q.Enqueue("email", `{"to":"a@example.com"}`); err != nil { + t.Fatalf("enqueue email: %v", err) + } + flaky, err := q.Enqueue("report", `{}`, queue.WithMaxAttempts(1), queue.WithPriority(1)) + if err != nil { + t.Fatalf("enqueue flaky: %v", err) + } + // A second report job stays pending, so the overview always has something + // to measure and the kind rows include every non-terminal state. + if _, err := q.Enqueue("report", `{}`); err != nil { + t.Fatalf("enqueue report: %v", err) + } + + job, err := q.Lease(ctx, "email", time.Minute) + if err != nil || job == nil { + t.Fatalf("lease email: %v %v", job, err) + } + if err := q.Acknowledge(job.ID); err != nil { + t.Fatalf("ack: %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) + } + + ov, err := NewViewer(s).Overview() + if err != nil { + t.Fatalf("overview: %v", err) + } + if ov.TotalJobs != 3 { + t.Errorf("expected 3 jobs, got %d", ov.TotalJobs) + } + if ov.Stats[queue.StatePending] != 1 || ov.Stats[queue.StateCompleted] != 1 || ov.Stats[queue.StateDeadLetter] != 1 { + t.Errorf("unexpected stats: %+v", ov.Stats) + } + if ov.OldestPending == nil { + t.Error("expected an oldest pending age for the pending report job") + } + // The failing report job consumed its only attempt, so it dead-lettered and + // the second report job stays pending. The kind rows reflect every state. + if len(ov.ByKind) != 3 { + t.Errorf("expected 3 kind rows, got %+v", ov.ByKind) + } + byType := map[queue.EventType]int{} + for _, e := range ov.EventCounts { + byType[e.EventType] = e.Count + } + if byType[queue.EventEnqueued] != 3 { + t.Errorf("expected 3 enqueued events, got %d", byType[queue.EventEnqueued]) + } + if byType[queue.EventDeadLettered] != 1 { + t.Errorf("expected 1 dead_lettered event, got %d", byType[queue.EventDeadLettered]) + } +} + +// TestOverviewOldestPendingUsesClock verifies that the oldest-pending age uses +// the injected clock. A scheduled job in the past reports an exact age. +func TestOverviewOldestPendingUsesClock(t *testing.T) { + s, q := newTestStore(t) + + now := time.Now().UTC() + if _, err := q.Enqueue("report", `{}`, queue.WithRunAt(now.Add(-45*time.Second))); err != nil { + t.Fatalf("enqueue: %v", err) + } + + ov, err := NewViewer(s, WithNow(func() time.Time { return now })).Overview() + if err != nil { + t.Fatalf("overview: %v", err) + } + if ov.OldestPending == nil || *ov.OldestPending != 45 { + t.Fatalf("expected oldest pending 45, got %v", ov.OldestPending) + } +} + +func do(t *testing.T, h http.Handler, method, target string) *httptest.ResponseRecorder { + t.Helper() + req := httptest.NewRequest(method, target, nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + return rec +} + +func TestPageRendersDashboard(t *testing.T) { + s, q := newTestStore(t) + if _, err := q.Enqueue("email", `{"to":"a@example.com"}`); err != nil { + t.Fatalf("enqueue: %v", err) + } + + h := Handler(s, WithDBPath("queue.db"), WithRefreshInterval(3*time.Second)) + rec := do(t, h, http.MethodGet, "/") + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", rec.Code) + } + body := rec.Body.String() + for _, want := range []string{ + "Local-first Durable Job Queue", + "db: queue.db", + `id="kpi-pending"`, + "id=\"jobs-table\"", + "id=\"events-body\"", + "email", + "fetch(\"/api/overview\")", + "/metrics", + } { + if !strings.Contains(body, want) { + t.Errorf("page missing %q", want) + } + } +} + +// TestPageEscapesUserData verifies that job payloads and metadata cannot inject +// markup into the dashboard. html/template escapes user data automatically. +func TestPageEscapesUserData(t *testing.T) { + s, q := newTestStore(t) + if _, err := q.Enqueue("email", ``); err != nil { + t.Fatalf("enqueue: %v", err) + } + + h := Handler(s) + rec := do(t, h, http.MethodGet, "/") + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", rec.Code) + } + body := rec.Body.String() + if strings.Contains(body, "") { + t.Error("page rendered an unescaped script payload") + } +} + +func TestOverviewEndpointJSON(t *testing.T) { + s, q := newTestStore(t) + if _, err := q.Enqueue("email", `{}`); err != nil { + t.Fatalf("enqueue: %v", err) + } + + h := Handler(s) + rec := do(t, h, http.MethodGet, "/api/overview") + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", rec.Code) + } + if ct := rec.Header().Get("Content-Type"); !strings.Contains(ct, "application/json") { + t.Errorf("expected JSON content type, got %q", ct) + } + var ov Overview + if err := json.Unmarshal(rec.Body.Bytes(), &ov); err != nil { + t.Fatalf("decode overview: %v", err) + } + if ov.TotalJobs != 1 || ov.Stats[queue.StatePending] != 1 { + t.Errorf("unexpected overview: %+v", ov) + } +} + +func TestJobsEndpointJSON(t *testing.T) { + s, q := newTestStore(t) + job, err := q.Enqueue("email", `{}`) + if err != nil { + t.Fatalf("enqueue: %v", err) + } + + h := Handler(s) + rec := do(t, h, http.MethodGet, "/api/jobs") + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", rec.Code) + } + var out struct { + Jobs []queue.Job `json:"jobs"` + Total int `json:"total"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &out); err != nil { + t.Fatalf("decode jobs: %v", err) + } + if out.Total != 1 || len(out.Jobs) != 1 || out.Jobs[0].ID != job.ID { + t.Errorf("unexpected jobs response: %+v", out) + } +} + +func TestJobDetailEndpointJSON(t *testing.T) { + s, q := newTestStore(t) + job, err := q.Enqueue("email", `{"n":1}`) + if err != nil { + t.Fatalf("enqueue: %v", err) + } + + h := Handler(s) + rec := do(t, h, http.MethodGet, "/api/jobs/"+job.ID) + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", rec.Code) + } + var out struct { + Job queue.Job `json:"job"` + Events []queue.Event `json:"events"` + } + if err := json.Unmarshal(rec.Body.Bytes(), &out); err != nil { + t.Fatalf("decode detail: %v", err) + } + if out.Job.ID != job.ID { + t.Errorf("expected job %s, got %s", job.ID, out.Job.ID) + } + if len(out.Events) != 1 || out.Events[0].EventType != queue.EventEnqueued { + t.Errorf("expected one enqueued event, got %+v", out.Events) + } +} + +func TestJobDetailEndpointNotFound(t *testing.T) { + s, _ := newTestStore(t) + h := Handler(s) + rec := do(t, h, http.MethodGet, "/api/jobs/does-not-exist") + if rec.Code != http.StatusNotFound { + t.Fatalf("expected 404, got %d", rec.Code) + } + var out map[string]string + if err := json.Unmarshal(rec.Body.Bytes(), &out); err != nil { + t.Fatalf("decode error body: %v", err) + } + if out["error"] == "" { + t.Error("expected an error message in the 404 body") + } +} + +func TestMetricsEndpointServed(t *testing.T) { + s, q := newTestStore(t) + if _, err := q.Enqueue("email", `{}`); err != nil { + t.Fatalf("enqueue: %v", err) + } + + h := Handler(s) + rec := do(t, h, http.MethodGet, "/metrics") + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", rec.Code) + } + if !strings.Contains(rec.Body.String(), "jobqueue_jobs") { + t.Errorf("expected Prometheus families, got %q", rec.Body.String()) + } +} + +func TestUnknownPathNotFound(t *testing.T) { + s, _ := newTestStore(t) + h := Handler(s) + if rec := do(t, h, http.MethodGet, "/no-such-path"); rec.Code != http.StatusNotFound { + t.Errorf("expected 404 for unknown path, got %d", rec.Code) + } + if rec := do(t, h, http.MethodPost, "/api/overview"); rec.Code != http.StatusMethodNotAllowed { + t.Errorf("expected 405 for POST, got %d", rec.Code) + } +} diff --git a/main.go b/main.go index 9119ad9..724c90b 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 read-only inspection dashboard 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: diff --git a/page2.html b/page2.html new file mode 100644 index 0000000..bb51b72 --- /dev/null +++ b/page2.html @@ -0,0 +1,227 @@ + + + + + +Local-first Durable Job Queue + + + +
+

Local-first Durable Job Queue

+ db: queue.db + refresh: 2s + connecting +
+ +
+
pending
1
+
leased
0
+
completed
0
+
dead letter
0
+
failed
0
+
oldest pending
35531.7s
+
events logged
1
+
+ +
+
+
+

Jobs (1)

+ + + + + + + + + + + + + + + +
idkindprioritystateattemptscreatedrun at
532b5662email20pending0/32026-08-03 12:38:20
+
+ +
+

Per kind

+ + + + + + + + +
kindstatecount
emailpending1
+ +
+
+ +
+
+

Recent events

+ + + + + + + + + + + + +
timejobtypedetails
2026-08-03 12:38:20532b5662enqueued{"kind":"email"}
+
+
+
+ +
+ Read-only dashboard. Scrape /metrics for Prometheus output. + Use the CLI for writes: enqueue, work, inspect, history, requeue. +
+ + + +