fm_be/internal/app/worker/file_processing_outbox_dispatcher.go

package worker

import (
	"context"
	"fmt"
	"log/slog"
	"os"
	"time"

	"wucher/internal/config"
	filemanager "wucher/internal/domain/file_manager"
	"wucher/internal/queue"
)

type FileProcessingOutboxDispatcher struct {
	repo            filemanager.FileProcessingOutboxRepository
	producer        queue.QueueProducer
	logger          *slog.Logger
	retryPolicy     queue.RetryPolicy
	pollInterval    time.Duration
	dispatchTimeout time.Duration
	lockTTL         time.Duration
	maxAttempts     int
	batchSize       int
	workerID        string
	metrics         *outboxMetrics
}

func NewFileProcessingOutboxDispatcher(
	cfg config.FileQueueConfig,
	repo filemanager.FileProcessingOutboxRepository,
	producer queue.QueueProducer,
	logger *slog.Logger,
) *FileProcessingOutboxDispatcher {
	if !cfg.OutboxEnabled || repo == nil || producer == nil {
		return nil
	}
	if logger == nil {
		logger = slog.Default()
	}
	workerID := fmt.Sprintf("%s-%d", hostname(), os.Getpid())
	return &FileProcessingOutboxDispatcher{
		repo:            repo,
		producer:        producer,
		logger:          logger.With("component", "file-outbox-dispatcher", "backend", producer.Backend()),
		retryPolicy:     queue.NewExponentialBackoffPolicy(cfg.WorkerBackoffMin, cfg.WorkerBackoffMax, cfg.WorkerPermanentFailureDelay),
		pollInterval:    cfg.OutboxPollInterval,
		dispatchTimeout: cfg.OutboxDispatchTimeout,
		lockTTL:         cfg.OutboxLockTTL,
		maxAttempts:     cfg.OutboxMaxAttempts,
		batchSize:       cfg.OutboxBatchSize,
		workerID:        workerID,
		metrics:         &outboxMetrics{},
	}
}

func (d *FileProcessingOutboxDispatcher) Metrics() OutboxMetricsSnapshot {
	if d == nil {
		return OutboxMetricsSnapshot{}
	}
	return d.metrics.Snapshot()
}

func (d *FileProcessingOutboxDispatcher) Run(ctx context.Context) error {
	if d == nil {
		return nil
	}
	if d.pollInterval <= 0 {
		d.pollInterval = time.Second
	}
	for {
		processed, err := d.dispatchOnce(ctx)
		if err != nil {
			if ctx.Err() != nil {
				return nil
			}
			d.metrics.update(func(snapshot *OutboxMetricsSnapshot) {
				snapshot.Failures++
				snapshot.LastFailureAt = time.Now().UTC()
				snapshot.LastError = err.Error()
			})
			d.logger.Error("file outbox dispatch cycle failed", slog.Any("error", err))
		}
		if ctx.Err() != nil {
			return nil
		}
		if processed > 0 {
			continue
		}
		select {
		case <-ctx.Done():
			return nil
		case <-time.After(d.pollInterval):
		}
	}
}

func (d *FileProcessingOutboxDispatcher) dispatchOnce(ctx context.Context) (int, error) {
	now := time.Now().UTC()
	messages, err := d.repo.ClaimPendingFileProcessingOutboxMessages(ctx, d.batchSize, now, d.lockTTL, d.workerID)
	if err != nil {
		return 0, err
	}
	if len(messages) == 0 {
		d.refreshPending(ctx, now)
		return 0, nil
	}
	d.metrics.update(func(snapshot *OutboxMetricsSnapshot) {
		snapshot.Claimed += int64(len(messages))
	})
	for i := range messages {
		d.dispatchMessage(ctx, &messages[i])
	}
	d.refreshPending(ctx, time.Now().UTC())
	return len(messages), nil
}

