package steps

import (
	"context"
	"database/sql"
	"fmt"
	"strings"

	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/caselib/pkg/mysql"
	"github.com/pingcap/test-infra/caselib/pkg/utils"
	apiresource "github.com/pingcap/test-infra/common/model/resource"
	"github.com/pingcap/test-infra/sdk/core"
	"github.com/pingcap/test-infra/sdk/resource"
	"go.uber.org/zap"
)

type executeSQL struct {
	res    *apiresource.ResourceHandle
	dbName string
	query  string
}

func (e executeSQL) String() string {
	return fmt.Sprintf("Execute query `%s` in database `%s` on resource `%s`", e.query, e.dbName, e.res.Name)
}

func (e executeSQL) Execute(ctx context.Context) error {
	resolved, err := core.ResolveResourceHandle(e.res)
	if err != nil {
		return err
	}
	mysql := resolved.(resource.MySQLCompatible)
	db, err := mysql.GetDB(e.dbName)
	if err != nil {
		return err
	}
	defer func() {
		if err := db.Close(); err != nil {
			log.Warn("close db failed", zap.Error(err))
		}
	}()

	_, err = db.Exec(e.query)
	if err != nil {
		log.Error("Execute sql error", zap.Error(err))
		return errors.Trace(err)
	}
	return nil
}

var _ Step = &executeSQL{}

func NewExecuteSQL(res *apiresource.ResourceHandle, dbName, query string) Step {
	return executeSQL{
		res:    res,
		dbName: dbName,
		query:  query,
	}
}

func (b *Builder) ExecuteSQL(res *apiresource.ResourceHandle, dbName, query string) *Builder {
	b.steps = append(b.steps, NewExecuteSQL(res, dbName, query))
	return b
}

type getMysqlNow struct {
	res      *apiresource.ResourceHandle
	mysqlNow *string
}

func (g getMysqlNow) String() string {
	return fmt.Sprintf("Get current time for resource `%s`", g.res.Name)
}

func (g getMysqlNow) Execute(ctx context.Context) error {
	mysqlInst := utils.Try(core.ResolveResourceHandle(g.res)).(resource.MySQLCompatible)
	db := utils.Try(mysqlInst.GetDB("")).(*sql.DB)
	defer db.Close()
	client := mysql.NewMySQLClient(db)
	var now string
	err := client.QueryRow("select now();").Scan(&now)
	if err != nil {
		log.Error("select now() error", zap.Error(err))
		return errors.Trace(err)
	} else {
		*(g.mysqlNow) = now
		log.Info(fmt.Sprintf("Get mysql now(): %s", now))
	}
	return nil
}

func (b *Builder) GetMysqlNow(res *apiresource.ResourceHandle, now *string) *Builder {
	b.steps = append(b.steps, getMysqlNow{res: res, mysqlNow: now})
	return b
}

// check expectedRows against lightning_task_info.conflict_error_v1
type checkLightningDuplicate struct {
	expectedRows *int32
	res          *apiresource.ResourceHandle
}

func (c checkLightningDuplicate) String() string {
	return fmt.Sprintf("Check lightning duplicate detection rows, expected: %d", *c.expectedRows)
}

func (c checkLightningDuplicate) Execute(ctx context.Context) error {
	mysqlInst := utils.Try(core.ResolveResourceHandle(c.res)).(resource.MySQLCompatible)
	db := utils.Try(mysqlInst.GetDB("")).(*sql.DB)
	defer db.Close()
	client := mysql.NewMySQLClient(db)
	var dupRowsNum int32
	err := client.QueryRow("select count(task_id) from lightning_task_info.conflict_error_v1 group by task_id limit 1;").Scan(&dupRowsNum)
	if err != nil {
		log.Warn("Get lightning duplicate rows error, force set it to 0.", zap.Error(err))
		dupRowsNum = 0
	} else {
		log.Info(fmt.Sprintf("Duplicated rows in lightning_task_info.conflict_error_v1: %d", dupRowsNum))
	}
	if *c.expectedRows != dupRowsNum {
		return fmt.Errorf("lightning duplicate row check failed: expected: %d, actual: %d", c.expectedRows, dupRowsNum)
	}
	return nil
}

