package main

import (
	"database/sql"
	"fmt"
	"log"
	"math/rand"
	"sync"
	"sync/atomic"
	"time"

	"github.com/juju/errors"
	client "github.com/pingcap/schrodinger/pkg/test_client"
)

func getDB(cli *client.Client) (*sql.DB, error) {
	dbAddr := cli.TiDBService()
	dbDSN := fmt.Sprintf("root:@tcp(%s)/test", dbAddr)
	db, _ := sql.Open("mysql", dbDSN)
	err := db.Ping()
	if err != nil {
		return nil, errors.Trace(err)
	}
	return db, nil
}

func mustQuery(cli *client.Client, query string) *sql.Rows {
	db, err := getDB(cli)
	if err != nil {
		client.Fatalf("get db error %v", err)
	}
	retryCount := 0
	rows, err := db.Query(query)
	for err != nil {
		if retryCount >= 10 {
			client.Fatal("try mustQuery too many times, test failed")
		}

		db, err = getDB(cli)
		if err != nil {
			client.Fatalf("get db error %v", err)
		}
		rows, err = db.Query(query)
		if err == nil {
			db.Close()
			return rows
		}
		retryCount++
		time.Sleep(time.Duration(7) * time.Second)
	}
	return rows
}

func mustExec(db *sql.DB, query string, args ...interface{}) {
	retryCount := 0
	_, err := db.Exec(query, args...)
	for err != nil {
		if retryCount >= 10 {
			log.Println("A test case failed.")
			log.Println("The case is: " + query)
			log.Println(err)
			break
		}
		_, err = db.Exec(query, args...)
		if err == nil {
			return
		}
		retryCount++
		time.Sleep(time.Duration(7) * time.Second)
	}
}

func dateImport(cli *client.Client, ch chan int) {
	log.Println("Importing test data....")
	db, err := getDB(cli)
	if err != nil {
		client.Fatalf("get db failed %+v", err)
	}
	mustExec(db, "create table test_delete (c1 int, c2 int, primary key(c1))")
	mustExec(db, "create table test_mixed (c1 int, c2 int, primary key(c1))")
	log.Println("Importing test data succsess")

	for _, tbl := range parallelDDLTbls {
		dataImportForParallelDDL(cli, tbl)
	}

	ch <- 1
}

func testDelete(cli *client.Client, ch chan int) {
	//run case
	time.Sleep(120 * time.Second)
	workerNum := 5
	rowCount := 10000
	batch := rowCount / workerNum
	start := time.Now()
	var wg sync.WaitGroup
	wg.Add(workerNum)
	for i := 0; i < workerNum; i++ {
		go func(i int) {
			defer wg.Done()
			db, err := getDB(cli)
			if err != nil {
				client.Fatalf("get db failed %+v", err)
			}
			for j := 0; j < batch; j++ {
				k := batch*i + j
				mustExec(db, fmt.Sprintf("insert into test_delete values (%d, %d)", k, k))
				mustExec(db, fmt.Sprintf("delete from test_delete where c1 = %d", k))
			}
		}(i)
	}
	wg.Wait()
	end := time.Now()
	log.Printf("[TestSimpleDelete][Time Cost]%v\n", end.Sub(start))
	//check
	cnt := 0
	rows := mustQuery(cli, "select * from test_delete")
	defer rows.Close()
	for rows.Next() {
		cnt++
	}
	if cnt != 0 {
		log.Printf("[TestSimpleDelete] Failed!\n")
	} else {
		log.Printf("[TestSimpleDelete] Success!\n")
	}
	ch <- 1
}

func testMixed(cli *client.Client, ch chan int) {
	time.Sleep(150 * time.Second)
	workerNum := 5
	rowCount := 10000
	batch := rowCount / workerNum

	start := time.Now()

	var wg sync.WaitGroup
	wg.Add(workerNum)
	for i := 0; i < workerNum; i++ {
		go func(i int) {
			defer wg.Done()
			db, err := getDB(cli)
			if err != nil {
				client.Fatalf("get db failed %+v", err)
			}
			for j := 0; j < batch; j++ {
				k := batch*i + j
				mustExec(db, fmt.Sprintf("insert into test_mixed values (%d, %d)", k, k))
			}
		}(i)
	}
	wg.Wait()

	end := time.Now()
	fmt.Printf("[TestSimpleMixed][Insert][Time Cost]%v\n", end.Sub(start))

	start = time.Now()

	rowID := int64(rowCount)
	defaultValue := int64(-1)

	wg.Add(workerNum)
	for i := 0; i < workerNum; i++ {
		go func() {
			defer wg.Done()
			db, err := getDB(cli)
			if err != nil {
				client.Fatalf("get db failed %+v", err)
			}
			for j := 0; j < batch; j++ {
				key := atomic.AddInt64(&rowID, 1)
				mustExec(db, fmt.Sprintf("insert into test_mixed values (%d, %d)", key, key))
				key = int64(rand.Intn(rowCount))
				mustExec(db, fmt.Sprintf("update test_mixed set c2 = %d where c1 = %d", defaultValue, key))
				key = int64(rand.Intn(rowCount))
				mustExec(db, fmt.Sprintf("delete from test_mixed where c1 = %d", key))
			}
		}()
	}
	wg.Wait()

	end = time.Now()
	fmt.Printf("[TestSimpleMixed][Mixed][Time Cost]%v\n", end.Sub(start))
	updateCount := int64(0)
	insertCount := int64(0)
	rows := mustQuery(cli, "select * from test_mixed")
	rowNum := 1
	defer rows.Close()
	for rows.Next() {
		var c1, c2 int64
		if err := rows.Scan(&c1, &c2); err != nil {
			log.Fatal(err)
		}
		if c1 == c2 {
			insertCount++
		} else if c2 == defaultValue && c1 < int64(rowCount) {
			updateCount++
		} else {
			log.Fatalf("[TestSimpleMixed fail]invalid row: %d", rowNum)
		}
		rowNum++
	}
	deleteCount := atomic.LoadInt64(&rowID) - insertCount - updateCount
	if insertCount > 0 && updateCount > 0 && deleteCount > 0 {
		log.Println("Mixed Test Success!")
	} else {
		log.Println("Mixed Test Failed!")
	}
	ch <- 1
}

