package server

import (
	"encoding/json"
	"fmt"
	"net/http"
	"net/url"
	"strconv"
	"strings"

	"github.com/pingcap/kvproto/pkg/keyspacepb"
	"github.com/pingcap/log"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	metametrics "github.com/tidbcloud/aws-shared-provider/pkg/metaserver/metrics"
	"github.com/tidbcloud/aws-shared-provider/pkg/metaserver/storage"
	"go.uber.org/zap"

	inframysql "github.com/tidbcloud/tidb-management-service/api/serverless_infra/client/mysql"
)

const (
	defaultListKeyspacesLimit = 100
	maxListKeyspacesLimit     = 1000
)

// ListKeyspacesPaginatedResponse is the response body for paginated list keyspace APIs.
type ListKeyspacesPaginatedResponse struct {
	Items  []*storage.KeyspaceMetaData `json:"items"`
	Total  int                         `json:"total"`
	Limit  int                         `json:"limit"`
	Offset int                         `json:"offset"`
}

func registerHandlers(server *Server) {
	registerHandler := func(route string, handler http.HandlerFunc) {
		server.mux.Handle(route, metametrics.InstrumentHandler(route, handler))
	}

	registerHandler("/api/v1/meta", server.handleKeyspaceDataLookup)
	registerHandler("/api/v1/meta:list", server.handleListKeyspaceMeta)
	registerHandler("/api/v1/clusters", server.handleClusterDataLookup)
	registerHandler("/api/v1/events", server.handleClusterEventList)
	// V2 APIs - support querying keyspaces with any state
	registerHandler("/api/v2/meta", server.handleKeyspaceDataLookupV2)
	registerHandler("/api/v2/meta:list", server.handleListKeyspaceMetaV2)
	server.mux.Handle("/metrics", promhttp.Handler())
}

// parseListKeyspacesPagination parses limit and offset from query. If limit or offset is set, returns (limit, offset, true). Otherwise (0, 0, false).
func parseListKeyspacesPagination(query url.Values) (limit, offset int, usePagination bool) {
	limitStr := query.Get("limit")
	offsetStr := query.Get("offset")
	if limitStr == "" && offsetStr == "" {
		return 0, 0, false
	}
	limit = defaultListKeyspacesLimit
	if limitStr != "" {
		if n, err := strconv.Atoi(limitStr); err == nil && n > 0 {
			limit = n
			if limit > maxListKeyspacesLimit {
				limit = maxListKeyspacesLimit
			}
		}
	}
	if offsetStr != "" {
		if n, err := strconv.Atoi(offsetStr); err == nil && n >= 0 {
			offset = n
		}
	}
	return limit, offset, true
}

func parseKeyspaceListFilter(query url.Values) (storage.KeyspaceListFilter, error) {
	var filter storage.KeyspaceListFilter
	stateStr := query.Get("state")
	if stateStr != "" {
		state, err := parseKeyspaceState(stateStr)
		if err != nil {
			return filter, err
		}
		filter.HasState = true
		filter.State = state
	}
	return filter, nil
}

func parseKeyspaceState(raw string) (keyspacepb.KeyspaceState, error) {
	if n, err := strconv.ParseInt(raw, 10, 32); err == nil {
		state := keyspacepb.KeyspaceState(n)
		if _, ok := keyspacepb.KeyspaceState_name[int32(state)]; ok {
			return state, nil
		}
		return 0, fmt.Errorf("unknown keyspace state %q", raw)
	}

	normalized := strings.TrimPrefix(strings.ToUpper(raw), "KEYSPACESTATE_")
	if v, ok := keyspacepb.KeyspaceState_value[normalized]; ok {
		return keyspacepb.KeyspaceState(v), nil
	}
	return 0, fmt.Errorf("unknown keyspace state %q", raw)
}

