Dead Letter Queue

package consumer

import (
	"context"
	"encoding/base64"
	"encoding/json"
	"time"

	kafkago "github.com/segmentio/kafka-go"
)

type deadLetter struct {
	FailedEventID string `json:"failed_event_id"`
	OriginalTopic string `json:"original_topic"`
	ErrorKind     string `json:"error_kind"`
	ErrorMessage  string `json:"error_message"`
	Attempts      int    `json:"attempts"`
	FailedAt      string `json:"failed_at"`
	RawPayload    string `json:"raw_payload"`
}

type DLQProducer struct {
	writer *kafkago.Writer
}

func NewDLQProducer(brokers []string) *DLQProducer {
	return &DLQProducer{
		writer: &kafkago.Writer{
			Addr:         kafkago.TCP(brokers...),
			Topic:        "payment.completed.dlq",
			RequiredAcks: kafkago.RequireAll,
		},
	}
}

// Send carries the original message base64-encoded rather than re-fetching
// it from anywhere. Whoever inspects the DLQ later — an operator, a replay
// tool — needs the exact bytes that failed, not a best-effort reconstruction
// pulled back out of a database row that may since have changed.
func (p *DLQProducer) Send(ctx context.Context, eventID, originalTopic, errorKind string, err error, attempts int, raw []byte) error {
	dl := deadLetter{
		FailedEventID: eventID,
		OriginalTopic: originalTopic,
		ErrorKind:     errorKind,
		ErrorMessage:  err.Error(),
		Attempts:      attempts,
		FailedAt:      time.Now().UTC().Format(time.RFC3339),
		RawPayload:    base64.StdEncoding.EncodeToString(raw),
	}
	body, marshalErr := json.Marshal(dl)
	if marshalErr != nil {
		return marshalErr
	}
	return p.writer.WriteMessages(ctx, kafkago.Message{Value: body})
}

ALTER TABLE notification_jobs ADD COLUMN dlq_published_at timestamptz;

package main

import (
	"context"
	"fmt"
	"os"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/rs/zerolog/log"

	"github.com/hoa-platform/backend/services/notification-service/internal/consumer"
)

// sweep runs once per invocation; a cron or a simple time.Ticker wrapper
// around it is an operational choice, not shown here. FOR UPDATE SKIP LOCKED
// matches Phase 14's worker and Lesson 2's relay — the same safe pattern
// for letting more than one sweeper instance run without double-publishing.
func sweep(ctx context.Context, pool *pgxpool.Pool, dlq *consumer.DLQProducer) error {
	tx, err := pool.Begin(ctx)
	if err != nil {
		return err
	}
	defer tx.Rollback(ctx)

	rows, err := tx.Query(ctx, `
		SELECT id, organization_id, payload, attempts, last_error
		FROM notification_jobs
		WHERE status = 'dead' AND dlq_published_at IS NULL
		ORDER BY id
		LIMIT 100
		FOR UPDATE SKIP LOCKED`,
	)
	if err != nil {
		return err
	}
	defer rows.Close()

	type deadJob struct {
		id         string
		attempts   int
		lastError  string
		payload    []byte
	}
	var jobs []deadJob
	for rows.Next() {
		var j deadJob
		var orgID string
		if err := rows.Scan(&j.id, &orgID, &j.payload, &j.attempts, &j.lastError); err != nil {
			return err
		}
		jobs = append(jobs, j)
	}
	if err := rows.Err(); err != nil {
		return err
	}

	for _, j := range jobs {
		if err := dlq.Send(ctx, j.id, "notification_jobs", "worker_exhausted",
			fmt.Errorf("%s", j.lastError), j.attempts, j.payload); err != nil {
			return err
		}
		if _, err := tx.Exec(ctx,
			"UPDATE notification_jobs SET dlq_published_at = now() WHERE id = $1", j.id); err != nil {
			return err
		}
	}
	return tx.Commit(ctx)
}

func main() {
	ctx := context.Background()
	pool, err := pgxpool.New(ctx, os.Getenv("NOTIFICATION_SERVICE_DB_URL"))
	if err != nil {
		log.Error().Err(err).Msg("dlq-sweeper: connect to database")
		os.Exit(1)
	}
	defer pool.Close()

	dlq := consumer.NewDLQProducer([]string{"localhost:9092"})
	if err := sweep(ctx, pool, dlq); err != nil {
		log.Error().Err(err).Msg("dlq-sweeper: sweep failed")
		os.Exit(1)
	}
	_ = time.Now
}

package consumer

import "github.com/prometheus/client_golang/prometheus"

type Metrics struct {
	DLQMessagesTotal *prometheus.CounterVec
}

func NewMetrics(reg prometheus.Registerer) *Metrics {
	m := &Metrics{
		DLQMessagesTotal: prometheus.NewCounterVec(prometheus.CounterOpts{
			Name: "notification_dlq_messages_total",
			Help: "Count of events published to a dead-letter topic, by original topic and error kind. Never labeled by organization_id or event_id — both are unbounded.",
		}, []string{"original_topic", "error_kind"}),
	}
	reg.MustRegister(m.DLQMessagesTotal)
	return m
}