package qaas

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"
	qaasgen "go.avito.ru/bill/service-billing-processing/internal/generated/qaas"
	"go.avito.ru/bill/service-billing-processing/pkg/logformatter"
	"go.avito.ru/bill/service-billing-processing/pkg/utils"
)

const (
	DefaultRetryLimit = 5000
	StagingRetryLimit = 50
)

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

type SubscriberStrategy interface {
	// GetConfig возвращает конфиг топика
	GetConfig() SubscriberConfig
	// ProcessMessage реализует бизнес логику обработки сообщения
	ProcessMessage(ctx context.Context, message any, meta qaasgen.Meta) error
	// Subscribe подписывается на топик
	Subscribe(ctx context.Context) error
	// SubscribeDeferred подписывается на топик отложенных сообщений
	SubscribeDeferred(ctx context.Context) error
}

type SubscriberConfig struct {
	// Topic название QaaS топика
	Topic string
	// RetryLimit лимит количества повторных обработок, 0 - без лимита
	RetryLimit int64
}

type SubscriberHandler func(messages []any, metas []qaasgen.Meta) (acks int, deferred bool, err error)

func GetSubscriberHandler(subscriber SubscriberStrategy, metric deps.Metrics) SubscriberHandler {
	return func(messages []any, metas []qaasgen.Meta) (acks int, deferred bool, err error) {
		observer := observability.New()

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

		cfg := subscriber.GetConfig()
		subscriberMetric := getMetrics(cfg.Topic)

		if utils.IsStaging() {
			cfg.RetryLimit = StagingRetryLimit
		}

		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(subscriberMetric.Deferred, len(messages)-acks)

				// "If deferred flag is false, then all messages after acks will be threatened as nacked."
				return acks, false, nil
			}

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

				acks++
				continue
			}

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

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

					acks++
					continue
				}

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

				// механизм лимита повторных обработок
				// при достижении лимита перестаем обрабатывать повторно, просто пропускаем
				if !errors.Is(err, ErrCantConvertMessage) &&
					(cfg.RetryLimit > 0 && meta.DeferCount() > cfg.RetryLimit) {
					logformatter.LogWithError(ctx, fmt.Sprintf("[QaaS %s] retry limit exceeded", cfg.Topic), err)
					metric.Increment(subscriberMetric.Failed)

					acks++
					continue
				}

				if !errors.Is(err, ErrToRetry) && !errors.Is(err, context.Canceled) {
					logProcessMessageError(ctx, err, cfg.Topic)
					metric.Increment(subscriberMetric.DeferredByError)
				}

				metric.Increment(subscriberMetric.Deferred)

				// принудительно вызвали моментальный ретрай
				if errors.Is(err, ErrToInstantRetry) {
					metric.Increment(subscriberMetric.DeferredInstantly)
					return acks, false, nil
				}

				return acks, true, nil
			}

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

		return acks, false, nil
	}
}

func InvalidMessage(ctx context.Context, msg string, args ...any) error {
	slog.WarnContext(ctx, msg, args...)
	return ErrInvalidMessage
}

func logProcessMessageError(ctx context.Context, err error, topic string) {
	// remove ErrToInstantRetry from source error
	if errors.Is(err, ErrToInstantRetry) {
		replaced := strings.Replace(err.Error(), ErrToInstantRetry.Error(), "", 1)
		replaced = strings.ReplaceAll(replaced, "\n", "")
		err = errors.New(replaced)
	}

	logformatter.LogWithError(ctx, fmt.Sprintf("[QaaS %s] process message error", topic), err)
}

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

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

	return SubscriberMetric{
		Success:           prefix + topic + ".success",
		Failed:            prefix + topic + ".failed",
		Deferred:          prefix + topic + ".deferred",
		DeferredByError:   prefix + topic + ".deferred_by_error",
		DeferredInstantly: prefix + topic + ".deferred_instantly",
		BadConvert:        prefix + topic + ".bad_convert",
		InvalidMessage:    prefix + topic + ".invalid_message",
		EmptyMessage:      prefix + topic + ".empty_message",
	}
}


объясни че здесь написано