package consistency

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

	_ "github.com/go-sql-driver/mysql"
	"github.com/pingcap/log"
	"github.com/pingcap/test-infra/caselib/pkg/utils"
	"go.uber.org/zap"
)

var logger *zap.Logger

func init() {
	_, logger = utils.GetCtxAndLogger()
}

type CRC32 struct {
	db *sql.DB
}

// NewCRC32 return a CRC32 instance, which can be used to get crc32 checksum
func NewCRC32(db *sql.DB) *CRC32 {
	return &CRC32{
		db: db,
	}
}

func (c *CRC32) getAllTables() (tables []string, err error) {
	query := fmt.Sprintf("SHOW TABLES")
	rows, err := c.db.Query(query)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	for rows.Next() {
		var t string
		if err := rows.Scan(&t); err != nil {
			return tables, err
		}
		tables = append(tables, t)
	}

	return tables, nil
}

func (c *CRC32) getColumns(tableName string) (result []columnInfo, err error) {
	rows, err := c.db.Query(fmt.Sprintf("SHOW COLUMNS FROM %s", tableName))
	if err != nil {
		return nil, err
	}
	defer func() {
		if err := rows.Close(); err != nil {
			logger.Error("close rows failed", zap.Error(err))
		}
	}()

	for rows.Next() {
		var t columnInfo
		if err := rows.Scan(&t.Field, &t.Type, &t.Null, &t.Key, &t.Default, &t.Extra); err != nil {
			return result, err
		}
		result = append(result, t)
	}

	return result, nil
}

// GetCheckSum get crc32 checksum for each table in the given database
func (c *CRC32) GetCheckSum() (result map[string]string, err error) {
	result = make(map[string]string)
	tables, err := c.getAllTables()
	if err != nil {
		return nil, err
	}

	for _, table := range tables {
		columns, err := c.getColumns(table)
		if err != nil {
			return result, err
		}
		checkSum, err := c.doChecksum(columns, table)
		if err != nil {
			return result, err
		}
		result[table] = checkSum
	}

	return result, nil
}

func (c *CRC32) doChecksum(columns []columnInfo, tableName string) (checkSum string, err error) {
	var columnNames, isNull []string
	for _, item := range columns {
		columnNames = append(columnNames, item.Field)

		t := fmt.Sprintf("ISNULL(%s)", item.Field)
		isNull = append(isNull, t)
	}

	a := strings.Join(columnNames, ",")
	b := strings.Join(isNull, ",")

	concat := fmt.Sprintf("CONCAT_WS(',', %s, %s)", a, b)
	query := fmt.Sprintf("SELECT BIT_XOR(CAST(CRC32(%s) AS UNSIGNED)) AS checksum FROM %s", concat, tableName)

	logger.Info("do checkSum", zap.String("table", tableName), zap.String("query", query))

	err = c.db.QueryRow(query).Scan(&checkSum)
	return checkSum, err
}

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

func CompareCRC32CheckSum(source, sink *sql.DB) error {
	sourceCRC32, err := NewCRC32(source).GetCheckSum()
	if err != nil {
		logger.Panic("get source crc32 failed", zap.Error(err))
		return err
	}

	sinkCRC32, err := NewCRC32(sink).GetCheckSum()
	if err != nil {
		logger.Error("get sink crc32 failed", zap.Error(err))
		return err
	}

	if len(sourceCRC32) != len(sinkCRC32) {
		logger.Warn("source and sink have different crc32 size",
			zap.Reflect("source", sourceCRC32), zap.Reflect("sink", sinkCRC32))
	}

	failed := false
	for k, v := range sourceCRC32 {
		target, ok := sinkCRC32[k]
		if !ok {
			logger.Warn("cannot find crc32 at sink, it may eligible to replicate",
				zap.String("tableName", k), zap.String("source crc32", v))
			continue
		}
		if v != target {
			failed = true
			logger.Error("crc32 mismatch", zap.String("source", v), zap.String("sink", target))
		}
	}

	if failed {
		return errors.New("check crc32 for consistency failed")
	}

	log.Info("check crc32 for consistency success")
	return nil
}