package mysql

import (
	"fmt"
	"strconv"
	"strings"

	"github.com/ngaut/log"
	"github.com/pingcap/tidb/ast"
	"github.com/pingcap/tidb/model"
	"github.com/pingcap/tidb/mysql"
	"github.com/pingcap/tidb/types"
	"github.com/pingcap/tidb/util/charset"
)

func defaultValueToSQL(opt *ast.ColumnOption) string {
	sql := " DEFAULT "
	datum := opt.Expr.GetDatum()
	switch datum.Kind() {
	case types.KindNull:
		expr, ok := opt.Expr.(*ast.FuncCallExpr)
		if ok {
			sql += expr.FnName.O
		} else {
			sql += "NULL"
		}

	case types.KindInt64:
		sql += strconv.FormatInt(datum.GetInt64(), 10)

	case types.KindString:
		sql += formatStringValue(datum.GetString())

	default:
		sql += fmt.Sprintf("%v", datum.GetValue())
	}

	return sql
}
func formatStringValue(s string) string {
	if s == "" {
		return "''"
	}
	return fmt.Sprintf("'%s'", s)
}

// for change/modify column use only.
func fieldTypeToSQL(ft *types.FieldType) string {
	strs := []string{ft.CompactStr()}
	log.Debugf("[syncer] tp %v, flag %v, flen %v, decimal %v, charset %v, collate %v, strs %v",
		ft.Tp, ft.Flag, ft.Flen, ft.Decimal, ft.Charset, ft.Collate, strs)
	if mysql.HasUnsignedFlag(ft.Flag) {
		strs = append(strs, "UNSIGNED")
	}
	if mysql.HasZerofillFlag(ft.Flag) {
		strs = append(strs, "ZEROFILL")
	}

	if mysql.HasBinaryFlag(ft.Flag) && (ft.Charset != charset.CharsetBin || (!types.IsTypeChar(ft.Tp) && !types.IsTypeBlob(ft.Tp))) {
		strs = append(strs, "BINARY")
	}

	return strings.Join(strs, " ")
}

// for add column use only. Work likes FieldType.String(), but bug free(?)
func fullFieldTypeToSQL(ft *types.FieldType) string {
	sql := fieldTypeToSQL(ft)
	strs := strings.Split(sql, " ")

	if types.IsTypeChar(ft.Tp) || types.IsTypeBlob(ft.Tp) {
		if ft.Charset != "" && ft.Charset != charset.CharsetBin {
			strs = append(strs, fmt.Sprintf("CHARACTER SET %s", ft.Charset))
		}
		if ft.Collate != "" && ft.Collate != charset.CharsetBin {
			strs = append(strs, fmt.Sprintf("COLLATE %s", ft.Collate))
		}
	}

	return strings.Join(strs, " ")
}

func columnOptionsToSQL(options []*ast.ColumnOption) string {
	sql := ""
	for _, opt := range options {
		switch opt.Tp {
		case ast.ColumnOptionNotNull:
			sql += " NOT NULL"
		case ast.ColumnOptionNull:
			sql += " NULL"
		case ast.ColumnOptionDefaultValue:
			sql += defaultValueToSQL(opt)
		case ast.ColumnOptionAutoIncrement:
			sql += " AUTO_INCREMENT"
		case ast.ColumnOptionUniqKey:
			sql += " UNIQUE KEY"
		case ast.ColumnOptionPrimaryKey:
			sql += " PRIMARY KEY"
		case ast.ColumnOptionComment:
			sql += fmt.Sprintf(" COMMENT '%v'", opt.Expr.GetValue())
		case ast.ColumnOptionOnUpdate: // For Timestamp and Datetime only.
			sql += " ON UPDATE CURRENT_TIMESTAMP"
		case ast.ColumnOptionFulltext:
			panic("[syncer]not implemented yet")
		default:
			panic("[syncer]not implemented yet")
		}
	}

	return sql
}