func (b *Builder) CheckLightningDuplicate(expectedRows *int32, res *apiresource.ResourceHandle) *Builder {
	b.steps = append(b.steps, checkLightningDuplicate{expectedRows: expectedRows, res: res})
	return b
}

// check expected max_error rows against target schema like lightning_task_info.type_error_v1
type checkLightningMaxError struct {
	expectedRows *int32
	schema       *string // parameter example: lightning_task_info.type_error_v1
	res          *apiresource.ResourceHandle
}

func (c checkLightningMaxError) String() string {
	return fmt.Sprintf("Check lightning max-error results, expected: %d against schema %s on TiDB: %s", *c.expectedRows, *c.schema, c.res.Name)
}

func (c checkLightningMaxError) Execute(ctx context.Context) error {
	// SQL := fmt.Sprintf("select count(task_id) from %s group by task_id limit 1", *c.schema)
	SQL := fmt.Sprintf("select count(task_id) from %s group by task_id", *c.schema)
	var actualRows int32
	err := ExecuteSqlGetResult(c.res, SQL).Scan(&actualRows)
	if err != nil {
		log.Warn("Get max_error from tidb failed, force set it to 0.", zap.Error(err))
		actualRows = 0
	} else {
		log.Info(fmt.Sprintf("max-error row in tidb: %d", actualRows))
	}
	if *c.expectedRows != actualRows {
		return fmt.Errorf("lightning max-error check failed, expected: %d, actual: %d", *c.expectedRows, actualRows)
	}
	return nil
}

func (b *Builder) CheckLightningMaxError(expectedRows *int32, schema *string, res *apiresource.ResourceHandle) *Builder {
	b.steps = append(b.steps, checkLightningMaxError{expectedRows: expectedRows, schema: schema, res: res})
	return b
}

func ExecuteSqlGetResult(res *apiresource.ResourceHandle, statements string) *sql.Row {
	mysqlInst := utils.Try(core.ResolveResourceHandle(res)).(resource.MySQLCompatible)
	db := utils.Try(mysqlInst.GetDB("")).(*sql.DB)
	defer db.Close()
	client := mysql.NewMySQLClient(db)
	return client.QueryRow(statements)
}

// check whether specified table is cached
type checkTableCache struct {
	isCached  *bool
	schema    string
	tableName string
	res       *apiresource.ResourceHandle
}

func (c checkTableCache) String() string {
	return fmt.Sprintf("Check whether table %s is cached agaist tidb %s, expected: %t", c.tableName, c.res.Name, *c.isCached)
}

func (c checkTableCache) Execute(ctx context.Context) error {
	var tableID, actual, expected int32
	if *c.isCached {
		expected = 1
	} else {
		expected = 0
	}

	err := ExecuteSqlGetResult(c.res, fmt.Sprintf("SELECT TIDB_TABLE_ID FROM `INFORMATION_SCHEMA`.`tables` WHERE `TABLE_SCHEMA`=\"%s\" and `TABLE_NAME` = \"%s\";", c.schema, c.tableName)).Scan(&tableID)
	if err != nil {
		log.Error(fmt.Sprintf("Failed to get table id for schema: %s, table : %s.", c.schema, c.tableName), zap.Error(err))
		return errors.New("checkTableCache failed")
	}
	err = ExecuteSqlGetResult(c.res, fmt.Sprintf("select count(tid) from mysql.table_cache_meta where tid=%d;", tableID)).Scan(&actual)
	if err != nil {
		log.Error(fmt.Sprintf("Failed to get info from table_cache_meta for table id for schema: %s, table : %s.", c.schema, c.tableName), zap.Error(err))
		return errors.New("checkTableCache failed")
	} else if actual != expected {
		log.Error(fmt.Sprintf("Failed to valid cache info for table id for schema: %s, table : %s", c.schema, c.tableName), zap.Error(err))
		return errors.New("checkTableCache failed")
	}
	return nil
}

