package metrics

import (
	"context"
	"net/http"
	"sync/atomic"
	"time"

	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promhttp"
)

// Settings configure metrics and slow-request logging behavior.
type Settings struct {
	Enabled              bool
	SlowRequestThreshold time.Duration
}

type metricsState struct {
	enabled       bool
	slowThreshold time.Duration
	registry      *prometheus.Registry
	handler       http.Handler
	stageHist     *prometheus.HistogramVec
	reqTotal      *prometheus.CounterVec
	reqDone       *prometheus.CounterVec
	reqLatency    *prometheus.HistogramVec
	reqInFlight   *prometheus.GaugeVec
	cqlQueryTotal *prometheus.CounterVec
	cqlQueryHist  *prometheus.HistogramVec
	cqlWriteRows  *prometheus.CounterVec
	cqlDisconnect *prometheus.CounterVec
	mysqlOps      *prometheus.CounterVec
	mysqlLatency  *prometheus.HistogramVec
	mysqlErrors   *prometheus.CounterVec
	writeRetries  *prometheus.CounterVec
	indexSelect   *prometheus.CounterVec
}

var globalState atomic.Value // *metricsState

var stageDurationBuckets = mustDurationBuckets(
	50*time.Microsecond,
	100*time.Microsecond,
	150*time.Microsecond,
	200*time.Microsecond,
	250*time.Microsecond,
	300*time.Microsecond,
	400*time.Microsecond,
	500*time.Microsecond,
	600*time.Microsecond,
	750*time.Microsecond,
	900*time.Microsecond,
	time.Millisecond,
	1250*time.Microsecond,
	1500*time.Microsecond,
	1750*time.Microsecond,
	2*time.Millisecond,
	2250*time.Microsecond,
	2500*time.Microsecond,
	2750*time.Microsecond,
	3*time.Millisecond,
	3250*time.Microsecond,
	3500*time.Microsecond,
	3750*time.Microsecond,
	4*time.Millisecond,
	4250*time.Microsecond,
	4500*time.Microsecond,
	4750*time.Microsecond,
	5*time.Millisecond,
	6*time.Millisecond,
	7*time.Millisecond,
	8*time.Millisecond,
	9*time.Millisecond,
	10*time.Millisecond,
	12*time.Millisecond,
	15*time.Millisecond,
	20*time.Millisecond,
	25*time.Millisecond,
	30*time.Millisecond,
	40*time.Millisecond,
	60*time.Millisecond,
	90*time.Millisecond,
	125*time.Millisecond,
	175*time.Millisecond,
	250*time.Millisecond,
	350*time.Millisecond,
	500*time.Millisecond,
	750*time.Millisecond,
	time.Second,
	1500*time.Millisecond,
	2500*time.Millisecond,
	4*time.Second,
	6*time.Second,
	10*time.Second,
	20*time.Second,
	40*time.Second,
	80*time.Second,
	160*time.Second,
	320*time.Second,
)

var requestDurationBuckets = mustDurationBuckets(
	250*time.Microsecond,
	500*time.Microsecond,
	750*time.Microsecond,
	time.Millisecond,
	1250*time.Microsecond,
	1500*time.Microsecond,
	1750*time.Microsecond,
	2*time.Millisecond,
	2250*time.Microsecond,
	2500*time.Microsecond,
	2750*time.Microsecond,
	3*time.Millisecond,
	3250*time.Microsecond,
	3500*time.Microsecond,
	3750*time.Microsecond,
	4*time.Millisecond,
	4250*time.Microsecond,
	4500*time.Microsecond,
	4750*time.Microsecond,
	5*time.Millisecond,
	6*time.Millisecond,
	7*time.Millisecond,
	8*time.Millisecond,
	9*time.Millisecond,
	10*time.Millisecond,
	12*time.Millisecond,
	15*time.Millisecond,
	20*time.Millisecond,
	25*time.Millisecond,
	30*time.Millisecond,
	40*time.Millisecond,
	50*time.Millisecond,
	65*time.Millisecond,
	80*time.Millisecond,
	100*time.Millisecond,
	125*time.Millisecond,
	150*time.Millisecond,
	200*time.Millisecond,
	300*time.Millisecond,
	500*time.Millisecond,
	750*time.Millisecond,
	time.Second,
	1500*time.Millisecond,
	2*time.Second,
	3*time.Second,
	5*time.Second,
	8*time.Second,
	12*time.Second,
	20*time.Second,
	40*time.Second,
	80*time.Second,
	160*time.Second,
	320*time.Second,
	640*time.Second,
)