func escapeName(name string) string {
	return strings.Replace(name, "`", "``", -1)
}

func tableNameToSQL(tbl *ast.TableName) string {
	sql := ""
	if tbl.Schema.O != "" {
		sql += fmt.Sprintf("`%s`.", tbl.Schema.O)
	}
	sql += fmt.Sprintf("`%s`", tbl.Name.O)
	return sql
}

func columnNameToSQL(name *ast.ColumnName) string {
	sql := ""
	if name.Schema.O != "" {
		sql += fmt.Sprintf("`%s`.", escapeName(name.Schema.O))
	}
	if name.Table.O != "" {
		sql += fmt.Sprintf("`%s`.", escapeName(name.Table.O))
	}
	sql += fmt.Sprintf("`%s`", escapeName(name.Name.O))
	return sql
}

func indexColNameToSQL(name *ast.IndexColName) string {
	sql := columnNameToSQL(name.Column)
	if name.Length != types.UnspecifiedLength {
		sql += fmt.Sprintf(" (%d)", name.Length)
	}
	return sql
}

func constraintKeysToSQL(keys []*ast.IndexColName) string {
	if len(keys) == 0 {
		panic("[syncer]unreachable")
	}
	sql := ""
	for i, indexColName := range keys {
		if i == 0 {
			sql += "("
		}
		sql += indexColNameToSQL(indexColName)
		if i != len(keys)-1 {
			sql += ", "
		}
	}
	sql += ")"
	return sql
}

func referenceDefToSQL(refer *ast.ReferenceDef) string {
	sql := fmt.Sprintf("%s ", tableNameToSQL(refer.Table))
	sql += constraintKeysToSQL(refer.IndexColNames)
	if refer.OnDelete != nil && refer.OnDelete.ReferOpt != ast.ReferOptionNoOption {
		sql += fmt.Sprintf(" ON DELETE %s", refer.OnDelete.ReferOpt)
	}
	if refer.OnUpdate != nil && refer.OnUpdate.ReferOpt != ast.ReferOptionNoOption {
		sql += fmt.Sprintf(" ON UPDATE %s", refer.OnUpdate.ReferOpt)
	}
	return sql
}

func indexTypeToSQL(opt *ast.IndexOption) string {
	// opt can be nil.
	if opt == nil {
		return ""
	}
	switch opt.Tp {
	case model.IndexTypeBtree:
		return "USING BTREE "
	case model.IndexTypeHash:
		return "USING HASH "
	default:
		// nothing to do
		return ""
	}
}

func constraintToSQL(constraint *ast.Constraint) string {
	sql := ""
	switch constraint.Tp {
	case ast.ConstraintKey, ast.ConstraintIndex:
		sql += "ADD INDEX "
		if constraint.Name != "" {
			sql += fmt.Sprintf("`%s` ", escapeName(constraint.Name))
		}
		sql += indexTypeToSQL(constraint.Option)
		sql += constraintKeysToSQL(constraint.Keys)
		sql += indexOptionToSQL(constraint.Option)

	case ast.ConstraintUniq, ast.ConstraintUniqKey, ast.ConstraintUniqIndex:
		sql += "ADD CONSTRAINT "
		if constraint.Name != "" {
			sql += fmt.Sprintf("`%s` ", escapeName(constraint.Name))
		}
		sql += "UNIQUE INDEX "
		sql += indexTypeToSQL(constraint.Option)
		sql += constraintKeysToSQL(constraint.Keys)
		sql += indexOptionToSQL(constraint.Option)

	case ast.ConstraintForeignKey:
		sql += "ADD CONSTRAINT "
		if constraint.Name != "" {
			sql += fmt.Sprintf("`%s` ", escapeName(constraint.Name))
		}
		sql += "FOREIGN KEY "
		sql += constraintKeysToSQL(constraint.Keys)
		sql += " REFERENCES "
		sql += referenceDefToSQL(constraint.Refer)

	case ast.ConstraintPrimaryKey:
		sql += "ADD CONSTRAINT "
		if constraint.Name != "" {
			sql += fmt.Sprintf("`%s` ", escapeName(constraint.Name))
		}
		sql += "PRIMARY KEY "
		sql += indexTypeToSQL(constraint.Option)
		sql += constraintKeysToSQL(constraint.Keys)
		sql += indexOptionToSQL(constraint.Option)

	case ast.ConstraintFulltext:
		sql += "ADD FULLTEXT INDEX "
		if constraint.Name != "" {
			sql += fmt.Sprintf("`%s` ", escapeName(constraint.Name))
		}
		sql += constraintKeysToSQL(constraint.Keys)
		sql += indexOptionToSQL(constraint.Option)

	default:
		panic("[syncer]not implemented yet")
	}
	return sql
}

