package virtualtables

import (
	"strings"
	"sync"
	"time"

	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/types"
)

// MetricsCollector collects and stores query performance metrics
type MetricsCollector struct {
	mu             sync.RWMutex
	readLatencies  map[string][]float64 // keyspace.table -> latencies
	writeLatencies map[string][]float64
	coordReadLat   map[string][]float64
	coordWriteLat  map[string][]float64
	requestCount   map[string]int64
	maxLatency     map[string]float64
	perSecond      map[string]float64
	lastReset      time.Time
}

// NewMetricsCollector creates a new metrics collector
func NewMetricsCollector() *MetricsCollector {
	return &MetricsCollector{
		readLatencies:  make(map[string][]float64),
		writeLatencies: make(map[string][]float64),
		coordReadLat:   make(map[string][]float64),
		coordWriteLat:  make(map[string][]float64),
		requestCount:   make(map[string]int64),
		maxLatency:     make(map[string]float64),
		perSecond:      make(map[string]float64),
		lastReset:      time.Now(),
	}
}

// RecordReadLatency records a read operation latency
func (mc *MetricsCollector) RecordReadLatency(keyspace, table string, latencyMs float64) {
	mc.mu.Lock()
	defer mc.mu.Unlock()

	key := keyspace + "." + table
	mc.readLatencies[key] = append(mc.readLatencies[key], latencyMs)
	mc.requestCount[key]++

	if latencyMs > mc.maxLatency[key] {
		mc.maxLatency[key] = latencyMs
	}

	mc.updatePerSecond(key)
}

// RecordWriteLatency records a write operation latency
func (mc *MetricsCollector) RecordWriteLatency(keyspace, table string, latencyMs float64) {
	mc.mu.Lock()
	defer mc.mu.Unlock()

	key := keyspace + "." + table
	mc.writeLatencies[key] = append(mc.writeLatencies[key], latencyMs)
	mc.requestCount[key]++

	if latencyMs > mc.maxLatency[key] {
		mc.maxLatency[key] = latencyMs
	}

	mc.updatePerSecond(key)
}

// RecordCoordinatorReadLatency records coordinator read latency
func (mc *MetricsCollector) RecordCoordinatorReadLatency(keyspace, table string, latencyMs float64) {
	mc.mu.Lock()
	defer mc.mu.Unlock()

	key := keyspace + "." + table
	mc.coordReadLat[key] = append(mc.coordReadLat[key], latencyMs)
	mc.requestCount[key]++

	if latencyMs > mc.maxLatency[key] {
		mc.maxLatency[key] = latencyMs
	}

	mc.updatePerSecond(key)
}

// RecordCoordinatorWriteLatency records coordinator write latency
func (mc *MetricsCollector) RecordCoordinatorWriteLatency(keyspace, table string, latencyMs float64) {
	mc.mu.Lock()
	defer mc.mu.Unlock()

	key := keyspace + "." + table
	mc.coordWriteLat[key] = append(mc.coordWriteLat[key], latencyMs)
	mc.requestCount[key]++

	if latencyMs > mc.maxLatency[key] {
		mc.maxLatency[key] = latencyMs
	}

	mc.updatePerSecond(key)
}

// GetReadLatencyMetrics returns read latency metrics for all tables
func (mc *MetricsCollector) GetReadLatencyMetrics() []types.Row {
	mc.mu.RLock()
	defer mc.mu.RUnlock()

	rows := make([]types.Row, 0, len(mc.readLatencies))
	for key, latencies := range mc.readLatencies {
		if len(latencies) == 0 {
			continue
		}

		parts := splitKey(key)
		p50 := percentile(latencies, 50)
		p99 := percentile(latencies, 99)
		count := mc.requestCount[key]
		max := mc.maxLatency[key]
		perSec := mc.perSecond[key]

		rows = append(rows, types.Row{
			Values: []interface{}{
				parts[0], // keyspace
				parts[1], // table
				p50,
				p99,
				count,
				max,
				perSec,
			},
		})
	}
	return rows
}

