queuedashboard

README

togo-framework/queue-dashboard

Queue & jobs monitoring for togo — Horizon / Sidekiq-Web for your queue.

Install

togo install togo-framework/queue-dashboard

Observe the togo queue: track every job's lifecycle (queued → running → done/failed/retried), aggregate throughput, latency and failure stats per queue, and retry or delete failed jobs — over a REST + Go API.

How it works

The togo Queue interface is Handle/Dispatch only, so the dashboard observes jobs by wrapping their handlers. Wrapped handlers record start/finish, duration and errors automatically.

Usage

import (
	queuedashboard "github.com/togo-framework/queue-dashboard"
	"github.com/togo-framework/togo"
)

func register(k *togo.Kernel) {
	qd, _ := queuedashboard.FromKernel(k)

	// Wrap handlers so jobs are tracked.
	k.Queue.Handle("emails", qd.Wrap("emails", "send", sendEmail))

	// (optional) record an enqueue for queued-depth visibility:
	id := qd.RecordDispatch("emails", "send", payload)
	_ = id
}

Inspect from Go:

qd.Jobs(queuedashboard.Filter{Queue: "emails", Status: queuedashboard.StatusFailed, Limit: 50})
qd.Stats()                       // []QueueStat{total, done, failed, failure_rate, avg_ms, ...}
qd.Retry(ctx, jobID)             // re-dispatch a failed job
qd.Delete(jobID)

REST API

Method	Path	Purpose
GET	/api/queue-dashboard/jobs?queue=&type=&status=&limit=	list recorded jobs
GET	/api/queue-dashboard/stats	per-queue health (throughput, failure rate, avg ms)
GET	/api/queue-dashboard/dashboard	stats + recent jobs (one call)
POST	/api/queue-dashboard/jobs/{id}/retry	re-dispatch a job
DELETE	/api/queue-dashboard/jobs/{id}	remove a recorded job

Configuration

No required env. Records are kept in a bounded in-memory ring (newest 2000) with a pluggable store interface for a DB-backed implementation. Pairs with togo install togo-framework/queue + a worker.

---

Premium sponsors

ID8 Media  ·  One Studio

_{Support togo — become a sponsor.}

Documentation

Overview

Package queuedashboard monitors togo queue jobs (Laravel Horizon / Sidekiq Web for togo): it records job lifecycle events, aggregates throughput / latency / failure stats, and exposes a REST + Go API to inspect, retry and delete jobs.

The togo Queue interface is Handle/Dispatch only, so the dashboard observes jobs by wrapping their handlers:

qd, _ := queuedashboard.FromKernel(k)
k.Queue.Handle("emails", qd.Wrap("emails", "send", sendEmail))

Wrapped handlers record running → done/failed with duration; RecordDispatch records an enqueue. Retry re-dispatches a recorded job over the kernel Queue.

Index

• Constants
• type Filter
• type Job
• type QueueStat
• type Service
  • func FromKernel(k *togo.Kernel) (*Service, bool)
  • func (s *Service) Delete(id string) bool
  • func (s *Service) Get(id string) (Job, bool)
  • func (s *Service) Jobs(f Filter) []Job
  • func (s *Service) RecordDispatch(queue, jobType string, payload any) string
  • func (s *Service) Retry(ctx context.Context, id string) error
  • func (s *Service) Stats() []QueueStat
  • func (s *Service) Wrap(queue, jobType string, h togo.QueueHandler) togo.QueueHandler

Constants

const (
	StatusQueued  = "queued"
	StatusRunning = "running"
	StatusDone    = "done"
	StatusFailed  = "failed"
	StatusRetried = "retried"
)

Job status values.

Types

type Filter

type Filter struct {
	Queue  string
	Type   string
	Status string
	Since  time.Time
	Limit  int
}

Filter narrows a Jobs query.

type Job

type Job struct {
	ID       string     `json:"id"`
	Queue    string     `json:"queue"`
	Type     string     `json:"type"`
	Status   string     `json:"status"`
	Attempts int        `json:"attempts"`
	Payload  any        `json:"payload,omitempty"`
	Queued   time.Time  `json:"queued"`
	Started  *time.Time `json:"started,omitempty"`
	Finished *time.Time `json:"finished,omitempty"`
	Duration int64      `json:"duration_ms,omitempty"`
	Error    string     `json:"error,omitempty"`
}

Job is a recorded queue job and its current state.

type QueueStat

type QueueStat struct {
	Queue       string  `json:"queue"`
	Total       int     `json:"total"`
	Done        int     `json:"done"`
	Failed      int     `json:"failed"`
	Running     int     `json:"running"`
	Queued      int     `json:"queued"`
	FailureRate float64 `json:"failure_rate"`
	AvgMs       int64   `json:"avg_ms"`
}

QueueStat is aggregate health for a queue.

type Service

type Service struct {
	// contains filtered or unexported fields
}

Service is the dashboard runtime stored on the kernel (k.Get("queue-dashboard")).

func FromKernel

func FromKernel(k *togo.Kernel) (*Service, bool)

FromKernel returns the dashboard Service registered on the kernel.

func (*Service) Delete

func (s *Service) Delete(id string) bool

Delete removes a recorded job.

func (*Service) Get

func (s *Service) Get(id string) (Job, bool)

Get returns a single recorded job.

func (*Service) Jobs

func (s *Service) Jobs(f Filter) []Job

Jobs returns recorded jobs matching the filter, newest first.

func (*Service) RecordDispatch

func (s *Service) RecordDispatch(queue, jobType string, payload any) string

RecordDispatch records an enqueued job and returns its id.

func (*Service) Retry

func (s *Service) Retry(ctx context.Context, id string) error

Retry re-dispatches a recorded job over the kernel Queue (best with failed jobs).

func (*Service) Stats

func (s *Service) Stats() []QueueStat

Stats aggregates per-queue health from the recorded jobs.

func (*Service) Wrap

func (s *Service) Wrap(queue, jobType string, h togo.QueueHandler) togo.QueueHandler

Wrap returns a QueueHandler that records running → done/failed with duration. Use it when registering a handler: k.Queue.Handle(name, qd.Wrap(name, type, h)).