package raftNode

import (
	"PPOSUtils/utils"
	"fmt"
	"github.com/hashicorp/raft"
	"reflect"

	//raftboltdb "github.com/hashicorp/raft-boltdb"
	"net"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"
)

type RaftConfig struct {
	NodeId      string `yaml:"node_id"`       // 节点编号，例：1
	NodeAddress string `yaml:"node_address"`  // 节点地址与端口，例：127.0.0.1:11086
	NodeCluster string `yaml:"node_cluster"`  // 群集节点列表，例：1/127.0.0.1:11086,2/127.0.0.1:11087,3/127.0.0.1:11088
	NodeDataDir string `yaml:"node_data_dir"` // 数据目录，例：data/node1
}

type RaftHost struct {
	nodeID        int64
	rwLock        sync.RWMutex
	nodeHost      *raft.Raft
	nodeFmt       FsmInterface
	leaderChange  chan bool
	leaderState   bool
	canWriteState bool
	leaderID      string
	lastIndex     uint64
	//nodeMember        map[uint64]string
	nodeMemberReverse map[string]string
}

func (that *RaftHost) init(config *RaftConfig) {
	//that.NotifyCh = make(chan bool, 1)
	that.leaderChange = make(chan bool)
	that.nodeMemberReverse = map[string]string{}
	_raftAddr := config.NodeAddress
	_raftId := config.NodeId
	that.nodeID = utils.ToInt64(_raftId)
	_raftCluster := config.NodeCluster
	_raftDir := config.NodeDataDir
	// 初始化raft
	_nodeHost, _fm, _err := that.newMyRaft(_raftAddr, _raftId, _raftDir)
	if _err != nil {
		log.Panic("NewMyRaft error: " + _err.Error())
		panic("NewMyRaft error: " + _err.Error())
		return
	}
	//
	_clusterSlice := strings.Split(_raftCluster, ",")
	for _, _v := range _clusterSlice {
		_cluster := strings.Split(_v, "/")
		that.nodeMemberReverse[_cluster[1]] = _cluster[0]
	}
	that.lastIndex = _nodeHost.LastIndex()
	that.nodeFmt = _fm
	that.nodeHost = _nodeHost
	go that.LeaderUpdated()
	// 启动raft
	that.bootstrap(_nodeHost, _raftId, _raftAddr, _raftCluster)
}

//func (that *RaftHost) MyRaft() *raft.Raft {
//	return that.myRaft
//}

func (that *RaftHost) LeaderID() string {
	that.rwLock.RLock()
	defer that.rwLock.RUnlock()
	return that.leaderID
}

func (that *RaftHost) SetLeaderID(leaderID string) {
	that.rwLock.RLock()
	defer that.rwLock.RUnlock()
	that.leaderID = leaderID
}

func (that *RaftHost) LeaderCh() <-chan bool {
	return that.leaderChange
}

func (that *RaftHost) SetLeaderState(leader bool) {
	that.rwLock.Lock()
	defer that.rwLock.Unlock()
	that.leaderState = leader
}

func (that *RaftHost) IsLeader() bool {
	that.rwLock.RLock()
	defer that.rwLock.RUnlock()
	return that.leaderState
}

func (that *RaftHost) GetCanWrite() bool {
	that.rwLock.RLock()
	defer that.rwLock.RUnlock()
	return that.canWriteState
}

func (that *RaftHost) SetCanWrite(canWrite bool) {
	that.rwLock.Lock()
	defer that.rwLock.Unlock()
	that.canWriteState = canWrite
}

func (that *RaftHost) CanWrite() bool {
	_i := 0
	for {
		if !that.IsLeader() {
			return false
		}
		if that.GetCanWrite() {
			return true
		}
		if _i > 20 {
			log.Info("[CanWrite] call depth more than 20")
			return false
		}
		time.Sleep(time.Duration(200) * time.Millisecond)
		_i++
	}
}

func (that *RaftHost) List(timeout time.Duration) ([]FsmDataInterface, error) {
	_fm := that.nodeFmt
	_slice := _fm.List()
	_rs := []FsmDataInterface{}
	for _, _v := range _slice {
		_val, _ok := _v.(string)
		if !_ok {
			fmt.Println("====== RaftHost.List error 001 ======", reflect.TypeOf(_val))
		}
		_data, _err := parseFsmData([]byte(_val))
		if _err != nil {
			fmt.Println(_val, "====== RaftHost.List error 002 ======", _err)
			continue
		}
		_rs = append(_rs, _data)
	}
	return _rs, nil
}

func (that *RaftHost) Find(tree, key string, timeout time.Duration) (FsmDataInterface, error) {
	_entryKey := newEntryKey(tree, key)
	_fm := that.nodeFmt
	_val := _fm.Find(_entryKey.GetTree(), _entryKey.CenerateEntryKey())
	_entry, _err := parseEntry([]byte(_val.(string)))
	if _err != nil {
		return nil, _err
	}
	_data, _err := _entry.GetEntry()
	if _err != nil {
		return nil, _err
	}
	return _data, _err
}

