// Retry queue с TTL - сообщение вернётся через 30 секунд
ch.QueueDeclare("retry.progress", true, false, false, false, amqp.Table{
    "x-dead-letter-exchange":    "events",           // вернуть в основной exchange
    "x-dead-letter-routing-key": "lesson.completed",
    "x-message-ttl":             int32(30000),        // 30 секунд задержка
})

// Финальная DLQ - сюда попадают после исчерпания ретраев
ch.QueueDeclare("dlq.progress.final", true, false, false, false, nil)

<?php
declare(strict_types=1);

use PhpAmqpLib\Channel\AMQPChannel;
use PhpAmqpLib\Wire\AMQPTable;

final class RetryTopology
{
    public function __construct(
        private readonly AMQPChannel $ch,
    ) {}

    public function declare(): void
    {
        // Retry queue с TTL - сообщение вернётся через 30 секунд
        $this->ch->queue_declare(
            queue: 'retry.progress',
            durable: true,
            arguments: new AMQPTable([
                'x-dead-letter-exchange'    => 'events',           // вернуть в основной exchange
                'x-dead-letter-routing-key' => 'lesson.completed',
                'x-message-ttl'             => 30000,              // 30 секунд задержка
            ]),
        );

        // Финальная DLQ - сюда попадают после исчерпания ретраев
        $this->ch->queue_declare('dlq.progress.final', durable: true);
    }
}

const maxRetries = 3

func (c *Consumer) handleSafely(ctx context.Context, msg amqp.Delivery, handler func(context.Context, []byte) error) {
    // Паника обработчика - тот же случай, что ошибка: сообщение
    // отправляем на разбор, а не роняем consumer.
    defer func() {
        if r := recover(); r != nil {
            log.Printf("panic on message %s: %v", msg.MessageId, r)
            c.reject(msg, fmt.Errorf("panic: %v", r))
        }
    }()

    if len(msg.Body) == 0 {
        // Пустое тело не станет валидным от повторов - сразу в финальную DLQ,
        // без круга по retry-очереди.
        log.Printf("empty body, straight to final DLQ")
        c.publishToFinalDLQ(ctx, msg, "empty body")
        msg.Ack(false)
        return
    }

    if err := handler(ctx, msg.Body); err != nil {
        c.reject(msg, err)
        return
    }

    msg.Ack(false)
}

// retryCount ищет запись x-death по очереди и причине. Брать первую нельзя:
// в retry-цикле их несколько - rejected из рабочей очереди и expired
// из retry-очереди.
func retryCount(msg amqp.Delivery, queue, reason string) int32 {
    deaths, ok := msg.Headers["x-death"].([]interface{})
    if !ok {
        return 0
    }
    for _, d := range deaths {
        entry, ok := d.(amqp.Table)
        if !ok {
            continue
        }
        if entry["queue"] == queue && entry["reason"] == reason {
            if count, ok := entry["count"].(int64); ok {
                return int32(count)
            }
        }
    }
    return 0
}

// reject решает, куда отправить сообщение: на очередной круг retry
// или в финальную DLQ. Счётчик берётся из x-death именно той очереди
// и той причины, которые нас интересуют.
func (c *Consumer) reject(msg amqp.Delivery, cause error) {
    attempts := retryCount(msg, c.queue, "rejected")

    if attempts >= maxRetries {
        log.Printf("message %s failed %d times, final DLQ: %v", msg.MessageId, attempts, cause)
        // Публикуем в финальную DLQ сами: так к сообщению можно приложить
        // причину последней ошибки - при dead-lettering её негде взять.
        c.publishToFinalDLQ(context.Background(), msg, cause.Error())
        msg.Ack(false)
        return
    }

    log.Printf("message %s attempt %d failed, retry later: %v", msg.MessageId, attempts+1, cause)
    // Без requeue: сообщение уходит через DLX в retry-очередь,
    // отлёживается там TTL и возвращается обратно. Именно этот проход
    // увеличивает x-death.
    msg.Nack(false, false)
}

<?php
declare(strict_types=1);

use PhpAmqpLib\Message\AMQPMessage;
use Psr\Log\LoggerInterface;

final class PoisonSafeConsumer
{
    private const MAX_RETRIES = 3;

    public function __construct(
        private readonly LoggerInterface $logger,
        private readonly FinalDlqPublisher $finalDlq,
        private readonly string $queue,
    ) {}

    /** @param callable(string): void $handler */
    public function handleSafely(AMQPMessage $msg, callable $handler): void
    {
        if ($msg->getBody() === '') {
            // Повторы не сделают пустое тело валидным.
            $this->logger->warning('empty body, straight to final DLQ');
            $this->finalDlq->publish($msg, 'empty body');
            $msg->ack();
            return;
        }

        try {
            $handler($msg->getBody());
            $msg->ack();
        } catch (Throwable $e) {
            $this->reject($msg, $e);
        }
    }

    private function reject(AMQPMessage $msg, Throwable $cause): void
    {
        $attempts = $this->retryCount($msg, $this->queue, 'rejected');

        if ($attempts >= self::MAX_RETRIES) {
            $this->logger->error('message {id} failed {n} times, final DLQ: {err}', [
                'id'  => $msg->get_properties()['message_id'] ?? '',
                'n'   => $attempts,
                'err' => $cause->getMessage(),
            ]);
            $this->finalDlq->publish($msg, $cause->getMessage());
            $msg->ack();
            return;
        }

        // requeue: false - обязательно. Только так сообщение пройдёт через DLX
        // в retry-очередь, и только так вырастет x-death.
        $msg->nack(requeue: false);
    }

    /** Ищем запись x-death по очереди и причине, а не берём первую. */
    private function retryCount(AMQPMessage $msg, string $queue, string $reason): int
    {
        $headers = $msg->get_properties()['application_headers'] ?? null;
        $deaths = $headers?->getNativeData()['x-death'] ?? [];

        foreach ($deaths as $entry) {
            if (($entry['queue'] ?? null) === $queue && ($entry['reason'] ?? null) === $reason) {
                return (int) ($entry['count'] ?? 0);
            }
        }
        return 0;
    }
}