Queviewd

A production-grade, modular background job processing system built in Go with Redis. Hand off slow work so your users never wait.

Producer (POST /enqueue)
        │
        ▼
┌───────────────────────────────────────────┐
│              Redis                         │
│  queviewd:queue:high    ← send_email      │
│  queviewd:queue:normal  ← resize_image    │
│  queviewd:queue:low     ← generate_pdf    │
│  queviewd:queue:scheduled  (delayed jobs) │
│  queviewd:processing    (crash recovery)  │
│  queviewd:queue:dead    (DLQ)             │
└───────────────────────────────────────────┘
        │
        ▼
Worker (concurrency=5, crash recovery, scheduler)
        │
        ▼
GET /metrics  →  queue depths, jobs done/failed/retried, uptime

Architecture Diagram

Features

Feature	Detail
Priority queues	HIGH (send_email) → NORMAL (resize_image) → LOW (generate_pdf). Workers drain high first.
Exponential backoff	`base * 2^attempt + jitter`, capped at 5 minutes. Respects distributed systems best practice.
Dead Letter Queue	Jobs exhausted after `MAX_RETRIES` land in a Redis sorted-set for inspection and future replay.
Scheduled jobs	`POST /schedule` with `run_at` — a scheduler goroutine promotes them at the right moment.
Crash recovery	On restart, jobs stuck mid-flight in the processing hash are automatically re-enqueued.
API key auth	`X-Api-Key` header on `/enqueue` and `/schedule`. Disabled when `API_KEY` env var is empty.
Structured logging	Every job event logged as JSON via `log/slog`. Fan-out to file + stdout.
Pluggable handlers	Add a new task type by creating one file — no switch-case to edit.
Atomic metrics	`sync/atomic` counters — zero race conditions across goroutines.
Graceful shutdown	SIGTERM drains in-flight jobs (up to 30s) before exiting.
Docker Compose	One command stands up Redis + both services.

Quick Start (Docker Compose)

# 1. Clone and copy config
cp .env.example .env          # edit API_KEY if you want

# 2. Start everything
docker-compose up --build

# 3. Enqueue a job (new terminal)
curl -X POST http://localhost:8080/enqueue \
  -H "Content-Type: application/json" \
  -H "X-Api-Key: dev-secret" \
  -d '{
    "type": "send_email",
    "payload": {
      "to": "you@example.com",
      "subject": "Hello from Queviewd!"
    },
    "max_retries": 3
  }'

# 4. Watch metrics
curl http://localhost:8081/metrics

Local Development (without Docker)

Prerequisites

Go 1.23+
Redis running on localhost:6379

# Start Redis (if you have it installed locally)
redis-server

# Copy config
cp .env.example .env

# Terminal 1 — producer
go run ./cmd/producer

# Terminal 2 — worker
go run ./cmd/worker

API Reference

Producer — `http://localhost:8080`

`GET /health`

Returns Redis connectivity status.

{ "status": "ok", "service": "queviewd-producer" }

`POST /enqueue`

Submit an immediate job.

Headers

Content-Type: application/json
X-Api-Key: <your-api-key>       (omit if API_KEY is not set)

Body

{
  "type":        "send_email",
  "payload":     { "to": "user@example.com", "subject": "Welcome!" },
  "max_retries": 3,
  "priority":    3
}

Field	Required	Description
`type`	✅	Task type — must match a registered handler
`payload`	✅	Handler-specific key-value data
`max_retries`	❌	Defaults to `MAX_RETRIES` env var (default 3)
`priority`	❌	1=low, 2=normal, 3=high. Defaults by task type.

Response 202 Accepted

{
  "job_id":   "f47ac10b-58cc-4372-a567-0e02b2c3d479",
  "type":     "send_email",
  "priority": "high",
  "queued_at": "2026-07-19T10:00:00Z",
  "message":  "job enqueued successfully — a worker will pick it up shortly"
}

Example Request:

`POST /schedule`

Submit a job to run at a specific future time.

Body — same as /enqueue plus a required run_at field:

{
  "type":    "generate_pdf",
  "payload": { "template": "report", "output_path": "/out/report.pdf" },
  "run_at":  "2026-07-19T23:00:00Z"
}

Response 202 Accepted

{
  "job_id":       "a1b2c3d4-...",
  "type":         "generate_pdf",
  "priority":     "low",
  "queued_at":    "2026-07-19T10:00:00Z",
  "scheduled_for": "2026-07-19T23:00:00Z",
  "message":      "job scheduled successfully — the worker scheduler will pick it up at run_at"
}

Worker — `http://localhost:8081`

`GET /metrics`

{
  "queue_depth_high":    0,
  "queue_depth_normal":  2,
  "queue_depth_low":     5,
  "processing_count":    3,
  "scheduled_count":     1,
  "dead_letter_count":   0,
  "jobs_done":           147,
  "jobs_failed":         2,
  "jobs_retried":        6,
  "uptime_seconds":      3604,
  "registered_task_types": ["generate_pdf", "resize_image", "send_email"]
}

Example Metrics Output:

Built-in Task Types

`send_email` — HIGH priority

{
  "type": "send_email",
  "payload": {
    "to":      "user@example.com",
    "subject": "Your order is confirmed",
    "body":    "Optional email body text"
  }
}

`resize_image` — NORMAL priority

{
  "type": "resize_image",
  "payload": {
    "src":    "/uploads/photo.jpg",
    "new_x":  800,
    "new_y":  600,
    "format": "webp"
  }
}

`generate_pdf` — LOW priority

{
  "type": "generate_pdf",
  "payload": {
    "template":    "invoice",
    "output_path": "/out/invoice-001.pdf",
    "data": {
      "customer": "Aryan",
      "amount":   499
    }
  }
}

