package auditlog

import (
	"encoding/json"
	"fmt"
	"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/auditlog"
	logstream "github.com/pingcap/tep/pkg/utils/es_stream"
	"github.com/wushilin/stream"
	"go.uber.org/zap"
	"k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"time"
)

type Audit struct {
	Time       strfmt.DateTime `json:"@timestamp"`
	EventClass string          `json:"event"`
	CostTime   float64         `json:"cost_time"`
	Client     string          `json:"client"`
	User       string          `json:"user"`
	Database   string          `json:"db"`
	Table      string          `json:"table"`
	Query      string          `json:"sql"`
	Result     int64           `json:"row"`
}

type TiDBAuditLogCxt struct {
	EventClass     string
	From           int
	Size           int
	StartTimeInSec int64
	EndTimeInSec   int64
	TiDBClusterID  string
	Database       string
	Table          string
	User           string
	OrderBy        string
	Desc           bool
}

func (s *TiDBAuditLogStorage) ListAuditLog(ctx TiDBAuditLogCxt) (*apiserver.TiDBAuditLogList, error) {
	c, err := es.GetClient(s.store.Conf)
	if err != nil {
		return nil, errors.NewInternalError(fmt.Errorf("connect log server failed"))
	}

	r, err := searchAuditLogQuery(c, ctx, s.store.Conf.AuditLogIndexPrefix)
	if err != nil {
		return nil, errors.NewInternalError(fmt.Errorf("search log from es server failed"))
	}

	return &apiserver.TiDBAuditLogList{
		Items: buildAuditLogLine(r),
		Total: r.TotalHits(),
	}, nil

}

func (s *TiDBAuditLogStorage) DownloadAuditLog(ctx TiDBAuditLogCxt) (runtime.Object, error) {
	c, err := es.GetClient(s.store.Conf)
	if err != nil {
		return nil, errors.NewInternalError(fmt.Errorf("connect log server failed"))
	}
	r := buildAuditLogQuery(ctx)
	return logstream.New(c, s.store.Conf.AuditLogIndexPrefix, r.ToQueryString(true), &CSVAuditLog{})
}

func searchAuditLogQuery(c *elasticsearch.Client, ctx TiDBAuditLogCxt, indexPrefix string) (*query.QueryResponse, error) {
	return buildAuditLogQuery(ctx).Do(c, indexPrefix)
}

func buildAuditLogQuery(ctx TiDBAuditLogCxt) *auditlog.AuditQuery {
	return auditlog.NewAuditLogQuery(auditlog.NameSpace(ctx.TiDBClusterID),
		auditlog.StartTimeInSec(ctx.StartTimeInSec),
		auditlog.EndTimeInSec(ctx.EndTimeInSec),
		auditlog.From(ctx.From),
		auditlog.Size(ctx.Size),
		auditlog.OrderBy(ctx.OrderBy),
		auditlog.Database(ctx.Database),
		auditlog.Table(ctx.Table),
		auditlog.User(ctx.User),
		auditlog.Desc(ctx.Desc),
		auditlog.EventClass(ctx.EventClass),
	)
}

func buildAuditLogLine(r *query.QueryResponse) []apiserver.TiDBAuditLog {
	list := r.HitsSlice()
	lines := make([]apiserver.TiDBAuditLog, len(list))
	stream.FromArray(list).Map(func(msg interface{}) interface{} {
		item := Audit{}
		hit := msg.(*query.Hit)
		if err := json.Unmarshal(*hit.Source, &item); err != nil {
			log.Error("failed to unmarshal es string", zap.String("msg", msg.(string)), zap.Error(err))
		}
		return apiserver.TiDBAuditLog{
			ObjectMeta: metav1.ObjectMeta{Name: hit.ID, CreationTimestamp: metav1.NewTime(time.Time(item.Time))},
			EventClass: item.EventClass,
			Timestamp:  item.Time.String(),
			SQL:        item.Query,
			User:       item.User,
			TableName:  item.Table,
			DbName:     item.Database,
			Cost:       item.CostTime / 1000,
			Result:     item.Result,
			SourceIP:   item.Client,
		}
	}).CollectTo(lines)

	return lines
}

type CSVAuditLog struct {
}

func (audit *CSVAuditLog) Header() []string {
	return []string{"time", "event class", "user", "database", "table", "client IP", "SQL", "cost", "rows"}
}

func (audit *CSVAuditLog) ToCSV(r *query.QueryResponse) [][]string {
	lines := buildAuditLogLine(r)
	var output [][]string
	for _, item := range lines {
		output = append(output, []string{fmt.Sprintf("%v", item.Timestamp), item.EventClass, item.User, item.DbName, item.TableName, item.SourceIP, item.SQL, fmt.Sprintf("%d", item.Cost), fmt.Sprintf("%d", item.Result)})

	}
	return output
}