// IsErrNotFound checks if err is a kind of NotFound error.
func IsErrNotFound(err error) bool {
	if errors2.ErrorEqual(err, leveldb.ErrNotFound) || errors2.ErrorEqual(err, ErrNotExist) {
		return true
	}

	return false
}

func mayExit(err error, line string) {
	if errors2.ErrorEqual(err, liner.ErrPromptAborted) || errors2.ErrorEqual(err, io.EOF) {
		fmt.Println("\nBye")
		saveHistory()
		os.Exit(0)
	}
	if err != nil {
		log.Fatal(errors.ErrorStack(err))
	}
}

func mayExit(err error, l string) bool {
	if errors2.ErrorEqual(err, liner.ErrPromptAborted) || errors2.ErrorEqual(err, io.EOF) {
		fmt.Println("\nBye")
		saveHistory()
		return true
	}
	if err != nil {
		log.Fatal(errors.ErrorStack(err))
	}
	return false
}

// IsRetryableError check if the err is not a fatal error and the under going operation is worth to retried.
func IsRetryableError(err error) bool {
	if err == nil {
		return false
	}

	if errors2.ErrorEqual(err, ErrLockConflict) || errors2.ErrorEqual(err, ErrConditionNotMatch) {
		return true
	}

	return false
}

// Get returns the value associated with key.
func (m *memDbBuffer) Get(k Key) ([]byte, error) {
	v, err := m.db.Get(k)
	if errors2.ErrorEqual(err, leveldb.ErrNotFound) {
		return nil, ErrNotExist
	}
	return v, nil
}

func (cc *clientConn) Run() {
	defer func() {
		r := recover()
		if r != nil {
			const size = 4096
			buf := make([]byte, size)
			buf = buf[:runtime.Stack(buf, false)]
			log.Errorf("lastCmd %s, %v, %s", cc.lastCmd, r, buf)
		}
		cc.Close()
	}()

	for {
		cc.alloc.Reset()
		data, err := cc.readPacket()
		if err != nil {
			if errors2.ErrorNotEqual(err, io.EOF) {
				log.Error(err)
			}
			return
		}

		if err := cc.dispatch(data); err != nil {
			if errors2.ErrorEqual(err, io.EOF) {
				return
			}
			log.Errorf("dispatch error %s, %s", errors.ErrorStack(err), cc)
			log.Errorf("cmd: %s", string(data[1:]))
			cc.writeError(err)
		}

		cc.pkg.sequence = 0
	}
}

// Exec implements the stmt.Statement Exec interface.
func (s *DropDatabaseStmt) Exec(ctx context.Context) (rset.Recordset, error) {
	err := sessionctx.GetDomain(ctx).DDL().DropSchema(ctx, model.NewCIStr(s.Name))
	if errors2.ErrorEqual(err, ddl.ErrNotExists) && s.IfExists {
		err = nil
	}
	return nil, errors.Trace(err)
}

// Both lock and unlock are used for simulating scenario of percolator papers.
func (s *dbStore) tryConditionLockKey(tid uint64, key string, snapshotVal []byte) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	if _, ok := s.keysLocked[key]; ok {
		return errors.Trace(kv.ErrLockConflict)
	}

	metaKey := codec.EncodeBytes(nil, []byte(key))
	currValue, err := s.db.Get(metaKey)
	if errors2.ErrorEqual(err, kv.ErrNotExist) || currValue == nil {
		// If it's a new key, we won't need to check its version
		return nil
	}
	if err != nil {
		return errors.Trace(err)
	}
	_, ver, err := codec.DecodeUint(currValue)
	if err != nil {
		return errors.Trace(err)
	}

	// If there's newer version of this key, returns error.
	if ver > tid {
		log.Warnf("txn:%d, tryLockKey condition not match for key %s, currValue:%q, snapshotVal:%q", tid, key, currValue, snapshotVal)
		return errors.Trace(kv.ErrConditionNotMatch)
	}

	s.keysLocked[key] = tid

	return nil
}

// Bootstrap initiates TiDB server.
func Bootstrap(store kv.Storage) {
	td := NewTiDBDriver(store)
	tc, err := td.OpenCtx(defaultCapability, mysql.DefaultCollationID, "")
	defer tc.Close()
	if err != nil {
		log.Fatal(err)
	}
	// Create a test database.
	_, err = tc.Execute("CREATE DATABASE IF NOT EXISTS test")
	if err != nil {
		log.Fatal(err)
	}

	//  Check if mysql db exists.
	_, err = tc.Execute("USE mysql;")
	if err == nil {
		// We have already finished bootstrap.
		return
	} else if !errors2.ErrorEqual(err, tidberrors.ErrDatabaseNotExist) {
		log.Fatal(err)
	}
	_, err = tc.Execute("CREATE DATABASE mysql;")
	if err != nil {
		log.Fatal(err)
	}
	_, err = tc.Execute("CREATE TABLE mysql.user (Host CHAR(64), User CHAR(16), Password CHAR(41), PRIMARY KEY (Host, User));")
	if err != nil {
		log.Fatal(err)
	}
	// Insert a default user with empty password.
	_, err = tc.Execute(`INSERT INTO mysql.user VALUES ("localhost", "root", ""), ("127.0.0.1", "root", "");`)
	if err != nil {
		log.Fatal(err)
	}
}

