package main

import (
	"context"
	"database/sql"
	"fmt"
	"github.com/pingcap/parser/mysql"
	"math/rand"
	"strconv"
	"strings"
	"sync"
	"time"

	mysqlDriver "github.com/go-sql-driver/mysql"
	"github.com/pingcap/errors"
	"github.com/pingcap/schrodinger/pkg/test_client"
)

type GCUnsafeDestroyRangeCase struct {
	db           *sql.DB
	dbName       string
	maxTables    int
	maxTableSize int
}

func NewGCUnsafeDestroyRangeCase(db *sql.DB, dbName string, maxTableSize int) *GCUnsafeDestroyRangeCase {
	return &GCUnsafeDestroyRangeCase{
		db,
		dbName,
		10,
		maxTableSize,
	}
}

func (c *GCUnsafeDestroyRangeCase) Run(ctx context.Context) error {
	tableIndex := int64(0)
	finishChan := make(chan error, 100)

	runningThreads := 0

	for i := 0; i < c.maxTables; i++ {
		tableName := "t" + strconv.FormatInt(tableIndex, 10)
		go func() {
			finishChan <- c.runOnce(ctx, tableName)
		}()

		runningThreads++
		tableIndex++
	}

	// When a thread stops, start a new one
LOOP:
	for {
		select {
		case <-ctx.Done():
			break LOOP
		case err := <-finishChan:
			if err != nil {
				client.Fatalf("[gc unsafe destroy range] Test failed: %v", err)
			}
			tableName := "t" + strconv.FormatInt(tableIndex, 10)
			go func() {
				finishChan <- c.runOnce(ctx, tableName)
			}()
			tableIndex++
		}
	}

	// Wait for all sub goroutines to finish
	for runningThreads > 0 {
		<-finishChan
		runningThreads--
	}
	return nil
}

// Creates a table with random number of rows, and then runs random queries and randomly adds and drops index
// and does admin check table.
// It will stop when `ctx` is canceled, or keep running for a random time.
// After stopping it will send a signal to finishChan.
func (c *GCUnsafeDestroyRangeCase) runOnce(ctx context.Context, tableName string) error {
	client.Infof("[gc unsafe destroy range] Creating table %v", tableName)

	// Create the table for testing
	tableSize := rand.Int() % c.maxTableSize
	err := c.generateTable(tableName, tableSize)
	if err != nil {
		return errors.Trace(err)
	}

	// Another goroutine will start to randomly run simple queries on the table. It will be stopped as soon as this
	// function finishes.
	queryThreadCtx, cancelQueryThread := context.WithCancel(ctx)
	wg := &sync.WaitGroup{}
	wg.Add(1)
	go c.query(queryThreadCtx, wg, tableName, tableSize)
	defer func() {
		// Cancel `c.query` and wait for if to finish
		client.Infof("[gc unsafe destroy range] Stopping query thread of table %v", tableName)
		cancelQueryThread()
		wg.Wait()

		// Drop the table
		client.Infof("[gc unsafe destroy range] Table %v is being dropped", tableName)
		_, err := c.db.Exec(fmt.Sprintf("drop table %v;", tableName))
		if err != nil {
			client.Errorf("[gc unsafe destroy range] Failed to drop table %v, err: %v", tableName, err)
		}
	}()

	// It will run on the table for a random time or until <-ctx.Done()
	tableLifeTime := time.Duration(rand.Int()%3600+120) * time.Second
	startTime := time.Now()

	// Flags that indicate whether each column of a, b and c has an index or not.
	aHasIndex := false
	bHasIndex := false
	cHasIndex := false

	for {
		if time.Since(startTime) >= tableLifeTime {
			client.Infof("[gc unsafe destroy range] Stop running on table %v due to timeout", tableName)
			return nil
		}

		select {
		case <-ctx.Done():
			client.Infof("[gc unsafe destroy range] Cancel running on table %v", tableName)
			return nil
		default:
		}

		// Randomly select a query to run
		queryType := rand.Int() % 4
		switch queryType {
		case 0:
			// Create or drop index on column a
			if aHasIndex {
				_, err = c.db.Exec(fmt.Sprintf("alter table %s drop index ia;", tableName))
				if err != nil {
					client.Errorf("[gc unsafe destroy range] Failed to drop index ia from table %v, err: %v", tableName, err)
					continue
				}
				aHasIndex = false
			} else {
				_, err = c.db.Exec(fmt.Sprintf("alter table %s add index ia(a);", tableName))
				if err != nil {
					client.Errorf("[gc unsafe destroy range] Failed to add index ia to table %v, err: %v", tableName, err)
					continue
				}
				aHasIndex = true
			}

		case 1:
			// Create or drop index on column b
			if bHasIndex {
				_, err = c.db.Exec(fmt.Sprintf("alter table %s drop index ib;", tableName))
				if err != nil {
					client.Errorf("[gc unsafe destroy range] Failed to drop index ib from table %v, err: %v", tableName, err)
					continue
				}
				bHasIndex = false
			} else {
				_, err = c.db.Exec(fmt.Sprintf("alter table %s add index ib(b);", tableName))
				if err != nil {
					client.Errorf("[gc unsafe destroy range] Failed to add index ib to table %v, err: %v", tableName, err)
					continue
				}
				bHasIndex = true
			}

		case 2:
			// Create or drop index on column c
			if cHasIndex {
				_, err = c.db.Exec(fmt.Sprintf("alter table %s drop index ic;", tableName))
				if err != nil {
					client.Errorf("[gc unsafe destroy range] Failed to drop index ic from table %v, err: %v", tableName, err)
					continue
				}
				cHasIndex = false
			} else {
				_, err = c.db.Exec(fmt.Sprintf("alter table %s add index ic(c);", tableName))
				if err != nil {
					client.Errorf("[gc unsafe destroy range] Failed to add index ic to table %v, err: %v", tableName, err)
					continue
				}
				cHasIndex = true
			}

		case 3:
			_, err = c.db.Exec(fmt.Sprintf("admin check table %s;", tableName))
			if err != nil {
				client.Errorf("[gc unsafe destroy range] Failed to check table %v, err: %v", tableName, err)
				return errors.Trace(err)
			}
		}
	}
}

