queue.go

package backstage

import (
	"context"
	"errors"
	"log/slog"
	"sync"
	"time"
)

// ErrQueueNotFound is returned by [Supervisor.Dispatch] when no queue with the
// given name has been registered.
var ErrQueueNotFound = errors.New("queue not found")

// ErrQueueFull is returned by [Supervisor.Dispatch] when the queue's channel
// buffer is full and the job cannot be accepted.
var ErrQueueFull = errors.New("queue full")

// Store is the persistence backend for a named queue. The default
// implementation is an in-memory buffered channel. Implement this interface to
// plug in a durable backend (e.g. a database-backed task table) so that jobs
// survive process restarts.
//
// All methods must be safe to call from multiple goroutines.
type Store interface {
	// Push enqueues a job. Returns [ErrQueueFull] if the store cannot accept
	// it at this time.
	Push(job Job) error

	// Pop blocks until a job is available or ctx is cancelled.
	// Returns (Job{}, ctx.Err()) when the context is done.
	Pop(ctx context.Context) (Job, error)

	// Len returns the number of jobs currently held in the store.
	Len() int
}

// memStore is the default in-memory Store backed by a buffered channel.
type memStore struct {
	ch chan Job
}

func newMemStore(buffer int) *memStore {
	return &memStore{ch: make(chan Job, buffer)}
}

func (s *memStore) Push(job Job) error {
	select {
	case s.ch <- job:
		return nil
	default:
		return ErrQueueFull
	}
}

func (s *memStore) Pop(ctx context.Context) (Job, error) {
	select {
	case <-ctx.Done():
		return Job{}, ctx.Err()
	case job := <-s.ch:
		return job, nil
	}
}

func (s *memStore) Len() int {
	return len(s.ch)
}

// QueueConfig configures a named worker pool.
type QueueConfig struct {
	// Workers is the number of concurrent worker goroutines. Defaults to 1.
	Workers int

	// Buffer is the depth of the default in-memory store. Only applied when
	// Store is nil. Defaults to 32.
	Buffer int

	// Store is the persistence backend for this queue. When nil, an in-memory
	// buffered channel of depth [Buffer] is used. Provide a custom
	// implementation to get durable, auditable job storage.
	Store Store
}

// queue is an internal named worker pool.
type queue struct {
	name    string
	workers int
	store   Store
	log     *slog.Logger
}

func newQueue(name string, workers int, store Store, parentLog *slog.Logger) *queue {
	return &queue{
		name:    name,
		workers: workers,
		store:   store,
		log:     parentLog.With(slog.String("queue", name)),
	}
}

// start launches the worker goroutines. Each worker adds itself to wg and
// calls wg.Done when it exits so the supervisor can wait for a clean drain.
func (q *queue) start(ctx context.Context, wg *sync.WaitGroup) {
	for i := 0; i < q.workers; i++ {
		wg.Add(1)
		go q.worker(ctx, i, wg)
	}
}

func (q *queue) worker(ctx context.Context, id int, wg *sync.WaitGroup) {
	defer wg.Done()
	q.log.Debug("worker started", slog.Int("worker_id", id))
	defer q.log.Debug("worker stopped", slog.Int("worker_id", id))

	for {
		job, err := q.store.Pop(ctx)
		if err != nil {
			return
		}
		q.runJob(ctx, job, id)
	}
}

// runJob executes a single job, retrying up to job.Retries times on failure.
// A panic aborts the job immediately without retrying.
func (q *queue) runJob(ctx context.Context, job Job, workerID int) {
	log := q.log
	if job.ID != "" {
		log = q.log.With(slog.String("job_id", job.ID))
	}

	start := time.Now()
	log.Debug("job started",
		slog.String("job", job.Name),
		slog.Int("worker_id", workerID),
	)

	defer func() {
		if r := recover(); r != nil {
			log.Error("job panicked",
				slog.String("job", job.Name),
				slog.Int("worker_id", workerID),
				slog.Any("panic", r),
			)
		}
	}()

	var err error
	for attempt := 0; attempt <= job.Retries; attempt++ {
		if attempt > 0 {
			if !sleepRetry(ctx, job.RetryDelay) {
				return
			}
			log.Debug("job retrying",
				slog.String("job", job.Name),
				slog.Int("worker_id", workerID),
				slog.Int("attempt", attempt+1),
				slog.Int("of", job.Retries+1),
			)
		}
		err = job.Run(ctx)
		if err == nil {
			break
		}
	}

	dur := time.Since(start)

	if err != nil {
		log.Warn("job finished with error",
			slog.String("job", job.Name),
			slog.Int("worker_id", workerID),
			slog.Duration("duration", dur),
			slog.Any("error", err),
		)
		return
	}

	log.Debug("job finished",
		slog.String("job", job.Name),
		slog.Int("worker_id", workerID),
		slog.Duration("duration", dur),
	)
}

// dispatch enqueues a job onto the store.
func (q *queue) dispatch(job Job) error {
	return q.store.Push(job)
}

func (q *queue) depth() int {
	return q.store.Len()
}