// 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 (
	"fmt"
	"net/http"
	"strings"
	"testing"
	"time"

	"github.com/elastic/go-elasticsearch/v7/esutil"
	"github.com/jarcoal/httpmock"
	"github.com/pingcap-inc/rule-engine/slowquery/config"
	"github.com/pingcap-inc/rule-engine/utils/docker"

	es "github.com/elastic/go-elasticsearch/v7"
	"github.com/stretchr/testify/require"
)

func TestKibanaLink(t *testing.T) {
	assert := require.New(t)

	eg, err := New(&Config{
		ESIndex:        "1e6f6b90-475a-11ec-bd35-a5f58f7843c2",
		KibanaIndex:    "1e6f6b90-475a-11ec-bd35-a5f58f7843c2",
		ESEndpoint:     "http://mock.es.com",
		KibanaEndpoint: "http://kibana-209e950d6bc28fd0.elb.cn-north-1.amazonaws.com.cn:5601",
		BeginTime:      time.Date(2021, 11, 17, 0, 0, 0, 0, time.UTC),
		EndTime:        time.Date(2021, 11, 18, 0, 0, 0, 0, time.UTC),
	})
	assert.Nil(err)
	{
		link := eg.buildKibanaLink("a", []string{"A", "B"})
		assert.Equal(`http://kibana-209e950d6bc28fd0.elb.cn-north-1.amazonaws.com.cn:5601/app/discover#/?_a=(index:'1e6f6b90-475a-11ec-bd35-a5f58f7843c2',query:(language:kuery,query:'digest:a and plan_digest:(A or B)'))&_g=(time:(from:'2021-11-17T00%3A00%3A00Z',to:'2021-11-18T00%3A00%3A00Z'))`, link)
	}
	{
		link := eg.buildKibanaLink("a", []string{"A"})
		assert.Equal(`http://kibana-209e950d6bc28fd0.elb.cn-north-1.amazonaws.com.cn:5601/app/discover#/?_a=(index:'1e6f6b90-475a-11ec-bd35-a5f58f7843c2',query:(language:kuery,query:'digest:a and plan_digest:A'))&_g=(time:(from:'2021-11-17T00%3A00%3A00Z',to:'2021-11-18T00%3A00%3A00Z'))`, link)
	}
}

var slowQueryLogMapping = `{
  "mappings": {
    "properties": {
      "level": {
        "type": "text",
        "fields": {
          "keyword": {"type": "keyword","ignore_above": 256}
        }
      },
      "message": {
        "type": "text",
        "fields": {
          "keyword": {"type": "keyword", "ignore_above": 256}
        }
      },
      "digest": {
        "type": "keyword",
        "fields": {
          "keyword": {"type": "keyword","ignore_above": 512}
        }
      },
      "plan_digest": {
        "type": "keyword",
        "fields": {
          "keyword": {"type": "keyword","ignore_above": 512}
        }
      },
      "process_keys": {
        "type": "long"
      },
      "process_time": {
        "type": "float"
      },
      "total_keys": {
        "type": "long"
      },
      "rocksdb_delete_skipped_count": {
        "type": "long"
      },
      "component": {
        "type": "keyword"
      },
      "type": {
        "type": "keyword"
      },
      "host": {
        "type": "keyword"
      },
      "instanceid": {
        "type": "keyword"
      },
      "packageid": {
        "type": "keyword"
      },
      "timestamp": {
        "type": "date"
      }
    }
  }
}`

type SlowQueryDoc struct {
	Timestamp                 int64   `json:"timestamp"`
	Message                   string  `json:"message"`
	Digest                    string  `json:"digest"`
	PlanDigest                string  `json:"plan_digest"`
	ProcessKeys               int64   `json:"process_keys"`
	ProcessTime               float64 `json:"process_time"`
	TotalKeys                 int64   `json:"total_keys"`
	RocksdbDeleteSkippedCount int64   `json:"rocksdb_delete_skipped_count"`
	Host                      string  `json:"host"`
	Component                 string  `json:"component"`
	PackageID                 string  `json:"packageid"`
	InstanceID                string  `json:"instanceid"`
	LogType                   string  `json:"type"`
}

