package importer

import (
	"archive/tar"
	"fmt"
	"io"
	"io/fs"
	"os"
	"path/filepath"
	"strings"
	"time"

	"github.com/DataDog/zstd"

	"github.com/pingcap-inc/clinic/pkg/db/model"
	"github.com/pingcap-inc/clinic/pkg/importer/utils"

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

	"github.com/pingcap-inc/clinic/pkg/cache"
	"github.com/pingcap-inc/clinic/pkg/cluster"
	"github.com/pingcap-inc/clinic/pkg/config"
	_const "github.com/pingcap-inc/clinic/pkg/const"
	"github.com/pingcap-inc/clinic/pkg/db"
	"github.com/pingcap-inc/clinic/pkg/event_bus"
	"github.com/pingcap-inc/clinic/pkg/events"
	executor2 "github.com/pingcap-inc/clinic/pkg/executor"
	"github.com/pingcap-inc/clinic/pkg/importer/fetchconf"
	log2 "github.com/pingcap-inc/clinic/pkg/importer/log"
	metrics2 "github.com/pingcap-inc/clinic/pkg/importer/metrics"
	"github.com/pingcap-inc/clinic/pkg/metrics"
	"github.com/pingcap-inc/clinic/pkg/storage/iface"
	"github.com/pingcap-inc/clinic/pkg/upload_file/manager"
)

type Importer struct {
	opt *ImporterOption

	clusterInfo              *cluster.ComponentsMeta
	expectedRebuildTaskTypes map[_const.TaskType]struct{}
	packageSize              int64
}

type ImporterOption struct {
	// static configuration
	Storage      iface.IStorage
	Manager      *manager.FileStoreageManager
	LogConfig    config.LogConfig
	DataPath     string
	MetricConfig config.MetricConfig // for report snapshot

	// cluster info
	FileUUID                 string
	ClusterID                string
	OrgID                    uint
	ExpectedRebuildTaskTypes []_const.TaskType
	BeginTS                  int64
	EndTS                    int64

	// for test
	localFilePath string
}

func NewImporter(opt *ImporterOption) *Importer {
	im := &Importer{
		opt: opt,
	}

	return im
}

func (i *Importer) Rebuild() error {
	if len(i.opt.ExpectedRebuildTaskTypes) == 0 {
		log.Warn("empty rebuild task", zap.String("fileUUID", i.opt.FileUUID), zap.String("clusterID", i.opt.ClusterID))
		return nil
	}

	startTime := time.Now().Unix()
	defer func() {
		log.Info("import the package done", zap.String("fileUUID", i.opt.FileUUID), zap.String("clusterID", i.opt.ClusterID))
		metrics.PackageRebuildCostTime.WithLabelValues(i.opt.FileUUID).Set(float64(time.Now().Unix() - startTime))
		if err := recover(); err != nil {
			log.Error("import task failed", zap.String("uuid", i.opt.FileUUID), zap.String("err", fmt.Sprintf("%v", err)))
			SendEvent(i.expectedRebuildTaskTypes, i.opt.FileUUID, db.StatusFailed)
		}
	}()

	err := i.decompress()
	if err != nil {
		log.Error("decompress file failed", zap.String("fileUUID", i.opt.FileUUID), zap.Error(err))
		return err
	}

	metrics.PackageDecompressCostTime.WithLabelValues(i.opt.FileUUID).Set(float64(time.Now().Unix() - startTime))
	log.Info("decompress package done", zap.String("fileUUID", i.opt.FileUUID))

	tasks := make(map[_const.TaskType]struct{}, 0)
	for _, rt := range i.opt.ExpectedRebuildTaskTypes {
		tasks[rt] = struct{}{}
	}

	clusterInfo, err := cluster.GetClusterInfoFromDataPath(i.opt.DataPath)
	if err != nil {
		log.Error("get cluster info failed", zap.String("fileUUID", i.opt.FileUUID), zap.Error(err))
		return err
	}
	i.clusterInfo = cluster.NewImportComponentsMeta(clusterInfo)
	i.expectedRebuildTaskTypes = tasks

	return i.run()
}

func (i *Importer) decompress() error {
	reader, err := i.getFile()
	if err != nil {
		return err
	}

	defer func() {
		_ = reader.Close()
	}()

	zdec := zstd.NewReader(reader)
	defer func() {
		_ = zdec.Close()
	}()

	packageSize, err := utils.ExtractDecompressFile(tar.NewReader(zdec), i.opt.DataPath)
	if err != nil {
		log.Error("extract file from decompress file failed", zap.Error(err), zap.String("path", i.opt.DataPath))
		return err
	}
	i.packageSize = packageSize

	clusterInfo, err := cluster.GetClusterInfoFromDataPath(i.opt.DataPath)
	if err != nil {
		log.Error("get cluster info failed", zap.String("uuid", i.opt.FileUUID), zap.Error(err))
		return err
	}
	return model.UpdateFileInfoAfterSubmitRebuild(i.opt.FileUUID, clusterInfo)
}

