package ddl

import (
	"strings"

	"github.com/juju/errors"
	"github.com/ngaut/log"
	"github.com/pingcap/tidb/ast"
	"github.com/pingcap/tidb/infoschema"
	"github.com/pingcap/tidb/model"
	"github.com/pingcap/tidb/mysql"
)

func (d *ddl) AddColumn(ti ast.Ident, spec *ast.AlterTableSpec) (err error) {
	specNewColumn := spec.NewColumns[0]

	d.dbs.mu.Lock()
	defer d.dbs.mu.Unlock()

	schema, ok := d.dbs.data[ti.Schema.O]
	if !ok {
		log.Warnf("[ddl] schema %s not exists", ti.Schema.O)
		return NewDBNotExistsError(ti.Schema.O)
	}

	table, ok := schema[ti.Name.O]
	if !ok {
		log.Errorf("[ddl] table %s not exists", ti.Name.O)
		return ErrTableNotExists
	}
	colName := specNewColumn.Name.Name.O
	col := findCol(table.Columns, colName)
	if col != nil {
		return ErrColumnExists
	}

	//  build column
	col = buildColumn(len(table.Columns), specNewColumn)

	_, offset, err := d.createColumnInfo(table, col, spec.Position)
	if err != nil {
		return errors.Trace(err)
	}

	// Adjust column offset.
	d.adjustColumnOffset(table.Columns, table.Indices, offset, true)

	return nil
}

func (d *ddl) createColumnInfo(tblInfo *model.TableInfo, colInfo *model.ColumnInfo, pos *ast.ColumnPosition) (*model.ColumnInfo, int, error) {
	// Check column name duplicate.
	cols := tblInfo.Columns
	position := len(cols)

	// Get column position.
	if pos.Tp == ast.ColumnPositionFirst {
		position = 0
	} else if pos.Tp == ast.ColumnPositionAfter {
		c := findCol(cols, pos.RelativeColumn.Name.O)
		if c == nil {
			return nil, 0, infoschema.ErrColumnNotExists.GenByArgs(pos.RelativeColumn, tblInfo.Name)
		}

		// Insert position is after the mentioned column.
		position = c.Offset + 1
	}
	colInfo.ID = allocateColumnID(tblInfo)
	// To support add column asynchronous, we should mark its offset as the last column.
	// So that we can use origin column offset to get value from row.
	colInfo.Offset = len(cols)

	// Insert col into the right place of the column list.
	newCols := make([]*model.ColumnInfo, 0, len(cols)+1)
	newCols = append(newCols, cols[:position]...)
	newCols = append(newCols, colInfo)
	newCols = append(newCols, cols[position:]...)

	tblInfo.Columns = newCols
	return colInfo, position, nil
}

// findCol finds column in cols by name.
func findCol(cols []*model.ColumnInfo, name string) *model.ColumnInfo {
	// column name don't distinguish capital and small letter
	name = strings.ToLower(name)
	for _, col := range cols {
		if col.Name.L == name {
			return col
		}
	}

	return nil
}

func (d *ddl) adjustColumnOffset(columns []*model.ColumnInfo, indices []*model.IndexInfo, offset int, added bool) {
	offsetChanged := make(map[int]int)
	if added {
		for i := offset + 1; i < len(columns); i++ {
			offsetChanged[columns[i].Offset] = i
			columns[i].Offset = i
		}
		columns[offset].Offset = offset
	} else {
		for i := offset + 1; i < len(columns); i++ {
			offsetChanged[columns[i].Offset] = i - 1
			columns[i].Offset = i - 1
		}
		columns[offset].Offset = len(columns) - 1
	}

	// TODO: Index can't cover the add/remove column with offset now, we may check this later.

	// Update index column offset info.
	for _, idx := range indices {
		for _, col := range idx.Columns {
			newOffset, ok := offsetChanged[col.Offset]
			if ok {
				col.Offset = newOffset
			}
		}
	}
}

