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 dropDatabases struct {
	res        *apiresource.ResourceHandle
	dbNameList *string
}

func (d dropDatabases) String() string {
	return fmt.Sprintf("Drop databases `%s` on resource `%s`", *(d.dbNameList), d.res.Name)
}

func (d dropDatabases) Execute(ctx context.Context) error {
	resolved, err := core.ResolveResourceHandle(d.res)
	if err != nil {
		return err
	}
	mysql := resolved.(resource.MySQLCompatible)
	db, err := mysql.GetDB("")
	if err != nil {
		return errors.Trace(err)
	}
	for _, dbName := range strings.Split(*d.dbNameList, ",") {
		// skip if dbName is not provided
		if dbName == "" {
			continue
		}
		_, err = db.Exec(fmt.Sprintf("drop database if exists `%s`", dbName))
		if err != nil {
			log.Error("drop database error", zap.Error(err))
			return errors.Trace(err)
		}
	}
	return nil
}

var _ Step = &dropDatabases{}

func DropDatabasesIfExists(res *apiresource.ResourceHandle, dbNameList *string) Step {
	return dropDatabases{
		res:        res,
		dbNameList: dbNameList,
	}
}

func (b *Builder) DropDatabasesIfExists(res *apiresource.ResourceHandle, dbList *string) *Builder {
	b.steps = append(b.steps, DropDatabasesIfExists(res, dbList))
	return b
}

func NewDropDatabasesIfExists(res *apiresource.ResourceHandle, dbList *string) Step {
	return &dropDatabases{res: res, dbNameList: dbList}
}

type createDatabase struct {
	handle *apiresource.ResourceHandle
	dbName string
}

func NewCreateDatabase(res *apiresource.ResourceHandle, dbName string) Step {
	return &createDatabase{
		handle: res,
		dbName: dbName,
	}
}

func (s *createDatabase) Execute(ctx context.Context) error {
	resolved, err := core.ResolveResourceHandle(s.handle)
	if err != nil {
		return err
	}

	mysql := resolved.(resource.MySQLCompatible)
	db, err := mysql.GetDB("")
	if err != nil {
		return errors.Trace(err)
	}
	defer func() {
		if err := db.Close(); err != nil {
			log.Warn("close db failed", zap.Error(err))
		}
	}()

	query := fmt.Sprintf("create database if not exists %s", s.dbName)
	_, err = db.Exec(query)

	return errors.Trace(err)
}

func (s *createDatabase) String() string {
	return fmt.Sprintf("createDatabase: mysqlHandler=%+v, dbName=%+v", s.handle, s.dbName)
}

// Create database in tiup cluster deployed TiDB.
type createDatabaseTiup struct {
	dbName string
}

func NewCreateDatabaseTiup(dbName string) Step {
	return &createDatabaseTiup{
		dbName: dbName,
	}
}

func (s *createDatabaseTiup) Execute(ctx context.Context) error {
	db, err := tiupTiDBCluster.GetDB("")
	if err != nil {
		return errors.Trace(err)
	}
	defer func() {
		if err := db.Close(); err != nil {
			log.Warn("close db failed", zap.Error(err))
		}
	}()

	query := fmt.Sprintf("create database if not exists %s", s.dbName)
	_, err = db.Exec(query)

	return errors.Trace(err)
}

func (s *createDatabaseTiup) String() string {
	return fmt.Sprintf("createDatabaseTiup: dbName=%+v", s.dbName)
}

// Check database available
type checkDatabaseAvailable struct {
	handle          *apiresource.ResourceHandle
	dbName          string
	filterTablename *string
}

var _ Step = &checkDatabaseAvailable{}

func NewCheckDatabaseAvailable(res *apiresource.ResourceHandle, dbName string, filterTablename *string) Step {
	return &checkDatabaseAvailable{
		handle:          res,
		dbName:          dbName,
		filterTablename: filterTablename,
	}
}

