package log

import (
	"compress/gzip"
	"fmt"
	"io"
	"os"
	"strings"
	"time"

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

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

const (
	logBatchSize = 50
	// consumeLogCount = 3
)

var importlogs = map[string]LogFileMeta{
	"tidb.": {
		LogType:        standardType,
		GuessComponent: cluster.TiDBComponent,
	},
	"tidb-": {
		LogType:        standardType,
		GuessComponent: cluster.TiDBComponent,
	},
	"pd.": {
		LogType:        standardType,
		GuessComponent: cluster.PDComponent,
	},
	"pd-": {
		LogType:        standardType,
		GuessComponent: cluster.PDComponent,
	},
	"tikv.": {
		LogType:        standardType,
		GuessComponent: cluster.TiKVComponent,
	},
	"tikv-": {
		LogType:        standardType,
		GuessComponent: cluster.TiKVComponent,
	},
	"tiflash.": {
		LogType:        standardType,
		GuessComponent: cluster.TiFlashComponent,
	},
	"tiflash-": {
		LogType:        standardType,
		GuessComponent: cluster.TiFlashComponent,
	},
	"tiflash_tikv.": {
		LogType:        standardType,
		GuessComponent: cluster.TiFlashComponent,
	},
	"tiflash_cluster_manager.": {
		LogType:        standardType,
		GuessComponent: cluster.TiFlashComponent,
	},
	"tidb_slow_query": {
		LogType:        slowQueryType,
		GuessComponent: cluster.TiDBComponent,
	},
	"dm-master.": {
		LogType:        standardType,
		GuessComponent: cluster.DMMasterComponent,
	},
	"dm-master-": {
		LogType:        standardType,
		GuessComponent: cluster.DMMasterComponent,
	},
	"dm-worker.": {
		LogType:        standardType,
		GuessComponent: cluster.DMWorkerComponent,
	},
	"dm-worker-": {
		LogType:        standardType,
		GuessComponent: cluster.DMWorkerComponent,
	},
	"cdc.": {
		LogType:        standardType,
		GuessComponent: cluster.TiCDCComponent,
	},
	"cdc-": {
		LogType:        standardType,
		GuessComponent: cluster.TiCDCComponent,
	},
}

type LogFileMeta struct {
	LogType        LogType
	GuessComponent cluster.ComponentName
}

func IsLogFile(path, fileName string, componentsMeta *cluster.ComponentsMeta) (*LogRecordOption, bool) {
	log.Info("is log file", zap.String("path", path), zap.String("fileName", fileName))
	if fileName == "" {
		return nil, false
	}

	if !strings.Contains(fileName, "log") {
		return nil, false
	}

	componentMeta, ok := componentsMeta.IsLog(path)
	if !ok {
		log.Info("skip the log file", zap.String("path", path), zap.String("fileName", fileName))
		return nil, false
	}

	for k, v := range importlogs {
		if strings.HasPrefix(fileName, k) {
			component := componentMeta.Component
			if component == cluster.UnKnownComponent {
				component = v.GuessComponent
			}

			return &LogRecordOption{
				host:       componentMeta.Host,
				instanceID: componentMeta.Instance,
				component:  string(component),
				logType:    v.LogType,
				fileName:   fileName,
			}, true
		}
	}

	log.Info("skip the log file", zap.String("path", path), zap.String("fileName", fileName))
	return nil, false
}

type logTask struct {
	*metrics2.TaskOption
	// logFilePath string
	// pathInTar   string
	// clusterID   uint64
	// fileUUID    string
	// callback    utils.ImportCallback
	opt  *LogRecordOption
	sink *LogSink
}

func NewLogTask(sink *LogSink, opt *LogRecordOption, taskOpt *metrics2.TaskOption) *logTask {
	return &logTask{
		TaskOption: taskOpt,
		opt:        opt,
		sink:       sink,
	}
}

func (lr logTask) Run() {
	startTime := time.Now().Unix()
	defer func() {
		cache.TaskStatusCache.Done(cache.TaskID(lr.FileUUID, _const.LogTask), lr.FileSize)
		// lr.Status.Done(lr.FileSize)
		metrics.RebuildLogCount.WithLabelValues(lr.ClusterID, lr.FileUUID).Inc()
		metrics.RebuildLogsCostTime.WithLabelValues(lr.ClusterID, lr.FileUUID).Add(float64(time.Now().Unix() - startTime))
	}()

	defer func() {
		log.Info("complete to insert logs", zap.String("fileUUID", lr.FileUUID), zap.String("filepath", lr.OriginalPathInTar))
	}()

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

	var r io.ReadCloser = f
	// tiflash may has compressed log
	if strings.HasSuffix(lr.FilePath, ".gz") {
		r, err = gzip.NewReader(f)
		if err != nil {
			log.Error("new gzip reader failed", zap.String("fileUUID", lr.FileUUID), zap.String("filepath", lr.OriginalPathInTar), zap.Error(err))
			return
		}
		defer r.Close()
	}

	// lr.Status.Start()
	metrics.ImportLogFileCount.WithLabelValues(lr.ClusterID, lr.FileUUID).Inc()
	log.Info("start to insert logs", zap.String("fileUUID", lr.FileUUID), zap.String("filepath", lr.OriginalPathInTar))
	err = stream.OfIterator(NewLogIterator(r, fmt.Sprintf("%s-%s", utils.FnvHash(lr.FileUUID), utils.FnvHash(lr.OriginalPathInTar)), lr.ClusterID, lr.FileUUID, lr.opt)).BatchSink(lr.sink.Sink, logBatchSize)
	if err != nil {
		log.Error("insert log failed", zap.String("filepath", lr.OriginalPathInTar), zap.Error(err))
		metrics.ImportLogFileProcessFailedCount.WithLabelValues(lr.ClusterID, lr.FileUUID).Inc()
	} else {
		metrics.ImportLogFileProcessSuccessfulCount.WithLabelValues(lr.ClusterID, lr.FileUUID).Inc()
	}
}