package metrics

import (
	"strconv"
	"time"

	"github.com/ThreeDotsLabs/watermill/message"
	"github.com/pingcap/log"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/tidbcloud/tiinf/transport/tieventbus/metadata"
	"go.uber.org/zap"
)

var (
	subscriberLabelKeys = []string{
		LabelKeyEventType,
		LabelKeySource,
		LabelKeyGroup,
		LabelKeyRetryTimes,
		LabelKeySuccess,
	}
)

// SubscriberPrometheusMetricsDecorator decorates a subscriber to capture Prometheus metrics.
type SubscriberPrometheusMetricsDecorator struct {
	message.Subscriber
	subscriberName               string
	eventsReceivedTotal          *prometheus.CounterVec
	eventsReceivedLatencySeconds *prometheus.HistogramVec
	closing                      chan struct{}
}

func (s SubscriberPrometheusMetricsDecorator) recordMetrics(msg *message.Message) {
	if msg == nil {
		return
	}
	consumingAt := time.Now()

	ctx := msg.Context()

	go func() {
		if subscribeAlreadyObserved(ctx) {
			// decorator idempotency when applied decorator multiple times
			return
		}
		select {
		case <-msg.Acked():
		case <-msg.Nacked():
		}
		labels := labelsFromCtx(msg.Context(), subscriberLabelKeys...)
		s.eventsReceivedTotal.With(labels).Inc()
		createdAtStr := msg.Metadata.Get(metadata.CreatedAtKey)
		createdAt, err := strconv.ParseInt(createdAtStr, 10, 64)
		if err != nil {
			log.Warn("Failed to log events_received_delay_seconds", zap.Error(err))
			return
		}
		s.eventsReceivedLatencySeconds.With(labels).Observe(time.Duration(consumingAt.UTC().UnixNano() - createdAt).Seconds())
	}()

	msg.SetContext(setSubscribeObservedToCtx(msg.Context()))
}