func (b *Builder) CheckTableCache(isCached *bool, schema, tableName string, res *apiresource.ResourceHandle) *Builder {
	b.steps = append(b.steps, checkTableCache{isCached: isCached, schema: schema, tableName: tableName, res: res})
	return b
}

// DatabasesExists Cases parameters for step CheckDatabasesExist, example {"databases": "sbtest,test", "exist": true}
type DatabasesExists struct {
	Databases string `json:"databases"`
	Exist     bool   `json:"exist"`
}

// check whether specified schema exists
type checkDatabasesExist struct {
	exist     *bool
	databases *string
	res       *apiresource.ResourceHandle
}

func (c checkDatabasesExist) String() string {
	return fmt.Sprintf("Checking whether databases %s exist in MySQL: %s, expected: %t", *c.databases, c.res.Name, *c.exist)
}

func (c checkDatabasesExist) Execute(ctx context.Context) error {
	if *c.databases == "" {
		log.Warn("databases is empty, skip the check.")
		return nil
	}
	databases := strings.Split(*c.databases, ",")
	for _, database := range databases {
		if CheckDatabaseExist(database, c.res) != *c.exist {
			return errors.New(fmt.Sprintf("checkDatabasesExist failed"))
		}
	}
	return nil
}

func (b *Builder) CheckDatabasesExist(exist *bool, databases *string, res *apiresource.ResourceHandle) *Builder {
	b.steps = append(b.steps, checkDatabasesExist{exist: exist, databases: databases, res: res})
	return b
}

// CheckDatabaseExist check whether a given database exists
func CheckDatabaseExist(database string, res *apiresource.ResourceHandle) bool {
	SQL := fmt.Sprintf("SHOW DATABASES LIKE '%s';", database)
	var checkResult string
	err := ExecuteSqlGetResult(res, SQL).Scan(&checkResult)
	if err != nil {
		if strings.Contains(err.Error(), "no rows in result set") {
			return false
		} else {
			log.Error(fmt.Sprintf("checkDatabaseExist failed, failed to execute SQL: %s", SQL), zap.Error(err))
			return false
		}
	} else if checkResult != database {
		log.Error(fmt.Sprintf("checkDatabaseExist failed, database: %s not exists", SQL), zap.Error(err))
		return false
	}
	return true
}

type Tables struct {
	Tables []string `json:"tables"`
}

// check whether specified tables placement options.
type checkTablesPlacementOptions struct {
	tables               *[]string
	withPlacementOptions *bool
	res                  *apiresource.ResourceHandle
}

func (c checkTablesPlacementOptions) String() string {
	return fmt.Sprintf("Check whether tables %s have placement options, expected: %t", *c.tables, *c.withPlacementOptions)
}

func (c checkTablesPlacementOptions) Execute(ctx context.Context) error {
	for _, table := range *c.tables {
		var target, placement, schedulingStat string
		sql := fmt.Sprintf("show placement for table %s;", table)
		err := ExecuteSqlGetResult(c.res, sql).Scan(&target, &placement, &schedulingStat)
		if err != nil {
			if *c.withPlacementOptions == false && strings.Contains(err.Error(), "no rows in result set") {
				return nil
			} else {
				log.Info("checkTablesPlacementOptions failed")
				return err
			}
		} else {
			if *c.withPlacementOptions == true && strings.Contains(target, table) {
				return nil
			} else {
				log.Info("checkTablesPlacementOptions failed")
				return errors.New(fmt.Sprintf("checkTablesPlacementOptions failed for table %s, expected: %t", table, *c.withPlacementOptions))
			}
		}
	}
	return nil
}

func (b *Builder) CheckTablesPlacementOptions(TablesPlacementOptions *Tables, WithPlacementOptions *bool, res *apiresource.ResourceHandle) *Builder {
	b.steps = append(b.steps, checkTablesPlacementOptions{tables: &TablesPlacementOptions.Tables, withPlacementOptions: WithPlacementOptions, res: res})
	return b
}