var cqlQueryDurationBuckets = mustDurationBuckets(
	250*time.Microsecond,
	500*time.Microsecond,
	750*time.Microsecond,
	time.Millisecond,
	1250*time.Microsecond,
	1500*time.Microsecond,
	1750*time.Microsecond,
	2*time.Millisecond,
	2250*time.Microsecond,
	2500*time.Microsecond,
	2750*time.Microsecond,
	3*time.Millisecond,
	3250*time.Microsecond,
	3500*time.Microsecond,
	3750*time.Microsecond,
	4*time.Millisecond,
	4250*time.Microsecond,
	4500*time.Microsecond,
	4750*time.Microsecond,
	5*time.Millisecond,
	6*time.Millisecond,
	7*time.Millisecond,
	8*time.Millisecond,
	9*time.Millisecond,
	10*time.Millisecond,
	12*time.Millisecond,
	15*time.Millisecond,
	20*time.Millisecond,
	25*time.Millisecond,
	30*time.Millisecond,
	40*time.Millisecond,
	50*time.Millisecond,
	65*time.Millisecond,
	80*time.Millisecond,
	100*time.Millisecond,
	125*time.Millisecond,
	150*time.Millisecond,
	200*time.Millisecond,
	300*time.Millisecond,
	400*time.Millisecond,
	600*time.Millisecond,
	800*time.Millisecond,
	time.Second,
	1500*time.Millisecond,
	2*time.Second,
	3*time.Second,
	5*time.Second,
	8*time.Second,
	12*time.Second,
	20*time.Second,
	40*time.Second,
	80*time.Second,
	160*time.Second,
	320*time.Second,
	640*time.Second,
)

var mysqlOperationDurationBuckets = mustDurationBuckets(
	100*time.Microsecond,
	250*time.Microsecond,
	500*time.Microsecond,
	750*time.Microsecond,
	time.Millisecond,
	1250*time.Microsecond,
	1500*time.Microsecond,
	1750*time.Microsecond,
	2*time.Millisecond,
	2250*time.Microsecond,
	2500*time.Microsecond,
	2750*time.Microsecond,
	3*time.Millisecond,
	3250*time.Microsecond,
	3500*time.Microsecond,
	3750*time.Microsecond,
	4*time.Millisecond,
	4250*time.Microsecond,
	4500*time.Microsecond,
	4750*time.Microsecond,
	5*time.Millisecond,
	6*time.Millisecond,
	7*time.Millisecond,
	8*time.Millisecond,
	9*time.Millisecond,
	10*time.Millisecond,
	12*time.Millisecond,
	15*time.Millisecond,
	20*time.Millisecond,
	25*time.Millisecond,
	30*time.Millisecond,
	40*time.Millisecond,
	50*time.Millisecond,
	65*time.Millisecond,
	80*time.Millisecond,
	100*time.Millisecond,
	125*time.Millisecond,
	150*time.Millisecond,
	200*time.Millisecond,
	300*time.Millisecond,
	500*time.Millisecond,
	750*time.Millisecond,
	time.Second,
	1500*time.Millisecond,
	2*time.Second,
	3*time.Second,
	5*time.Second,
	8*time.Second,
	12*time.Second,
	20*time.Second,
	30*time.Second,
	60*time.Second,
	120*time.Second,
	240*time.Second,
	480*time.Second,
	960*time.Second,
)

var opcodeLabels = func() [256]string {
	var out [256]string
	out[0x00] = "ERROR"
	out[0x01] = "STARTUP"
	out[0x02] = "READY"
	out[0x03] = "AUTHENTICATE"
	out[0x05] = "OPTIONS"
	out[0x06] = "SUPPORTED"
	out[0x07] = "QUERY"
	out[0x08] = "RESULT"
	out[0x09] = "PREPARE"
	out[0x0A] = "EXECUTE"
	out[0x0B] = "REGISTER"
	out[0x0C] = "EVENT"
	out[0x0D] = "BATCH"
	out[0x0E] = "AUTH_CHALLENGE"
	out[0x0F] = "AUTH_RESPONSE"
	out[0x10] = "AUTH_SUCCESS"
	return out
}()

