Asynqpg

https://github.com/yakser/asynqpg
Admin
Asynqpg is a distributed task queue for Go backed by PostgreSQL, enabling safe concurrent task...
Tokens:15,614
Snippets:138
Trust Score:8.4
License:MIT
Update:6 months ago
Show doc for...
Tokens:
Raw
### Install asynqpg

Source: https://github.com/yakser/asynqpg/blob/master/README.md

Use go get to install the asynqpg library. Requires Go 1.25+ and PostgreSQL 14+.

```bash
go get github.com/yakser/asynqpg
```

--------------------------------

### Development Setup Commands

Source: https://github.com/yakser/asynqpg/blob/master/CONTRIBUTING.md

Commands to set up the development environment, including starting services and migrating databases.

```bash
make up
make migrate
```

--------------------------------

### Implement Full Asynqpg Application

Source: https://context7.com/yakser/asynqpg/llms.txt

Demonstrates a complete setup including producer initialization, consumer configuration with middleware, task handler registration, and graceful shutdown handling.

```go
package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "log/slog"
    "net/http"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/jmoiron/sqlx"
    _ "github.com/lib/pq"
    "github.com/yakser/asynqpg"
    "github.com/yakser/asynqpg/consumer"
    "github.com/yakser/asynqpg/producer"
    "github.com/yakser/asynqpg/ui"
)

type EmailPayload struct {
    To      string `json:"to"`
    Subject string `json:"subject"`
    Body    string `json:"body"`
}

func main() {
    db, err := sqlx.Connect("postgres", "postgres://postgres:password@localhost:5432/asynqpg?sslmode=disable")
    if err != nil {
        log.Fatal(err)
    }

    // Create producer
    p, err := producer.New(producer.Config{
        Pool:            db,
        DefaultMaxRetry: 3,
    })
    if err != nil {
        log.Fatal(err)
    }

    // Create consumer
    c, err := consumer.New(consumer.Config{
        Pool:               db,
        ClientID:           "worker-main",
        FetchInterval:      100 * time.Millisecond,
        ShutdownTimeout:    30 * time.Second,
        CompletedRetention: 24 * time.Hour,
        FailedRetention:    7 * 24 * time.Hour,
    })
    if err != nil {
        log.Fatal(err)
    }

    // Global logging middleware
    _ = c.Use(func(next consumer.TaskHandler) consumer.TaskHandler {
        return consumer.TaskHandlerFunc(func(ctx context.Context, task *asynqpg.TaskInfo) error {
            start := time.Now()
            err := next.Handle(ctx, task)
            slog.Info("task processed", "type", task.Type, "id", task.ID, "duration", time.Since(start), "error", err)
            return err
        })
    })

    // Register email handler
    _ = c.RegisterTaskHandler("email:send",
        consumer.TaskHandlerFunc(func(ctx context.Context, task *asynqpg.TaskInfo) error {
            var payload EmailPayload
            if err := json.Unmarshal(task.Payload, &payload); err != nil {
                return fmt.Errorf("invalid payload: %w", asynqpg.ErrSkipRetry)
            }
            slog.Info("sending email", "to", payload.To, "subject", payload.Subject)
            time.Sleep(100 * time.Millisecond) // Simulate work
            return nil
        }),
        consumer.WithWorkersCount(5),
        consumer.WithTimeout(30*time.Second),
    )

    // Start consumer
    if err := c.Start(); err != nil {
        log.Fatal(err)
    }

    // Setup web UI
    uiHandler, err := ui.NewHandler(ui.HandlerOpts{
        Pool:      db,
        Prefix:    "/asynqpg",
        BasicAuth: &ui.BasicAuth{Username: "admin", Password: "admin"},
    })
    if err != nil {
        log.Fatal(err)
    }

    // Enqueue some test tasks
    go func() {
        time.Sleep(time.Second)
        ctx := context.Background()
        for i := 0; i < 10; i++ {
            payload, _ := json.Marshal(EmailPayload{
                To:      fmt.Sprintf("user%d@example.com", i),
                Subject: fmt.Sprintf("Test email %d", i),
                Body:    "Hello!",
            })
            _, _ = p.Enqueue(ctx, asynqpg.NewTask("email:send", payload))
        }
        slog.Info("enqueued 10 test tasks")
    }()

    // HTTP server
    http.Handle("/asynqpg/", uiHandler)
    go func() {
        slog.Info("Dashboard at http://localhost:8080/asynqpg/ (admin:admin)")
        _ = http.ListenAndServe(":8080", nil)
    }()

    // Wait for shutdown signal
    sigCh := make(chan os.Signal, 1)
    signal.Notify(sigCh, os.Interrupt, syscall.SIGTERM)
    <-sigCh

    slog.Info("shutting down...")
    if err := c.Stop(); err != nil {
        slog.Error("shutdown error", "error", err)
    }
    slog.Info("shutdown complete")
}
```

--------------------------------

### Run Full Demo

Source: https://github.com/yakser/asynqpg/blob/master/README.md

Execute this command to start all services including producers, consumers, web UI, and the observability stack. Use 'make demo-down' to stop all services.

```bash
make demo
```

```bash
make demo-down
```

--------------------------------

### Producer Example

Source: https://github.com/yakser/asynqpg/blob/master/README.md

Example of how to enqueue a task using the asynqpg producer. Ensure your PostgreSQL connection string is correct and the database is set up.

