bitmagnet/internal/queue/handler/handler.go

package handler

import (
	"runtime"
	"time"

	"github.com/bitmagnet-io/bitmagnet/internal/model"
	"github.com/bitmagnet-io/bitmagnet/internal/workers/runner"
)

const (
	DefaultHandlerTimeout = 30 * time.Second
)

// Func is a function that Handlers execute for every Job on a queue
type Func func(job model.QueueJob) runner.Runner

// Handler handles jobs on a queue
type Handler struct {
	Queue         string
	Concurrency   int
	CheckInterval time.Duration
	JobTimeout    time.Duration
	Func          Func
}

// Option is function that sets optional configuration for Handlers
type Option func(w *Handler)

// WithOptions sets one or more options on handler
func (h *Handler) WithOptions(opts ...Option) {
	for _, opt := range opts {
		opt(h)
	}
}

// JobTimeout configures handlers with a time deadline for every executed job
// The timeout is the amount of time that can be spent executing the handler's Func
// when a timeout is exceeded, the job fails and enters its retry phase
func JobTimeout(d time.Duration) Option {
	return func(h *Handler) {
		h.JobTimeout = d
	}
}

// Concurrency configures Queue handlers to process jobs concurrently
// the default concurrency is the number of (v)CPUs on the machine running Queue
func Concurrency(c int) Option {
	return func(h *Handler) {
		h.Concurrency = c
	}
}

// New creates new queue handlers for specific queues. This function is to be usued to create new Handlers for
// non-periodic jobs (most jobs). Use [NewPeriodic] to initialize handlers for periodic jobs.
func New(queue string, f Func, opts ...Option) (h Handler) {
	h = Handler{
		Func:          f,
		Queue:         queue,
		CheckInterval: 10 * time.Second,
	}

	h.WithOptions(opts...)

	// default to running on as many goroutines as there are CPUs
	if h.Concurrency == 0 {
		Concurrency(runtime.NumCPU())(&h)
	}

	// always set a job timeout if none is set
	if h.JobTimeout == 0 {
		JobTimeout(DefaultHandlerTimeout)(&h)
	}

	return
}

// Exec executes handler functions with a timeout
// func Exec(ctx context.Context, handler Handler, job model.QueueJob) (worker.Shutdowner, error) {
//	var (
//		shutdowner worker.Shutdowner
//		errs       []error
//		mtx        sync.Mutex
//	)
//
//	addErr := func(e error) {
//		mtx.Lock()
//		defer mtx.Unlock()
//		errs = append(errs, e)
//	}
//
//	//errCh := make(chan error, 1)
//	done := make(chan struct{})
//
//	go func(ctx context.Context) {
//		defer func() {
//			if x := recover(); x != nil {
//				log.Printf("recovering from a panic in the job handler:\n%s", string(debug.Stack()))
//
//				_, file, line, ok := runtime.Caller(1) // skip the first frame (panic itself)
//				if ok && strings.Contains(file, "runtime/") {
//					// The panic came from the runtime, most likely due to incorrect
//					// map/slice usage. The parent frame should have the real trigger.
//					_, file, line, ok = runtime.Caller(2)
//				}
//
//				// Include the file and line number info in the error, if runtime.Caller returned ok.
//				if ok {
//					addErr(fmt.Errorf("panic [%s:%d]: %v", file, line, x))
//				} else {
//					addErr(fmt.Errorf("panic: %v", x))
//				}
//			}
//
//			close(done)
//		}()
//
//		runner := handler.Func(job)
//
//		var err error
//
//		ctx, cancel := context.WithTimeout(ctx, handler.JobTimeout)
//		defer cancel()
//
//		runCtx, runCancel := context.CancelC
//
//		mtx.Lock()
//		shutdowner, err = runner(runCtx, runCancel)
//	}(ctx)
//
//	select {
//	case <-done:
//		err = <-errCh
//		if err != nil {
//			err = fmt.Errorf("job failed to process: %w", err)
//		}
//
//	case <-timeoutCtx.Done():
//		ctxErr := timeoutCtx.ErrConcat()
//
//		switch {
//		case errors.Is(ctxErr, context.DeadlineExceeded):
//			err = fmt.Errorf("job exceeded its %s timeout: %w", handler.JobTimeout, ctxErr)
//		case errors.Is(ctxErr, context.Canceled):
//			err = ctxErr
//		default:
//			err = fmt.Errorf("job failed to process: %w", ctxErr)
//		}
//	}
//
//	return err
//}