package flow

import (
	"flag"
	"strconv"
	"strings"
	"validate/dbctrl"
	"validate/transactional_operation"
	"validate/util"

	"github.com/pingcap/errors"
)

type ComponentType int32

const (
	ComponentType_COMPONENT_TYPE_UNSPECIFIED ComponentType = 0
	ComponentType_TIKV                       ComponentType = 1
	ComponentType_TIDB                       ComponentType = 2
	ComponentType_TIFLASH                    ComponentType = 3
	ComponentType_TIFLASH_LEARNER            ComponentType = 4
	ComponentType_PD                         ComponentType = 5
	ComponentType_TIPROXY                    ComponentType = 6
)

// Enum value maps for ComponentType.
var (
	ComponentType_name = map[ComponentType]string{
		0: "COMPONENT_TYPE_UNSPECIFIED",
		1: "TIKV",
		2: "TIDB",
		3: "TIFLASH",
		4: "TIFLASH_LEARNER",
		5: "PD",
		6: "TIPROXY",
	}
	ComponentType_value = map[string]int32{
		"COMPONENT_TYPE_UNSPECIFIED": 0,
		"TIKV":                       1,
		"TIDB":                       2,
		"TIFLASH":                    3,
		"TIFLASH_LEARNER":            4,
		"PD":                         5,
		"TIPROXY":                    6,
	}
)

func VerifyDownstreamCommand() *VerifyDownstreamCommandCtx {
	ca := &VerifyDownstreamCommandCtx{
		ClusterMigrationSubCommandCtx{
			fs: flag.NewFlagSet(VerifyDownstreamClusterCommand, flag.ContinueOnError),
		},
	}

	ca.fs.StringVar(&ca.CommandTarget, "target", "all", "Sub-actions of the command.")

	ca.fs.StringVar(&ca.PrimaryAddress.Host, PrefixPrimaryClusterAddr+"host", "127.0.0.1", "Host address of the primary TiDB Cluster, 'localhost' as default.")
	ca.fs.IntVar(&ca.PrimaryAddress.Port, PrefixPrimaryClusterAddr+"port", 4000, "Port of the primary TiDB Cluster, 4000 as default.")
	ca.fs.StringVar(&ca.PrimaryAddress.Database, PrefixPrimaryClusterAddr+"dbname", "mysql", "Database name of the primary TiDB Cluster, 'mysql' as default.")
	ca.fs.StringVar(&ca.PrimaryAddress.Account, PrefixPrimaryClusterAddr+"account", "root", "Database account of the primary TiDB Cluster, 'root' as default.")
	ca.fs.StringVar(&ca.PrimaryAddress.Password, PrefixPrimaryClusterAddr+"password", "", "Host address of the primary TiDB Cluster, '' as default.")

	ca.fs.StringVar(&ca.DownstreamAddress.Host, PrefixDownstreamClusterAddr+"host", "127.0.0.1", "Host address of the primary TiDB Cluster, 'localhost' as default.")
	ca.fs.IntVar(&ca.DownstreamAddress.Port, PrefixDownstreamClusterAddr+"port", 4000, "Port of the primary TiDB Cluster, 4000 as default.")
	ca.fs.StringVar(&ca.DownstreamAddress.Database, PrefixDownstreamClusterAddr+"dbname", "mysql", "Database name of the primary TiDB Cluster, 'mysql' as default.")
	ca.fs.StringVar(&ca.DownstreamAddress.Account, PrefixDownstreamClusterAddr+"account", "root", "Database account of the primary TiDB Cluster, 'root' as default.")
	ca.fs.StringVar(&ca.DownstreamAddress.Password, PrefixDownstreamClusterAddr+"password", "", "Host address of the primary TiDB Cluster, '' as default.")

	return ca
}

type VerifyDownstreamCommandCtx struct {
	ClusterMigrationSubCommandCtx
}

func (vdc *VerifyDownstreamCommandCtx) verifyDBAccounts() error {
	util.Logger.Printf("-- DB Accounts validation between the primary and downstream.\n")
	primaryDBAccountList, err := transactional_operation.ExportAllDBUserDefinition(vdc.PrimaryDBConn)
	if err != nil {
		return err
	}

	downstreamDBAccountList, err := transactional_operation.ExportAllDBUserDefinition(vdc.DownstreamDBConn)
	if err != nil {
		return err
	}

	if len(primaryDBAccountList) != len(downstreamDBAccountList) {
		return errors.New("the count of databases is not identical, primary [" +
			strconv.Itoa(len(primaryDBAccountList)) + "] and downstream [" +
			strconv.Itoa(len(downstreamDBAccountList)) + "]")
	}

	for idx := 0; idx < len(primaryDBAccountList); idx += 1 {
		isEqual, differences, err := util.Compare(primaryDBAccountList[idx], downstreamDBAccountList[idx], nil)
		if !isEqual || err != nil {
			for _, diff := range differences {
				util.Logger.Printf(diff.String() + "\n")
			}
			break
		}
	}
	util.Logger.Printf("-- DB Accounts are identical between the two TiDB Cluster.\n")
	util.Logger.Printf("-- DB Accounts validation finished.\n")

	return nil
}

