package queue import ( "context" "errors" "testing" "time" "wucher/internal/config" "wucher/internal/shared/pkg/logger" ) type stubHandler struct { err error } func (h stubHandler) Handle(context.Context, *Envelope) error { return h.err } type stubIdempotencyStore struct { state IdempotencyState err error } func (s stubIdempotencyStore) Acquire(context.Context, string, time.Duration) (IdempotencyState, error) { if s.err != nil { return "", s.err } if s.state != "" { return s.state, nil } return IdempotencyStateAcquired, nil } func (s stubIdempotencyStore) MarkCompleted(context.Context, string, time.Duration) error { return nil } func (s stubIdempotencyStore) Release(context.Context, string) error { return nil } func TestWorkerProcessDeliveryDuplicate(t *testing.T) { worker := NewWorker(WorkerConfig{}, nil, NewJSONSerializer("v1", true, QueueTypeStandard, ""), stubHandler{}, nil, stubIdempotencyStore{state: IdempotencyStateDuplicate}, nil, nil, logger.New(config.LoggingConfig{Format: "json"})) message, err := NewJSONSerializer("v1", true, QueueTypeStandard, "").Serialize(context.Background(), EmailJob{ To: "user@example.com", Subject: "Hello", Body: "World", }) if err != nil { t.Fatalf("serialize: %v", err) } result := worker.processDelivery(context.Background(), &Delivery{ID: message.ID, Body: message.Body, Attributes: message.Attributes, ReceiveCount: 1}) if result.decision.Action != RetryActionAck { t.Fatalf("expected ack decision, got %s", result.decision.Action) } if worker.Metrics().Snapshot().Duplicates != 1 { t.Fatalf("expected duplicate metric to increment") } } func TestWorkerProcessDeliveryMalformedPayload(t *testing.T) { worker := NewWorker(WorkerConfig{}, nil, NewJSONSerializer("v1", false, QueueTypeStandard, ""), stubHandler{}, nil, stubIdempotencyStore{}, nil, nil, logger.New(config.LoggingConfig{Format: "json"})) result := worker.processDelivery(context.Background(), &Delivery{ID: "msg-1", Body: []byte(`{"bad":true}`), ReceiveCount: 1}) if result.decision.Action != RetryActionDeadLetter { t.Fatalf("expected dead-letter decision, got %s", result.decision.Action) } if worker.Metrics().Snapshot().Malformed != 1 { t.Fatalf("expected malformed metric to increment") } } func TestWorkerProcessDeliveryTransientError(t *testing.T) { worker := NewWorker(WorkerConfig{}, nil, NewJSONSerializer("v1", true, QueueTypeStandard, ""), stubHandler{err: Transient(errors.New("ses timeout"))}, nil, stubIdempotencyStore{}, nil, nil, logger.New(config.LoggingConfig{Format: "json"})) message, err := NewJSONSerializer("v1", true, QueueTypeStandard, "").Serialize(context.Background(), EmailJob{ To: "user@example.com", Subject: "Hello", Body: "World", }) if err != nil { t.Fatalf("serialize: %v", err) } result := worker.processDelivery(context.Background(), &Delivery{ID: message.ID, Body: message.Body, Attributes: message.Attributes, ReceiveCount: 2}) if result.decision.Action != RetryActionRetry { t.Fatalf("expected retry decision, got %s", result.decision.Action) } if result.decision.Delay <= 0 { t.Fatalf("expected positive retry delay") } }