package cat

import (
	"context"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/schrodinger/pkg/storage"
	"github.com/pingcap/schrodinger/pkg/types"
	"github.com/pingcap/schrodinger/util"
	oputil "github.com/pingcap/tidb-operator/pkg/util"
	"golang.org/x/sync/errgroup"
)

// Operator provides a special interface for control cluster.
type Operator interface {
	// CreateCluster creates a tidb cluster
	CreateCluster(ctx context.Context, cluster *types.Cluster, waitOk bool) error
	// UpdateCluster updates a tidb cluster
	UpdateCluster(ctx context.Context, cluster *types.Cluster, waitOk bool) error
	// ListCluster lists all clusters
	ListCluster(spec *types.Cluster) ([]*types.Cluster, error)
	// QueryCluster queries cluster by name, if waitOk is true, then will wait cluster ready
	QueryCluster(ctx context.Context, name string, waitOk bool) (*types.Cluster, error)
	// DeleteCluster deletes cluster by name
	DeleteCluster(name string) error
	// BackupCluster backups tidb cluster
	BackupCluster(ctx context.Context, tidbClusterBackup *types.TiDBClusterBackup) error
	// RestoreCluster restores tidb cluster
	RestoreCluster(ctx context.Context, tidbClusterBackup *types.TiDBClusterBackup) error
}

// Cat is cat
type Cat struct {
	sync.RWMutex
	ctx    context.Context
	cancel context.CancelFunc

	cluster *types.Cluster

	process *types.CatProcess

	store  *storage.Storage
	labels map[string]map[string]string

	closed int32

	operator Operator
}

// NewCat creates cat
func NewCat(
	ctx context.Context,
	process *types.CatProcess,
	store *storage.Storage,
	operator Operator,
) *Cat {
	cctx, cancel := context.WithCancel(ctx)
	cat := &Cat{
		ctx:      cctx,
		cancel:   cancel,
		process:  process,
		store:    store,
		operator: operator,
	}
	cat.labels = cat.genCatLabel(process.UseBoxLabel, process.IgnoreLabels)

	cat.saveCat()
	return cat
}

// Start starts cat
func (c *Cat) Start() error {
	c.Lock()

	defer func() {
		c.saveCat()
		c.Unlock()
	}()

	// if cat state is abnormal, just return nil
	// starts ok, but process contains state abnormal by design
	if c.getState() == types.CatStateAbnormal {
		return nil
	}

	var err error
	switch c.getStage() {
	case types.CatStageNew, types.CatStageCreating:
		c.setStage(types.CatStageCreating)
		log.Infof("start to create cluster %s", c.process.Namespace)
		err = c.createCluster()
	case types.CatStageRunning, types.CatStageUpdating:
		err = c.checkAndSetCluster()
	case types.CatStageBackuping:
		err = c.backup(c.process.BackupName)
	case types.CatStageRestoring:
		backup, ok := c.store.GetTiDBClusterBackupByCatIDAndName(c.process.ID, c.process.BackupName)
		if !ok {
			err = errors.NotFoundf("backup %s", c.process.BackupName)
			break
		}
		err = c.restore(backup)
	default:
		err = errors.NotValidf("cat stage %s", c.getStage())
	}

	if err == nil {
		log.Infof("start cat %d [namespace %s] success", c.Identify(), c.Namespace())
		c.setStage(types.CatStageRunning)
		return errors.Trace(c.syncStatus())
	}

	if strings.Contains(err.Error(), "context canceled") {
		log.Infof("start cat %s canceled", c.Name())
		return errors.Trace(err)
	}

	c.setState(types.CatStateAbnormal)
	return errors.Trace(err)
}

// Stop stops cat
func (c *Cat) Stop() error {
	log.Infof("stop cat %d [namespace %s]", c.Identify(), c.Namespace())
	if !atomic.CompareAndSwapInt32(&c.closed, 0, 1) {
		return errors.MethodNotAllowedf("cat %d [namespace %s] was already stopped", c.Identify(), c.Namespace())
	}

	c.process.BeDestroyed()
	c.cancel()

	c.Lock()
	defer func() {
		c.saveCat()
		c.Unlock()
	}()

	err := c.destroyCluster()
	if err != nil {
		return errors.Trace(err)
	}

	log.Infof("stop cat %d [namespace %s] success", c.Identify(), c.Namespace())

	return nil
}