func (i *Importer) getFile() (io.ReadCloser, error) {
	if i.opt.localFilePath != "" {
		return os.Open(i.opt.localFilePath)
	}

	file, _, ferr := i.opt.Manager.GetFile(i.opt.FileUUID, "")
	if ferr != nil {
		log.Error("error on get file", zap.String("uuid", i.opt.FileUUID), zap.Error(ferr))
		return nil, ferr
	}

	return file.Body, nil
}

func (i *Importer) run() error {
	fastExecutor := executor2.NewExecutorWithQueueSize(8, 100_000)
	slowExecutor := executor2.NewExecutorWithQueueSize(8, 100_000)
	defer func() {
		fastExecutor.Close()
		slowExecutor.Close()
	}()

	rebuildTasks := []RebuildTask{NewFastRebuildTask(fastExecutor, i), NewSlowRebuildTask(slowExecutor, i)}
	err := filepath.Walk(i.opt.DataPath, func(path string, info fs.FileInfo, err error) error {
		// isFastType
		isTSDBDir := metrics2.IsTSDBDir(path)
		if info.IsDir() && !isTSDBDir {
			return nil
		}
		for _, task := range rebuildTasks {
			if task == nil {
				continue
			}

			if taskType, ok := task.IsMatch(info, path); ok {
				task.Submit(info, path, taskType)
				break
			}
		}
		// if path id tsdb dir path, skip the path under the tsdb dir path
		if isTSDBDir {
			return filepath.SkipDir
		}
		return nil
	})
	if err != nil {
		log.Error("walk data path failed", zap.String("FileUUID", i.opt.FileUUID), zap.Error(err))
		return err
	}

	for _, task := range rebuildTasks {
		if task != nil {
			task.WaitDone()
		}
	}

	log.Info("rebuild the metrics and logs done", zap.String("fileUUID", i.opt.FileUUID), zap.String("clusterID", i.opt.ClusterID))
	return nil
}

func SendEvent(sortingTasks map[_const.TaskType]struct{}, fileUUID string, status db.StatusType) {
	for t := range sortingTasks {
		// view *RebuildEventListener.SubscribeRebuildEvent for processing logic
		if err := event_bus.PostSync(events.RebuildEvent{
			FileUUID:      fileUUID,
			TaskType:      t,
			RebuildStatus: status,
		}); err != nil {
			log.Error("send rebuild event failed", zap.String("file", fileUUID), zap.Int("task", int(t)), zap.Error(err))
		}
	}
}

type FastRebuildTask struct {
	importer      *Importer
	executor      executor2.Executor
	startTime     int64
	metricSink    *metrics2.MetricSink
	fetchConfSink *fetchconf.FetchConfSink

	sortingTasks map[_const.TaskType]struct{}
}

func NewFastRebuildTask(executor executor2.Executor, importer *Importer) RebuildTask {
	tasks := make(map[_const.TaskType]struct{})

	for _, t := range []_const.TaskType{_const.MetricTask, _const.RebuildConfTask} {
		if _, ok := importer.expectedRebuildTaskTypes[t]; ok {
			tasks[t] = struct{}{}
			cache.TaskStatusCache.Init(cache.TaskID(importer.opt.FileUUID, t))

			cache.TaskStatusCache.TotalIncr(cache.TaskID(importer.opt.FileUUID, t), importer.packageSize)
		}
	}

	if len(tasks) == 0 {
		return nil
	}

	fst := &FastRebuildTask{
		importer:     importer,
		executor:     executor,
		sortingTasks: tasks,
	}

	fst.metricSink = &metrics2.MetricSink{
		Host:         importer.opt.MetricConfig.Host,
		Port:         importer.opt.MetricConfig.Port,
		ClusterModel: true,
		OrgID:        importer.opt.OrgID,
	}

	sink, err := fetchconf.NewFetchConfSink(fst.importer.opt.ClusterID, fst.importer.opt.FileUUID, importer.opt.Storage)
	if err != nil {
		log.Error("init fast runner failed", zap.String("file", fst.importer.opt.FileUUID))
		metrics.RebuildTaskFailed.WithLabelValues(importer.opt.FileUUID).Inc()
		SendEvent(tasks, importer.opt.FileUUID, db.StatusFailed)
		return nil
	}
	fst.fetchConfSink = sink
	fst.startTime = time.Now().Unix()

	SendEvent(tasks, importer.opt.FileUUID, db.StatusRunning)
	return fst
}

