package slowlog

import (
	"encoding/json"
	"github.com/pingcap/log"
	"github.com/pingcap/tep/pkg/apis/apiserver"
	"go.uber.org/zap"
	"k8s.io/apimachinery/pkg/api/errors"

	"github.com/elastic/go-elasticsearch/v6"
	"github.com/go-openapi/strfmt"
	"github.com/pingcap/tep/pkg/utils/es"
	"github.com/pingcap/tep/pkg/utils/es/query"
	"github.com/pingcap/tep/pkg/utils/es/query/slowlog"
	"github.com/wushilin/stream"
)

type SlowLog struct {
	Time    strfmt.DateTime `json:"@timestamp"`
	Message string          `json:"log"`
}

type Context struct {
	From           int
	Size           int
	StartTimeInSec int64
	EndTimeInSec   int64
	Namespace      string
	Items          []interface{}
}

func (s *TiDBSlowLogStorage) ListSlowLog(ctx Context) (*apiserver.TiDBSlowLogList, error) {
	c, err := es.GetClient(s.store.Conf)
	if err != nil {
		log.Error("get es client failed", zap.Error(err))
		return nil, errors.NewInternalError(err)
	}

	r, err := searchSlowQueryLog(c, ctx, s.store.Conf.SlowLogIndexPrefix)
	if err != nil {
		log.Error("query slow log failed", zap.Error(err))
		return nil, errors.NewInternalError(err)
	}

	return &apiserver.TiDBSlowLogList{
		Items: buildSlowLogLine(r),
		Total: r.TotalHits(),
	}, nil
}

func searchSlowQueryLog(c *elasticsearch.Client, ctx Context, indexPrefix string) (*query.QueryResponse, error) {
	tidbLog := slowlog.NewSlowQuery(slowlog.NameSpace(ctx.Namespace),
		slowlog.StartTimeInSec(ctx.StartTimeInSec),
		slowlog.EndTimeInSec(ctx.EndTimeInSec),
		slowlog.From(ctx.From),
		slowlog.Size(ctx.Size),
	)

	return tidbLog.Do(c, indexPrefix)
}

func buildSlowLogLine(r *query.QueryResponse) []apiserver.TiDBSlowLog {
	list := r.Result()
	lines := make([]apiserver.TiDBSlowLog, len(list))
	stream.FromArray(list).Map(func(msg interface{}) interface{} {
		item := SlowLog{}
		if err := json.Unmarshal([]byte(msg.(string)), &item); err != nil {
			log.Error("failed to unmarshal log item", zap.Error(err))
		}
		return apiserver.TiDBSlowLog{
			Timestamp: item.Time.String(),
			Log:       item.Message,
		}
	}).CollectTo(lines)

	return lines
}