func (s *checkDatabaseAvailable) Execute(ctx context.Context) error {
	resolved, err := core.ResolveResourceHandle(s.handle)
	if err != nil {
		return err
	}

	mysql := resolved.(resource.MySQLCompatible)
	db, err := mysql.GetDB("")
	if err != nil {
		return err
	}

	defer func() {
		if err := db.Close(); err != nil {
			log.Warn("close db failed", zap.Error(err))
		}
	}()
	// use select COUNT(*) from dbname.tablename to check if the db&table available
	query := fmt.Sprintf("select COUNT(*) from %s.%s", s.dbName, *(s.filterTablename))
	_, err = db.Exec(query)

	return errors.Trace(err)
}

func (s *checkDatabaseAvailable) String() string {
	return fmt.Sprintf("checkDatabaseAvailable: kafkaHandler=%+v, dbName=%+v, tableName=%+v", s.handle, s.dbName, *(s.filterTablename))
}

// Execute sql statement
type executeSqlStatement struct {
	handle     *apiresource.ResourceHandle
	statements *string
}

var _ Step = &executeSqlStatement{}

func NewExecuteSqlStatement(res *apiresource.ResourceHandle, statements *string) Step {
	return &executeSqlStatement{
		handle:     res,
		statements: statements,
	}
}

func (s *executeSqlStatement) Execute(ctx context.Context) error {
	resolved, err := core.ResolveResourceHandle(s.handle)
	if err != nil {
		return err
	}

	mysql := resolved.(resource.MySQLCompatible)
	db, err := mysql.GetDB("")
	if err != nil {
		return err
	}

	defer func() {
		if err := db.Close(); err != nil {
			log.Warn("close db failed", zap.Error(err))
		}
	}()
	// Execute sql statements
	query := fmt.Sprintf("%s", *(s.statements))
	_, err = db.Exec(query)

	return errors.Trace(err)
}

func (s *executeSqlStatement) String() string {
	return fmt.Sprintf("Execute Sql Statement: kafkaHandler=%+v, statement=%+v", s.handle, *(s.statements))
}

type getUserDatabases struct {
	mysqlRes  *apiresource.ResourceHandle
	databases *[]string
}

func (g getUserDatabases) String() string {
	return fmt.Sprintf("Get databases from %s", g.mysqlRes.Name)
}

func (g getUserDatabases) Execute(ctx context.Context) (err error) {
	mysqlInst := utils.Try(core.ResolveResourceHandle(g.mysqlRes)).(resource.MySQLCompatible)
	db := utils.Try(mysqlInst.GetDB("")).(*sql.DB)
	defer db.Close()
	client := mysql.NewMySQLClient(db)
	*g.databases, err = client.GetUserDatabases()
	if err != nil {
		log.Error("Get user databases failed", zap.Error(err))
	}
	return errors.Trace(err)
}

var _ Step = &getUserDatabases{}

func GetUserDatabases(mysqlRes *apiresource.ResourceHandle, databases *[]string) Step {
	return &getUserDatabases{
		mysqlRes,
		databases,
	}
}

func (b *Builder) GetUserDatabases(mysqlRes *apiresource.ResourceHandle, databases *[]string) *Builder {
	b.steps = append(b.steps, GetUserDatabases(mysqlRes, databases))
	return b
}

type dropUserDatabases struct {
	mysqlRes *apiresource.ResourceHandle
}

func (d dropUserDatabases) String() string {
	return fmt.Sprintf("drop user databases on %s", d.mysqlRes.Name)
}

func (d dropUserDatabases) Execute(ctx context.Context) error {
	mysqlInst := utils.Try(core.ResolveResourceHandle(d.mysqlRes)).(resource.MySQLCompatible)
	db := utils.Try(mysqlInst.GetDB("")).(*sql.DB)
	defer db.Close()
	client := mysql.NewMySQLClient(db)
	err := client.DropUserDatabases()
	if err != nil {
		log.Error("Drop user databases failed", zap.Error(err))
	}
	return errors.Trace(err)
}

var _ Step = &dropUserDatabases{}

func DropUserDatabases(mysqlRes *apiresource.ResourceHandle) Step {
	return &dropUserDatabases{mysqlRes: mysqlRes}
}

