package slowlog_explorer

import (
	"encoding/json"
	"fmt"
	"math"
	"reflect"
	"strconv"
	"strings"
	"time"
)

// transformRows converts raw Data Proxy rows into the format expected by the frontend.
// This is a faithful port of the Python _transform_rows() function.
func transformRows(result *DataProxyResponse, clusterID string) []map[string]interface{} {
	if result == nil || len(result.Rows) == 0 {
		return nil
	}

	colIndex := make(map[string]int)
	for i, col := range result.Columns {
		colIndex[col] = i
	}

	get := func(row []interface{}, key string) interface{} {
		idx, ok := colIndex[key]
		if !ok || idx >= len(row) {
			return nil
		}
		return row[idx]
	}

	getStr := func(row []interface{}, key string) *string {
		v := get(row, key)
		if v == nil {
			return nil
		}
		s := fmt.Sprintf("%v", v)
		if s == "" {
			return nil
		}
		return &s
	}

	items := make([]map[string]interface{}, 0, len(result.Rows))
	for _, row := range result.Rows {
		timeVal := get(row, "time")
		eventTime := ""
		var timeFloat float64
		if timeVal != nil {
			timeStr := fmt.Sprintf("%v", timeVal)
			f, err := strconv.ParseFloat(timeStr, 64)
			if err == nil {
				timeFloat = f
				// Convert to fixed-point string first to preserve precision
				// (raw value may be scientific notation like "1.775726429658004E9")
				fixedStr := strconv.FormatFloat(f, 'f', -1, 64)
				eventTime = fixedStrToISO(fixedStr)
			} else {
				eventTime = timeStr
			}
		}

		queryText := ""
		if q := get(row, "query"); q != nil {
			queryText = fmt.Sprintf("%v", q)
		}
		queryHead := queryText
		queryHead = truncateString(queryHead, maxQueryHead)

		digest := ""
		if d := getStr(row, "digest"); d != nil {
			digest = *d
		}
		connID := ""
		if c := get(row, "conn_id"); c != nil {
			connID = fmt.Sprintf("%v", c)
		}

		// Use raw time string (preserves precision like Python's Decimal)
		timeRaw := fmt.Sprintf("%v", get(row, "time"))
		if timeFloat > 0 {
			timeRaw = strconv.FormatFloat(timeFloat, 'f', -1, 64)
		}
		eventID := fmt.Sprintf("%s:%s:%s:%s", clusterID, digest, timeRaw, connID)

		item := map[string]interface{}{
			"event_id":                      eventID,
			"cluster_id":                    clusterID,
			"instance":                      getStr(row, "instance"),
			"event_time":                    eventTime,
			"db":                            getStr(row, "db"),
			"user_name":                     getStr(row, "user"),
			"conn_id":                       get(row, "conn_id"),
			"txn_start_ts":                  get(row, "txn_start_ts"),
			"digest":                        digest,
			"plan_digest":                   getStr(row, "plan_digest"),
			"query_time_ms":                 toMsOrZero(get(row, "query_time")),
			"parse_time_ms":                 toMs(get(row, "parse_time")),
			"compile_time_ms":               toMs(get(row, "compile_time")),
			"optimize_time_ms":              toMs(get(row, "optimize_time")),
			"rewrite_time_ms":               toMs(get(row, "rewrite_time")),
			"wait_ts_ms":                    toMs(get(row, "wait_ts")),
			"process_time_ms":               toMs(get(row, "process_time")),
			"wait_time_ms":                  toMs(get(row, "wait_time")),
			"backoff_time_ms":               toMs(get(row, "backoff_time")),
			"kv_total_ms":                   toMs(get(row, "kv_total")),
			"pd_total_ms":                   toMs(get(row, "pd_total")),
			"mem_max_bytes":                 toInt(get(row, "mem_max")),
			"disk_max_bytes":                toInt(get(row, "disk_max")),
			"total_keys":                    toInt(get(row, "total_keys")),
			"process_keys":                  toInt(get(row, "process_keys")),
			"result_rows":                   toInt(get(row, "result_rows")),
			"write_keys":                    toInt(get(row, "write_keys")),
			"write_size":                    toInt(get(row, "write_size")),
			"prewrite_region":               toInt(get(row, "prewrite_region")),
			"request_count":                 toInt(get(row, "request_count")),
			"txn_retry":                     toInt(get(row, "txn_retry")),
			"cop_proc_max_ms":               toMs(get(row, "cop_proc_max")),
			"cop_proc_avg_ms":               toMs(get(row, "cop_proc_avg")),
			"cop_proc_p90_ms":               toMs(get(row, "cop_proc_p90")),
			"cop_proc_addr":                 getStr(row, "cop_proc_addr"),
			"cop_wait_max_ms":               toMs(get(row, "cop_wait_max")),
			"cop_wait_avg_ms":               toMs(get(row, "cop_wait_avg")),
			"cop_wait_p90_ms":               toMs(get(row, "cop_wait_p90")),
			"cop_wait_addr":                 getStr(row, "cop_wait_addr"),
			"exec_retry_count":              toInt(get(row, "exec_retry_count")),
			"exec_retry_time_ms":            toMs(get(row, "exec_retry_time")),
			"preproc_subqueries":            toInt(get(row, "preproc_subqueries")),
			"preproc_subqueries_time_ms":    toMs(get(row, "preproc_subqueries_time")),
			"rocksdb_delete_skipped_count":  toInt(get(row, "rocksdb_delete_skipped_count")),
			"rocksdb_key_skipped_count":     toInt(get(row, "rocksdb_key_skipped_count")),
			"rocksdb_block_cache_hit_count": toInt(get(row, "rocksdb_block_cache_hit_count")),
			"rocksdb_block_read_count":      toInt(get(row, "rocksdb_block_read_count")),
			"rocksdb_block_read_byte":       toInt(get(row, "rocksdb_block_read_byte")),
			"succ":                          toBool(get(row, "succ")),
			"is_internal":                   toBoolFalse(get(row, "is_internal")),
			"is_explicit_txn":               toBool(get(row, "is_explicit_txn")),
			"prepared":                      toBool(get(row, "prepared")),
			"plan_from_cache":               toBoolFalse(get(row, "plan_from_cache")),
			"plan_from_binding":             toBoolFalse(get(row, "plan_from_binding")),
			"resource_group":                getStr(row, "resource_group"),
			"ru_read":                       toFloat(get(row, "request_unit_read")),
			"ru_write":                      toFloat(get(row, "request_unit_write")),
			"time_queued_by_rc_ms":          toMs(get(row, "time_queued_by_rc")),
			"cop_time_ms":                   toMs(get(row, "cop_time")),
			"index_names":                   getStr(row, "index_names"),
			"query_head":                    queryHead,
			"query_text":                    nilIfEmpty(queryText),
		}

		// Keep raw time for cursor pagination
		item["_time"] = timeFloat

		items = append(items, item)
	}

	return items
}