// Keeps running random queries on the table until `ctx` is canceled. Sends a signal to `stopChan` when it stops.
func (c *GCUnsafeDestroyRangeCase) query(ctx context.Context, wg *sync.WaitGroup, tableName string, tableSize int) {
	defer func() {
		wg.Done()
	}()

	lastLogTime := time.Now()
	totalDoneQueries := 0

	for {
		select {
		case <-ctx.Done():
			client.Infof("[gc unsafe destroy range] Query thread on table %v exited, %v queries executed", tableName, totalDoneQueries)
			return
		default:
		}

		if time.Since(lastLogTime) > time.Second*time.Duration(300) {
			client.Infof("[gc unsafe destroy range] %v: executed queries: %v", tableName, totalDoneQueries)
			lastLogTime = time.Now()
		}

		id1 := rand.Int() % tableSize
		id2 := rand.Int() % tableSize

		diffA := rand.Int()%2333333 - (2333333 / 2)
		diffC := rand.Int()%2333333 - (2333333 / 2)

		txn, err := c.db.Begin()
		if err != nil {
			client.Errorf("[gc unsafe destroy range] Failed begin transaction on table %v, err: %v", tableName, err)
			time.Sleep(time.Second)
			continue
		}

		_, err = txn.Exec(fmt.Sprintf("update %s set a = a + ?, b = ?, c = c + ? where id = ?;", tableName), diffA, randStr(rand.Int()%128), diffC, id1)
		if err != nil {
			client.Errorf("[gc unsafe destroy range] Failed to do query on table %v, err: %v", tableName, err)
			_ = txn.Rollback()
			continue
		}
		_, err = txn.Exec(fmt.Sprintf("update %s set a = a + ?, b = ?, c = c + ? where id = ?;", tableName), -diffA, randStr(rand.Int()%128), -diffC, id2)
		if err != nil {
			client.Errorf("[gc unsafe destroy range] Failed to do query on table %v, err: %v", tableName, err)
			_ = txn.Rollback()
			continue
		}

		err = txn.Commit()

		if err != nil {
			ignoreErr := false
			if mysqlErr, ok := err.(*mysqlDriver.MySQLError); ok {
				if mysqlErr.Number == mysql.ErrWriteConflict ||
					mysqlErr.Number == mysql.ErrWriteConflictInTiDB ||
					strings.Contains(mysqlErr.Message, "Information schema is changed") {
					ignoreErr = true
				}
			}

			if !ignoreErr {
				client.Errorf("[gc unsafe destroy range] Failed to commit transaction on table %v, err: %v", tableName, err)
			}
		} else {
			totalDoneQueries++
		}
	}
}

// Creates a table with specified name and load specified amount of rows.
func (c *GCUnsafeDestroyRangeCase) generateTable(tableName string, tableSize int) error {
	_, err := c.db.Exec(fmt.Sprintf("create table %s (id integer not null primary key, a integer, b varchar(128), c integer);", tableName))
	if err != nil {
		client.Errorf("[gc unsafe destroy range] Failed to create table %v, err: %v", tableName, err)
		return errors.Trace(err)
	}
	for i := 0; i < tableSize; i++ {
		_, err = c.db.Exec(fmt.Sprintf("insert into %s values (?, ?, ?, ?);", tableName), i, rand.Int()%2333333, randStr(rand.Int()%128), rand.Int()%2333333)
		if err != nil {
			client.Errorf("[gc unsafe destroy range] Failed to insert row %v, err: %v", i, err)
			return errors.Trace(err)
		}
	}
	client.Infof("[gc unsafe destroy range] Created table %v and loaded %v rows", tableName, tableSize)
	return nil
}

func randStr(len int) string {
	res := ""
	for i := 0; i < len; i++ {
		res += string('a' + (rand.Int() % 26))
	}
	return res
}