// Copyright 2021 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package engine

import (
	"encoding/base64"
	"encoding/json"
	"fmt"
	elasticsearch "github.com/opensearch-project/opensearch-go"
	"github.com/pingcap-inc/rule-engine/log/config"
	"io"
	"net/url"
	"strings"
	"time"
)

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

type EngineConfig struct {
	ClusterID      string
	KibanaIndex    string
	ESEndpoint     string
	KibanaEndpoint string
	BeginTime      time.Time
	EndTime        time.Time
	Username       string
	Password       string
}

type EngineOpt func(eng *Engine) error

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

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

func (e *Engine) Execute() ([]Symptom, error) {
	symptoms := []Symptom{}
	for _, rule := range e.rules {
		query, err := e.buildQuery(&rule)
		if err != nil {
			return nil, err
		}
		hit, err := e.isQueryHit(query)
		if err != nil {
			return nil, err
		}
		if hit {
			symptoms = append(symptoms, Symptom{
				ID:     rule.ID,
				Name:   rule.Symptom,
				Kibana: e.buildKibanaLink(&rule),
				KB:     rule.KB,
				Jira:   rule.Jira,
			})
		}
	}
	return symptoms, nil
}

func (e *Engine) isQueryHit(query string) (bool, error) {
	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
	}
	res, err := e.esClient.Search(
		e.esClient.Search.WithIndex(e.config.ClusterID),
		e.esClient.Search.WithBody(strings.NewReader(query)),
		e.esClient.Search.WithTrackTotalHits(true),
		e.esClient.Search.WithHeader(h),
	)
	if err != nil {
		return false, err
	}
	defer func() {
		// req is a POST http request, we need to drain the body before close it.
		_, _ = io.Copy(io.Discard, res.Body)
		_ = res.Body.Close()
	}()

	if res.IsError() {
		return false, fmt.Errorf("elastic search returns error with code %d", res.StatusCode)
	}
	r := struct {
		Hits struct {
			Total struct {
				Value uint `json:"value"`
			} `json:"total"`
		} `json:"hits"`
	}{}
	if err := json.NewDecoder(res.Body).Decode(&r); err != nil {
		return false, err
	}
	return r.Hits.Total.Value > 0, nil
}

func (e *Engine) buildKibanaLink(rule *config.RuleItem) string {
	qs := []string{}
	if rule.Patterns.Component != "" {
		qs = append(qs, fmt.Sprintf("component:%s", rule.Patterns.Component))
	}
	if rule.Patterns.Level != "" {
		qs = append(qs, fmt.Sprintf("level:%s", rule.Patterns.Level))
	}
	for _, msg := range rule.Patterns.Message {
		qs = append(qs, fmt.Sprintf(`message:"%s"`, msg))
	}

	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,
		strings.Join(qs, " and "),
		url.QueryEscape(e.config.BeginTime.Format(time.RFC3339)),
		url.QueryEscape(e.config.EndTime.Format(time.RFC3339)),
	)
}

func (e *Engine) buildQuery(rule *config.RuleItem) (string, error) {
	qs := []map[string]interface{}{}
	if rule.Patterns.Component != "" {
		qs = append(qs, map[string]interface{}{
			"bool": map[string]interface{}{
				"should": []map[string]interface{}{{
					"match_phrase": map[string]interface{}{
						"component": rule.Patterns.Component,
					},
				}},
			},
		})
	}
	if rule.Patterns.Level != "" {
		qs = append(qs, map[string]interface{}{
			"bool": map[string]interface{}{
				"should": []map[string]interface{}{{
					"match_phrase": map[string]interface{}{
						"level": rule.Patterns.Level,
					},
				}},
			},
		})
	}
	for _, m := range rule.Patterns.Message {
		qs = append(qs, map[string]interface{}{
			"bool": map[string]interface{}{
				"should": []map[string]interface{}{{
					"match_phrase": map[string]interface{}{
						"message": m,
					},
				}},
			},
		})
	}
	query := map[string]interface{}{
		"size": 0,
		"query": map[string]interface{}{
			"bool": map[string]interface{}{
				"filter": []map[string]interface{}{{
					"bool": map[string]interface{}{
						"filter": qs,
					},
				}, {
					"range": map[string]interface{}{
						"timestamp": map[string]interface{}{
							"gte": e.config.BeginTime.Format(time.RFC3339),
							"lte": e.config.EndTime.Format(time.RFC3339),
						},
					},
				}},
			},
		},
	}
	q, err := json.Marshal(query)
	if err != nil {
		return "", err
	}
	return string(q), nil
}