Queue

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

queue is the kit's Redis/asynq foundation: everything that follows in this category (events, outbox, scheduling, worker) builds on this package. It combines connection config, task enqueueing, and per-task options (queue name, delay, retries) in one place; the server side is asynq, which provides the worker processing and monitoring.

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

Configuration

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

type Config struct {
    Redis queue.Config `envPrefix:"REDIS_"`
}

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

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

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

Client

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

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

NewClient opens the Redis 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 Redis 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 below).
• 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 Redis 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.

Enqueue options

The kit re-exports the useful subset of asynq's options so common cases don't require importing asynq:

The options look the same everywhere a task is born: at Enqueue, at the events.Listen listener registration, and on scheduler entries. The shop's mail queue, for example, appears in the listener registration:

events.Listen(reg, "sendOrderConfirmation", sendOrderConfirmation(deps), queue.OnQueue("mail"))

When you need the full asynq option set, queue.AsynqOptions(opts...) converts kit options into raw []asynq.Option, and asynq's own API remains open alongside Enqueue: an escape hatch, not a forbidden zone.

The Envelope

Every event task payload is the same wire format, queue.Envelope:

type Envelope struct {
    ID         string            `json:"id"`         // dispatch id, doubles as the asynq TaskID
    Name       string            `json:"name"`       // the event name (EventName())
    Payload    json.RawMessage   `json:"payload"`    // the JSON-encoded event value
    Metadata   map[string]string `json:"metadata,omitempty"` // W3C trace context
    OccurredAt time.Time         `json:"occurred_at"`
}

The envelope travels unchanged from the dispatcher (or the outbox) through the fan-out task to every listener task: the listener sees the same ID, payload and trace context the producer set. env.Task() renders it as the fan-out task; queue.DecodeEnvelope(task) unpacks it on the other side.

Trace propagation across the queue

env.InjectTrace(ctx) writes the context's trace context into Metadata in W3C form (traceparent/tracestate); on the worker side, env.ExtractTrace(ctx) continues the same trace. That makes the HTTP request → outbox → relay → worker → listener path a single trace, in the shop from placing the order to the confirmation e-mail. The dispatchers and the worker middleware do this automatically; see the OpenTelemetry page for details.

Task types

The kit uses four task-type prefixes, and the helpers are the single source of the wire names:

The worker routes tasks to its four built-in handlers by these prefixes; the Is*TaskType / JobName helpers give the same classification to anyone working with raw tasks (say, in a handler registered via WithMux). These are also the names you see in Asynqmon and in the worker's logs: you can tell at a glance which task is which listener of which event.

When to reach for it directly

• Your own, non-event task: build an asynq.NewTask(...), submit it with client.Enqueue, and register a handler for it in the worker with WithMux.
• An idempotent "act exactly once" enqueue: TaskID + ErrDuplicate handling, as above.
• Everything else (event dispatch, scheduling) belongs to the higher layers: start at events.