gobullmq

https://pkg.go.dev/go.codycody31.dev/gobullmq
Admin
gobullmq is a BullMQ-compatible, Redis-backed job queue for Go that allows Go and Node.js processes...
Tokens:56,328
Snippets:400
Trust Score:9.6
Update:2 months ago
Show doc for...
Tokens:
Raw
### Install gobullmq via go get

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Commands to install specific versions of the gobullmq module.

```bash
go get go.codycody31.dev/gobullmq@v1.1.0  # exact release
go get go.codycody31.dev/gobullmq@v1.1    # latest fix on BullMQ v4.12.2
```

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

### Start Worker

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Starts the worker execution using the provided context.

```go
func (w *Worker[D, R]) Run(ctx context.Context) error
```

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

### func (*QueueEvents) Run

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Starts the queue events consumer.

```APIDOC
## func (*QueueEvents) Run

### Description
Run starts the queue events and listens for events from the redis stream.

### Signature
`func (qe *QueueEvents) Run(ctx context.Context) error`
```

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

### func (*Worker[D, R]) Run

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Starts the worker.

```APIDOC
## func (*Worker[D, R]) Run

### Description
Run starts the worker with the given context.

### Signature
`func (w *Worker[D, R]) Run(ctx context.Context) error`
```

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

### QueueEvents Lifecycle Management

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Methods to start, stop, and verify the status of the event consumer.

```go
func (qe *QueueEvents) Close() error
```

```go
func (qe *QueueEvents) Name() string
```

```go
func (qe *QueueEvents) Ping(ctx context.Context) error
```

```go
func (qe *QueueEvents) Run(ctx context.Context) error
```

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

### Lua Redis Scripting Variables and Setup

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/lua

Initializes variables and unpacks arguments from Redis calls.

```lua
local jobId
local jobIdKey
local rcall = redis.call
local args = cmsgpack.unpack(ARGV[1])
local data = ARGV[2]
local opts = cmsgpack.unpack(ARGV[3])
local parentKey = args[5]
local repeatJobKey = args[9]
local parent = args[8]
local parentData
```

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

### ProcessedOn

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Returns when processing started.

```go
func (j *Job[D]) ProcessedOn() time.Time
```

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

### Get Target Queue List

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/lua

Determines whether to return the wait queue or the paused queue based on the queue meta state.

```lua
local function getTargetQueueList(queueMetaKey, waitKey, pausedKey)
  if rcall("HEXISTS", queueMetaKey, "paused") ~= 1 then
    return waitKey, false
  else
    return pausedKey, true
  end
end
```

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

### Initialize Queue and Worker

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Demonstrates setting up a queue and worker with typed payloads and graceful shutdown handling.

```go
package main

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

	"github.com/redis/go-redis/v9"
	"go.codycody31.dev/gobullmq"
)

// EmailJob is the typed job payload (anything JSON-marshalable works).
type EmailJob struct {
	To      string `json:"to"`
	Subject string `json:"subject"`
}

func main() {
	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer cancel()

	queueClient := redis.NewClient(&redis.Options{Addr: "127.0.0.1:6379"})
	workerClient := redis.NewClient(&redis.Options{Addr: "127.0.0.1:6379"})
	defer queueClient.Close()
	defer workerClient.Close()

	// Queue[EmailJob]: constructor does no I/O and takes no context.
	queue, err := gobullmq.NewQueue[EmailJob]("mail", queueClient, nil)
	if err != nil {
		log.Fatal(err)
	}
	defer queue.Close()

	job, err := queue.Add(ctx, "welcome", EmailJob{To: "user@example.com", Subject: "Welcome!"},
		gobullmq.AddWithAttempts(3),
		gobullmq.AddWithBackoff(gobullmq.BackoffOptions{Type: "exponential", Delay: 500}),
	)
	if err != nil {
		log.Fatal(err)
	}
	fmt.Println("added job", job.ID())

	// Worker[EmailJob, string]: processes EmailJob payloads, returns string results.
	process := func(ctx context.Context, job *gobullmq.Job[EmailJob]) (string, error) {
		fmt.Printf("sending %q to %s\n", job.Data().Subject, job.Data().To)
		return "sent", nil
	}
	worker, err := gobullmq.NewWorker[EmailJob, string]("mail", workerClient, process, &gobullmq.WorkerOptions{
		Concurrency: 4,
	})
	if err != nil {
		log.Fatal(err)
	}

	if err := worker.Run(ctx); err != nil { // starts processing; returns immediately
		log.Fatal(err)
	}

	<-ctx.Done() // wait for SIGINT/SIGTERM

	// Graceful shutdown: stop fetching, drain in-flight jobs, bounded by ctx.
	shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer shutdownCancel()
	if err := worker.Shutdown(shutdownCtx); err != nil {
		log.Printf("shutdown: %v", err)
	}
}
```

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

### Options

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Returns the options the job was created with.

```go
func (j *Job[D]) Options() JobOptions
```

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

### Initialize and Manage Queue

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Core methods for creating and interacting with a Redis-backed queue.

```go
type Queue[D any] struct {
	// contains filtered or unexported fields
}
```

```go
func NewQueue[D any](name string, client redis.Cmdable, opts *QueueOptions) (*Queue[D], error)
```

```go
func (q *Queue[D]) ActiveCount(ctx context.Context) (int64, error)
```

```go
func (q *Queue[D]) ActiveJobs(ctx context.Context, start, end int) ([]*Job[D], error)
```

