fm_be/internal/resilience/breaker.go

package resilience

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"net"
	"strings"
	"time"

	"github.com/aws/smithy-go"
	"github.com/sony/gobreaker/v2"

	"wucher/internal/config"
)

type Classification int

const (
	ClassificationFailure Classification = iota
	ClassificationExcluded
)

type ErrorClassifier func(error) Classification

type Executor struct {
	name       string
	breaker    *gobreaker.CircuitBreaker[any]
	classifier ErrorClassifier
}

type OpenError struct {
	Dependency string
	Err        error
}

func (e *OpenError) Error() string {
	if e == nil {
		return ""
	}
	return fmt.Sprintf("%s circuit breaker: %v", e.Dependency, e.Err)
}

func (e *OpenError) Unwrap() error {
	if e == nil {
		return nil
	}
	return e.Err
}

type HTTPStatusError struct {
	Operation  string
	StatusCode int
	Body       string
}

func (e *HTTPStatusError) Error() string {
	if e == nil {
		return ""
	}
	body := strings.TrimSpace(e.Body)
	if len(body) > 256 {
		body = body[:256]
	}
	if body == "" {
		return fmt.Sprintf("%s: status=%d", e.Operation, e.StatusCode)
	}
	return fmt.Sprintf("%s: status=%d body=%s", e.Operation, e.StatusCode, body)
}

type classifiedError struct {
	err      error
	excluded bool
}

func (e *classifiedError) Error() string {
	if e == nil || e.err == nil {
		return ""
	}
	return e.err.Error()
}

func (e *classifiedError) Unwrap() error {
	if e == nil {
		return nil
	}
	return e.err
}

func NewExecutor(name string, policy config.CircuitBreakerPolicy, logger *slog.Logger, classifier ErrorClassifier) *Executor {
	policy = normalizePolicy(policy)
	if strings.TrimSpace(name) == "" {
		name = "dependency"
	}
	executor := &Executor{
		name:       name,
		classifier: classifier,
	}
	if !policy.Enabled {
		return executor
	}
	if logger == nil {
		logger = slog.Default()
	}
	cbLogger := logger.With("component", "circuit_breaker", "dependency", name)
	settings := gobreaker.Settings{
		Name:         name,
		MaxRequests:  policy.MaxRequests,
		Interval:     policy.Interval,
		BucketPeriod: policy.BucketPeriod,
		Timeout:      policy.Timeout,
		ReadyToTrip: func(counts gobreaker.Counts) bool {
			if policy.ConsecutiveFailures > 0 && counts.ConsecutiveFailures >= policy.ConsecutiveFailures {
				return true
			}
			total := counts.TotalSuccesses + counts.TotalFailures
			if total < policy.MinRequests || policy.FailureRatio <= 0 {
				return false
			}
			return float64(counts.TotalFailures)/float64(total) >= policy.FailureRatio
		},
		OnStateChange: func(name string, from, to gobreaker.State) {
			level := slog.LevelInfo
			if to == gobreaker.StateOpen {
				level = slog.LevelWarn
			}
			cbLogger.Log(context.Background(), level, "state changed",
				"from", from.String(),
				"to", to.String(),
			)
		},
		IsSuccessful: func(err error) bool {
			return err == nil
		},
		IsExcluded: func(err error) bool {
			var classified *classifiedError
			return errors.As(err, &classified) && classified.excluded
		},
	}
	executor.breaker = gobreaker.NewCircuitBreaker[any](settings)
	return executor
}

func (e *Executor) Enabled() bool {
	return e != nil && e.breaker != nil
}

func (e *Executor) Counts() gobreaker.Counts {
	if e == nil || e.breaker == nil {
		return gobreaker.Counts{}
	}
	return e.breaker.Counts()
}

func (e *Executor) Execute(ctx context.Context, fn func(context.Context) error) error {
	_, err := Do(ctx, e, func(ctx context.Context) (struct{}, error) {
		return struct{}{}, fn(ctx)
	})
	return err
}

