package data

import (
	"context"
	"database/sql"
	"fmt"
	"math/rand"
	"os/exec"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/schrodinger/util"
)

func init() {
	rand.Seed(time.Now().UnixNano())
}

// Index is index
type Index struct {
	Name     string
	Category int
	Type     string
}

// Client is client to generat ddl & dml data
type Client struct {
	ctx context.Context
	sync.RWMutex
	columnMap map[string]string
	viewMap   map[string]bool
	indexMap  map[string]*Index
}

// DBInfo stores db paramters, used by some external tools such as pt-osc
type DBInfo struct {
	user     string
	password string
	host     string
	port     int
}

type ConsumeDDLArgs struct {
	ConsumeDDLInfo
	ConsumeDDLOpts
}

type ConsumeDDLInfo struct {
	Db        *sql.DB
	DbInfo    *DBInfo
	DbName    string
	TableName string
	Ddl       string
}
type ConsumeDDLOpts struct {
	Done          chan struct{}
	ErrCh         chan error
	OnlineDDL     bool
	OnlineDDLTool string
}

func NewDBInfo(user, password, host string, port int) *DBInfo {
	return &DBInfo{
		user:     user,
		password: password,
		host:     host,
		port:     port,
	}
}

// NewClient creates new client
func NewClient(ctx context.Context) *Client {
	return &Client{
		ctx:       ctx,
		columnMap: make(map[string]string),
		viewMap:   make(map[string]bool),
		indexMap:  make(map[string]*Index),
	}
}

