What:

На данный момент, если мы попадаем в секцию empty messages, то сообщение просто акается и пишется в логи. Но логи тротлятся и в случае потерь не все можно восстановить из них, остальные пути восстановления слишком дорогие.

Секция с empty messages есть в универсальном хэндлере как QaaS так и Databus, туда, как правило, попадают сообщения, у которых payload не смог распарситься на структуру сообщения из brief.

Acceptance criteria:

Выработано решение
Согласовано с командой
Заведена задача/задачи

вот код databus
package dbus

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"strings"

	"go.avito.ru/gl/observability/v2"

	"go.avito.ru/bill/service-billing-processing/cmd"
	"go.avito.ru/bill/service-billing-processing/internal/deps"
	dbusgen "go.avito.ru/bill/service-billing-processing/internal/generated/databus"
	"go.avito.ru/bill/service-billing-processing/pkg/logformatter"
	"go.avito.ru/bill/service-billing-processing/pkg/utils"
)

var (
	// ErrToRetry ошибка для механизма повторной обработки сообщений
	ErrToRetry = errors.New("retrying message")
	// ErrCantConvertMessage не получилось конвертировать сообщение в нужный тип топика
	ErrCantConvertMessage = errors.New("can't convert message")
	// ErrInvalidMessage сообщение не прошло валидацию
	ErrInvalidMessage = errors.New("invalid message")
)

type ConsumerStrategy interface {
	// GetConfig возвращает конфиг топика
	GetConfig() ConsumerConfig
	// ProcessMessage реализует бизнес логику обработки сообщения
	ProcessMessage(ctx context.Context, message any) error
	// Consume подписывается на топик
	Consume(ctx context.Context) error
}

type ConsumerConfig struct {
	// Topic название databus топика
	Topic string
}

type ConsumerHandler func(messages []any, metas []dbusgen.Meta) (acks int, fatalErr error)

func GetConsumerHandler(consumer ConsumerStrategy, metric deps.Metrics) ConsumerHandler {
	return func(messages []any, metas []dbusgen.Meta) (acks int, fatalErr error) {
		observer := observability.New()

		if len(messages) == 0 {
			return 0, nil
		}

		cfg := consumer.GetConfig()
		consumerMetric := getMetrics(cfg.Topic)

		for i, message := range messages {
			meta := metas[i]
			ctx := meta.Context()
			ctx = logformatter.WithField(ctx, "context.requestId", observer.RequestID(ctx))

			// сервис выключается и не сможет обработать эти сообщения - даем nack
			// новые поды подхватят сообщения и обработают их
			if cmd.IsShuttingDown {
				metric.Count(consumerMetric.Deferred, len(messages)-acks)
				return acks, nil
			}

			if utils.IsNil(message) {
				ctx = logformatter.WithField(ctx, "payload.meta", string(meta.Payload()))
				slog.WarnContext(ctx, fmt.Sprintf("[Databus %s] empty message", cfg.Topic))
				metric.Increment(consumerMetric.EmptyMessage)

				acks++
				continue
			}

			ctx = logformatter.LogConsumedMessage(ctx, cfg.Topic, message)

			err := consumer.ProcessMessage(ctx, message)
			if err != nil {
				// ошибка валидации полей
				// нет смысла повторно обрабатывать -> пропускаем сообщение
				if errors.Is(err, ErrInvalidMessage) {
					metric.Increment(consumerMetric.InvalidMessage)

					acks++
					continue
				}

				// не получилось конвертировать в структуру сообщения
				// скорее всего проблема на нашей стороне, поэтому не игнорируем
				if errors.Is(err, ErrCantConvertMessage) {
					metric.Increment(consumerMetric.BadConvert)
				}

				if !errors.Is(err, ErrToRetry) && !errors.Is(err, context.Canceled) {
					logformatter.LogWithError(ctx, fmt.Sprintf("[Databus %s] process message error", cfg.Topic), err)
					metric.Increment(consumerMetric.DeferredByError)
				}

				metric.Increment(consumerMetric.Deferred)

				// Не возвращаем ошибку, так как не хотим останавливать consumer:
				// "Usually you do not want to return fatalErr even in case of your handler errors: just return zero acks.
				// Return fatalErr as non-nil ONLY if you want to stop consuming."
				return acks, nil
			}

			metric.Increment(consumerMetric.Success)
			acks++
		}

		return acks, nil
	}
}

func InvalidMessageWithError(ctx context.Context, cfg ConsumerConfig, err error) error {
	slog.WarnContext(
		ctx,
		fmt.Sprintf("validation error on %s", cfg.Topic),
		slog.String("error", err.Error()),
	)

	return errors.Join(
		ErrInvalidMessage,
		err,
	)
}

type ConsumerMetric struct {
	// Success метрика кол-ва успешно обработанных сообщении
	Success string
	// Deferred метрика кол-ва сообщений ушедших в повторную обработку
	Deferred string
	// DeferredByError метрика кол-ва сообщений ушедших в повторную обработку из-за ошибки
	DeferredByError string
	// BadConvert метрика кол-ва сообщений с неправильным типом сообщения
	BadConvert string
	// InvalidMessage метрика кол-ва сообщений с ошибками валидации
	InvalidMessage string
	// EmptyMessage метрика кол-ва пустых сообщений (nil)
	EmptyMessage string
}

func getMetrics(topic string) ConsumerMetric {
	const prefix = "dbus."
	topic = strings.ReplaceAll(topic, ".", "_")

	return ConsumerMetric{
		Success:         prefix + topic + ".success",
		Deferred:        prefix + topic + ".deferred",
		DeferredByError: prefix + topic + ".deferred_by_error",
		BadConvert:      prefix + topic + ".bad_convert",
		InvalidMessage:  prefix + topic + ".invalid_message",
		EmptyMessage:    prefix + topic + ".empty_message",
	}
}

объясни мне проблему и накидай решения