func opcodeLabel(opcode byte) (string, bool) {
	label := opcodeLabels[int(opcode)]
	return label, label != ""
}

func requestCompletionLabels(opcode byte, rm *RequestMetrics) (opcodeName, opType string, ok bool) {
	opcodeName, ok = opcodeLabel(opcode)
	if !ok {
		return "", "", false
	}
	opType = opcodeName
	if rm != nil {
		if current := normalizeLabel(rm.OperationType(), ""); current != "" && current != "unknown" {
			opType = current
		}
	}
	return opcodeName, opType, true
}

// Init initializes global metrics state. It is safe to call multiple times; the last call wins.
func Init(settings Settings, poolProvider PoolStatsProvider) {
	SetPoolStatsProvider(poolProvider)
	SetPreparedStmtCacheStatsProvider(nil)
	SetConnectionStatsProvider(nil)

	state := &metricsState{
		enabled:       settings.Enabled,
		slowThreshold: settings.SlowRequestThreshold,
	}
	if settings.Enabled {
		reg := prometheus.NewRegistry()
		stageHist := prometheus.NewHistogramVec(
			prometheus.HistogramOpts{
				Namespace: "proxy",
				Subsystem: "request",
				Name:      "stage_seconds",
				Help:      "Latency per request stage in seconds.",
				Buckets:   stageDurationBuckets,
			},
			[]string{"stage", "KeyspaceName", "ClusterId"},
		)
		reg.MustRegister(stageHist)
		state.stageHist = stageHist

		reqTotal := prometheus.NewCounterVec(
			prometheus.CounterOpts{
				Namespace: "proxy",
				Subsystem: "request",
				Name:      "total",
				Help:      "Total requests by opcode.",
			},
			[]string{"opcode", "KeyspaceName", "ClusterId"},
		)
		reg.MustRegister(reqTotal)
		state.reqTotal = reqTotal

		reqDone := prometheus.NewCounterVec(
			prometheus.CounterOpts{
				Namespace: "proxy",
				Subsystem: "request",
				Name:      "completed_total",
				Help:      "Completed requests by opcode, operation type, and status.",
			},
			[]string{"opcode", "op_type", "status", "KeyspaceName", "ClusterId"},
		)
		reg.MustRegister(reqDone)
		state.reqDone = reqDone

		reqLatency := prometheus.NewHistogramVec(
			prometheus.HistogramOpts{
				Namespace: "proxy",
				Subsystem: "request",
				Name:      "duration_seconds",
				Help:      "End-to-end request latency by opcode, operation type, and status.",
				Buckets:   requestDurationBuckets,
			},
			[]string{"opcode", "op_type", "status", "KeyspaceName", "ClusterId"},
		)
		reg.MustRegister(reqLatency)
		state.reqLatency = reqLatency

		reqInFlight := prometheus.NewGaugeVec(
			prometheus.GaugeOpts{
				Namespace: "proxy",
				Subsystem: "request",
				Name:      "in_flight",
				Help:      "Current number of in-flight requests.",
			},
			[]string{"KeyspaceName", "ClusterId"},
		)
		reg.MustRegister(reqInFlight)
		state.reqInFlight = reqInFlight

		cqlQueryTotal := prometheus.NewCounterVec(
			prometheus.CounterOpts{
				Namespace: "proxy",
				Subsystem: "cql",
				Name:      "query_total",
				Help:      "Total CQL queries by query type and status.",
			},
			[]string{"query_type", "status", "KeyspaceName", "ClusterId"},
		)
		reg.MustRegister(cqlQueryTotal)
		state.cqlQueryTotal = cqlQueryTotal

		cqlQueryHist := prometheus.NewHistogramVec(
			prometheus.HistogramOpts{
				Namespace: "proxy",
				Subsystem: "cql",
				Name:      "query_duration_seconds",
				Help:      "End-to-end CQL query latency by query type and status.",
				Buckets:   cqlQueryDurationBuckets,
			},
			[]string{"query_type", "status", "KeyspaceName", "ClusterId"},
		)
		reg.MustRegister(cqlQueryHist)
		state.cqlQueryHist = cqlQueryHist

		cqlWriteRows := prometheus.NewCounterVec(
			prometheus.CounterOpts{
				Namespace: "proxy",
				Subsystem: "cql",
				Name:      "write_rows_total",
				Help:      "Total requested logical rows written by CQL operation type.",
			},
			[]string{"operation_type", "KeyspaceName", "ClusterId"},
		)
		reg.MustRegister(cqlWriteRows)
		state.cqlWriteRows = cqlWriteRows

		cqlDisconnect := prometheus.NewCounterVec(
			prometheus.CounterOpts{
				Namespace: "proxy",
				Subsystem: "cql",
				Name:      "disconnection_total",
				Help:      "Total CQL client disconnections by scope.",
			},
			[]string{"scope", "KeyspaceName", "ClusterId"},
		)
		reg.MustRegister(cqlDisconnect)
		state.cqlDisconnect = cqlDisconnect

		mysqlOps := prometheus.NewCounterVec(
			prometheus.CounterOpts{
				Namespace: "proxy",
				Subsystem: "mysql",
				Name:      "operation_total",
				Help:      "Total MySQL operations by operation and status.",
			},
			[]string{"operation", "status", "KeyspaceName", "ClusterId"},
		)
		reg.MustRegister(mysqlOps)
		state.mysqlOps = mysqlOps

		mysqlLatency := prometheus.NewHistogramVec(
			prometheus.HistogramOpts{
				Namespace: "proxy",
				Subsystem: "mysql",
				Name:      "operation_seconds",
				Help:      "MySQL operation latency by operation and status.",
				Buckets:   mysqlOperationDurationBuckets,
			},
			[]string{"operation", "status", "KeyspaceName", "ClusterId"},
		)
		reg.MustRegister(mysqlLatency)
		state.mysqlLatency = mysqlLatency

		mysqlErrors := prometheus.NewCounterVec(
			prometheus.CounterOpts{
				Namespace: "proxy",
				Subsystem: "mysql",
				Name:      "error_total",
				Help:      "Total MySQL errors by operation and error type.",
			},
			[]string{"operation", "error_type", "KeyspaceName", "ClusterId"},
		)
		reg.MustRegister(mysqlErrors)
		state.mysqlErrors = mysqlErrors

		writeRetries := prometheus.NewCounterVec(
			prometheus.CounterOpts{
				Namespace: "proxy",
				Subsystem: "mysql",
				Name:      "write_retry_total",
				Help:      "Total backend write retries by retry kind and outcome.",
			},
			[]string{"kind", "outcome", "KeyspaceName", "ClusterId"},
		)
		reg.MustRegister(writeRetries)
		state.writeRetries = writeRetries

		indexSelect := prometheus.NewCounterVec(
			prometheus.CounterOpts{
				Namespace: "proxy",
				Subsystem: "index",
				Name:      "selection_total",
				Help:      "Index selection outcomes (sampled).",
			},
			[]string{"outcome", "rule", "index_type", "covering", "order_by", "back_table", "KeyspaceName", "ClusterId"},
		)
		reg.MustRegister(indexSelect)
		state.indexSelect = indexSelect

		reg.MustRegister(newPoolStatsCollector())
		reg.MustRegister(newPreparedStmtCacheStatsCollector())
		reg.MustRegister(newConnectionStatsCollector())

		state.registry = reg
		state.handler = promhttp.HandlerFor(reg, promhttp.HandlerOpts{})
	}

	globalState.Store(state)
}

