package importer

import (
	"archive/tar"
	"fmt"
	"io"
	"os"
	"strings"
	"testing"
	"time"

	"github.com/prometheus/common/model"
	"go.uber.org/atomic"

	"github.com/pingcap-inc/clinic/pkg/cache"
	"github.com/pingcap-inc/clinic/pkg/config"
	_const "github.com/pingcap-inc/clinic/pkg/const"
	"github.com/pingcap-inc/clinic/pkg/es"
	executor2 "github.com/pingcap-inc/clinic/pkg/executor"
	log2 "github.com/pingcap-inc/clinic/pkg/importer/log"
	metrics2 "github.com/pingcap-inc/clinic/pkg/importer/metrics"

	"github.com/onsi/gomega"
	"github.com/pingcap/log"
	"go.uber.org/zap"

	"github.com/pingcap-inc/clinic/pkg/metrics"

	dto "github.com/prometheus/client_model/go"
)

const (
	timeout  = time.Second * 120
	interval = time.Second * 1
)

func Test_ImportWithTarAsync(t *testing.T) {
	clusterID := "123124421"
	orgID := uint(1)

	cfg := config.LogConfig{
		Endpoint: fmt.Sprintf("http://%s:%d", "127.0.0.1", es.EsPort),
		User:     "admin",
		Password: "admin",
		KibanaConfig: config.KibanaConfig{
			AddressConfig: config.AddressConfig{
				Host: "127.0.0.1",
				Port: es.KibanaPort,
			},
		},
	}

	im := NewImporter(&ImporterOption{
		MetricConfig: config.MetricConfig{
			VMClusterAddressConfig: config.VMClusterAddressConfig{
				AddressConfig: config.AddressConfig{
					Host: "127.0.0.1",
					Port: metrics2.VmInsert,
				},
			},
		},
		LogConfig:                cfg,
		FileUUID:                 "8df9e3075057432-2be068847b212ee7-a71cd20639ea6108-a71cd20639ea6108",
		ClusterID:                clusterID,
		DataPath:                 dataPath,
		OrgID:                    orgID,
		localFilePath:            "testdata/test.tar.zst",
		ExpectedRebuildTaskTypes: _const.AllTaskType(),
		BeginTS:                  1635000000000,
		EndTS:                    1636000000000,
	})

	g := gomega.NewWithT(t)

	g.Expect(im.Rebuild()).To(gomega.Succeed())

	g.Eventually(func() bool {
		var m dto.Metric
		err := metrics.ImportMetricFileProcessSuccessfulCount.WithLabelValues(clusterID, "8df9e3075057432-2be068847b212ee7-a71cd20639ea6108-a71cd20639ea6108").Write(&m)
		if err != nil {
			log.Error("write metric failed", zap.Error(err))
			return false
		}

		return int(*m.Gauge.Value) >= 2
	}, timeout, interval).Should(gomega.BeTrue())

	insertCount := 2
	r := log2.QueryESEventually(&log2.ESQueryOption{
		ClusterID:  clusterID,
		Host:       "172.16.5.137",
		Component:  "tidb",
		StartTime:  1635240046000,
		EndTime:    1635240346000,
		Level:      "info",
		InstanceID: "tidb-4000",
		OrgID:      orgID,
	}, g, insertCount)

	actualLogs := make([]string, insertCount)
	for i := 0; i < insertCount; i++ {
		actualLogs[i] = r[i].Message
	}

	g.Expect(actualLogs).To(gomega.ContainElements("[main.go:336] [\"disable Prometheus push client\"]"))
	g.Expect(actualLogs).To(gomega.ContainElements("[store.go:68] [\"new store\"] [path=tikv://172.16.5.137:2379,172.16.5.208:2379,172.16.5.220:2379]"))
}

func Test_importMetrics(t *testing.T) {
	g := gomega.NewWithT(t)

	filepath := "Test_ImportMetrics"
	clusterID := "2222"
	orgID := uint(2234)
	importCount := 19

	timestamp := time.Now().Unix() - 11*60

	importer := NewImporter(&ImporterOption{
		MetricConfig: config.MetricConfig{
			VMClusterAddressConfig: config.VMClusterAddressConfig{
				AddressConfig: config.AddressConfig{
					Host: "127.0.0.1",
					Port: metrics2.VmInsert,
				},
			},
			SelectAddr: config.MetricSelectAddressConfig{
				OpAddress: config.VMClusterAddressConfig{
					AddressConfig: config.AddressConfig{
						Host: "127.0.0.1",
						Port: metrics2.VmSelect,
					},
				},
			},
		},

		DataPath:                 dataPath,
		FileUUID:                 "8df9e3075057432-2be068847b212ee7-a71cd20639ea6108-a71cd20639ea6108",
		ClusterID:                clusterID,
		ExpectedRebuildTaskTypes: []_const.TaskType{_const.MetricTask},
		OrgID:                    orgID,
		BeginTS:                  1635000000000,
		EndTS:                    1636000000000,
		// MetricHost: "127.0.0.1",
		// MetricPort: metrics2.VmPort,
	})
	// importer.metricClusterModel = false

	for i := 0; i < importCount; i++ {
		path, err := metrics2.NewTar(fmt.Sprintf("%s%d", filepath, i), timestamp, true, "1.1.1.1-9090")
		g.Expect(err).To(gomega.Succeed())

		importer.opt.localFilePath = path

		g.Expect(importer.Rebuild()).To(gomega.Succeed())
		// _ = os.RemoveAll(dataPath)
		_ = os.Remove(path)
	}

	g.Eventually(func() bool {
		var m dto.Metric
		err := metrics.ImportMetricFileProcessSuccessfulCount.WithLabelValues(clusterID, "8df9e3075057432-2be068847b212ee7-a71cd20639ea6108-a71cd20639ea6108").Write(&m)
		if err != nil {
			log.Error("write metric failed", zap.Error(err))
			return false
		}

		return int(*m.Gauge.Value) >= importCount
	}, timeout, interval).Should(gomega.BeTrue())

	//g.Eventually(func() bool{
	//	return mockReaderClosedCount.Load() >= int32(importCount)
	//}, timeout, interval).Should(gomega.BeTrue())

	for i := 0; i < importCount; i++ {
		result := metrics2.QueryEventually(importer.opt.MetricConfig.SelectAddr.OpAddress.Host, importer.opt.MetricConfig.SelectAddr.OpAddress.Port, metrics2.QueryOption{
			Query: fmt.Sprintf(`{filepath="%s%d"}`, filepath, i),
		}, orgID, g)

		g.Expect(string(result[0].Metric[model.LabelName(metrics2.ClusterLabel)])).To(gomega.Equal(clusterID))
		g.Expect(strings.HasPrefix(string(result[0].Metric[model.LabelName("instance")]), filepath)).To(gomega.BeTrue())
		g.Expect(string(result[0].Metric[model.LabelName(metrics2.HostLabel)])).To(gomega.Equal("1.1.1.1"))
		g.Expect(len(result[0].Values) > 0).To(gomega.BeTrue())
		g.Expect(float64(result[0].Values[0].Value) < 1000).To(gomega.BeTrue())
		g.Expect(float64(result[0].Values[0].Value) >= 0).To(gomega.BeTrue())
		g.Expect(int64(result[0].Values[0].Timestamp)).To(gomega.Equal(timestamp * 1000))
	}
}

