package metrics

import (
	"bytes"
	"fmt"
	"io"
	"os"
	"path"
	"strings"

	"github.com/DataDog/zstd"
	"github.com/pingcap/log"
	"go.uber.org/zap"

	"github.com/pingcap-inc/clinic/pkg/cache"
	_const "github.com/pingcap-inc/clinic/pkg/const"
	"github.com/pingcap-inc/clinic/pkg/metrics"
	"github.com/pingcap-inc/clinic/pkg/stream"
)

const (
	metricBatchSize = 30

	SubMonitorDir = "monitor/metrics"

	TDBSuffix = "monitor/raw"
)

func IsMetricFile(fileName, path string) bool {
	if strings.HasSuffix(fileName, "json") && !strings.Contains(fileName, "ALERTS") && strings.Contains(path, SubMonitorDir) {
		return true
	}

	if strings.Contains(path, TDBSuffix) {
		return true
	}

	return false
}

type MetricTask struct {
	//data        []byte
	//status    *status.Status
	//filePath  string
	//
	//clusterID uint64
	*TaskOption
	hostName string
	// path      string
	sink *MetricSink

	// fileUUID string
}

type TaskOption struct {
	// Status    *status.Status
	FilePath string
	FileSize int64
	FileUUID string

	OriginalPathInTar string

	ClusterID string
	// for vmctl to process tsdb blocks
	ProcessAsTSDB bool
	TSDBDirPath   string
}

func NewMetricTask(sink *MetricSink, opt *TaskOption) *MetricTask {
	hostName := getHostName(opt.OriginalPathInTar)

	return &MetricTask{
		TaskOption: opt,
		hostName:   hostName,
		sink:       sink,
	}
}

func (mt *MetricTask) Run() {
	defer func() {
		cache.TaskStatusCache.Done(cache.TaskID(mt.FileUUID, _const.MetricTask), mt.FileSize)
		log.Info("complete to insert metric", zap.String("fileUUID", mt.FileUUID), zap.String("cluster", mt.ClusterID), zap.String("filepath", mt.OriginalPathInTar))
	}()
	if mt.ProcessAsTSDB {
		log.Debug("process tsdb block",
			zap.String("fileUUID", mt.FileUUID),
			zap.String("cluster", mt.ClusterID),
			zap.String("tsdbDir", mt.TSDBDirPath),
			zap.Int64("fileSize", mt.FileSize))
		if err := mt.sink.SinkTSDBVmCtl(mt.TSDBDirPath, []string{fmt.Sprintf("tidb_cluster=%s", mt.ClusterID)}); err != nil {
			metrics.ImportMetricFileProcessFailedCount.WithLabelValues(mt.ClusterID, mt.FileUUID).Inc()
			log.Error("sink tsdb dir with vmctl failed",
				zap.String("fileUUID", mt.FileUUID),
				zap.String("pathInTar", mt.OriginalPathInTar),
				zap.String("tsdbDir", mt.TSDBDirPath),
				zap.String("cluster", mt.ClusterID),
				zap.Error(err))
		} else {
			metrics.ImportMetricFileProcessSuccessfulCount.WithLabelValues(mt.ClusterID, mt.FileUUID).Inc()
		}
		return
	}

	f, err := os.Open(mt.FilePath)
	if err != nil {
		log.Error("open tmp file failed", zap.String("fileUUID", mt.FileUUID), zap.String("cluster", mt.ClusterID), zap.String("filepath", mt.OriginalPathInTar), zap.Error(err))
		return
	}
	defer func() {
		_ = f.Close()
	}()

	var r io.ReadCloser = f
	header := make([]byte, 4)
	_, _ = f.Read(header)
	_, _ = f.Seek(0, io.SeekStart)
	// header of zstd, https://github.com/facebook/zstd/blob/dev/doc/zstd_compression_format.md
	if bytes.Equal(header, []byte{0x28, 0xb5, 0x2f, 0xfd}) {
		r = zstd.NewReader(f)
		defer r.Close()
	}

	// TODO all the following metrics is not correct when process tsdb block via vmctl
	metrics.ImportMetricFileCount.WithLabelValues(mt.ClusterID, mt.FileUUID).Inc()

	iter, err := NewMetricIterator(r, mt.OriginalPathInTar, mt.ClusterID, mt.hostName, mt.FileUUID)
	if err != nil {
		log.Error("create metric iterator failed", zap.String("cluster", mt.ClusterID), zap.String("path", mt.OriginalPathInTar), zap.Error(err))
		metrics.ImportMetricFileProcessFailedCount.WithLabelValues(mt.ClusterID, mt.FileUUID).Inc()
		return
	}
	if iter == nil {
		log.Warn("the metrics iterator is nil", zap.String("fileUUID", mt.FileUUID), zap.String("path", mt.OriginalPathInTar))
		return
	}

	err = stream.OfIterator(iter).BatchSink(mt.sink.SinkBatch, metricBatchSize)
	if err != nil {
		log.Error("import metric failed", zap.String("cluster", mt.ClusterID), zap.String("host", mt.hostName), zap.Error(err))
		metrics.ImportMetricFileProcessFailedCount.WithLabelValues(mt.ClusterID, mt.FileUUID).Inc()
		return
	}
	metrics.ImportMetricFileProcessSuccessfulCount.WithLabelValues(mt.ClusterID, mt.FileUUID).Inc()
}

func getHostName(filePath string) string {
	fileNameStart := strings.LastIndex(filePath, "/")
	if fileNameStart < 0 {
		return ""
	}
	dir := filePath[:fileNameStart]

	hpStart := strings.LastIndex(dir, "/")
	if hpStart < 0 {
		return ""
	}
	hostWithPort := dir[hpStart+1:]

	hostStart := strings.LastIndex(hostWithPort, "-")
	if hostStart < 0 {
		return ""
	}
	return hostWithPort[:hostStart]
}

// func isTSDBPath(filePath string) (string, bool) {
// 	index := strings.LastIndex(filePath, TDBSuffix)
// 	if index < 0 {
// 		return "", false
// 	}

// 	tsdbPathWithPromHost := strings.Split(filePath[index+len(TDBSuffix)+1:], "/")

// 	if len(tsdbPathWithPromHost) <= 2 {
// 		return "", false
// 	}
// 	return strings.Join(tsdbPathWithPromHost[1:len(tsdbPathWithPromHost)-1], "/"), true
// }

// only /xxx/metrics/raw/<host>-<port> with or without trailing slash is considered as tsdbDir
func IsTSDBDir(dirPath string) bool {
	if !path.IsAbs(dirPath) {
		return false
	}
	p := strings.TrimSuffix(dirPath, "/")
	index := strings.LastIndex(p, TDBSuffix)
	if index < 0 {
		return false
	}
	promHost := strings.Split(strings.TrimPrefix(p[index+len(TDBSuffix):], "/"), "/")
	return len(promHost) == 1 && promHost[0] != ""
}