// Enabled reports whether Prometheus metrics are enabled.
func Enabled() bool {
	state := activeState()
	return state != nil && state.enabled
}

// SlowThreshold returns the configured slow-request threshold (0 disables slow logs).
func SlowThreshold() time.Duration {
	state := activeState()
	if state == nil {
		return 0
	}
	return state.slowThreshold
}

// RequestTrackingEnabled reports whether per-request timing should be captured.
func RequestTrackingEnabled() bool {
	state := activeState()
	if state == nil {
		return false
	}
	return state.enabled || state.slowThreshold > 0
}

// Handler returns the Prometheus HTTP handler when metrics are enabled.
func Handler() http.Handler {
	state := activeState()
	if state == nil {
		return nil
	}
	return state.handler
}

// WriteRetryTotal returns the total backend write retries recorded across all
// label combinations. It is primarily for tests and diagnostics; it returns 0
// when metrics are disabled.
func WriteRetryTotal() float64 {
	state := activeState()
	if state == nil || state.registry == nil {
		return 0
	}
	families, err := state.registry.Gather()
	if err != nil {
		return 0
	}
	var total float64
	for _, family := range families {
		if family.GetName() != "proxy_mysql_write_retry_total" {
			continue
		}
		for _, metric := range family.GetMetric() {
			total += metric.GetCounter().GetValue()
		}
	}
	return total
}

