package pkg

import (
	"context"
	"database/sql"
	"fmt"
	"math/rand"
	"time"

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

	"github.com/pingcap/endless/pkg/client/tidb"
)

const addDropColumnPrefix = "adc_"
const changeColumnTypePrefix = "cct_"

// AddDropColumn execute the ddl and dml in the sequence
// add column ,Update limit 2(value-> null,boundary value) ,select column limit 2 ,drop column
func AddDropColumn(ctx context.Context, db *sql.DB, dbName string, errorHandler func(error) error) error {
	rand.Seed(time.Now().Unix())
	cli := tidb.NewClient(db)
	tables, err := cli.ShowBaseTableNames(dbName)
	if err != nil {
		return err
	}
	optionalColumnDescs := []struct {
		desc   string
		values []string
	}{
		{"int", []string{"null", "-2147483648", "2147483647", "0"}},
		{"int default null", []string{"null", "-2147483648", "2147483647", "0"}},
		{"int default 10", []string{"null", "-2147483648", "2147483647", "0"}},
		{"int unsigned", []string{"null", "4294967295", "0", "10"}},
		{"bigint", []string{"null", "-9223372036854775808", "9223372036854775807", "0"}},
		{"tinyint", []string{"null", "-10", "20", "0"}},
		{"smallint", []string{"null", "-40", "50", "0"}},
		{"mediumint", []string{"null", "-123", "456", "0"}},

		{"decimal(5,2)", []string{"null", "-999.99", "999.99", "0"}},
		{"float", []string{"null", "-999.99", "999.99", "0"}},
		{"double", []string{"null", "-999.99", "999.99", "0"}},

		{"tinyblob", []string{"null", "'aa'", "'bb'", "0"}},
		{"tinytext", []string{"null", "'aa'", "'bb'", "0"}},
		{"blob", []string{"null", "'aa'", "'bb'", "0"}},
		{"text", []string{"null", "'aa'", "'bb'", "0"}},
		{"mediumblob", []string{"null", "'aa'", "'bb'", "0"}},
		{"mediumtext", []string{"null", "'aa'", "'bb'", "0"}},
		{"longblob", []string{"null", "'aa'", "'bb'", "0"}},
		{"longtext", []string{"null", "'aa'", "'bb'", "0"}},

		{"json", []string{"null", `'{"a":"b"}'`, `'{}'`, `'[1,2,3]'`}},

		{"datetime(6)", []string{"null", "'1000-01-01 00:00:00.000000'", "'9999-12-31 23:59:59.999999'", "'2020-10-10 10:10:10'"}},
		{"date", []string{"null", "'1000-01-01'", "'9999-12-31'", "'2020-10-10'"}},
		{"datetime default current_timestamp on Update current_timestamp", []string{"null", "'1000-01-01'", "'9999-12-31'", "'2020-10-10'"}},
		{"timestamp default current_timestamp on Update current_timestamp", []string{"null", "'2010-01-01'", "'1991-12-31'", "'2020-10-10'"}},
		{"time", []string{"null", "'-838:59:59'", "'838:59:59'", "'12:12:01'"}},
	}
	index := 0
	for {
		index++
		select {
		case <-ctx.Done():
			return nil
		default:
		}
		columnName := fmt.Sprintf("%s%d", addDropColumnPrefix, index)
		optionalColumnDesc := optionalColumnDescs[rand.Intn(len(optionalColumnDescs))]
		for _, t := range tables {
			sqls := []string{
				// 多个 ddl case 构成一个执行 plan 的时候，有可能会因为 column 没有被上一个 case 删除而导致后续失败
				// 先执行 drop column，然后忽略 ERROR 1091 (42000): Can't DROP 'e'; check that column/key exists 错误
				fmt.Sprintf("alter table %s drop column %s", t, columnName),
				fmt.Sprintf("alter table %s add column %s %s", t, columnName, optionalColumnDesc.desc),
				fmt.Sprintf("Update %s set %s = %s limit 2", t, columnName,
					optionalColumnDesc.values[rand.Intn(len(optionalColumnDesc.values))]),
				fmt.Sprintf("select /*+ read_from_storage(TIFLASH[%s]) */ %s from %s limit 2", t, columnName, t),
				"select sleep(30)",
				fmt.Sprintf("alter table %s drop column %s", t, columnName),
			}
			for _, s := range sqls {
				_, err := db.ExecContext(ctx, s)
				log.Info("ddl sequence", zap.String("sql", s))
				if err != nil {
					log.Error("sql error", zap.String("sql", s), zap.Error(err))
					if CtxDoneHandler(err) == nil {
						return nil
					}
					if errorHandler != nil {
						err = errorHandler(err)
						if err != nil {
							return err
						}
					} else {
						return err
					}
					continue
				}
			}
		}
	}
}