Adding a New Task Type

Create a new file in internal/registry/handlers/:

// internal/registry/handlers/send_sms.go
package handlers

import (
    "context"
    "fmt"
    "queviewd/internal/registry"
)

func init() {
    registry.Register("send_sms", handleSendSMS)
}

func handleSendSMS(ctx context.Context, payload map[string]interface{}) error {
    to, ok := payload["to"].(string)
    if !ok || to == "" {
        return fmt.Errorf("send_sms: missing 'to' field")
    }
    // Your SMS logic here
    return nil
}

Optionally add a default priority in internal/job/job.go:

var PriorityFromType = map[string]Priority{
    "send_email":   PriorityHigh,
    "resize_image": PriorityNormal,
    "generate_pdf": PriorityLow,
    "send_sms":     PriorityHigh,   // ← add this line
}

That's it. No switch-cases to edit. Restart the worker and the new type is live.

Configuration

Copy .env.example to .env and edit as needed:

Variable	Default	Description
`REDIS_URL`	`redis://localhost:6379`	Redis connection URL
`PRODUCER_PORT`	`8080`	Producer HTTP port
`WORKER_PORT`	`8081`	Worker metrics HTTP port
`WORKER_CONCURRENCY`	`5`	Parallel goroutines processing jobs
`MAX_RETRIES`	`3`	Default max retries per job
`BACKOFF_BASE_SECONDS`	`5`	Base seconds for exponential backoff
`API_KEY`	(empty)	Required value of `X-Api-Key` header. Empty = no auth.
`LOG_FILE_PATH`	`logs/queviewd.log`	Where structured JSON logs are written
`SCHEDULER_INTERVAL_MS`	`1000`	How often the scheduler checks for ready jobs

Logs

Logs are written as JSON (fan-out to file + stdout):

{"time":"2026-07-19T10:00:01Z","level":"INFO","msg":"job enqueued","job_id":"f47a...","type":"send_email","priority":"high"}
{"time":"2026-07-19T10:00:02Z","level":"INFO","msg":"job started","worker_id":0,"job_id":"f47a...","type":"send_email","attempt":1}
{"time":"2026-07-19T10:00:02Z","level":"INFO","msg":"job completed","job_id":"f47a...","type":"send_email","duration_ms":203}
{"time":"2026-07-19T10:00:05Z","level":"WARN","msg":"job failed — scheduling retry","job_id":"...","attempt":1,"retry_in":"5.3s"}
{"time":"2026-07-19T10:00:20Z","level":"ERROR","msg":"job failed permanently — moving to DLQ","job_id":"...","total_attempts":3}

Example Worker Logs:

Project Structure

queviewd/
├── cmd/
│   ├── producer/main.go          # HTTP API: /enqueue, /schedule, /health
│   └── worker/main.go            # Job runner + /metrics + /health
├── internal/
│   ├── config/config.go          # Centralised .env loading
│   ├── job/job.go                # Core Job struct + API types
│   ├── queue/queue.go            # All Redis operations
│   ├── registry/
│   │   ├── registry.go           # Handler registration system
│   │   └── handlers/
│   │       ├── send_email.go     # HIGH priority handler
│   │       ├── resize_image.go   # NORMAL priority handler
│   │       └── generate_pdf.go   # LOW priority handler
│   ├── logger/logger.go          # Structured slog JSON logger
│   ├── metrics/metrics.go        # Atomic counters
│   └── middleware/auth.go        # X-Api-Key auth
├── logs/queviewd.log             # Runtime log file
├── .env.example                  # Configuration template
├── docker-compose.yml
├── Dockerfile.producer
├── Dockerfile.worker
├── Makefile                      # Dev convenience commands
└── go.mod

Author

Built by Aryan — GitHub: (coming soon)

Queviewd — because your users shouldn't wait for background work.

Name		Name	Last commit message	Last commit date
Latest commit History 1 Commit
cmd		cmd
internal		internal
logs		logs
.env.example		.env.example
.gitignore		.gitignore
Dockerfile.producer		Dockerfile.producer
Dockerfile.worker		Dockerfile.worker
Makefile		Makefile
README.md		README.md
WorkQueue.png		WorkQueue.png
docker-compose.yml		docker-compose.yml
go.mod		go.mod
go.sum		go.sum
image-1.png		image-1.png
image-2.png		image-2.png
image.png		image.png

Navigation Menu

Repository files navigation

Queviewd

Architecture Diagram

Features

Quick Start (Docker Compose)

Local Development (without Docker)

Prerequisites

API Reference

Producer — `http://localhost:8080`

`GET /health`

`POST /enqueue`

`POST /schedule`

Worker — `http://localhost:8081`

`GET /metrics`

Built-in Task Types

`send_email` — HIGH priority

`resize_image` — NORMAL priority

`generate_pdf` — LOW priority

Adding a New Task Type

Configuration

Logs

Project Structure

Author

About

Releases

Packages

Contributors

Languages

Folders and files

Latest commit

History

Repository files navigation

Queviewd

Architecture Diagram

Features

Quick Start (Docker Compose)

Local Development (without Docker)

Prerequisites

API Reference

Producer — http://localhost:8080

GET /health

POST /enqueue

POST /schedule

Worker — http://localhost:8081

GET /metrics

Built-in Task Types

send_email — HIGH priority

resize_image — NORMAL priority

generate_pdf — LOW priority

Adding a New Task Type

Configuration

Logs

Project Structure

Author

About

Topics

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages

Producer — `http://localhost:8080`

`GET /health`

`POST /enqueue`

`POST /schedule`

Worker — `http://localhost:8081`

`GET /metrics`

`send_email` — HIGH priority

`resize_image` — NORMAL priority

`generate_pdf` — LOW priority