Run background jobs

Two pieces that are often used together but are independent. gas.Worker is the framework without a router or HTTP server. gas/queue is an SQS-backed gas.JobQueueProvider. You can enqueue from a web app and consume in a worker, or use either alone.

go get github.com/gasmod/gas/queue

Worker instead of App

cmd/worker/main.go

// Worker is App without the router and HTTP server: same container, same
// service lifecycle, same migrations. Run blocks until SIGINT or SIGTERM.
func main() {
  cfg := config.New(config.WithProvider(providers.NewEnvProvider()))
  if err := cfg.Load(); err != nil {
    log.Fatal(err)
  }

  w := gas.NewWorker(
    gas.WithServiceInstance[gas.ConfigProvider](cfg),
    gas.WithSingletonService[gas.Logger](gaslog.NewSlogLogger()),
    gas.WithSingletonService[gas.JobQueueProvider](queuesqs.New()),
  )

  if err := w.Run(); err != nil {
    log.Fatal(err)
  }
}

Worker gives you the same DI container, service lifecycle, events, and migrations as App. App actually embeds Worker, so every registration option works with both. Only the HTTP-specific options, such as WithErrorHandler and WithTrustedOrigin, are App-only.

Run blocks until SIGINT or SIGTERM. Start does the same setup without blocking, which is what you want when something else owns the process loop.

Enqueueing

func enqueue(ctx context.Context, q gas.JobQueueProvider, queueURL string, payload []byte) error {
  return q.Enqueue(ctx, queueURL, payload,
    gas.WithDelay(10*time.Second),
    gas.WithGroupID("order-123"),
    gas.WithDedupeID("order-123-confirmation"),
    gas.WithJobAttributes(map[string]string{"env": "prod"}),
  )
}

The queue argument is the full SQS queue URL. WithGroupID and WithDedupeID map to FIFO ordering and deduplication and are ignored on standard queues.

Consuming

The interface is deliberately pull-based, so you own the loop, the concurrency, and the failure policy.

// The interface is pull-based: run your own loop, then acknowledge. Ack removes
// the message; Nack makes it immediately visible again for retry.
func consume(ctx context.Context, q gas.JobQueueProvider, queueURL string, logger gas.Logger) {
  for {
    jobs, err := q.Dequeue(ctx, queueURL, 10, 20*time.Second)
    if err != nil {
      logger.Error("dequeue failed").Err("error", err).Send()
      continue
    }

    for _, job := range jobs {
      if err := process(job); err != nil {
        _ = q.Nack(ctx, queueURL, job)
        continue
      }
      _ = q.Ack(ctx, queueURL, job)
    }
  }
}

Dequeue long-polls for up to the wait duration. Ack deletes the message. Nack makes it immediately visible again rather than waiting out the visibility timeout, which is what you want for a fast retry. Do neither and the message reappears when its visibility timeout expires, which is the safe default if your process dies mid-job.

At-least-once, so make handlers idempotent

SQS can deliver the same message more than once, and a crash between finishing work and calling Ack guarantees it. Key your side effects on something stable from the payload rather than assuming exactly-once delivery.

On Lambda

cmd/lambda/main.go

// In Lambda, build the worker once at package init so it survives across
// invocations, and resolve the handler from the container.
func lambdaSetup() *gas.Worker {
  w := gas.NewWorker(
    gas.WithSingletonService[gas.Logger](gaslog.NewSlogLogger()),
  )
  if err := w.Start(); err != nil { // Start does not block
    log.Fatal(err)
  }
  return w
}

Build the worker in init so it survives across invocations, resolve your handler from the container once, then hand it to lambda.Start. Wire Worker.Shutdown into lambda.WithEnableSIGTERM so services close cleanly when the environment is frozen. The lambda-worker example is a complete version of this.

Config

Field	Default	Description
Queue.Region	us-east-1	AWS region
Queue.Endpoint		Custom endpoint, for ElasticMQ. Empty uses AWS
Queue.AccessKeyID		Static access key. Empty uses the default chain
Queue.SecretAccessKey		Static secret key
Queue.VisibilityTimeout	30s	How long a dequeued message stays hidden
Queue.WaitTimeSeconds	20	Long-poll duration, 0 to 20 (an SQS limit)

Testing

queuetest.MockQueue records calls and delegates to EnqueueFn, DequeueFn, AckFn, and NackFn, so a consumer loop can be driven through a fixed set of jobs with no AWS involved.