// Upgrade upgrades cat
func (c *Cat) Upgrade(config *types.Cat) error {
	c.Lock()
	defer func() {
		c.saveCat()
		c.Unlock()
	}()

	c.process.Creator = config.Creator

	log.Infof("upgrade cat %d [namespace %s]", c.Identify(), c.Namespace())

	if c.getStage() != types.CatStageRunning || c.getState() != types.CatStateNormal {
		return errors.NotSupportedf("update cat from stage %s state", c.getStage(),
			c.getState())
	}

	c.setStage(types.CatStageUpdating)
	err := c.upgrade(config)
	if err != nil {
		if err == context.Canceled {
			return errors.Errorf("update cat %s canceld", c.Name())
		}
		c.setState(types.CatStateAbnormal)
		return errors.Trace(err)
	}

	log.Infof("upgrade cat %d [namespace %s] success", c.Identify(), c.Namespace())

	c.setStage(types.CatStageRunning)
	return errors.Trace(c.syncStatus())
}

// Backup backups cat owner physically
func (c *Cat) Backup(backupName string) error {
	c.Lock()
	defer func() {
		c.saveCat()
		c.Unlock()
	}()

	log.Infof("backup cat %d [namespace %s]", c.Identify(), c.Namespace())
	if c.getStage() != types.CatStageRunning {
		return errors.NotSupportedf("backup cat from stage %s", c.getStage())
	}

	c.setStage(types.CatStageBackuping)
	c.process.BackupName = backupName
	c.saveCat()

	go func() {
		defer c.saveCat()
		if err := c.backup(backupName); err != nil {
			c.setState(types.CatStateAbnormal)
			log.Errorf("backup cat %d [namespace %s] failed: %v", c.Identify(), c.Namespace(), err)
			return
		}
		log.Infof("backup cat %d [namespace %s] success", c.Identify(), c.Namespace())
		c.setStage(types.CatStageRunning)
	}()

	return nil
}

// ListBackups lists all backups of the cat
func (c *Cat) ListBackups() ([]*types.TiDBClusterBackup, error) {
	c.Lock()
	defer func() {
		c.Unlock()
	}()

	backups := c.store.GetTiDBClusterBackupsByCatID(c.process.ID)

	return backups, nil
}

// Restore restores cat owner with a pyhsical backup
func (c *Cat) Restore(backupName string) error {
	c.Lock()
	defer func() {
		c.saveCat()
		c.Unlock()
	}()

	backup, ok := c.store.GetTiDBClusterBackupByCatIDAndName(c.process.ID, backupName)
	if !ok {
		log.Errorf("cat %s does not have a %s backup", c.Namespace(), backupName)
		return errors.Errorf("cat %s does not have a %s backup", c.Namespace(), backupName)
	}

	log.Infof("restore cat %d [namespace %s]", c.Identify(), c.Namespace())
	if c.getStage() == types.CatStageNew || c.getState() == types.CatStageCreating {
		return errors.NotSupportedf("restore cat from stage %s", c.getStage())
	}
	if backup.State != types.BackupSuccess {
		log.Errorf("cat %s backup %s can not be used to backup, the state is %s", c.Namespace(), backupName, backup.State)
		return errors.Errorf("cat %s backup %s can not be used to backup, the state is %s", c.Namespace(), backupName, backup.State)
	}

	c.setStage(types.CatStageRestoring)
	c.setState(types.CatStateNormal)
	c.process.BackupName = backupName
	c.saveCat()

	go func() {
		defer c.saveCat()
		if err := c.restore(backup); err != nil {
			c.setState(types.CatStateAbnormal)
			log.Errorf("restore cat %d [namespace %s] failed: %v", c.Identify(), c.Namespace(), err)
			return
		}
		log.Infof("restore cat %d [namespace %s] success", c.Identify(), c.Namespace())
		c.setStage(types.CatStageRunning)
	}()

	return nil
}

