fm_be/internal/queue/worker.go

package queue

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

	"wucher/internal/shared/pkg/metrics"
)

type WorkerConfig struct {
	Concurrency      int
	MaxBatchSize     int
	ProcessTimeout   time.Duration
	ProcessingTTL    time.Duration
	IdempotencyTTL   time.Duration
	PollErrorDelay   time.Duration
	MetricsComponent string
}

type Worker struct {
	cfg        WorkerConfig
	consumer   QueueConsumer
	serializer MessageSerializer
	handler    EmailJobHandler
	retry      RetryPolicy
	idem       IdempotencyStore
	metrics    *Metrics
	health     *HealthReporter
	logger     *slog.Logger
	component  string
}

type processResult struct {
	delivery *Delivery
	envelope *Envelope
	decision RetryDecision
	err      error
}

func NewWorker(
	cfg WorkerConfig,
	consumer QueueConsumer,
	serializer MessageSerializer,
	handler EmailJobHandler,
	retry RetryPolicy,
	idem IdempotencyStore,
	metrics *Metrics,
	health *HealthReporter,
	logger *slog.Logger,
) *Worker {
	if cfg.Concurrency <= 0 {
		cfg.Concurrency = 1
	}
	if cfg.MaxBatchSize <= 0 || cfg.MaxBatchSize > cfg.Concurrency {
		cfg.MaxBatchSize = cfg.Concurrency
	}
	if cfg.ProcessTimeout <= 0 {
		cfg.ProcessTimeout = 30 * time.Second
	}
	if cfg.ProcessingTTL <= 0 {
		cfg.ProcessingTTL = 15 * time.Minute
	}
	if cfg.IdempotencyTTL <= 0 {
		cfg.IdempotencyTTL = 24 * time.Hour
	}
	if cfg.PollErrorDelay <= 0 {
		cfg.PollErrorDelay = time.Second
	}
	if retry == nil {
		retry = NewExponentialBackoffPolicy(time.Second, 5*time.Minute, 30*time.Second)
	}
	if idem == nil {
		idem = NoopIdempotencyStore{}
	}
	if metrics == nil {
		metrics = NewMetrics()
	}
	if health == nil {
		health = NewHealthReporter(3)
	}
	if logger == nil {
		logger = slog.Default()
	}
	return &Worker{
		cfg:        cfg,
		consumer:   consumer,
		serializer: serializer,
		handler:    handler,
		retry:      retry,
		idem:       idem,
		metrics:    metrics,
		health:     health,
		logger:     logger,
		component:  normalizeWorkerMetricsComponent(cfg.MetricsComponent),
	}
}

func (w *Worker) Run(ctx context.Context) error {
	if w.consumer == nil {
		return errors.New("queue consumer is required")
	}
	if w.serializer == nil {
		return errors.New("message serializer is required")
	}
	if w.handler == nil {
		return errors.New("message handler is required")
	}
	for {
		select {
		case <-ctx.Done():
			w.health.MarkDraining()
			return nil
		default:
		}

		deliveries, err := w.consumer.Receive(ctx, w.cfg.MaxBatchSize)
		if err != nil {
			if ctx.Err() != nil {
				w.health.MarkDraining()
				return nil
			}
			w.health.MarkPollFailure(err)
			w.logger.Error("queue receive failed",
				slog.String("backend", w.consumer.Backend()),
				slog.Any("error", err),
			)
			select {
			case <-ctx.Done():
				w.health.MarkDraining()
				return nil
			case <-time.After(w.cfg.PollErrorDelay):
			}
			continue
		}
		w.health.MarkPollSuccess()
		if len(deliveries) == 0 {
			continue
		}

		w.metrics.RecordReceive(len(deliveries))
		results := make(chan processResult, len(deliveries))
		for _, delivery := range deliveries {
			delivery := delivery
			go func() {
				results <- w.processDelivery(ctx, delivery)
			}()
		}

		acks := make([]*Delivery, 0, len(deliveries))
		for range deliveries {
			result := <-results
			switch result.decision.Action {
			case RetryActionAck:
				acks = append(acks, result.delivery)
			case RetryActionRetry:
				if err := w.consumer.Retry(ctx, result.delivery, result.decision.Delay); err != nil {
					w.metrics.RecordRetryFailure()
					w.logger.Error("queue retry scheduling failed",
						slog.String("backend", w.consumer.Backend()),
						slog.String("message_id", result.delivery.ID),
						slog.Any("error", err),
					)
				} else {
					w.metrics.RecordRetry()
				}
			case RetryActionDeadLetter:
				if err := w.consumer.DeadLetter(ctx, result.delivery, result.decision.Reason, result.err); err != nil {
					w.metrics.RecordDeadLetterFailure()
					w.logger.Error("queue dead-letter failed",
						slog.String("backend", w.consumer.Backend()),
						slog.String("message_id", result.delivery.ID),
						slog.Any("error", err),
					)
				} else {
					w.metrics.RecordDeadLetter()
				}
			}
		}

		if len(acks) > 0 {
			if err := w.consumer.Ack(ctx, acks); err != nil {
				w.metrics.RecordAckFailure()
				w.logger.Error("queue ack failed",
					slog.String("backend", w.consumer.Backend()),
					slog.Int("count", len(acks)),
					slog.Any("error", err),
				)
			}
		}
	}
}

