package dbctrl

import (
	"database/sql"
	"fmt"
	"strings"
	"time"
	"validate/util"

	"github.com/pingcap/errors"
)

const ShowTableStmt = "show tables;"
const DescribeTableStmt = "desc `%s`;"
const ShowIndexFromTableStmt = "show index from `%s`;"

const ShowTableFromInformationSchemaStmt = "SELECT * FROM information_schema.tables where TABLE_NAME = '%s'and TABLE_SCHEMA = '%s';"
const ShowAllTableFromInformationSchemaStmt = "SELECT * FROM information_schema.tables;"
const ShowAllPartitionTableFromInformationSchemaStmt = "SELECT * FROM information_schema.partitions;"

type TableColumnKeyType string

const (
	PrimaryKey TableColumnKeyType = "PRI"
	UniqueKey  TableColumnKeyType = "UNI"
	NoneKey    TableColumnKeyType = ""
)

func (keyType TableColumnKeyType) ToString() string {
	switch keyType {
	case PrimaryKey:
		return "PRI"
	case UniqueKey:
		return "UNI"
	default:
		return ""
	}
}

type ColumnInfo struct {
	FieldName       string
	Type            string
	IsNull          bool
	KeyType         TableColumnKeyType
	DefaultValue    string
	Extra           string
	IsAutoIncreased bool
}

type TableIndexType string

const (
	BtreeIndexType TableIndexType = "btree"
	NoneIndexType  TableIndexType = ""
)

func (idxType *TableIndexType) ToString() string {
	switch *idxType {
	case BtreeIndexType:
		return "btree"
	default:
		return ""
	}
}

type IndexInfo struct {
	Table        string
	IsNoneUnique bool
	KeyName      string
	SeqInIndex   int
	ColumnName   string
	Collation    string
	Cardinality  int
	SubPart      string
	Packed       string
	IsNull       bool
	IndexType    TableIndexType
	Comment      string
	IndexComment string
	IsVisible    bool
	Expression   string
	IsClustered  bool
}

type ForeignKeyInfo struct{}

type TableInformationSchema struct {
	TABLE_CATALOG              string
	TABLE_SCHEMA               string
	TABLE_NAME                 string
	TABLE_TYPE                 string
	ENGINE                     string
	VERSION                    int64
	ROW_FORMAT                 string
	TABLE_ROWS                 int64
	AVG_ROW_LENGTH             int64
	DATA_LENGTH                int64
	MAX_DATA_LENGTH            int64
	INDEX_LENGTH               int64
	DATA_FREE                  int64
	AUTO_INCREMENT             int64
	CREATE_TIME                time.Time
	UPDATE_TIME                time.Time
	CHECK_TIME                 time.Time
	TABLE_COLLATION            string
	CHECKSUM                   int64
	CREATE_OPTIONS             string
	TABLE_COMMENT              string
	TIDB_TABLE_ID              int64
	TIDB_ROW_ID_SHARDING_INFO  string
	TIDB_PK_TYPE               string
	TIDB_PLACEMENT_POLICY_NAME string

	ColumnList     []*ColumnInfo
	IndexList      []*IndexInfo
	ForeignKeyList []*ForeignKeyInfo
}

type PartitionTableInformationSchema struct {
	TABLE_CATALOG                 string
	TABLE_SCHEMA                  string
	TABLE_NAME                    string
	PARTITION_NAME                string
	SUBPARTITION_NAME             string
	PARTITION_ORDINAL_POSITION    int64
	SUBPARTITION_ORDINAL_POSITION int64
	PARTITION_METHOD              string
	SUBPARTITION_METHOD           string
	PARTITION_EXPRESSION          string
	SUBPARTITION_EXPRESSION       string
	PARTITION_DESCRIPTION         string
	TABLE_ROWS                    int64
	AVG_ROW_LENGTH                int64
	DATA_LENGTH                   int64
	MAX_DATA_LENGTH               int64
	INDEX_LENGTH                  int64
	DATA_FREE                     int64
	CREATE_TIME                   time.Time
	UPDATE_TIME                   time.Time
	CHECK_TIME                    time.Time
	CHECKSUM                      int64
	PARTITION_COMMENT             string
	NODEGROUP                     string
	TABLESPACE_NAME               string
	TIDB_PARTITION_ID             int64
	TIDB_PLACEMENT_POLICY_NAME    string
}

