package pkg

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

	"github.com/onsi/gomega"
	"github.com/pingcap/log"
	"go.uber.org/zap"

	"github.com/pingcap/endless/pkg/matcher"
)

func isQuery(q string) bool {
	preHandleQ := strings.ToLower(strings.TrimSpace(q))
	if strings.HasPrefix(preHandleQ, "select") || strings.HasPrefix(preHandleQ, "with") {
		return true
	}
	return false
}

func isAlterTiFlashReplica(q string) (bool, error) {
	pattern := `(^alter)\s*(database|table).+set\s*tiflash\s*replica\s*[0-9]+`
	preHandleQ := strings.ToLower(strings.TrimSpace(q))
	return regexp.MatchString(pattern, preHandleQ)
}

func isAllTablesTiFlashReplicaReady(db *sql.DB) (bool, error) {
	rows, err := db.Query("select * from  information_schema.tiflash_replica where AVAILABLE!=1")
	if err != nil {
		return false, err
	}
	return !rows.Next(), nil
}

// CheckQueryResultTiKVTiFlash will check all queries result consistency of tikv and tiflash
// 查询遇到错误后会根据 errorIgnore 来判断是否忽略该错误继续执行
// 该函数会返回两种错误 1. InconsistentErr; 2. errorIgnore 未忽略的错误
func CheckQueryResultTiKVTiFlash(sqls []string, engineTikVDB *sql.DB, engineTiFlashDB *sql.DB, errorIgnore func(error) error) error {
	for i, q := range sqls {
		log.Info("execute sequence", zap.Int("row number", i+1), zap.String("sql", q))
		ifAlterTiFlashRep, err := isAlterTiFlashReplica(q)
		if err != nil {
			return err
		}
		if isQuery(q) {
			tikvRes, _, tikvErr := Query(context.Background(), engineTikVDB, q, false)
			if tikvErr != nil {
				log.Error("raise error when execute sql that is query using tikv", zap.Int("row number", i+1), zap.String("sql", q), zap.Error(tikvErr))
				if errorIgnore == nil || (errorIgnore != nil && errorIgnore(tikvErr) != nil) {
					return tikvErr
				}
			}
			tiflashRes, _, tiflashErr := Query(context.Background(), engineTiFlashDB, q, false)
			if tiflashErr != nil {
				log.Error("raise error when execute sql that is query using tiflash", zap.Int("row number", i+1), zap.String("sql", q), zap.Error(tikvErr))
				if errorIgnore == nil || (errorIgnore != nil && errorIgnore(tiflashErr) != nil) {
					return tiflashErr
				}
			}
			if tikvErr != tiflashErr {
				log.Error("the error of query using tiflash and tikv is not consistency", zap.Int("row number", i+1), zap.String("sql", q),
					zap.Any("tikv error", tikvErr), zap.Any("tiflash error", tiflashErr))
				return fmt.Errorf("%w, tikv :%v, tiflash %v", InconsistentErr, tikvErr, tiflashErr)
			}
			if !testEq(tikvRes, tiflashRes) {
				log.Error("the result of query using tiflash and tikv is not consistency", zap.Int("row number", i+1), zap.String("sql", q),
					zap.Strings("tikv result", tikvRes), zap.Strings("tiflash result", tiflashRes))
				return fmt.Errorf("%w, tikv: %v, tiflash %v", InconsistentErr, tikvRes, tiflashRes)
			}
			continue
		}
		_, err = engineTikVDB.Exec(q)
		if err != nil {
			log.Error("raise error when execute sql that is not query using tikv", zap.Int("row number", i+1), zap.String("sql", q), zap.Error(err))
			if errorIgnore != nil && errorIgnore(err) != nil {
				return err
			}
		}
		if ifAlterTiFlashRep {
			log.Info("wait tiflash replica ready, timeout 10m")
			gomega.Eventually(func() bool {
				return matcher.Try(isAllTablesTiFlashReplicaReady(engineTikVDB)).(bool)
			}, 10*time.Minute, 2*time.Second).Should(gomega.BeTrue())
		}
	}
	return nil
}

func DiffTiFlashTiKV(ctx context.Context, db *sql.DB, q string, isSort bool, isLogResult bool) (err error) {
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return err
	}
	defer tx.Commit()
	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(ctx, tx, q, false)
	if err != nil {
		return err
	}
	if _, err = tx.Exec("set @@tidb_isolation_read_engines='tikv';"); err != nil {
		return err
	}
	rowValuesTiKV, _, err := Query(ctx, tx, q, false)
	if err != nil {
		return err
	}
	if isSort {
		sort.Strings(rowValuesTiFlash)
		sort.Strings(rowValuesTiKV)
	}
	if !testEq(rowValuesTiKV, rowValuesTiFlash) {
		return fmt.Errorf("%w tso %d, tikv: %v, tiflash %v", InconsistentErr, tso, rowValuesTiKV, rowValuesTiFlash)
	} else {
		if isLogResult {
			log.Info("verify successfully", zap.Strings("tikv", rowValuesTiKV), zap.Strings("tiflash", rowValuesTiFlash))
		} else {
			log.Info("verify successfully", zap.Int("result length", len(rowValuesTiKV)))
		}
	}
	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
}

type query interface {
	QueryContext(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error)
}

func Query(ctx context.Context, tx query, q string, skipScan bool) ([]string, int, error) {
	rowValues := make([]string, 0)
	count := 0
	rows, err := tx.QueryContext(ctx, q)
	if err != nil {
		return rowValues, count, err
	}
	defer rows.Close()
	cols, err := rows.Columns()
	if err != nil {
		return rowValues, count, 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, count, err
		}
		if skipScan {
			count++
			continue
		}
		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, count, err
	}
	return rowValues, count, nil
}