package tidblog

import (
	"encoding/json"
	"github.com/elastic/go-elasticsearch/v6"
	"github.com/go-openapi/strfmt"
	"github.com/pingcap/log"
	"github.com/pingcap/tep/pkg/apis/apiserver"
	"github.com/pingcap/tep/pkg/utils/es"
	"github.com/pingcap/tep/pkg/utils/es/query"
	"github.com/pingcap/tep/pkg/utils/es/query/tidblog"
	"github.com/wushilin/stream"
	"go.uber.org/zap"
	"k8s.io/apimachinery/pkg/api/errors"
	"strings"
)

type TiDBLog struct {
	Time       strfmt.DateTime `json:"@timestamp"`
	Level      string          `json:"level"`
	Msg        string          `json:"log"`
	Kubernetes Kubernetes      `json:"kubernetes"`
}

type Kubernetes struct {
	PodName string `json:"pod_name"`
	Labels  Labels `json:"labels"`
}

type Labels struct {
	Component string `json:"app_kubernetes_io/component"`
}

type Context struct {
	From           int
	Size           int
	Level          string
	StartTimeInSec int64
	EndTimeInSec   int64
	Namespace      string
	Component      string
}

func (s *Storage) ListLog(ctx Context) (*apiserver.TiDBLogList, 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 := do(c, ctx, s.store.Conf.TiDBLogIndexPrefix)
	if err != nil {
		log.Error("query tidb log failed", zap.Error(err))
		return nil, errors.NewInternalError(err)
	}

	return &apiserver.TiDBLogList{
		Items: buildLogLine(r),
		Total: r.TotalHits(),
	}, nil
}

func do(c *elasticsearch.Client, ctx Context, indexPrefix string) (*query.QueryResponse, error) {
	tidbLog := tidblog.NewTiDBQuery(tidblog.NameSpace(ctx.Namespace),
		tidblog.Component(ctx.Component),
		tidblog.StartTimeInSec(ctx.StartTimeInSec),
		tidblog.EndTimeInSec(ctx.EndTimeInSec),
		tidblog.Level(tidblog.LogLevel(strings.ToUpper(ctx.Level))),
		tidblog.From(ctx.From),
		tidblog.Size(ctx.Size),
	)

	return tidbLog.Do(c, indexPrefix)
}

func buildLogLine(r *query.QueryResponse) []apiserver.TiDBLog {
	list := r.Result()
	lines := make([]apiserver.TiDBLog, len(list))
	stream.FromArray(list).Map(func(msg interface{}) interface{} {
		item := TiDBLog{}
		if err := json.Unmarshal([]byte(msg.(string)), &item); err != nil {
			log.Error("failed to unmarshal es string", zap.Error(err))
		}
		return apiserver.TiDBLog{
			Level:     item.Level,
			Message:   item.Msg,
			Timestamp: item.Time.String(),
			Component: item.Kubernetes.Labels.Component,
			PodName:   item.Kubernetes.PodName,
		}
	}).CollectTo(lines)

	return lines
}