package worker

import "github.com/prometheus/client_golang/prometheus"

type Metrics struct {
	MessagesReadTotal                *prometheus.CounterVec
	MessagesSentTotal                *prometheus.CounterVec
	MessagesFailedToBePublishedTotal *prometheus.CounterVec
	KafkaSuccessfulAcksTotal         *prometheus.CounterVec
	KafkaFailedAcksTotal             *prometheus.CounterVec
	KafkaLastAckedMessageGauge       *prometheus.GaugeVec
}

func NewMetrics() *Metrics {
	m := &Metrics{
		MessagesReadTotal: prometheus.NewCounterVec(
			prometheus.CounterOpts{
				Name: "kafka_connect_pubsub__messages_read_total",
				Help: "total number of messages the connector has read from kafka topic",
			}, []string{"topic"}),
		MessagesSentTotal: prometheus.NewCounterVec(
			prometheus.CounterOpts{
				Name: "kafka_connect_pubsub__messages_sent_total",
				Help: "total number of messages that the connector sent to pub/sub",
			}, []string{"topic"}),
		MessagesFailedToBePublishedTotal: prometheus.NewCounterVec(
			prometheus.CounterOpts{
				Name: "kafka_connect_pubsub__messages_failed_to_be_published_total",
				Help: "total number of messages that failed to be published to pub/sub",
			}, []string{"topic"}),
		KafkaSuccessfulAcksTotal: prometheus.NewCounterVec(
			prometheus.CounterOpts{
				Name: "kafka_connect_pubsub__successful_kafka_acks_total",
				Help: "total number of successful kafka acks",
			}, []string{"topic"}),
		KafkaFailedAcksTotal: prometheus.NewCounterVec(
			prometheus.CounterOpts{
				Name: "kafka_connect_pubsub__failed_kafka_acks_total",
				Help: "total number of failed kafka acks",
			}, []string{"topic"}),
		KafkaLastAckedMessageGauge: prometheus.NewGaugeVec(
			prometheus.GaugeOpts{
				Name: "kafka_connect_pubsub__last_acked_kafka_message",
				Help: "the last kafka acked message",
			}, []string{"topic", "partition"}),
	}

	prometheus.MustRegister(
		m.MessagesReadTotal,
		m.MessagesSentTotal,
		m.MessagesFailedToBePublishedTotal,
		m.KafkaSuccessfulAcksTotal,
		m.KafkaFailedAcksTotal,
		m.KafkaLastAckedMessageGauge,
	)

	return m
}

func (m *Metrics) Reset() {
	m.MessagesReadTotal.Reset()
	m.MessagesSentTotal.Reset()
	m.KafkaSuccessfulAcksTotal.Reset()
	m.KafkaFailedAcksTotal.Reset()
	m.MessagesFailedToBePublishedTotal.Reset()
}