```go
func (q *Queue[D]) Add(ctx context.Context, jobName string, data D, addOpts ...AddOption) (*Job[D], error)
```

```go
func (q *Queue[D]) AddBulk(ctx context.Context, jobs []BulkJob[D]) ([]*Job[D], error)
```

```go
func (q *Queue[D]) AddJobLog(ctx context.Context, jobId string, message string, keepLogs int64) (int64, error)
```

```go
func (q *Queue[D]) ChildrenValues(ctx context.Context, jobId string) (map[string]any, error)
```

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

### QueueOptions Configuration

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Configuration options for creating a new Queue instance.

```go
type QueueOptions struct {
	Prefix             string
	StreamEventsMaxLen int64
	// DefaultJobOptions are merged under the per-call options of every Add
	// and AddBulk, like upstream defaultJobOptions.
	DefaultJobOptions *JobOptions
}
```

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

### Load function signature

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/compatibility

Reads and validates a compatibility manifest.

```go
func Load(path string) (Metadata, error)
```

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

### Get Job Attempts

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Method to return the number of processing attempts made by a job.

```go
func (j *Job[D]) AttemptsMade() int
```

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

### Prepare job for processing

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/lua

Initializes job processing by checking rate limits and setting up the job key.

```lua
local function prepareJobForProcessing(keys, keyPrefix, targetKey, jobId, processedOn,
    maxJobs, expireTime, opts)
  local jobKey = keyPrefix .. jobId
  -- Check if we need to perform rate limiting.
  if maxJobs then
    local rateLimiterKey = keys[6];
```

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

### Get Rate Limit TTL

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/lua

Calculates the TTL for rate limiting based on the current job count.

```lua
local function getRateLimitTTL(maxJobs, rateLimiterKey)
  if maxJobs and maxJobs <= tonumber(rcall("GET", rateLimiterKey) or 0) then
    local pttl = rcall("PTTL", rateLimiterKey)
    if pttl == 0 then
      rcall("DEL", rateLimiterKey)
    end
    if pttl > 0 then
      return pttl
    end
  end
  return 0
end
```

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

### QueueEventsOptions Configuration

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Configuration options for initializing a QueueEvents instance.

```go
type QueueEventsOptions struct {
	Autorun     bool   // If true, run the queue events immediately after creation
	Prefix      string // Prefix for the queue events key
	LastEventId string // Resume reading the stream after this event id (default: only new events)
}
```

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

### Get Next Delayed Job Timestamp

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/lua

Retrieves the timestamp of the next delayed job from a sorted set.

```lua
local function getNextDelayedTimestamp(delayedKey)
  local result = rcall("ZRANGE", delayedKey, 0, 0, "WITHSCORES")
  if #result then
    local nextTimestamp = tonumber(result[2])
    if (nextTimestamp ~= nil) then 
      nextTimestamp = nextTimestamp / 0x1000
    end
    return nextTimestamp
  end
end
```

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

### Get latest timestamp from job attributes

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/lua

Retrieves the first available timestamp from a list of provided job attributes.

```lua
local function getTimestamp(jobKey, attributes)
  if #attributes == 1 then
    return rcall("HGET", jobKey, attributes[1])
  end
  local jobTs
  for _, ts in ipairs(rcall("HMGET", jobKey, unpack(attributes))) do
    if (ts) then
      jobTs = ts
      break
    end
  end
  return jobTs
end
```

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

### MoveStalledJobsToWaitScript

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/lua

Initializes the script for moving stalled jobs back to the wait queue.

