package ddl

import (
	"encoding/json"

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

	"github.com/pingcap/wormhole/pkg/whpb"
)

func (d *ddl) CreateTable(s *ast.CreateTableStmt) (err error) {
	if s.ReferTable == nil {
		err = d.createTable(s)
	} else {
		err = d.createTableWithLike(s)
	}
	return errors.Trace(err)
}

func (d *ddl) createTable(s *ast.CreateTableStmt) (err error) {
	d.dbs.mu.Lock()
	defer d.dbs.mu.Unlock()

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

	_, ok = schema[s.Table.Name.O]
	if ok {
		if s.IfNotExists {
			return nil
		}
		log.Warnf("[ddl] table %s already exists", s.Table.Name.O)
		return nil
	}

	// validation.
	if err = checkTooLongTable(s.Table.Name); err != nil {
		return errors.Trace(err)
	}
	if err = checkDuplicateColumn(s.Cols); err != nil {
		return errors.Trace(err)
	}
	if err = checkGeneratedColumn(s.Cols); err != nil {
		return errors.Trace(err)
	}
	if err = checkTooLongColumn(s.Cols); err != nil {
		return errors.Trace(err)
	}

	cols := buildColumns(s.Cols)

	tbInfo, err := d.buildTableInfo(s.Table.Name, cols)
	if err != nil {
		return errors.Trace(err)
	}

	data, err := json.Marshal(tbInfo)
	if err != nil {
		return errors.Trace(err)
	}
	detail := &whpb.SchemaDetail{
		Schema: s.Table.Schema.String(),
		Table:  s.Table.Name.String(),
		Detail: data,
	}
	err = d.etcdClient.SaveSchemaDetail(d.dbinfo.GetHost(), d.dbinfo.GetPort(), s.Table.Schema.String(), s.Table.Name.String(), detail)
	if err != nil {
		return errors.Trace(err)
	}
	schema[s.Table.Name.O] = tbInfo

	return nil
}

func (d *ddl) createTableWithLike(s *ast.CreateTableStmt) (err error) {
	d.dbs.mu.Lock()
	defer d.dbs.mu.Unlock()

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

	_, ok = schema[s.Table.Name.O]
	if ok {
		if s.IfNotExists {
			return nil
		}
		log.Warnf("[ddl] table %s already exists", s.Table.Name.O)
		return nil
	}

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

	refTable, ok := refSchema[s.ReferTable.Name.O]
	if !ok {
		log.Errorf("[ddl] refertable %s not exists", s.ReferTable.Name.O)
		return errors.Trace(ErrTableNotExists)
	}

	data, err := json.Marshal(refTable)
	if err != nil {
		return errors.Trace(err)
	}
	detail := &whpb.SchemaDetail{
		Schema: s.Table.Schema.String(),
		Table:  s.Table.Name.String(),
		Detail: data,
	}
	err = d.etcdClient.SaveSchemaDetail(d.dbinfo.GetHost(), d.dbinfo.GetPort(), s.Table.Schema.String(), s.Table.Name.String(), detail)
	if err != nil {
		return errors.Trace(err)
	}
	schema[s.Table.Name.O] = refTable
	return nil
}

func (d *ddl) DropTable(s *ast.DropTableStmt) (err error) {
	for _, table := range s.Tables {
		if table.Schema.String() == "" {
			return errors.Errorf("schema is empty %+v", s)
		}
		err = d.dropTable(table, s.IfExists)
		if err != nil {
			return errors.Trace(err)
		}
	}

	return nil
}

func (d *ddl) dropTable(table *ast.TableName, ifExists bool) error {
	d.dbs.mu.Lock()
	defer d.dbs.mu.Unlock()

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

	_, ok = schema[table.Name.O]
	if !ok {
		if ifExists {
			return nil
		}
		log.Errorf("[ddl] table %s not exists", table.Name.O)
		return errors.Trace(ErrTableNotExists)
	}
	err := d.etcdClient.DeleteSchemaDetail(d.dbinfo.GetHost(), d.dbinfo.GetPort(), table.Schema.String(), table.Name.String())
	if err != nil {
		return errors.Trace(err)
	}
	delete(schema, table.Name.O)
	return nil
}