func TestEngine_Execute(t *testing.T) {
	assert := require.New(t)
	port, err := docker.StartES()
	assert.Nil(err)
	defer docker.CleanESKibanaDocker()
	esEndpoint := fmt.Sprintf("http://127.0.0.1:%d", port)
	esCli, err := es.NewClient(es.Config{
		RetryOnStatus:        []int{502, 503, 504, 429},
		Addresses:            []string{esEndpoint},
		EnableRetryOnTimeout: true,
		MaxRetries:           3,
	})
	assert.Nil(err)
	esIndex := "test-slow-query"
	resp, err := esCli.Indices.Delete([]string{esIndex}, esCli.Indices.Delete.WithIgnoreUnavailable(true))
	// prepare es data
	assert.Nil(err)
	assert.False(resp.IsError())
	resp, err = esCli.Indices.Create(
		esIndex,
		esCli.Indices.Create.WithBody(strings.NewReader(slowQueryLogMapping)),
		esCli.Indices.Create.WithWaitForActiveShards("1"))
	assert.Nil(err)
	assert.False(resp.IsError())
	docs := []SlowQueryDoc{
		{Timestamp: 1566441991000, Message: "xxx", Digest: "a", PlanDigest: "a", ProcessKeys: 1000001, ProcessTime: 10, TotalKeys: 1000, RocksdbDeleteSkippedCount: 10, Component: "tidb", Host: "127.0.0.1", LogType: "slowquery"},
		{Timestamp: 1566442051000, Message: "xxx", Digest: "a", PlanDigest: "a", ProcessKeys: 1000002, ProcessTime: 20, TotalKeys: 2000, RocksdbDeleteSkippedCount: 10, Component: "tidb", Host: "127.0.0.1", LogType: "slowquery"},
		{Timestamp: 1566442111000, Message: "xxx", Digest: "a", PlanDigest: "b", ProcessKeys: 1000003, ProcessTime: 5000, TotalKeys: 3000, RocksdbDeleteSkippedCount: 10, Component: "tidb", Host: "127.0.0.1", LogType: "slowquery"},
		{Timestamp: 1566442171000, Message: "xxx", Digest: "a", PlanDigest: "c", ProcessKeys: 1000004, ProcessTime: 200, TotalKeys: 4000, RocksdbDeleteSkippedCount: 10, Component: "tidb", Host: "127.0.0.1", LogType: "slowquery"},
		{Timestamp: 1566442231000, Message: "xxx", Digest: "b", PlanDigest: "a", ProcessKeys: 10010, ProcessTime: 100, TotalKeys: 15, RocksdbDeleteSkippedCount: 0, Component: "tidb", Host: "127.0.0.1", LogType: "slowquery"},
		{Timestamp: 1566442291000, Message: "xxx", Digest: "b", PlanDigest: "b", ProcessKeys: 10010, ProcessTime: 100, TotalKeys: 100100, RocksdbDeleteSkippedCount: 0, Component: "tidb", Host: "127.0.0.1", LogType: "slowquery"},
		{Timestamp: 1566442301000, Message: "xxx", Digest: "b", PlanDigest: "c", ProcessKeys: 10, ProcessTime: 100, TotalKeys: 5, RocksdbDeleteSkippedCount: 0, Component: "tidb", Host: "127.0.0.1", LogType: "slowquery"},
		{Timestamp: 1566442401000, Message: "xxx", Digest: "b", PlanDigest: "a", ProcessKeys: 100, ProcessTime: 100, TotalKeys: 100, RocksdbDeleteSkippedCount: 0, Component: "tidb", Host: "127.0.0.1", LogType: "slowquery"},
		{Timestamp: 1566442411000, Message: "xxx", Digest: "c", PlanDigest: "a", ProcessKeys: 110000, ProcessTime: 0, TotalKeys: 0, RocksdbDeleteSkippedCount: 510000, Component: "tidb", Host: "127.0.0.1", LogType: "slowquery"},
		{Timestamp: 1566442451000, Message: "xxx", Digest: "c", PlanDigest: "b", ProcessKeys: 110000, ProcessTime: 0, TotalKeys: 0, RocksdbDeleteSkippedCount: 110050, Component: "tidb", Host: "127.0.0.1", LogType: "slowquery"},
	}
	for _, doc := range docs {
		resp, err := esCli.Index(esIndex, esutil.NewJSONReader(&doc), esCli.Index.WithRefresh("true"))
		assert.Nil(err)
		assert.False(resp.IsError())
	}
	// execute engine
	spec, err := config.LoadBetaRuleSpec()
	assert.Nil(err)
	testLoc := time.FixedZone("UTC+8", 8*60*60)
	eng, err := New(&Config{
		ESIndex:        esIndex,
		KibanaIndex:    esIndex,
		ESEndpoint:     esEndpoint,
		KibanaEndpoint: "http://kibana-209e950d6bc28fd0.elb.cn-north-1.amazonaws.com.cn:5601",
		BeginTime:      time.Unix(1566403200, 0).In(testLoc),
		EndTime:        time.Unix(1566446400, 0).In(testLoc),
	}, WithRules(spec.Rule))
	assert.Nil(err)
	syms, err := eng.Execute()
	assert.Nil(err)
	// check result
	for _, sym := range syms {
		assert.Contains(sym.Kibana, "_g=(time:(from:'2019-08-22T00%3A00%3A00%2B08%3A00',to:'2019-08-22T12%3A00%3A00%2B08%3A00'))")
		switch sym.RuleID {
		case "0x4001":
			assert.Contains(sym.Kibana, "_a=(index:'test-slow-query',query:(language:kuery,query:'digest:a and plan_digest:(b or a)'))")
		case "0x4002":
			assert.Contains(sym.Kibana, "_a=(index:'test-slow-query',query:(language:kuery,query:'digest:b and plan_digest:b'))")
		case "0x4003":
			assert.Contains(sym.Kibana, "_a=(index:'test-slow-query',query:(language:kuery,query:'digest:c and plan_digest:b'))")
		}
	}
}

