package raftNode2

import (
	"PPOSUtils/utils"
	"context"
	"encoding/json"
	"fmt"
	"github.com/lni/dragonboat/v3"
	"github.com/lni/dragonboat/v3/config"
	"github.com/lni/dragonboat/v3/raftio"
	"os"
	"strings"
	"sync"
	"time"
)

const (
	exampleClusterID uint64 = 128
)

type RaftHost struct {
	nodeID            int64
	leaderID          string
	nodeHost          *dragonboat.NodeHost
	nodeMember        map[uint64]string
	nodeMemberReverse map[string]string
	leaderState       bool
	leaderChange      chan bool
	rwLock            sync.RWMutex
}

func (that *RaftHost) Init(cfg *RaftConfig) {
	that.nodeMember = map[uint64]string{}
	that.nodeMemberReverse = map[string]string{}
	that.leaderChange = make(chan bool, 1)
	_nodeAddress := cfg.NodeAddress
	_raftId := cfg.NodeId
	that.nodeID = utils.ToInt64(_raftId)
	_raftCluster := cfg.NodeCluster
	_raftDir := cfg.NodeDataDir
	_clusterSlice := strings.Split(_raftCluster, ",")
	for _, _v := range _clusterSlice {
		_cluster := strings.Split(_v, "/")
		that.nodeMemberReverse[_cluster[1]] = _cluster[0]
		that.nodeMember[uint64(utils.ToInt64(_cluster[0]))] = _cluster[1]
	}
	//
	_nh, _err := dragonboat.NewNodeHost(config.NodeHostConfig{
		WALDir:            _raftDir,
		NodeHostDir:       _raftDir,
		RTTMillisecond:    200,
		RaftAddress:       _nodeAddress,
		RaftEventListener: that,
	})
	if _err != nil {
		panic(_err)
	}
	if _err := _nh.StartOnDiskCluster(that.nodeMember, false, NewDiskKV, config.Config{
		// ClusterID and NodeID of the raft node
		NodeID:             uint64(utils.ToInt64(_raftId)),
		ClusterID:          exampleClusterID,
		ElectionRTT:        10,
		HeartbeatRTT:       1,
		CheckQuorum:        true,
		SnapshotEntries:    10,
		CompactionOverhead: 5,
	}); _err != nil {
		fmt.Fprintf(os.Stderr, "failed to add cluster, %v\n", _err)
		os.Exit(1)
	}
	that.nodeHost = _nh
}

func (that *RaftHost) Find(tree, key string, timeout time.Duration) (FsmDataInterface, error) {
	_nh := that.nodeHost
	_ctx, _cancel := context.WithTimeout(context.Background(), timeout)
	defer _cancel()
	_data, _err := _nh.SyncRead(_ctx, exampleClusterID, createEntryKey(tree, key))
	if _err != nil {
		return nil, _err
	}
	_fsmEntry := &FsmEntry{}
	if _err := json.Unmarshal([]byte(fmt.Sprintf("%s", _data)), &_fsmEntry); _err != nil {
		fmt.Println("======= Find error =======", _err.Error())
		return nil, _err
	}
	_entry, _err := parseEntry([]byte(_fsmEntry.Val))
	return _entry, nil
}

func (that *RaftHost) List(timeout time.Duration) ([]FsmDataInterface, error) {
	_nh := that.nodeHost
	_ctx, _cancel := context.WithTimeout(context.Background(), timeout)
	defer _cancel()
	_data, _err := _nh.SyncRead(_ctx, exampleClusterID, []byte(findAllKey))
	if _err != nil {
		return nil, _err
	}
	_slice := []*FsmEntry{}
	_dataStr := fmt.Sprintf("%s", _data)
	fmt.Println("======= List data =======", _dataStr)
	if _err := json.Unmarshal([]byte(_dataStr), &_slice); _err != nil {
		fmt.Println("======= List error =======", _err.Error())
		return nil, _err
	}
	_rs := []FsmDataInterface{}

	for _k, _v := range _slice {
		if _v.Key == appliedIndexKey {
			continue
		}
		fmt.Println(_k, _v.Key, len(_slice), "======= List =======", _v.Val)
		_entry, _err := parseEntry([]byte(_v.Val))
		if _err != nil {
			fmt.Println("======= List error2 =======", _err)
			continue
		}
		_rs = append(_rs, _entry)
	}
	return _rs, nil
}

func (that *RaftHost) Apply(data FsmDataInterface, timeout time.Duration) error {
	//
	_data := createEntry(data)
	//
	_nh := that.nodeHost
	_ctx, _cancel := context.WithTimeout(context.Background(), timeout)
	_cs := _nh.GetNoOPSession(exampleClusterID)
	if _, _err := _nh.SyncPropose(_ctx, _cs, _data); _err != nil {
		return _err
	}
	_cancel()
	return nil
}

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) 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) LeaderUpdated(info raftio.LeaderInfo) {
	//fmt.Println("========= LeaderUpdated =========", info.LeaderID, info.NodeID, info.Term, info.ClusterID)
	_leader := false
	that.SetLeaderID(utils.ToString(info.LeaderID, true))
	if info.LeaderID == uint64(that.nodeID) {
		_leader = true
	}
	if that.IsLeader() != _leader {
		that.leaderChange <- _leader
		that.SetLeaderState(_leader)
	}
}

func (that *RaftHost) Close() {
	that.nodeHost.Stop()
	that.nodeHost = nil
}