// AddRecord implements table.Table AddRecord interface.
func (t *Table) AddRecord(ctx context.Context, r []interface{}) (recordID int64, err error) {
	id := variable.GetSessionVars(ctx).LastInsertID
	// Already have auto increment ID
	if id != 0 {
		recordID = int64(id)
	} else {
		recordID, err = t.alloc.Alloc(t.ID)
		if err != nil {
			return 0, err
		}
	}
	txn, err := ctx.GetTxn(false)
	if err != nil {
		return 0, err
	}
	for _, v := range t.indices {
		if v == nil {
			continue
		}
		colVals, _ := v.FetchValues(r)
		if err = v.X.Create(txn, colVals, recordID); err != nil {
			if errors2.ErrorEqual(err, kv.ErrKeyExists) {
				// Get the duplicate row handle
				iter, _, terr := v.X.Seek(txn, colVals)
				if terr != nil {
					return 0, errors.Trace(terr)
				}
				_, h, terr := iter.Next()
				if terr != nil {
					return 0, errors.Trace(terr)
				}
				return h, errors.Trace(err)
			}
			return 0, errors.Trace(err)
		}
	}

	// split a record into multiple kv pair
	// first key -> LOCK
	k := t.RecordKey(recordID, nil)
	// A new row with current txn-id as lockKey
	err = txn.Set([]byte(k), []byte(txn.String()))
	if err != nil {
		return 0, err
	}
	// column key -> column value
	for _, c := range t.Cols() {
		colKey := t.RecordKey(recordID, c)
		data, err := t.EncodeValue(r[c.Offset])
		if err != nil {
			return 0, err
		}
		err = txn.Set([]byte(colKey), data)
		if err != nil {
			return 0, err
		}
	}
	variable.GetSessionVars(ctx).AddAffectedRows(1)
	return recordID, nil
}

func (d *ddl) deleteTableData(ctx context.Context, t table.Table) error {
	// Remove data
	err := t.Truncate(ctx)
	if err != nil {
		return errors.Trace(err)
	}
	txn, err := ctx.GetTxn(false)
	if err != nil {
		return errors.Trace(err)
	}
	// Remove indices
	for _, v := range t.Indices() {
		if v != nil && v.X != nil {
			if err = v.X.Drop(txn); err != nil {
				return errors.Trace(err)
			}
		}
	}
	// Remove auto ID key
	err = txn.Delete([]byte(meta.AutoIDKey(t.TableID())))

	// Auto ID meta is created when the first time used, so it may not exist.
	if errors2.ErrorEqual(err, kv.ErrNotExist) {
		return nil
	}
	return errors.Trace(err)
}

// HGet gets the value of a hash field.
func (t *TxStructure) HGet(key []byte, field []byte) ([]byte, error) {
	dataKey := t.encodeHashDataKey(key, field)
	value, err := t.txn.Get(dataKey)
	if errors2.ErrorEqual(err, kv.ErrNotExist) {
		err = nil
	}
	return value, errors.Trace(err)
}

// Clear removes the string value of the key.
func (t *TxStructure) Clear(key []byte) error {
	ek := t.encodeStringDataKey(key)
	err := t.txn.Delete(ek)
	if errors2.ErrorEqual(err, kv.ErrNotExist) {
		err = nil
	}
	return errors.Trace(err)
}

// Get gets the string value of a key.
func (t *TxStructure) Get(key []byte) ([]byte, error) {
	ek := t.encodeStringDataKey(key)
	value, err := t.txn.Get(ek)
	if errors2.ErrorEqual(err, kv.ErrNotExist) {
		err = nil
	}
	return value, errors.Trace(err)
}

// Inc increments the integer value of a key by step, returns
// the value after the increment.
func (t *TxStructure) Inc(key []byte, step int64) (int64, error) {
	ek := t.encodeStringDataKey(key)
	// txn Inc will lock this key, so we don't lock it here.
	n, err := t.txn.Inc(ek, step)
	if errors2.ErrorEqual(err, kv.ErrNotExist) {
		err = nil
	}
	return n, errors.Trace(err)
}

