story/internal/outbox/worker_test.go

package outbox

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"net"
	"strings"
	"testing"
	"time"

	"github.com/jackc/pgx/v5/pgconn"
)

func TestWorkerDefaultsAndRetryCap(t *testing.T) {
	worker := NewWorker(nil, discardLogger(), WorkerOptions{})
	if worker.options.PollInterval != 2*time.Second || worker.options.BatchSize != 20 || worker.options.MaxAttempts != 8 {
		t.Fatalf("unexpected worker defaults: %#v", worker.options)
	}
	worker.options.BaseRetry = time.Second
	worker.options.MaxRetry = 3 * time.Second
	delay := time.Until(worker.nextAttempt(10))
	if delay < 1400*time.Millisecond || delay > 3100*time.Millisecond {
		t.Fatalf("capped retry delay = %s", delay)
	}
	if !strings.HasPrefix(worker.workerID, "worker-") {
		t.Fatalf("unexpected worker id: %s", worker.workerID)
	}
}

func TestStoreRejectsInvalidEventBeforeDatabaseAccess(t *testing.T) {
	store := NewStore(nil)
	if _, err := store.Enqueue(context.Background(), nil, NewEvent{}); err == nil {
		t.Fatal("empty event should be rejected")
	}
	if _, err := store.Enqueue(context.Background(), nil, NewEvent{
		AggregateType: "test", AggregateID: "id", EventType: "type", Payload: json.RawMessage(`not-json`),
	}); err == nil {
		t.Fatal("invalid JSON payload should be rejected")
	}
}

func TestHashPayloadIsStable(t *testing.T) {
	first := hashPayload([]byte(`{"ok":true}`))
	if first != hashPayload([]byte(`{"ok":true}`)) || first == hashPayload([]byte(`{"ok":false}`)) {
		t.Fatal("payload hash is not stable")
	}
}

func TestRetryableErrorAllowList(t *testing.T) {
	permanent := errors.New("malformed payload")
	if IsRetryable(permanent) {
		t.Fatal("ordinary errors must be permanent by default")
	}
	if !IsRetryable(Retryable(permanent)) || !errors.Is(Retryable(permanent), permanent) {
		t.Fatal("explicit retryable marker was not preserved")
	}
	if !IsRetryable(context.DeadlineExceeded) || !IsRetryable(&net.DNSError{IsTimeout: true}) {
		t.Fatal("timeouts must remain retryable")
	}
	if !IsRetryable(&pgconn.PgError{Code: "40001"}) || IsRetryable(&pgconn.PgError{Code: "23505"}) {
		t.Fatal("PostgreSQL transient and constraint errors were misclassified")
	}
	if Retryable(nil) != nil {
		t.Fatal("wrapping nil must return nil")
	}
	if !IsRetryable(fmt.Errorf("dependency: %w", Retryable(permanent))) {
		t.Fatal("wrapped marker must remain discoverable")
	}
}

func TestWorkerHonorsOptionalDependencyRetryDelay(t *testing.T) {
	worker := NewWorker(nil, discardLogger(), WorkerOptions{BaseRetry: time.Second, MaxRetry: 3 * time.Second})
	cause := errors.New("provider rate limited")
	delayed := fmt.Errorf("privacy deletion: %w", RetryAfter(cause, time.Minute))
	if !IsRetryable(delayed) || !errors.Is(delayed, cause) || RetryDelay(delayed) != time.Minute {
		t.Fatalf("retry metadata lost through wrapping: %v", delayed)
	}
	if delay := time.Until(worker.nextAttemptForError(1, delayed)); delay < 59*time.Second || delay > 61*time.Second {
		t.Fatalf("dependency cooldown was not honored: %s", delay)
	}
	if delay := time.Until(worker.nextAttemptForError(1, Retryable(cause))); delay < 400*time.Millisecond || delay > 1100*time.Millisecond {
		t.Fatalf("ordinary retries changed: %s", delay)
	}
	if RetryAfter(nil, time.Minute) != nil || RetryDelay(cause) != 0 || RetryDelay(RetryAfter(cause, -time.Second)) != 0 {
		t.Fatal("nil, ordinary or negative retry delay was not handled")
	}
	if RetryDelay(RetryAfter(delayed, time.Second)) != time.Minute {
		t.Fatal("outer wrapper shortened dependency cooldown")
	}
}