func (that *RaftHost) Apply(data []byte, timeout time.Duration) error {
	_data := make([]byte, len(data))
	copy(_data, data)
	_future := that.nodeHost.Apply(_data, timeout)
	//
	if _err := _future.Error(); _err != nil {
		return _err
	}
	//...
	return nil
}

func (that *RaftHost) newMyRaft(raftAddr, raftId, raftDir string) (*raft.Raft, FsmInterface, error) {
	_config := raft.DefaultConfig()
	_config.LocalID = raft.ServerID(raftId)
	//_config.NotifyCh = that.NotifyCh
	_config.SnapshotInterval = 10 * time.Second
	_config.SnapshotThreshold = 100
	// config.HeartbeatTimeout = 1000 * time.Millisecond
	// config.ElectionTimeout = 1000 * time.Millisecond
	// config.CommitTimeout = 1000 * time.Millisecond

	_addr, _err := net.ResolveTCPAddr("tcp", raftAddr)
	if _err != nil {
		return nil, nil, _err
	}
	_transport, _err := raft.NewTCPTransport(raftAddr, _addr, 2, 10*time.Second, os.Stderr)
	if _err != nil {
		return nil, nil, _err
	}
	_snapshots, _err := raft.NewFileSnapshotStore(raftDir, 2, os.Stderr)
	if _err != nil {
		return nil, nil, _err
	}
	//_logStore, _err := raftboltdb.NewBoltStore(filepath.Join(raftDir, "raft-log.db"))
	//if _err != nil {
	//	return nil, nil, _err
	//}
	//stableStore, err := raftboltdb.NewBoltStore(filepath.Join(raftDir, "raft-stable.db"))
	//if err != nil {
	//	return nil, nil, err
	//}

	//
	_store, _err := NewLevelDBStore(filepath.Join(raftDir, "raft-log.db"), 0)
	if _err != nil {
		return nil, nil, _err
	}
	_storeStable, _err := NewLevelDBStore(filepath.Join(raftDir, "raft-stable.db"), 0)
	if _err != nil {
		return nil, nil, _err
	}
	_lastIndex, _ := _store.LastIndex()
	fmt.Println("======== start =======", _lastIndex)
	//fm, _ := NewRaftFsm2(filepath.Join(raftDir, "raft-fsm.db"), 0)
	fm := NewRaftFsm()
	rf, err := raft.NewRaft(_config, fm, _store, _storeStable, _snapshots, _transport)
	if err != nil {
		return nil, nil, err
	}
	//rf, err := raft.NewRaft(_config, fm, _logStore, stableStore, _snapshots, _transport)
	//if err != nil {
	//	return nil, nil, err
	//}
	return rf, fm, nil
}

func (that *RaftHost) bootstrap(rf *raft.Raft, raftId, raftAddr, raftCluster string) {
	servers := rf.GetConfiguration().Configuration().Servers
	if len(servers) > 0 {
		return
	}
	peerArray := strings.Split(raftCluster, ",")
	if len(peerArray) == 0 {
		return
	}

	var configuration raft.Configuration
	for _, peerInfo := range peerArray {
		peer := strings.Split(peerInfo, "/")
		id := peer[0]
		addr := peer[1]
		server := raft.Server{
			ID:      raft.ServerID(id),
			Address: raft.ServerAddress(addr),
		}
		configuration.Servers = append(configuration.Servers, server)
	}
	rf.BootstrapCluster(configuration)
	return
}

func (that *RaftHost) LeaderUpdated() {
	for {
		_t := time.NewTicker(time.Duration(500) * time.Millisecond)
		select {
		case _leaderCh := <-that.nodeHost.LeaderCh():
			//fmt.Println("===== LeaderUpdated 001 =====", _leaderCh)
			_leader := that.nodeHost.Leader()
			if _leader != "" {
				_clusterID, _ok := that.nodeMemberReverse[string(_leader)]
				if _ok {
					that.SetLeaderID(_clusterID)
				}
			}
			if that.IsLeader() != _leaderCh {
				that.leaderChange <- _leaderCh
			}
			//fmt.Println("===== LeaderUpdated 002 =====", _leaderCh)
			that.SetLeaderState(_leaderCh)
			//fmt.Println("===== LeaderUpdated 003 =====", _leaderCh)
		case <-_t.C:
			_leader := that.nodeHost.Leader()
			if _leader != "" {
				_clusterID, _ok := that.nodeMemberReverse[string(_leader)]
				if _ok {
					that.SetLeaderID(_clusterID)
				}
			}
		}
	}
}

func (that *RaftHost) Close() {
	if _err := that.nodeHost.Shutdown(); _err != nil {
		fmt.Println("shutdown raft error", _err.Error())
		fmt.Printf("shutdown raft error:%v \n", _err)
	}
}