```go
package main

import (
    "context"
    "encoding/json"
    "log"
    "time"

    "github.com/jmoiron/sqlx"
    _ "github.com/lib/pq"
    "github.com/yakser/asynqpg"
    "github.com/yakser/asynqpg/producer"
)

func main() {
    db, err := sqlx.Connect("postgres", "postgres://postgres:password@localhost:5432/asynqpg?sslmode=disable")
    if err != nil {
        log.Fatal(err)
    }

    p, err := producer.New(producer.Config{Pool: db})
    if err != nil {
        log.Fatal(err)
    }

    payload, _ := json.Marshal(map[string]string{"to": "user@example.com", "subject": "Hello"})

    _, err = p.Enqueue(context.Background(), asynqpg.NewTask("email:send", payload,
        asynqpg.WithMaxRetry(5),
        asynqpg.WithDelay(10*time.Second),
    ))
    if err != nil {
        log.Fatal(err)
    }
}
```

--------------------------------

### OpenTelemetry Observability Setup

Source: https://context7.com/yakser/asynqpg/llms.txt

Integrate OpenTelemetry for metrics and tracing with asynqpg producer and consumer. Configure OTLP exporters and set global tracer and meter providers.

```go
package main

import (
    "context"
    "log"

    "github.com/jmoiron/sqlx"
    _ "github.com/lib/pq"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc"
    "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
    sdkmetric "go.opentelemetry.io/otel/sdk/metric"
    sdktrace "go.opentelemetry.io/otel/sdk/trace"

    "github.com/yakser/asynqpg/consumer"
    "github.com/yakser/asynqpg/producer"
)

func main() {
    ctx := context.Background()

    // Setup OTLP exporters (to Jaeger, Prometheus, etc.)
    traceExporter, _ := otlptracegrpc.New(ctx)
    metricExporter, _ := otlpmetricgrpc.New(ctx)

    tp := sdktrace.NewTracerProvider(sdktrace.WithBatcher(traceExporter))
    mp := sdkmetric.NewMeterProvider(sdkmetric.WithReader(
        sdkmetric.NewPeriodicReader(metricExporter),
    ))

    otel.SetTracerProvider(tp)
    otel.SetMeterProvider(mp)

    db, _ := sqlx.Connect("postgres", "postgres://postgres:password@localhost:5432/asynqpg?sslmode=disable")

    // Producer with observability
    p, _ := producer.New(producer.Config{
        Pool:           db,
        MeterProvider:  mp, // Optional: uses global if nil
        TracerProvider: tp, // Optional: uses global if nil
    })

    // Consumer with observability
    c, _ := consumer.New(consumer.Config{
        Pool:           db,
        MeterProvider:  mp,
        TracerProvider: tp,
    })

    _, _ = p, c
    log.Println("Observability configured")

    // Available metrics:
    // - asynqpg.tasks.enqueued (counter)      - Tasks enqueued
    // - asynqpg.tasks.processed (counter)     - Tasks finished processing
    // - asynqpg.tasks.errors (counter)        - Processing or enqueue errors
    // - asynqpg.task.duration (histogram)     - Handler execution duration (seconds)
    // - asynqpg.task.enqueue_duration (hist)  - Enqueue latency (seconds)
    // - asynqpg.tasks.in_flight (updown)      - Currently executing tasks
    //
    // All metrics tagged with: task_type, status
}
```

--------------------------------

### Start PostgreSQL and Observability Stack

Source: https://github.com/yakser/asynqpg/blob/master/README.md

Use this command to start the PostgreSQL database along with the full observability stack (Jaeger, Prometheus, Grafana, OTel Collector).

```bash
make demo-up
```

--------------------------------

### Consumer Example

Source: https://github.com/yakser/asynqpg/blob/master/README.md

Example of setting up and running a consumer to process tasks. This registers a handler for the 'email:send' task type and configures worker concurrency and timeouts.

```go
package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/jmoiron/sqlx"
    _ "github.com/lib/pq"
    "github.com/yakser/asynqpg"
    "github.com/yakser/asynqpg/consumer"
)

func main() {
    db, err := sqlx.Connect("postgres", "postgres://postgres:password@localhost:5432/asynqpg?sslmode=disable")
    if err != nil {
        log.Fatal(err)
    }

    c, err := consumer.New(consumer.Config{Pool: db})
    if err != nil {
        log.Fatal(err)
    }

    if err := c.RegisterTaskHandler("email:send",
        consumer.TaskHandlerFunc(func(ctx context.Context, task *asynqpg.TaskInfo) error {
            fmt.Printf("Processing task %d: %s\n", task.ID, task.Type)
            // process task...
            return nil
        }),
        consumer.WithWorkersCount(5),
        consumer.WithTimeout(30*time.Second),
    ); err != nil {
        log.Fatal(err)
    }

    if err := c.Start(); err != nil {
        log.Fatal(err)
    }

    sigCh := make(chan os.Signal, 1)
    signal.Notify(sigCh, os.Interrupt, syscall.SIGTERM)
    <-sigCh

    if err := c.Stop(); err != nil {
        log.Printf("shutdown error: %v", err)
    }
}
```

--------------------------------

### Manage Consumer Lifecycle

Source: https://github.com/yakser/asynqpg/blob/master/README.md

Controls the start and shutdown processes for the consumer.

```go
c.Start()               // start processing
c.Stop()                 // graceful shutdown (uses configured ShutdownTimeout)
c.Shutdown(timeout)      // graceful shutdown with custom timeout
```

--------------------------------

### Get Task Metadata from Context

Source: https://github.com/yakser/asynqpg/blob/master/README.md

Task metadata such as ID, retry count, and max retries can be accessed from the context using helper functions like `asynqpg.GetTaskID` and `asynqpg.GetTaskMetadata`.

