package copr_paging

import (
	"context"
	"database/sql"
	"fmt"
	"log"
	"strings"
	"sync"
	"time"

	"github.com/go-sql-driver/mysql"
	"github.com/pingcap/errors"
	"github.com/pingcap/test-infra/sdk/resource"
	_ "github.com/pingcap/test-infra/sdk/resource/impl/k8s"

	"github.com/pingcap/endless/pkg/chaos"
	"github.com/pingcap/endless/pkg/client/fail"
	"github.com/pingcap/endless/pkg/client/pd"
	pdclient "github.com/pingcap/endless/pkg/client/pd"
	. "github.com/pingcap/endless/pkg/matcher"
	"github.com/pingcap/endless/testcase/oltp/oltppkg"
)

func getDB(tc resource.TiDBCluster) *sql.DB {
	return Try(tc.GetDB("test")).(*sql.DB)
}

var ignoreMySQLErrors = map[uint16]string{
	1105: "no available connections",
	8027: "Information schema is out of date",
	9002: "TiKV server timeout",
	9005: "Region is unavailable",
}

var ignoreErrorPrefix = map[string]struct{}{
	"invalid connection": {},
	"socks connect":      {},
	"context canceled":   {},
}

// filterError filter common errors like server timeout for reducing false positive.
func filterError(err error) error {
	if mysqlErr, ok := err.(*mysql.MySQLError); ok {
		if _, ok := ignoreMySQLErrors[mysqlErr.Number]; ok {
			return nil
		}
	}
	errStr := err.Error()
	for prefix := range ignoreErrorPrefix {
		if strings.HasPrefix(errStr, prefix) {
			return nil
		}
	}
	return err
}

// execAndCompare accept an in-transaction db connection and execute query in non-paging and paging mode.
// After execution, this function will compare their results.
// It returns error in any of the following conditions:
// 1. the execution meet error
// 2. the result is not same.
func execAndCompare(ctx context.Context, txn *sql.Tx, query string, args ...interface{}) (err error) {
	var (
		rows1   *sql.Rows
		rows2   *sql.Rows
		rows    *sql.Rows
		startTS string
	)
	rows, err = txn.QueryContext(ctx, "select @@tidb_current_ts")
	if err != nil {
		return errors.Trace(err)
	}
	data, err := ReadAll(rows)
	if err != nil {
		return errors.Trace(err)
	}
	startTS = data[0][0]
	defer func() {
		if err != nil {
			err = errors.Wrapf(err, "start_ts: %s", startTS)
		}
	}()
	// execution with paging
	if _, err = txn.ExecContext(ctx, "set session tidb_enable_paging=1"); err != nil {
		err = errors.Trace(filterError(err))
		return
	}

	rows1, err = txn.QueryContext(ctx, query, args...)
	if err != nil {
		err = errors.Trace(filterError(err))
		return
	}
	data1, err := ReadAll(rows1)
	if err != nil {
		return errors.Trace(err)
	}
	// execution without paging
	if _, err = txn.ExecContext(ctx, "set session tidb_enable_paging=0"); err != nil {
		err = errors.Trace(filterError(err))
		return
	}
	rows2, err = txn.QueryContext(ctx, query, args...)
	if err != nil {
		err = errors.Trace(filterError(err))
		return
	}
	data2, err := ReadAll(rows2)
	if err != nil {
		return errors.Trace(err)
	}

	err = errors.Trace(compareData(data1, data2))
	return
}

func compareData(data1, data2 [][]string) error {
	errWithResult := func(desc string) error {
		return errors.Errorf("%s\ndata1(paging):\n%v\ndata2(non-paging):\n%v", desc, data1, data2)
	}
	if len(data1) != len(data2) {
		return errWithResult(fmt.Sprintf("length of data1(%d) and data2(%d) not same.", len(data1), len(data2)))
	}
	if len(data1) == 0 {
		return nil
	}
	if len(data1[0]) != len(data2[0]) {
		return errWithResult("length of data1[0] and data2[0] not same.")
	}

	for i := 0; i < len(data1); i++ {
		row1, row2 := data1[i], data2[i]
		for j := 0; j < len(row1); j++ {
			if row1[j] != row2[j] {
				return errWithResult(fmt.Sprintf("[%d, %d] not same, data1[%d][%d]=%s, data2[%d][%d]=%s",
					i, j, i, j, row1[j], i, j, row2[j]))
			}
		}
	}
	return nil
}

func ReadAll(rows *sql.Rows) ([][]string, error) {
	defer func() {
		Try(rows.Close())
	}()

	data := make([][]string, 0)
	for rows.Next() {
		columns, err := rows.Columns()
		if err != nil {
			return nil, err
		}
		row := make([]string, len(columns))
		scan := make([]interface{}, len(columns))
		for i := range row {
			scan[i] = &row[i]
		}
		if err := rows.Scan(scan...); err != nil {
			return nil, err
		}
		data = append(data, row)
	}
	return data, nil
}

// MustExec must execute sql or fatal
func MustExec(DB *sql.DB, query string, args ...interface{}) sql.Result {
	return Try(DB.Exec(query, args...)).(sql.Result)
}

