// Package observability also exposes Prometheus metrics for push-gateway.
//
// The metrics cover the full surface area described in 02 §6.4: active
// connections, messages pushed/dropped, heartbeats, disconnects, Redis Pub/Sub
// latency, Kafka consumption and Redis SET rebuilds. All metrics are
// registered with the global prometheus.DefaultRegisterer on package init.
package observability

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

// Metrics holds all push-gateway Prometheus metric handles.
type Metrics struct {
	ActiveConnections   prometheus.Gauge
	MessagesPushed      *prometheus.CounterVec
	MessagesDropped     *prometheus.CounterVec
	Heartbeats          prometheus.Counter
	Disconnects         *prometheus.CounterVec
	RedisPubSubLatency  *prometheus.HistogramVec
	KafkaConsumed       *prometheus.CounterVec
	RedisSetRebuild     prometheus.Counter
	ConnectionsPerUser  prometheus.Gauge
}

// metrics is the process-wide metric set, initialized once by NewMetrics.
var metrics *Metrics

// NewMetrics registers and returns the push-gateway metric set. Subsequent
// calls return the same instance to avoid duplicate-registration panics.
func NewMetrics() *Metrics {
	if metrics != nil {
		return metrics
	}
	metrics = &Metrics{
		ActiveConnections: promauto.NewGauge(prometheus.GaugeOpts{
			Namespace: "push_gateway",
			Name:      "active_connections",
			Help:      "Current number of live WebSocket connections.",
		}),
		MessagesPushed: promauto.NewCounterVec(prometheus.CounterOpts{
			Namespace: "push_gateway",
			Name:      "messages_pushed_total",
			Help:      "Total messages pushed, partitioned by event and result.",
		}, []string{"event", "result"}),
		MessagesDropped: promauto.NewCounterVec(prometheus.CounterOpts{
			Namespace: "push_gateway",
			Name:      "messages_dropped_total",
			Help:      "Messages dropped, partitioned by reason.",
		}, []string{"reason"}),
		Heartbeats: promauto.NewCounter(prometheus.CounterOpts{
			Namespace: "push_gateway",
			Name:      "heartbeat_total",
			Help:      "Total WebSocket Ping heartbeats received.",
		}),
		Disconnects: promauto.NewCounterVec(prometheus.CounterOpts{
			Namespace: "push_gateway",
			Name:      "disconnect_total",
			Help:      "Disconnects, partitioned by reason.",
		}, []string{"reason"}),
		RedisPubSubLatency: promauto.NewHistogramVec(prometheus.HistogramOpts{
			Namespace: "push_gateway",
			Name:      "redis_pubsub_latency_seconds",
			Help:      "Redis Pub/Sub round-trip latency in seconds.",
			Buckets:   prometheus.ExponentialBuckets(0.001, 2, 12),
		}, []string{"direction"}),
		KafkaConsumed: promauto.NewCounterVec(prometheus.CounterOpts{
			Namespace: "push_gateway",
			Name:      "kafka_consumed_total",
			Help:      "Kafka messages consumed, partitioned by topic and partition.",
		}, []string{"topic", "partition"}),
		RedisSetRebuild: promauto.NewCounter(prometheus.CounterOpts{
			Namespace: "push_gateway",
			Name:      "redis_set_rebuild_total",
			Help:      "Total Redis online-presence SET rebuilds (startup or recovery).",
		}),
		ConnectionsPerUser: promauto.NewGauge(prometheus.GaugeOpts{
			Namespace: "push_gateway",
			Name:      "connections_per_user_max",
			Help:      "Maximum concurrent connections held by a single user.",
		}),
	}
	return metrics
}

// Metrics returns the process-wide metric set. When NewMetrics has not been
// called a new set is registered on demand.
func MetricsInstance() *Metrics {
	if metrics == nil {
		return NewMetrics()
	}
	return metrics
}

// IncMessageDropped is a convenience helper for the channel-full case.
func (m *Metrics) IncMessageDropped(reason string) {
	m.MessagesDropped.WithLabelValues(reason).Inc()
}

// IncPushed is a convenience helper for the push result counter.
func (m *Metrics) IncPushed(event, result string) {
	m.MessagesPushed.WithLabelValues(event, result).Inc()
}

// IncDisconnect is a convenience helper for the disconnect counter.
func (m *Metrics) IncDisconnect(reason string) {
	m.Disconnects.WithLabelValues(reason).Inc()
}

// ObservePubSubLatency records a Redis Pub/Sub round-trip duration.
func (m *Metrics) ObservePubSubLatency(direction string, seconds float64) {
	m.RedisPubSubLatency.WithLabelValues(direction).Observe(seconds)
}

// IncKafkaConsumed increments the Kafka consumed counter.
func (m *Metrics) IncKafkaConsumed(topic, partition string) {
	m.KafkaConsumed.WithLabelValues(topic, partition).Inc()
}

// IncRedisSetRebuild increments the SET rebuild counter (ISSUE-058 metric).
func (m *Metrics) IncRedisSetRebuild() {
	m.RedisSetRebuild.Inc()
}