package mysql

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

	"github.com/pingcap/errors"
	"github.com/pingcap/log"
	"go.uber.org/zap"
)

func isSysDatabase(database string) bool {
	sysDbs := []string{"information_schema", "performance_schema", "mysql", "sys", "metrics_schema"}
	for _, sysDb := range sysDbs {
		if sysDb == strings.ToLower(database) {
			return true
		}
	}
	return false
}

type MySQLClient struct {
	db *sql.DB
}

func NewMySQLClient(db *sql.DB) *MySQLClient {
	return &MySQLClient{db: db}
}

func (m *MySQLClient) Query(query string, args ...interface{}) (*sql.Rows, error) {
	begin := time.Now()
	log.Debug("executing query", zap.String("query", query))
	rows, err := m.db.Query(query, args...)
	end := time.Now()
	duration := end.Sub(begin)
	log.Debug("execute query finished", zap.Duration("spend time", duration))
	if err != nil {
		log.Error("execute query error", zap.Error(err))
		return nil, errors.Trace(err)
	}
	log.Debug("execute query successfully")
	return rows, nil
}

func (m *MySQLClient) QueryRow(query string, args ...interface{}) *sql.Row {
	log.Debug("executing query", zap.String("query", query))
	row := m.db.QueryRow(query, args...)
	log.Debug("execute query successfully")
	return row
}

func (m *MySQLClient) GetTablesInDatabase(database string) ([]string, error) {
	query := fmt.Sprintf("show tables from `%s`", database)
	rows, err := m.Query(query)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var tables []string
	for rows.Next() {
		var table string
		if err = rows.Scan(&table); err != nil {
			return nil, err
		}
		tables = append(tables, table)
	}
	log.Debug("Get tables", zap.String("database", database), zap.Strings("tables", tables))

	return tables, nil
}

func (m *MySQLClient) GetTablesInDatabases(databases []string, tableNamePrefix string) (map[string][]string, error) {
	dbTableMap := make(map[string][]string)
	for _, database := range databases {
		tables, err := m.GetTablesInDatabase(database)
		if err != nil {
			log.Error("Failed to get tables", zap.String("database", database))
			return nil, err
		}
		if tableNamePrefix != "" {
			dbTableMap[database] = make([]string, 0)
			for _, table := range tables {
				if strings.HasPrefix(table, tableNamePrefix) {
					dbTableMap[database] = append(dbTableMap[database], table)
				}
			}
		} else {
			dbTableMap[database] = tables
		}
	}
	return dbTableMap, nil
}

type ColumnInfo struct {
	Field   string
	Type    string
	Null    string
	Key     string
	Default *string
	Extra   string
}

func (m *MySQLClient) GetColumnsInTable(database, table string) ([]ColumnInfo, error) {
	query := fmt.Sprintf("show columns from `%s`.`%s`", database, table)
	rows, err := m.Query(query)
	if err != nil {
		return nil, errors.Trace(err)
	}
	defer rows.Close()

	var columns []ColumnInfo
	for rows.Next() {
		var c ColumnInfo
		if err = rows.Scan(&c.Field, &c.Type, &c.Null, &c.Key, &c.Default, &c.Extra); err != nil {
			return nil, err
		}
		columns = append(columns, c)
	}
	log.Debug("Get columns", zap.String("database", database), zap.String("table", table), zap.Reflect("columns", columns))
	return columns, nil
}

func (m *MySQLClient) GetUserDatabases() ([]string, error) {
	rows, err := m.Query("show databases")
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var databases []string
	for rows.Next() {
		var db string
		if err = rows.Scan(&db); err != nil {
			return nil, err
		}
		if !isSysDatabase(db) {
			databases = append(databases, db)
		}
	}
	log.Debug("Get databases", zap.Strings("databases", databases))
	return databases, nil
}

func (m *MySQLClient) DropUserDatabases() error {
	dbs, err := m.GetUserDatabases()
	if err != nil {
		return err
	}
	for _, db := range dbs {
		if _, err := m.db.Exec(fmt.Sprintf("drop database if exists `%s`", db)); err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}

func (m *MySQLClient) Crc32Table(database, table string) (uint32, error) {
	columns, err := m.GetColumnsInTable(database, table)
	if err != nil {
		return 0, err
	}
	var columnNames []string
	for _, c := range columns {
		columnNames = append(columnNames, c.Field)
	}
	concat := fmt.Sprintf("concat_ws(',', `%s`)", strings.Join(columnNames, "`,`"))
	query := fmt.Sprintf("select bit_xor(crc32(%s)) as checksum from `%s`.`%s`", concat, database, table)
	var checksum uint32
	err = m.QueryRow(query).Scan(&checksum)
	log.Info("Get table checksum", zap.String("database", database), zap.String("table", table), zap.Uint32("table crc32 checksum", checksum))
	return checksum, errors.Trace(err)
}

type DbTable struct {
	Database string
	Table    string
}

func (m *MySQLClient) Crc32TableWorker(dbTableChan <-chan DbTable, checksumChan chan<- uint32, errChan chan<- error) {
	for dbTable := range dbTableChan {
		tableChecksum, err := m.Crc32Table(dbTable.Database, dbTable.Table)
		if err != nil {
			log.Error("Crc32Table error", zap.Error(err))
			errChan <- err
		}
		checksumChan <- tableChecksum
	}
}

func (m *MySQLClient) Crc32Database(database string, workerNum int) (uint32, error) {
	var dbChecksum uint32
	tables, err := m.GetTablesInDatabase(database)
	if err != nil {
		return 0, err
	}
	if len(tables) == 0 {
		return 0, nil
	}
	tableChan := make(chan DbTable, workerNum)
	checksumChan := make(chan uint32, workerNum)
	errChan := make(chan error, 1)
	for workerId := 0; workerId < workerNum; workerId++ {
		go m.Crc32TableWorker(tableChan, checksumChan, errChan)
	}
	go func() {
		for _, table := range tables {
			tableChan <- DbTable{Database: database, Table: table}
		}
		close(tableChan)
	}()
	for i := 0; i < len(tables); i++ {
		select {
		case err := <-errChan:
			return 0, err
		case tableChecksum := <-checksumChan:
			dbChecksum = dbChecksum ^ tableChecksum
		}
	}
	close(checksumChan)
	log.Info("Get db checksum", zap.String("database", database), zap.Uint32("db crc32 checksum", dbChecksum))
	return dbChecksum, nil
}

func (m *MySQLClient) Crc32Databases(databases []string, workerNum int) (uint32, error) {
	var totalChecksum uint32
	for _, db := range databases {
		dbChecksum, err := m.Crc32Database(db, workerNum)
		if err != nil {
			return 0, err
		}
		totalChecksum = totalChecksum ^ dbChecksum
	}
	log.Info("Get databases checksum", zap.Strings("databases", databases), zap.Uint32("databases crc32 checksum", totalChecksum))
	return totalChecksum, nil
}

func (m *MySQLClient) Crc32UserDatabases(workerNum int) (uint32, error) {
	databases, err := m.GetUserDatabases()
	if err != nil {
		return 0, err
	}
	checksum, err := m.Crc32Databases(databases, workerNum)
	log.Info("Get user databases checksum", zap.Uint32("user databases crc32 checksum", checksum))
	return checksum, err
}