func (s *Server) handleKeyspaceDataLookup(w http.ResponseWriter, r *http.Request) {
	query := r.URL.Query()
	log.Info("[http server] meta lookup request receive", zap.String("query", r.URL.RawQuery))
	var (
		meta            []*storage.KeyspaceMetaData
		err             error
		validIndexFound bool
	)
	for _, index := range storage.KeyspaceSupportedIndex {
		if key := query.Get(string(index)); key != "" {
			validIndexFound = true
			meta, err = s.store.LookupKeyspace(index, key, true) // v1 only returns enabled keyspaces
			break
		}
	}
	if !validIndexFound {
		log.Error("[http server] missing index")
		http.Error(w, "No valid index provided in the query", http.StatusBadRequest)
		return
	}
	if err != nil {
		log.Error("[http server] lookup failed", zap.Error(err))
		http.Error(w, fmt.Sprintf("Failed to lookup metadata: %s", err), http.StatusInternalServerError)
		return
	}
	if len(meta) == 0 {
		log.Error("[http server] no matching record")
		http.Error(w, "No matching records found", http.StatusNotFound)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	if err = json.NewEncoder(w).Encode(meta); err != nil {
		log.Error("[http server] failed to encode response", zap.Error(err))
		http.Error(w, fmt.Sprintf("Failed to encode response metadata: %s", err), http.StatusInternalServerError)
	}
}

func (s *Server) handleListKeyspaceMeta(w http.ResponseWriter, r *http.Request) {
	log.Info("[http server] list keyspace meta request received")
	query := r.URL.Query()
	filter := storage.KeyspaceListFilter{HasState: true, State: keyspacepb.KeyspaceState_ENABLED}
	if limit, offset, usePag := parseListKeyspacesPagination(query); usePag {
		items, total, err := s.store.ListKeyspacesPaginated(filter, limit, offset)
		if err != nil {
			log.Error("[http server] list keyspace meta failed", zap.Error(err))
			http.Error(w, fmt.Sprintf("Failed to list keyspace metadata: %s", err), http.StatusInternalServerError)
			return
		}
		w.Header().Set("Content-Type", "application/json")
		resp := ListKeyspacesPaginatedResponse{Items: items, Total: total, Limit: limit, Offset: offset}
		if err = json.NewEncoder(w).Encode(resp); err != nil {
			log.Error("[http server] failed to encode response", zap.Error(err))
			http.Error(w, fmt.Sprintf("Failed to encode response metadata: %s", err), http.StatusInternalServerError)
		}
		return
	}
	meta, err := s.store.ListKeyspaces(filter) // v1 only returns enabled keyspaces
	if err != nil {
		log.Error("[http server] list keyspace meta failed", zap.Error(err))
		http.Error(w, fmt.Sprintf("Failed to list keyspace metadata: %s", err), http.StatusInternalServerError)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	if err = json.NewEncoder(w).Encode(meta); err != nil {
		log.Error("[http server] failed to encode response", zap.Error(err))
		http.Error(w, fmt.Sprintf("Failed to encode response metadata: %s", err), http.StatusInternalServerError)
	}
}

func (s *Server) handleClusterDataLookup(w http.ResponseWriter, r *http.Request) {
	query := r.URL.Query()
	log.Info("[http server] cluster lookup request receive", zap.String("query", r.URL.RawQuery))
	var (
		cluster         []*inframysql.Cluster
		err             error
		validIndexFound bool
	)
	for _, index := range storage.ClusterSupportedIndex {
		if key := query.Get(string(index)); key != "" {
			validIndexFound = true
			cluster, err = s.store.LookupCluster(index, key)
			break
		}
	}
	if !validIndexFound {
		log.Error("[http server] missing index")
		http.Error(w, "No valid index provided in the query", http.StatusBadRequest)
		return
	}
	if err != nil {
		log.Error("[http server] lookup failed", zap.Error(err))
		http.Error(w, fmt.Sprintf("Failed to lookup cluster: %s", err), http.StatusInternalServerError)
		return
	}
	if len(cluster) == 0 {
		log.Error("[http server] no matching record")
		http.Error(w, "No matching records found", http.StatusNotFound)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	if err := json.NewEncoder(w).Encode(cluster); err != nil {
		log.Error("[http server] failed to encode response", zap.Error(err))
		http.Error(w, fmt.Sprintf("Failed to encode response metadata: %s", err), http.StatusInternalServerError)
	}
}

func (s *Server) handleClusterEventList(w http.ResponseWriter, r *http.Request) {
	if r.Method != http.MethodGet {
		w.Header().Set("Allow", http.MethodGet)
		http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
		return
	}

	eventWindow, err := s.store.GetClusterEventWindow()
	if err != nil {
		log.Error("[http server] get cluster event window failed", zap.Error(err))
		http.Error(w, fmt.Sprintf("Failed to get cluster event window: %s", err), http.StatusInternalServerError)
		return
	}

	w.Header().Set("Content-Type", "application/json")
	if err = json.NewEncoder(w).Encode(eventWindow); err != nil {
		log.Error("[http server] failed to encode cluster event window", zap.Error(err))
		http.Error(w, fmt.Sprintf("Failed to encode cluster event window: %s", err), http.StatusInternalServerError)
	}
}

// handleKeyspaceDataLookupV2 handles v2 API for keyspace lookup, returns keyspaces with any state
func (s *Server) handleKeyspaceDataLookupV2(w http.ResponseWriter, r *http.Request) {
	query := r.URL.Query()
	log.Info("[http server] v2 meta lookup request receive", zap.String("query", r.URL.RawQuery))
	var (
		meta            []*storage.KeyspaceMetaData
		err             error
		validIndexFound bool
	)
	for _, index := range storage.KeyspaceSupportedIndex {
		if key := query.Get(string(index)); key != "" {
			validIndexFound = true
			meta, err = s.store.LookupKeyspace(index, key, false) // v2 returns keyspaces with any state
			break
		}
	}
	if !validIndexFound {
		log.Error("[http server] missing index")
		http.Error(w, "No valid index provided in the query", http.StatusBadRequest)
		return
	}
	if err != nil {
		log.Error("[http server] lookup failed", zap.Error(err))
		http.Error(w, fmt.Sprintf("Failed to lookup metadata: %s", err), http.StatusInternalServerError)
		return
	}
	if len(meta) == 0 {
		log.Error("[http server] no matching record")
		http.Error(w, "No matching records found", http.StatusNotFound)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	if err = json.NewEncoder(w).Encode(meta); err != nil {
		log.Error("[http server] failed to encode response", zap.Error(err))
		http.Error(w, fmt.Sprintf("Failed to encode response metadata: %s", err), http.StatusInternalServerError)
	}
}

// handleListKeyspaceMetaV2 handles v2 API for listing all keyspaces, returns keyspaces with any state
func (s *Server) handleListKeyspaceMetaV2(w http.ResponseWriter, r *http.Request) {
	log.Info("[http server] v2 list keyspace meta request received")
	query := r.URL.Query()
	filter, err := parseKeyspaceListFilter(query)
	if err != nil {
		log.Error("[http server] invalid list keyspace filter", zap.Error(err))
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}
	if limit, offset, usePag := parseListKeyspacesPagination(query); usePag {
		items, total, err := s.store.ListKeyspacesPaginated(filter, limit, offset)
		if err != nil {
			log.Error("[http server] list keyspace meta failed", zap.Error(err))
			http.Error(w, fmt.Sprintf("Failed to list keyspace metadata: %s", err), http.StatusInternalServerError)
			return
		}
		w.Header().Set("Content-Type", "application/json")
		resp := ListKeyspacesPaginatedResponse{Items: items, Total: total, Limit: limit, Offset: offset}
		if err = json.NewEncoder(w).Encode(resp); err != nil {
			log.Error("[http server] failed to encode response", zap.Error(err))
			http.Error(w, fmt.Sprintf("Failed to encode response metadata: %s", err), http.StatusInternalServerError)
		}
		return
	}
	items, err := s.store.ListKeyspaces(filter) // v2 returns keyspaces with any state when filter is empty
	if err != nil {
		log.Error("[http server] list keyspace meta failed", zap.Error(err))
		http.Error(w, fmt.Sprintf("Failed to list keyspace metadata: %s", err), http.StatusInternalServerError)
		return
	}
	w.Header().Set("Content-Type", "application/json")
	if err = json.NewEncoder(w).Encode(items); err != nil {
		log.Error("[http server] failed to encode response", zap.Error(err))
		http.Error(w, fmt.Sprintf("Failed to encode response metadata: %s", err), http.StatusInternalServerError)
	}
}