// Copyright 2017 TiKV Project Authors.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package api

import (
	"net/http"
	"regexp"
	"strings"

	"github.com/pingcap/errors"
	"github.com/pingcap/kvproto/pkg/metapb"
	"github.com/tikv/pd/server"
	"github.com/unrolled/render"
)

type labelsHandler struct {
	svr *server.Server
	rd  *render.Render
}

func newLabelsHandler(svr *server.Server, rd *render.Render) *labelsHandler {
	return &labelsHandler{
		svr: svr,
		rd:  rd,
	}
}

// @Tags label
// @Summary List all label values.
// @Produce json
// @Success 200 {array} metapb.StoreLabel
// @Router /labels [get]
func (h *labelsHandler) Get(w http.ResponseWriter, r *http.Request) {
	rc := getCluster(r.Context())
	var labels []*metapb.StoreLabel
	m := make(map[string]struct{})
	stores := rc.GetStores()
	for _, s := range stores {
		ls := s.GetLabels()
		for _, l := range ls {
			if _, ok := m[strings.ToLower(l.Key+l.Value)]; !ok {
				m[strings.ToLower(l.Key+l.Value)] = struct{}{}
				labels = append(labels, l)
			}
		}
	}
	h.rd.JSON(w, http.StatusOK, labels)
}

// @Tags label
// @Summary List stores that have specific label values.
// @Param name query string true "name of store label filter"
// @Param value query string true "value of store label filter"
// @Produce json
// @Success 200 {object} StoresInfo
// @Failure 500 {string} string "PD server failed to proceed the request."
// @Router /labels/stores [get]
func (h *labelsHandler) GetStores(w http.ResponseWriter, r *http.Request) {
	rc := getCluster(r.Context())
	name := r.URL.Query().Get("name")
	value := r.URL.Query().Get("value")
	filter, err := newStoresLabelFilter(name, value)
	if err != nil {
		h.rd.JSON(w, http.StatusInternalServerError, err.Error())
		return
	}

	stores := rc.GetMetaStores()
	storesInfo := &StoresInfo{
		Stores: make([]*StoreInfo, 0, len(stores)),
	}

	stores = filter.filter(stores)
	for _, s := range stores {
		storeID := s.GetId()
		store := rc.GetStore(storeID)
		if store == nil {
			h.rd.JSON(w, http.StatusInternalServerError, server.ErrStoreNotFound(storeID).Error())
			return
		}

		storeInfo := newStoreInfo(h.svr.GetScheduleConfig(), store)
		storesInfo.Stores = append(storesInfo.Stores, storeInfo)
	}
	storesInfo.Count = len(storesInfo.Stores)

	h.rd.JSON(w, http.StatusOK, storesInfo)
}

type storesLabelFilter struct {
	keyPattern   *regexp.Regexp
	valuePattern *regexp.Regexp
}

func newStoresLabelFilter(name, value string) (*storesLabelFilter, error) {
	// add (?i) to set a case-insensitive flag
	keyPattern, err := regexp.Compile("(?i)" + name)
	if err != nil {
		return nil, errors.WithStack(err)
	}
	valuePattern, err := regexp.Compile("(?i)" + value)
	if err != nil {
		return nil, errors.WithStack(err)
	}
	return &storesLabelFilter{
		keyPattern:   keyPattern,
		valuePattern: valuePattern,
	}, nil
}

func (filter *storesLabelFilter) filter(stores []*metapb.Store) []*metapb.Store {
	ret := make([]*metapb.Store, 0, len(stores))
	for _, s := range stores {
		ls := s.GetLabels()
		for _, l := range ls {
			isKeyMatch := filter.keyPattern.MatchString(l.Key)
			isValueMatch := filter.valuePattern.MatchString(l.Value)
			if isKeyMatch && isValueMatch {
				ret = append(ret, s)
				break
			}
		}

	}
	return ret
}