6. Handling poison messages

// Паника при отсутствии поля в теле сообщения
orderID := payload.Customer.Profile.ID 

package consumer

import (
    "context"
    "fmt"
    "log/slog"
    "runtime/debug"
)

type Worker struct {
    logger *slog.Logger
}

// Безопасная оболочка (Panic Boundary) для вызова бизнес-логики
func (w *Worker) safeHandle(ctx context.Context, msg []byte) (err error) {
    // Спасаем процесс операционной системы от крушения
    defer func() {
        if r := recover(); r != nil {
            // Извлекаем полный Stack Trace для последующего анализа в Sentry
            stackTrace := string(debug.Stack())
            w.logger.Error("КРИТИЧЕСКАЯ ПАНИКА в обработчике сообщения!",
                "panic", r,
                "stack", stackTrace,
            )

            // Преобразуем неперехваченную панику в типизированную ошибку яда
            err = fmt.Errorf("%w: panic recovered: %v", ErrFatalPoisonPill, r)
        }
    }()

    // Выполнение реальной доменной логики
    return w.businessLogic(ctx, msg)
}

package consumer

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

// Sentinel-ошибки классификации сбоев
var (
    // ErrFatalPoisonPill — сообщение структурно или логически ядовито (ретрай бессмысленен)
    ErrFatalPoisonPill = errors.New("fatal poison message")

    // ErrTransientRetry — временный инфраструктурный сбой (нужно повторить с backoff)
    ErrTransientRetry = errors.New("transient error, retry needed")
)

type OrderPayload struct {
    ID     string  `json:"id"`
    Amount float64 `json:"amount"`
}

func (p *OrderPayload) Validate() error {
    if p.ID == "" || p.Amount <= 0 {
        return errors.New("order_id is empty or amount is non-positive")
    }
    return nil
}

func (w *Worker) businessLogic(ctx context.Context, data []byte) error {
    var payload OrderPayload

    // 1. Синтаксическая ошибка JSON — это 100% неисправимый яд
    if err := json.Unmarshal(data, &payload); err != nil {
        return fmt.Errorf("%w: syntax error in json payload: %v", ErrFatalPoisonPill, err)
    }

    // 2. Бизнес-валидация инвариантов
    if err := payload.Validate(); err != nil {
        return fmt.Errorf("%w: domain validation rejected: %v", ErrFatalPoisonPill, err)
    }

    // 3. Взаимодействие с внешними зависимостями (СУБД)
    err := w.saveOrder(ctx, payload)
    if err != nil {
        // Сетевой таймаут или deadlock в БД — это временная проблема!
        return fmt.Errorf("%w: database transaction failed: %v", ErrTransientRetry, err)
    }

    return nil
}

func (w *Worker) saveOrder(ctx context.Context, p OrderPayload) error {
    // Имитация записи в БД
    return nil
}

package consumer

import (
    "context"
    "errors"

    amqp "github.com/rabbitmq/amqp091-go"
)

func (w *Worker) ConsumeRabbit(deliveries <-chan amqp.Delivery) {
    for d := range deliveries {
        err := w.safeHandle(context.Background(), d.Body)

        if err == nil {
            // Успешная обработка: подтверждаем удаление
            _ = d.Ack(false)
            continue
        }

        if errors.Is(err, ErrFatalPoisonPill) {
            // ФАТАЛЬНЫЙ ЯД (Poison Message):
            // requeue=false приказывает RabbitMQ немедленно удалить сообщение из очереди.
            // Поскольку настроен DLX, брокер автоматически перенаправит яд в Dead Letter Queue!
            w.logger.Warn("Ядовитое сообщение изолировано в DLQ", "err", err)
            _ = d.Nack(false, false)
        } else {
            // ВРЕМЕННАЯ ОШИБКА (Transient Error):
            // requeue=true возвращает сообщение в очередь для повторной попытки.
            // В продакшене здесь применяется пауза (Backoff) или маршрутизация в Retry-очередь.
            w.logger.Info("Временный сбой, возвращаем сообщение в очередь", "err", err)
            _ = d.Nack(false, true)
        }
    }
}

package consumer

import (
    "context"
    "errors"
    "time"
)

type KafkaMessage struct {
    Key   []byte
    Value []byte
    Topic string
}

type KafkaDLQWorker struct {
    producer KafkaSyncProducer
    consumer KafkaConsumerClient
}

func (w *KafkaDLQWorker) ProcessKafkaMessage(ctx context.Context, msg KafkaMessage) {
    err := w.safeHandle(ctx, msg.Value)

    if err == nil {
        // Успешно: фиксируем смещение в основном топике
        w.consumer.Commit(msg)
        return
    }

    if errors.Is(err, ErrFatalPoisonPill) {
        // Блокирующий цикл гарантированной доставки яда в DLQ
        for {
            dlqErr := w.producer.ProduceSync(ctx, "dlq-topic", msg.Key, msg.Value, err.Error())
            if dlqErr == nil {
                // Яд успешно изолирован в DLQ!
                // ТЕПЕРЬ мы имеем полное право сдвинуть оффсет в основной очереди
                w.consumer.Commit(msg)
                return
            }

            // Если записать в DLQ не удалось, ждем и повторяем: данные терять нельзя
            time.Sleep(2 * time.Second)
        }
    } else {
        // Обработка временных сбоев (например, отправка в retry-топик)
        w.handleTransient(ctx, msg)
    }
}

func (w *KafkaDLQWorker) safeHandle(ctx context.Context, val []byte) error {
    return nil
}

func (w *KafkaDLQWorker) handleTransient(ctx context.Context, msg KafkaMessage) {}