func GetInformationSchemaTableByStmt(tx *sql.Tx, stmt string) ([]*TableInformationSchema, error) {
	if tx == nil || len(stmt) < 1 {
		return nil, errors.New("empty query statement")
	}

	var tableList []*TableInformationSchema

	rows, err := tx.Query(stmt)
	if err != nil {
		return nil, err
	}

	for rows.Next() {
		var updateTime, checkTime sql.NullTime
		var checkSum sql.NullInt64
		var policyName sql.NullString

		var shardingInfo sql.NullString
		var autoIncrementalID sql.NullInt64
		var tbl = &TableInformationSchema{}
		if err = rows.Scan(
			&tbl.TABLE_CATALOG,
			&tbl.TABLE_SCHEMA,
			&tbl.TABLE_NAME,
			&tbl.TABLE_TYPE,
			&tbl.ENGINE,
			&tbl.VERSION,
			&tbl.ROW_FORMAT,
			&tbl.TABLE_ROWS,
			&tbl.AVG_ROW_LENGTH,
			&tbl.DATA_LENGTH,
			&tbl.MAX_DATA_LENGTH,
			&tbl.INDEX_LENGTH,
			&tbl.DATA_FREE,
			&autoIncrementalID,
			&tbl.CREATE_TIME,
			&updateTime,
			&checkTime,
			&tbl.TABLE_COLLATION,
			&checkSum,
			&tbl.CREATE_OPTIONS,
			&tbl.TABLE_COMMENT,
			&tbl.TIDB_TABLE_ID,
			&shardingInfo,
			&tbl.TIDB_PK_TYPE,
			&policyName,
		); err != nil {
			util.Logger.Fatalf(err.Error())
		}

		if checkTime.Valid {
			tbl.UPDATE_TIME = updateTime.Time
		}
		if updateTime.Valid {
			tbl.CHECK_TIME = checkTime.Time
		}
		if checkSum.Valid {
			tbl.CHECKSUM = checkSum.Int64
		}
		if policyName.Valid {
			tbl.TIDB_PLACEMENT_POLICY_NAME = policyName.String
		}
		if autoIncrementalID.Valid {
			tbl.AUTO_INCREMENT = autoIncrementalID.Int64
		}
		if shardingInfo.Valid {
			tbl.TIDB_ROW_ID_SHARDING_INFO = shardingInfo.String
		}

		tableList = append(tableList, tbl)
	}
	if err = rows.Err(); err != nil {
		return nil, err
	}

	return tableList, nil
}

func GetAllInformationSchemaTable(tx *sql.Tx) ([]*TableInformationSchema, error) {
	return GetInformationSchemaTableByStmt(tx, ShowAllTableFromInformationSchemaStmt)
}