```go
func (h *MyHandler) Handle(ctx context.Context, task *asynqpg.TaskInfo) error {
    id, _        := asynqpg.GetTaskID(ctx)         // database ID
    retry, _     := asynqpg.GetRetryCount(ctx)     // attempts already elapsed
    max, _       := asynqpg.GetMaxRetry(ctx)       // total max retry count
    createdAt, _ := asynqpg.GetCreatedAt(ctx)       // task creation time

    // Or get all metadata at once:
    meta, ok := asynqpg.GetTaskMetadata(ctx)
    // meta.ID, meta.RetryCount, meta.MaxRetry, meta.CreatedAt
    // ...
}
```

--------------------------------

### Configure and Run Asynqpg Consumer

Source: https://context7.com/yakser/asynqpg/llms.txt

Sets up a new consumer with a database connection and registers a task handler for 'email:send' tasks. Includes graceful shutdown handling.

```go
package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "log/slog"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/jmoiron/sqlx"
    _ "github.com/lib/pq"
    "github.com/yakser/asynqpg"
    "github.com/yakser/asynqpg/consumer"
)

type EmailPayload struct {
    To      string `json:"to"`
    Subject string `json:"subject"`
}

func main() {
    db, err := sqlx.Connect("postgres", "postgres://postgres:password@localhost:5432/asynqpg?sslmode=disable")
    if err != nil {
        log.Fatal(err)
    }

    // Create consumer with full configuration
    c, err := consumer.New(consumer.Config{
        Pool:              db,                    // Required: database pool
        ClientID:          "worker-1",            // Optional: for leader election
        FetchInterval:     100 * time.Millisecond,
        ShutdownTimeout:   30 * time.Second,
        DefaultWorkersCount: 10,
        DefaultMaxAttempts:  3,
        DefaultTimeout:      30 * time.Second,

        // Retention for completed/failed/cancelled tasks (auto-cleanup)
        CompletedRetention: 24 * time.Hour,
        FailedRetention:    7 * 24 * time.Hour,
        CancelledRetention: 24 * time.Hour,

        // RetryPolicy: &asynqpg.ConstantRetryPolicy{Delay: 5 * time.Second},
        // DisableMaintenance: true,     // Disable rescuer + cleaner
        // DisableBatchCompleter: true,  // Disable batch completions
    })
    if err != nil {
        log.Fatal(err)
    }

    // Register task handler with per-type options
    err = c.RegisterTaskHandler("email:send",
        consumer.TaskHandlerFunc(func(ctx context.Context, task *asynqpg.TaskInfo) error {
            var payload EmailPayload
            if err := json.Unmarshal(task.Payload, &payload); err != nil {
                // Invalid payload - skip all retries
                return fmt.Errorf("bad payload: %w", asynqpg.ErrSkipRetry)
            }

            fmt.Printf("Sending email to %s (attempt %d/%d)\n",
                payload.To, task.AttemptsElapsed+1, task.AttemptsLeft+task.AttemptsElapsed)

            // Simulate work
            // return nil for success, error for retry
            return nil
        }),
        consumer.WithWorkersCount(5),           // 5 goroutines for this task type
        consumer.WithTimeout(30*time.Second),   // Per-task execution timeout
        consumer.WithMaxAttempts(5),            // Override default max attempts
    )
    if err != nil {
        log.Fatal(err)
    }

    // Start consumer
    if err := c.Start(); err != nil {
        log.Fatal(err)
    }

    slog.Info("Consumer started, press Ctrl+C to stop")

    // Graceful shutdown on signal
    sigCh := make(chan os.Signal, 1)
    signal.Notify(sigCh, os.Interrupt, syscall.SIGTERM)
    <-sigCh

    if err := c.Stop(); err != nil {
        log.Printf("Shutdown error: %v", err)
    }
}
```

--------------------------------

### Initialize Asynqpg Producer

Source: https://context7.com/yakser/asynqpg/llms.txt

Set up the producer with a database connection pool and optional configurations like logger and retry settings. Ensure the database connection is established before initializing the producer.

```go
package main

import (
    "context"
    "encoding/json"
    "log"
    "log/slog"
    "time"

    "github.com/jmoiron/sqlx"
    _ "github.com/lib/pq"
    "github.com/yakser/asynqpg"
    "github.com/yakser/asynqpg/producer"
)

func main() {
    db, err := sqlx.Connect("postgres", "postgres://postgres:password@localhost:5432/asynqpg?sslmode=disable")
    if err != nil {
        log.Fatal(err)
    }

    // Create producer with configuration
    p, err := producer.New(producer.Config{
        Pool:            db,              // Required: database connection pool
        Logger:          slog.Default(),  // Optional: structured logger
        DefaultMaxRetry: 3,               // Optional: default retry count (default: 3)
        // MeterProvider: mp,             // Optional: OpenTelemetry metrics
        // TracerProvider: tp,            // Optional: OpenTelemetry tracing
    })
    if err != nil {
        log.Fatal(err)
    }

    ctx := context.Background()

    // Single task enqueue
    payload, _ := json.Marshal(map[string]string{"to": "user@example.com", "subject": "Hello"})
    result, err := p.Enqueue(ctx, asynqpg.NewTask("email:send", payload,
        asynqpg.WithMaxRetry(5),
        asynqpg.WithDelay(10*time.Second),
    ))
    if err != nil {
        log.Fatal(err)
    }
    log.Printf("Enqueued task ID: %d, Duplicate: %v", result.ID, result.Duplicate)

    // Batch enqueue - efficient single SQL call with UNNEST
    tasks := []*asynqpg.Task{
        asynqpg.NewTask("email:send", []byte(`{"to":"a@example.com"}`),
            asynqpg.WithIdempotencyToken("email-a")),
        asynqpg.NewTask("email:send", []byte(`{"to":"b@example.com"}`),
            asynqpg.WithIdempotencyToken("email-b")),
        asynqpg.NewTask("email:send", []byte(`{"to":"c@example.com"}`),
            asynqpg.WithIdempotencyToken("email-c")),
    }
    batchResult, err := p.EnqueueMany(ctx, tasks)
    if err != nil {
        log.Fatal(err)
    }
    log.Printf("Batch enqueued: %d inserted, %d duplicates",
        batchResult.InsertedCount(), len(tasks)-batchResult.InsertedCount())

    // Transactional enqueue - atomic with your business logic
    tx, _ := db.BeginTxx(ctx, nil)
    defer tx.Rollback()

    // Your business logic here...
    _, _ = tx.ExecContext(ctx, "INSERT INTO orders (id) VALUES ($1)", 123)

    // Enqueue task in same transaction
    _, err = p.EnqueueTx(ctx, tx, asynqpg.NewTask("order:process", []byte(`{"order_id":123}`)))
    if err != nil {
        log.Fatal(err)
    }

    tx.Commit() // Both order insert and task enqueue commit together
}
```