func (d *FileProcessingOutboxDispatcher) dispatchMessage(ctx context.Context, message *filemanager.FileProcessingOutboxMessage) {
	if message == nil {
		return
	}
	outbound, err := message.OutboundMessage()
	if err != nil {
		d.failPermanently(ctx, message, err)
		return
	}
	queueMessage := queue.OutboundMessage{
		ID:              outbound.ID,
		Body:            outbound.Body,
		Attributes:      outbound.Attributes,
		MessageGroupID:  outbound.MessageGroupID,
		DeduplicationID: outbound.DeduplicationID,
	}

	dispatchCtx := ctx
	cancel := func() {}
	if d.dispatchTimeout > 0 {
		dispatchCtx, cancel = context.WithTimeout(ctx, d.dispatchTimeout)
	}
	err = d.producer.Publish(dispatchCtx, queueMessage)
	cancel()
	if err == nil {
		publishedAt := time.Now().UTC()
		if markErr := d.repo.MarkFileProcessingOutboxMessagePublished(ctx, message.ID, publishedAt); markErr != nil {
			d.metrics.update(func(snapshot *OutboxMetricsSnapshot) {
				snapshot.Failures++
				snapshot.LastFailureAt = time.Now().UTC()
				snapshot.LastError = markErr.Error()
			})
			d.logger.Error("file outbox publish marked failed",
				slog.String("message_id", message.MessageID),
				slog.Any("error", markErr),
			)
			return
		}
		d.metrics.update(func(snapshot *OutboxMetricsSnapshot) {
			snapshot.Published++
			snapshot.LastSuccessAt = publishedAt
			snapshot.LastError = ""
		})
		d.logger.Info("file outbox message published",
			slog.String("message_id", message.MessageID),
			slog.String("job_type", message.JobType),
			slog.String("correlation_id", message.CorrelationID),
			slog.Int("attempt", message.Attempts),
		)
		return
	}

	if message.Attempts >= d.maxAttempts {
		d.failPermanently(ctx, message, err)
		return
	}

	decision := d.retryPolicy.Decide(&queue.Delivery{ID: message.MessageID, ReceiveCount: message.Attempts}, nil, err)
	nextAttemptAt := time.Now().UTC().Add(decision.Delay)
	if retryErr := d.repo.MarkFileProcessingOutboxMessageRetry(ctx, message.ID, nextAttemptAt, err.Error()); retryErr != nil {
		d.metrics.update(func(snapshot *OutboxMetricsSnapshot) {
			snapshot.Failures++
			snapshot.LastFailureAt = time.Now().UTC()
			snapshot.LastError = retryErr.Error()
		})
		d.logger.Error("file outbox retry scheduling failed",
			slog.String("message_id", message.MessageID),
			slog.Any("error", retryErr),
		)
		return
	}
	d.metrics.update(func(snapshot *OutboxMetricsSnapshot) {
		snapshot.Retried++
		snapshot.LastFailureAt = time.Now().UTC()
		snapshot.LastError = err.Error()
	})
	d.logger.Warn("file outbox publish failed; scheduled retry",
		slog.String("message_id", message.MessageID),
		slog.String("job_type", message.JobType),
		slog.Duration("retry_in", decision.Delay),
		slog.Int("attempt", message.Attempts),
		slog.Any("error", err),
	)
}

func (d *FileProcessingOutboxDispatcher) failPermanently(ctx context.Context, message *filemanager.FileProcessingOutboxMessage, err error) {
	if message == nil || err == nil {
		return
	}
	failedAt := time.Now().UTC()
	if markErr := d.repo.MarkFileProcessingOutboxMessageDead(ctx, message.ID, failedAt, err.Error()); markErr != nil {
		d.metrics.update(func(snapshot *OutboxMetricsSnapshot) {
			snapshot.Failures++
			snapshot.LastFailureAt = failedAt
			snapshot.LastError = markErr.Error()
		})
		d.logger.Error("file outbox dead-letter marking failed",
			slog.String("message_id", message.MessageID),
			slog.Any("error", markErr),
		)
		return
	}
	d.metrics.update(func(snapshot *OutboxMetricsSnapshot) {
		snapshot.Dead++
		snapshot.LastFailureAt = failedAt
		snapshot.LastError = err.Error()
	})
	d.logger.Error("file outbox message marked dead",
		slog.String("message_id", message.MessageID),
		slog.String("job_type", message.JobType),
		slog.Int("attempt", message.Attempts),
		slog.Any("error", err),
	)
}

func (d *FileProcessingOutboxDispatcher) refreshPending(ctx context.Context, now time.Time) {
	count, err := d.repo.CountPendingFileProcessingOutboxMessages(ctx, now)
	if err != nil {
		return
	}
	d.metrics.update(func(snapshot *OutboxMetricsSnapshot) {
		snapshot.ApproximatePending = count
	})
}