// Exec implements the stmt.Statement Exec interface.
func (s *CreateTableStmt) Exec(ctx context.Context) (_ rset.Recordset, err error) {
	err = sessionctx.GetDomain(ctx).DDL().CreateTable(ctx, s.Ident.Full(ctx), s.Cols, s.Constraints)
	if errors2.ErrorEqual(err, ddl.ErrExists) {
		if s.IfNotExists {
			return nil, nil
		}
		return nil, errors.Errorf("CREATE TABLE: table exists %s", s.Ident)
	}
	return nil, errors.Trace(err)
}

func (t *TxStructure) loadHashValue(dataKey []byte) ([]byte, error) {
	v, err := t.txn.Get(dataKey)
	if errors2.ErrorEqual(err, kv.ErrNotExist) {
		err = nil
		v = nil
	} else if err != nil {
		return nil, errors.Trace(err)
	}

	return v, nil
}

// Inc increments the integer value of a key by step, returns
// the value after the increment.
func (t *TxStructure) Inc(key []byte, step int64) (int64, error) {
	ek := t.encodeStringDataKey(key)
	if err := t.txn.LockKeys(ek); err != nil {
		return 0, errors.Trace(err)
	}

	n, err := t.txn.Inc(ek, step)
	if errors2.ErrorEqual(err, kv.ErrNotExist) {
		err = nil
	}
	return n, errors.Trace(err)
}

// Exec implements the stmt.Statement Exec interface.
func (s *ReplaceIntoStmt) Exec(ctx context.Context) (_ rset.Recordset, err error) {
	t, err := getTable(ctx, s.TableIdent)
	if err != nil {
		return nil, errors.Trace(err)
	}
	cols, err := s.getColumns(t.Cols())
	if err != nil {
		return nil, errors.Trace(err)
	}

	// Process `replace ... (select ...)`
	// TODO: handles the duplicate-key in a primary key or a unique index.
	if s.Sel != nil {
		return s.execSelect(t, cols, ctx)
	}
	// Process `replace ... set x=y ...`
	if err = s.fillValueList(); err != nil {
		return nil, errors.Trace(err)
	}
	m, err := s.getDefaultValues(ctx, t.Cols())
	if err != nil {
		return nil, errors.Trace(err)
	}
	replaceValueCount := len(s.Lists[0])

	for i, list := range s.Lists {
		if err = s.checkValueCount(replaceValueCount, len(list), i, cols); err != nil {
			return nil, errors.Trace(err)
		}
		row, err := s.getRow(ctx, t, cols, list, m)
		if err != nil {
			return nil, errors.Trace(err)
		}
		h, err := t.AddRecord(ctx, row)
		if err == nil {
			continue
		}
		if err != nil && !errors2.ErrorEqual(err, kv.ErrKeyExists) {
			return nil, errors.Trace(err)
		}

		// While the insertion fails because a duplicate-key error occurs for a primary key or unique index,
		// a storage engine may perform the REPLACE as an update rather than a delete plus insert.
		// See: http://dev.mysql.com/doc/refman/5.7/en/replace.html.
		if err = replaceRow(ctx, t, h, row); err != nil {
			return nil, errors.Trace(err)
		}
		variable.GetSessionVars(ctx).AddAffectedRows(1)
	}

	return nil, nil
}

// AddRecord implements table.Table AddRecord interface.
func (t *Table) AddRecord(ctx context.Context, r []interface{}) (recordID int64, err error) {
	id := variable.GetSessionVars(ctx).LastInsertID
	// Already have auto increment ID
	if id != 0 {
		recordID = int64(id)
	} else {
		recordID, err = t.alloc.Alloc(t.ID)
		if err != nil {
			return 0, errors.Trace(err)
		}
	}
	txn, err := ctx.GetTxn(false)
	if err != nil {
		return 0, errors.Trace(err)
	}
	for _, v := range t.indices {
		if v == nil {
			continue
		}
		colVals, _ := v.FetchValues(r)
		if err = v.X.Create(txn, colVals, recordID); err != nil {
			if errors2.ErrorEqual(err, kv.ErrKeyExists) {
				// Get the duplicate row handle
				// For insert on duplicate syntax, we should update the row
				iter, _, err1 := v.X.Seek(txn, colVals)
				if err1 != nil {
					return 0, errors.Trace(err1)
				}
				_, h, err1 := iter.Next()
				if err1 != nil {
					return 0, errors.Trace(err1)
				}
				return h, errors.Trace(err)
			}
			return 0, errors.Trace(err)
		}
	}

	if err := t.LockRow(ctx, recordID); err != nil {
		return 0, errors.Trace(err)
	}

	// column key -> column value
	for _, c := range t.Cols() {
		k := t.RecordKey(recordID, c)
		if err := t.SetColValue(txn, k, r[c.Offset]); err != nil {
			return 0, errors.Trace(err)
		}
	}
	variable.GetSessionVars(ctx).AddAffectedRows(1)
	return recordID, nil
}