```lua
var MoveStalledJobsToWaitScript = redis.NewScript(`--[[
  Move stalled jobs to wait.
    Input:
      KEYS[1] 'stalled' (SET)
      KEYS[2] 'wait',   (LIST)
      KEYS[3] 'active', (LIST)
      KEYS[4] 'failed', (ZSET)
```

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

### Worker Configuration Options

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Defines the configuration structure for initializing a Worker.

```go
type WorkerOptions struct {
	Autorun          bool
	Concurrency      int
	Limiter          *RateLimiterOptions
	Metrics          *MetricsOptions
	Prefix           string
	MaxStalledCount  int
	StalledInterval  time.Duration
	RemoveOnComplete *KeepJobs
	RemoveOnFail     *KeepJobs
	SkipStalledCheck bool
	SkipLockRenewal  bool
	DrainDelay       time.Duration
	LockDuration     time.Duration
	LockRenewTime    time.Duration
	RunRetryDelay    time.Duration
	Backoff          *BackoffOptions
	BackoffStrategy  BackoffStrategyFunc
}
```

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

### Initialize QueueEvents

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Creates a new instance to consume Redis event streams.

```go
func NewQueueEvents(ctx context.Context, name string, client redis.Cmdable, opts *QueueEventsOptions) (*QueueEvents, error)
```

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

### Define FlowGetOptions

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Configuration structure for fetching flow trees.

```go
type FlowGetOptions struct {
	ID          string
	QueueName   string
	Prefix      string // defaults to the producer's prefix
	Depth       int    // how many levels of children to fetch (default 10)
	MaxChildren int    // maximum children fetched per node (default 20)
}
```

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

### Job Management and Querying

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Methods for retrieving, counting, promoting, and removing jobs from the queue.

```go
func (q *Queue[D]) PrioritizedCount(ctx context.Context) (int64, error)
```

```go
func (q *Queue[D]) PrioritizedJobs(ctx context.Context, start, end int) ([]*Job[D], error)
```

```go
func (q *Queue[D]) PromoteJobs(ctx context.Context) error
```

```go
func (q *Queue[D]) Remove(ctx context.Context, jobID string) error
```

```go
func (q *Queue[D]) RemoveWithChildren(ctx context.Context, jobID string) error
```

```go
func (q *Queue[D]) RetryJobs(ctx context.Context, opts RetryJobsOptions) error
```

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

### Initialize Job and Handle Duplicates

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/lua

Logic for validating parent existence, incrementing job counters, and handling duplicate job IDs.

```lua
if parentKey ~= nil then
  if rcall("EXISTS", parentKey) ~= 1 then
    return -5
  end
  parentData = cjson.encode(parent)
end
local jobCounter = rcall("INCR", KEYS[4])
-- Trim events before emiting them to avoid trimming events emitted in this script
trimEvents(KEYS[3], KEYS[8])
local parentDependenciesKey = args[7]
local timestamp = args[4]
if args[2] == "" then
  jobId = jobCounter
  jobIdKey = args[1] .. jobId
else
  jobId = args[2]
  jobIdKey = args[1] .. jobId
  if rcall("EXISTS", jobIdKey) == 1 then
    if parentKey ~= nil then
      if rcall("ZSCORE", KEYS[7], jobId) ~= false then
        local returnvalue = rcall("HGET", jobIdKey, "returnvalue")
        updateParentDepsIfNeeded(parentKey, parent['queueKey'], parentDependenciesKey,
          parent['id'], jobIdKey, returnvalue, timestamp)
      else
        if parentDependenciesKey ~= nil then
          rcall("SADD", parentDependenciesKey, jobIdKey)
        end
      end
      rcall("HMSET", jobIdKey, "parentKey", parentKey, "parent", parentData)
    end
    rcall("XADD", KEYS[8], "*", "event", "duplicated", "jobId", jobId)
    return jobId .. "" -- convert to string
  end
end
```

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

### func Load

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/compatibility

Reads and validates a compatibility manifest from a file path.

```APIDOC
## func Load(path string) (Metadata, error)

### Description
Reads and validates a compatibility manifest from the specified file path.

### Parameters
- **path** (string) - The file path to the manifest.

### Returns
- **Metadata** - The loaded compatibility metadata.
- **error** - Returns an error if the file cannot be read or validated.
```

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

### Schedule Repeatable Jobs

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Add jobs with repeat options using cron patterns or millisecond intervals, and manage existing repeatables.

```go
_, err = queue.Add(ctx, "report", EmailJob{To: "ops@example.com", Subject: "Hourly report"},
	gobullmq.AddWithRepeat(gobullmq.JobRepeatOptions{
		Pattern: "0 * * * *", // hourly; or Every: 10000 for every 10s
	}),
)
```

```go
repeatables, err := queue.RepeatableJobs(ctx, 0, -1, gobullmq.SortAsc)
if err != nil {
	log.Fatal(err)
}
for _, r := range repeatables {
	if err := queue.RemoveRepeatableByKey(ctx, r.Key); err != nil {
		log.Printf("remove %s: %v", r.Key, err)
	}
}
```

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

### Queue Status Methods

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/fifoqueue

Methods to inspect the current state and metrics of the queue.

```go
func (q *FifoQueue[T]) IsClosed() bool
```

```go
func (q *FifoQueue[T]) NumPending() int
```

```go
func (q *FifoQueue[T]) NumQueued() int
```

```go
func (q *FifoQueue[T]) NumTotal() int
```

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

### RepeatKeyOpts Struct

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/utils/repeat

Holds the configuration fields required to construct a repeat key.

```go
type RepeatKeyOpts struct {
	EndDate int64 // Unix epoch milliseconds, 0 when unset
	TZ      string
	Pattern string
	Every   int
	JobId   string
}
```

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

### Log

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Adds a log entry to the job, respecting the KeepLogs configuration.

```go
func (j *Job[D]) Log(ctx context.Context, message string) (int64, error)
```

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

### Initialize EventEmitter

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/eventEmitter

Creates a new instance of the event emitter.

```go
func NewEventEmitter() *EventEmitter
```

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

### Define backoff options

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/utils/backoff

Configuration structure for defining the backoff strategy and base delay in milliseconds.

```go
type Options struct {
	Type  string `json:"type" msgpack:"type"`
	Delay int    `json:"delay" msgpack:"delay"`
}
```

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

### Timestamp

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Returns the job's creation time.

```go
func (j *Job[D]) Timestamp() time.Time
```

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

### Promote

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Promotes a delayed job to the waiting state.

```go
func (j *Job[D]) Promote(ctx context.Context) error
```

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

### RateLimiterOptions Configuration

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Configuration for capping worker throughput.

```go
type RateLimiterOptions struct {
	Max      int `msgpack:"max"`
	Duration int `msgpack:"duration"`
}
```

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

### Worker Lifecycle Methods

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Methods for managing worker state and lifecycle.

```go
func (w *Worker[D, R]) Close() error
```

```go
func (w *Worker[D, R]) IsPaused() bool
```

```go
func (w *Worker[D, R]) IsRunning() bool
```

```go
func (w *Worker[D, R]) Name() string
```

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

### MoveToCompleted

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Transitions the job to the completed state with a return value, optionally fetching the next job.

```go
func (j *Job[D]) MoveToCompleted(ctx context.Context, returnValue any, fetchNext bool) (*Job[D], error)
```

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

### Initialize Event Stream

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Writes metadata for the events stream; required only when the instance is used exclusively for reads.

```go
func (q *Queue[D]) InitStream(ctx context.Context) error
```

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

### Queue.Add

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Adds a new job to the queue.

```APIDOC
## func (q *Queue[D]) Add(ctx context.Context, jobName string, data D, addOpts ...AddOption) (*Job[D], error)

### Description
Adds a new job to the queue. For repeatable jobs, a nil job with nil error is returned when no iteration is due.
```

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

### Define Obliterate and Parent Options

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Configuration structures for queue maintenance and job relationships.

```go
type ObliterateOptions struct {
	Force bool
	Count int
}
```

```go
type ParentOptions struct {
	ID    string `json:"id" msgpack:"id"`
	Queue string `json:"queue" msgpack:"queue"`
}
```

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

### Query Job States and Retry

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Retrieve job counts, list jobs by state, and retry failed jobs.

```go
counts, err := queue.JobCounts(ctx, gobullmq.JobStateWaiting, gobullmq.JobStateActive, gobullmq.JobStateFailed)
if err != nil {
	log.Fatal(err)
}
fmt.Println("waiting:", counts[gobullmq.JobStateWaiting])

failed, err := queue.Jobs(ctx, []gobullmq.JobState{gobullmq.JobStateFailed}, 0, 9, gobullmq.SortDesc)
if err != nil {
	log.Fatal(err)
}
for _, j := range failed {
	// Retry back onto the wait list, at the front (LIFO) or back (FIFO).
	if err := j.Retry(ctx, gobullmq.JobStateFailed, gobullmq.FIFO); err != nil {
		log.Printf("retry %s: %v", j.ID(), err)
	}
}

state, err := queue.JobState(ctx, job.ID())
if err != nil {
	log.Fatal(err)
}
fmt.Println(state == gobullmq.JobStateCompleted)
```

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

### Initialize FifoQueue

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/fifoqueue

Creates a new queue instance with a specified number of workers.

```go
func NewFifoQueue[T any](workers int, ignoreErrors bool) *FifoQueue[T]
```

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

### Worker Job Processing Methods

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Methods for manual job fetching and event listener management.

```go
func (w *Worker[D, R]) NextJob(ctx context.Context, token string) (*Job[D], error)
```

```go
func (w *Worker[D, R]) Off(event string, id ListenerID)
```

```go
func (w *Worker[D, R]) On(event string, listener func(...any)) ListenerID
```

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

### Define JobRepeatOptions Structure

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Configuration for repeatable jobs. Date fields are expected as Unix epoch milliseconds.

```go
type JobRepeatOptions struct {
	CurrentDate  int64  `json:"currentDate,omitempty" msgpack:"currentDate,omitempty"`
	StartDate    int64  `json:"startDate,omitempty" msgpack:"startDate,omitempty"`
	EndDate      int64  `json:"endDate,omitempty" msgpack:"endDate,omitempty"`
	UTC          bool   `json:"utc,omitempty" msgpack:"utc,omitempty"`
	TZ           string `json:"tz,omitempty" msgpack:"tz,omitempty"`
	NthDayOfWeek int    `json:"nthDayOfWeek,omitempty" msgpack:"nthDayOfWeek,omitempty"`
	Pattern      string `json:"pattern,omitempty" msgpack:"pattern,omitempty"`
	Limit        int    `json:"limit,omitempty" msgpack:"limit,omitempty"`
	Every        int    `json:"every,omitempty" msgpack:"every,omitempty"`
	Immediately  bool   `json:"immediately,omitempty" msgpack:"immediately,omitempty"`
	Count        int    `json:"count,omitempty" msgpack:"count,omitempty"`
	Offset       int64  `json:"offset,omitempty" msgpack:"offset,omitempty"`
	JobID        string `json:"jobId,omitempty" msgpack:"jobId,omitempty"`
	// contains filtered or unexported fields
}
```

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

### Event Handling Methods

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Methods for registering listeners for queue events.

```APIDOC
## func (*Queue[D]) On

### Description
Listens for the event and returns a ListenerID that can be used with Off.

### Signature
`func (q *Queue[D]) On(event string, listener func(...any)) ListenerID`

## func (*Queue[D]) Off

### Description
Removes a specific listener by its ListenerID.

### Signature
`func (q *Queue[D]) Off(event string, id ListenerID)`
```

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

### Name

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Returns the job's name.

```go
func (j *Job[D]) Name() string
```

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

### State

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Returns the current state of the job.

```go
func (j *Job[D]) State(ctx context.Context) (JobState, error)
```

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

### Define DrainOptions

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Configuration structure for the Drain operation.

```go
type DrainOptions struct {
	IncludeDelayed bool
}
```

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

### Define JobOptions Structure

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Configuration structure for job settings including priority, retry attempts, and delay. Fields use JSON and msgpack tags for serialization compatibility.

```go
type JobOptions struct {
	Priority         int       `json:"priority,omitempty" msgpack:"priority,omitempty"`
	RemoveOnComplete *KeepJobs `json:"removeOnComplete,omitempty" msgpack:"removeOnComplete,omitempty"`
	RemoveOnFail     *KeepJobs `json:"removeOnFail,omitempty" msgpack:"removeOnFail,omitempty"`
	// Attempts and Delay are always serialized: upstream stores
	// { attempts: 0, delay: 0 } defaults on every job.
	Attempts  int    `json:"attempts" msgpack:"attempts"`
	Delay     int    `json:"delay" msgpack:"delay"`
	Timestamp int64  `json:"timestamp,omitempty" msgpack:"timestamp,omitempty"`
	Lifo      bool   `json:"lifo,omitempty" msgpack:"lifo,omitempty"`
	JobID     string `json:"jobId,omitempty" msgpack:"jobId,omitempty"`
	// RepeatJobKey is carried on the job (rjk hash field) but stripped from
	// the serialized opts, like upstream's Job constructor destructuring.
	RepeatJobKey string            `json:"repeatJobKey,omitempty" msgpack:"repeatJobKey,omitempty"`
	Repeat       *JobRepeatOptions `json:"repeat,omitempty" msgpack:"repeat,omitempty"`
	// FailParentOnFailure, RemoveDependencyOnFailure and KeepLogs use
	// upstream's shortened Redis keys (RedisJobOptions: fpof/rdof/kl).
	FailParentOnFailure       bool           `json:"fpof,omitempty" msgpack:"fpof,omitempty"`
	Parent                    *ParentOptions `json:"parent,omitempty" msgpack:"parent,omitempty"`
	RemoveDependencyOnFailure bool           `json:"rdof,omitempty" msgpack:"rdof,omitempty"`
	// KeepLogs caps the number of log entries kept per job (0 = unlimited).
	KeepLogs int `json:"kl,omitempty" msgpack:"kl,omitempty"`
	// StackTraceLimit caps the stacktrace entries kept on failures (0 = unlimited).
	StackTraceLimit int `json:"stackTraceLimit,omitempty" msgpack:"stackTraceLimit,omitempty"`
	// SizeLimit rejects jobs whose JSON data exceeds this many bytes (0 = unlimited).
	SizeLimit int             `json:"sizeLimit,omitempty" msgpack:"sizeLimit,omitempty"`
	Backoff   *BackoffOptions `json:"backoff,omitempty" msgpack:"backoff,omitempty"`
	// PrevMillis is the slot of the previous repeatable iteration (epoch ms).
	// Upstream stores it top-level on the job opts, not inside repeat.
	PrevMillis   int64 `json:"prevMillis,omitempty" msgpack:"prevMillis,omitempty"`
	WaitChildren bool  `json:"-" msgpack:"-"`
}
```

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

### Initialize FlowProducer

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Function to create a new FlowProducer instance using an externally managed Redis client.

```go
func NewFlowProducer(client redis.Cmdable, opts *FlowProducerOptions) (*FlowProducer, error)
```

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

### RepeatableJobInfo Metadata

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Metadata structure for repeatable job entries.

```go
type RepeatableJobInfo struct {
	Key     string
	Name    string
	ID      string
	EndDate string
	TZ      string
	Pattern string
	Every   string
	Next    int64
}
```

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

### Fetch main branch directly

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Command to bypass the proxy and fetch the main branch directly.

```bash
GOPROXY=direct go get go.codycody31.dev/gobullmq@main
```

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

### Execute UpdateProgress Lua script

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/lua

Executes the updateProgress Lua script on Redis using 2 keys.

```go
func UpdateProgress(ctx context.Context, client redis.Cmdable, keys []string, args ...any) (any, error)
```

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

### Redis Lua Script Go Wrappers

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/lua

Go function signatures for executing various Lua scripts on Redis.

```go
func AddJob(ctx context.Context, client redis.Cmdable, keys []string, args ...any) (any, error)
```

```go
func ChangeDelay(ctx context.Context, client redis.Cmdable, keys []string, args ...any) (any, error)
```

```go
func ChangePriority(ctx context.Context, client redis.Cmdable, keys []string, args ...any) (any, error)
```

```go
func CleanJobsInSet(ctx context.Context, client redis.Cmdable, keys []string, args ...any) (any, error)
```

```go
func Drain(ctx context.Context, client redis.Cmdable, keys []string, args ...any) (any, error)
```

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

### Define Queue Metrics Structures

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Structures for tracking and reporting queue performance metrics.

```go
type MetricsMeta struct {
	Count     int64
	PrevTS    int64
	PrevCount int64
}
```

```go
type MetricsOptions struct {
	MaxDataPoints int
}
```

```go
type MetricsResult struct {
	Meta  MetricsMeta
	Data  []string
	Count int64
}
```

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

### On

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/eventEmitter

Registers a listener function for a specific event.

```APIDOC
## func (*EventEmitter) On

### Description
Registers a function to be called whenever the specified event is emitted. Returns a ListenerID for later removal.

### Signature
`func (e *EventEmitter) On(event string, fn func(...any)) ListenerID`
```

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

### NewWorker Constructor

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Creates a new Worker instance with optional configuration.

```go
func NewWorker[D any, R any](name string, client redis.Cmdable, processor ProcessFunc[D, R], opts *WorkerOptions) (*Worker[D, R], error)
```

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

### Logs

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Retrieves paginated logs for the job, ordered from oldest to newest.

```go
func (j *Job[D]) Logs(ctx context.Context, start, end int64) ([]string, int64, error)
```

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

### Metadata struct definition

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/compatibility

Defines the branch-local BullMQ compatibility contract.

```go
type Metadata struct {
	SchemaVersion int    `json:"schemaVersion"`
	ReleaseLine   string `json:"releaseLine"`
	Branch        string `json:"branch"`
	BullMQ        struct {
		Version string `json:"version"`
		Tag     string `json:"tag"`
		Commit  string `json:"commit"`
	} `json:"bullmq"`
	Go struct {
		Minimum string `json:"minimum"`
		CI      string `json:"ci"`
	} `json:"go"`
	Node struct {
		Version string `json:"version"`
	} `json:"node"`
	Redis struct {
		Version string   `json:"version"`
		Mode    []string `json:"mode"`
	} `json:"redis"`
}
```

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

### Queue Maintenance

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Utility methods for queue maintenance and configuration cleanup.

```go
func (q *Queue[D]) RateLimitTTL(ctx context.Context) (time.Duration, error)
```

```go
func (q *Queue[D]) RemoveDeprecatedPriorityKey(ctx context.Context) error
```

```go
func (q *Queue[D]) TrimEvents(ctx context.Context, max int64) (int64, error)
```

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

### Retrieve Queue Name

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Returns the name of the queue instance.

```go
func (q *Queue[D]) Name() string
```

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

### Log

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Adds a log entry for the job.

```APIDOC
## func (*Job[D]) Log

### Description
Log adds a log entry for the job, honoring the job's KeepLogs option.

### Signature
`func (j *Job[D]) Log(ctx context.Context, message string) (int64, error)`
```

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

### NewQueue

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Creates a new Queue instance using the provided Redis client and configuration options.

```APIDOC
## func NewQueue[D any](name string, client redis.Cmdable, opts *QueueOptions) (*Queue[D], error)

### Description
Creates a new Queue instance. The redis client is externally managed and will not be closed by the queue.
```

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

### Queue Event Data Structures

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Definitions for event data and the event consumer structure.

```go
type QueueEventData struct {
	JobID    string
	StreamID string
	Data     map[string]any
}
```

```go
type QueueEvents struct {
	// contains filtered or unexported fields
}
```

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

### CleanJobsInSet

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/lua

Executes the cleanJobsInSet Lua script.

```APIDOC
## CleanJobsInSet

### Description
Executes the cleanJobsInSet Lua script on Redis with 2 keys.

### Parameters
- **ctx** (context.Context) - Execution context
- **client** (redis.Cmdable) - Redis client
- **keys** ([]string) - 2 required keys
- **args** (...any) - Arguments for the script
```

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

### Delay

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Returns the initial processing delay configured for the job.

```go
func (j *Job[D]) Delay() time.Duration
```

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

### Repeatable Job Management

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Methods for managing jobs that repeat on a schedule.

```go
func (q *Queue[D]) RemoveRepeatable(ctx context.Context, name string, repeatOpts JobRepeatOptions) error
```

```go
func (q *Queue[D]) RemoveRepeatableByKey(ctx context.Context, key string) error
```

```go
func (q *Queue[D]) RepeatableJobs(ctx context.Context, start, end int, order SortOrder) ([]RepeatableJobInfo, error)
```

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

### Define FlowProducerOptions

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Configuration structure for the FlowProducer.

```go
type FlowProducerOptions struct {
	Prefix string
}
```

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

### Store and Queue New Job

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/lua

Logic for persisting job data to Redis and placing the job into the appropriate queue based on priority or delay.

```lua
-- Store the job.
local jsonOpts = cjson.encode(opts)
local delay = opts['delay'] or 0
local priority = opts['priority'] or 0
local optionalValues = {}
if parentKey ~= nil then
  table.insert(optionalValues, "parentKey")
  table.insert(optionalValues, parentKey)
  table.insert(optionalValues, "parent")
  table.insert(optionalValues, parentData)
end
if repeatJobKey ~= nil then
  table.insert(optionalValues, "rjk")
  table.insert(optionalValues, repeatJobKey)
end
rcall("HMSET", jobIdKey, "name", args[3], "data", ARGV[2], "opts", jsonOpts,
  "timestamp", timestamp, "delay", delay, "priority", priority, unpack(optionalValues))
rcall("XADD", KEYS[8], "*", "event", "added", "jobId", jobId, "name", args[3])
-- Check if job is delayed
local delayedTimestamp = (delay > 0 and (timestamp + delay)) or 0
-- Check if job is a parent, if so add to the parents set
local waitChildrenKey = args[6]
if waitChildrenKey ~= nil then
  rcall("ZADD", waitChildrenKey, timestamp, jobId)
  rcall("XADD", KEYS[8], "*", "event", "waiting-children", "jobId", jobId)
elseif (delayedTimestamp ~= 0) then
  local score = delayedTimestamp * 0x1000 + bit.band(jobCounter, 0xfff)
  rcall("ZADD", KEYS[5], score, jobId)
  rcall("XADD", KEYS[8], "*", "event", "delayed", "jobId", jobId, "delay",
        delayedTimestamp)
  -- If wait list is empty, and this delayed job is the next one to be processed,
  -- then we need to signal the workers by adding a dummy job (jobId 0:delay) to the wait list.
  local target = getTargetQueueList(KEYS[3], KEYS[1], KEYS[2])
  addDelayMarkerIfNeeded(target, KEYS[5])
else
  local target, paused = getTargetQueueList(KEYS[3], KEYS[1], KEYS[2])
  -- Standard or priority add
  if priority == 0 then
    -- LIFO or FIFO
    local pushCmd = opts['lifo'] and 'RPUSH' or 'LPUSH'
    rcall(pushCmd, target, jobId)
  else
    addJobWithPriority(KEYS[1], KEYS[6], priority, paused, jobId, KEYS[9])
  end
  -- Emit waiting event
  rcall("XADD", KEYS[8], "*", "event", "waiting", "jobId", jobId)
end
```

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

### Manage Event Listeners

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Methods for registering and removing event listeners on the queue.

```go
func (q *Queue[D]) Off(event string, id ListenerID)
```

```go
func (q *Queue[D]) On(event string, listener func(...any)) ListenerID
```

```go
func (q *Queue[D]) OnCleaned(fn func(jobIDs []string, state JobState)) ListenerID
```

```go
func (q *Queue[D]) OnError(fn func(err error)) ListenerID
```

```go
func (q *Queue[D]) OnPaused(fn func()) ListenerID
```

```go
func (q *Queue[D]) OnResumed(fn func()) ListenerID
```

```go
func (q *Queue[D]) OnWaiting(fn func(job *Job[D])) ListenerID
```

```go
func (q *Queue[D]) Once(event string, listener func(...any)) ListenerID
```

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

### Execute RetryJob Lua script

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/lua

Executes the retryJob Lua script on Redis using 9 keys.

```go
func RetryJob(ctx context.Context, client redis.Cmdable, keys []string, args ...any) (any, error)
```

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

### Metadata validation methods

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/compatibility

Methods for validating branches, documents, dependencies, and manifest schemas.

```go
func (metadata Metadata) CheckCurrentBranch(branch string) error
```

```go
func (metadata Metadata) CheckDocuments(root string) error
```

```go
func (metadata Metadata) CheckNodeLock(root string) error
```

```go
func (metadata Metadata) CheckUpstream(root, upstream string) error
```

```go
func (metadata Metadata) Validate() error
```

```go
func (metadata Metadata) ValidateBranch(branch string) error
```

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

### Process Job Completion and Queue Management

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/lua

Handles job completion logic, including parent dependency updates, job removal, and fetching the next job from the queue.

```lua
    trimEvents(KEYS[9], KEYS[4])
    -- If job has a parent we need to
    -- 1) remove this job id from parents dependencies
    -- 2) move the job Id to parent "processed" set
    -- 3) push the results into parent "results" list
    -- 4) if parent's dependencies is empty, then move parent to "wait/paused". Note it may be a different queue!.
    if parentId == "" and parentKey ~= "" then
        parentId = getJobIdFromKey(parentKey)
        parentQueueKey = getJobKeyPrefix(parentKey, ":" .. parentId)
    end
    if parentId ~= "" then
        if ARGV[5] == "completed" then
            local dependenciesSet = parentKey .. ":dependencies"
            if rcall("SREM", dependenciesSet, jobIdKey) == 1 then
                updateParentDepsIfNeeded(parentKey, parentQueueKey,
                                         dependenciesSet, parentId, jobIdKey,
                                         ARGV[4], timestamp)
            end
        else
            if opts['fpof'] then
                moveParentFromWaitingChildrenToFailed(parentQueueKey, parentKey,
                                            parentId, jobIdKey, timestamp)
            elseif opts['rdof'] then
                local dependenciesSet = parentKey .. ":dependencies"
                if rcall("SREM", dependenciesSet, jobIdKey) == 1 then
                    moveParentToWaitIfNeeded(parentQueueKey, dependenciesSet,
                        parentKey, parentId, timestamp)
                end
            end
        end
    end
    -- Remove job?
    if maxCount ~= 0 then
        local targetSet = KEYS[11]
        -- Add to complete/failed set
        rcall("ZADD", targetSet, timestamp, jobId)
        rcall("HMSET", jobIdKey, ARGV[3], ARGV[4], "finishedOn", timestamp)
        -- "returnvalue" / "failedReason" and "finishedOn"
        -- Remove old jobs?
        local prefix = ARGV[8]
        if maxAge ~= nil then
            removeJobsByMaxAge(timestamp, maxAge, targetSet, prefix)
        end
        if maxCount ~= nil and maxCount > 0 then
            removeJobsByMaxCount(maxCount, targetSet, prefix)
        end
    else
        rcall("DEL", jobIdKey, jobIdKey .. ':logs', jobIdKey .. ':processed')
        if parentKey ~= "" then
            removeParentDependencyKey(jobIdKey, false, parentKey)
        end
    end
    rcall("XADD", KEYS[4], "*", "event", ARGV[5], "jobId", jobId, ARGV[3],
          ARGV[4])
    if ARGV[5] == "failed" then
        if tonumber(attemptsMade) >= tonumber(attempts) then
            rcall("XADD", KEYS[4], "*", "event", "retries-exhausted", "jobId",
                  jobId, "attemptsMade", attemptsMade)
        end
    end
    -- Collect metrics
    if maxMetricsSize ~= "" then
        collectMetrics(KEYS[13], KEYS[13] .. ':data', maxMetricsSize, timestamp)
    end
    -- Try to get next job to avoid an extra roundtrip if the queue is not closing,
    -- and not rate limited.
    if (ARGV[7] == "1") then
        local target, paused = getTargetQueueList(KEYS[9], KEYS[1], KEYS[8])
        -- Check if there are delayed jobs that can be promoted
        promoteDelayedJobs(KEYS[7], KEYS[1], target, KEYS[3],
                           KEYS[4], ARGV[8], timestamp, paused, KEYS[10])
        local maxJobs = tonumber(opts['limiter'] and opts['limiter']['max'])
        -- Check if we are rate limited first.
        local expireTime = getRateLimitTTL(maxJobs, KEYS[6])
        if expireTime > 0 then return {0, 0, expireTime, 0} end
        -- paused queue
        if paused then return {0, 0, 0, 0} end
        jobId = rcall("RPOPLPUSH", KEYS[1], KEYS[2])
        if jobId then
            if string.sub(jobId, 1, 2) == "0:" then
                rcall("LREM", KEYS[2], 1, jobId)
                -- If jobId is special ID 0:delay (delay greater than 0), then there is no job to process
                -- but if ID is 0:0, then there is at least 1 prioritized job to process
                if jobId == "0:0" then
                    jobId = moveJobFromPriorityToActive(KEYS[3], KEYS[2], KEYS[10])
                    return prepareJobForProcessing(KEYS, ARGV[8], target, jobId, timestamp,
                        maxJobs, expireTime, opts)
                end
            else
                return prepareJobForProcessing(KEYS, ARGV[8], target, jobId, timestamp, maxJobs,
                    expireTime, opts)
            end
        else
            jobId = moveJobFromPriorityToActive(KEYS[3], KEYS[2], KEYS[10])
            if jobId then
                return prepareJobForProcessing(KEYS, ARGV[8], target, jobId, timestamp, maxJobs,
                    expireTime, opts)
            end
        end
        -- Return the timestamp for the next delayed job if any.
        local nextTimestamp = getNextDelayedTimestamp(KEYS[7])
        if nextTimestamp ~= nil then
            -- The result is guaranteed to be positive, since the
            -- ZRANGEBYSCORE command would have return a job otherwise.
            return {0, 0, 0, nextTimestamp}