// CreateDDLFile creates ddl file
func (c *Client) CreateDDLFile(filename string, ddlCount int, ddlType []int, indexCategorys []int) error {
	ddls := c.CreateRandomDDLData(ddlCount, ddlType, indexCategorys)
	for _, ddl := range ddls {
		err := util.WriteToFile(fmt.Sprintf("%s\n", ddl), filename)
		if err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}

// ConsumeDDL consumes ddl
func (c *Client) ConsumeDDL(args *ConsumeDDLArgs) {
	db := args.ConsumeDDLInfo.Db
	dbInfo := args.ConsumeDDLInfo.DbInfo
	dbName := args.ConsumeDDLInfo.DbName
	tableName := args.ConsumeDDLInfo.TableName
	ddl := args.ConsumeDDLInfo.Ddl

	done := args.ConsumeDDLOpts.Done
	errCh := args.ConsumeDDLOpts.ErrCh
	onlineDDL := args.ConsumeDDLOpts.OnlineDDL
	onlineDDLTool := args.ConsumeDDLOpts.OnlineDDLTool

	defer func() {
		done <- struct{}{}
	}()

	sql, err := c.generateSQLByDdlLine(dbName, tableName, ddl)
	if err != nil {
		errCh <- errors.Trace(err)
		return
	}

	if onlineDDL {
		log.Debugf("[consume ddl] online ddl with pt-osc %s", sql)
		alterSQL := strings.Join(strings.Split(sql, " ")[3:], " ")
		err = c.onlineDDL(alterSQL, dbName, tableName, onlineDDLTool, dbInfo)
		if err != nil {
			errCh <- errors.Trace(err)
			return
		}
	} else {
		log.Debugf("[consume ddl] execute ddl %s", sql)
		err = execSQL(db, sql)
		if err != nil {
			errCh <- errors.Trace(err)
			return
		}
	}

	if len(c.columnMap) < 1 {
		return
	}

	for i := 0; i < rand.Intn(10000); i++ {
		sql, err = c.genDml(db, dbName, tableName, []int{Insert, Update, Delete})
		if err != nil {
			errCh <- errors.Trace(err)
			return
		}
		log.Debugf("[consume ddl] execute dml %d %s", i, sql)
		err = execSQL(db, sql)
		if err != nil {
			errCh <- errors.Trace(err)
			return
		}
	}
}

// StartDML starts dml
func (c *Client) StartDML(db *sql.DB, schemaName, tableName string) error {
	for i := 0; i < rand.Intn(100); i++ {
		time.Sleep(30 * time.Millisecond)
		sql, err := c.genDml(db, schemaName, tableName, []int{Insert, Update, Delete})
		if err != nil {
			return errors.Trace(err)
		}
		log.Debugf("[consume dml] execute dml %d %s", i, sql)
		err = execSQL(db, sql)
		if err != nil {
			return errors.Trace(err)
		}
	}
	return nil
}

// CreateRandomDataWithFixedDDL create random data with fixed ddl
func (c *Client) CreateRandomDataWithFixedDDL(db *sql.DB, dbName, tableName string, ddlFile string) error {
	err := cleanTable(db, dbName, tableName)
	if err != nil {
		return errors.Trace(err)
	}

	ddls, err := loadDDLFile(ddlFile)
	if err != nil {
		return errors.Trace(err)
	}

	for _, ddl := range ddls {
		select {
		case <-c.ctx.Done():
			return nil
		default:
		}

		sql, err := c.generateSQLByDdlLine(dbName, tableName, ddl)
		if err != nil {
			return errors.Trace(err)
		}

		err = execSQL(db, sql)
		if err != nil {
			return errors.Trace(err)
		}

		if len(c.columnMap) < 1 {
			continue
		}

		for i := 0; i < rand.Intn(10000); i++ {
			time.Sleep(30 * time.Millisecond)
			sql, err = c.genDml(db, dbName, tableName, []int{Insert, Update, Delete})
			if err != nil {
				return errors.Trace(err)
			}
			err = execSQL(db, sql)
			if err != nil {
				return errors.Trace(err)
			}
		}
	}
	return nil
}

// GetColumnType returns definition type of specific column and whether column exists
func (c *Client) GetColumnType(columnName string) (string, bool) {
	ct, ok := c.columnMap[columnName]
	return ct, ok
}

func (c *Client) genDml(db *sql.DB, schemaName, tableName string, dmlType []int) (string, error) {
	columns := getMapKey(c.columnMap)
	switch dmlType[rand.Intn(len(dmlType))] {
	case Insert:
		sql, err := c.genInsertTable(schemaName, tableName, columns)
		return sql, errors.Trace(err)
	case Update:
		sql, err := c.genUpdateTable(db, schemaName, tableName, columns)
		return sql, errors.Trace(err)
	case Delete:
		sql, err := c.genDeleteTable(db, schemaName, tableName)
		return sql, errors.Trace(err)
	}

	return "", errors.NotSupportedf("dml type")
}

// CreateRandomDDLData creates random ddl file. can only be parse by this client
func (c *Client) CreateRandomDDLData(ddlCount int, ddlType []int, indexCategorys []int) []string {
	c.Lock()
	defer c.Unlock()

	var ddlLines []string

	for i := 0; i < ddlCount; i++ {
		var ddlLine string
		switch ddlType[rand.Intn(len(ddlType))] {
		case AddColumn:
			for {
				column := genRandomString(20)
				_, ok := c.columnMap[column]
				if !ok {
					def := genRandomColumnDef()
					c.columnMap[column] = def
					ddlLine = fmt.Sprintf("%d %s %s", AddColumn, column, def)
					break
				}
			}

		case DeletColumn:
			if len(c.columnMap) < 2 {
				i--
				continue
			}
			column := getMapKey(c.columnMap)[rand.Intn(len(c.columnMap))]
			delete(c.columnMap, column)
			delete(c.indexMap, column)
			ddlLine = fmt.Sprintf("%d %s", DeletColumn, column)

		case AlterColumn:
			if len(c.columnMap) < 1 {
				i--
				continue
			}
			column := getMapKey(c.columnMap)[rand.Intn(len(c.columnMap))]
			def := c.columnMap[column]
			for {
				newDef := genRandomColumnDef()
				if newDef != def {
					c.columnMap[column] = newDef
					ddlLine = fmt.Sprintf("%d %s %s", AlterColumn, column, newDef)
					break
				}
			}

		case AddAndAlterColumn:
			if len(c.columnMap) < 1 {
				i--
				continue
			}

			for {
				column := genRandomString(20)
				_, ok := c.columnMap[column]
				if !ok {
					def := "VARCHAR(40)"
					ddlLine = fmt.Sprintf("%d %s %s", AddColumn, column, def)
					// This may cause generated SQLs count one more than ddlCount
					ddlLines = append(ddlLines, ddlLine)
					i++
					newdef := "VARCHAR(20)"
					c.columnMap[column] = newdef
					ddlLine = fmt.Sprintf("%d %s %s", AlterColumn, column, newdef)
					break
				}
			}

		case AddWithIndexAndDeleteColumn:
			if len(c.columnMap) < 1 {
				i--
				continue
			}

			for {
				column := genRandomString(20)
				_, ok := c.columnMap[column]
				if !ok {
					def := "INT"
					ddlLine = fmt.Sprintf("%d %s %s", AddColumn, column, def)
					ddlLines = append(ddlLines, ddlLine)
					c.columnMap[column] = def

					// This may cause generated SQLs count two more than ddlCount
					i += 2

					index := c.createIndexObject(column, indexCategorys)
					ddlLine = fmt.Sprintf("%d %s %s %d %s", CreateIndex, column, index.Name, index.Category, index.Type)
					ddlLines = append(ddlLines, ddlLine)

					delete(c.columnMap, column)
					delete(c.indexMap, column)
					ddlLine = fmt.Sprintf("%d %s", DeletColumn, column)
					break
				}
			}

		case CreateIndex:
			if len(c.columnMap) < 1 {
				i--
				continue
			}

			column, ok := c.getUnIndexedColumn()
			if !ok {
				i--
				continue
			}

			index := c.createIndexObject(column, indexCategorys)

			ddlLine = fmt.Sprintf("%d %s %s %d %s", CreateIndex, column, index.Name, index.Category, index.Type)
		case DropIndex:
			if len(c.indexMap) < 1 {
				i--
				continue
			}
			index := c.removeIndexObject()
			ddlLine = fmt.Sprintf("%d %s", DropIndex, index.Name)

		case Flush:
			if len(c.columnMap) < 1 {
				i--
				continue
			}

			switch rand.Intn(2) {
			case 0:
				ddlLine = fmt.Sprintf("%d %s", Flush, genRandomFlushOption())
			case 1:
				ddlLine = fmt.Sprintf("%d %s", Flush, "TABLES")
			}

		case OptimizeTable:
			if len(c.columnMap) < 1 {
				i--
				continue
			}

			ddlLine = fmt.Sprintf("%d", OptimizeTable)

		case CreateView:
			if len(c.columnMap) < 1 {
				i--
				continue
			}

			for {
				viewName := genRandomString(20)
				_, ok := c.viewMap[viewName]
				if !ok {
					c.viewMap[viewName] = true
					ddlLine = fmt.Sprintf("%d %s", CreateView, viewName)
					break
				}
			}
		}

		ddlLines = append(ddlLines, ddlLine)
	}
	return ddlLines
}

func (c *Client) getUnIndexedColumn() (string, bool) {
	if len(c.columnMap) < 1 {
		return "", false
	}

	for column, columnDef := range c.columnMap {
		if !supportIndex(columnDef) {
			continue
		}
		_, ok := c.indexMap[column]
		if !ok {
			return column, true
		}
	}

	return "", false
}

func (c *Client) createIndexObject(column string, indexCategorys []int) *Index {
	keys := make(map[string]struct{})
	for _, item := range c.indexMap {
		keys[item.Name] = struct{}{}
	}

	for {
		indexName := genRandomString(20)
		_, ok := keys[indexName]
		if !ok {
			c.indexMap[column] = &Index{
				Name:     indexName,
				Category: indexCategorys[rand.Intn(len(indexCategorys))],
				Type:     []string{"BTREE", "HASH"}[rand.Intn(2)],
			}
			return c.indexMap[column]
		}
	}
}

func (c *Client) removeIndexObject() *Index {
	column := getMapKeyForIndex(c.indexMap)[rand.Intn(len(c.indexMap))]
	index := c.indexMap[column]
	delete(c.indexMap, column)
	return index
}

func (c *Client) generateSQLByDdlLine(schemaName, tableName, ddlLine string) (string, error) {
	details := strings.Split(ddlLine, " ")

	ddlType, err := strconv.Atoi(details[0])
	if err != nil {
		return "", errors.Trace(err)
	}

	switch ddlType {
	case AddColumn:
		c.columnMap[details[1]] = details[2]
		return genAddColumnSQL(schemaName, tableName, details[1], details[2]), nil
	case DeletColumn:
		delete(c.columnMap, details[1])
		return genDelColumnSQL(schemaName, tableName, details[1]), nil
	case AlterColumn:
		c.columnMap[details[1]] = details[2]
		return genAlterColumnDefSQL(schemaName, tableName, details[1], details[2]), nil
	case CreateIndex:
		category, err := strconv.Atoi(details[3])
		if err != nil {
			return "", errors.Trace(err)
		}
		return genCreateIndex(schemaName, tableName, details[1], details[2], category, details[4]), nil
	case DropIndex:
		return genDropIndex(schemaName, tableName, details[1]), nil
	case Flush:
		return genFlushSQL(schemaName, tableName, details[1:]...), nil
	case OptimizeTable:
		return genOptimizeSQL(schemaName, tableName), nil
	case CreateView:
		return genCreateViewSQL(schemaName, tableName, details[1]), nil
	}
	return "", errors.NotSupportedf("ddl type %d", ddlType)
}

func (c *Client) genInsertTable(schemaName, tableName string, columnNames []string) (string, error) {
	keys := strings.Join(columnNames, ",")
	values, err := c.genColumnValue(columnNames)
	if err != nil {
		return "", errors.Trace(err)
	}

	return fmt.Sprintf("INSERT INTO `%s`.`%s` (%s) values (%s)", schemaName, tableName, keys, values), nil
}

func (c *Client) genUpdateTable(db *sql.DB, schemaName, tableName string, columnNames []string) (string, error) {
	id, err := getRandID(db, schemaName, tableName)
	if err != nil {
		return "", errors.Trace(err)
	}
	columnName := columnNames[rand.Intn(len(columnNames))]
	return fmt.Sprintf("UPDATE `%s`.`%s` SET %s = %s WHERE id=%d", schemaName, tableName, columnName, genRandomValue(c.columnMap[columnName]), id), nil
}

func (c *Client) genDeleteTable(db *sql.DB, schemaName, tableName string) (string, error) {
	id, err := getRandID(db, schemaName, tableName)
	if err != nil {
		return "", errors.Trace(err)
	}

	return fmt.Sprintf("DELETE FROM `%s`.`%s` WHERE id=%d", schemaName, tableName, id), nil
}

func (c *Client) genColumnValue(columnNames []string) (string, error) {
	var vals []string

	for _, columnName := range columnNames {
		columnDef, ok := c.columnMap[columnName]
		if !ok {
			return "", errors.NotFoundf("column def for %s", columnName)
		}
		vals = append(vals, genRandomValue(columnDef))
	}

	values := strings.Join(vals, ",")
	return values, nil
}

func (c *Client) onlineDDL(alter, schema, table, tool string, dbInfo *DBInfo) error {
	if tool == "pt" {
		return c.ptOnlineDDL(alter, schema, table, dbInfo)
	} else if tool == "gh-ost" {
		return c.ghostOnlineDDL(alter, schema, table, dbInfo)
	}
	return errors.Errorf("unknown online ddl tool %s", tool)
}

func (c *Client) ptOnlineDDL(alter, schema, table string, dbInfo *DBInfo) error {
	fields := []string{
		"--user=" + dbInfo.user, "--host=" + dbInfo.host,
		fmt.Sprintf("--port=%d", dbInfo.port), "--alter=" + alter,
		fmt.Sprintf("D=%s,t=%s", schema, table), "--print",
	}
	if dbInfo.password != "" {
		fields = append(fields, "--password="+dbInfo.password)
	}

	cmd := "pt-online-schema-change"
	var err error
	err = c.onlineDDLRunCmd(cmd, fields, "--dry-run")
	if err != nil {
		return errors.Trace(err)
	}
	err = c.onlineDDLRunCmd(cmd, fields, "--execute")
	if err != nil {
		return errors.Trace(err)
	}

	return nil
}

func (c *Client) ghostOnlineDDL(alter, schema, table string, dbInfo *DBInfo) error {
	fields := []string{
		"--user=" + dbInfo.user, "--host=" + dbInfo.host,
		fmt.Sprintf("--port=%d", dbInfo.port),
		"--database=" + schema, "--table=" + table, "--alter=" + alter,
		"--allow-on-master", "--initially-drop-ghost-table",
		"--initially-drop-old-table", "-ok-to-drop-table",
		"-serve-socket-file=" + fmt.Sprintf("/tmp/gh-ost.%s.%d.%s.%s.sock", dbInfo.host, dbInfo.port, schema, table),
	}
	if dbInfo.password != "" {
		fields = append(fields, "--password="+dbInfo.password)
	}

	cmd := "gh-ost"
	var err error
	// dry run without --execute option
	err = c.onlineDDLRunCmd(cmd, fields)
	if err != nil {
		return errors.Trace(err)
	}
	err = c.onlineDDLRunCmd(cmd, fields, "--execute")
	if err != nil {
		return errors.Trace(err)
	}

	return nil
}

func (c *Client) onlineDDLRunCmd(cmd string, fields []string, args ...string) error {
	fields = append(fields, args...)

	out, err := exec.Command(cmd, fields...).Output()
	log.Infof("%s(%v) online ddl output: %s", cmd, fields, string(out))

	if err != nil {
		if ee, ok := err.(*exec.ExitError); ok {
			log.Errorf("online ddl ExitError: %s, cmd: %s %v", string(ee.Stderr), cmd, fields)
		} else {
			log.Errorf("online ddl error: %s, cmd: %s %v", err, cmd, fields)
		}
		return errors.Trace(err)
	}
	return nil
}