package ddl

import (
	"encoding/json"
	"fmt"
	"sync"

	"github.com/juju/errors"
	"github.com/pingcap/tidb/ast"
	"github.com/pingcap/tidb/model"
	"github.com/pingcap/tidb/mysql"
	"github.com/pingcap/tidb/terror"

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

const (
	codeBadField                terror.ErrCode = 1054
	codeGeneratedColumnNonPrior                = 3107
)

var (
	// errGeneratedColumnNonPrior forbiddens to refer generated column non prior to it.
	errGeneratedColumnNonPrior = terror.ClassDDL.New(codeGeneratedColumnNonPrior, mysql.MySQLErrName[mysql.ErrGeneratedColumnNonPrior])
	errBadField                = terror.ClassDDL.New(codeBadField, "Unknown column '%s' in '%s'")
)

// DBNotExistsError defines error of db not exists.
type DBNotExistsError struct {
	db string
}

// NewDBNotExistsError creates a DBNotExistsError.
func NewDBNotExistsError(db string) *DBNotExistsError {
	return &DBNotExistsError{db}
}

func (e *DBNotExistsError) Error() string {
	return fmt.Sprintf("db %s not exists", e.db)
}

// Name returns the name of the database.
func (e *DBNotExistsError) Name() string {
	return e.db
}

var (
	// ErrTableNotExists is the err of table not exits
	ErrTableNotExists = errors.New("table not exits")
	// ErrDBAlreadyExists is the err of db already exists
	ErrDBAlreadyExists = errors.New("db already exists")
	// ErrTableAlreadyExists is the err of table already exists
	ErrTableAlreadyExists = errors.New("table already exists")
	// ErrColumnExists is the err of column already exists
	ErrColumnExists = errors.New("column already exists")
	// ErrColumnNoExists is the err of column not exists
	ErrColumnNoExists = errors.New("column not exists")
	// ErrCantDropColWithIndex is the err of can't drop column with index
	ErrCantDropColWithIndex = errors.New("can't drop column with index")
	// ErrUnsupportedPKHandle is the err of unsupported drop integer primary key
	ErrUnsupportedPKHandle = errors.New("unsupported drop integer primary key")
	// ErrUnsupportedModifyColumn is the err of unsupported modify column
	ErrUnsupportedModifyColumn = errors.New("unsupported modify column")
	// ErrUnsupportedOnGeneratedColumn is the err of unsupported on generated column
	ErrUnsupportedOnGeneratedColumn = errors.New("unsupported on generated column")
	// ErrWrongDBName is the err of wrong db name
	ErrWrongDBName = errors.New("wrong db name")
	// ErrWrongTableName is the err of wrong table name
	ErrWrongTableName = errors.New("wrong table name")
	// ErrNotInitialized is the err of need to initialized from data source
	ErrNotInitialized = errors.New("need to initialized from data source")
)

// DDL is an interface for transaction and executive args environment
type DDL interface {
	Init() error
	Execute(stmt ast.StmtNode) error
	Snapshot(schema, table string) (*model.TableInfo, error)
	SetTable(schemaName string, table *model.TableInfo)
	SetSchema(schemaName string)
	GetSchemaSummary() map[string][]string
	GetTables(schema string) []string
}

// NewDDL creates a new ddl.
func NewDDL(etcdClient *etcdutil.Client, dbinfo whpb.DBAccount) DDL {
	d := &ddl{
		quitCh:     make(chan struct{}),
		etcdClient: etcdClient,
		dbinfo:     dbinfo,
	}
	d.dbs.data = make(map[string]map[string]*model.TableInfo)
	return d
}

type ddl struct {
	dbs struct {
		data map[string]map[string]*model.TableInfo
		mu   sync.RWMutex
	}
	quitCh     chan struct{}
	etcdClient *etcdutil.Client
	dbinfo     whpb.DBAccount
}

// Init initialize data from ether-client.
func (d *ddl) Init() (err error) {
	err = d.initFromEther()
	if err != nil {
		return errors.Trace(err)
	}

	d.dbs.mu.RLock()
	defer d.dbs.mu.RUnlock()
	if len(d.dbs.data) == 0 {
		return errors.Trace(ErrNotInitialized)
	}
	return nil
}

func (d *ddl) initFromEther() (err error) {
	host := d.dbinfo.GetHost()
	port := d.dbinfo.GetPort()

	var details map[string]map[string]*whpb.SchemaDetail
	err = d.etcdClient.LoadSchemaDetails(host, port, details)
	if err != nil {
		return errors.Trace(err)
	}

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

	for schema, tables := range details {
		_, ok := d.dbs.data[schema]
		if !ok {
			d.dbs.data[schema] = make(map[string]*model.TableInfo)
		}
		for table, detail := range tables {
			tableInfo := &model.TableInfo{}
			err = json.Unmarshal(detail.Detail, tableInfo)
			if err != nil {
				return errors.Trace(err)
			}
			d.dbs.data[schema][table] = tableInfo
		}
	}
	return nil
}

// Execute applies ddl changes.
func (d *ddl) Execute(stmt ast.StmtNode) (err error) {
	switch s := stmt.(type) {
	case *ast.CreateDatabaseStmt:
		err = d.CreateSchema(s) //todo
	case *ast.DropDatabaseStmt:
		err = d.DropSchema(s)
	case *ast.CreateTableStmt:
		err = d.CreateTable(s)
	case *ast.DropTableStmt:
		err = d.DropTable(s)
	case *ast.RenameTableStmt:
		oldIdent := ast.Ident{Schema: s.OldTable.Schema, Name: s.OldTable.Name}
		newIdent := ast.Ident{Schema: s.NewTable.Schema, Name: s.NewTable.Name}
		err = d.RenameTable(oldIdent, newIdent)
	case *ast.AlterTableStmt:
		err = d.AlterTable(s)
	case *ast.TruncateTableStmt:
		err = d.TruncateTable(s)
	case *ast.CreateIndexStmt:
		err = d.CreateIndex(s)
	case *ast.DropIndexStmt:
		err = d.DropIndex(s)
	}
	if err != nil {
		return errors.Trace(err)
	}

	return nil
}

// Snapshot returns the snapshot of specific table.
func (d *ddl) Snapshot(schema, table string) (*model.TableInfo, error) {
	return d.snapshot(schema, table)
}

func (d *ddl) snapshot(schema, table string) (*model.TableInfo, error) {
	d.dbs.mu.RLock()
	defer d.dbs.mu.RUnlock()

	db, ok := d.dbs.data[schema]
	if !ok {
		return nil, NewDBNotExistsError(schema)
	}

	tableInfo, ok := db[table]
	if !ok {
		return nil, errors.Annotatef(ErrTableNotExists, "table %s", table)
	}

	return tableInfo, nil

}

func (d *ddl) SetTable(schemaName string, table *model.TableInfo) {
	d.dbs.mu.Lock()
	_, ok := d.dbs.data[schemaName]
	if !ok {
		d.dbs.data[schemaName] = make(map[string]*model.TableInfo)
	}
	d.dbs.data[schemaName][table.Name.O] = table

	d.dbs.mu.Unlock()
}

func (d *ddl) SetSchema(schemaName string) {
	d.dbs.mu.Lock()
	defer d.dbs.mu.Unlock()
	_, ok := d.dbs.data[schemaName]
	if ok {
		return
	}
	d.dbs.data[schemaName] = make(map[string]*model.TableInfo)
}

func (d *ddl) GetSchemaSummary() map[string][]string {
	d.dbs.mu.RLock()
	defer d.dbs.mu.RUnlock()

	summary := make(map[string][]string)
	for schema, tables := range d.dbs.data {
		summary[schema] = make([]string, 0, len(tables))
		for table := range tables {
			summary[schema] = append(summary[schema], table)
		}
	}
	return summary
}

func (d *ddl) GetTables(schema string) []string {
	d.dbs.mu.RLock()
	defer d.dbs.mu.RUnlock()

	tables, ok := d.dbs.data[schema]
	if !ok {
		return nil
	}
	tableNames := make([]string, 0, 0)
	for name := range tables {
		tableNames = append(tableNames, name)
	}
	return tableNames
}