func WithErrQMap(f ErrQMap) WorkerOption {
	return func(w *Worker) {
		w.ErrQMap = f
	}
}
WithErrQMap sets a function that maps from inbox queue names to error queue names.
Defaults to DefaultErrQMap.
package entroq
import (
	"context"
	"errors"
	"fmt"
	"log"
	"time"
	"github.com/google/uuid"
)
const DefaultRetryDelay = 30 * time.Second
type ErrQMap func(inbox string) string
type DependencyHandler func(err DependencyError) error
type Worker struct {
	Qs []string
	ErrQMap ErrQMap
	OnDepErr DependencyHandler
	MaxAttempts int32
	eqc *EntroQ
	lease          time.Duration
	baseRetryDelay time.Duration  
}
type MoveTaskError struct {
	Err error
}
func NewMoveTaskError(err error) *MoveTaskError {
	return &MoveTaskError{Err: err}
}
func MoveTaskErrorf(format string, values ...interface{}) *MoveTaskError {
	return NewMoveTaskError(fmt.Errorf(format, values...))
}
func (e *MoveTaskError) Error() string {
	return e.Err.Error()
}
type RetryTaskError struct {
	Err error
}
func NewRetryTaskError(err error) *RetryTaskError {
	return &RetryTaskError{Err: err}
}
func RetryTaskErrorf(format string, values ...interface{}) *RetryTaskError {
	return NewRetryTaskError(fmt.Errorf(format, values...))
}
func (e *RetryTaskError) Error() string {
	return e.Err.Error()
}
func NewWorker(eq *EntroQ, qs ...string) *Worker {
	return &Worker{
		Qs:      qs,
		ErrQMap: DefaultErrQMap,
		eqc:   eq,
		lease: DefaultClaimDuration,
		baseRetryDelay: DefaultRetryDelay,
	}
}
func (c *EntroQ) NewWorker(qs ...string) *Worker {
	return NewWorker(c, qs...)
}
func (w *Worker) WithOpts(opts ...WorkerOption) *Worker {
	for _, opt := range opts {
		opt(w)
	}
	return w
}
type Work func(ctx context.Context, task *Task) ([]ModifyArg, error)
func taskErrMsg(err error) string {
	if err == nil {
		return ""
	}
	var firstErr error
	for e := err; e != nil; e = errors.Unwrap(e) {
		firstErr = e
	}
	return firstErr.Error()
}
func (w *Worker) Run(ctx context.Context, f Work) (err error) {
	if len(w.Qs) == 0 {
		return fmt.Errorf("No queues specified to work on")
	}
	for {
		select {
		case <-ctx.Done():
			return fmt.Errorf("worker quit: %w", ctx.Err())
		default:
		}
		task, err := w.eqc.Claim(ctx, From(w.Qs...), ClaimFor(w.lease))
		if err != nil {
			return fmt.Errorf("worker claim (%q): %w", w.Qs, err)
		}
		errQ := w.ErrQMap(task.Queue)
		var args []ModifyArg
		renewed, workErr := w.eqc.DoWithRenew(ctx, task, w.lease, func(ctx context.Context) error {
			var err error
			if args, err = f(ctx, task); err != nil {
				if e := new(RetryTaskError); errors.As(err, &e) {
					changeArgs := []ChangeArg{ErrTo(taskErrMsg(e)), AttemptToNext()}
					if w.MaxAttempts == 0 || task.Attempt+1 < w.MaxAttempts {
						log.Printf("Worker received retryable error, incrementing attempt: %v", e)
						changeArgs = append(changeArgs, ArrivalTimeBy(w.baseRetryDelay))
					} else {
						log.Printf("Worker max attempts reached, moving to %q instead of retrying: %v", errQ, e)
						changeArgs = append(changeArgs, QueueTo(errQ))
					}
					args = []ModifyArg{task.AsChange(changeArgs...)}
					return nil
				}
				if e := new(MoveTaskError); errors.As(err, &e) {
					log.Printf("Worker moving to %q: %v", errQ, err)
					args = []ModifyArg{task.AsChange(QueueTo(errQ), ErrTo(taskErrMsg(e)))}
					return nil
				}
				return fmt.Errorf("work (%q): %w", w.Qs, err)
			}
			return nil
		})
		if workErr != nil {
			if _, ok := AsDependency(workErr); ok {
				log.Printf("Worker continuing after dependency (%q)", w.Qs)
				continue
			}
			if IsTimeout(workErr) {
				log.Printf("Worker continuing after timeout (%q)", w.Qs)
				continue
			}
			if IsCanceled(workErr) {
				return nil
			}
			return fmt.Errorf("worker error: %w", workErr)
		}
		modification := NewModification(uuid.Nil, args...)
		for _, task := range modification.Changes {
			if task.ID == renewed.ID && task.Version != renewed.Version {
				if task.Version > renewed.Version {
					return fmt.Errorf("task updated inside worker body, expected version <= %v, got %v", renewed.Version, task.Version)
				}
				log.Printf("Rewriting change version %v => %v", task.Version, renewed.Version)
				task.Version = renewed.Version
			}
		}
		for _, id := range modification.Depends {
			if id.ID == renewed.ID && task.Version != renewed.Version {
				if task.Version > renewed.Version {
					return fmt.Errorf("task updated inside worker body, expected version <= %v, got %v", renewed.Version, task.Version)
				}
				log.Printf("Rewriting depend version %v => %v", task.Version, renewed.Version)
				id.Version = renewed.Version
			}
		}
		for _, id := range modification.Deletes {
			if id.ID == renewed.ID && task.Version != renewed.Version {
				if task.Version > renewed.Version {
					return fmt.Errorf("task updated inside worker body, expected version <= %v, got %v", renewed.Version, task.Version)
				}
				log.Printf("Rewriting delete version %v => %v", task.Version, renewed.Version)
				id.Version = renewed.Version
			}
		}
		if _, _, err := w.eqc.Modify(ctx, WithModification(modification)); err != nil {
			if depErr, ok := AsDependency(err); ok {
				if w.OnDepErr != nil {
					if err := w.OnDepErr(depErr); err != nil {
						log.Printf("Dependency error upgraded to fatal: %v", err)
						return fmt.Errorf("worker depdency error upgraded to fatal: %w", err)
					}
				}
				log.Printf("Worker ack failed (%q), throwing away: %v", w.Qs, err)
				continue
			}
			if IsTimeout(err) {
				log.Printf("Worker continuing (%q) after ack timeout: %v", w.Qs, err)
				continue
			}
			if IsCanceled(err) {
				log.Printf("Worker exiting cleanly (%q) instead of acking: %v", w.Qs, err)
				return nil
			}
			return fmt.Errorf("worker ack (%q): %w", w.Qs, err)
		}
	}
}
type WorkerOption func(*Worker)
func WithLease(d time.Duration) WorkerOption {
	return func(w *Worker) {
		w.lease = d
	}
}