func (b *Builder) DropUserDatabases(mysqlRes *apiresource.ResourceHandle) *Builder {
	b.steps = append(b.steps, DropUserDatabases(mysqlRes))
	return b
}

// Execute sql statements
type executeSqlStatements struct {
	handle     *apiresource.ResourceHandle
	statements *string
}

// var Step = &executeSqlStatements{}

func NewExecuteSqlStatements(res *apiresource.ResourceHandle, statements *string) Step {
	return &executeSqlStatements{
		handle:     res,
		statements: statements,
	}
}

func (s *executeSqlStatements) Execute(ctx context.Context) error {
	resolved, err := core.ResolveResourceHandle(s.handle)
	if err != nil {
		return err
	}

	mysql := resolved.(resource.MySQLCompatible)
	db, err := mysql.GetDB("")
	if err != nil {
		return err
	}

	defer func() {
		if err := db.Close(); err != nil {
			log.Warn("close db failed", zap.Error(err))
		}
	}()
	// Execute sql statements
	sqls := strings.Split(*(s.statements), "|")
	for _, query := range sqls {
		sql := fmt.Sprintf("%s", query)
		// query := fmt.Sprintf("%s", s.statement)
		_, err = db.Exec(sql)

	}
	return errors.Trace(err)
}

func (s *executeSqlStatements) String() string {
	return fmt.Sprintf("Execute Sql Statement: mysqlHandler=%+v, statement=%+v", s.handle, *(s.statements))
}

// insert result to specified database
type executeSqlOnPulibcDatabase struct {
	dsn          *string
	sqlStatement *string
}

// var Step = &executeSqlStatements{}

func NewExecuteSqlOnPulibcDatabase(dsn, sqlStatement *string) Step {
	return &executeSqlOnPulibcDatabase{
		dsn:          dsn,
		sqlStatement: sqlStatement,
	}
}

func (s *executeSqlOnPulibcDatabase) Execute(ctx context.Context) error {
	db, err := sql.Open("mysql", *s.dsn)
	if err != nil {
		return errors.Trace(err)
	}
	defer func() {
		if err := db.Close(); err != nil {
			log.Warn("close db failed", zap.Error(err))
		}
	}()
	sqls := strings.Split(*(s.sqlStatement), "|")
	for _, query := range sqls {
		sql := fmt.Sprintf("%s", query)
		// query := fmt.Sprintf("%s", s.statement)
		_, err = db.Exec(sql)

	}
	return errors.Trace(err)
}

func (s *executeSqlOnPulibcDatabase) String() string {
	return fmt.Sprintf("Execute Sql Statement: dsn=%s, statement=%s", *(s.dsn), *(s.sqlStatement))
}

func (b *Builder) ExecuteSqlOnPulibcDatabase(dsn, sqlStatement *string) *Builder {
	b.steps = append(b.steps, NewExecuteSqlOnPulibcDatabase(dsn, sqlStatement))
	return b
}

type createDatabases struct {
	res        *apiresource.ResourceHandle
	dbNameList *string // dbs like db1,db2
}

func (c createDatabases) String() string {
	return fmt.Sprintf("Create databases `%s` on resource `%s`", *(c.dbNameList), c.res.Name)
}

func (c createDatabases) Execute(ctx context.Context) error {
	resolved, err := core.ResolveResourceHandle(c.res)
	if err != nil {
		return err
	}
	mysql := resolved.(resource.MySQLCompatible)
	db, err := mysql.GetDB("")
	if err != nil {
		return errors.Trace(err)
	}
	for _, dbName := range strings.Split(*c.dbNameList, ",") {
		// skip if dbName is not provided
		if dbName == "" {
			continue
		}
		_, err = db.Exec(fmt.Sprintf("create database if not exists `%s`", dbName))
		if err != nil {
			log.Error("create database error", zap.Error(err))
			return errors.Trace(err)
		}
	}
	return nil
}

func (b *Builder) CreateDatabasesIfNotExists(res *apiresource.ResourceHandle, dbList *string) *Builder {
	b.steps = append(b.steps, createDatabases{res, dbList})
	return b
}