// Status returns the stage and state of the cat
func (c *Cat) Status() types.CatStatus {
	return types.CatStatus{
		Stage: c.getStage(),
		State: c.getState(),
	}
}

func (c *Cat) backup(backupName string) error {
	var backup *types.TiDBClusterBackup

	backup, exist := c.store.GetTiDBClusterBackupByCatIDAndName(c.process.ID, backupName)
	if !exist {
		backup = &types.TiDBClusterBackup{
			Name:        backupName,
			Namespace:   c.Namespace(),
			ClusterName: "tidb-cluster",
			State:       types.BackupRunning,
			CatID:       c.process.ID,
		}

		if err := c.store.CreateTiDBClusterBackup(backup); err != nil {
			return errors.Trace(err)
		}
	}

	if err := c.operator.BackupCluster(c.ctx, backup); err != nil {
		backup.State = types.BackupFailed
		if err1 := c.store.UpdateTiDBClusterBackup(backup); err1 != nil {
			log.Errorf("update backup %v failed: %v", backup, backupName)
		}
		return errors.Trace(err)
	}

	backup.State = types.BackupSuccess
	log.Infof("update backup info %v", backup)
	if err := c.store.UpdateTiDBClusterBackup(backup); err != nil {
		return errors.Trace(err)
	}

	return nil
}

func (c *Cat) restore(backup *types.TiDBClusterBackup) error {
	return errors.Trace(c.operator.RestoreCluster(c.ctx, backup))
}

func (c *Cat) syncStatus() error {
	cluster, err := c.operator.QueryCluster(context.Background(), c.process.Namespace, false)
	if err != nil {
		log.Errorf("[namespace: %s]fail to query cluster %v", c.process.Namespace, err)
		return errors.Trace(err)
	}
	c.cluster = cluster
	c.cluster.Meta = c.process.Meta

	return nil
}

// Health returns true if the cat is healthy
func (c *Cat) Health() (bool, error) {
	// sync the status of the cat
	if err := c.syncStatus(); err != nil {
		return false, errors.Trace(err)
	}

	for _, tikv := range c.cluster.TikvStatus {
		if tikv.ProcessStatus != oputil.StoreUpState &&
			tikv.ProcessStatus != oputil.StoreOfflineState &&
			tikv.ProcessStatus != oputil.StoreTombstoneState {
			log.Errorf("cat %s tikv %s is %s", c.Name(), tikv.Name, tikv.ProcessStatus)
			return false, errors.Errorf("tikv %s is %s", tikv.Name, tikv.ProcessStatus)
		}
	}

	for _, pd := range c.cluster.PdStatus {
		if pd.ProcessStatus != oputil.StoreUpState {
			log.Errorf("cat %s pd %s is unhealthy", c.Name(), pd.Name)
			return false, errors.Errorf("pd %s is unhealthy", pd.Name)
		}
	}

	for _, tidb := range c.cluster.TidbStatus {
		if tidb.ProcessStatus != oputil.StoreUpState {
			log.Errorf("cat %s tidb %s is unhealthy", c.Name(), tidb.Name)
			return false, errors.Errorf("tidb %s is unhealthy", tidb.Name)
		}
	}

	return true, nil
}

func (c *Cat) checkAndSetCluster() error {
	var err error
	c.cluster, err = c.operator.QueryCluster(c.ctx, c.process.Namespace, true)
	if err == nil {
		c.cluster.Meta = c.process.Meta
		return nil
	}
	if err == context.Canceled {
		log.Infof("cancel query cluster %s", c.process.Namespace)
		return errors.Trace(err)
	}
	return errors.Annotatef(err, "get cluster %s", c.process.Namespace)
}