func (vdc *VerifyDownstreamCommandCtx) verifySchemas() error {
	util.Logger.Printf("-- Schemas validation between the primary and downstream.\n")
	var primaryDBList []*dbctrl.DatabaseInfo
	dbList, err := transactional_operation.ExportDatabaseList(vdc.PrimaryDBConn)
	if err != nil {
		return err
	}
	for _, db := range dbList {
		if db.IsSystemDatabase() {
			continue
		}
		dbInfo, err := transactional_operation.ExportDatabase(db, vdc.PrimaryAddress)
		if err != nil {
			return errors.Wrap(err, "failed to export database detail '"+db.SCHEMA_NAME+"'in primary TiDB Cluster")
		}
		primaryDBList = append(primaryDBList, dbInfo)
	}

	var downstreamDBList []*dbctrl.DatabaseInfo
	dbList, err = transactional_operation.ExportDatabaseList(vdc.DownstreamDBConn)
	if err != nil {
		return err
	}
	for _, db := range dbList {
		if db.IsSystemDatabase() {
			continue
		}
		dbInfo, err := transactional_operation.ExportDatabase(db, vdc.DownstreamAddress)
		if err != nil {
			return errors.Wrap(err, "failed to export database detail '"+db.SCHEMA_NAME+"'in downstream TiDB Cluster")
		}
		downstreamDBList = append(downstreamDBList, dbInfo)
	}

	var areEqual = true
	var missedSchemaList = make([]*dbctrl.DatabaseInfo, 0)
	var extraSchemaList = make([]*dbctrl.DatabaseInfo, 0)
	for primaryIdx := 0; primaryIdx < len(primaryDBList); primaryIdx += 1 {
		downstreamIdx := 0
		var found = false
		for ; downstreamIdx < len(downstreamDBList); downstreamIdx += 1 {
			if 0 == strings.Compare(primaryDBList[primaryIdx].SCHEMA_NAME, downstreamDBList[downstreamIdx].SCHEMA_NAME) {
				found = true
				break
			}
		}

		if found {
			isEqual, differences, err := util.Compare(primaryDBList[primaryIdx], downstreamDBList[downstreamIdx], nil)
			if !isEqual || err != nil {
				for _, diff := range differences {
					util.Logger.Printf(diff.String() + "\n")
				}
			}
			areEqual = areEqual && isEqual
		} else {
			missedSchemaList = append(missedSchemaList, primaryDBList[primaryIdx])
		}
	}
	for downstreamIdx := 0; downstreamIdx < len(downstreamDBList); downstreamIdx += 1 {
		primaryIdx := 0
		var found = false
		for primaryIdx = 0; primaryIdx < len(primaryDBList); primaryIdx += 1 {
			if 0 == strings.Compare(primaryDBList[primaryIdx].SCHEMA_NAME, downstreamDBList[downstreamIdx].SCHEMA_NAME) {
				found = true
				break
			}
		}

		if !found {
			extraSchemaList = append(extraSchemaList, downstreamDBList[downstreamIdx])
		}
	}

	if len(missedSchemaList) > 0 {
		areEqual = false
		util.Logger.Printf("-- Schemas only in primary TiDB Cluster:\n")
		for idx, dbInfo := range missedSchemaList {
			util.Logger.Printf("---- " + strconv.Itoa(idx+1) + ". " + dbInfo.PrintDatabase() + "\n")
		}
	}
	if len(extraSchemaList) > 0 {
		areEqual = false
		util.Logger.Printf("-- Schemas only in the downstream TiDB Cluster:\n")
		for idx, dbInfo := range extraSchemaList {
			util.Logger.Printf("---- " + strconv.Itoa(idx+1) + ". " + dbInfo.PrintDatabase())
		}
	}

	if areEqual {
		util.Logger.Printf("-- Schemas/Tables are identical between the two TiDB Cluster.\n")
	}

	util.Logger.Printf("-- Schemas validation finished.\n")
	return nil
}