func (fst *FastRebuildTask) WaitDone() {
	defer func() {
		metrics.FastRebuildCostTime.WithLabelValues(fst.importer.opt.FileUUID).Set(float64(time.Now().Unix() - fst.startTime))
		for t := range fst.sortingTasks {
			cache.TaskStatusCache.Remove(cache.TaskID(fst.importer.opt.FileUUID, t))
		}

		SendEvent(fst.sortingTasks, fst.importer.opt.FileUUID, db.StatusSuccess)
	}()
	for t := range fst.sortingTasks {
		fst.executor.Wait(string(cache.TaskID(fst.importer.opt.FileUUID, t)))
	}

	taskNames := ""
	for t := range fst.sortingTasks {
		taskNames += _const.TaskTypeString(t) + " "
	}

	log.Info("rebuild fast task success", zap.String("task", taskNames), zap.String("fileUUID", fst.importer.opt.FileUUID), zap.String("cluster", fst.importer.opt.ClusterID))

	fst.fetchConfSink.Close()
}

func getOriginalPathInTar(dataPath, absolutePath string) string {
	return strings.TrimPrefix(absolutePath, fmt.Sprintf("%s/", dataPath))
}

func (fst *FastRebuildTask) IsMatch(info fs.FileInfo, path string) (_const.TaskType, bool) {
	if metrics2.IsMetricFile(info.Name(), path) {
		_, ok := fst.sortingTasks[_const.MetricTask]
		return _const.MetricTask, ok
	}

	if fetchconf.IsConfFile(getOriginalPathInTar(fst.importer.opt.DataPath, path), info.Name(), fst.importer.clusterInfo) && !fetchconf.IsStderrLog(info.Name()) {
		_, ok := fst.sortingTasks[_const.RebuildConfTask]
		return _const.RebuildConfTask, ok
	}

	return _const.UnsupportTask, false
}

func (fst *FastRebuildTask) Submit(info fs.FileInfo, path string, taskType _const.TaskType) {
	// taskType := _const.UnsupportTask
	// var task executor2.Task
	if taskType == _const.MetricTask {
		// check if the path is tsdb dir

		taskOpt := &metrics2.TaskOption{
			FilePath:          path,
			FileSize:          info.Size(),
			OriginalPathInTar: getOriginalPathInTar(fst.importer.opt.DataPath, path),
			ClusterID:         fst.importer.opt.ClusterID,
			FileUUID:          fst.importer.opt.FileUUID,
		}
		log.Debug("check path before submit metrics task",
			zap.String("uuid", taskOpt.FileUUID),
			zap.String("path", path),
			zap.String("cluster", fst.importer.opt.ClusterID),
			zap.Bool("isTSDBDir", metrics2.IsTSDBDir(path)),
		)
		if metrics2.IsTSDBDir(path) {
			taskOpt.ProcessAsTSDB = true
			taskOpt.TSDBDirPath = path
			// recalculate the total file size under the path
			// TODO the following calculation is just a estimation
			totalSize := int64(0)
			_ = filepath.WalkDir(path, func(path string, entry fs.DirEntry, err error) error {
				if !entry.Type().IsRegular() {
					return nil
				}
				if eInfo, err := entry.Info(); err == nil {
					totalSize += eInfo.Size()
				}
				return nil
			})

			taskOpt.FileSize = totalSize
		}
		task := metrics2.NewMetricTask(fst.metricSink, taskOpt)
		fst.executor.Submit(string(cache.TaskID(fst.importer.opt.FileUUID, taskType)), task)
		return
	}

	if taskType == _const.RebuildConfTask {
		hdr := &tar.Header{
			Name: getOriginalPathInTar(fst.importer.opt.DataPath, path),
			Mode: int64(info.Mode()),
			Size: info.Size(),
		}

		// taskType = _const.RebuildConfTask
		task := fetchconf.NewFetchConfTask(fst.fetchConfSink, hdr, &metrics2.TaskOption{
			// Status:            ft.tasks[_const.RebuildConfTask],
			FilePath:          path,
			FileSize:          info.Size(),
			OriginalPathInTar: getOriginalPathInTar(fst.importer.opt.DataPath, path),
			ClusterID:         fst.importer.opt.ClusterID,
			FileUUID:          fst.importer.opt.FileUUID,
		})

		fst.executor.Submit(string(cache.TaskID(fst.importer.opt.FileUUID, taskType)), task)
	}
}

