package raftNode

import (
	"bytes"
	"encoding/binary"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/hashicorp/raft"
	"github.com/syndtr/goleveldb/leveldb"
	"github.com/syndtr/goleveldb/leveldb/opt"
	"io"
)

const (
	// Permissions to use on the db file. This is only used if the
	// database file does not exist and needs to be created.
	dbFileMode = 0600
)

var (
	// Bucket names we perform transactions in
	dbLogs = []byte("logs")
	dbConf = []byte("conf")

	// An error indicating a given key does not exist
	ErrKeyNotFound = errors.New("not found")

	errInvalidLog = errors.New("invalid log")
)

type Level int

const (
	Low    Level = -1
	Medium Level = 0
	High   Level = 1
)

// LevelDBStore provides access to BoltDB for Raft to store and retrieve
// log entries. It also provides key/value storage, and can be used as
// a LogStore and StableStore.
type LevelDBStore struct {
	// db is the underlying handle to the db.
	db *leveldb.DB
	// The path to the Bolt database file
	path string
	wo   *opt.WriteOptions
}

// NewLevelDBStore takes a file path and returns a connected Raft backend.
func NewLevelDBStore(path string, durability Level) (*LevelDBStore, error) {
	var wo opt.WriteOptions
	var opts opt.Options
	if durability <= Low {
		opts.NoSync = true
		wo.Sync = false
	} else {
		opts.NoSync = false
		wo.Sync = true
	}
	// Try to connect
	db, err := leveldb.OpenFile(path, &opts)
	if err != nil {
		return nil, err
	}

	// Create the new store
	store := &LevelDBStore{
		db:   db,
		path: path,
		wo:   &wo,
	}
	return store, nil
}

// Close is used to gracefully close the DB connection.
func (that *LevelDBStore) Close() error {
	return that.db.Close()
}

// FirstIndex returns the first known index from the Raft log.
func (that *LevelDBStore) FirstIndex() (uint64, error) {
	var n uint64
	iter := that.db.NewIterator(nil, nil)
	for ok := iter.Seek(dbLogs); ok; ok = iter.Next() {
		// Use key/value.
		key := iter.Key()
		if !bytes.HasPrefix(key, dbLogs) {
			break
		}
		n = bytesToUint64(key[len(dbLogs):])
		break
	}
	iter.Release()
	err := iter.Error()
	if err != nil {
		return 0, err
	}
	return n, nil
}

// LastIndex returns the last known index from the Raft log.
func (that *LevelDBStore) LastIndex() (uint64, error) {
	var n uint64
	iter := that.db.NewIterator(nil, nil)
	for ok := iter.Last(); ok; ok = iter.Prev() {
		key := iter.Key()
		if !bytes.HasPrefix(key, dbLogs) {
			break
		}
		n = bytesToUint64(key[len(dbLogs):])
		break
	}
	iter.Release()
	err := iter.Error()
	if err != nil {
		return 0, err
	}
	return n, nil
}

// GetLog is used to retrieve a log from BoltDB at a given index.
func (that *LevelDBStore) GetLog(idx uint64, log *raft.Log) error {
	key := append(dbLogs, uint64ToBytes(idx)...)
	val, err := that.db.Get(key, nil)
	if err != nil {
		if err == leveldb.ErrNotFound {
			return raft.ErrLogNotFound
		}
		return err
	}
	return decodeLog(val, log)
}

// StoreLog is used to store a single raft log
func (that *LevelDBStore) StoreLog(log *raft.Log) error {
	return that.StoreLogs([]*raft.Log{log})
}

// StoreLogs is used to store a set of raft logs
func (that *LevelDBStore) StoreLogs(logs []*raft.Log) error {
	var batch leveldb.Batch
	for _, log := range logs {
		key := append(dbLogs, uint64ToBytes(log.Index)...)
		batch.Put(key, encodeLog(log))
	}
	return that.db.Write(&batch, that.wo)
}

// DeleteRange is used to delete logs within a given range inclusively.
func (that *LevelDBStore) DeleteRange(min, max uint64) error {
	var batch leveldb.Batch
	prefix := append(dbLogs, uint64ToBytes(min)...)
	iter := that.db.NewIterator(nil, nil)
	for ok := iter.Seek(prefix); ok; ok = iter.Next() {
		key := iter.Key()
		if !bytes.HasPrefix(key, dbLogs) {
			break
		}
		if bytesToUint64(key[len(dbLogs):]) > max {
			break
		}
		batch.Delete(key)
	}
	iter.Release()
	err := iter.Error()
	if err != nil {
		return err
	}
	return that.db.Write(&batch, that.wo)
}

