nexarch/internal/notify/dispatcher_test.go

package notify

import (
	"context"
	"errors"
	"os"
	"sync"
	"testing"
	"time"

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

func setupTest(t *testing.T) (*pgxpool.Pool, func()) {
	t.Helper()
	adminDSN := os.Getenv("TEST_ADMIN_DSN")
	if adminDSN == "" {
		t.Skip("TEST_ADMIN_DSN nicht gesetzt, Integrationstest uebersprungen")
	}
	ctx := context.Background()

	pool, err := pgxpool.New(ctx, adminDSN)
	if err != nil {
		t.Fatalf("pool: %v", err)
	}
	if _, err := pool.Exec(ctx, `
		CREATE TABLE IF NOT EXISTS notification_jobs (
			id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
			channel TEXT NOT NULL,
			recipient TEXT NOT NULL,
			payload JSONB NOT NULL DEFAULT '{}'::jsonb,
			status TEXT NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'sent', 'failed')),
			attempts INT NOT NULL DEFAULT 0,
			max_attempts INT NOT NULL DEFAULT 5,
			next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT now(),
			last_error TEXT,
			created_at TIMESTAMPTZ NOT NULL DEFAULT now(),
			updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
		)`); err != nil {
		t.Fatalf("schema: %v", err)
	}

	cleanup := func() { pool.Close() }
	return pool, cleanup
}

type fakeSender struct {
	mu        sync.Mutex
	sentIDs   []string
	failUntil int
	calls     int
}

func (f *fakeSender) Send(ctx context.Context, n Notification) error {
	f.mu.Lock()
	defer f.mu.Unlock()
	f.calls++
	if f.calls <= f.failUntil {
		return errors.New("simulierter zustellfehler")
	}
	f.sentIDs = append(f.sentIDs, n.ID)
	return nil
}

func (f *fakeSender) sentCount() int {
	f.mu.Lock()
	defer f.mu.Unlock()
	return len(f.sentIDs)
}

// Akzeptanzkriterium 1: Module loesen ueber Enqueue aus, keine eigene
// Versandlogik noetig.
func TestDispatcher_EnqueueAndProcess(t *testing.T) {
	pool, cleanup := setupTest(t)
	defer cleanup()
	ctx := context.Background()

	d := NewDispatcher(pool)
	id, err := d.Enqueue(ctx, "email", "alice@example.com", map[string]any{"subject": "Willkommen"})
	if err != nil {
		t.Fatalf("enqueue: %v", err)
	}
	if id == "" {
		t.Fatal("erwartet nicht-leere id")
	}

	sender := &fakeSender{}
	sent, failed, err := d.ProcessDue(ctx, sender, 10)
	if err != nil {
		t.Fatalf("process: %v", err)
	}
	if sent != 1 || failed != 0 {
		t.Fatalf("erwartet sent=1 failed=0, habe sent=%d failed=%d", sent, failed)
	}
	if sender.sentCount() != 1 {
		t.Fatalf("erwartet 1 zustellung, habe %d", sender.sentCount())
	}
}

// Akzeptanzkriterium 2 + Pruefung 2: Wiederholungslogik greift bei
// simuliertem Fehler und bricht nach definierter Anzahl kontrolliert ab.
func TestProcessDue_RetriesThenGivesUpAfterMaxAttempts(t *testing.T) {
	pool, cleanup := setupTest(t)
	defer cleanup()
	ctx := context.Background()

	d := NewDispatcher(pool).WithRetryPolicy(3, time.Millisecond)
	id, err := d.Enqueue(ctx, "email", "bob@example.com", nil)
	if err != nil {
		t.Fatalf("enqueue: %v", err)
	}

	sender := &fakeSender{failUntil: 100} // schlaegt bei jedem versuch fehl

	for i := 0; i < 3; i++ {
		time.Sleep(5 * time.Millisecond) // next_attempt_at abwarten
		if _, _, err := d.ProcessDue(ctx, sender, 10); err != nil {
			t.Fatalf("process %d: %v", i, err)
		}
	}

	var status string
	var attempts int
	if err := pool.QueryRow(ctx, `SELECT status, attempts FROM notification_jobs WHERE id = $1`, id).Scan(&status, &attempts); err != nil {
		t.Fatalf("status lesen: %v", err)
	}
	if status != "failed" {
		t.Fatalf("erwartet status failed nach max_attempts, habe %q", status)
	}
	if attempts != 3 {
		t.Fatalf("erwartet 3 versuche, habe %d", attempts)
	}

	// Weiteres ProcessDue darf den bereits aufgegebenen job nicht mehr anfassen.
	sent, failed, err := d.ProcessDue(ctx, sender, 10)
	if err != nil {
		t.Fatalf("process nach abbruch: %v", err)
	}
	if sent != 0 || failed != 0 {
		t.Fatalf("erwartet keine weitere verarbeitung, habe sent=%d failed=%d", sent, failed)
	}
}

// Akzeptanzkriterium 3 + Pruefung 1: Neustart des Dienstes waehrend offener
// Zustellung verliert keine Nachricht — simuliert durch eine komplett neue
// Dispatcher/Pool-Instanz nach dem Enqueue, bevor irgendetwas verarbeitet wurde.
func TestQueue_SurvivesRestartWithoutMessageLoss(t *testing.T) {
	pool, cleanup := setupTest(t)
	defer cleanup()
	ctx := context.Background()

	firstInstance := NewDispatcher(pool)
	id, err := firstInstance.Enqueue(ctx, "email", "carol@example.com", nil)
	if err != nil {
		t.Fatalf("enqueue: %v", err)
	}

	// "Neustart": eine voellig neue Dispatcher-Instanz (repraesentiert einen
	// neuen Prozess) verbindet sich neu und verarbeitet die Warteschlange —
	// die Nachricht existiert ausschliesslich in Postgres, nicht im
	// Prozessspeicher der ersten Instanz.
	restartedInstance := NewDispatcher(pool)
	sender := &fakeSender{}
	sent, failed, err := restartedInstance.ProcessDue(ctx, sender, 10)
	if err != nil {
		t.Fatalf("process nach neustart: %v", err)
	}
	if sent != 1 || failed != 0 {
		t.Fatalf("erwartet sent=1 nach neustart, habe sent=%d failed=%d", sent, failed)
	}
	if len(sender.sentIDs) != 1 || sender.sentIDs[0] != id {
		t.Fatalf("erwartet zustellung der urspruenglichen nachricht %q, habe %v", id, sender.sentIDs)
	}
}

// Akzeptanzkriterium 3 + Pruefung 3: zwei gleichzeitig ausloesende Module,
// beide Nachrichten werden korrekt (und nicht doppelt) zugestellt.
func TestProcessDue_ConcurrentDispatchBothDelivered(t *testing.T) {
	pool, cleanup := setupTest(t)
	defer cleanup()
	ctx := context.Background()

	d := NewDispatcher(pool)
	idA, err := d.Enqueue(ctx, "email", "modul-a@example.com", nil)
	if err != nil {
		t.Fatalf("enqueue a: %v", err)
	}
	idB, err := d.Enqueue(ctx, "email", "modul-b@example.com", nil)
	if err != nil {
		t.Fatalf("enqueue b: %v", err)
	}

	sender := &fakeSender{}
	var wg sync.WaitGroup
	for i := 0; i < 2; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			if _, _, err := d.ProcessDue(ctx, sender, 10); err != nil {
				t.Errorf("process: %v", err)
			}
		}()
	}
	wg.Wait()

	if sender.sentCount() != 2 {
		t.Fatalf("erwartet genau 2 zustellungen, habe %d: %v", sender.sentCount(), sender.sentIDs)
	}
	seen := map[string]bool{}
	for _, id := range sender.sentIDs {
		if seen[id] {
			t.Fatalf("nachricht %q wurde doppelt zugestellt", id)
		}
		seen[id] = true
	}
	if !seen[idA] || !seen[idB] {
		t.Fatalf("erwartet beide nachrichten zugestellt, habe %v", sender.sentIDs)
	}
}