package cluster

import (
	"fmt"
	"path/filepath"
	"strings"

	"github.com/pingcap/diag/pkg/models"
	"github.com/pingcap/log"
	"go.uber.org/zap"

	"github.com/pingcap-inc/clinic/pkg/stream"
)

type ComponentsMeta struct {
	logs         map[string]ComponentMeta
	confs        map[string]ComponentMeta
	isNewVersion bool
}

type ComponentMeta struct {
	Component ComponentName
	Host      string
	Instance  string
}

type ComponentName string

var (
	PDComponent       = ComponentName("pd")
	TiDBComponent     = ComponentName("tidb")
	TiKVComponent     = ComponentName("tikv")
	TiFlashComponent  = ComponentName("tiflash")
	TiCDCComponent    = ComponentName("ticdc")
	PUMPComponent     = ComponentName("pump")
	DrainerComponent  = ComponentName("drainer")
	TiSparkComponent  = ComponentName("tispark")
	DMMasterComponent = ComponentName("dm-master")
	DMWorkerComponent = ComponentName("dm-worker")

	UnKnownComponent = ComponentName("unknown")

	logFiles    = "log_files"
	configFiles = "config_files"
)

func NewImportComponentsMeta(clusterInfo *ClusterInfo) *ComponentsMeta {
	results := &ComponentsMeta{
		logs:         make(map[string]ComponentMeta),
		confs:        make(map[string]ComponentMeta),
		isNewVersion: false,
	}

	appendComponentMeta := func(compoent ComponentName, spec models.ComponentSpec) {
		meta := ComponentMeta{
			Component: compoent,
			Host:      spec.Host,
			Instance:  fmt.Sprintf("%v-%d", compoent, spec.Port),
		}
		logfiles, isNewVersion := getLogFiles(spec)
		if len(logfiles) != 0 {
			for _, lf := range logfiles {
				if lf == "" {
					continue
				}

				v, ok := results.logs[lf]
				if ok && v.Component != meta.Component {
					meta.Component = UnKnownComponent
				}
				results.logs[lf] = meta
			}
		}
		results.isNewVersion = isNewVersion || results.isNewVersion

		conffiles, isNewVersion := getConfFiles(spec)
		if len(conffiles) != 0 {
			for _, cf := range conffiles {
				if cf == "" {
					continue
				}
				v, ok := results.confs[cf]
				if ok && v.Component != meta.Component {
					meta.Component = UnKnownComponent
				}
				results.confs[cf] = meta
			}
		}
		results.isNewVersion = isNewVersion || results.isNewVersion
	}

	stream.OfSlice(clusterInfo.Topology.PD).Each(func(spec *models.PDSpec) {
		appendComponentMeta(PDComponent, spec.ComponentSpec)
	})

	stream.OfSlice(clusterInfo.Topology.TiKV).Each(func(spec *models.TiKVSpec) {
		appendComponentMeta(TiKVComponent, spec.ComponentSpec)
	})

	stream.OfSlice(clusterInfo.Topology.TiDB).Each(func(spec *models.TiDBSpec) {
		appendComponentMeta(TiDBComponent, spec.ComponentSpec)
	})

	stream.OfSlice(clusterInfo.Topology.TiFlash).Each(func(spec *models.TiFlashSpec) {
		appendComponentMeta(TiFlashComponent, spec.ComponentSpec)
	})

	stream.OfSlice(clusterInfo.Topology.Pump).Each(func(spec *models.PumpSpec) {
		appendComponentMeta(PUMPComponent, spec.ComponentSpec)
	})

	stream.OfSlice(clusterInfo.Topology.TiCDC).Each(func(spec *models.TiCDCSpec) {
		appendComponentMeta(TiCDCComponent, spec.ComponentSpec)
	})

	stream.OfSlice(clusterInfo.Topology.Drainer).Each(func(spec *models.DrainerSpec) {
		appendComponentMeta(DrainerComponent, spec.ComponentSpec)
	})

	stream.OfSlice(clusterInfo.Topology.TiSpark).Each(func(spec *models.TiSparkCSpec) {
		appendComponentMeta(TiSparkComponent, spec.ComponentSpec)
	})

	stream.OfSlice(clusterInfo.Topology.DMMaster).Each(func(spec *models.DMMasterSpec) {
		appendComponentMeta(DMMasterComponent, spec.ComponentSpec)
	})

	stream.OfSlice(clusterInfo.Topology.DMWorker).Each(func(spec *models.DMWorkerSpec) {
		appendComponentMeta(DMWorkerComponent, spec.ComponentSpec)
	})
	return results
}

