package systemlog

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/systemlog"
	"github.com/wushilin/stream"
	"go.uber.org/zap"
	"k8s.io/apimachinery/pkg/api/errors"
)

type Context struct {
	From           int
	Size           int
	Host           string
	StartTimeInSec int64
	EndTimeInSec   int64
}

type SystemLog struct {
	Host string          `json:"host"`
	Msg  string          `json:"message"`
	Time strfmt.DateTime `json:"@timestamp"`
}

func (s *Storage) SystemLogs(ctx Context) (*apiserver.SystemLogList, 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 := systemLogQuery(c, ctx, s.store.Conf.SystemLogIndexPrefix)
	if err != nil {
		log.Error("query system log failed", zap.Error(err))
		return nil, errors.NewInternalError(err)
	}

	return &apiserver.SystemLogList{
		Items: buildSystemLogLine(r),
		Total: r.TotalHits(),
	}, nil
}

func systemLogQuery(c *elasticsearch.Client, ctx Context, indexPrefix string) (*query.QueryResponse, error) {
	sysLogQuery := systemlog.NewSystemQuery(
		systemlog.StartTimeInSec(ctx.StartTimeInSec),
		systemlog.EndTimeInSec(ctx.EndTimeInSec),
		systemlog.Host(ctx.Host),
		systemlog.From(ctx.From),
		systemlog.Size(ctx.Size),
	)

	return sysLogQuery.Do(c, indexPrefix)
}

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

	return lines
}