func Do[T any](ctx context.Context, executor *Executor, fn func(context.Context) (T, error)) (T, error) {
	var zero T
	if executor == nil || executor.breaker == nil {
		return fn(ctx)
	}
	value, err := executor.breaker.Execute(func() (any, error) {
		result, err := fn(ctx)
		if err != nil {
			if executor.classifier != nil && executor.classifier(err) == ClassificationExcluded {
				return nil, &classifiedError{err: err, excluded: true}
			}
			return nil, &classifiedError{err: err}
		}
		return any(result), nil
	})
	if err != nil {
		if errors.Is(err, gobreaker.ErrOpenState) || errors.Is(err, gobreaker.ErrTooManyRequests) {
			return zero, &OpenError{Dependency: executor.name, Err: err}
		}
		var classified *classifiedError
		if errors.As(err, &classified) {
			return zero, classified.err
		}
		return zero, err
	}
	if value == nil {
		return zero, nil
	}
	typed, ok := value.(T)
	if !ok {
		return zero, fmt.Errorf("%s circuit breaker returned unexpected type %T", executor.name, value)
	}
	return typed, nil
}

func ClassifyMicrosoftSSOError(err error) Classification {
	switch {
	case err == nil:
		return ClassificationExcluded
	case errors.Is(err, context.Canceled):
		return ClassificationExcluded
	case errors.Is(err, context.DeadlineExceeded):
		return ClassificationFailure
	}
	var netErr net.Error
	if errors.As(err, &netErr) {
		return ClassificationFailure
	}
	var statusErr *HTTPStatusError
	if errors.As(err, &statusErr) {
		if statusErr.StatusCode == 429 || statusErr.StatusCode >= 500 {
			return ClassificationFailure
		}
		return ClassificationExcluded
	}
	return ClassificationFailure
}

func ClassifySESError(err error) Classification {
	switch {
	case err == nil:
		return ClassificationExcluded
	case errors.Is(err, context.Canceled):
		return ClassificationExcluded
	case errors.Is(err, context.DeadlineExceeded):
		return ClassificationFailure
	}
	var netErr net.Error
	if errors.As(err, &netErr) {
		return ClassificationFailure
	}
	var apiErr smithy.APIError
	if errors.As(err, &apiErr) {
		code := strings.ToLower(strings.TrimSpace(apiErr.ErrorCode()))
		if apiErr.ErrorFault() == smithy.FaultServer || isThrottleCode(code) {
			return ClassificationFailure
		}
		return ClassificationExcluded
	}
	message := strings.ToLower(strings.TrimSpace(err.Error()))
	if strings.Contains(message, "ses is not configured") ||
		strings.Contains(message, "recipient is required") ||
		strings.Contains(message, "invalid email job") {
		return ClassificationExcluded
	}
	if strings.Contains(message, "eof") {
		return ClassificationFailure
	}
	return ClassificationExcluded
}

func ClassifySQSError(err error) Classification {
	switch {
	case err == nil:
		return ClassificationExcluded
	case errors.Is(err, context.Canceled):
		return ClassificationExcluded
	case errors.Is(err, context.DeadlineExceeded):
		return ClassificationFailure
	}
	var netErr net.Error
	if errors.As(err, &netErr) {
		return ClassificationFailure
	}
	var apiErr smithy.APIError
	if errors.As(err, &apiErr) {
		code := strings.ToLower(strings.TrimSpace(apiErr.ErrorCode()))
		if apiErr.ErrorFault() == smithy.FaultServer || isThrottleCode(code) {
			return ClassificationFailure
		}
		return ClassificationExcluded
	}
	return ClassificationFailure
}

func normalizePolicy(policy config.CircuitBreakerPolicy) config.CircuitBreakerPolicy {
	if policy.MaxRequests == 0 {
		policy.MaxRequests = 1
	}
	if policy.Interval <= 0 {
		policy.Interval = time.Minute
	}
	if policy.BucketPeriod <= 0 {
		policy.BucketPeriod = 10 * time.Second
	}
	if policy.Timeout <= 0 {
		policy.Timeout = 30 * time.Second
	}
	if policy.MinRequests == 0 {
		policy.MinRequests = 5
	}
	if policy.FailureRatio <= 0 || policy.FailureRatio > 1 {
		policy.FailureRatio = 0.6
	}
	if policy.ConsecutiveFailures == 0 {
		policy.ConsecutiveFailures = 5
	}
	return policy
}

func isThrottleCode(code string) bool {
	switch code {
	case "throttling", "throttlingexception", "requestthrottled", "toomanyrequestsexception", "slowdown":
		return true
	default:
		return false
	}
}