package mysql

import (
	"database/sql"
	"database/sql/driver"
	"fmt"
	"strconv"
	"strings"
	"time"

	"github.com/go-sql-driver/mysql"
	"github.com/juju/errors"
	"github.com/ngaut/log"
	tmysql "github.com/pingcap/tidb/mysql"
	"github.com/pingcap/wormhole/pkg/util"
	"github.com/pingcap/wormhole/pkg/whpb"
	gmysql "github.com/siddontang/go-mysql/mysql"
	"github.com/siddontang/go-mysql/schema"
)

const (
	defaultDialTimeout = "5s"
	maxRetryCount      = 100
	retryTimeout       = 3 * time.Second
)

// TransactionLevel is the level of mysql transaction
type TransactionLevel string

const (
	// RepeatableRead is mysql transaction level REPEATABLE READ
	RepeatableRead TransactionLevel = "REPEATABLE READ"
	// ReadCommitted is mysql transaction level READ COMMITTED
	ReadCommitted = "READ COMMITTED"
	// ReadUnCommitted is mysql transaction level READ UNCOMMITTED
	ReadUnCommitted = "READ UNCOMMITTED"
	// Serializable is mysql transaction level SERIALIZABLE
	Serializable = "SERIALIZABLE"
)

//CheckBinlog checks binlog token in this request is right or not
func CheckBinlog(db *sql.DB) error {
	var variable, value string
	err := db.QueryRow(`SHOW GLOBAL VARIABLES LIKE "log_bin";`).Scan(&variable, &value)
	if err != nil {
		return errors.Trace(err)
	}
	if variable != "log_bin" || value != "ON" {
		return errors.New("binlog not open")
	}

	err = db.QueryRow(`SHOW GLOBAL VARIABLES LIKE "binlog_format";`).Scan(&variable, &value)
	if err != nil {
		return errors.Trace(err)
	}
	if variable != "binlog_format" || value != "ROW" {
		return errors.Errorf("binlog format is not ROW, but %s", value)
	}

	err = db.QueryRow(`SHOW GLOBAL VARIABLES LIKE "binlog_row_image";`).Scan(&variable, &value)
	if err != nil {
		return errors.Trace(err)
	}
	if variable != "binlog_row_image" || value != "FULL" {
		return errors.Errorf("binlog_row_image is not FULL, but %s", value)
	}

	return nil
}

// GetMasterStatus returns mysql status.
func GetMasterStatus(db *sql.DB) (gmysql.Position, GTIDSet, error) {
	var (
		binlogPos gmysql.Position
		gs        GTIDSet
	)
	rows, err := db.Query(`SHOW MASTER STATUS`)
	if err != nil {
		return binlogPos, gs, errors.Trace(err)
	}
	defer rows.Close()

	rowColumns, err := rows.Columns()
	if err != nil {
		return binlogPos, gs, errors.Trace(err)
	}

	// Show an example.
	/*
		MySQL [test]> SHOW MASTER STATUS;
		+-----------+----------+--------------+------------------+--------------------------------------------+
		| File      | Position | Binlog_Do_DB | Binlog_Ignore_DB | Executed_Gtid_Set                          |
		+-----------+----------+--------------+------------------+--------------------------------------------+
		| ON.000001 |     4822 |              |                  | 85ab69d1-b21f-11e6-9c5e-64006a8978d2:1-46
		+-----------+----------+--------------+------------------+--------------------------------------------+
	*/
	var (
		gtid       string
		binlogName string
		pos        uint32
		nullPtr    interface{}
	)
	for rows.Next() {
		if len(rowColumns) == 5 {
			err = rows.Scan(&binlogName, &pos, &nullPtr, &nullPtr, &gtid)
		} else {
			err = rows.Scan(&binlogName, &pos, &nullPtr, &nullPtr)
		}
		if err != nil {
			return binlogPos, gs, errors.Trace(err)
		}

		binlogPos = gmysql.Position{
			Name: binlogName,
			Pos:  pos,
		}

		gs, err = ParseGTIDSet(gtid)
		if err != nil {
			return binlogPos, gs, errors.Trace(err)
		}
	}
	if rows.Err() != nil {
		return binlogPos, gs, errors.Trace(rows.Err())
	}

	return binlogPos, gs, nil
}