func positionToSQL(pos *ast.ColumnPosition) string {
	var sql string
	switch pos.Tp {
	case ast.ColumnPositionNone:
	case ast.ColumnPositionFirst:
		sql = " FIRST"
	case ast.ColumnPositionAfter:
		colName := pos.RelativeColumn.Name.O
		sql = fmt.Sprintf(" AFTER `%s`", escapeName(colName))
	default:
		panic("[syncer] unreachable code")
	}
	return sql
}

// Convert constraint indexoption to sql. Currently only support comment.
func indexOptionToSQL(option *ast.IndexOption) string {
	if option == nil {
		return ""
	}

	if option.Comment != "" {
		return fmt.Sprintf(" COMMENT '%s'", option.Comment)
	}

	return ""
}

func tableOptionsToSQL(options []*ast.TableOption) string {
	sql := ""
	if len(options) == 0 {
		return sql
	}

	for _, opt := range options {
		switch opt.Tp {
		case ast.TableOptionEngine:
			if opt.StrValue == "" {
				sql += fmt.Sprintf(" ENGINE = ''")
			} else {
				sql += fmt.Sprintf(" ENGINE = %s", opt.StrValue)
			}

		case ast.TableOptionCharset:
			sql += fmt.Sprintf(" CHARACTER SET = %s", opt.StrValue)

		case ast.TableOptionCollate:
			sql += fmt.Sprintf(" COLLATE = %s", opt.StrValue)

		case ast.TableOptionAutoIncrement:
			sql += fmt.Sprintf(" AUTO_INCREMENT = %d", opt.UintValue)

		case ast.TableOptionComment:
			sql += fmt.Sprintf(" COMMENT '%s'", opt.StrValue)

		case ast.TableOptionAvgRowLength:
			sql += fmt.Sprintf(" AVG_ROW_LENGTH = %d", opt.UintValue)

		case ast.TableOptionCheckSum:
			sql += fmt.Sprintf(" CHECKSUM = %d", opt.UintValue)

		case ast.TableOptionCompression:
			// In TiDB parser.y, the rule is "COMPRESSION" EqOpt Identifier. No single quote here.
			sql += fmt.Sprintf(" COMPRESSION = '%s'", opt.StrValue)

		case ast.TableOptionConnection:
			sql += fmt.Sprintf(" CONNECTION = '%s'", opt.StrValue)

		case ast.TableOptionPassword:
			sql += fmt.Sprintf(" PASSWORD = '%s'", opt.StrValue)

		case ast.TableOptionKeyBlockSize:
			sql += fmt.Sprintf(" KEY_BLOCK_SIZE = %d", opt.UintValue)

		case ast.TableOptionMaxRows:
			sql += fmt.Sprintf(" MAX_ROWS = %d", opt.UintValue)

		case ast.TableOptionMinRows:
			sql += fmt.Sprintf(" MIN_ROWS = %d", opt.UintValue)

		case ast.TableOptionDelayKeyWrite:
			sql += fmt.Sprintf(" DELAY_KEY_WRITE = %d", opt.UintValue)

		case ast.TableOptionRowFormat:
			sql += fmt.Sprintf(" ROW_FORMAT = %s", formatRowFormat(opt.UintValue))

		case ast.TableOptionStatsPersistent:
			// Since TiDB doesn't support this feature, we just give a default value.
			sql += " STATS_PERSISTENT = DEFAULT"
		default:
			panic("[syncer]unreachable")
		}

	}

	// trim prefix space
	sql = strings.TrimPrefix(sql, " ")
	return sql
}