func batchInsertRows(db *sql.DB, tbl string, startKey, endKey int) error {
	sql := fmt.Sprintf("insert into %s values", tbl)
	for i := startKey; i < endKey; i++ {
		sql += fmt.Sprintf("(%d, '%d', '%d', %d, %d, %d)", i, i, i, i, i, i)
		if i != endKey-1 {
			sql += ","
		}
	}
	log.Printf("insert startKey %d, endKey %d", startKey, endKey)
	_, err := db.Exec(sql)
	if err != nil {
		log.Printf("sql: %v, err: %v", sql, err)
	}
	return err
}

func dataImportForParallelDDL(cli *client.Client, tbl string) {
	db, err := getDB(cli)
	if err != nil {
		client.Fatalf("get db error %v", err)
	}
	sql := fmt.Sprintf("create table if not exists %s(a int, b varchar(256), c varchar(256), d int, e int, f int)", tbl)
	_, err = db.Exec(sql)
	if err != nil {
		log.Fatalf("create table failed %v", err)
	}
	db.Close()

	totalCnt := 10000
	goCnt := 50
	step := totalCnt / goCnt
	batch := 100
	if step/batch < 1 {
		batch = 1
	}
	wg := sync.WaitGroup{}
	for i := 0; i < goCnt; i++ {
		sKey := i * step
		wg.Add(1)
		go func(num, startKey int) {
			defer wg.Done()
			db, err := getDB(cli)
			if err != nil {
				client.Fatalf("get db error %v", err)
			}
			defer db.Close()
			for j := 0; j < step/batch; j++ {
				err = batchInsertRows(db, tbl, startKey, startKey+batch)
				if err != nil {
					log.Printf("insert data err %v", err)
					return
				}
				startKey += batch

			}
			log.Printf("exec sql startKey %d, endKey %d end", startKey, startKey+step)
		}(i, sKey)
	}
	wg.Wait()
}

// buildDDL generates the statements of DDL.
func buildDDL(tbls ...string) [][]string {
	sqlsSli := make([][]string, 0, 31)
	for _, tbl := range tbls {
		sqls := make([]string, 0, 31)
		sql := fmt.Sprintf("alter table %s add index idx0(a)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx1(b)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx2(c)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx3(d)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx4(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx5(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx6(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx7(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx8(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx9(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx10(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx11(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx12(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx13(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx14(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx15(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx16(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx17(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx18(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx19(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx20(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx21(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx22(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx23(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx24(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx25(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx26(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx27(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx28(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx29(e)", tbl)
		sqls = append(sqls, sql)
		sql = fmt.Sprintf("alter table %s add index idx30(e)", tbl)
		sqls = append(sqls, sql)
		sqlsSli = append(sqlsSli, sqls)
		log.Printf("tbl %v, sqls %v", tbl, sqls)

		sqls1 := make([]string, 0, 31)
		sql = fmt.Sprintf("alter table %s add column col int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col1 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col2 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col3 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col4 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col5 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col6 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col7 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col8 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col9 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col10 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col11 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col12 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col13 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col14 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col15 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col16 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col17 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col18 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col19 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col20 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col21 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col22 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col23 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col24 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col25 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col26 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col27 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col28 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s add column col29 int", tbl)
		sqls1 = append(sqls1, sql)
		sql = fmt.Sprintf("alter table %s drop column f", tbl)
		sqls1 = append(sqls1, sql)
		sqlsSli = append(sqlsSli, sqls1)
		log.Printf("tbl %v, sqls1 %v", tbl, sqls1)
	}
	log.Printf("sqls len %d ", len(sqlsSli))
	return sqlsSli
}

var parallelDDLTbls = []string{"parallel_ddl_tbl_0", "parallel_ddl_tbl_1", "parallel_ddl_tbl_2"}

func execParallelDDL(cli *client.Client, wg *sync.WaitGroup, tbls ...string) {
	time.Sleep(120 * time.Second)
	sqlsSli := buildDDL(tbls...)
	for i, sqls := range sqlsSli {
		wg.Add(1)
		go func(num int, sqls []string) {
			defer wg.Done()
			db, err := getDB(cli)
			if err != nil {
				client.Fatalf("get db error %v", err)
			}
			defer db.Close()
			log.Printf("start run ddl no.%d", num)
			for _, sql := range sqls {
				_, err := db.Exec(sql)
				if err != nil {
					log.Printf("no.%d, sql: %s, err: %v", num, sql, err)
				} else {
					log.Printf("no.%d, sql: %s", num, sql)
				}
			}
		}(i, sqls)
	}
}
func checkParallelDDLTables(cli *client.Client) {
	db, err := getDB(cli)
	if err != nil {
		client.Fatalf("get db error %v", err)
	}
	defer db.Close()
	for _, tbl := range parallelDDLTbls {
		sql := fmt.Sprintf("admin check table %s", tbl)
		_, err := db.Exec(sql)
		if err != nil {
			log.Fatalf("admin check table failed %v", err)
		}
	}
}