//CreateDB for dumper syncer and stargate
func CreateDB(dbs whpb.DBAccount) (*sql.DB, error) {
	dbName := fmt.Sprintf("%s:%s@tcp(%s:%d)/?charset=utf8&interpolateParams=true&timeout=%s", dbs.Username, dbs.Password, dbs.Host, dbs.Port, defaultDialTimeout)
	db, err := sql.Open("mysql", dbName)
	if err != nil {
		return nil, errors.Trace(err)
	}

	err = db.Ping()
	return db, errors.Trace(err)
}

//SetTimeout is the timeout for wait and netWrite
func SetTimeout(db *sql.DB, wait int, netWrite int) error {
	if wait != 0 {
		sql := fmt.Sprintf("SET SESSION wait_timeout = %d", wait)
		_, err := db.Exec(sql)
		if err != nil {
			return errors.Trace(err)
		}
	}
	if netWrite != 0 {
		sql := fmt.Sprintf("SET SESSION net_write_timeout = %d", netWrite)
		_, err := db.Exec(sql)
		if err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}

// ShowTables get tables for a schema(database).
func ShowTables(db *sql.DB, dbname string) (tables []string, err error) {
	query := fmt.Sprintf("SHOW TABLES IN `%s`", dbname)
	rows, err := db.Query(query)
	if err != nil {
		return nil, errors.Trace(err)
	}
	defer rows.Close()

	// Show an example.
	/*
		mysql> show tables in foo;
		+---------------+
		| Tables_in_foo |
		+---------------+
		| bar           |
		+---------------+
	*/

	tables = make([]string, 0, 8)
	for rows.Next() {
		var table string
		err = rows.Scan(&table)
		if err != nil {
			return nil, errors.Trace(err)
		}

		tables = append(tables, table)
	}
	if rows.Err() != nil {
		return nil, errors.Trace(rows.Err())
	}

	return tables, nil
}

// ShowSchemas get schemas from a db instance.
func ShowSchemas(db *sql.DB) (schemas []string, err error) {
	rows, err := db.Query(`SHOW SCHEMAS;`)
	if err != nil {
		return nil, errors.Trace(err)
	}
	defer rows.Close()

	// Show an example.
	/*
		mysql> show schemas;
		+--------------------+
		| Database           |
		+--------------------+
		| information_schema |
		| foo                |
		| mysql              |
		| performance_schema |
		| sys                |
		+--------------------+
	*/
	schemas = make([]string, 0, 8)
	for rows.Next() {
		var schema string

		err = rows.Scan(&schema)
		if err != nil {
			return nil, errors.Trace(err)
		}

		schemas = append(schemas, schema)
	}
	if rows.Err() != nil {
		return nil, errors.Trace(rows.Err())
	}

	return schemas, nil
}

//LockTables add read locks for table
func LockTables(db *sql.DB) error {
	_, err := db.Exec("FLUSH TABLES WITH READ LOCK")
	return errors.Trace(err)
}

//UnlockTables release locks for table
func UnlockTables(db *sql.DB) error {
	_, err := db.Exec("UNLOCK TABLES")
	return errors.Trace(err)
}

//CreateSnapshot is to start transaction with consistent snapshot
func CreateSnapshot(db *sql.DB) error {
	_, err := db.Exec("START TRANSACTION /*!40108 WITH CONSISTENT SNAPSHOT */")
	return errors.Trace(err)
}

//SetTransactionIsolation is to set session transaction isolation level
func SetTransactionIsolation(db *sql.DB, level TransactionLevel) error {
	sql := fmt.Sprintf("SET SESSION TRANSACTION ISOLATION LEVEL %s", level)
	_, err := db.Exec(sql)
	return errors.Trace(err)
}

// QuerySQL with retry.
func QuerySQL(db *sql.DB, query string) (*sql.Rows, error) {
	var (
		err  error
		rows *sql.Rows
	)

	for i := 0; i < maxRetryCount; i++ {
		if i > 0 {
			log.Warnf("[mysql]sql query retry %d: %s", i, query)
			time.Sleep(retryTimeout)
		}

		log.Debugf("[mysql][query][sql]%s", query)

		rows, err = db.Query(query)
		if err != nil {
			if !isRetryableError(err) {
				return rows, errors.Trace(err)
			}
			log.Warnf("[mysql][query][sql]%s[error]%v", query, err)
			continue
		}

		return rows, nil
	}

	if err != nil {
		log.Errorf("[mysql]query sql[%s] failed %v", query, errors.ErrorStack(err))
		return nil, errors.Trace(err)
	}

	return nil, errors.Errorf("[mysql]query sql[%s] failed", query)
}

// GetServerUUID returns mysql server_uuid.
func GetServerUUID(db *sql.DB) (string, error) {
	var masterUUID string
	rows, err := db.Query(`select @@server_uuid;`)
	if err != nil {
		return "", errors.Trace(err)
	}
	defer rows.Close()

	// Show an example.
	/*
	   MySQL [test]> SHOW MASTER STATUS;
	   +--------------------------------------+
	   | @@server_uuid                        |
	   +--------------------------------------+
	   | 53ea0ed1-9bf8-11e6-8bea-64006a897c73 |
	   +--------------------------------------+
	*/
	for rows.Next() {
		err = rows.Scan(&masterUUID)
		if err != nil {
			return "", errors.Trace(err)
		}
	}
	if rows.Err() != nil {
		return "", errors.Trace(rows.Err())
	}

	return masterUUID, nil
}

// IsGTIDModeEnabled tells MySQL gtid_mode enable or not.
func IsGTIDModeEnabled(db *sql.DB) (bool, error) {
	var (
		variable string
		gtidMode string
	)
	err := db.QueryRow("SHOW GLOBAL VARIABLES LIKE 'gtid_mode';").Scan(&variable, &gtidMode)
	if err != nil {
		return false, errors.Trace(err)
	}

	return strings.ToUpper(variable) == "GTID_MODE" && strings.ToUpper(gtidMode) == "ON", nil
}

// GetSchemaTable returns table information.
func GetSchemaTable(db *sql.DB, schemaName, tableName string) (table *schema.Table, err error) {
	maxRetry := 3
	for i := 0; i < maxRetry; i++ {
		table, err = schema.NewTableFromSqlDB(db, schemaName, tableName)
		if err != nil {
			log.Warnf("[syncer] get table error %s", errors.ErrorStack(err))
			time.Sleep(time.Millisecond * 200)
			continue
		}
		log.Debugf("[syncer] get table %+v", table)
		return table, nil
	}

	return nil, errors.Errorf("get table reached max retry %d and failed", maxRetry)
}

// EstimateCount estimate count rows of the table for specific field.
func EstimateCount(db *sql.DB, dbname string, table string, field string) (int64, error) {
	// Show an example.
	/*
		mysql> EXPLAIN SELECT `id` from `test`.`test`;
		+----+-------------+-------+------------+-------+---------------+------+---------+------+--------+----------+-------------+
		| id | select_type | table | partitions | type  | possible_keys | key  | key_len | ref  | rows   | filtered | Extra       |
		+----+-------------+-------+------------+-------+---------------+------+---------+------+--------+----------+-------------+
		|  1 | SIMPLE      | test  | NULL       | index | NULL          | id   | 5       | NULL | 108920 |   100.00 | Using index |
		+----+-------------+-------+------------+-------+---------------+------+---------+------+--------+----------+-------------+
	*/

	query := fmt.Sprintf("EXPLAIN SELECT `%s` from `%s`.`%s`", field, dbname, table)
	rows, err := db.Query(query)
	if err != nil {
		return 0, errors.Trace(err)
	}
	defer rows.Close()

	var fields map[string][]byte
	if rows.Next() {
		fields, err = ScanRow(rows)
	}
	if err != nil {
		return 0, errors.Trace(err)
	}

	cntStr, ok := fields["rows"]
	if !ok {
		return 0, errors.New("[dumper] `rows` field not found in explain select sql result")
	}
	cnt, err := strconv.ParseInt(string(cntStr), 10, 64)
	return cnt, errors.Trace(err)
}

// ShowIndex returns result of execute `show index`
func ShowIndex(db *sql.DB, dbname string, table string) ([]map[string][]byte, error) {
	/*
		show index example result:
		mysql> show index from test;
		+-------+------------+----------+--------------+-------------+-----------+-------------+----------+--------+------+------------+---------+---------------+
		| Table | Non_unique | Key_name | Seq_in_index | Column_name | Collation | Cardinality | Sub_part | Packed | Null | Index_type | Comment | Index_comment |
		+-------+------------+----------+--------------+-------------+-----------+-------------+----------+--------+------+------------+---------+---------------+
		| test  | 0          | PRIMARY  | 1            | id          | A         | 0           | NULL     | NULL   |      | BTREE      |         |               |
		| test  | 0          | aid      | 1            | aid         | A         | 0           | NULL     | NULL   | YES  | BTREE      |         |               |
		+-------+------------+----------+--------------+-------------+-----------+-------------+----------+--------+------+------------+---------+---------------+

	*/

	query := fmt.Sprintf("SHOW INDEX FROM `%s`.`%s`", dbname, table)
	rows, err := db.Query(query)
	if err != nil {
		return nil, errors.Trace(err)
	}
	defer rows.Close()

	var rowsData []map[string][]byte
	for rows.Next() {
		fields, err1 := ScanRow(rows)
		if err1 != nil {
			return nil, errors.Trace(err1)
		}
		rowsData = append(rowsData, fields)
	}

	return rowsData, nil
}

func FindSuitableIndex(db *sql.DB, dbName string, table string, numberFirst bool) (*schema.TableColumn, error) {
	rowsData, err := ShowIndex(db, dbName, table)
	if err != nil {
		return nil, errors.Trace(err)
	}

	tableInfo, err := GetSchemaTable(db, dbName, table)
	if err != nil {
		return nil, errors.Trace(err)
	}

	var indexColumn *schema.TableColumn
	// seek pk
	for _, fields := range rowsData {
		if string(fields["Key_name"]) == "PRIMARY" && string(fields["Seq_in_index"]) == "1" {
			column, valid := getColumnByName(tableInfo, string(fields["Column_name"]))
			if !valid {
				return nil, errors.New(fmt.Sprintf("can't find column %s in %s.%s", string(fields["Column_name"]), dbName, table))
			}
			if !numberFirst || column.Type == schema.TYPE_NUMBER {
				return column, nil
			}
			indexColumn = column
		}
	}

	// no pk found, seek unique index
	for _, fields := range rowsData {
		if string(fields["Non_unique"]) == "0" && string(fields["Seq_in_index"]) == "1" {
			column, valid := getColumnByName(tableInfo, string(fields["Column_name"]))
			if !valid {
				return nil, errors.New(fmt.Sprintf("can't find column %s in %s.%s", string(fields["Column_name"]), dbName, table))
			}
			if !numberFirst || column.Type == schema.TYPE_NUMBER {
				return column, nil
			}
			if indexColumn == nil {
				indexColumn = column
			}
		}
	}

	// no unique index found, seek index with max cardinality
	var c *schema.TableColumn
	var maxCardinality int
	for _, fields := range rowsData {
		if string(fields["Seq_in_index"]) == "1" {
			cardinality, err := strconv.Atoi(string(fields["Cardinality"]))
			if err != nil {
				return nil, errors.Trace(err)
			}
			if cardinality > maxCardinality {
				column, valid := getColumnByName(tableInfo, string(fields["Column_name"]))
				if !valid {
					return nil, errors.New(fmt.Sprintf("can't find column %s in %s.%s", string(fields["Column_name"]), dbName, table))
				}
				if !numberFirst || column.Type == schema.TYPE_NUMBER {
					maxCardinality = cardinality
					c = column
				}
			}
		}
	}

	if c == nil {
		// if can't find number index, return the first index
		return indexColumn, nil
	}

	return c, nil
}

// FindNumberColumn find a number type column in the table
func FindNumberColumn(db *sql.DB, dbname string, table string) (*schema.TableColumn, error) {
	tableInfo, err := GetSchemaTable(db, dbname, table)
	if err != nil {
		return nil, errors.Trace(err)
	}
	var numberColumn *schema.TableColumn
	for _, column := range tableInfo.Columns {
		if column.Type == schema.TYPE_NUMBER {
			numberColumn = &column
			break
		}
	}

	return numberColumn, nil
}

// GetFirstColumn returns the first column in the table
func GetFirstColumn(db *sql.DB, dbname string, table string) (*schema.TableColumn, error) {
	tableInfo, err := GetSchemaTable(db, dbname, table)
	if err != nil {
		return nil, errors.Trace(err)
	}
	return &tableInfo.Columns[0], nil
}

// GetRandomValues returns some random value of a table
func GetRandomValues(db *sql.DB, dbname string, table string, field string, num int64, min, max interface{}) ([]string, error) {
	randomValue := make([]string, 0, num)
	query := fmt.Sprintf("SELECT `%s` FROM (SELECT `%s` FROM `%s`.`%s` WHERE `%s` > \"%v\" AND `%s` < \"%v\" ORDER BY RAND() LIMIT %d)rand_tmp ORDER BY `%s`", field, field, dbname, table, field, min, field, max, num, field)
	log.Infof("GetRandomValues sql: %s", query)
	rows, err := db.Query(query)
	if err != nil {
		return nil, errors.Trace(err)
	}
	defer rows.Close()

	for rows.Next() {
		var value string
		err = rows.Scan(&value)
		if err != nil {
			return nil, errors.Trace(err)
		}
		randomValue = append(randomValue, value)
	}

	return randomValue, nil
}

func getColumnByName(table *schema.Table, name string) (*schema.TableColumn, bool) {
	var c *schema.TableColumn
	for _, column := range table.Columns {
		if column.Name == name {
			c = &column
			break
		}
	}

	if c != nil {
		return c, true
	}

	return nil, false
}

// ScanRowsToInterfaces scans rows to interfaces.
func ScanRowsToInterfaces(rows *sql.Rows) ([][]interface{}, error) {
	var rowsData [][]interface{}
	cols, err := rows.Columns()
	if err != nil {
		return nil, errors.Trace(err)
	}

	for rows.Next() {
		colVals := make([]interface{}, len(cols))
		colValsPtr := make([]interface{}, len(cols))
		for i := range colVals {
			colValsPtr[i] = &colVals[i]
		}

		err = rows.Scan(colValsPtr...)
		if err != nil {
			return nil, errors.Trace(err)
		}
		rowsData = append(rowsData, colVals)
	}

	return rowsData, nil
}

// ScanRow scans rows into a map.
func ScanRow(rows *sql.Rows) (map[string][]byte, error) {
	cols, err := rows.Columns()
	if err != nil {
		return nil, errors.Trace(err)
	}

	colVals := make([][]byte, len(cols))
	colValsI := make([]interface{}, len(colVals))
	for i := range colValsI {
		colValsI[i] = &colVals[i]
	}

	err = rows.Scan(colValsI...)
	if err != nil {
		return nil, errors.Trace(err)
	}

	result := make(map[string][]byte)
	for i := range colVals {
		result[cols[i]] = colVals[i]
	}

	return result, nil
}

func isRetryableError(err error) bool {
	if err == driver.ErrBadConn {
		return true
	}
	err = util.OriginError(err)
	mysqlErr, ok := err.(*mysql.MySQLError)
	if ok {
		if mysqlErr.Number == tmysql.ErrUnknown {
			return true
		}
		return false
	}

	return true
}