func formatRowFormat(rf uint64) string {
	var s string
	switch rf {
	case ast.RowFormatDefault:
		s = "DEFAULT"
	case ast.RowFormatDynamic:
		s = "DYNAMIC"
	case ast.RowFormatFixed:
		s = "FIXED"
	case ast.RowFormatCompressed:
		s = "COMPRESSED"
	case ast.RowFormatRedundant:
		s = "REDUNDANT"
	case ast.RowFormatCompact:
		s = "COMPACT"
	default:
		panic("[syncer]unreachable")
	}
	return s
}

func columnToSQL(typeDef string, newColumn *ast.ColumnDef) string {
	return fmt.Sprintf("%s %s%s", columnNameToSQL(newColumn.Name), typeDef, columnOptionsToSQL(newColumn.Options))
}

func alterTableSpecToSQL(spec *ast.AlterTableSpec, newTable *ast.TableName, ctxSchema string) []string {
	var (
		suffixes []string
		suffix   string
	)

	switch spec.Tp {
	case ast.AlterTableOption:
		suffix += tableOptionsToSQL(spec.Options)
		suffixes = append(suffixes, suffix)

	case ast.AlterTableAddColumns:
		for _, newColumn := range spec.NewColumns {
			suffix = ""
			typeDef := fullFieldTypeToSQL(newColumn.Tp)
			suffix += fmt.Sprintf("ADD COLUMN %s", columnToSQL(typeDef, newColumn))
			if spec.Position != nil {
				suffix += positionToSQL(spec.Position)
			}
			suffixes = append(suffixes, suffix)
		}

	case ast.AlterTableDropColumn:
		suffix += fmt.Sprintf("DROP COLUMN %s", columnNameToSQL(spec.OldColumnName))
		suffixes = append(suffixes, suffix)

	case ast.AlterTableDropIndex:
		suffix += fmt.Sprintf("DROP INDEX `%s`", escapeName(spec.Name))
		suffixes = append(suffixes, suffix)

	case ast.AlterTableAddConstraint:
		suffix += constraintToSQL(spec.Constraint)
		suffixes = append(suffixes, suffix)

	case ast.AlterTableDropForeignKey:
		suffix += fmt.Sprintf("DROP FOREIGN KEY `%s`", escapeName(spec.Name))
		suffixes = append(suffixes, suffix)

	case ast.AlterTableModifyColumn:
		// TiDB doesn't support alter table modify column charset and collation.
		typeDef := fieldTypeToSQL(spec.NewColumns[0].Tp)
		suffix += fmt.Sprintf("MODIFY COLUMN %s", columnToSQL(typeDef, spec.NewColumns[0]))
		if spec.Position != nil {
			suffix += positionToSQL(spec.Position)
		}
		suffixes = append(suffixes, suffix)

	// FIXME: should support [FIRST|AFTER col_name], but tidb parser not support this currently.
	case ast.AlterTableChangeColumn:
		typeDef := fieldTypeToSQL(spec.NewColumns[0].Tp)
		suffix += "CHANGE COLUMN "
		suffix += fmt.Sprintf("%s %s",
			columnNameToSQL(spec.OldColumnName),
			columnToSQL(typeDef, spec.NewColumns[0]))
		if spec.Position != nil {
			suffix += positionToSQL(spec.Position)
		}
		suffixes = append(suffixes, suffix)

	case ast.AlterTableRenameTable:
		setSchemaIfNotExists(spec.NewTable, ctxSchema)
		*newTable = *spec.NewTable
		suffix += fmt.Sprintf("RENAME TO %s", tableNameToSQL(spec.NewTable))
		suffixes = append(suffixes, suffix)

	case ast.AlterTableAlterColumn:
		suffix += fmt.Sprintf("ALTER COLUMN %s ", columnNameToSQL(spec.NewColumns[0].Name))
		if options := spec.NewColumns[0].Options; options != nil {
			suffix += fmt.Sprintf("SET DEFAULT %v", options[0].Expr.GetValue())
		} else {
			suffix += "DROP DEFAULT"
		}
		suffixes = append(suffixes, suffix)

	case ast.AlterTableDropPrimaryKey:
		suffix += "DROP PRIMARY KEY"
		suffixes = append(suffixes, suffix)

	case ast.AlterTableLock:
		// just ignore it
	default:
	}
	return suffixes
}