// transformDetail adds detail-only fields on top of transformRows output.
// Faithful port of the Python get_slowlog_detail endpoint.
func transformDetail(result *DataProxyResponse, clusterID string) map[string]interface{} {
	items := transformRows(result, clusterID)
	if len(items) == 0 {
		return nil
	}
	record := items[0]
	delete(record, "_time")

	// Build raw row_dict for accessing detail-only columns
	colIndex := make(map[string]int)
	for i, col := range result.Columns {
		colIndex[col] = i
	}
	row := result.Rows[0]
	getRaw := func(key string) interface{} {
		idx, ok := colIndex[key]
		if !ok || idx >= len(row) {
			return nil
		}
		return row[idx]
	}
	getRawStr := func(key string) *string {
		v := getRaw(key)
		if v == nil {
			return nil
		}
		s := fmt.Sprintf("%v", v)
		if s == "" {
			return nil
		}
		return &s
	}

	// plan_text
	record["plan_text"] = getRawStr("plan")

	// decoded_plan — try JSON parse
	record["decoded_plan"] = nil
	if dp := getRaw("decoded_plan"); dp != nil {
		dpStr := fmt.Sprintf("%v", dp)
		if dpStr != "" {
			var parsed interface{}
			if err := json.Unmarshal([]byte(dpStr), &parsed); err == nil {
				record["decoded_plan"] = parsed
			}
		}
	}

	// Other detail-only text fields
	record["binary_plan_text"] = getRawStr("binary_plan")
	record["prev_stmt_text"] = getRawStr("prev_stmt")
	record["warnings_json"] = getRawStr("warnings")
	record["backoff_detail_text"] = getRawStr("backoff_detail")

	// Detail-only timing fields (seconds → ms with 2 decimal places)
	for _, field := range []string{
		"prewrite_time", "commit_time", "get_commit_ts_time",
		"commit_backoff_time", "resolve_lock_time",
		"local_latch_wait_time", "lockkeys_time",
		"backoff_total", "write_sql_response_total",
		"wait_prewrite_binlog_time",
	} {
		record[field+"_ms"] = toMsFloat(getRaw(field))
	}

	// backoff_types as string
	record["backoff_types"] = getRawStr("backoff_types")

	return record
}

