package auditlog

import (
	"context"
	"errors"
	"fmt"
	"io/ioutil"
	"strings"
	"time"

	"github.com/elastic/go-elasticsearch/v6"
	"github.com/pingcap/log"
	query2 "github.com/pingcap/tep/pkg/utils/es/query"
	"github.com/wushilin/stream"
	"go.uber.org/zap"
)

type AuditQuery struct {
	eventClass string
	from       int
	size       int
	startTime  string
	endTime    string
	namespace  string
	database   string
	table      string
	user       string
	orderBy    string
	desc       bool
}

type query func(q *AuditQuery)

func NewAuditLogQuery(q ...query) *AuditQuery {
	tidbQ := &AuditQuery{}
	stream.FromArray(q).Each(func(f interface{}) {
		f.(query)(tidbQ)
	})

	return tidbQ
}

func EventClass(eventClass string) query {
	return func(q *AuditQuery) {
		q.eventClass = eventClass
	}
}

func From(from int) query {
	if from < 0 {
		from = 0
	}

	return func(q *AuditQuery) {
		q.from = from
	}
}

func Size(size int) query {
	if size <= 0 {
		size = 10
	}

	return func(q *AuditQuery) {
		q.size = size
	}
}

func StartTimeInSec(startTimeInSec int64) query {
	if startTimeInSec <= 0 {
		startTimeInSec = time.Now().Unix() - 24*60*60
	}

	return func(q *AuditQuery) {
		q.startTime = query2.ConvertTimeStamp(startTimeInSec)
	}
}

func EndTimeInSec(endTimeInSec int64) query {
	if endTimeInSec <= 0 {
		endTimeInSec = time.Now().Unix()
	}

	return func(q *AuditQuery) {
		q.endTime = query2.ConvertTimeStamp(endTimeInSec)
	}
}

func NameSpace(namespace string) query {
	return func(q *AuditQuery) {
		q.namespace = namespace
	}
}

func Database(database string) query {
	return func(q *AuditQuery) {
		q.database = database
	}
}

func Table(table string) query {
	return func(q *AuditQuery) {
		q.table = table
	}
}

func User(user string) query {
	return func(q *AuditQuery) {
		q.user = user
	}
}

func OrderBy(orderBy string) query {
	if orderBy == "" || orderBy == "timestamp" {
		orderBy = "@timestamp"
	}
	return func(q *AuditQuery) {
		q.orderBy = orderBy
	}
}

func Desc(desc bool) query {
	return func(q *AuditQuery) {
		q.desc = desc
	}
}

func (q *AuditQuery) Do(client *elasticsearch.Client, indexPrefix string) (*query2.QueryResponse, error) {
	query := q.ToQueryString(false)
	////currently, the index name is not made
	res, err := client.Search(
		client.Search.WithContext(context.Background()),
		client.Search.WithIndex(fmt.Sprintf("%s*", indexPrefix)),
		client.Search.WithBody(strings.NewReader(query)),
		client.Search.WithPretty(),
	)

	if err != nil {
		log.Error("Query from elasticsearch failed", zap.String("query", query), zap.Error(err))
		return nil, err
	}
	defer res.Body.Close()
	if res.StatusCode != 200 {
		buf, _ := ioutil.ReadAll(res.Body)
		strErr := string(buf)
		log.Error("query from es failed", zap.String("query", query), zap.String("err", strErr))
		return nil, errors.New(strErr)
	}

	return query2.NewResponse(res.Body), nil
}

func (q *AuditQuery) Scan(client *elasticsearch.Client, indexPrefix string, consume func(*query2.QueryResponse)) error {
	query := q.ToQueryString(true)
	res, err := client.Search(
		client.Search.WithContext(context.Background()),
		client.Search.WithIndex(fmt.Sprintf("%s*", indexPrefix)),
		client.Search.WithBody(strings.NewReader(query)),
		client.Search.WithPretty(),
		client.Search.WithScroll(time.Duration(time.Minute)),
	)
	if err != nil {
		log.Error("Query from elasticsearch failed", zap.String("query", query), zap.Error(err))
		return err
	}
	searchResponse := query2.NewResponse(res.Body)
	res.Body.Close()
	consume(searchResponse)
	for len(searchResponse.Result()) > 0 && searchResponse.ScrollID != "" {
		res, err := client.Scroll(
			client.Scroll.WithContext(context.Background()),
			client.Scroll.WithPretty(),
			client.Scroll.WithScroll(time.Duration(time.Minute)),
			client.Scroll.WithScrollID(searchResponse.ScrollID),
		)
		if err != nil {
			log.Error("get scroll result from elasticsearch failed", zap.String("query", query), zap.Error(err))
			break
		}
		searchResponse = query2.NewResponse(res.Body)
		res.Body.Close()
		consume(searchResponse)
	}
	return nil
}

func (q *AuditQuery) ToQueryString(scroll bool) string {
	builder := query2.NewQueryBuilder().
		AddEqual("kubernetes.namespace_name", q.namespace).
		AddRange("@timestamp", q.startTime, q.endTime)
	if q.user != "*" {
		builder.AddEqual("user", strings.ToLower(q.user))
	}
	if q.database != "*" {
		builder.AddEqual("db", strings.ToLower(q.database))
	}
	if q.table != "*" {
		builder.AddEqual("table", strings.ToLower(q.table))
	}
	if q.eventClass != "*" {
		builder.AddEqual("event", strings.ToLower(q.eventClass))
	}

	if scroll {
		builder = builder.Sort("_doc", false).From(0).Size(1000)
	} else {
		builder = builder.Sort(q.orderBy, q.desc).
			From(q.from).
			Size(q.size)
	}
	return builder.BuildQuery()
}