func alterTableStmtToSQL(table *ast.TableName, newTable *ast.TableName, spec *ast.AlterTableSpec, ctxSchema string) []string {
	var (
		sqls   []string
		prefix string
	)
	if newTable.Name.O != "" {
		prefix = fmt.Sprintf("ALTER TABLE %s ", tableNameToSQL(newTable))
	} else {
		prefix = fmt.Sprintf("ALTER TABLE %s ", tableNameToSQL(table))
	}

	log.Infof("spec %+v", spec)
	for _, suffix := range alterTableSpecToSQL(spec, newTable, ctxSchema) {
		sqls = append(sqls, prefix+suffix)
	}

	log.Debugf("alter table stmt to sql:%v", sqls)
	return sqls
}

// if not exists by default for fanout use.
func createDatabaseStmtToSQL(stmt *ast.CreateDatabaseStmt) string {
	options := databaseOptionsToSQL(stmt.Options)
	return fmt.Sprintf("CREATE DATABASE IF NOT EXISTS `%s`%s", stmt.Name, options)
}

func databaseOptionsToSQL(options []*ast.DatabaseOption) string {
	var result string
	if len(options) == 0 {
		return ""
	}
	if len(options) == 1 && options[0].Tp == ast.DatabaseOptionNone {
		// unreachable code. DatabaseOptionNone is never used in TiDB.
		return ""
	}

	//FIXME: need sort ?
	for _, option := range options {
		if option.Tp == ast.DatabaseOptionCharset {
			result += fmt.Sprintf(" CHARACTER SET %s", option.Value)
		}
		if option.Tp == ast.DatabaseOptionCollate {
			result += fmt.Sprintf(" COLLATE %s", option.Value)
		}
	}
	return result
}

// if exists by default. for fanout use.
func dropDatabaseStmtToSQL(stmt *ast.DropDatabaseStmt) string {
	return fmt.Sprintf("DROP DATABASE IF EXISTS `%s`", stmt.Name)
}

func dropTableStmtToSQL(table *ast.TableName, ifExists bool) string {
	var ex string
	if ifExists {
		ex = "IF EXISTS "
	}
	return fmt.Sprintf("DROP TABLE %s%s", ex, tableNameToSQL(table))
}

func renameTableStmtToSQL(t2t *ast.TableToTable) string {
	return fmt.Sprintf("RENAME TABLE %s TO %s", tableNameToSQL(t2t.OldTable), tableNameToSQL(t2t.NewTable))
}

func truncateTableStmtToSQL(stmt *ast.TruncateTableStmt) string {
	return fmt.Sprintf("TRUNCATE TABLE %s", tableNameToSQL(stmt.Table))
}

// func createIndexStmtToSQL(stmt *ast.CreateIndexStmt) string {
// TODO
// }

func dropIndexStmtToSQL(stmt *ast.DropIndexStmt) string {
	return fmt.Sprintf("DROP INDEX `%s` ON %s", escapeName(stmt.IndexName), tableNameToSQL(stmt.Table))
}