--------------------------------

### Create Asynqpg Task with Options

Source: https://context7.com/yakser/asynqpg/llms.txt

Demonstrates creating a new Asynqpg task with a type and payload. Shows how to configure max retries, processing delay, and idempotency tokens.

```go
package main

import (
    "time"
    "github.com/yakser/asynqpg"
)

func main() {
    // Basic task with type and payload
    task := asynqpg.NewTask("email:send", []byte(`{"to":"user@example.com"}`))

    // Task with options: max retries, delay before processing, idempotency token
    taskWithOpts := asynqpg.NewTask("email:send", []byte(`{"to":"user@example.com"}`),
        asynqpg.WithMaxRetry(5),                           // Max 5 retry attempts
        asynqpg.WithDelay(10*time.Second),                 // Delay 10 seconds before first run
        asynqpg.WithIdempotencyToken("unique-order-123"),  // Deduplicate by token
    )

    // Schedule task for a specific time using ProcessAt
    scheduledTask := asynqpg.NewTask("report:generate", []byte(`{"id":1}`))
    scheduledTask.ProcessAt = time.Date(2026, 1, 1, 9, 0, 0, 0, time.UTC)

    _, _, _ = task, taskWithOpts, scheduledTask
}
```

--------------------------------

### Create Consumer Configuration

Source: https://github.com/yakser/asynqpg/blob/master/README.md

Initialize a consumer with configuration options. Key settings include the database pool, client ID for leader election, retry policy, fetch interval, and retention periods for task states.

```go
c, err := consumer.New(consumer.Config{
    Pool:              db,
    ClientID:          "worker-1",
    RetryPolicy:       retryPolicy,
    FetchInterval:     100*time.Millisecond,
    ShutdownTimeout:   30*time.Second,
    CompletedRetention: 24*time.Hour,
    FailedRetention:    7*24*time.Hour,
    CancelledRetention: 24*time.Hour,
})
```

--------------------------------

### Implement Consumer Middleware in Go

Source: https://context7.com/yakser/asynqpg/llms.txt

Demonstrates global middleware for logging and panic recovery, as well as per-task middleware registration.

```go
package main

import (
    "context"
    "log/slog"
    "time"

    "github.com/jmoiron/sqlx"
    _ "github.com/lib/pq"
    "github.com/yakser/asynqpg"
    "github.com/yakser/asynqpg/consumer"
)

func main() {
    db, _ := sqlx.Connect("postgres", "postgres://postgres:password@localhost:5432/asynqpg?sslmode=disable")
    c, _ := consumer.New(consumer.Config{Pool: db})

    // Global middleware - applies to ALL task types
    _ = c.Use(func(next consumer.TaskHandler) consumer.TaskHandler {
        return consumer.TaskHandlerFunc(func(ctx context.Context, task *asynqpg.TaskInfo) error {
            start := time.Now()
            slog.Info("processing task", "type", task.Type, "id", task.ID)

            err := next.Handle(ctx, task)

            slog.Info("task completed",
                "type", task.Type,
                "id", task.ID,
                "duration", time.Since(start),
                "error", err,
            )
            return err
        })
    })

    // Recovery middleware - catch panics
    _ = c.Use(func(next consumer.TaskHandler) consumer.TaskHandler {
        return consumer.TaskHandlerFunc(func(ctx context.Context, task *asynqpg.TaskInfo) (err error) {
            defer func() {
                if r := recover(); r != nil {
                    slog.Error("task panic", "task_id", task.ID, "panic", r)
                    err = asynqpg.ErrSkipRetry // Don't retry panics
                }
            }()
            return next.Handle(ctx, task)
        })
    })

    // Per-task-type middleware
    rateLimitMiddleware := func(next consumer.TaskHandler) consumer.TaskHandler {
        return consumer.TaskHandlerFunc(func(ctx context.Context, task *asynqpg.TaskInfo) error {
            // Rate limiting logic here
            return next.Handle(ctx, task)
        })
    }

    _ = c.RegisterTaskHandler("email:send",
        consumer.TaskHandlerFunc(func(ctx context.Context, task *asynqpg.TaskInfo) error {
            return nil
        }),
        consumer.WithMiddleware(rateLimitMiddleware), // Per-task middleware
        consumer.WithWorkersCount(5),
    )
}
```

--------------------------------

### Create Producer Configuration

Source: https://github.com/yakser/asynqpg/blob/master/README.md

Initialize a producer with configuration options. The database connection is required, while logger, max retry, and OpenTelemetry providers are optional.

```go
p, err := producer.New(producer.Config{
    Pool:            db,
    Logger:          slog.Default(),
    DefaultMaxRetry: 3,
    MeterProvider:   mp,
    TracerProvider:  tp,
})
```

--------------------------------

### Web UI with Basic Authentication

Source: https://context7.com/yakser/asynqpg/llms.txt