func NewImportComponentsMetaWithByte(clusterInfo []byte) (*ComponentsMeta, error) {
	ci, err := UnmarshalJSON(clusterInfo)
	if err != nil {
		return nil, err
	}

	return NewImportComponentsMeta(ci), nil
}

func (cm *ComponentsMeta) IsLog(path string) (*ComponentMeta, bool) {
	return getComponentMeta(cm.logs, path, cm.isNewVersion)
}

func (cm *ComponentsMeta) IsConf(path string) (*ComponentMeta, bool) {
	return getComponentMeta(cm.confs, path, cm.isNewVersion)
}

func getComponentMeta(metas map[string]ComponentMeta, path string, isNewVersion bool) (*ComponentMeta, bool) {
	p := path
	if strings.HasPrefix(path, "/") {
		p = path[1:]
	}
	if strings.HasPrefix(path, "./") {
		p = path[2:]
	}
	lastIndex := strings.LastIndex(p, "/")
	if lastIndex <= 0 {
		return nil, false
	}
	ppath := p[:lastIndex]
	meta, ok := metas[ppath]
	if ok {
		return &meta, true
	}
	if isNewVersion {
		for path, meta := range metas {
			if strings.HasSuffix(p, path) {
				return &meta, true
			}
		}
	}
	return &meta, false
}

func getLogPath(spec models.ComponentSpec) string {
	logDir, ok := spec.Attributes["log_dir"].(string)
	if !ok {
		log.Warn("empty log dir", zap.String("host", spec.Host), zap.Int("port", spec.Port))
		logDir = ""
	}

	deployDir, ok := spec.Attributes["deploy_dir"].(string)
	if !ok {
		log.Warn("empty deploy dir", zap.String("host", spec.Host), zap.Int("port", spec.Port))
		deployDir = ""
	}

	if logDir == "" && deployDir == "" {
		log.Error("empty deploy dir and log dir", zap.String("host", spec.Host), zap.Int("port", spec.Port))
		return ""
	}

	if logDir == "" {
		logDir = "log"
	}

	if !strings.HasPrefix(logDir, "/") {
		logDir = filepath.Join(deployDir, logDir)
	}
	return filepath.Join(spec.Host, logDir)
}

func getConfPath(spec models.ComponentSpec) string {
	dir, ok := spec.Attributes["deploy_dir"].(string)
	if !ok {
		log.Error("get log path failed")
		return ""
	}
	return filepath.Join(spec.Host, dir, "conf")
}

func getLogFiles(spec models.ComponentSpec) (logs []string, addition bool) {
	logpath := getLogPath(spec)
	logs, addition = spec.Attributes[logFiles].([]string)
	if !addition {
		log.Warn("no log_files attribute", zap.String("host", spec.Host), zap.Int("port", spec.Port), zap.Any("spec", spec), zap.String("logpath", logpath))
	}
	logs = append(logs, logpath)
	if pod, ok := spec.Attributes["pod"].(string); ok {
		logs = append(logs, filepath.Join("logs", pod))
	}
	return
}

func getConfFiles(spec models.ComponentSpec) ([]string, bool) {
	confpath := getConfPath(spec)
	configs, ok := spec.Attributes[configFiles].([]any)
	if !ok {
		log.Warn("no config_files attribute", zap.Any("spec", spec))
		return []string{confpath}, false
	}
	configrs := make([]string, 0)
	for _, c := range configs {
		cr, ok := c.(string)
		if !ok {
			continue
		}
		configrs = append(configrs, cr)
	}
	return configrs, true
}