func rawRecord(result *DataProxyResponse) map[string]interface{} {
	if result == nil || len(result.Rows) == 0 {
		return nil
	}

	row := result.Rows[0]
	record := make(map[string]interface{}, len(result.Columns))
	for i, col := range result.Columns {
		if i < len(row) {
			record[col] = row[i]
		} else {
			record[col] = nil
		}
	}
	return record
}

func publicItems(items []map[string]interface{}) []map[string]interface{} {
	if items == nil {
		return nil
	}
	out := make([]map[string]interface{}, 0, len(items))
	for _, item := range items {
		public := make(map[string]interface{}, len(item))
		for k, v := range item {
			if k == "_time" {
				continue
			}
			public[k] = v
		}
		out = append(out, public)
	}
	return out
}

func extractLastTime(items []map[string]interface{}) *float64 {
	if len(items) == 0 {
		return nil
	}
	last := items[len(items)-1]
	if t, ok := last["_time"]; ok {
		if f, ok := t.(float64); ok && f > 0 {
			return &f
		}
	}
	return nil
}

// fixedStrToISO converts a fixed-point unix timestamp string (e.g. "1775726429.658004")
// to ISO 8601 format, preserving microsecond precision without float64 loss.
func fixedStrToISO(s string) string {
	parts := strings.SplitN(s, ".", 2)
	secs, err := strconv.ParseInt(parts[0], 10, 64)
	if err != nil {
		return s
	}
	fracStr := ""
	if len(parts) == 2 {
		fracStr = parts[1]
	}
	// Pad/truncate to 6 digits (microseconds)
	for len(fracStr) < 6 {
		fracStr += "0"
	}
	if len(fracStr) > 6 {
		fracStr = fracStr[:6]
	}
	micros, _ := strconv.ParseInt(fracStr, 10, 64)
	t := time.Unix(secs, micros*1000).UTC()
	return t.Format("2006-01-02T15:04:05.999999+00:00")
}

// toNumber converts any value to float64.
func toNumber(v interface{}) (float64, error) {
	switch val := v.(type) {
	case float64:
		return val, nil
	case string:
		return parseNumericString(val)
	case json.Number:
		return val.Float64()
	case nil:
		return 0, fmt.Errorf("nil")
	default:
		return parseNumericString(fmt.Sprintf("%v", val))
	}
}

func parseNumericString(s string) (float64, error) {
	s = strings.TrimSpace(s)
	if s == "" {
		return 0, fmt.Errorf("empty")
	}
	if strings.Count(s, "/") == 1 {
		parts := strings.SplitN(s, "/", 2)
		numerator, err := strconv.ParseFloat(strings.TrimSpace(parts[0]), 64)
		if err != nil {
			return 0, err
		}
		denominator, err := strconv.ParseFloat(strings.TrimSpace(parts[1]), 64)
		if err != nil {
			return 0, err
		}
		if denominator == 0 {
			return 0, fmt.Errorf("zero denominator")
		}
		return numerator / denominator, nil
	}
	return strconv.ParseFloat(s, 64)
}