func (c *Cat) createCluster() error {
	err := c.prepareImage()
	if err != nil {
		return errors.Trace(err)
	}

	c.cluster = getClusterSpec(c.store, c.process, c.labels)

	err = c.operator.CreateCluster(c.ctx, c.cluster, true)
	if err == nil {
		return nil
	}

	if strings.Contains(err.Error(), "already exists") {
		c.cluster, err = c.operator.QueryCluster(c.ctx, c.process.Namespace, true)
		if err != nil {
			return errors.Trace(err)
		}
	}

	return errors.Trace(err)
}

func (c *Cat) genCatLabel(useBoxLabel bool, ignoreLabels map[string]string) map[string]map[string]string {
	labels := make(map[string]map[string]string)

	nodeLabel := getNodeLabels(c.process.Meta)

	// set node label for tidb/pd/tikv
	labels["tidb"] = util.MapCopy(nodeLabel)
	labels["pd"] = util.MapCopy(nodeLabel)
	labels["tikv"] = util.MapCopy(nodeLabel)
	labels["monitor"] = util.MapCopy(nodeLabel)

	return labels
}

func (c *Cat) destroyCluster() error {
	log.Infof("destroy cluster namespace %s", c.process.Namespace)

	err := c.operator.DeleteCluster(c.process.Namespace)
	if err != nil && !strings.Contains(err.Error(), "not found") {
		return errors.Annotatef(err, "[namespace: %s] delete cluster namespace", c.process.Namespace)
	}

	log.Infof("destroy cluster namespace %s success", c.process.Namespace)

	return nil
}

func (c *Cat) prepareImage() error {
	log.Infof("start prepare image")

	tasks := c.initClusterTasks()

	if err := c.runBuildTasks(c.ctx, tasks); err != nil {
		return errors.Trace(err)
	}

	runVer, err := genRunVersion(tasks)
	if err != nil {
		return errors.Trace(err)
	}

	if !runVer.Valid() {
		return errors.NotValidf("version %+v", runVer)
	}

	log.Infof("cat %d [namespace %s] prepare image success, tidb version %s, tikv version %s, pd version %s",
		c.Identify(), c.Namespace(), runVer.TiDBVersion.Value, runVer.TiKVVersion.Value, runVer.PDVersion.Value)

	c.process.RunVersion = runVer

	return nil
}

func (c *Cat) initClusterTasks() map[string]*types.BuildTask {
	tidbVer := c.process.Config.TiDBConfig.Version
	tikvVer := c.process.Config.TiKVConfig.Version
	tikvFuse := c.process.Config.TiKVConfig.Fuse
	pdVer := c.process.Config.PDConfig.Version

	if !util.IsEmpty(c.process.RunVersion.TiDBVersion.Value) {
		tidbVer = c.process.RunVersion.TiDBVersion
	}
	if util.IsEmpty(tidbVer.Type) {
		tidbVer.Type = util.GitBranch
		tidbVer.Value = "master"
	}

	if !util.IsEmpty(c.process.RunVersion.TiKVVersion.Value) {
		tikvVer = c.process.RunVersion.TiKVVersion
	}
	if util.IsEmpty(tikvVer.Type) {
		tikvVer.Type = util.GitBranch
		tikvVer.Value = "master"
	}

	if !util.IsEmpty(c.process.RunVersion.PDVersion.Value) {
		pdVer = c.process.RunVersion.PDVersion
	}
	if util.IsEmpty(pdVer.Type) {
		pdVer.Type = util.GitBranch
		pdVer.Value = "master"
	}

	tasks := make(map[string]*types.BuildTask)

	tasks[tidbVer.Name] = newBuildTask(tidbVer, false)
	tasks[pdVer.Name] = newBuildTask(pdVer, false)
	tasks[tikvVer.Name] = newBuildTask(tikvVer, tikvFuse)

	return tasks
}

func genRunVersion(tasks map[string]*types.BuildTask) (types.RunVersion, error) {
	var runVer types.RunVersion

	for key, task := range tasks {
		versionInfo := types.VersionInfo{
			Name:          task.Information.Repo,
			Type:          util.GitHash,
			Value:         task.Information.Githash,
			CompileOption: task.Information.CompileOption,
			Platform:      task.Information.Platform,
			BuildTime:     task.Information.BuildTime,
		}
		switch key {
		case "tidb":
			runVer.TiDBVersion = versionInfo
		case "pd":
			runVer.PDVersion = versionInfo
		case "tikv":
			runVer.TiKVVersion = versionInfo
		default:
			return runVer, errors.NotSupportedf("task %s", key)
		}
	}

	return runVer, nil
}