// ChangeColumnType execute the ddl and dml in the sequence
// add column, Update limit 2, select limit 2, change column type , select limit 2
func ChangeColumnType(ctx context.Context, db *sql.DB, dbName string, errorHandler func(error) error) error {
	rand.Seed(time.Now().Unix())
	cli := tidb.NewClient(db)
	tables, err := cli.ShowBaseTableNames(dbName)
	if err != nil {
		return err
	}
	optionalColumnDescs := []struct {
		desc       string
		values     []string
		modifyDesc []string
	}{
		// meta-only change
		// reorg-data change
		// 当所变更的类型与已经存在的数据行产生冲突时，TiDB 会进行报错处理。在上述例子中，TiDB 将进行如下报错，如
		// ERROR 1265 (01000): Data truncated for column 'a', value is 'abcdefg'
		//
		{"int", []string{"null", "-2147483648", "2147483647", "0"}, []string{"bigint", "varchar(20)", "json"}},
		{"int", []string{"null", "20201010", "10000101"}, []string{"datetime", "int unsigned", "json"}},
		{"int default 10", []string{"20201010", "10000101"}, []string{"int default null", "json"}},
		{"varchar(10)", []string{"null", "'10'", "'20'", "0"}, []string{"int", "bigint", "decimal", "json", "time"}},
		{"varchar(20)", []string{"null", "'[1,2,3]'", `'{"a":"b"}'`, "0"}, []string{"varchar(10)", "json"}},
		{"decimal(5,2)", []string{"null", "-999.99", "999.99", "0"}, []string{"varchar(10)", "int", "decimal(6,3)", "decimal(5,1)", "json"}},
		{"json", []string{"null", "20201010", "10000101"}, []string{"int", "datetime"}},
		// set global tidb_enable_change_column_type=true;
		// may not support in low version. ERROR 8200 (HY000): Unsupported modify column: column type conversion between 'varchar' and 'non-varchar' is currently unsupported yet
	}
	index := 0
	for {
		index++
		select {
		case <-ctx.Done():
			return nil
		default:
		}
		columnName := fmt.Sprintf("%s%d", changeColumnTypePrefix, index)
		optionalColumnDesc := optionalColumnDescs[rand.Intn(len(optionalColumnDescs))]
		for _, t := range tables {
			sqls := []string{
				fmt.Sprintf("alter table %s drop column %s", t, columnName),
				fmt.Sprintf("alter table %s add column %s %s", t, columnName, optionalColumnDesc.desc),
				fmt.Sprintf("Update %s set %s = %s limit 2", t, columnName,
					optionalColumnDesc.values[rand.Intn(len(optionalColumnDesc.values))]),
				fmt.Sprintf("select /*+ read_from_storage(TIFLASH[%s]) */ %s from %s limit 2", t, columnName, t),
				"select sleep(30)",
				fmt.Sprintf("alter table %s modify %s %s", t,
					columnName, optionalColumnDesc.modifyDesc[rand.Intn(len(optionalColumnDesc.modifyDesc))]),
				fmt.Sprintf("select /*+ read_from_storage(TIFLASH[%s]) */ %s from %s limit 2", t, columnName, t),
				fmt.Sprintf("alter table %s drop column %s", t, columnName),
			}
			for _, s := range sqls {
				_, err := db.ExecContext(ctx, s)
				log.Info("ddl sequence", zap.String("sql", s))
				if err != nil {
					log.Error("sql error", zap.String("sql", s), zap.Error(err))
					if CtxDoneHandler(err) == nil {
						return nil
					}
					if errorHandler != nil {
						err = errorHandler(err)
						if err != nil {
							return err
						}
					} else {
						return err
					}
					continue
				}
			}
		}
	}
}

func RandomTruncateTable(ctx context.Context, tidbClient *tidb.Client, db *sql.DB, dbName string, errorHandler func(error) error) error {
	tables, err := tidbClient.ShowTableNames(dbName)
	if err != nil {
		return err
	}
	for {
		select {
		case <-ctx.Done():
			return nil
		default:
		}
		tableName := tables[rand.Intn(len(tables))]
		partitions, err := tidbClient.ShowTablePartitionNames(dbName, tableName)
		if err != nil {
			return err
		}
		if len(partitions) == 0 {
			continue
		}
		partition := partitions[rand.Intn(len(partitions))]
		s := fmt.Sprintf("ALTER TABLE %s TRUNCATE PARTITION %s", tableName, partition)
		log.Info("ddl sequence", zap.String("sql", s))
		if _, err = db.Exec(s); err != nil {
			if err != nil {
				log.Error("sql error", zap.String("sql", s), zap.Error(err))
				if CtxDoneHandler(err) == nil {
					return nil
				}
				if errorHandler != nil {
					err = errorHandler(err)
					if err != nil {
						return err
					}
				} else {
					return err
				}
				continue
			}
		}
		if Sleep(ctx, 30*time.Minute) {
			return nil
		}
	}
}