func (w *Worker) RefreshQueueDepth(ctx context.Context) error {
	if w.consumer == nil {
		return errors.New("queue consumer is required")
	}
	depth, err := w.consumer.ApproximateDepth(ctx)
	if err != nil {
		return err
	}
	w.metrics.SetQueueDepth(depth)
	return nil
}

func (w *Worker) Metrics() *Metrics {
	return w.metrics
}

func (w *Worker) Health() *HealthReporter {
	return w.health
}

func (w *Worker) processDelivery(ctx context.Context, delivery *Delivery) (result processResult) {
	started := time.Now()
	jobType := "unknown"
	var jobErr error
	defer func() {
		metrics.ObserveWorkerJob(w.component, jobType, jobErr, started)
	}()

	envelope, err := w.serializer.Deserialize(delivery.Body, delivery.Attributes)
	if err != nil {
		w.metrics.RecordMalformed()
		w.metrics.RecordFailure()
		jobErr = err
		return processResult{
			delivery: delivery,
			decision: w.retry.Decide(delivery, nil, Permanent(err)),
			err:      err,
		}
	}
	jobType = strings.TrimSpace(envelope.Payload.NormalizedType())

	idempotencyKey := envelope.DeduplicationKey()
	state, err := w.idem.Acquire(ctx, idempotencyKey, w.cfg.ProcessingTTL)
	if err != nil {
		w.metrics.RecordFailure()
		jobErr = err
		return processResult{
			delivery: delivery,
			envelope: envelope,
			decision: w.retry.Decide(delivery, envelope, Transient(err)),
			err:      err,
		}
	}
	switch state {
	case IdempotencyStateDuplicate:
		w.metrics.RecordDuplicate()
		w.logger.Info("queue duplicate skipped",
			slog.String("message_id", envelope.MessageID),
			slog.String("job_type", envelope.Payload.NormalizedType()),
			slog.String("recipient_hash", envelope.Payload.RecipientHash()),
			slog.String("correlation_id", envelope.CorrelationID),
		)
		return processResult{
			delivery: delivery,
			envelope: envelope,
			decision: RetryDecision{Action: RetryActionAck},
		}
	case IdempotencyStateInProgress:
		inProgressErr := Transient(errors.New("message already in progress"))
		jobErr = inProgressErr
		return processResult{
			delivery: delivery,
			envelope: envelope,
			decision: w.retry.Decide(delivery, envelope, inProgressErr),
			err:      inProgressErr,
		}
	}

	processCtx, cancel := context.WithTimeout(ctx, w.cfg.ProcessTimeout)
	defer cancel()

	if err := w.handler.Handle(processCtx, envelope); err != nil {
		_ = w.idem.Release(ctx, idempotencyKey)
		w.metrics.RecordFailure()
		jobErr = err
		return processResult{
			delivery: delivery,
			envelope: envelope,
			decision: w.retry.Decide(delivery, envelope, err),
			err:      err,
		}
	}

	if err := w.idem.MarkCompleted(ctx, idempotencyKey, w.cfg.IdempotencyTTL); err != nil {
		w.metrics.RecordFailure()
		w.logger.Error("queue idempotency completion failed",
			slog.String("message_id", envelope.MessageID),
			slog.String("correlation_id", envelope.CorrelationID),
			slog.Any("error", err),
		)
		jobErr = err
		return processResult{
			delivery: delivery,
			envelope: envelope,
			decision: w.retry.Decide(delivery, envelope, Transient(err)),
			err:      err,
		}
	}

	w.metrics.RecordProcessed(time.Since(started))
	jobErr = nil
	return processResult{
		delivery: delivery,
		envelope: envelope,
		decision: RetryDecision{Action: RetryActionAck},
	}
}

func normalizeWorkerMetricsComponent(component string) string {
	component = strings.ToLower(strings.TrimSpace(component))
	if component == "" {
		return "worker"
	}
	return component
}