func GetInformationSchemaPartitionTableByStmt(tx *sql.Tx, stmt string) ([]*PartitionTableInformationSchema, error) {
	if tx == nil || len(stmt) < 1 {
		return nil, errors.New("empty query statement")
	}

	var tableList []*PartitionTableInformationSchema

	rows, err := tx.Query(stmt)
	if err != nil {
		return nil, err
	}

	for rows.Next() {
		var updateTime, checkTime sql.NullTime
		var checkSum sql.NullInt64
		var policyName sql.NullString
		var tbl = &PartitionTableInformationSchema{}
		if err = rows.Scan(
			&tbl.TABLE_CATALOG,
			&tbl.TABLE_SCHEMA,
			&tbl.TABLE_NAME,
			&tbl.PARTITION_NAME,
			&tbl.SUBPARTITION_NAME,
			&tbl.PARTITION_ORDINAL_POSITION,
			&tbl.SUBPARTITION_ORDINAL_POSITION,
			&tbl.PARTITION_METHOD,
			&tbl.SUBPARTITION_METHOD,
			&tbl.PARTITION_EXPRESSION,
			&tbl.SUBPARTITION_EXPRESSION,
			&tbl.PARTITION_DESCRIPTION,
			&tbl.TABLE_ROWS,
			&tbl.AVG_ROW_LENGTH,
			&tbl.DATA_LENGTH,
			&tbl.MAX_DATA_LENGTH,
			&tbl.INDEX_LENGTH,
			&tbl.DATA_FREE,
			&tbl.CREATE_TIME,
			&updateTime,
			&checkTime,
			&checkSum,
			&tbl.PARTITION_COMMENT,
			&tbl.NODEGROUP,
			&tbl.TABLESPACE_NAME,
			&tbl.TIDB_PARTITION_ID,
			&checkSum,
		); err != nil {
			util.Logger.Fatalf(err.Error())
		}

		if updateTime.Valid {
			tbl.UPDATE_TIME = updateTime.Time
		}
		if checkTime.Valid {
			tbl.CHECK_TIME = checkTime.Time
		}
		if checkSum.Valid {
			tbl.CHECKSUM = checkSum.Int64
		}
		if policyName.Valid {
			tbl.TIDB_PLACEMENT_POLICY_NAME = policyName.String
		}

		tableList = append(tableList, tbl)
	}
	if err = rows.Err(); err != nil {
		return nil, err
	}

	return tableList, nil
}

func GetAllInformationSchemaPartitionTable(tx *sql.Tx) ([]*PartitionTableInformationSchema, error) {
	return GetInformationSchemaPartitionTableByStmt(tx, ShowAllPartitionTableFromInformationSchemaStmt)
}

func getTableIndexList(tableName string, tx *sql.Tx) ([]*IndexInfo, error) {
	var stmt = fmt.Sprintf(ShowIndexFromTableStmt, tableName)
	rows, err := tx.Query(stmt)
	if err != nil {
		return nil, err
	}

	var indexList []*IndexInfo
	for rows.Next() {

	}

	return indexList, nil
}

func getTableForeignKeyList(tableName string, tx *sql.Tx) ([]*ForeignKeyInfo, error) {
	return nil, nil
}

func getTableColumnList(tableName string, tx *sql.Tx) ([]*ColumnInfo, error) {
	var stmt = fmt.Sprintf(DescribeTableStmt, tableName)
	rows, err := tx.Query(stmt)
	if err != nil {
		return nil, err
	}

	var columnList []*ColumnInfo
	for rows.Next() {
		var fieldName string
		var columnType string
		var isNullString string
		var keyType string
		var defaultValue sql.NullString
		var extra string
		if err = rows.Scan(&fieldName, &columnType, &isNullString,
			&keyType, &defaultValue, &extra); err != nil {
			util.Logger.Fatalf(err.Error())
		}

		columnInfo := &ColumnInfo{}

		columnInfo.FieldName = fieldName
		columnInfo.Type = columnType
		columnInfo.IsNull = isNullString == "YES"
		switch keyType {
		case "PRI":
			columnInfo.KeyType = PrimaryKey
		case "UNI":
			columnInfo.KeyType = UniqueKey
		default:
			columnInfo.KeyType = NoneKey
		}

		if defaultValue.Valid {
			columnInfo.DefaultValue = defaultValue.String
		} else {
			columnInfo.DefaultValue = ""
		}

		columnInfo.Extra = extra
		if strings.Contains(columnInfo.Extra, "auto_increment") {
			columnInfo.IsAutoIncreased = true
		}

		columnList = append(columnList, columnInfo)
	}

	return columnList, nil
}