func (vdc *VerifyDownstreamCommandCtx) verifyGlobalVariables() error {
	util.Logger.Printf("-- Global variables validation between the primary and downstream.\n")
	primaryVariableList, err := transactional_operation.GetGlobalVariableList(vdc.PrimaryDBConn)
	if err != nil {
		return err
	}

	downstreamVariableList, err := transactional_operation.GetGlobalVariableList(vdc.DownstreamDBConn)
	if err != nil {
		return err
	}

	if len(primaryVariableList) != len(downstreamVariableList) {
		return errors.New("the count of databases is not identical, primary [" +
			strconv.Itoa(len(primaryVariableList)) + "] and downstream [" +
			strconv.Itoa(len(downstreamVariableList)) + "]")
	}

	var diffCount = 0
	var diffVariableListInPrimary = make([]*dbctrl.GlobalVariables, 0)
	var diffVariableListInDownstream = make([]*dbctrl.GlobalVariables, 0)
	for idx := 0; idx < len(primaryVariableList); idx += 1 {
		if transactional_operation.IsVariableInWhiteList(primaryVariableList[idx]) {
			continue
		}

		isEqual, _, err := util.Compare(primaryVariableList[idx], downstreamVariableList[idx], nil)
		if !isEqual || err != nil {
			diffCount += 1
			diffVariableListInPrimary = append(diffVariableListInPrimary, primaryVariableList[idx])
			diffVariableListInDownstream = append(diffVariableListInDownstream, downstreamVariableList[idx])
		}
	}

	if diffCount > 0 {
		util.Logger.Printf("---- [" + strconv.Itoa(diffCount) + "] variables are NOT identical in total:")
		for idx := 0; idx < diffCount; idx += 1 {
			util.Logger.Printf("---- Name: \"" + diffVariableListInPrimary[idx].Name + "\",\t Value in primary: \"" +
				diffVariableListInPrimary[idx].Value + "\", Value in downstream: \"" +
				diffVariableListInDownstream[idx].Value + "\"")
		}
	} else {
		util.Logger.Printf("---- Global variables are identical between the primary and the downstream.\n")
	}

	util.Logger.Printf("-- Global variables validation finished.\n")
	return nil
}

func (vdc *VerifyDownstreamCommandCtx) verifyConfigurations() error {
	util.Logger.Printf("-- Configuration validation between the primary and the downstream.\n")
	primaryConfigList, err := transactional_operation.GetConfigurationList(vdc.PrimaryDBConn)
	if err != nil {
		return err
	}

	downstreamConfigList, err := transactional_operation.GetConfigurationList(vdc.DownstreamDBConn)
	if err != nil {
		return err
	}

	if len(primaryConfigList) != len(downstreamConfigList) {
		return errors.New("the count of databases is not identical, primary [" +
			strconv.Itoa(len(primaryConfigList)) + "] and downstream [" +
			strconv.Itoa(len(downstreamConfigList)) + "]")
	}

	var diffCount = 0
	var diffConfigListInPrimary = make([]*dbctrl.Configuration, 0)
	var diffConfigListInDownstream = make([]*dbctrl.Configuration, 0)
	for idx := 0; idx < len(primaryConfigList); idx += 1 {
		if transactional_operation.IsConfigInWhiteList(primaryConfigList[idx]) {
			continue
		}

		isEqual, differences, err := util.Compare(primaryConfigList[idx], downstreamConfigList[idx], []string{"Instance"})
		if !isEqual || err != nil {
			diffConfigListInPrimary = append(diffConfigListInPrimary, primaryConfigList[idx])
			diffConfigListInDownstream = append(diffConfigListInDownstream, downstreamConfigList[idx])
			diffCount += 1

			for _, diff := range differences {
				util.Logger.Debugf(diff.String() + "\n")
			}
		}
	}

	if diffCount > 0 {
		util.Logger.Printf("---- [" + strconv.Itoa(diffCount) + "] configurations are NOT identical in total:\n")
		for idx := 0; idx < diffCount; idx += 1 {
			util.Logger.Printf("---- {Type:\"" + diffConfigListInPrimary[idx].Type +
				"\", Name: \"" + diffConfigListInPrimary[idx].Name + "\", }, Value in primary: \"" +
				diffConfigListInPrimary[idx].Value + "\", Value in downstream: \"" +
				diffConfigListInDownstream[idx].Value + "\"\n")
		}
	} else {
		util.Logger.Printf("---- All configurations are identical between the primary and the downstream.\n")
	}

	util.Logger.Printf("-- Configuration validation finished.\n")

	return nil
}