```

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

### Add

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Creates a job flow (dependency tree) atomically in a single Redis transaction.

```APIDOC
## func (*FlowProducer) Add(ctx context.Context, flow FlowJob) (*FlowJobResult, error)

### Description
Creates a job flow (dependency tree). The whole tree is queued in a single Redis transaction like upstream, so a partial failure cannot strand a parent in waiting-children.
```

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

### AddJob

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/lua

Executes the addJob Lua script.

```APIDOC
## AddJob

### Description
Executes the addJob Lua script on Redis with 9 keys.

### Parameters
- **ctx** (context.Context) - Execution context
- **client** (redis.Cmdable) - Redis client
- **keys** ([]string) - 9 required keys
- **args** (...any) - Arguments for the script
```

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

### Fetch Flow Tree

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Method to fetch a flow tree rooted at a specific job.

```go
func (fp *FlowProducer) Flow(ctx context.Context, opts FlowGetOptions) (*FlowJobResult, error)
```

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

### Execute SaveStacktrace Lua script

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/lua

Executes the saveStacktrace Lua script on Redis using 1 key.

```go
func SaveStacktrace(ctx context.Context, client redis.Cmdable, keys []string, args ...any) (any, error)
```

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

### GetKey Function

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq/internal/utils/repeat

Returns the key for a repeatable job, formatted as name:jobId:endDate:tz:suffix.

```go
func GetKey(name string, opts RepeatKeyOpts) string
```

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

### SortOrder Enumeration

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Ordering options for paginated getters.

```go
type SortOrder int
```

```go
const (
	// SortAsc returns entries oldest-first.
	SortAsc SortOrder = iota
	// SortDesc returns entries newest-first.
	SortDesc
)
```

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

### Manage Worker and Queue Lifecycle

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Control worker shutdown, pausing, and resuming, as well as queue-level state management.

```go
// Worker, graceful: stop fetching, drain in-flight jobs, bounded by ctx.
// When ctx expires first, remaining jobs are abandoned and the error wraps
// gobullmq.ErrShutdownTimeout.
shutdownCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
err = worker.Shutdown(shutdownCtx)

