nexarch/internal/notify/worker.go

package notify

import (
	"context"
	"encoding/json"
	"fmt"
	"time"
)

// ProcessDue holt bis zu limit faellige Benachrichtigungen und versucht sie
// ueber sender zuzustellen. FOR UPDATE SKIP LOCKED serialisiert konkurrierende
// Aufrufe (Akzeptanzkriterium 3 / Pruefung 3: zwei gleichzeitig ausloesende
// Module duerfen sich nicht gegenseitig blockieren oder Nachrichten doppelt
// zustellen) — dieselbe Konvention wie internal/tenant.Lifecycle.ProcessDueDeletions.
func (d *Dispatcher) ProcessDue(ctx context.Context, sender Sender, limit int) (sent, failed int, err error) {
	tx, err := d.pool.Begin(ctx)
	if err != nil {
		return 0, 0, fmt.Errorf("transaktion starten: %w", err)
	}
	defer func() { _ = tx.Rollback(ctx) }()

	rows, err := tx.Query(ctx, `
		SELECT id, channel, recipient, payload, attempts, max_attempts
		FROM notification_jobs
		WHERE status = 'pending' AND next_attempt_at <= now()
		ORDER BY created_at
		FOR UPDATE SKIP LOCKED
		LIMIT $1
	`, limit)
	if err != nil {
		return 0, 0, fmt.Errorf("faellige benachrichtigungen abfragen: %w", err)
	}

	type due struct {
		id, channel, recipient string
		payload                []byte
		attempts, maxAttempts  int
	}
	var candidates []due
	for rows.Next() {
		var c due
		if err := rows.Scan(&c.id, &c.channel, &c.recipient, &c.payload, &c.attempts, &c.maxAttempts); err != nil {
			rows.Close()
			return 0, 0, fmt.Errorf("faellige benachrichtigung lesen: %w", err)
		}
		candidates = append(candidates, c)
	}
	rows.Close()
	if err := rows.Err(); err != nil {
		return 0, 0, err
	}

	for _, c := range candidates {
		var payload map[string]any
		if err := json.Unmarshal(c.payload, &payload); err != nil {
			payload = map[string]any{}
		}

		sendErr := sender.Send(ctx, Notification{
			ID: c.id, Channel: c.channel, Recipient: c.recipient, Payload: payload, Attempts: c.attempts,
		})

		if sendErr == nil {
			if _, err := tx.Exec(ctx, `
				UPDATE notification_jobs SET status = 'sent', updated_at = now() WHERE id = $1
			`, c.id); err != nil {
				return sent, failed, fmt.Errorf("erfolg speichern: %w", err)
			}
			sent++
			continue
		}

		newAttempts := c.attempts + 1
		if newAttempts >= c.maxAttempts {
			// Akzeptanzkriterium 2: kontrollierter Abbruch nach definierter
			// Anzahl Versuche, kein endloses Wiederholen.
			if _, err := tx.Exec(ctx, `
				UPDATE notification_jobs
				SET status = 'failed', attempts = $2, last_error = $3, updated_at = now()
				WHERE id = $1
			`, c.id, newAttempts, sendErr.Error()); err != nil {
				return sent, failed, fmt.Errorf("fehlschlag speichern: %w", err)
			}
			failed++
			continue
		}

		nextAttempt := time.Now().Add(time.Duration(newAttempts) * d.retryBackoff)
		if _, err := tx.Exec(ctx, `
			UPDATE notification_jobs
			SET attempts = $2, next_attempt_at = $3, last_error = $4, updated_at = now()
			WHERE id = $1
		`, c.id, newAttempts, nextAttempt, sendErr.Error()); err != nil {
			return sent, failed, fmt.Errorf("wiederholung planen: %w", err)
		}
	}

	if err := tx.Commit(ctx); err != nil {
		return 0, 0, fmt.Errorf("transaktion committen: %w", err)
	}
	return sent, failed, nil
}