package configuration

import (
	"archive/tar"
	"io"
	"os"

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

	"github.com/pingcap-inc/clinic/pkg/cluster"
	"github.com/pingcap-inc/clinic/pkg/config"
	"github.com/pingcap-inc/clinic/pkg/event_bus"
	"github.com/pingcap-inc/clinic/pkg/events"
	"github.com/pingcap-inc/clinic/pkg/importer/utils"
	"github.com/pingcap-inc/clinic/pkg/stream"
)

type ConfigurationTask struct {
	logConfig      config.LogConfig
	clusterID      uint64
	packagePath    string
	fileUUID       string
	callback       utils.ImportCallback
	componentsMeta *cluster.ComponentsMeta
}

func NewConfigurationTask(cfg config.LogConfig, clusterID uint64, packagePath, fileUUID string, componentsMeta *cluster.ComponentsMeta, callback utils.ImportCallback) *ConfigurationTask {
	return &ConfigurationTask{
		logConfig:      cfg,
		clusterID:      clusterID,
		packagePath:    packagePath,
		fileUUID:       fileUUID,
		callback:       callback,
		componentsMeta: componentsMeta,
	}
}

func (ct *ConfigurationTask) Run() {
	readerFunc := func() (io.ReadCloser, error) {
		return os.Open(ct.packagePath)
	}
	cb := func(err error) {
		if ct.callback != nil {
			ct.callback(err)
		}
	}

	err := ct.importConfiguration(readerFunc, func(r io.ReadCloser) *tar.Reader {
		return tar.NewReader(r)
	}, ct.componentsMeta)
	cb(err)
	if err != nil {
		log.Error("import configuration failed", zap.Uint64("cluster", ct.clusterID), zap.String("tar", ct.packagePath), zap.Error(err))
	} else {
		err = event_bus.PostSync(events.RebuildConfigurationSuccessEvent{FileUUID: ct.fileUUID})
		if err != nil {
			log.Error("handle rebuild configuration event failed", zap.String("fileUUID", ct.fileUUID), zap.Error(err))
		}
	}
}

func (ct *ConfigurationTask) importConfiguration(freader func() (io.ReadCloser, error), tarreader func(r io.ReadCloser) *tar.Reader, componentsMeta *cluster.ComponentsMeta) error {
	f, err := freader()
	if err != nil {
		log.Error("open tar file failed", zap.Uint64("cluster", ct.clusterID), zap.Error(err))
		return err
	}
	defer func() {
		f.Close()
	}()

	reader := tarreader(f)

	configSink, err := NewConfigurationSink(ct.logConfig /* It's ok to use LogConfig here */, ct.clusterID)
	if err != nil {
		return err
	}
	defer func() {
		err = configSink.Close()
		if err != nil {
			log.Error("close sink failed", zap.Error(err))
		}
	}()

	submitTask := func(fileName string, header *tar.Header, r io.Reader) {
		path := header.Name
		componentMeta, isConf := isConfigurationFile(path, fileName, componentsMeta)
		if isConf {
			log.Info("start to insert configurations", zap.String("fileUUID", ct.fileUUID), zap.String("filepath", path))
			err = stream.OfIterator(NewConfigurationIterator(r, path, ct.fileUUID, componentMeta.Instance, componentMeta.Host, string(componentMeta.Component))).BatchSink(configSink.Sink, 1)
			if err != nil {
				log.Error("insert configuration failed", zap.String("filepath", path), zap.Error(err))
			}
		}
	}
	return utils.ReadTarWithReader(reader, submitTask)
}

func isConfigurationFile(path, fileName string, componentsMeta *cluster.ComponentsMeta) (*cluster.ComponentMeta, bool) {
	if fileName == "" {
		return nil, false
	}

	return componentsMeta.IsConf(path)
}