// Worker, immediate: in-flight jobs are abandoned (their process contexts
// are cancelled); the stalled checker of another worker will recover them.
err = worker.Close()

// Worker pause/resume without closing.
worker.Pause()                      // immediate; in-flight jobs keep running
err = worker.PauseAndWait(pauseCtx) // pause, then wait for in-flight jobs
worker.Resume()

// Queue pausing affects all workers on the queue (state lives in Redis).
err = queue.Pause(ctx)
err = queue.Resume(ctx)

// Queue and QueueEvents close locally, without a context; the Redis clients
// are yours to close.
err = queue.Close()
err = events.Close()
```

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

### Queue Add Functional Options

Source: https://pkg.go.dev/go.codycody31.dev/gobullmq

Various functional options for configuring job addition to a queue.

```go
func AddWithAttempts(times int) AddOption
func AddWithBackoff(opts BackoffOptions) AddOption
func AddWithDelay(delay time.Duration) AddOption
func AddWithFailParentOnFailure(fail bool) AddOption
func AddWithJobID(id string) AddOption
func AddWithKeepLogs(n int) AddOption
func AddWithLifo() AddOption
func AddWithParent(parentOpts ParentOptions) AddOption
func AddWithPriority(priority int) AddOption
func AddWithRemoveDependencyOnFailure(remove bool) AddOption
func AddWithRemoveOnComplete(keep KeepJobs) AddOption
func AddWithRemoveOnFail(keep KeepJobs) AddOption
func AddWithRepeat(repeatOpts JobRepeatOptions) AddOption
func AddWithSizeLimit(n int) AddOption
func AddWithStackTraceLimit(n int) AddOption
func AddWithTimestamp(tsMillis int64) AddOption
```
Tokens:
Raw