package elasticsearch

import (
	"context"
	"reflect"

	"github.com/juju/errors"
	"gopkg.in/olivere/elastic.v6"
)

const (
	namespaceFilter = "kubernetes.namespace_name"
	podNameFilter   = "kubernetes.pod_name"
	timestampFilter = "@timestamp"
	logStrFilter    = "log"
)

// Config defines the config of elasticsearch service
type Config struct {
	Address string `toml:"address"`
	Index   string `toml:"index"`
}

// Validate checks if the config is valid.
func (c *Config) Validate() error {
	if len(c.Address) == 0 {
		return errors.New("address of elastic is empty")
	}

	if len(c.Index) == 0 {
		return errors.New("index of elastic is empty")
	}

	return nil
}

// Client defines the client of elasticsearch
type Client struct {
	conf *Config

	EsClient *elastic.Client
}

// NewClient returns a instance of elasticsearch client
func NewClient(conf *Config) (*Client, error) {
	if err := conf.Validate(); err != nil {
		return nil, errors.Trace(err)
	}

	cli := &Client{
		conf: conf,
	}

	var err error
	cli.EsClient, err = elastic.NewClient(elastic.SetURL(conf.Address), elastic.SetSniff(false))
	if err != nil {
		return nil, errors.Trace(err)
	}

	return cli, nil
}

// FilterLog filters the logs from elasticsearch,
// returns the total number of the all hit rows and the `size` rows real log data.
// `namespace` parameter defines the pod namespace in kubernetes.
// `podName` parameter defines the pod name, if the `podName` is empty, will filter the whole namespace.
// `filterStr` parameter defines the filter string, like: "region xxx".
// `gte` parameter represents that date is greater than `gte`, like: "now-1d".
// `lte` parameter represents that date is less than `lte`, like: "now".
// `from` parameter defines the offset from the first result you want to fetch, `from` default to 0.
// `size` parameter allows you to configure the maximum amount of hits to be returned, `size` default to 10.
// Note that from + size can not be more than the index.max_result_window index setting which defaults to 10,000.
// date-match refer: https://www.elastic.co/guide/en/elasticsearch/reference/6.7/common-options.html#date-math
func (c *Client) FilterLog(
	namespace string,
	podName string,
	filterStr string,
	gte string,
	lte string,
	from int,
	size int,
) (int64, []*LogRow, error) {
	var filters []elastic.Query

	if len(namespace) > 0 {
		filters = append(filters, elastic.NewTermQuery(namespaceFilter, namespace))
	}

	if len(podName) > 0 {
		// prefix match, we can use this to match all tidbs / pds / tikvs
		filters = append(filters, elastic.NewMatchPhrasePrefixQuery(podNameFilter, podName))
	}

	rangeFilter := elastic.NewRangeQuery(timestampFilter)

	if len(gte) > 0 {
		rangeFilter = rangeFilter.Gte(gte)
	}
	if len(lte) > 0 {
		rangeFilter = rangeFilter.Lte(lte)
	}

	filters = append(filters, rangeFilter)

	q := elastic.NewBoolQuery().Filter(filters...)

	if len(filterStr) > 0 {
		q = q.Must(elastic.NewMatchPhraseQuery(logStrFilter, filterStr))
	}

	res, err := c.EsClient.Search().
		Index(c.conf.Index).
		Query(q).
		FetchSourceContext(elastic.NewFetchSourceContext(true).Include(logStrFilter)).
		Sort(timestampFilter, true).
		From(from).
		Size(size).
		Do(context.Background())
	if err != nil {
		return 0, nil, errors.Trace(err)
	}

	if res.Hits.TotalHits == 0 {
		return 0, nil, nil
	}

	var row LogRow
	var rows []*LogRow
	for _, item := range res.Each(reflect.TypeOf(row)) {
		r, ok := item.(LogRow)
		if !ok {
			return 0, nil, errors.NotValidf("log: %v", item)
		}

		rows = append(rows, &r)
	}

	return res.Hits.TotalHits, rows, nil
}

// LogRow defines the log struct,
// which is used to unmarshal json data from elasticsearch.
type LogRow struct {
	Log string `json:"log"`
}