func (d *ddl) RenameTable(oldIdent, newIdent ast.Ident) (err error) {
	d.dbs.mu.Lock()
	defer d.dbs.mu.Unlock()

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

	oldTable, ok := oldSchema[oldIdent.Name.O]
	if !ok {
		log.Errorf("[ddl] old table %s not exists", oldIdent.Name.O)
		return errors.Trace(ErrTableNotExists)
	}

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

	_, ok = d.dbs.data[newIdent.Name.O]
	if ok {
		log.Errorf("[ddl] new table %s already exists", newIdent.Name.O)
		return errors.Trace(ErrTableAlreadyExists)
	}

	data, err := json.Marshal(oldTable)
	if err != nil {
		return errors.Trace(err)
	}
	detail := &whpb.SchemaDetail{
		Schema: newIdent.Schema.String(),
		Table:  newIdent.Name.String(),
		Detail: data,
	}

	err = d.etcdClient.RenameTable(
		d.dbinfo.GetHost(),
		d.dbinfo.GetPort(),
		oldIdent.Schema.String(),
		oldIdent.Name.String(),
		newIdent.Schema.String(),
		newIdent.Name.String(),
		detail,
	)
	if err != nil {
		return errors.Trace(err)
	}

	// oldSchema and newSchema may be same one, never mind.
	delete(oldSchema, oldIdent.Name.O)
	oldTable.Name = newIdent.Name
	newSchema[newIdent.Name.O] = oldTable

	return nil
}

func (d *ddl) TruncateTable(s *ast.TruncateTableStmt) (err error) {
	// Nothing to do. Truncate table just delete all the data, and will not change the table schema.
	return nil
}

func (d *ddl) AlterTable(s *ast.AlterTableStmt) (err error) {
	if len(s.Specs) != 1 {
		log.Warnf("[ddl] alter table specs length:%d", len(s.Specs))
		return nil
	}

	ti := ast.Ident{Schema: s.Table.Schema, Name: s.Table.Name}
	err = d.alterTableSpec(ti, s.Specs[0])
	if err != nil {
		return errors.Trace(err)
	}

	// FIXME: how to save schemadetail to ether then save to memory here?
	return errors.Trace(d.saveTable(ti))
}

func (d *ddl) saveTable(ti ast.Ident) error {
	host := d.dbinfo.GetHost()
	port := d.dbinfo.GetPort()

	d.dbs.mu.RLock()
	data, err := json.Marshal(d.dbs.data)
	if err != nil {
		d.dbs.mu.RUnlock()
		return errors.Trace(err)
	}
	d.dbs.mu.RUnlock()

	detail := &whpb.SchemaDetail{
		Schema: ti.Schema.String(),
		Table:  ti.Name.String(),
		Detail: data,
	}
	err = d.etcdClient.SaveSchemaDetail(host, port, ti.Schema.String(), ti.Name.String(), detail)
	if err != nil {
		return errors.Trace(err)
	}
	return nil
}

func (d *ddl) alterTableSpec(ti ast.Ident, spec *ast.AlterTableSpec) (err error) {
	switch spec.Tp {
	case ast.AlterTableOption:
		// nothing to do now.
	case ast.AlterTableAddColumns:
		err = d.AddColumn(ti, spec)
	case ast.AlterTableAddConstraint:
		// nothing to do now.
	case ast.AlterTableDropColumn:
		err = d.DropColumn(ti, spec.OldColumnName.Name)
	case ast.AlterTableDropPrimaryKey:
		// nothing to do now.
	case ast.AlterTableDropIndex:
		// nothing to do now.
	case ast.AlterTableDropForeignKey:
		// nothing to do now.
	case ast.AlterTableModifyColumn:
		err = d.ModifyColumn(ti, spec)
	case ast.AlterTableChangeColumn:
		err = d.ChangeColumn(ti, spec)
	case ast.AlterTableAlterColumn:
		err = d.AlterColumn(ti, spec)
	case ast.AlterTableRenameTable:
		newIdent := ast.Ident{Schema: spec.NewTable.Schema, Name: spec.NewTable.Name}
		err = d.RenameTable(ti, newIdent)
	case ast.AlterTableLock:
		// do nothing
	}

	if err != nil {
		return errors.Trace(err)
	}
	return nil
}

func (d *ddl) buildTableInfo(tableName model.CIStr, cols []*model.ColumnInfo) (tbInfo *model.TableInfo, err error) {
	tbInfo = &model.TableInfo{
		Name: tableName,
	}
	for _, v := range cols {
		v.ID = allocateColumnID(tbInfo)
		tbInfo.Columns = append(tbInfo.Columns, v)
	}

	return
}