func quantileAt(v interface{}, idx int) interface{} {
	if v == nil || idx < 0 {
		return nil
	}
	rv := reflect.ValueOf(v)
	if rv.Kind() != reflect.Slice && rv.Kind() != reflect.Array {
		return nil
	}
	if idx >= rv.Len() {
		return nil
	}
	return rv.Index(idx).Interface()
}

// --- Type conversion helpers matching Python _to_ms, _to_int, _to_bool ---

// toMs converts seconds to integer milliseconds, or nil if empty/invalid.
func toMs(v interface{}) interface{} {
	if v == nil {
		return nil
	}
	s := fmt.Sprintf("%v", v)
	if s == "" {
		return nil
	}
	f, err := parseNumericString(s)
	if err != nil {
		return nil
	}
	return int(math.Round(f * 1000))
}

// toMsOrZero is like toMs but returns 0 instead of nil.
func toMsOrZero(v interface{}) interface{} {
	r := toMs(v)
	if r == nil {
		return 0
	}
	return r
}

// toMsFloat converts seconds to ms with 2 decimal places, or nil.
func toMsFloat(v interface{}) interface{} {
	if v == nil {
		return nil
	}
	s := fmt.Sprintf("%v", v)
	if s == "" {
		return nil
	}
	f, err := parseNumericString(s)
	if err != nil {
		return nil
	}
	return math.Round(f*1000*100) / 100
}

// toMsFloatOrZero is like toMsFloat but returns 0 instead of nil.
func toMsFloatOrZero(v interface{}) interface{} {
	r := toMsFloat(v)
	if r == nil {
		return 0
	}
	return r
}

// toInt converts to int, or nil if empty/invalid.
func toInt(v interface{}) interface{} {
	if v == nil {
		return nil
	}
	s := fmt.Sprintf("%v", v)
	if s == "" {
		return nil
	}
	f, err := parseNumericString(s)
	if err != nil {
		return nil
	}
	return int(f)
}

// toIntOrZero is like toInt but returns 0 instead of nil.
func toIntOrZero(v interface{}) interface{} {
	r := toInt(v)
	if r == nil {
		return 0
	}
	return r
}

// toFloat converts to float64, or nil if empty/invalid.
func toFloat(v interface{}) interface{} {
	if v == nil {
		return nil
	}
	s := fmt.Sprintf("%v", v)
	if s == "" {
		return nil
	}
	f, err := parseNumericString(s)
	if err != nil {
		return nil
	}
	return f
}

// toFloatOrZero is like toFloat but returns 0 instead of nil.
func toFloatOrZero(v interface{}) interface{} {
	r := toFloat(v)
	if r == nil {
		return 0
	}
	return r
}

// toBool converts to bool, or nil if empty/invalid.
func toBool(v interface{}) interface{} {
	if v == nil {
		return nil
	}
	s := fmt.Sprintf("%v", v)
	if s == "" {
		return nil
	}
	switch s {
	case "1", "true", "True":
		return true
	case "0", "false", "False":
		return false
	}
	return nil
}

// toBoolFalse is like toBool but returns false instead of nil.
func toBoolFalse(v interface{}) interface{} {
	r := toBool(v)
	if r == nil {
		return false
	}
	return r
}

// rawAt safely gets element at index from a row slice.
func rawAt(row []interface{}, idx int) interface{} {
	if idx >= len(row) {
		return nil
	}
	return row[idx]
}

// rawStr safely gets string at index from a row slice.
func rawStr(row []interface{}, idx int) string {
	if idx >= len(row) || row[idx] == nil {
		return ""
	}
	return fmt.Sprintf("%v", row[idx])
}

func nilIfEmpty(s string) interface{} {
	if s == "" {
		return nil
	}
	return s
}