func GetTableDetail(dbName string, tableName string, tx *sql.Tx) (*TableInformationSchema, error) {
	stmt := fmt.Sprintf(ShowTableFromInformationSchemaStmt, tableName, dbName)
	tbls, err := GetInformationSchemaTableByStmt(tx, stmt)
	if err != nil {
		return nil, err
	}
	if len(tbls) > 1 {
		return nil, errors.New("more than one table found by '" + dbName + "'.'" + tableName + "'")
	}

	tblInfo := tbls[0]

	columnList, err := getTableColumnList(tableName, tx)
	if err != nil {
		return nil, err
	}
	tblInfo.ColumnList = columnList
	indexList, err := getTableIndexList(tableName, tx)
	if err != nil {
		return nil, err
	}
	tblInfo.IndexList = indexList

	foreignKeyList, err := getTableForeignKeyList(tableName, tx)
	if err != nil {
		return nil, err
	}
	tblInfo.ForeignKeyList = foreignKeyList

	return tblInfo, nil
}

func GetTableNameList(tx *sql.Tx) ([]string, error) {
	var tableNameList []string

	rows, err := tx.Query(ShowTableStmt)
	if err != nil {
		return nil, err
	}

	for rows.Next() {
		var tblName string
		if err = rows.Scan(&tblName); err != nil {
			util.Logger.Fatalf(err.Error())
		}

		tableNameList = append(tableNameList, tblName)
	}
	if err = rows.Err(); err != nil {
		return nil, err
	}

	return tableNameList, nil
}

func IsSystemTableInMySQL(dbName string, tableName string) bool {
	if dbName != MySQLDatabase {
		return false
	}
	for _, name := range SystemTableInMySQLList {
		if tableName == name {
			return true
		}
	}

	return false
}

func PrintTableBriefHeader() string {
	var tblInfo string
	tblInfo += fmt.Sprintf("+------------------------+----------------------+-------------+-------------------+---------------------------+\n")
	tblInfo += fmt.Sprintf("| Table Name             | Estimated Storage    |   UK Or PK  |   Column Count    |  Auto-incremental Column  |\n")
	tblInfo += fmt.Sprintf("+------------------------+----------------------+-------------+-------------------+---------------------------+\n")

	return tblInfo
}

func PrintTableBriefTail() string {
	return fmt.Sprintf("+------------------------+----------------------+-------------+-------------------+---------------------------+")
}

func (tbl *TableInformationSchema) HasPKOrUK() bool {
	for _, column := range tbl.ColumnList {
		if column.KeyType == PrimaryKey || column.KeyType == UniqueKey {
			return true
		}
	}

	return false
}

func (tbl *TableInformationSchema) PKOrUK() string {
	var hasPK = false
	var hasUK = false

	for _, column := range tbl.ColumnList {
		if column.KeyType == PrimaryKey {
			hasPK = true
		}
		if column.KeyType == UniqueKey {
			hasUK = true
		}
	}

	var tag = ""
	if hasPK {
		tag += PrimaryKey.ToString()
	}
	if hasUK {
		if hasPK {
			tag += ", "
		}
		tag += UniqueKey.ToString()
	}

	return tag
}

func (tbl *TableInformationSchema) AutoIncrementalColumnList() (clmList []*ColumnInfo) {
	for _, column := range tbl.ColumnList {
		if column.IsAutoIncreased {
			clmList = append(clmList, column)
		}
	}

	return clmList
}

func (tbl *TableInformationSchema) PrintTableBrief() string {
	var tblInfo string

	var clmListString = ""
	for _, column := range tbl.AutoIncrementalColumnList() {
		if len(clmListString) > 0 {
			clmListString += ", "
		}
		clmListString += column.FieldName
	}

	tblInfo += fmt.Sprintf("| %-22s | %-20d | %-11s | %-17d | %-25s |\n", tbl.TABLE_NAME, 0, tbl.PKOrUK(), len(tbl.ColumnList), clmListString)

	return tblInfo
}

func (tbl *TableInformationSchema) IsSystemTableInMySQL() bool {
	if tbl.TABLE_SCHEMA != MySQLDatabase {
		return false
	}
	for _, name := range SystemTableInMySQLList {
		if 0 == strings.Compare(strings.TrimSpace(tbl.TABLE_NAME), strings.TrimSpace(name)) {
			return true
		}
	}

	return false
}