func (d *ddl) DropColumn(ti ast.Ident, colName model.CIStr) (err error) {
	d.dbs.mu.Lock()
	defer d.dbs.mu.Unlock()

	schema, ok := d.dbs.data[ti.Schema.O]
	if !ok {
		log.Warnf("[ddl] schema %s not exists", ti.Schema.O)
		return NewDBNotExistsError(ti.Schema.O)
	}

	table, ok := schema[ti.Name.O]
	if !ok {
		log.Errorf("[ddl] table %s not exists", ti.Name.O)
		return ErrTableNotExists
	}

	col := findCol(table.Columns, colName.O)
	if col == nil {
		log.Errorf("[ddl] column %s not exists", colName.O)
		return ErrColumnNoExists
	}

	// TODO: check dependences of generated column
	// Check whether there are other columns depend on this column or not.
	// for _, col := range t.Cols() {
	// 	for dep := range col.Dependences {
	// 		if dep == colName.L {
	// 			return errDependentByGeneratedColumn.GenByArgs(dep)
	// 		}
	// 	}
	// }

	// We don't support dropping column with index covered now.
	// We must drop the index first, then drop the column.
	if isColumnWithIndex(colName.L, table.Indices) {
		return ErrCantDropColWithIndex
	}

	// We don't support dropping column with PK handle covered now.
	if isPKHandleColumn(col, table) {
		return ErrUnsupportedPKHandle
	}

	// Set this column's offset to the last and reset all following columns' offsets.
	d.adjustColumnOffset(table.Columns, table.Indices, col.Offset, false)

	// All reorganization jobs are done, drop this column.
	newColumns := make([]*model.ColumnInfo, 0, len(table.Columns))
	for _, col := range table.Columns {
		if col.Name.L != colName.L {
			newColumns = append(newColumns, col)
		}
	}
	table.Columns = newColumns

	return nil
}

func (d *ddl) ModifyColumn(ti ast.Ident, spec *ast.AlterTableSpec) (err error) {
	specNewColumn := spec.NewColumns[0]
	if specNewColumn.Name.Schema.O != "" && ti.Schema.O != specNewColumn.Name.Schema.O {
		return ErrWrongDBName
	}
	if specNewColumn.Name.Table.O != "" && ti.Name.O != specNewColumn.Name.Table.O {
		return ErrWrongTableName
	}

	oldColName := specNewColumn.Name.Name
	err = d.modifyColumn(ti, oldColName, spec)
	if err != nil {
		return errors.Trace(err)
	}
	return nil
}

func (d *ddl) ChangeColumn(ti ast.Ident, spec *ast.AlterTableSpec) (err error) {
	specNewColumn := spec.NewColumns[0]
	if specNewColumn.Name.Schema.O != "" && ti.Schema.O != specNewColumn.Name.Schema.O {
		return ErrWrongDBName
	}
	if spec.OldColumnName.Schema.O != "" && ti.Schema.O != spec.OldColumnName.Schema.O {
		return ErrWrongDBName
	}
	if specNewColumn.Name.Table.O != "" && ti.Name.O != specNewColumn.Name.Table.O {
		return ErrWrongTableName
	}
	if spec.OldColumnName.Table.O != "" && ti.Name.O != spec.OldColumnName.Table.O {
		return ErrWrongTableName
	}

	err = d.modifyColumn(ti, spec.OldColumnName.Name, spec)
	if err != nil {
		return errors.Trace(err)
	}

	return nil
}

