// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

package internal

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

var topicLabelNames = []string{"pulsar_tenant", "pulsar_namespace", "topic"}

type Metrics struct {
	messagesPublished *prometheus.CounterVec
	bytesPublished    *prometheus.CounterVec
	messagesPending   *prometheus.GaugeVec
	bytesPending      *prometheus.GaugeVec
	publishErrors     *prometheus.CounterVec
	publishLatency    *prometheus.HistogramVec
	publishRPCLatency *prometheus.HistogramVec

	messagesReceived   *prometheus.CounterVec
	bytesReceived      *prometheus.CounterVec
	prefetchedMessages *prometheus.GaugeVec
	prefetchedBytes    *prometheus.GaugeVec
	acksCounter        *prometheus.CounterVec
	nacksCounter       *prometheus.CounterVec
	dlqCounter         *prometheus.CounterVec
	processingTime     *prometheus.HistogramVec

	producersOpened     *prometheus.CounterVec
	producersClosed     *prometheus.CounterVec
	producersPartitions *prometheus.GaugeVec
	consumersOpened     *prometheus.CounterVec
	consumersClosed     *prometheus.CounterVec
	consumersPartitions *prometheus.GaugeVec
	readersOpened       *prometheus.CounterVec
	readersClosed       *prometheus.CounterVec

	// Metrics that are not labeled with topic, are immediately available
	ConnectionsOpened                     prometheus.Counter
	ConnectionsClosed                     prometheus.Counter
	ConnectionsEstablishmentErrors        prometheus.Counter
	ConnectionsHandshakeErrors            prometheus.Counter
	LookupRequestsCount                   prometheus.Counter
	PartitionedTopicMetadataRequestsCount prometheus.Counter
	RPCRequestCount                       prometheus.Counter
}

type TopicMetrics struct {
	MessagesPublished        prometheus.Counter
	BytesPublished           prometheus.Counter
	MessagesPending          prometheus.Gauge
	BytesPending             prometheus.Gauge
	PublishErrorsTimeout     prometheus.Counter
	PublishErrorsMsgTooLarge prometheus.Counter
	PublishLatency           prometheus.Observer
	PublishRPCLatency        prometheus.Observer

	MessagesReceived   prometheus.Counter
	BytesReceived      prometheus.Counter
	PrefetchedMessages prometheus.Gauge
	PrefetchedBytes    prometheus.Gauge
	AcksCounter        prometheus.Counter
	NacksCounter       prometheus.Counter
	DlqCounter         prometheus.Counter
	ProcessingTime     prometheus.Observer

	ProducersOpened     prometheus.Counter
	ProducersClosed     prometheus.Counter
	ProducersPartitions prometheus.Gauge
	ConsumersOpened     prometheus.Counter
	ConsumersClosed     prometheus.Counter
	ConsumersPartitions prometheus.Gauge
	ReadersOpened       prometheus.Counter
	ReadersClosed       prometheus.Counter
}