func TestBasicAuth(t *testing.T) {
	assert := require.New(t)
	httpmock.Activate()
	defer httpmock.DeactivateAndReset()

	mockESEndpoint := "http://mock.com:9200"

	httpmock.RegisterResponder(http.MethodPost, fmt.Sprintf("%s/clinic/_search", mockESEndpoint),
		func(req *http.Request) (*http.Response, error) {
			if username, password, ok := req.BasicAuth(); !ok {
				return httpmock.NewJsonResponse(http.StatusUnauthorized, struct{}{})
			} else if username != "root" || password != "root" {
				return httpmock.NewJsonResponse(http.StatusUnauthorized, struct{}{})
			}
			resp, err := httpmock.NewJsonResponse(200, struct{}{})
			if err != nil {
				return httpmock.NewStringResponse(500, ""), nil
			}
			return resp, nil
		},
	)
	httpmock.RegisterResponder(http.MethodGet, "/", func(request *http.Request) (*http.Response, error) {
		resp, _ := httpmock.NewJsonResponse(http.StatusOK, struct{}{})
		return resp, nil
	})
	cfg := Config{
		ESIndex:    "clinic",
		ESEndpoint: mockESEndpoint,
		BeginTime:  time.Now(),
		EndTime:    time.Now(),
	}
	eng, err := New(&cfg, WithBasicAuth("root", "root"))
	assert.Nil(err)
	rules, err := config.LoadBetaRuleSpec()
	assert.Nil(err)
	_, err = eng.queryAggregation(rules.Rule[0])
	assert.Nil(err)
}