func (vdc *VerifyDownstreamCommandCtx) verifyPlacementRules() error {
	util.Logger.Printf("-- Placement Rules validation between the primary and the downstream.\n")
	primaryPlacementList, err := transactional_operation.GetPlacementPolicyList(vdc.PrimaryDBConn)
	if err != nil {
		return err
	}

	downstreamPlacementList, err := transactional_operation.GetPlacementPolicyList(vdc.DownstreamDBConn)
	if err != nil {
		return err
	}

	if len(primaryPlacementList) != len(downstreamPlacementList) {
		return errors.New("the count of databases is not identical, primary [" +
			strconv.Itoa(len(primaryPlacementList)) + "] and downstream [" +
			strconv.Itoa(len(downstreamPlacementList)) + "]")
	}

	var diffCount = 0
	for idx := 0; idx < len(primaryPlacementList); idx += 1 {
		isEqual, differences, err := util.Compare(primaryPlacementList[idx], downstreamPlacementList[idx], nil)
		if !isEqual || err != nil {
			for _, diff := range differences {
				util.Logger.Printf(diff.String())
			}
			diffCount += 1
		}
	}

	if diffCount > 0 {
		util.Logger.Printf("---- [" + strconv.Itoa(diffCount) + "] Placement Rules are NOT identical in total.\n")
	} else {
		util.Logger.Printf("---- All Placement Rules are identical between the primary and the downstream.\n")
	}

	util.Logger.Printf("-- Placement Rules validation finished.\n")
	return nil
}

func (vdc *VerifyDownstreamCommandCtx) verifySQLBinding() error {
	util.Logger.Printf("-- SQL Binding validation between the primary and the downstream.\n")
	primarySQLBindingList, err := transactional_operation.GetSQLBindingList(vdc.PrimaryDBConn)
	if err != nil {
		return err
	}

	downstreamSQLBindingList, err := transactional_operation.GetSQLBindingList(vdc.DownstreamDBConn)
	if err != nil {
		return err
	}

	if len(primarySQLBindingList) != len(downstreamSQLBindingList) {
		return errors.New("the count of databases is not identical, primary [" +
			strconv.Itoa(len(primarySQLBindingList)) + "] and downstream [" +
			strconv.Itoa(len(downstreamSQLBindingList)) + "]")
	}

	var diffCount = 0
	for idx := 0; idx < len(primarySQLBindingList); idx += 1 {
		isEqual, differences, err := util.Compare(primarySQLBindingList[idx], downstreamSQLBindingList[idx], nil)
		if !isEqual || err != nil {
			for _, diff := range differences {
				util.Logger.Printf(diff.String())
			}
			diffCount += 1
		}
	}

	if diffCount > 0 {
		util.Logger.Printf("---- [" + strconv.Itoa(diffCount) + "] SQL Bindings are NOT identical in total.\n")
	} else {
		util.Logger.Printf("---- All SQL Bindings are identical between the primary and the downstream.\n")
	}
	util.Logger.Printf("-- SQL Binding validation finished.\n")
	return nil
}

func (vdc *VerifyDownstreamCommandCtx) verifyAll() error {
	err := vdc.verifyDBAccounts()
	if err != nil {
		return err
	}
	err = vdc.verifySchemas()
	if err != nil {
		return err
	}
	err = vdc.verifyGlobalVariables()
	if err != nil {
		return err
	}
	err = vdc.verifyConfigurations()
	if err != nil {
		return err
	}
	err = vdc.verifyPlacementRules()
	if err != nil {
		return err
	}
	err = vdc.verifySQLBinding()
	if err != nil {
		return err
	}

	return nil
}

func (vdc *VerifyDownstreamCommandCtx) Name() string {
	return vdc.fs.Name()
}

func (vdc *VerifyDownstreamCommandCtx) Init(args []string) error {
	return vdc.fs.Parse(args)
}

func (vdc *VerifyDownstreamCommandCtx) Execute() error {
	return vdc.VerifyDownstreamCommandMain()
}

func (vdc *VerifyDownstreamCommandCtx) VerifyDownstreamCommandMain() (err error) {
	vdc.PrimaryDBConnPool, err = dbctrl.NewDBConnInfoByAdd(&vdc.PrimaryAddress)
	if err != nil {
		return err
	}
	vdc.PrimaryDBConn, err = vdc.PrimaryDBConnPool.Connect()
	if err != nil {
		return err
	}
	defer vdc.PrimaryDBConn.Close()

	vdc.DownstreamDBConnPool, err = dbctrl.NewDBConnInfoByAdd(&vdc.DownstreamAddress)
	if err != nil {
		return err
	}
	vdc.DownstreamDBConn, err = vdc.DownstreamDBConnPool.Connect()
	if err != nil {
		return err
	}
	defer vdc.PrimaryDBConn.Close()

	switch vdc.CommandTarget {
	case CommandTargetAll:
		return vdc.verifyAll()
	case CommandTargetDBAccount:
		return vdc.verifyDBAccounts()
	case CommandTargetSchema:
		return vdc.verifySchemas()
	case CommandTargetGlobalVariable:
		return vdc.verifyGlobalVariables()
	case CommandTargetConfiguration:
		return vdc.verifyConfigurations()
	case CommandTargetPlacementRule:
		return vdc.verifyPlacementRules()
	case CommandTargetSQLBinding:
		return vdc.verifySQLBinding()
	default:
		return nil
	}
}