func NewMetricsProvider(userDefinedLabels map[string]string) *Metrics {
	constLabels := map[string]string{
		"client": "go",
	}
	for k, v := range userDefinedLabels {
		constLabels[k] = v
	}

	metrics := &Metrics{
		messagesPublished: prometheus.NewCounterVec(prometheus.CounterOpts{
			Name:        "pulsar_client_messages_published",
			Help:        "Counter of messages published by the client",
			ConstLabels: constLabels,
		}, topicLabelNames),

		bytesPublished: prometheus.NewCounterVec(prometheus.CounterOpts{
			Name:        "pulsar_client_bytes_published",
			Help:        "Counter of messages published by the client",
			ConstLabels: constLabels,
		}, topicLabelNames),

		messagesPending: prometheus.NewGaugeVec(prometheus.GaugeOpts{
			Name:        "pulsar_client_producer_pending_messages",
			Help:        "Counter of messages pending to be published by the client",
			ConstLabels: constLabels,
		}, topicLabelNames),

		bytesPending: prometheus.NewGaugeVec(prometheus.GaugeOpts{
			Name:        "pulsar_client_producer_pending_bytes",
			Help:        "Counter of bytes pending to be published by the client",
			ConstLabels: constLabels,
		}, topicLabelNames),

		publishErrors: prometheus.NewCounterVec(prometheus.CounterOpts{
			Name:        "pulsar_client_producer_errors",
			Help:        "Counter of publish errors",
			ConstLabels: constLabels,
		}, append(topicLabelNames, "error")),

		publishLatency: prometheus.NewHistogramVec(prometheus.HistogramOpts{
			Name:        "pulsar_client_producer_latency_seconds",
			Help:        "Publish latency experienced by the client",
			ConstLabels: constLabels,
			Buckets:     []float64{.0005, .001, .005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10},
		}, topicLabelNames),

		publishRPCLatency: prometheus.NewHistogramVec(prometheus.HistogramOpts{
			Name:        "pulsar_client_producer_rpc_latency_seconds",
			Help:        "Publish RPC latency experienced internally by the client when sending data to receiving an ack",
			ConstLabels: constLabels,
			Buckets:     []float64{.0005, .001, .005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10},
		}, topicLabelNames),

		producersOpened: prometheus.NewCounterVec(prometheus.CounterOpts{
			Name:        "pulsar_client_producers_opened",
			Help:        "Counter of producers created by the client",
			ConstLabels: constLabels,
		}, topicLabelNames),

		producersClosed: prometheus.NewCounterVec(prometheus.CounterOpts{
			Name:        "pulsar_client_producers_closed",
			Help:        "Counter of producers closed by the client",
			ConstLabels: constLabels,
		}, topicLabelNames),

		producersPartitions: prometheus.NewGaugeVec(prometheus.GaugeOpts{
			Name:        "pulsar_client_producers_partitions_active",
			Help:        "Counter of individual partitions the producers are currently active",
			ConstLabels: constLabels,
		}, topicLabelNames),

		consumersOpened: prometheus.NewCounterVec(prometheus.CounterOpts{
			Name:        "pulsar_client_consumers_opened",
			Help:        "Counter of consumers created by the client",
			ConstLabels: constLabels,
		}, topicLabelNames),

		consumersClosed: prometheus.NewCounterVec(prometheus.CounterOpts{
			Name:        "pulsar_client_consumers_closed",
			Help:        "Counter of consumers closed by the client",
			ConstLabels: constLabels,
		}, topicLabelNames),

		consumersPartitions: prometheus.NewGaugeVec(prometheus.GaugeOpts{
			Name:        "pulsar_client_consumers_partitions_active",
			Help:        "Counter of individual partitions the consumers are currently active",
			ConstLabels: constLabels,
		}, topicLabelNames),

		messagesReceived: prometheus.NewCounterVec(prometheus.CounterOpts{
			Name:        "pulsar_client_messages_received",
			Help:        "Counter of messages received by the client",
			ConstLabels: constLabels,
		}, topicLabelNames),

		bytesReceived: prometheus.NewCounterVec(prometheus.CounterOpts{
			Name:        "pulsar_client_bytes_received",
			Help:        "Counter of bytes received by the client",
			ConstLabels: constLabels,
		}, topicLabelNames),

		prefetchedMessages: prometheus.NewGaugeVec(prometheus.GaugeOpts{
			Name:        "pulsar_client_consumer_prefetched_messages",
			Help:        "Number of messages currently sitting in the consumer pre-fetch queue",
			ConstLabels: constLabels,
		}, topicLabelNames),

		prefetchedBytes: prometheus.NewGaugeVec(prometheus.GaugeOpts{
			Name:        "pulsar_client_consumer_prefetched_bytes",
			Help:        "Total number of bytes currently sitting in the consumer pre-fetch queue",
			ConstLabels: constLabels,
		}, topicLabelNames),

		acksCounter: prometheus.NewCounterVec(prometheus.CounterOpts{
			Name:        "pulsar_client_consumer_acks",
			Help:        "Counter of messages acked by client",
			ConstLabels: constLabels,
		}, topicLabelNames),

		nacksCounter: prometheus.NewCounterVec(prometheus.CounterOpts{
			Name:        "pulsar_client_consumer_nacks",
			Help:        "Counter of messages nacked by client",
			ConstLabels: constLabels,
		}, topicLabelNames),

		dlqCounter: prometheus.NewCounterVec(prometheus.CounterOpts{
			Name:        "pulsar_client_consumer_dlq_messages",
			Help:        "Counter of messages sent to Dead letter queue",
			ConstLabels: constLabels,
		}, topicLabelNames),

		processingTime: prometheus.NewHistogramVec(prometheus.HistogramOpts{
			Name:        "pulsar_client_consumer_processing_time_seconds",
			Help:        "Time it takes for application to process messages",
			Buckets:     []float64{.0005, .001, .005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10},
			ConstLabels: constLabels,
		}, topicLabelNames),

		readersOpened: prometheus.NewCounterVec(prometheus.CounterOpts{
			Name:        "pulsar_client_readers_opened",
			Help:        "Counter of readers created by the client",
			ConstLabels: constLabels,
		}, topicLabelNames),

		readersClosed: prometheus.NewCounterVec(prometheus.CounterOpts{
			Name:        "pulsar_client_readers_closed",
			Help:        "Counter of readers closed by the client",
			ConstLabels: constLabels,
		}, topicLabelNames),

		ConnectionsOpened: prometheus.NewCounter(prometheus.CounterOpts{
			Name:        "pulsar_client_connections_opened",
			Help:        "Counter of connections created by the client",
			ConstLabels: constLabels,
		}),

		ConnectionsClosed: prometheus.NewCounter(prometheus.CounterOpts{
			Name:        "pulsar_client_connections_closed",
			Help:        "Counter of connections closed by the client",
			ConstLabels: constLabels,
		}),

		ConnectionsEstablishmentErrors: prometheus.NewCounter(prometheus.CounterOpts{
			Name:        "pulsar_client_connections_establishment_errors",
			Help:        "Counter of errors in connections establishment",
			ConstLabels: constLabels,
		}),

		ConnectionsHandshakeErrors: prometheus.NewCounter(prometheus.CounterOpts{
			Name:        "pulsar_client_connections_handshake_errors",
			Help:        "Counter of errors in connections handshake (eg: authz)",
			ConstLabels: constLabels,
		}),

		LookupRequestsCount: prometheus.NewCounter(prometheus.CounterOpts{
			Name:        "pulsar_client_lookup_count",
			Help:        "Counter of lookup requests made by the client",
			ConstLabels: constLabels,
		}),

		PartitionedTopicMetadataRequestsCount: prometheus.NewCounter(prometheus.CounterOpts{
			Name:        "pulsar_client_partitioned_topic_metadata_count",
			Help:        "Counter of partitioned_topic_metadata requests made by the client",
			ConstLabels: constLabels,
		}),

		RPCRequestCount: prometheus.NewCounter(prometheus.CounterOpts{
			Name:        "pulsar_client_rpc_count",
			Help:        "Counter of RPC requests made by the client",
			ConstLabels: constLabels,
		}),
	}

	prometheus.DefaultRegisterer.Register(metrics.messagesPublished)
	prometheus.DefaultRegisterer.Register(metrics.bytesPublished)
	prometheus.DefaultRegisterer.Register(metrics.messagesPending)
	prometheus.DefaultRegisterer.Register(metrics.bytesPending)
	prometheus.DefaultRegisterer.Register(metrics.publishErrors)
	prometheus.DefaultRegisterer.Register(metrics.publishLatency)
	prometheus.DefaultRegisterer.Register(metrics.publishRPCLatency)

	prometheus.DefaultRegisterer.Register(metrics.messagesReceived)
	prometheus.DefaultRegisterer.Register(metrics.bytesReceived)
	prometheus.DefaultRegisterer.Register(metrics.prefetchedMessages)
	prometheus.DefaultRegisterer.Register(metrics.prefetchedBytes)
	prometheus.DefaultRegisterer.Register(metrics.acksCounter)
	prometheus.DefaultRegisterer.Register(metrics.nacksCounter)
	prometheus.DefaultRegisterer.Register(metrics.dlqCounter)
	prometheus.DefaultRegisterer.Register(metrics.processingTime)

	prometheus.DefaultRegisterer.Register(metrics.producersOpened)
	prometheus.DefaultRegisterer.Register(metrics.producersClosed)
	prometheus.DefaultRegisterer.Register(metrics.producersPartitions)
	prometheus.DefaultRegisterer.Register(metrics.consumersOpened)
	prometheus.DefaultRegisterer.Register(metrics.consumersClosed)
	prometheus.DefaultRegisterer.Register(metrics.consumersPartitions)
	prometheus.DefaultRegisterer.Register(metrics.readersOpened)
	prometheus.DefaultRegisterer.Register(metrics.readersClosed)

	prometheus.DefaultRegisterer.Register(metrics.ConnectionsOpened)
	prometheus.DefaultRegisterer.Register(metrics.ConnectionsClosed)
	prometheus.DefaultRegisterer.Register(metrics.ConnectionsEstablishmentErrors)
	prometheus.DefaultRegisterer.Register(metrics.ConnectionsHandshakeErrors)
	prometheus.DefaultRegisterer.Register(metrics.LookupRequestsCount)
	prometheus.DefaultRegisterer.Register(metrics.PartitionedTopicMetadataRequestsCount)
	prometheus.DefaultRegisterer.Register(metrics.RPCRequestCount)
	return metrics
}