func newBuildTask(ver types.VersionInfo, fuse bool) *types.BuildTask {
	branch, tag, hash := ver.GetBranchTagHash()

	return &types.BuildTask{
		Status: types.TaskNew,
		Information: types.Infomation{
			Repo:          ver.Name,
			Branch:        branch,
			Tag:           tag,
			Githash:       hash,
			Platform:      ver.Platform,
			CompileOption: ver.CompileOption,
			Fuse:          fuse,
		},
	}
}

func (c *Cat) runBuildTasks(
	ctx context.Context,
	tasks map[string]*types.BuildTask,
) error {
	g := errgroup.Group{}
	for _, task := range tasks {
		taskT := task
		g.Go(func() error {
			return errors.Trace(c.runBuildTaskAction(ctx, taskT))
		})
	}

	return errors.Trace(g.Wait())
}

func (c *Cat) runBuildTaskAction(ctx context.Context, task *types.BuildTask) error {
	if err := c.store.CreateBuildTask(task); err != nil {
		return errors.Trace(err)
	}

	cctx, cancel := context.WithTimeout(ctx, 20*time.Minute)
	defer cancel()

	if err := waitTaskFinished(cctx, c.store, task); err != nil {
		return errors.Trace(err)
	}

	return nil
}

func waitTaskFinished(
	ctx context.Context,
	store *storage.Storage,
	task *types.BuildTask,
) error {
	ticker := time.NewTicker(5 * time.Second)

	for {
		select {
		case <-ctx.Done():
			if ctx.Err() == context.Canceled {
				return context.Canceled
			}
			return errors.Errorf("run task %d timeout", task.ID)
		case <-ticker.C:
			taskT, ok := store.GetBuildTaskByID(task.ID)
			if !ok {
				return errors.Errorf("sync build task %d failed, error: can not get task from store", task.ID)
			}

			switch taskT.Status {
			case types.TaskRunning:
				continue
			case types.TaskFinished:
				if err := task.Copy(taskT); err != nil {
					return errors.Trace(err)
				}
				return nil
			case types.TaskFailed:
				return errors.Errorf("run task %d failed, build log: %s",
					taskT.ID, taskT.Information.BuildLog)
			}
		}
	}
}

func (c *Cat) upgrade(config *types.Cat) error {
	c.upgradeCatProcess(config)

	err := c.store.UpdateCatProcess(c.process)
	if err != nil {
		log.Warningf("save cat process error %v", err)
	}

	err = c.prepareImage()
	if err != nil {
		return errors.Trace(err)
	}

	c.upgradeClusterImage(config)
	return errors.Trace(c.operator.UpdateCluster(c.ctx, c.cluster, true))
}

func (c *Cat) upgradeCatProcess(config *types.Cat) {
	if !util.IsEmpty(config.Name) {
		c.process.Name = config.Name
	}

	c.process.Config.TiDBConfig = config.TiDBConfig
	c.process.Config.TiKVConfig = config.TiKVConfig
	c.process.Config.PDConfig = config.PDConfig
}

func (c *Cat) upgradeClusterImage(config *types.Cat) {
	c.cluster.TidbCluster.Spec.PD.ContainerSpec.Image = getImage("pd", c.process.RunVersion.PDVersion, false)
	c.cluster.TidbCluster.Spec.TiDB.ContainerSpec.Image = getImage("tidb", c.process.RunVersion.TiDBVersion, false)
	c.cluster.TidbCluster.Spec.TiKV.ContainerSpec.Image = getImage("tikv", c.process.RunVersion.TiKVVersion, config.TiKVConfig.Fuse)
	c.cluster.TidbCluster.Spec.PD.Replicas = config.PDConfig.Size
	c.cluster.TidbCluster.Spec.TiDB.Replicas = config.TiDBConfig.Size
	c.cluster.TidbCluster.Spec.TiKV.Replicas = config.TiKVConfig.Size

	c.cluster.TiDBConfig = genConfigContent(c.store, config.TiDBConfig)
	c.cluster.TiKVConfig = genConfigContent(c.store, config.TiKVConfig)
	c.cluster.PDConfig = genConfigContent(c.store, config.PDConfig)
}