// Set is used to set a key/value set outside of the raft log
func (that *LevelDBStore) Set(k, v []byte) error {
	return that.db.Put(append(dbConf, k...), v, that.wo)
}

// Get is used to retrieve a value from the k/v store by key
func (that *LevelDBStore) Get(k []byte) ([]byte, error) {
	val, err := that.db.Get(append(dbConf, k...), nil)
	if err != nil {
		if err == leveldb.ErrNotFound {
			return nil, ErrKeyNotFound
		}
		return nil, err
	}
	return bcopy(val), nil
}

// SetUint64 is like Set, but handles uint64 values
func (that *LevelDBStore) SetUint64(key []byte, val uint64) error {
	return that.Set(key, uint64ToBytes(val))
}

// GetUint64 is like Get, but handles uint64 values
func (that *LevelDBStore) GetUint64(key []byte) (uint64, error) {
	val, err := that.Get(key)
	if err != nil {
		return 0, err
	}
	return bytesToUint64(val), nil
}

// Peers returns raft peers
func (b *LevelDBStore) Peers() ([]string, error) {
	var peers []string
	val, err := b.Get([]byte("peers"))
	if err != nil {
		if err == ErrKeyNotFound {
			return []string{}, nil
		}
		return nil, err
	}
	if err := json.Unmarshal(val, &peers); err != nil {
		return nil, err
	}
	return peers, nil
}

// SetPeers sets raft peers
func (b *LevelDBStore) SetPeers(peers []string) error {
	data, err := json.Marshal(peers)
	if err != nil {
		return err
	}
	return b.Set([]byte("peers"), data)
}

//Apply flower同步写入
func (that *LevelDBStore) Apply(l *raft.Log) interface{} {
	_entry, _err := parseEntry(l.Data)
	if _err != nil {
		fmt.Println(string(l.Data), "====== raftFsm.Apply error 001 ======", _err)
		return nil
	}
	_key := _entry.GetKey()
	_val := _entry.GetVal()
	//
	if _val == "" {
		that.db.Delete([]byte(_key), that.wo)
		return nil
	}
	that.db.Put([]byte(_key), []byte(_val), that.wo)
	return nil
}

//Snapshot 返回快照接口类型
func (that *LevelDBStore) Snapshot() (raft.FSMSnapshot, error) {
	return that, nil
}

//Restore 从快照中恢复到内存
func (that *LevelDBStore) Restore(r io.ReadCloser) error {
	return nil
}

//Persist 生成快照
func (that LevelDBStore) Persist(sink raft.SnapshotSink) error {
	return nil
}

func (that LevelDBStore) Release() {
}

func bcopy(b []byte) []byte {
	r := make([]byte, len(b))
	copy(r, b)
	return r
}

// Decode reverses the encode operation on a byte slice input
func decodeLog(buf []byte, log *raft.Log) error {
	if len(buf) < 25 {
		return errInvalidLog
	}
	log.Index = binary.LittleEndian.Uint64(buf[0:8])
	log.Term = binary.LittleEndian.Uint64(buf[8:16])
	log.Type = raft.LogType(buf[16])
	log.Data = make([]byte, binary.LittleEndian.Uint64(buf[17:25]))
	if len(buf[25:]) < len(log.Data) {
		return errInvalidLog
	}
	copy(log.Data, buf[25:])
	return nil
}

// Encode writes an encoded object to a new bytes buffer
func encodeLog(log *raft.Log) []byte {
	var buf []byte
	var num = make([]byte, 8)
	binary.LittleEndian.PutUint64(num, log.Index)
	buf = append(buf, num...)
	binary.LittleEndian.PutUint64(num, log.Term)
	buf = append(buf, num...)
	buf = append(buf, byte(log.Type))
	binary.LittleEndian.PutUint64(num, uint64(len(log.Data)))
	buf = append(buf, num...)
	buf = append(buf, log.Data...)
	return buf
}

// Converts bytes to an integer
func bytesToUint64(b []byte) uint64 {
	return binary.BigEndian.Uint64(b)
}

// Converts a uint to a byte slice
func uint64ToBytes(u uint64) []byte {
	buf := make([]byte, 8)
	binary.BigEndian.PutUint64(buf, u)
	return buf
}