type MockSortingBaggingTask struct {
	types []_const.TaskType

	// waitMill int
	err error
	// count    int
	expected *atomic.Int32
	task     *MockTask
}

func (msbt *MockSortingBaggingTask) SortingTypes() []_const.TaskType {
	return msbt.types
}

func (msbt *MockSortingBaggingTask) Sorting(fileName string, fh *tar.Header) (_const.TaskType, bool) {
	msbt.expected.Inc()
	return msbt.types[0], true
}

func (msbt *MockSortingBaggingTask) Bagging(fileName string, fh *tar.Header, reader io.Reader) (_const.TaskType, executor2.Task, error) {
	// msbt.waitMill = msbt.waitMill + 50
	return msbt.types[0], msbt.task, nil
}

func (msbt *MockSortingBaggingTask) Init() error {
	return msbt.err
}

func (msbt *MockSortingBaggingTask) Close() {
}

type MockTask struct {
	// waitMill *atomic.Int32
	count *atomic.Int32

	fileUUID  string
	dtype     _const.TaskType
	processed *atomic.Int32
}

func NewMockTest(fileUUID string, dtype _const.TaskType) *MockTask {
	return &MockTask{
		count:     atomic.NewInt32(0),
		processed: atomic.NewInt32(0),
		dtype:     dtype,
		fileUUID:  fileUUID,
	}
}

func (mt *MockTask) Run() {
	time.Sleep(time.Duration(50))

	cache.TaskStatusCache.Done(cache.TaskID(mt.fileUUID, mt.dtype), 1)
	processed, ok := cache.TaskStatusCache.Processed(cache.TaskID(mt.fileUUID, mt.dtype))
	if ok {
		mt.processed.Store(int32(processed))
	}

	mt.count.Inc()
}

//func Test_FastRunner(t *testing.T) {
//	rt := &rebuildTask{
//		needToRebuildTask: map[_const.TaskType]struct{}{
//			_const.MetricTask:      {},
//			_const.RebuildConfTask: {},
//		},
//
//		RebuildTaskOption: &RebuildTaskOption{
//			FileUUID: "Test_FastRunner",
//			TarPath:  "./testdata/test.tar",
//		},
//	}
//
//	g := gomega.NewWithT(t)
//
//	exec := executor2.NewExecutor(2)
//	sortingBagging := executor2.NewExecutor(10)
//
//	msbt := &MockSortingBaggingTask{
//		expected: atomic.NewInt32(0),
//		task:     NewMockTest(rt.FileUUID, _const.MetricTask),
//		types:    []_const.TaskType{_const.MetricTask, _const.RebuildConfTask},
//		err:      nil,
//	}
//
//	sortingBaggingTask := func(sbt *SortingTable, err error) {
//		if err != nil {
//			log.Error("init fast runner failed", zap.String("uuid", rt.FileUUID), zap.Error(err))
//			return
//		}
//
//		if sbt != nil {
//			exec.Submit(rt.FileUUID, sbt)
//		}
//	}
//
//	sortingBaggingTask(NewSortingTable(sortingBagging, msbt, rt))
//	//sortingBaggingTask(NewSortingTable(sortingBagging, msbt, rt))
//
//	exec.Wait(rt.FileUUID)
//
//	g.Expect(msbt.task.processed.Load() > 0).To(gomega.BeTrue())
//	g.Expect(msbt.task.count.Load() > 0).To(gomega.BeTrue())
//	g.Expect(msbt.expected.Load()).To(gomega.Equal(msbt.task.count.Load()))
//	//g := gomega.NewWithT(t)
//	//
//	//fr, err := NewFastRunner(&rebuildTask{
//	//	importer: NewImporter(ImporterOption{
//	//		MetricConfig: config.MetricConfig{},
//	//	}),
//	//})
//	//g.Expect(err).To(gomega.Succeed())
//
//	//fr.Submit()
//}

//type SortingBaggingTask interface {
//	SortingTypes() []_const.TaskType
//	GetTaskType(fileName string, fh *tar.Header) (_const.TaskType, bool)
//	Sorting() bool
//	Bagging() (executor.Task, error)
//	Init() error
//	Close()
//}