func (mp *Metrics) GetTopicMetrics(t string) *TopicMetrics {
	tn, _ := ParseTopicName(t)
	topic := TopicNameWithoutPartitionPart(tn)
	labels := map[string]string{
		"pulsar_tenant":    tn.Tenant,
		"pulsar_namespace": tn.Namespace,
		"topic":            topic,
	}

	tm := &TopicMetrics{
		MessagesPublished:        mp.messagesPublished.With(labels),
		BytesPublished:           mp.bytesPublished.With(labels),
		MessagesPending:          mp.messagesPending.With(labels),
		BytesPending:             mp.bytesPending.With(labels),
		PublishErrorsTimeout:     mp.publishErrors.With(mergeMaps(labels, map[string]string{"error": "timeout"})),
		PublishErrorsMsgTooLarge: mp.publishErrors.With(mergeMaps(labels, map[string]string{"error": "msg_too_large"})),
		PublishLatency:           mp.publishLatency.With(labels),
		PublishRPCLatency:        mp.publishRPCLatency.With(labels),

		MessagesReceived:   mp.messagesReceived.With(labels),
		BytesReceived:      mp.bytesReceived.With(labels),
		PrefetchedMessages: mp.prefetchedMessages.With(labels),
		PrefetchedBytes:    mp.prefetchedBytes.With(labels),
		AcksCounter:        mp.acksCounter.With(labels),
		NacksCounter:       mp.nacksCounter.With(labels),
		DlqCounter:         mp.dlqCounter.With(labels),
		ProcessingTime:     mp.processingTime.With(labels),

		ProducersOpened:     mp.producersOpened.With(labels),
		ProducersClosed:     mp.producersClosed.With(labels),
		ProducersPartitions: mp.producersPartitions.With(labels),
		ConsumersOpened:     mp.consumersOpened.With(labels),
		ConsumersClosed:     mp.consumersClosed.With(labels),
		ConsumersPartitions: mp.consumersPartitions.With(labels),
		ReadersOpened:       mp.readersOpened.With(labels),
		ReadersClosed:       mp.readersClosed.With(labels),
	}

	return tm
}

func mergeMaps(a, b map[string]string) map[string]string {
	res := make(map[string]string)
	for k, v := range a {
		res[k] = v
	}
	for k, v := range b {
		res[k] = v
	}

	return res
}