Configure the asynqpg dashboard with HTTP Basic Authentication. Ensure the database connection and handler options are correctly set.

```go
package main

import (
    "log"
    "net/http"

    "github.com/jmoiron/sqlx"
    _ "github.com/lib/pq"
    "github.com/yakser/asynqpg/ui"
)

func main() {
    db, err := sqlx.Connect("postgres", "postgres://postgres:password@localhost:5432/asynqpg?sslmode=disable")
    if err != nil {
        log.Fatal(err)
    }

    handler, err := ui.NewHandler(ui.HandlerOpts{
        Pool:   db,
        Prefix: "/asynqpg",
        BasicAuth: &ui.BasicAuth{
            Username: "admin",
            Password: "secret",
        },
    })
    if err != nil {
        log.Fatal(err)
    }

    http.Handle("/asynqpg/", handler)
    log.Println("Dashboard at http://localhost:8080/asynqpg/ (admin:secret)")
    log.Fatal(http.ListenAndServe(":8080", nil))
}
```

--------------------------------

### Manage Tasks with Asynqpg Client

Source: https://context7.com/yakser/asynqpg/llms.txt

Demonstrates task inspection, filtering, and lifecycle management using the client API, including transactional support.

```go
package main

import (
    "context"
    "fmt"
    "log"

    "github.com/jmoiron/sqlx"
    _ "github.com/lib/pq"
    "github.com/yakser/asynqpg"
    "github.com/yakser/asynqpg/client"
)

func main() {
    db, err := sqlx.Connect("postgres", "postgres://postgres:password@localhost:5432/asynqpg?sslmode=disable")
    if err != nil {
        log.Fatal(err)
    }

    cl, err := client.New(client.Config{
        Pool: db,
        // Logger: slog.Default(),
        // TracerProvider: tp,
    })
    if err != nil {
        log.Fatal(err)
    }

    ctx := context.Background()

    // Get a single task by ID
    taskInfo, err := cl.GetTask(ctx, 123)
    if err != nil {
        if err == client.ErrTaskNotFound {
            fmt.Println("Task not found")
        } else {
            log.Fatal(err)
        }
    } else {
        fmt.Printf("Task %d: type=%s, status=%s, attempts=%d/%d\n",
            taskInfo.ID, taskInfo.Type, taskInfo.Status,
            taskInfo.AttemptsElapsed, taskInfo.AttemptsLeft+taskInfo.AttemptsElapsed)
    }

    // List tasks with filtering and pagination
    result, err := cl.ListTasks(ctx, client.NewListParams().
        States(asynqpg.TaskStatusFailed, asynqpg.TaskStatusPending).
        Types("email:send", "order:process").
        Limit(50).
        Offset(0).
        OrderBy(client.OrderByCreatedAt, client.SortDesc),
    )
    if err != nil {
        log.Fatal(err)
    }
    fmt.Printf("Found %d tasks (total: %d)\n", len(result.Tasks), result.Total)

    // Cancel a task (pending/failed -> cancelled)
    // Running tasks: handler context is cancelled
    cancelledTask, err := cl.CancelTask(ctx, 456)
    if err != nil {
        switch err {
        case client.ErrTaskNotFound:
            fmt.Println("Task not found")
        case client.ErrTaskAlreadyFinalized:
            fmt.Println("Task already completed, cannot cancel")
        default:
            log.Fatal(err)
        }
    } else {
        fmt.Printf("Cancelled task %d\n", cancelledTask.ID)
    }

    // Retry a failed/cancelled task (moves to pending)
    retriedTask, err := cl.RetryTask(ctx, 789)
    if err != nil {
        switch err {
        case client.ErrTaskNotFound:
            fmt.Println("Task not found")
        case client.ErrTaskAlreadyAvailable:
            fmt.Println("Task already pending")
        case client.ErrTaskRunning:
            fmt.Println("Task currently running")
        case client.ErrTaskAlreadyFinalized:
            fmt.Println("Task completed, cannot retry")
        default:
            log.Fatal(err)
        }
    } else {
        fmt.Printf("Retried task %d, now pending\n", retriedTask.ID)
    }

    // Delete a task (any non-running status)
    deletedTask, err := cl.DeleteTask(ctx, 101)
    if err != nil {
        switch err {
        case client.ErrTaskNotFound:
            fmt.Println("Task not found")
        case client.ErrTaskRunning:
            fmt.Println("Cannot delete running task")
        default:
            log.Fatal(err)
        }
    } else {
        fmt.Printf("Deleted task %d (was %s)\n", deletedTask.ID, deletedTask.Status)
    }

    // Transactional operations
    tx, _ := db.BeginTxx(ctx, nil)
    _, _ = cl.GetTaskTx(ctx, tx, 123)
    _, _ = cl.CancelTaskTx(ctx, tx, 456)
    _, _ = cl.RetryTaskTx(ctx, tx, 789)
    _, _ = cl.DeleteTaskTx(ctx, tx, 101)
    tx.Commit()
}
```

--------------------------------

### Run Benchmarks

Source: https://github.com/yakser/asynqpg/blob/master/README.md

This command runs Go benchmarks for SQL-level operations, batch completion, producer throughput, and consumer processing. It requires Docker and uses testcontainers.

```bash
make bench
```

--------------------------------

### Run Tests and Benchmarks

Source: https://github.com/yakser/asynqpg/blob/master/README.md

Commands for running unit tests, integration tests, benchmarks, and linting. Integration tests and benchmarks require Docker.

```bash
make test
```

```bash
make test-integration
```

```bash
make test-all
```

```bash
make bench
```

```bash
make lint
```

--------------------------------

### Serve Asynqpg Web UI Dashboard

Source: https://context7.com/yakser/asynqpg/llms.txt