type QueryExecptedResult struct {
	Query    string `json:"query"`
	Expected string `json:"expected"`
}

var QueryExecptedResultList []QueryExecptedResult

// Run queries and verify results
// Notes:
//   1. query should return only one column
//   2. set expected to "" if the query doesn't have return.
// queryExpectedResultList example:
//   [{"query": "select TABLE_SCHEMA from information_schema.tiflash_replica where TABLE_SCHEMA='tiflash_replica_table';", "expected": ""}, {"query": "set tidb_isolation_read_engines='tiflash'", "expected": ""}, {"query": "select count(*) from tiflash_replica_table.t1;", "expected": "2"}]

type verifyQueryResult struct {
	queryExpectedResultList *[]QueryExecptedResult
	res                     *apiresource.ResourceHandle
}

func (v verifyQueryResult) String() string {
	return fmt.Sprintf("Run verifyQueryResult, queryExpectedResultList: %v, tidb: %s", v.queryExpectedResultList, v.res.Name)
}

func (v verifyQueryResult) Execute(ctx context.Context) error {
	for _, queryExpectedResult := range *v.queryExpectedResultList {
		log.Info(fmt.Sprintf("query: %s, expected: %s", queryExpectedResult.Query, queryExpectedResult.Expected))
		var actual string
		err := ExecuteSqlGetResult(v.res, queryExpectedResult.Query).Scan(&actual)
		if err != nil {
			if strings.Contains(err.Error(), "no rows in result set") && queryExpectedResult.Expected == "" {
				log.Info("verify query result OK")
			} else {
				return err
			}
		} else {
			if queryExpectedResult.Expected == actual {
				log.Info("verify query result OK")
			} else {
				log.Info("verify query result failed")
				return errors.New(fmt.Sprintf("verify query result failed, expected: %v, actual %v",
					queryExpectedResult.Expected, actual))
			}
		}
	}
	return nil
}

func (b *Builder) VerifyQueryResult(queryExpectedResultList *[]QueryExecptedResult, res *apiresource.ResourceHandle) *Builder {
	b.steps = append(b.steps, verifyQueryResult{queryExpectedResultList: queryExpectedResultList, res: res})
	return b
}

type getTsNow struct {
	res               *apiresource.ResourceHandle
	ts                *string
	timeHumanReadable *string
}

func (g getTsNow) String() string {
	return fmt.Sprintf("Get ts now by running show master status")
}

func (g getTsNow) Execute(ctx context.Context) error {
	mysqlInst := utils.Try(core.ResolveResourceHandle(g.res)).(resource.MySQLCompatible)
	db := utils.Try(mysqlInst.GetDB("")).(*sql.DB)
	defer db.Close()
	client := mysql.NewMySQLClient(db)
	var tsNow, ignore, tsHumanReadable string
	err := client.QueryRow("show master status;").Scan(&ignore, &tsNow, &ignore, &ignore, &ignore)
	if err != nil {
		log.Error("show master status", zap.Error(err))
		return errors.Trace(err)
	} else {
		*(g.ts) = tsNow
		sql := fmt.Sprintf("select TIDB_PARSE_TSO(%s);", tsNow)
		err = client.QueryRow(sql).Scan(&tsHumanReadable)
		if err != nil {
			log.Error("TIDB_PARSE_TSO error", zap.Error(err))
			return errors.Trace(err)
		} else {
			*(g.timeHumanReadable) = tsHumanReadable
			log.Info(fmt.Sprintf("show master status, current tso: %s, time: %s", tsNow, tsHumanReadable))
		}

	}
	return nil
}

func (b *Builder) GetTsNow(res *apiresource.ResourceHandle, ts *string, timeHumanReadable *string) *Builder {
	b.steps = append(b.steps, &getTsNow{res, ts, timeHumanReadable})
	return b
}

func NewGetTsNow(res *apiresource.ResourceHandle, ts *string, timeHumanReadable *string) Step {
	return &getTsNow{res, ts, timeHumanReadable}
}