type SlowRebuildTask struct {
	executor executor2.Executor

	importer *Importer

	startTime    int64
	sink         *log2.LogSink
	sortingTasks map[_const.TaskType]struct{}
}

func NewSlowRebuildTask(executor executor2.Executor, importer *Importer) RebuildTask {
	tasks := make(map[_const.TaskType]struct{})
	for _, t := range []_const.TaskType{_const.LogTask} {
		if _, ok := importer.expectedRebuildTaskTypes[t]; ok {
			tasks[t] = struct{}{}
			cache.TaskStatusCache.Init(cache.TaskID(importer.opt.FileUUID, t))

			cache.TaskStatusCache.TotalIncr(cache.TaskID(importer.opt.FileUUID, t), importer.packageSize)
		}
	}

	// TODO: check
	if len(tasks) == 0 {
		return nil
	}

	t := &SlowRebuildTask{
		importer:     importer,
		executor:     executor,
		sortingTasks: tasks,
	}

	logsink, err := log2.NewLogSink(importer.opt.LogConfig, importer.opt.OrgID, importer.opt.BeginTS, importer.opt.EndTS)
	if err != nil {
		log.Error("init log sink failed", zap.String("uuid", importer.opt.FileUUID), zap.Error(err))
		return nil
	}

	t.sink = logsink
	t.startTime = time.Now().Unix()

	SendEvent(tasks, importer.opt.FileUUID, db.StatusRunning)
	return t
}

func (srt *SlowRebuildTask) IsMatch(info fs.FileInfo, path string) (_const.TaskType, bool) {
	if _, isLogFile := log2.IsLogFile(getOriginalPathInTar(srt.importer.opt.DataPath, path), info.Name(), srt.importer.clusterInfo); isLogFile {
		_, ok := srt.sortingTasks[_const.LogTask]
		return _const.LogTask, ok
	}

	return _const.UnsupportTask, false
}

func (srt *SlowRebuildTask) Submit(info fs.FileInfo, path string, taskType _const.TaskType) {
	if opt, isLogFile := log2.IsLogFile(getOriginalPathInTar(srt.importer.opt.DataPath, path), info.Name(), srt.importer.clusterInfo); isLogFile {
		// cache.TaskStatusCache.TotalIncr(cache.TaskID(srt.importer.opt.FileUUID, _const.LogTask), info.Size())

		t := log2.NewLogTask(srt.sink, opt, &metrics2.TaskOption{
			// Status:            sr.tasks[_const.LogTask],
			FilePath:          path,
			FileSize:          info.Size(),
			OriginalPathInTar: getOriginalPathInTar(srt.importer.opt.DataPath, path),
			ClusterID:         srt.importer.opt.ClusterID,
			FileUUID:          srt.importer.opt.FileUUID,
		})

		srt.executor.Submit(string(cache.TaskID(srt.importer.opt.FileUUID, _const.LogTask)), t)
	}
}

func (srt *SlowRebuildTask) WaitDone() {
	defer func() {
		metrics.SlowRebuildCostTime.WithLabelValues(srt.importer.opt.FileUUID).Set(float64(time.Now().Unix() - srt.startTime))
		for t := range srt.sortingTasks {
			cache.TaskStatusCache.Remove(cache.TaskID(srt.importer.opt.FileUUID, t))
		}

		SendEvent(srt.sortingTasks, srt.importer.opt.FileUUID, db.StatusSuccess)
	}()

	for t := range srt.sortingTasks {
		srt.executor.Wait(string(cache.TaskID(srt.importer.opt.FileUUID, t)))
	}

	err := srt.sink.Close()
	if err != nil {
		log.Error("close sink failed", zap.String("uuid", srt.importer.opt.FileUUID), zap.Error(err))
	}

	err = event_bus.PostSync(events.RebuildLogSuccessEvent{FileUUID: srt.importer.opt.FileUUID, OrgID: srt.importer.opt.OrgID})
	if err != nil {
		log.Error("handle rebuild log event failed", zap.String("fileUUID", srt.importer.opt.FileUUID), zap.Error(err))
	}

	taskNames := ""
	for t := range srt.sortingTasks {
		taskNames += _const.TaskTypeString(t) + " "
	}

	log.Info("rebuild slow task success", zap.String("task", taskNames), zap.String("fileUUID", srt.importer.opt.FileUUID), zap.String("cluster", srt.importer.opt.ClusterID))
}

type RebuildTask interface {
	WaitDone()
	Submit(info fs.FileInfo, path string, taskType _const.TaskType)
	IsMatch(info fs.FileInfo, path string) (_const.TaskType, bool)
}