Mounts an HTTP handler to serve the REST API and React SPA for task management.

```go
package main

import (
    "log"
    "net/http"
    "time"

    "github.com/jmoiron/sqlx"
    _ "github.com/lib/pq"
    "github.com/yakser/asynqpg/ui"
)

func main() {
    db, err := sqlx.Connect("postgres", "postgres://postgres:password@localhost:5432/asynqpg?sslmode=disable")
    if err != nil {
        log.Fatal(err)
    }

    // Basic handler without authentication
    handler, err := ui.NewHandler(ui.HandlerOpts{
        Pool:   db,
        Prefix: "/asynqpg",
        // Logger: slog.Default(),
        // HidePayloadByDefault: true,      // Hide payloads in list/detail views
        // AllowedOrigins: []string{"*"},   // CORS allowed origins
    })
    if err != nil {
        log.Fatal(err)
    }

    http.Handle("/asynqpg/", handler)
    log.Println("Dashboard at http://localhost:8080/asynqpg/")
    log.Fatal(http.ListenAndServe(":8080", nil))
}
```

--------------------------------

### Web UI with OAuth Authentication

Source: https://context7.com/yakser/asynqpg/llms.txt

Set up the asynqpg dashboard with session-based authentication using OAuth providers. Implement the ui.AuthProvider interface for your specific OAuth provider.

```go
package main

import (
    "log"
    "net/http"
    "time"

    "github.com/jmoiron/sqlx"
    _ "github.com/lib/pq"
    "github.com/yakser/asynqpg/ui"
)

// Implement ui.AuthProvider interface for your OAuth provider
// See examples/demo/github_provider.go for a complete GitHub OAuth implementation

func main() {
    db, err := sqlx.Connect("postgres", "postgres://postgres:password@localhost:5432/asynqpg?sslmode=disable")
    if err != nil {
        log.Fatal(err)
    }

    // Example OAuth configuration (implement AuthProvider interface)
    // githubProvider := NewGitHubProvider(clientID, clientSecret, callbackURL)

    handler, err := ui.NewHandler(ui.HandlerOpts{
        Pool:   db,
        Prefix: "/asynqpg",
        // AuthProviders: []ui.AuthProvider{githubProvider},
        SecureCookies: true,              // Set true for HTTPS
        SessionMaxAge: 24 * time.Hour,
        // SessionStore: ui.NewMemorySessionStore(), // Default in-memory store
    })
    if err != nil {
        log.Fatal(err)
    }

    http.Handle("/asynqpg/", handler)
    log.Fatal(http.ListenAndServe(":8080", nil))
}
```

--------------------------------

### Configure Web UI Authentication

Source: https://github.com/yakser/asynqpg/blob/master/README.md

Sets up basic or OAuth authentication for the dashboard.

```go
ui.HandlerOpts{
    BasicAuth: &ui.BasicAuth{Username: "admin", Password: "secret"},
}
```

```go
ui.HandlerOpts{
    AuthProviders: []ui.AuthProvider{githubProvider},
    SecureCookies: true,    // for HTTPS
    SessionMaxAge: 24*time.Hour,
}
```

--------------------------------

### Create Task with Options

Source: https://github.com/yakser/asynqpg/blob/master/README.md

Create a new task with specified type, payload, and various options like max retries, delay, and idempotency token.

```go
asynqpg.NewTask("type", payload,
    asynqpg.WithMaxRetry(5),
    asynqpg.WithDelay(10*time.Second),
    asynqpg.WithIdempotencyToken("unique-token"),
)
```

--------------------------------

### Initialize Theme Based on User Preference

Source: https://github.com/yakser/asynqpg/blob/master/ui/frontend/index.html

This script initializes the application's theme by checking local storage for a saved theme preference. If no preference is found, it defaults to the user's system preference for dark mode. Ensure this script runs early in the page load to apply the theme correctly.

```javascript
asynqpg  (function() { var t = localStorage.getItem("theme"); if (t === "dark" || (t !== "light" && matchMedia("(prefers-color-scheme: dark)").matches)) { document.documentElement.classList.add("dark"); } })();
```

--------------------------------

### Initialize Theme Script

Source: https://github.com/yakser/asynqpg/blob/master/ui/frontend/dist/index.html

Applies the 'dark' class to the document element based on local storage settings or system color scheme preferences.

```javascript
(function() { var t = localStorage.getItem("theme"); if (t === "dark" || (t !== "light" && matchMedia("(prefers-color-scheme: dark)").matches)) { document.documentElement.classList.add("dark"); } })();
```

--------------------------------

### Manage Tasks via Client

Source: https://github.com/yakser/asynqpg/blob/master/README.md

Provides methods to inspect, list, and modify task states in the database.

```go
cl, err := client.New(client.Config{Pool: db})

// Get a single task
info, err := cl.GetTask(ctx, taskID)

// List tasks with filtering
result, err := cl.ListTasks(ctx, client.NewListParams().
    States(asynqpg.TaskStatusFailed, asynqpg.TaskStatusPending).
    Types("email:send").
    Limit(50).
    OrderBy(client.OrderByCreatedAt, client.SortDesc),
)
// result.Tasks, result.Total

// Manage tasks
_, err = cl.CancelTask(ctx, id)   // pending/failed → cancelled
_, err = cl.RetryTask(ctx, id)    // failed/cancelled → pending
_, err = cl.DeleteTask(ctx, id)   // remove from database
```

--------------------------------

### Create Test Context with Task Metadata

Source: https://github.com/yakser/asynqpg/blob/master/README.md

Use `asynqpg.WithTaskMetadata` to create a context with specific task metadata for testing handlers without a running asynqpg server.

