middleware.go

package executor

import (
	"context"
	"fmt"
	"log/slog"
	"math/rand/v2"
	"sync/atomic"
	"time"

	"github.com/zalgonoise/x/errs"
)

const (
	// ErrPanic indicates an unhandled panic recovered during runner execution.
	ErrPanic = errs.Kind("panic")
	// ErrExecution identifies the runner execution entity in error values.
	ErrExecution = errs.Entity("runner execution")
)

// ErrRecoveredPanic is returned when a panic is recovered during a Runner execution.
var ErrRecoveredPanic = errs.WithDomain(errDomain, ErrPanic, ErrExecution)

// Recover wraps a Runner to recover from panics during execution, logging the panic if logger is non-nil
// and returning an error wrapping ErrRecoveredPanic.
func Recover(r Runner, logger *slog.Logger) Runner {
	if r == nil {
		return Runnable(nil)
	}

	return Runnable(func(ctx context.Context) (err error) {
		defer func() {
			if rec := recover(); rec != nil {
				err = fmt.Errorf("%w: %v", ErrRecoveredPanic, rec)

				if logger != nil {
					logger.ErrorContext(ctx, "recovered panic in runner execution",
						slog.Any("panic", rec),
					)
				}
			}
		}()

		return r.Run(ctx)
	})
}

// SkipIfRunning wraps a Runner so that concurrent invocations are skipped if a previous execution is still running.
func SkipIfRunning(r Runner, logger *slog.Logger) Runner {
	if r == nil {
		return Runnable(nil)
	}

	var running atomic.Bool

	return Runnable(func(ctx context.Context) error {
		if !running.CompareAndSwap(false, true) {
			if logger != nil {
				logger.DebugContext(ctx, "skipping runner execution: previous run still active")
			}

			return nil
		}

		defer running.Store(false)

		return r.Run(ctx)
	})
}

// Jitter wraps a Runner to sleep for a random duration in [0, maxJitter) before executing.
// If the context is canceled during the jitter delay, the context error is returned without running the job.
func Jitter(r Runner, maxJitter time.Duration) Runner {
	if r == nil {
		return Runnable(nil)
	}

	if maxJitter <= 0 {
		return r
	}

	return Runnable(func(ctx context.Context) error {
		//nolint:gosec // jitter delay does not require cryptographic randomness
		delay := rand.N(maxJitter)
		if delay > 0 {
			timer := time.NewTimer(delay)
			defer timer.Stop()

			select {
			case <-ctx.Done():
				return ctx.Err()
			case <-timer.C:
			}
		}

		return r.Run(ctx)
	})
}

// Retry wraps a Runner to retry failed executions up to maxRetries additional times,
// waiting backoff duration between attempts. If the context is canceled, retries stop immediately.
func Retry(r Runner, maxRetries int, backoff time.Duration) Runner {
	if r == nil {
		return Runnable(nil)
	}

	if maxRetries <= 0 {
		return r
	}

	return Runnable(func(ctx context.Context) error {
		var err error

		for attempt := 0; attempt <= maxRetries; attempt++ {
			if attempt > 0 && backoff > 0 {
				timer := time.NewTimer(backoff)

				select {
				case <-ctx.Done():
					timer.Stop()

					return ctx.Err()
				case <-timer.C:
				}
			}

			err = r.Run(ctx)
			if err == nil {
				return nil
			}

			if ctx.Err() != nil {
				return err
			}
		}

		return err
	})
}