// ObserveStage records a stage duration and updates the request tracker when provided.
func ObserveStage(stage Stage, d time.Duration, rm *RequestMetrics) {
	if rm != nil {
		rm.AddStage(stage, d)
	}
	state := activeState()
	if state == nil || !state.enabled || state.stageHist == nil {
		return
	}
	if stage < 0 || stage >= stageCount {
		return
	}
	labels := labelsFromRequestMetrics(nil, rm)
	state.stageHist.WithLabelValues(stageNames[stage], labels.KeyspaceName, labels.ClusterID).Observe(d.Seconds())
}

// IncRequests increments the request counter for an opcode.
func IncRequests(opcode byte, rm *RequestMetrics) {
	state := activeState()
	if state == nil || !state.enabled || state.reqTotal == nil {
		return
	}
	opcodeLabel, ok := opcodeLabel(opcode)
	if !ok {
		return
	}
	labels := labelsFromRequestMetrics(nil, rm)
	state.reqTotal.WithLabelValues(opcodeLabel, labels.KeyspaceName, labels.ClusterID).Inc()
}

// IncInFlightRequests increments the in-flight request gauge.
func IncInFlightRequests(rm *RequestMetrics) {
	state := activeState()
	if state == nil || !state.enabled || state.reqInFlight == nil {
		return
	}
	labels := defaultMetricLabels()
	if rm != nil {
		labels = rm.CaptureInFlightMetricLabels(nil)
	}
	state.reqInFlight.WithLabelValues(labels.KeyspaceName, labels.ClusterID).Inc()
}

// DecInFlightRequests decrements the in-flight request gauge.
func DecInFlightRequests(rm *RequestMetrics) {
	state := activeState()
	if state == nil || !state.enabled || state.reqInFlight == nil {
		return
	}
	labels := defaultMetricLabels()
	if rm != nil {
		labels = rm.InFlightMetricLabels(nil)
	}
	state.reqInFlight.WithLabelValues(labels.KeyspaceName, labels.ClusterID).Dec()
}

// ObserveRequestCompleted records request completion status and end-to-end latency.
func ObserveRequestCompleted(opcode byte, success bool, d time.Duration, rm *RequestMetrics) {
	state := activeState()
	if state == nil || !state.enabled || state.reqDone == nil || state.reqLatency == nil {
		return
	}
	opcodeLabel, opType, ok := requestCompletionLabels(opcode, rm)
	if !ok {
		return
	}
	labels := labelsFromRequestMetrics(nil, rm)
	status := requestStatusLabel(success)
	state.reqDone.WithLabelValues(opcodeLabel, opType, status, labels.KeyspaceName, labels.ClusterID).Inc()
	state.reqLatency.WithLabelValues(opcodeLabel, opType, status, labels.KeyspaceName, labels.ClusterID).Observe(d.Seconds())
}

