package slowlog

import (
	"context"
	"fmt"
	"github.com/pingcap/log"
	"github.com/pingcap/tep/pkg/apis/apiserver"
	"github.com/pingcap/tep/pkg/operator"
	store "github.com/pingcap/tep/pkg/registry/apiserver"
	"github.com/pingcap/tep/pkg/registry/apiserver/tidbcluster"
	"github.com/pingcap/tep/pkg/utils"
	"github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"github.com/pingcap/tidb-operator/pkg/label"
	"github.com/wushilin/stream"
	"go.uber.org/zap"
	"k8s.io/apimachinery/pkg/api/errors"
	metainternalversion "k8s.io/apimachinery/pkg/apis/meta/internalversion"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"reflect"
	"strings"
)

type SlowInstancesStorage struct {
	store *store.Store
}

func NewSlowInstancesStorage(store *store.Store) *SlowInstancesStorage {
	return &SlowInstancesStorage{
		store: store,
	}
}

func (s *SlowInstancesStorage) Kind() string {
	return "TiDBClusterSlowQuery"
}

func (s *SlowInstancesStorage) New() runtime.Object {
	return &apiserver.TiDBClusterSlowQuery{}
}

func (s *SlowInstancesStorage) NewList() runtime.Object {
	return &apiserver.TiDBClusterSlowQueryList{}
}

func (s *SlowInstancesStorage) NamespaceScoped() bool {
	return false
}

func (s *SlowInstancesStorage) NewExtraListOptions() runtime.Object {
	return &apiserver.TiDBClusterSlowQueryListOptions{}
}

func (s *SlowInstancesStorage) List(ctx context.Context, options *metainternalversion.ListOptions, extraOptions runtime.Object) (runtime.Object, error) {
	listOpts := extraOptions.(*apiserver.TiDBClusterSlowQueryListOptions)

	opts := metav1.ListOptions{}
	opts.LabelSelector = fmt.Sprintf("%s=%s", operator.CreatedByLabelKey, operator.CreatedByLabelVal)

	if listOpts.Name != "" {
		if err := tidbcluster.ValidateName(listOpts.Name); err != nil {
			msg := fmt.Sprintf("instance id is invalid: %v", err)
			return nil, errors.NewBadRequest(msg)
		}
		opts.LabelSelector += fmt.Sprintf(",%s=%s", label.InstanceLabelKey, listOpts.Name)
	}

	if listOpts.ClusterName != "" {
		if err := tidbcluster.ValidateClusterName(listOpts.ClusterName); err != nil {
			msg := fmt.Sprintf("instance name is invalid: %v", err)
			return nil, errors.NewBadRequest(msg)
		}
		opts.LabelSelector += fmt.Sprintf(",%s=%s", operator.NameLabelKey, listOpts.ClusterName)
	}

	var tidbClusters []v1alpha1.TidbCluster
	// get instance list with tag filter
	// multiple tags separated by comma，e.g. tags=pingcap,tep,cloud
	if listOpts.Tags != "" {
		tags := strings.Split(listOpts.Tags, ",")
		tagLables := make([]string, 0, len(tags))
		for _, tag := range tags {
			tagLables = append(tagLables, operator.TagLabelKeyPrefix+strings.TrimSpace(tag))
		}

		var tidbClusterList []v1alpha1.TidbCluster
		for _, tagLabel := range tagLables {
			opts.LabelSelector = tagLabel
			tcl, err := s.store.Controller().GetTidbClusters(metav1.NamespaceAll, opts)
			if err != nil {
				msg := fmt.Sprintf("failed to get instance list: %v", err)
				return nil, errors.NewBadRequest(msg)
			}
			tidbClusterList = append(tidbClusterList, tcl.Items...)
		}

		m := make(map[string]bool)
		for _, tc := range tidbClusterList {
			ns := tc.GetNamespace()
			if _, ok := m[ns]; !ok {
				m[ns] = true
				tidbClusters = append(tidbClusters, tc)
			}
		}
	} else {
		tcl, err := s.store.Controller().GetTidbClusters(metav1.NamespaceAll, opts)
		if err != nil {
			msg := fmt.Errorf("failed to get instance list: %v", err)
			return nil, errors.NewInternalError(msg)
		}
		tidbClusters = tcl.Items
	}

	name2TC := make(map[string]*v1alpha1.TidbCluster)
	stream.FromArray(tidbClusters).Each(func(tc v1alpha1.TidbCluster) {
		name2TC[tc.Name] = &tc
	})

	return s.query(ctx, options, extraOptions, name2TC)
}

func (s *SlowInstancesStorage) query(ctx context.Context, options *metainternalversion.ListOptions, extraOptions runtime.Object, name2TC map[string]*v1alpha1.TidbCluster) (runtime.Object, error) {
	listOptions := extraOptions.(*apiserver.TiDBClusterSlowQueryListOptions)

	items := make([]interface{}, len(name2TC))
	stream.FromMapKeys(name2TC).Map(func(name string) interface{} {
		return name
	}).CollectTo(items)

	tcInES, err := s.ListSlowAgg(Context{
		From:           utils.GetStartIndex(listOptions.Page, listOptions.ItemsPerPage),
		Size:           utils.GetSize(listOptions.ItemsPerPage),
		StartTimeInSec: listOptions.StartTimeInSec,
		EndTimeInSec:   listOptions.EndTimeInSec,
		Items:          items,
	})

	if err != nil {
		log.Error("query slow log failed", zap.Error(err))
		return nil, errors.NewBadRequest("query slow log failed")
	}

	if len(tcInES) == 0 {
		return &apiserver.TiDBClusterSlowQueryList{
			Total: int64(utils.GetStartIndex(listOptions.Page, listOptions.ItemsPerPage)),
		}, nil
	}

	r := &apiserver.TiDBClusterSlowQueryList{
		Total: int64(utils.GetStartIndex(listOptions.Page, listOptions.ItemsPerPage) + len(tcInES)),
		Items: make([]apiserver.TiDBClusterSlowQuery, len(tcInES)),
	}

	stream.FromMapEntries(tcInES).Map(func(entry stream.MapEntry) apiserver.TiDBClusterSlowQuery {
		tc := name2TC[entry.Key.(reflect.Value).String()]

		q := apiserver.TiDBClusterSlowQuery{
			Count: entry.Value.(int),
		}
		q.Name = tc.Name
		q.CreationTimestamp = metav1.NewTime(tc.CreationTimestamp.Time)
		q.ClusterName, q.Tags = s.store.Controller().GetNameAndTags(tc)
		q.Status = s.store.StatusMonitor.GetClusterStatus(tc)

		return q

	}).CollectTo(r.Items)

	return r, nil
}