package engine

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"net/url"
	"strings"
	"time"

	"encoding/base64"
	elasticsearch "github.com/opensearch-project/opensearch-go"
	//"github.com/elastic/go-elasticsearch/v7"
	"github.com/pingcap-inc/rule-engine/slowquery/config"
	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"go.uber.org/zap"
)

type Engine struct {
	config   *Config
	esClient *elasticsearch.Client
	rules    []config.RuleItem
	username string
	password string
}

// some must have field
type Config struct {
	ESIndex        string
	KibanaIndex    string
	ESEndpoint     string
	KibanaEndpoint string
	BeginTime      time.Time
	EndTime        time.Time
	Username       string
	Password       string
}

type EngineOpt func(eng *Engine) error

func WithRules(rules []config.RuleItem) EngineOpt {
	return func(eng *Engine) error {
		eng.rules = rules
		return nil
	}
}

func WithBasicAuth(username string, password string) EngineOpt {
	return func(eng *Engine) error {
		eng.username = username
		eng.password = password
		return nil
	}
}

func New(cfg *Config, opts ...EngineOpt) (*Engine, error) {
	es, err := elasticsearch.NewClient(elasticsearch.Config{
		Addresses: []string{cfg.ESEndpoint},
	})
	if err != nil {
		return nil, err
	}
	eng := &Engine{
		config:   cfg,
		esClient: es,
		rules:    make([]config.RuleItem, 0),
		username: cfg.Username,
		password: cfg.Password,
	}
	for _, opt := range opts {
		if err := opt(eng); err != nil {
			return nil, err
		}
	}
	return eng, nil
}

// Execute query es and check if any abnormal case exists.
func (e *Engine) Execute() ([]Symptom, error) {
	symptoms := make([]Symptom, 0)
	for _, rule := range e.rules {
		sqlBkt, err := e.queryAggregation(rule)
		if err != nil {
			return nil, err
		}
		abnormalPlan, err := sqlBkt.GetAbnormalPlan()
		if err != nil {
			return nil, err
		}
		for sqlDigest, plans := range abnormalPlan {
			symptoms = append(symptoms, Symptom{
				RuleID: fmt.Sprintf("%#x", rule.ID),
				Name:   rule.Symptom,
				Kibana: e.buildKibanaLink(sqlDigest, plans),
				KB:     []string{},
				Jira:   []string{},
			})
		}
	}
	return symptoms, nil
}

func (e *Engine) queryAggregation(rule config.RuleItem) (AggregatedPlan, error) {
	query, err := rule.BuildQuery(e.config.BeginTime, e.config.EndTime)
	if err != nil {
		return nil, err
	}
	result := QueryResult{}
	result.Aggregations.SQL = BuildAggregatedPlan(int(rule.ID))
	buf := bytes.NewBufferString(query)
	esc := e.esClient
	h := make(map[string]string)
	if len(e.username) != 0 && len(e.password) != 0 {
		auth := e.username + ":" + e.password
		token := base64.StdEncoding.EncodeToString([]byte(auth))
		h["Authorization"] = "Basic " + token
	}
	resp, err := e.esClient.Search(
		esc.Search.WithContext(context.Background()),
		esc.Search.WithIndex(e.config.ESIndex),
		esc.Search.WithBody(buf),
		esc.Search.WithHeader(h),
	)
	if err != nil {
		log.Error("search es failed", zap.Error(err))
		return nil, err
	}
	defer resp.Body.Close()

	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
		return nil, err
	}
	if resp.IsError() {
		log.Error("search es failed", zap.String("status", resp.Status()), zap.String("reason", result.Error.Reason))
		return nil, fmt.Errorf("es status %v", resp.Status())
	}
	if result.TimeOut {
		return nil, errors.New("es return resp with timeout")
	}
	return result.Aggregations.SQL, nil
}

func (e *Engine) buildKibanaLink(sqlDigest string, planDigests []string) string {
	planDigestFilter := strings.Join(planDigests, " or ")
	if len(planDigests) > 1 {
		planDigestFilter = fmt.Sprintf("(%s)", planDigestFilter)
	}
	boolQuery := fmt.Sprintf("digest:%s and plan_digest:%s", sqlDigest, planDigestFilter)
	// only RFC3339 format time may contain `+` which need to do url encode
	return fmt.Sprintf(
		"%s/app/discover#/?_a=(index:'%s',query:(language:kuery,query:'%s'))&_g=(time:(from:'%s',to:'%s'))",
		e.config.KibanaEndpoint,
		e.config.KibanaIndex,
		boolQuery,
		url.QueryEscape(e.config.BeginTime.Format(time.RFC3339)),
		url.QueryEscape(e.config.EndTime.Format(time.RFC3339)),
	)
}