func (c *Cat) setStage(stage types.CatStage) {
	c.process.SetStage(stage)
}

func (c *Cat) getStage() types.CatStage {
	return c.process.Stage
}

func (c *Cat) setState(state types.CatState) {
	c.process.SetState(state)
}

func (c *Cat) getState() types.CatState {
	return c.process.State
}

func (c *Cat) saveCat() error {
	c.process.Report = c.Report()
	return errors.Trace(c.store.UpdateCatProcess(c.process))
}

// Identify returns cat's id
func (c *Cat) Identify() int64 {
	return c.process.ID
}

// Namespace returns cat's namespace
func (c *Cat) Namespace() string {
	return c.process.Namespace
}

// Name returns cat's name
func (c *Cat) Name() string {
	return c.process.Name
}

// IsRunning returns true if cat is running
func (c *Cat) IsRunning() bool {
	return c.process.IsRunning()
}

// IsNew returns true if cat is new
func (c *Cat) IsNew() bool {
	return c.process.IsNew()
}

// IsCreating returns true if cat is creating
func (c *Cat) IsCreating() bool {
	return c.process.IsCreating()
}

// IsDestroyed returns true if cat is Destroyed
func (c *Cat) IsDestroyed() bool {
	return c.process.IsDestroyed()
}

// Config returns config version
func (c *Cat) Config() types.TiDBClusterConfig {
	return c.process.Config
}

// RunVersion returns run hash
func (c *Cat) RunVersion() types.RunVersion {
	return c.process.RunVersion
}

// SetOwner sets owner for cat
func (c *Cat) SetOwner(name string) {
	c.Lock()
	c.process.Owner = name
	c.Unlock()
}

// RemoveOwner removes owner for cat
func (c *Cat) RemoveOwner() {
	c.Lock()
	c.process.Owner = ""
	c.Unlock()
}

// GetOwner gets cat owner
func (c *Cat) GetOwner() (string, bool) {
	c.RLock()
	defer c.RUnlock()
	if !util.IsEmpty(c.process.Owner) {
		return c.process.Owner, true
	}
	return "", false
}

// Report reports cat
func (c *Cat) Report() *types.CatReport {
	if !c.IsDestroyed() && !c.IsNew() && !c.IsCreating() {
		c.syncStatus()
	}
	report := &types.CatReport{
		Stage:      c.process.Stage,
		State:      c.process.State,
		Destroyed:  c.process.Destroyed,
		ID:         c.process.ID,
		Name:       c.process.Name,
		Namespace:  c.process.Namespace,
		Creator:    c.process.Creator,
		TidbSize:   c.process.Config.TiDBConfig.Size,
		TikvSize:   c.process.Config.TiKVConfig.Size,
		PdSize:     c.process.Config.PDConfig.Size,
		Config:     c.process.Config,
		RunVersion: c.process.RunVersion,
	}

	if c.cluster != nil {
		report.Cluster = c.cluster.Report()
	}

	if c.process.Meta != nil {
		report.LabelStr = util.LabelsMapToLabels(c.process.Meta.Labels)
	}
	return report
}

// Nodes returns all nodes used by cat
func (c *Cat) Nodes() []string {
	var nodes []string
	for _, pod := range c.cluster.TidbStatus {
		nodes = append(nodes, pod.NodeIP)
	}

	for _, pod := range c.cluster.PdStatus {
		nodes = append(nodes, pod.NodeIP)
	}

	for _, pod := range c.cluster.TikvStatus {
		nodes = append(nodes, pod.NodeIP)
	}

	return util.RemoveRep(nodes)
}