```go
ctx := asynqpg.WithTaskMetadata(context.Background(), asynqpg.TaskMetadata{
    ID: 42, RetryCount: 0, MaxRetry: 3, CreatedAt: time.Now(),
})
err := handler.Handle(ctx, task)
```

--------------------------------

### Clone Repository and Set Upstream

Source: https://github.com/yakser/asynqpg/blob/master/CONTRIBUTING.md

Clone the asynqpg repository and add the upstream remote for keeping your fork in sync.

```bash
git clone https://github.com/<your-user>/asynqpg.git
cd asynqpg
git remote add upstream https://github.com/yakser/asynqpg.git
```

--------------------------------

### Mount Web UI Handler

Source: https://github.com/yakser/asynqpg/blob/master/README.md

Exposes the asynqpg dashboard as an HTTP handler.

```go
handler, err := ui.NewHandler(ui.HandlerOpts{
    Pool:   db,
    Prefix: "/asynqpg",
})
http.Handle("/asynqpg/", handler)
```

--------------------------------

### Configure Global Middleware

Source: https://github.com/yakser/asynqpg/blob/master/README.md

Applies middleware to all task types using the TaskHandler pattern.

```go
c, _ := consumer.New(config)

_ = c.Use(func(next consumer.TaskHandler) consumer.TaskHandler {
    return consumer.TaskHandlerFunc(func(ctx context.Context, task *asynqpg.TaskInfo) error {
        slog.Info("processing task", "type", task.Type, "id", task.ID)
        err := next.Handle(ctx, task)
        slog.Info("task done", "type", task.Type, "id", task.ID, "error", err)
        return err
    })
})
```

--------------------------------

### Schedule Task for Specific Time

Source: https://github.com/yakser/asynqpg/blob/master/README.md

Create a task and schedule it to be processed at a specific time in the future. The `ProcessAt` field determines the earliest processing time.

```go
task := asynqpg.NewTask("report:generate", payload)
task.ProcessAt = time.Date(2026, 1, 1, 9, 0, 0, 0, time.UTC)
```

--------------------------------

### Core Module Release

Source: https://github.com/yakser/asynqpg/blob/master/CONTRIBUTING.md

Commands to create a release for the core module, specifying the version and pushing tags.

```bash
make release-core V=v0.5.0
git push origin master --tags
```

--------------------------------

### Manage Task Retries and Error Handling in Go

Source: https://context7.com/yakser/asynqpg/llms.txt

Shows how to use ErrSkipRetry, TaskSnooze, and TaskSnoozeWithError to control task lifecycle and inspect error types.

```go
package main

import (
    "context"
    "encoding/json"
    "errors"
    "fmt"
    "time"

    "github.com/yakser/asynqpg"
)

type OrderPayload struct {
    OrderID int `json:"order_id"`
}

// Handler demonstrating all error handling patterns
func HandleOrder(ctx context.Context, task *asynqpg.TaskInfo) error {
    var payload OrderPayload
    if err := json.Unmarshal(task.Payload, &payload); err != nil {
        // ErrSkipRetry - permanently fail, no more retries
        // Use for: invalid payload, business logic rejection, non-recoverable errors
        return fmt.Errorf("invalid payload: %w", asynqpg.ErrSkipRetry)
    }

    // Check if external service is ready
    if !isExternalServiceReady() {
        // TaskSnooze - reschedule WITHOUT counting as an attempt
        // attempts_left and attempts_elapsed remain unchanged
        // Use for: external dependency not ready, rate limiting
        return asynqpg.TaskSnooze(30 * time.Second)
    }

    // Call external API
    if err := callExternalAPI(payload.OrderID); err != nil {
        if isRateLimited(err) {
            // TaskSnoozeWithError - reschedule AND count as failed attempt
            // attempts_left decremented, error message stored
            // Use for: transient errors with custom delay
            return fmt.Errorf("rate limited: %w", asynqpg.TaskSnoozeWithError(1*time.Minute))
        }
        // Regular error - uses RetryPolicy for delay, counts as attempt
        return fmt.Errorf("api call failed: %w", err)
    }

    return nil // Success - task marked completed
}

// Detecting error types
func analyzeError(err error) {
    // Check for SkipRetry
    if errors.Is(err, asynqpg.ErrSkipRetry) {
        fmt.Println("Task should not be retried")
    }

    // Check for Snooze (no attempt counted)
    var snoozeErr *asynqpg.TaskSnoozeError
    if errors.As(err, &snoozeErr) {
        fmt.Printf("Task snoozed for %s (no attempt counted)\n", snoozeErr.Duration)
    }

    // Check for SnoozeWithError (attempt counted)
    var snoozeWithErr *asynqpg.TaskSnoozeWithErrError
    if errors.As(err, &snoozeWithErr) {
        fmt.Printf("Task snoozed with error for %s (attempt counted)\n", snoozeWithErr.Duration)
    }
}

func isExternalServiceReady() bool { return true }
func callExternalAPI(id int) error { return nil }
func isRateLimited(err error) bool { return false }

func main() {}
```

--------------------------------

### Register Per-Task-Type Middleware

Source: https://github.com/yakser/asynqpg/blob/master/README.md

Applies specific middleware and worker configurations to a single task type.

```go
c.RegisterTaskHandler("email:send", emailHandler,
    consumer.WithMiddleware(rateLimitMiddleware),
    consumer.WithWorkersCount(5),
)
```

--------------------------------

### Configure Retry Policies in Go

Source: https://context7.com/yakser/asynqpg/llms.txt

Define exponential or constant backoff strategies for task retries and apply them to the consumer configuration.

