Overview

queue is a standalone Go module (github.com/gp-system/queue): the Valkey/asynq foundation everything else in the async category builds on: events, outbox, scheduling, and the kit's worker chassis. It combines connection config and task enqueueing in one place; the server side is asynq, which provides the worker processing and monitoring. The higher-level options, the event envelope, and task-type naming are covered on the next page, Tasks; this page stays at the level of connecting and enqueueing.

import "github.com/gp-system/queue"

Day to day you rarely call it directly: events reach the queue through a dispatcher, jobs through the scheduler. But the concepts here (config, client, the envelope on the next page) reappear in every higher layer, so this is the right place to start.

Install

go get github.com/gp-system/queue@v0.1.0 # the kit itself uses this tag
go get github.com/gp-system/queue@latest

Go 1.25+. queue depends on errs, hibiken/asynq and redis/go-redis/v9, plus OTel's tracing API for the producer span.

Configuration

queue.Config is the single Valkey connection description: the client, the worker, the scheduler and the outbox relay all receive it. Compose it under a prefix:

type Config struct {
    Valkey queue.Config `envPrefix:"VALKEY_"`
}

The config opens connections in two directions, and both libraries are configured from the same single source:

• cfg.ValkeyConnOpt() → asynq.RedisClientOpt (asynq client, server, scheduler),
• cfg.ValkeyOptions() → *redis.Options (direct go-redis calls, e.g. the scheduler's lease).

The worker's Config already embeds it under the VALKEY_ prefix. In a generated project there is nothing extra to wire.

Client

client, err := queue.NewClient(ctx, cfg.Valkey)   // or queue.MustNewClient in main()
defer client.Close()

info, err := client.Enqueue(ctx, task, queue.OnQueue("mail"))

NewClient opens the Valkey connection and verifies it with a PING at construction, the same convention as pg.NewPool: a wrong address fails at startup, not at the first enqueue. MustNewClient is the panicking variant for main().

Enqueue(ctx, *asynq.Task, opts...) submits the task to Valkey and returns an *asynq.TaskInfo. It adds two things on top of raw asynq:

• A producer span: every enqueue opens an OpenTelemetry span (enqueue <task-type>), so handing work to the queue shows up in the trace (see trace propagation on the next page).
• queue.ErrDuplicate: when a TaskID or Unique constraint means the task is already there, you get a single sentinel error instead of asynq's two distinct ones. Callers that rely on idempotent delivery (the outbox relay, the event fan-out) treat it as success:

if _, err := client.Enqueue(ctx, task, queue.TaskID(id)); err != nil {
    if errors.Is(err, queue.ErrDuplicate) {
        return nil // already enqueued, exactly what we wanted
    }
    return err
}

Close() closes the shared Valkey connection (the asynq client is built from it and doesn't own it). The worker does this for you as part of its shutdown order. The rest of Enqueue's options (OnQueue, MaxRetry, Unique, ...) are covered on the Tasks page, alongside the envelope and task-type naming that build on this client.

Standalone example

queue needs nothing from the kit: a client, a task, and a minimal asynq server are enough to see work flow end to end.

package main

import (
    "context"
    "log"

    "github.com/gp-system/queue"
    "github.com/hibiken/asynq"
)

func main() {
    ctx := context.Background()
    cfg := queue.Config{Addr: "localhost:6379"}

    client := queue.MustNewClient(ctx, cfg)
    defer client.Close()

    task := asynq.NewTask("email:welcome", []byte(`{"user_id":"u_1"}`))
    if _, err := client.Enqueue(ctx, task, queue.OnQueue("mail")); err != nil {
        log.Fatal(err)
    }

    srv := asynq.NewServer(cfg.ValkeyConnOpt(), asynq.Config{Concurrency: 5})
    mux := asynq.NewServeMux()
    mux.HandleFunc("email:welcome", func(ctx context.Context, t *asynq.Task) error {
        log.Printf("sending welcome email: %s", t.Payload())
        return nil
    })
    if err := srv.Run(mux); err != nil {
        log.Fatal(err)
    }
}

Where to next

• Tasks: the rest of Enqueue's options, the Envelope wire format, OTel trace propagation, and the task-type naming conventions the events module and the kit's worker build on.