// ObserveCQLQuery records CQL query count and latency by query type.
func ObserveCQLQuery(ctx context.Context, queryType string, success bool, d time.Duration) {
	queryType = normalizeLabel(queryType, "unknown")
	if queryType != "unknown" {
		SetRequestOperationType(ctx, queryType)
	}
	state := activeState()
	if state == nil || !state.enabled || state.cqlQueryTotal == nil || state.cqlQueryHist == nil {
		return
	}
	labels := labelsFromContext(ctx)
	status := requestStatusLabel(success)
	state.cqlQueryTotal.WithLabelValues(queryType, status, labels.KeyspaceName, labels.ClusterID).Inc()
	state.cqlQueryHist.WithLabelValues(queryType, status, labels.KeyspaceName, labels.ClusterID).Observe(d.Seconds())
}

// ObserveCQLWriteRows records rows written by CQL operation type.
func ObserveCQLWriteRows(ctx context.Context, operationType string, rows int64) {
	if rows <= 0 {
		return
	}
	state := activeState()
	if state == nil || !state.enabled || state.cqlWriteRows == nil {
		return
	}
	opLabel := normalizeLabel(operationType, "unknown")
	switch opLabel {
	case "insert", "update", "batch":
	default:
		return
	}
	labels := labelsFromContext(ctx)
	state.cqlWriteRows.WithLabelValues(opLabel, labels.KeyspaceName, labels.ClusterID).Add(float64(rows))
}

// RecordCQLDisconnection increments the CQL client disconnection counter.
func RecordCQLDisconnection(username string) {
	state := activeState()
	if state == nil || !state.enabled || state.cqlDisconnect == nil {
		return
	}
	labels := ResolveMetricLabels(nil, username)
	state.cqlDisconnect.WithLabelValues("all", labels.KeyspaceName, labels.ClusterID).Inc()
}

// ObserveMySQLOperation records a MySQL operation with latency and status.
func ObserveMySQLOperation(ctx context.Context, operation string, success bool, d time.Duration) {
	state := activeState()
	if state == nil || !state.enabled || state.mysqlOps == nil || state.mysqlLatency == nil {
		return
	}
	labels := labelsFromContext(ctx)
	opLabel := normalizeLabel(operation, "unknown")
	status := requestStatusLabel(success)
	state.mysqlOps.WithLabelValues(opLabel, status, labels.KeyspaceName, labels.ClusterID).Inc()
	state.mysqlLatency.WithLabelValues(opLabel, status, labels.KeyspaceName, labels.ClusterID).Observe(d.Seconds())
}

// RecordMySQLError increments MySQL error counters.
func RecordMySQLError(ctx context.Context, operation, errType string) {
	state := activeState()
	if state == nil || !state.enabled || state.mysqlErrors == nil {
		return
	}
	labels := labelsFromContext(ctx)
	opLabel := normalizeLabel(operation, "unknown")
	typeLabel := normalizeLabel(errType, "other")
	state.mysqlErrors.WithLabelValues(opLabel, typeLabel, labels.KeyspaceName, labels.ClusterID).Inc()
}

// RecordWriteRetry increments the backend write-retry counter for a retry kind
// (e.g. deadlock) and outcome (succeeded_after_retry or exhausted).
func RecordWriteRetry(ctx context.Context, kind, outcome string) {
	state := activeState()
	if state == nil || !state.enabled || state.writeRetries == nil {
		return
	}
	labels := labelsFromContext(ctx)
	kindLabel := normalizeLabel(kind, "unknown")
	outcomeLabel := normalizeLabel(outcome, "unknown")
	state.writeRetries.WithLabelValues(kindLabel, outcomeLabel, labels.KeyspaceName, labels.ClusterID).Inc()
}

func requestStatusLabel(success bool) string {
	if success {
		return "ok"
	}
	return "error"
}

func activeState() *metricsState {
	if value := globalState.Load(); value != nil {
		if state, ok := value.(*metricsState); ok {
			return state
		}
	}
	return nil
}

func mustDurationBuckets(values ...time.Duration) []float64 {
	if len(values) == 0 {
		panic("metrics: histogram buckets must not be empty")
	}
	out := make([]float64, len(values))
	prev := time.Duration(0)
	for i, v := range values {
		if v <= 0 {
			panic("metrics: histogram buckets must be positive")
		}
		if i > 0 && v <= prev {
			panic("metrics: histogram buckets must be strictly increasing")
		}
		out[i] = v.Seconds()
		prev = v
	}
	return out
}