// GetWriteLatencyMetrics returns write latency metrics for all tables
func (mc *MetricsCollector) GetWriteLatencyMetrics() []types.Row {
	mc.mu.RLock()
	defer mc.mu.RUnlock()

	rows := make([]types.Row, 0, len(mc.writeLatencies))
	for key, latencies := range mc.writeLatencies {
		if len(latencies) == 0 {
			continue
		}

		parts := splitKey(key)
		p50 := percentile(latencies, 50)
		p99 := percentile(latencies, 99)
		count := mc.requestCount[key]
		max := mc.maxLatency[key]
		perSec := mc.perSecond[key]

		rows = append(rows, types.Row{
			Values: []interface{}{
				parts[0],
				parts[1],
				p50,
				p99,
				count,
				max,
				perSec,
			},
		})
	}
	return rows
}

// GetCoordinatorReadLatencyMetrics returns coordinator read metrics
func (mc *MetricsCollector) GetCoordinatorReadLatencyMetrics() []types.Row {
	mc.mu.RLock()
	defer mc.mu.RUnlock()

	rows := make([]types.Row, 0, len(mc.coordReadLat))
	for key, latencies := range mc.coordReadLat {
		if len(latencies) == 0 {
			continue
		}

		parts := splitKey(key)
		p50 := percentile(latencies, 50)
		p99 := percentile(latencies, 99)
		count := mc.requestCount[key]
		max := mc.maxLatency[key]
		perSec := mc.perSecond[key]

		rows = append(rows, types.Row{
			Values: []interface{}{
				parts[0],
				parts[1],
				p50,
				p99,
				count,
				max,
				perSec,
			},
		})
	}
	return rows
}

// GetCoordinatorWriteLatencyMetrics returns coordinator write metrics
func (mc *MetricsCollector) GetCoordinatorWriteLatencyMetrics() []types.Row {
	mc.mu.RLock()
	defer mc.mu.RUnlock()

	rows := make([]types.Row, 0, len(mc.coordWriteLat))
	for key, latencies := range mc.coordWriteLat {
		if len(latencies) == 0 {
			continue
		}

		parts := splitKey(key)
		p50 := percentile(latencies, 50)
		p99 := percentile(latencies, 99)
		count := mc.requestCount[key]
		max := mc.maxLatency[key]
		perSec := mc.perSecond[key]

		rows = append(rows, types.Row{
			Values: []interface{}{
				parts[0],
				parts[1],
				p50,
				p99,
				count,
				max,
				perSec,
			},
		})
	}
	return rows
}

// Reset clears all metrics
func (mc *MetricsCollector) Reset() {
	mc.mu.Lock()
	defer mc.mu.Unlock()

	mc.readLatencies = make(map[string][]float64)
	mc.writeLatencies = make(map[string][]float64)
	mc.coordReadLat = make(map[string][]float64)
	mc.coordWriteLat = make(map[string][]float64)
	mc.requestCount = make(map[string]int64)
	mc.maxLatency = make(map[string]float64)
	mc.perSecond = make(map[string]float64)
	mc.lastReset = time.Now()
}

// updatePerSecond updates the per-second rate for a key
func (mc *MetricsCollector) updatePerSecond(key string) {
	elapsed := time.Since(mc.lastReset).Seconds()
	if elapsed > 0 {
		mc.perSecond[key] = float64(mc.requestCount[key]) / elapsed
	}
}

// percentile calculates the p-th percentile of a slice of values
func percentile(values []float64, p float64) float64 {
	if len(values) == 0 {
		return 0.0
	}

	// Create a copy to avoid modifying the original
	sorted := make([]float64, len(values))
	copy(sorted, values)

	// Simple bubble sort for small datasets
	for i := 0; i < len(sorted)-1; i++ {
		for j := 0; j < len(sorted)-i-1; j++ {
			if sorted[j] > sorted[j+1] {
				sorted[j], sorted[j+1] = sorted[j+1], sorted[j]
			}
		}
	}

	// Calculate index
	index := int(float64(len(sorted)-1) * p / 100.0)
	if index < 0 {
		index = 0
	}
	if index >= len(sorted) {
		index = len(sorted) - 1
	}

	return sorted[index]
}

// splitKey splits a keyspace.table key into its parts
func splitKey(key string) []string {
	parts := []string{"", ""}
	idx := strings.Index(key, ".")
	if idx >= 0 {
		parts[0] = key[:idx]
		parts[1] = key[idx+1:]
	} else {
		parts[0] = key
		parts[1] = ""
	}
	return parts
}

// Global metrics collector instance
var globalMetricsCollector = NewMetricsCollector()

// GetMetricsCollector returns the global metrics collector
func GetMetricsCollector() *MetricsCollector {
	return globalMetricsCollector
}