func (d *ddl) modifyColumn(ti ast.Ident, oldColName model.CIStr, spec *ast.AlterTableSpec) (err error) {
	specNewColumn := spec.NewColumns[0]

	d.dbs.mu.Lock()
	defer d.dbs.mu.Unlock()

	schema, ok := d.dbs.data[ti.Schema.O]
	if !ok {
		log.Warnf("[ddl] schema %s not exists", ti.Schema.O)
		return NewDBNotExistsError(ti.Schema.O)
	}

	table, ok := schema[ti.Name.O]
	if !ok {
		log.Errorf("[ddl] table %s not exists", ti.Name.O)
		return ErrTableNotExists
	}

	oldCol := findCol(table.Columns, oldColName.O)
	if oldCol == nil {
		return ErrColumnNoExists
	}

	if specNewColumn.Tp == nil {
		return ErrUnsupportedModifyColumn
	}

	newCol := &model.ColumnInfo{
		Offset:             oldCol.Offset,
		OriginDefaultValue: oldCol.OriginDefaultValue,
		FieldType:          *specNewColumn.Tp,
		Name:               specNewColumn.Name.Name,
	}

	//check modifiable
	err = modifiable(&oldCol.FieldType, &newCol.FieldType)
	if err != nil {
		return errors.Trace(err)
	}

	// Copy index related options to the new spec.
	indexFlags := oldCol.FieldType.Flag & (mysql.PriKeyFlag | mysql.UniqueKeyFlag | mysql.MultipleKeyFlag)
	newCol.FieldType.Flag |= indexFlags
	if mysql.HasPriKeyFlag(oldCol.FieldType.Flag) {
		newCol.FieldType.Flag |= mysql.NotNullFlag
		// TODO: If user explicitly set NULL, we should throw error ErrPrimaryCantHaveNull.
	}

	// We don't support modifying column from not_auto_increment to auto_increment.
	if !mysql.HasAutoIncrementFlag(oldCol.Flag) && mysql.HasAutoIncrementFlag(newCol.Flag) {
		return ErrUnsupportedModifyColumn
	}

	// We don't support modifying the type definitions from 'null' to 'not null' now.
	if !mysql.HasNotNullFlag(oldCol.Flag) && mysql.HasNotNullFlag(newCol.Flag) {
		return ErrUnsupportedModifyColumn
	}
	// As same with MySQL, we don't support modifying the stored status for generated columns.
	if err = checkModifyGeneratedColumn(table.Columns, oldCol, newCol); err != nil {
		return errors.Trace(err)
	}

	pos := spec.Position
	// Calculate column's new position.
	oldPos, newPos := oldCol.Offset, oldCol.Offset
	if pos.Tp == ast.ColumnPositionAfter {
		if oldColName.L == pos.RelativeColumn.Name.L {
			// `alter table tableName modify column b int after b` will return ver,ErrColumnNotExists.
			return infoschema.ErrColumnNotExists.GenByArgs(oldColName, table.Name)
		}

		relative := findCol(table.Columns, pos.RelativeColumn.Name.L)
		if relative == nil {
			return infoschema.ErrColumnNotExists.GenByArgs(pos.RelativeColumn, table.Name)
		}

		if relative.Offset < oldPos {
			newPos = relative.Offset + 1
		} else {
			newPos = relative.Offset
		}
	} else if pos.Tp == ast.ColumnPositionFirst {
		newPos = 0
	}

	columnChanged := make(map[string]*model.ColumnInfo)
	columnChanged[oldColName.L] = newCol

	if newPos == oldPos {
		table.Columns[newPos] = newCol
	} else {
		cols := table.Columns

		// Reorder columns in place.
		if newPos < oldPos {
			copy(cols[newPos+1:], cols[newPos:oldPos])
		} else {
			copy(cols[oldPos:], cols[oldPos+1:newPos+1])
		}
		cols[newPos] = newCol

		for i, newCol := range table.Columns {
			if newCol.Offset != i {
				columnChanged[newCol.Name.L] = newCol
				newCol.Offset = i
			}
		}
	}

	// Change offset and name in indices.
	for _, idx := range table.Indices {
		for _, c := range idx.Columns {
			if newCol, ok := columnChanged[c.Name.L]; ok {
				c.Name = newCol.Name
				c.Offset = newCol.Offset
			}
		}
	}

	return nil
}

func (d *ddl) AlterColumn(ti ast.Ident, spec *ast.AlterTableSpec) (err error) {
	// alter column just set default value.
	// so, we have nothing to do now.
	return nil
}