package datagen

import (
	"context"
	"database/sql"
	"fmt"
	"sort"
	"strings"

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

const inconsistentErr = "result inconsistent"

func DiffTiFlashTiKV(db *sql.DB, ctx context.Context, q string, isSort bool) (err error) {
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer func() {
		if err != nil && !strings.Contains(err.Error(), inconsistentErr) {
			err2 := tx.Rollback()
			if err == sql.ErrTxDone && err2 == sql.ErrTxDone {
				err = nil
			}
			if err == sql.ErrTxDone && err2 != sql.ErrTxDone {
				err = err2
			}
		}
	}()
	var tso uint64
	if err := tx.QueryRow("select @@tidb_current_ts").Scan(&tso); err != nil {
		return err
	}
	if _, err = tx.Exec("set @@tidb_isolation_read_engines='tiflash';"); err != nil {
		return err
	}
	rowValuesTiFlash, err := query(tx, q)
	if err != nil {
		return err
	}
	if _, err = tx.Exec("set @@tidb_isolation_read_engines='tikv';"); err != nil {
		return err
	}
	rowValuesTiKV, err := query(tx, q)
	if err != nil {
		return err
	}
	if isSort {
		sort.Strings(rowValuesTiFlash)
		sort.Strings(rowValuesTiKV)
	}
	if !testEq(rowValuesTiKV, rowValuesTiFlash) {
		err = fmt.Errorf("%s tso %d, tikv: %v, tiflash %v", inconsistentErr, tso, rowValuesTiKV, rowValuesTiFlash)
	} else {
		log.Info("verify successfully", zap.Strings("tikv", rowValuesTiKV), zap.Strings("tiflash", rowValuesTiFlash))
	}
	err = tx.Commit()
	return
}

func testEq(a, b []string) bool {
	if len(a) != len(b) {
		return false
	}
	for i := range a {
		if a[i] != b[i] {
			return false
		}
	}
	return true
}

func query(tx *sql.Tx, q string) ([]string, error) {
	rowValues := make([]string, 0)
	rows, err := tx.Query(q)
	if err != nil {
		return rowValues, err
	}
	defer rows.Close()
	cols, err := rows.Columns()
	if err != nil {
		return rowValues, err
	}
	values := make([]interface{}, len(cols))
	results := make([]sql.NullString, len(cols))
	for i := range values {
		values[i] = &results[i]
	}
	for rows.Next() {
		if err := rows.Scan(values...); err != nil {
			return rowValues, err
		}
		allFields := ""
		for _, v := range results {
			if !v.Valid {
				allFields += "NULL"
				continue
			}
			allFields += v.String
		}
		rowValues = append(rowValues, allFields)
	}
	if err := rows.Err(); err != nil {
		return rowValues, err
	}
	return rowValues, nil
}