func prepareChaos(tc resource.TiDBCluster, tp resource.TiDBCompType, name string, spec resource.ExperimentSpec) *chaos.Chaos {
	instanceIdList := oltppkg.GetInstanceIdList(tc, tp)
	instanceId, _ := oltppkg.GetRandInstance(tc, tp)
	c := Try(oltppkg.InitChaos(tc, tp, name, spec, &instanceIdList[instanceId])).(chaos.Chaos)
	return &c
}

// doRegionError injects region error by leader transfer.
func doRegionError(tc resource.TiDBCluster, errCh chan<- error, finishCh <-chan struct{}, wg *sync.WaitGroup) {
	pdcli := Try(pdclient.NewPDClient(tc)).(*pd.Client)
	state := false
	defer func() {
		wg.Done()
		if state {
			if err := pdcli.DeleteScheduler("shuffle-leader-scheduler"); err != nil {
				log.Println("clean scheduler error", err)
			}
		}
	}()

	timer := time.NewTimer(0)
	for {
		select {
		case <-finishCh:
			return
		case <-timer.C:
		}
		if !state {
			faultDuration := oltppkg.GetRandDuration(time.Minute, 5)
			log.Println("turn on network chaos for", faultDuration)
			if _, err := pdcli.CreateScheduler("shuffle-leader-scheduler"); err != nil {
				errCh <- err
				return
			}
			timer = time.NewTimer(faultDuration)
		} else {
			if err := pdcli.DeleteScheduler("shuffle-leader-scheduler"); err != nil {
				errCh <- err
				return
			}
			timer = time.NewTimer(5 * time.Minute)
		}
		state = !state
	}
}

// doNetworkError injects network errors for TiKV nodes.
func doNetworkError(tc resource.TiDBCluster, errCh chan<- error, finishCh <-chan struct{}, wg *sync.WaitGroup) {
	var c *chaos.Chaos
	defer func() {
		if c != nil {
			if err := c.Clean(); err != nil {
				log.Println("clean network error chaos error", err)
			}
		}
		wg.Done()
	}()

	timer := time.NewTimer(0)
	for {
		select {
		case <-finishCh:
			return
		case <-timer.C:
		}
		if c == nil {
			spec := &resource.NetworkLossSpec{
				Loss:        "25",
				Correlation: "25",
			}
			c = prepareChaos(tc, resource.TiKV, "network-loss", spec)
			faultDuration := oltppkg.GetRandDuration(time.Minute, 5)
			log.Println("turn on network chaos for", faultDuration)
			if err := c.Do(); err != nil {
				errCh <- err
				return
			}
			timer = time.NewTimer(faultDuration)
		} else {
			log.Println("turn off network chaos")
			if err := c.Clean(); err != nil {
				errCh <- err
				return
			}
			c = nil
			timer = time.NewTimer(5 * time.Minute)
		}
	}
}

// doLockedError injects locked errors by network delay.
func doLockedError(tc resource.TiDBCluster, errCh chan<- error, finishCh <-chan struct{}, wg *sync.WaitGroup) {
	failClient := fail.NewFailClient(tc, resource.TiDB)
	failpointPath := "tikvclient/beforeCommit"
	failpointAction := "sleep(1000)"
	status := false
	defer func() {
		if status {
			failClient.Disable(failpointPath)
		}
		wg.Done()
	}()

	timer := time.NewTimer(0)
	for {
		select {
		case <-finishCh:
			return
		case <-timer.C:
		}
		if !status {
			faultDuration := oltppkg.GetRandDuration(time.Minute, 5)
			log.Println("turn on locked error for", faultDuration)
			if err := failClient.Enable(failpointPath, failpointAction); err != nil {
				errCh <- err
				return
			}
			timer = time.NewTimer(faultDuration)
		} else {
			log.Println("turn off locked error")
			if err := failClient.Disable(failpointPath); err != nil {
				errCh <- err
				return
			}
			timer = time.NewTimer(5 * time.Minute)
		}
		status = !status
	}
}

// doLockedError injects locked errors by network delay.
func doPodFailure(tc resource.TiDBCluster, errCh chan<- error, finishCh <-chan struct{}, wg *sync.WaitGroup) {
	var c *chaos.Chaos
	defer func() {
		if c != nil {
			if err := c.Clean(); err != nil {
				log.Println("clean locked error chaos error", err)
			}
		}
		wg.Done()
	}()

	timer := time.NewTimer(0)
	for {
		select {
		case <-finishCh:
			return
		case <-timer.C:
		}
		if c == nil {
			spec := &resource.FailureExperimentSpec{}
			c = prepareChaos(tc, resource.TiKV, "tikv failure", spec)
			faultDuration := time.Minute
			log.Println("turn on locked error for", faultDuration)
			if err := c.Do(); err != nil {
				errCh <- err
				return
			}
			timer = time.NewTimer(faultDuration)
		} else {
			log.Println("turn off locked error")
			if err := c.Clean(); err != nil {
				errCh <- err
				return
			}
			c = nil
			timer = time.NewTimer(5 * time.Minute)
		}
	}
}