```go
package main

import (
    "fmt"
    "time"

    "github.com/jmoiron/sqlx"
    _ "github.com/lib/pq"
    "github.com/yakser/asynqpg"
    "github.com/yakser/asynqpg/consumer"
)

func main() {
    db, _ := sqlx.Connect("postgres", "postgres://postgres:password@localhost:5432/asynqpg?sslmode=disable")

    // Default exponential backoff: attempt^4 seconds with ±10% jitter
    // Attempt 1: ~1s, Attempt 2: ~16s, Attempt 3: ~81s, Attempt 4: ~256s, etc.
    exponentialPolicy := &asynqpg.DefaultRetryPolicy{
        MaxRetryDelay: 24 * time.Hour, // Cap at 24 hours (default)
    }

    // Constant delay - same wait between all retries
    constantPolicy := &asynqpg.ConstantRetryPolicy{
        Delay: 5 * time.Second,
    }

    // Use in consumer
    c, _ := consumer.New(consumer.Config{
        Pool:        db,
        RetryPolicy: exponentialPolicy, // or constantPolicy
    })

    _ = c

    // Policy usage examples
    fmt.Printf("Exponential attempt 1: %v\n", exponentialPolicy.NextRetry(1)) // ~1s
    fmt.Printf("Exponential attempt 2: %v\n", exponentialPolicy.NextRetry(2)) // ~16s
    fmt.Printf("Exponential attempt 3: %v\n", exponentialPolicy.NextRetry(3)) // ~81s
    fmt.Printf("Constant any attempt: %v\n", constantPolicy.NextRetry(5))     // 5s
}
```

--------------------------------

### UI Module Release

Source: https://github.com/yakser/asynqpg/blob/master/CONTRIBUTING.md

Commands to create a release for the UI module, specifying the version and pushing tags. It can optionally pin to a specific core version.

```bash
make release-ui V=v0.1.0
git push origin master --tags
```

--------------------------------

### UI Module Release with Specific Core Version

Source: https://github.com/yakser/asynqpg/blob/master/CONTRIBUTING.md

Release the UI module pinned to a specific core version using a Python script.

```bash
python3 scripts/release.py ui v0.1.0 --core-version v0.4.0
git push origin master --tags
```

--------------------------------

### Access Task Metadata in Go

Source: https://context7.com/yakser/asynqpg/llms.txt

Retrieve task information from the context using individual accessors or the bulk metadata function, and use WithTaskMetadata for testing.

```go
package main

import (
    "context"
    "fmt"
    "time"

    "github.com/yakser/asynqpg"
)

// Handler using context utilities
func HandleWithContext(ctx context.Context, task *asynqpg.TaskInfo) error {
    // Individual accessors
    id, _ := asynqpg.GetTaskID(ctx)             // Database ID
    retry, _ := asynqpg.GetRetryCount(ctx)      // Attempts already elapsed
    maxRetry, _ := asynqpg.GetMaxRetry(ctx)     // Total max retry count
    createdAt, _ := asynqpg.GetCreatedAt(ctx)   // Task creation time

    fmt.Printf("Task %d: attempt %d/%d, created %v\n", id, retry+1, maxRetry, createdAt)

    // Or get all metadata at once
    meta, ok := asynqpg.GetTaskMetadata(ctx)
    if ok {
        fmt.Printf("Full metadata: ID=%d, Retry=%d, MaxRetry=%d, Created=%v\n",
            meta.ID, meta.RetryCount, meta.MaxRetry, meta.CreatedAt)
    }

    return nil
}

// Testing handlers with mock context
func TestHandler() {
    // Create context with task metadata for testing
    ctx := asynqpg.WithTaskMetadata(context.Background(), asynqpg.TaskMetadata{
        ID:         42,
        RetryCount: 0,
        MaxRetry:   3,
        CreatedAt:  time.Now(),
    })

    task := &asynqpg.TaskInfo{
        ID:      42,
        Type:    "test:task",
        Payload: []byte(`{}`),
    }

    err := HandleWithContext(ctx, task)
    fmt.Printf("Handler error: %v\n", err)
}

func main() {
    TestHandler()
}
```

--------------------------------

### Task Handler Interface

Source: https://github.com/yakser/asynqpg/blob/master/README.md

Defines the interface for task handlers. Implement this interface or use the `TaskHandlerFunc` adapter to process tasks.

```go
type TaskHandler interface {
    Handle(ctx context.Context, task *asynqpg.TaskInfo) error
}
```

--------------------------------

### Asynqpg PostgreSQL Database Schema

Source: https://context7.com/yakser/asynqpg/llms.txt

SQL statements to create the necessary PostgreSQL table and enum type for Asynqpg task management. Includes indexes for efficient querying.

```sql
CREATE TYPE asynqpg_task_status AS ENUM (
    'pending', 'running', 'completed', 'failed', 'cancelled'
);

CREATE TABLE IF NOT EXISTS asynqpg_tasks (
    id BIGSERIAL PRIMARY KEY,
    attempts_left SMALLINT NOT NULL,
    attempts_elapsed SMALLINT DEFAULT 0 NOT NULL,
    blocked_till TIMESTAMPTZ NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    finalized_at TIMESTAMPTZ,
    attempted_at TIMESTAMPTZ,
    type TEXT NOT NULL,
    idempotency_token TEXT,
    status asynqpg_task_status NOT NULL DEFAULT 'pending',
    payload BYTEA NOT NULL,
    messages TEXT[] DEFAULT ARRAY[]::TEXT[] NOT NULL
);

CREATE UNIQUE INDEX IF NOT EXISTS asynqpg_tasks_idempotency_idx
    ON asynqpg_tasks (type, idempotency_token)
    WHERE idempotency_token IS NOT NULL;

CREATE INDEX IF NOT EXISTS asynqpg_tasks_ready_idx
    ON asynqpg_tasks (type, blocked_till)
    WHERE status IN ('pending', 'running');
```
Tokens:
Raw