// Copyright 2018 The dexon-consensus Authors
// This file is part of the dexon-consensus library.
//
// The dexon-consensus library is free software: you can redistribute it
// and/or modify it under the terms of the GNU Lesser General Public License as
// published by the Free Software Foundation, either version 3 of the License,
// or (at your option) any later version.
//
// The dexon-consensus library is distributed in the hope that it will be
// useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser
// General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the dexon-consensus library. If not, see
// <http://www.gnu.org/licenses/>.

package test

import (
	"bytes"
	"errors"
	"sort"
	"sync"
	"time"

	"github.com/dexon-foundation/dexon-consensus/common"
	"github.com/dexon-foundation/dexon-consensus/core/crypto"
	"github.com/dexon-foundation/dexon-consensus/core/crypto/ecdsa"
	"github.com/dexon-foundation/dexon-consensus/core/types"
	typesDKG "github.com/dexon-foundation/dexon-consensus/core/types/dkg"
	"github.com/dexon-foundation/dexon/rlp"
)

var (
	// ErrDuplicatedChange means the change request is already applied.
	ErrDuplicatedChange = errors.New("duplicated change")
	// ErrForkedCRS means a different CRS for one round is proposed.
	ErrForkedCRS = errors.New("forked CRS")
	// ErrMissingPreviousCRS means previous CRS not found when
	// proposing a specific round of CRS.
	ErrMissingPreviousCRS = errors.New("missing previous CRS")
	// ErrUnknownStateChangeType means a StateChangeType is not recognized.
	ErrUnknownStateChangeType = errors.New("unknown state change type")
	// ErrProposerMPKIsReady means a proposer of one mpk is ready.
	ErrProposerMPKIsReady = errors.New("proposer mpk is ready")
	// ErrProposerIsFinal means a proposer of one complaint is finalized.
	ErrProposerIsFinal = errors.New("proposer is final")
	// ErrStateConfigNotEqual means configuration part of two states is not
	// equal.
	ErrStateConfigNotEqual = errors.New("config not equal")
	// ErrStateLocalFlagNotEqual means local flag of two states is not equal.
	ErrStateLocalFlagNotEqual = errors.New("local flag not equal")
	// ErrStateNodeSetNotEqual means node sets of two states are not equal.
	ErrStateNodeSetNotEqual = errors.New("node set not equal")
	// ErrStateDKGComplaintsNotEqual means DKG complaints for two states are not
	// equal.
	ErrStateDKGComplaintsNotEqual = errors.New("dkg complaints not equal")
	// ErrStateDKGMasterPublicKeysNotEqual means DKG master public keys of two
	// states are not equal.
	ErrStateDKGMasterPublicKeysNotEqual = errors.New(
		"dkg master public keys not equal")
	// ErrStateDKGMPKReadysNotEqual means DKG readys of two states are not
	// equal.
	ErrStateDKGMPKReadysNotEqual = errors.New("dkg readys not equal")
	// ErrStateDKGFinalsNotEqual means DKG finalizations of two states are not
	// equal.
	ErrStateDKGFinalsNotEqual = errors.New("dkg finalizations not equal")
	// ErrStateDKGSuccessesNotEqual means DKG successes of two states are not
	// equal.
	ErrStateDKGSuccessesNotEqual = errors.New("dkg successes not equal")
	// ErrStateCRSsNotEqual means CRSs of two states are not equal.
	ErrStateCRSsNotEqual = errors.New("crs not equal")
	// ErrStateDKGResetCountNotEqual means dkgResetCount of two states are not
	// equal.
	ErrStateDKGResetCountNotEqual = errors.New("dkg reset count not equal")
	// ErrStatePendingChangesNotEqual means pending change requests of two
	// states are not equal.
	ErrStatePendingChangesNotEqual = errors.New("pending changes not equal")
	// ErrChangeWontApply means the state change won't be applied for some
	// reason.
	ErrChangeWontApply = errors.New("change won't apply")
	// ErrNotInRemoteMode means callers attempts to call functions for remote
	// mode when the State instance is still in local mode.
	ErrNotInRemoteMode = errors.New(
		"attempting to use remote functions in local mode")
)

type crsAdditionRequest struct {
	Round uint64      `json:"round"`
	CRS   common.Hash `json:"crs"`
}

// State emulates what the global state in governace contract on a fullnode.
type State struct {
	// Configuration related.
	lambdaBA         time.Duration
	lambdaDKG        time.Duration
	notarySetSize    uint32
	roundInterval    uint64
	minBlockInterval time.Duration
	// Nodes
	nodes map[types.NodeID]crypto.PublicKey
	// DKG & CRS
	dkgComplaints       map[uint64]map[types.NodeID][]*typesDKG.Complaint
	dkgMasterPublicKeys map[uint64]map[types.NodeID]*typesDKG.MasterPublicKey
	dkgReadys           map[uint64]map[types.NodeID]*typesDKG.MPKReady
	dkgFinals           map[uint64]map[types.NodeID]*typesDKG.Finalize
	dkgSuccesses        map[uint64]map[types.NodeID]*typesDKG.Success
	crs                 []common.Hash
	dkgResetCount       map[uint64]uint64
	// Other stuffs
	local           bool
	logger          common.Logger
	lock            sync.RWMutex
	appliedRequests map[common.Hash]struct{}
	// Pending change requests.
	ownRequests    map[common.Hash]*StateChangeRequest
	globalRequests map[common.Hash]*StateChangeRequest
}

// NewState constructs an State instance with genesis information, including:
//  - node set
//  - crs
func NewState(
	dkgDelayRound uint64,
	nodePubKeys []crypto.PublicKey,
	lambda time.Duration,
	logger common.Logger,
	local bool) *State {
	nodes := make(map[types.NodeID]crypto.PublicKey)
	for _, key := range nodePubKeys {
		nodes[types.NewNodeID(key)] = key
	}
	genesisCRS := crypto.Keccak256Hash([]byte("__ DEXON"))
	crs := make([]common.Hash, dkgDelayRound+1)
	for i := range crs {
		crs[i] = genesisCRS
		genesisCRS = crypto.Keccak256Hash(genesisCRS[:])
	}
	return &State{
		local:            local,
		logger:           logger,
		lambdaBA:         lambda,
		lambdaDKG:        lambda * 10,
		roundInterval:    1000,
		minBlockInterval: 4 * lambda,
		crs:              crs,
		nodes:            nodes,
		notarySetSize:    uint32(len(nodes)),
		ownRequests:      make(map[common.Hash]*StateChangeRequest),
		globalRequests:   make(map[common.Hash]*StateChangeRequest),
		dkgReadys: make(
			map[uint64]map[types.NodeID]*typesDKG.MPKReady),
		dkgFinals: make(
			map[uint64]map[types.NodeID]*typesDKG.Finalize),
		dkgSuccesses: make(
			map[uint64]map[types.NodeID]*typesDKG.Success),
		dkgComplaints: make(
			map[uint64]map[types.NodeID][]*typesDKG.Complaint),
		dkgMasterPublicKeys: make(
			map[uint64]map[types.NodeID]*typesDKG.MasterPublicKey),
		dkgResetCount:   make(map[uint64]uint64),
		appliedRequests: make(map[common.Hash]struct{}),
	}
}

// SwitchToRemoteMode turn this State instance into remote mode: all changes
// are pending, and need to be packed/unpacked to apply. Once this state switch
// to remote mode, there would be no way to switch back to local mode.
func (s *State) SwitchToRemoteMode() {
	s.lock.Lock()
	defer s.lock.Unlock()
	s.local = false
}

// Snapshot returns configration that could be snapshotted.
func (s *State) Snapshot() (*types.Config, []crypto.PublicKey) {
	s.lock.RLock()
	defer s.lock.RUnlock()
	// Clone a node set.
	nodes := make([]crypto.PublicKey, 0, len(s.nodes))
	for _, key := range s.nodes {
		nodes = append(nodes, key)
	}
	cfg := &types.Config{
		LambdaBA:         s.lambdaBA,
		LambdaDKG:        s.lambdaDKG,
		NotarySetSize:    s.notarySetSize,
		RoundLength:      s.roundInterval,
		MinBlockInterval: s.minBlockInterval,
	}
	s.logger.Info("Snapshot config", "config", cfg)
	return cfg, nodes
}

// AttachLogger allows to attach custom logger.
func (s *State) AttachLogger(logger common.Logger) {
	s.logger = logger
}

func (s *State) unpackPayload(
	raw *rawStateChangeRequest) (v interface{}, err error) {
	switch raw.Type {
	case StateAddCRS:
		v = &crsAdditionRequest{}
		err = rlp.DecodeBytes(raw.Payload, v)
	case StateAddDKGComplaint:
		v = &typesDKG.Complaint{}
		err = rlp.DecodeBytes(raw.Payload, v)
	case StateAddDKGMasterPublicKey:
		v = &typesDKG.MasterPublicKey{}
		err = rlp.DecodeBytes(raw.Payload, v)
	case StateAddDKGMPKReady:
		v = &typesDKG.MPKReady{}
		err = rlp.DecodeBytes(raw.Payload, v)
	case StateAddDKGFinal:
		v = &typesDKG.Finalize{}
		err = rlp.DecodeBytes(raw.Payload, v)
	case StateAddDKGSuccess:
		v = &typesDKG.Success{}
		err = rlp.DecodeBytes(raw.Payload, v)
	case StateResetDKG:
		var tmp common.Hash
		err = rlp.DecodeBytes(raw.Payload, &tmp)
		v = tmp
	case StateChangeLambdaBA:
		var tmp uint64
		err = rlp.DecodeBytes(raw.Payload, &tmp)
		v = tmp
	case StateChangeLambdaDKG:
		var tmp uint64
		err = rlp.DecodeBytes(raw.Payload, &tmp)
		v = tmp
	case StateChangeRoundLength:
		var tmp uint64
		err = rlp.DecodeBytes(raw.Payload, &tmp)
		v = tmp
	case StateChangeMinBlockInterval:
		var tmp uint64
		err = rlp.DecodeBytes(raw.Payload, &tmp)
		v = tmp
	case StateChangeNotarySetSize:
		var tmp uint32
		err = rlp.DecodeBytes(raw.Payload, &tmp)
		v = tmp
	case StateAddNode:
		var tmp []byte
		err = rlp.DecodeBytes(raw.Payload, &tmp)
		v = tmp
	default:
		err = ErrUnknownStateChangeType
	}
	if err != nil {
		return
	}
	return
}

func (s *State) unpackRequests(
	b []byte) (reqs []*StateChangeRequest, err error) {
	// Try to unmarshal this byte stream into []*StateChangeRequest.
	rawReqs := []*rawStateChangeRequest{}
	if err = rlp.DecodeBytes(b, &rawReqs); err != nil {
		return
	}
	for _, r := range rawReqs {
		var payload interface{}
		if payload, err = s.unpackPayload(r); err != nil {
			return
		}
		reqs = append(reqs, &StateChangeRequest{
			Type:      r.Type,
			Payload:   payload,
			Hash:      r.Hash,
			Timestamp: r.Timestamp,
		})
	}
	return
}

// Equal checks equality between State instance.
func (s *State) Equal(other *State) error {
	// Check configuration part.
	configEqual := s.lambdaBA == other.lambdaBA &&
		s.lambdaDKG == other.lambdaDKG &&
		s.notarySetSize == other.notarySetSize &&
		s.roundInterval == other.roundInterval &&
		s.minBlockInterval == other.minBlockInterval
	if !configEqual {
		return ErrStateConfigNotEqual
	}
	// Check local flag.
	if s.local != other.local {
		return ErrStateLocalFlagNotEqual
	}
	// Check node set.
	if len(s.nodes) != len(other.nodes) {
		return ErrStateNodeSetNotEqual
	}
	for nID, key := range s.nodes {
		otherKey, exists := other.nodes[nID]
		if !exists {
			return ErrStateNodeSetNotEqual
		}
		if bytes.Compare(key.Bytes(), otherKey.Bytes()) != 0 {
			return ErrStateNodeSetNotEqual
		}
	}
	// Check DKG Complaints, here I assume the addition sequence of complaints
	// proposed by one node would be identical on each node (this should be true
	// when state change requests are carried by blocks and executed in order).
	if len(s.dkgComplaints) != len(other.dkgComplaints) {
		return ErrStateDKGComplaintsNotEqual
	}
	for round, compsForRound := range s.dkgComplaints {
		otherCompsForRound, exists := other.dkgComplaints[round]
		if !exists {
			return ErrStateDKGComplaintsNotEqual
		}
		if len(compsForRound) != len(otherCompsForRound) {
			return ErrStateDKGComplaintsNotEqual
		}
		for nID, comps := range compsForRound {
			otherComps, exists := otherCompsForRound[nID]
			if !exists {
				return ErrStateDKGComplaintsNotEqual
			}
			if len(comps) != len(otherComps) {
				return ErrStateDKGComplaintsNotEqual
			}
			for idx, comp := range comps {
				if !comp.Equal(otherComps[idx]) {
					return ErrStateDKGComplaintsNotEqual
				}
			}
		}
	}
	// Check DKG master public keys.
	if len(s.dkgMasterPublicKeys) != len(other.dkgMasterPublicKeys) {
		return ErrStateDKGMasterPublicKeysNotEqual
	}
	for round, mKeysForRound := range s.dkgMasterPublicKeys {
		otherMKeysForRound, exists := other.dkgMasterPublicKeys[round]
		if !exists {
			return ErrStateDKGMasterPublicKeysNotEqual
		}
		if len(mKeysForRound) != len(otherMKeysForRound) {
			return ErrStateDKGMasterPublicKeysNotEqual
		}
		for nID, mKey := range mKeysForRound {
			otherMKey, exists := otherMKeysForRound[nID]
			if !exists {
				return ErrStateDKGMasterPublicKeysNotEqual
			}
			if !mKey.Equal(otherMKey) {
				return ErrStateDKGMasterPublicKeysNotEqual
			}
		}
	}
	// Check DKG readys.
	if len(s.dkgReadys) != len(other.dkgReadys) {
		return ErrStateDKGMPKReadysNotEqual
	}
	for round, readysForRound := range s.dkgReadys {
		otherReadysForRound, exists := other.dkgReadys[round]
		if !exists {
			return ErrStateDKGMPKReadysNotEqual
		}
		if len(readysForRound) != len(otherReadysForRound) {
			return ErrStateDKGMPKReadysNotEqual
		}
		for nID, ready := range readysForRound {
			otherReady, exists := otherReadysForRound[nID]
			if !exists {
				return ErrStateDKGMPKReadysNotEqual
			}
			if !ready.Equal(otherReady) {
				return ErrStateDKGMPKReadysNotEqual
			}
		}
	}
	// Check DKG finals.
	if len(s.dkgFinals) != len(other.dkgFinals) {
		return ErrStateDKGFinalsNotEqual
	}
	for round, finalsForRound := range s.dkgFinals {
		otherFinalsForRound, exists := other.dkgFinals[round]
		if !exists {
			return ErrStateDKGFinalsNotEqual
		}
		if len(finalsForRound) != len(otherFinalsForRound) {
			return ErrStateDKGFinalsNotEqual
		}
		for nID, final := range finalsForRound {
			otherFinal, exists := otherFinalsForRound[nID]
			if !exists {
				return ErrStateDKGFinalsNotEqual
			}
			if !final.Equal(otherFinal) {
				return ErrStateDKGFinalsNotEqual
			}
		}
	}
	// Check DKG successes.
	if len(s.dkgSuccesses) != len(other.dkgSuccesses) {
		return ErrStateDKGSuccessesNotEqual
	}
	for round, successesForRound := range s.dkgSuccesses {
		otherSuccessesForRound, exists := other.dkgSuccesses[round]
		if !exists {
			return ErrStateDKGSuccessesNotEqual
		}
		if len(successesForRound) != len(otherSuccessesForRound) {
			return ErrStateDKGSuccessesNotEqual
		}
		for nID, success := range successesForRound {
			otherSuccesse, exists := otherSuccessesForRound[nID]
			if !exists {
				return ErrStateDKGSuccessesNotEqual
			}
			if !success.Equal(otherSuccesse) {
				return ErrStateDKGSuccessesNotEqual
			}
		}
	}
	// Check CRS part.
	if len(s.crs) != len(other.crs) {
		return ErrStateCRSsNotEqual
	}
	for idx, crs := range s.crs {
		if crs != other.crs[idx] {
			return ErrStateCRSsNotEqual
		}
	}
	// Check dkgResetCount.
	if len(s.dkgResetCount) != len(other.dkgResetCount) {
		return ErrStateDKGResetCountNotEqual
	}
	for idx, count := range s.dkgResetCount {
		if count != other.dkgResetCount[idx] {
			return ErrStateDKGResetCountNotEqual
		}
	}
	// Check pending changes.
	checkPending := func(
		src, target map[common.Hash]*StateChangeRequest) error {
		if len(src) != len(target) {
			return ErrStatePendingChangesNotEqual
		}
		for k, v := range src {
			otherV, exists := target[k]
			if !exists {
				return ErrStatePendingChangesNotEqual
			}
			if err := v.Equal(otherV); err != nil {
				return err
			}
		}
		return nil
	}
	if err := checkPending(s.ownRequests, other.ownRequests); err != nil {
		return err
	}
	if err := checkPending(s.globalRequests, other.globalRequests); err != nil {
		return err
	}
	return nil
}

// Clone returns a copied State instance.
func (s *State) Clone() (copied *State) {
	// Clone configuration parts.
	copied = &State{
		lambdaBA:         s.lambdaBA,
		lambdaDKG:        s.lambdaDKG,
		notarySetSize:    s.notarySetSize,
		roundInterval:    s.roundInterval,
		minBlockInterval: s.minBlockInterval,
		local:            s.local,
		logger:           s.logger,
		nodes:            make(map[types.NodeID]crypto.PublicKey),
		dkgComplaints: make(
			map[uint64]map[types.NodeID][]*typesDKG.Complaint),
		dkgMasterPublicKeys: make(
			map[uint64]map[types.NodeID]*typesDKG.MasterPublicKey),
		dkgReadys:       make(map[uint64]map[types.NodeID]*typesDKG.MPKReady),
		dkgFinals:       make(map[uint64]map[types.NodeID]*typesDKG.Finalize),
		dkgSuccesses:    make(map[uint64]map[types.NodeID]*typesDKG.Success),
		appliedRequests: make(map[common.Hash]struct{}),
	}
	// Nodes
	for nID, key := range s.nodes {
		copied.nodes[nID] = key
	}
	// DKG & CRS
	for round, complaintsForRound := range s.dkgComplaints {
		copied.dkgComplaints[round] =
			make(map[types.NodeID][]*typesDKG.Complaint)
		for nID, comps := range complaintsForRound {
			tmpComps := []*typesDKG.Complaint{}
			for _, comp := range comps {
				tmpComps = append(tmpComps, CloneDKGComplaint(comp))
			}
			copied.dkgComplaints[round][nID] = tmpComps
		}
	}
	for round, mKeysForRound := range s.dkgMasterPublicKeys {
		copied.dkgMasterPublicKeys[round] =
			make(map[types.NodeID]*typesDKG.MasterPublicKey)
		for nID, mKey := range mKeysForRound {
			copied.dkgMasterPublicKeys[round][nID] =
				CloneDKGMasterPublicKey(mKey)
		}
	}
	for round, readysForRound := range s.dkgReadys {
		copied.dkgReadys[round] = make(map[types.NodeID]*typesDKG.MPKReady)
		for nID, ready := range readysForRound {
			copied.dkgReadys[round][nID] = CloneDKGMPKReady(ready)
		}
	}
	for round, finalsForRound := range s.dkgFinals {
		copied.dkgFinals[round] = make(map[types.NodeID]*typesDKG.Finalize)
		for nID, final := range finalsForRound {
			copied.dkgFinals[round][nID] = CloneDKGFinalize(final)
		}
	}
	for round, successesForRound := range s.dkgSuccesses {
		copied.dkgSuccesses[round] = make(map[types.NodeID]*typesDKG.Success)
		for nID, success := range successesForRound {
			copied.dkgSuccesses[round][nID] = CloneDKGSuccess(success)
		}
	}
	for _, crs := range s.crs {
		copied.crs = append(copied.crs, crs)
	}
	copied.dkgResetCount = make(map[uint64]uint64, len(s.dkgResetCount))
	for round, count := range s.dkgResetCount {
		copied.dkgResetCount[round] = count
	}
	for hash := range s.appliedRequests {
		copied.appliedRequests[hash] = struct{}{}
	}
	// Pending Changes
	copied.ownRequests = make(map[common.Hash]*StateChangeRequest)
	for k, req := range s.ownRequests {
		copied.ownRequests[k] = req.Clone()
	}
	copied.globalRequests = make(map[common.Hash]*StateChangeRequest)
	for k, req := range s.globalRequests {
		copied.globalRequests[k] = req.Clone()
	}
	return
}

type reqByTime []*StateChangeRequest

func (req reqByTime) Len() int      { return len(req) }
func (req reqByTime) Swap(i, j int) { req[i], req[j] = req[j], req[i] }
func (req reqByTime) Less(i, j int) bool {
	return req[i].Timestamp < req[j].Timestamp
}

// Apply change requests, this function would also
// be called when we extract these request from delivered blocks.
func (s *State) Apply(reqsAsBytes []byte) (err error) {
	if len(reqsAsBytes) == 0 {
		return
	}
	// Try to unmarshal this byte stream into []*StateChangeRequest.
	reqs, err := s.unpackRequests(reqsAsBytes)
	if err != nil {
		return
	}
	sort.Sort(reqByTime(reqs))
	s.lock.Lock()
	defer s.lock.Unlock()
	for _, req := range reqs {
		s.logger.Debug("Apply Request", "req", req)
		// Remove this request from pending set once it's about to apply.
		delete(s.globalRequests, req.Hash)
		delete(s.ownRequests, req.Hash)
		if _, exist := s.appliedRequests[req.Hash]; exist {
			continue
		}
		if err = s.isValidRequest(req); err != nil {
			if err == ErrDuplicatedChange {
				err = nil
				continue
			}
			return
		}
		if err = s.applyRequest(req); err != nil {
			return
		}
		s.appliedRequests[req.Hash] = struct{}{}
	}
	return
}

// AddRequestsFromOthers add requests from others, they won't be packed by
// 'PackOwnRequests'.
func (s *State) AddRequestsFromOthers(reqsAsBytes []byte) (err error) {
	if s.local {
		err = ErrNotInRemoteMode
		return
	}
	reqs, err := s.unpackRequests(reqsAsBytes)
	if err != nil {
		return
	}
	s.lock.Lock()
	defer s.lock.Unlock()
	for _, req := range reqs {
		s.globalRequests[req.Hash] = req
	}
	return
}

// PackRequests pack all current pending requests, include those from others.
func (s *State) PackRequests() (b []byte, err error) {
	if s.local {
		// Convert own requests to global one for packing.
		if _, err = s.PackOwnRequests(); err != nil {
			return
		}
	}
	// Pack requests in global pool.
	packed := []*StateChangeRequest{}
	s.lock.Lock()
	defer s.lock.Unlock()
	for _, v := range s.globalRequests {
		s.logger.Debug("Pack Request", "req", v)
		packed = append(packed, v)
	}
	return rlp.EncodeToBytes(packed)
}

// PackOwnRequests pack current pending requests as byte slice, which
// could be sent as blocks' payload and unmarshall back to apply.
//
// Once a request is packed as own request, it would be turned into a normal
// pending request and won't be packed by this function. This would ensure
// each request broadcasted(gossip) once.
//
// This function is not required to call in local mode.
func (s *State) PackOwnRequests() (b []byte, err error) {
	packed := []*StateChangeRequest{}
	s.lock.Lock()
	defer s.lock.Unlock()
	for k, v := range s.ownRequests {
		delete(s.ownRequests, k)
		s.globalRequests[k] = v
		packed = append(packed, v)
	}
	if b, err = rlp.EncodeToBytes(packed); err != nil {
		return
	}
	return
}

// isValidRequest checks if this request is valid to proceed or not.
func (s *State) isValidRequest(req *StateChangeRequest) error {
	// NOTE: there would be no lock in this helper, callers should be
	//       responsible for acquiring appropriate lock.
	switch req.Type {
	case StateAddDKGMPKReady:
		ready := req.Payload.(*typesDKG.MPKReady)
		if ready.Reset != s.dkgResetCount[ready.Round] {
			return ErrChangeWontApply
		}
	case StateAddDKGFinal:
		final := req.Payload.(*typesDKG.Finalize)
		if final.Reset != s.dkgResetCount[final.Round] {
			return ErrChangeWontApply
		}
	case StateAddDKGSuccess:
		success := req.Payload.(*typesDKG.Success)
		if success.Reset != s.dkgResetCount[success.Round] {
			return ErrChangeWontApply
		}
	case StateAddDKGMasterPublicKey:
		mpk := req.Payload.(*typesDKG.MasterPublicKey)
		if mpk.Reset != s.dkgResetCount[mpk.Round] {
			return ErrChangeWontApply
		}
		// If we've received identical MPK, ignore it.
		mpkForRound, exists := s.dkgMasterPublicKeys[mpk.Round]
		if exists {
			if oldMpk, exists := mpkForRound[mpk.ProposerID]; exists {
				if !oldMpk.Equal(mpk) {
					return ErrDuplicatedChange
				}
				return ErrChangeWontApply
			}
		}
		// If we've received MPK from that proposer, we would ignore
		// its mpk.
		if _, exists := s.dkgReadys[mpk.Round][mpk.ProposerID]; exists {
			return ErrProposerMPKIsReady
		}
	case StateAddDKGComplaint:
		comp := req.Payload.(*typesDKG.Complaint)
		if comp.Reset != s.dkgResetCount[comp.Round] {
			return ErrChangeWontApply
		}
		// If we've received DKG final from that proposer, we would ignore
		// its complaint.
		if _, exists := s.dkgFinals[comp.Round][comp.ProposerID]; exists {
			return ErrProposerIsFinal
		}
		// If we've received identical complaint, ignore it.
		compForRound, exists := s.dkgComplaints[comp.Round]
		if !exists {
			break
		}
		comps, exists := compForRound[comp.ProposerID]
		if !exists {
			break
		}
		for _, tmpComp := range comps {
			if tmpComp == comp {
				return ErrDuplicatedChange
			}
		}
	case StateAddCRS:
		crsReq := req.Payload.(*crsAdditionRequest)
		if uint64(len(s.crs)) > crsReq.Round {
			if !s.crs[crsReq.Round].Equal(crsReq.CRS) {
				return ErrForkedCRS
			}
			return ErrDuplicatedChange
		} else if uint64(len(s.crs)) == crsReq.Round {
			return nil
		} else {
			return ErrMissingPreviousCRS
		}
	case StateResetDKG:
		newCRS := req.Payload.(common.Hash)
		if s.crs[len(s.crs)-1].Equal(newCRS) {
			return ErrDuplicatedChange
		}
		// TODO(mission): find a smart way to make sure the caller call request
		//                this change with correct resetCount.
	}
	return nil
}

// applyRequest applies a single StateChangeRequest.
func (s *State) applyRequest(req *StateChangeRequest) error {
	// NOTE: there would be no lock in this helper, callers should be
	//       responsible for acquiring appropriate lock.
	switch req.Type {
	case StateAddNode:
		pubKey, err := ecdsa.NewPublicKeyFromByteSlice(req.Payload.([]byte))
		if err != nil {
			return err
		}
		s.nodes[types.NewNodeID(pubKey)] = pubKey
	case StateAddCRS:
		crsRequest := req.Payload.(*crsAdditionRequest)
		if crsRequest.Round != uint64(len(s.crs)) {
			return ErrDuplicatedChange
		}
		s.crs = append(s.crs, crsRequest.CRS)
	case StateAddDKGComplaint:
		comp := req.Payload.(*typesDKG.Complaint)
		if _, exists := s.dkgComplaints[comp.Round]; !exists {
			s.dkgComplaints[comp.Round] = make(
				map[types.NodeID][]*typesDKG.Complaint)
		}
		s.dkgComplaints[comp.Round][comp.ProposerID] = append(
			s.dkgComplaints[comp.Round][comp.ProposerID], comp)
	case StateAddDKGMasterPublicKey:
		mKey := req.Payload.(*typesDKG.MasterPublicKey)
		if _, exists := s.dkgMasterPublicKeys[mKey.Round]; !exists {
			s.dkgMasterPublicKeys[mKey.Round] = make(
				map[types.NodeID]*typesDKG.MasterPublicKey)
		}
		s.dkgMasterPublicKeys[mKey.Round][mKey.ProposerID] = mKey
	case StateAddDKGMPKReady:
		ready := req.Payload.(*typesDKG.MPKReady)
		if _, exists := s.dkgReadys[ready.Round]; !exists {
			s.dkgReadys[ready.Round] = make(map[types.NodeID]*typesDKG.MPKReady)
		}
		s.dkgReadys[ready.Round][ready.ProposerID] = ready
	case StateAddDKGFinal:
		final := req.Payload.(*typesDKG.Finalize)
		if _, exists := s.dkgFinals[final.Round]; !exists {
			s.dkgFinals[final.Round] = make(map[types.NodeID]*typesDKG.Finalize)
		}
		s.dkgFinals[final.Round][final.ProposerID] = final
	case StateAddDKGSuccess:
		success := req.Payload.(*typesDKG.Success)
		if _, exists := s.dkgSuccesses[success.Round]; !exists {
			s.dkgSuccesses[success.Round] =
				make(map[types.NodeID]*typesDKG.Success)
		}
		s.dkgSuccesses[success.Round][success.ProposerID] = success
	case StateResetDKG:
		round := uint64(len(s.crs) - 1)
		s.crs[round] = req.Payload.(common.Hash)
		s.dkgResetCount[round]++
		delete(s.dkgMasterPublicKeys, round)
		delete(s.dkgReadys, round)
		delete(s.dkgComplaints, round)
		delete(s.dkgFinals, round)
		delete(s.dkgSuccesses, round)
	case StateChangeLambdaBA:
		s.lambdaBA = time.Duration(req.Payload.(uint64))
	case StateChangeLambdaDKG:
		s.lambdaDKG = time.Duration(req.Payload.(uint64))
	case StateChangeRoundLength:
		s.roundInterval = req.Payload.(uint64)
	case StateChangeMinBlockInterval:
		s.minBlockInterval = time.Duration(req.Payload.(uint64))
	case StateChangeNotarySetSize:
		s.notarySetSize = req.Payload.(uint32)
	default:
		return errors.New("you are definitely kidding me")
	}
	return nil
}

// ProposeCRS propose a new CRS for a specific round.
func (s *State) ProposeCRS(round uint64, crs common.Hash) (err error) {
	err = s.RequestChange(StateAddCRS, &crsAdditionRequest{
		Round: round,
		CRS:   crs,
	})
	return
}

// RequestChange submits a state change request.
func (s *State) RequestChange(
	t StateChangeType, payload interface{}) (err error) {
	s.logger.Info("Request Change to State", "type", t, "value", payload)
	// Patch input parameter's type.
	switch t {
	case StateAddNode:
		payload = payload.(crypto.PublicKey).Bytes()
	case StateChangeLambdaBA,
		StateChangeLambdaDKG,
		StateChangeMinBlockInterval:
		payload = uint64(payload.(time.Duration))
	// These cases for for type assertion, make sure callers pass expected types.
	case StateAddCRS:
		payload = payload.(*crsAdditionRequest)
	case StateAddDKGMPKReady:
		payload = payload.(*typesDKG.MPKReady)
	case StateAddDKGFinal:
		payload = payload.(*typesDKG.Finalize)
	case StateAddDKGSuccess:
		payload = payload.(*typesDKG.Success)
	case StateAddDKGMasterPublicKey:
		payload = payload.(*typesDKG.MasterPublicKey)
	case StateAddDKGComplaint:
		payload = payload.(*typesDKG.Complaint)
	case StateResetDKG:
		payload = payload.(common.Hash)
	}
	req := NewStateChangeRequest(t, payload)
	s.lock.Lock()
	defer s.lock.Unlock()
	if s.local {
		if err = s.isValidRequest(req); err != nil {
			return
		}
		err = s.applyRequest(req)
	} else {
		if err = s.isValidRequest(req); err != nil {
			return
		}
		s.ownRequests[req.Hash] = req
	}
	return
}

// CRS access crs proposed for that round.
func (s *State) CRS(round uint64) common.Hash {
	s.lock.RLock()
	defer s.lock.RUnlock()
	if round >= uint64(len(s.crs)) {
		return common.Hash{}
	}
	return s.crs[round]
}

// DKGComplaints access current received dkg complaints for that round.
// This information won't be snapshot, thus can't be cached in test.Governance.
func (s *State) DKGComplaints(round uint64) []*typesDKG.Complaint {
	s.lock.RLock()
	defer s.lock.RUnlock()
	comps, exists := s.dkgComplaints[round]
	if !exists {
		return nil
	}
	tmpComps := make([]*typesDKG.Complaint, 0, len(comps))
	for _, compProp := range comps {
		for _, comp := range compProp {
			tmpComps = append(tmpComps, CloneDKGComplaint(comp))
		}
	}
	return tmpComps
}

// DKGMasterPublicKeys access current received dkg master public keys for that
// round. This information won't be snapshot, thus can't be cached in
// test.Governance.
func (s *State) DKGMasterPublicKeys(round uint64) []*typesDKG.MasterPublicKey {
	s.lock.RLock()
	defer s.lock.RUnlock()
	masterPublicKeys, exists := s.dkgMasterPublicKeys[round]
	if !exists {
		return nil
	}
	mpks := make([]*typesDKG.MasterPublicKey, 0, len(masterPublicKeys))
	for _, mpk := range masterPublicKeys {
		mpks = append(mpks, CloneDKGMasterPublicKey(mpk))
	}
	return mpks
}

// IsDKGMPKReady checks if current received dkg readys exceeds threshold.
// This information won't be snapshot, thus can't be cached in test.Governance.
func (s *State) IsDKGMPKReady(round uint64, threshold int) bool {
	s.lock.RLock()
	defer s.lock.RUnlock()
	return len(s.dkgReadys[round]) >= threshold
}

// IsDKGFinal checks if current received dkg finals exceeds threshold.
// This information won't be snapshot, thus can't be cached in test.Governance.
func (s *State) IsDKGFinal(round uint64, threshold int) bool {
	s.lock.RLock()
	defer s.lock.RUnlock()
	return len(s.dkgFinals[round]) >= threshold
}

// IsDKGSuccess checks if current received dkg successes exceeds threshold.
// This information won't be snapshot, thus can't be cached in test.Governance.
func (s *State) IsDKGSuccess(round uint64, threshold int) bool {
	s.lock.RLock()
	defer s.lock.RUnlock()
	return len(s.dkgSuccesses[round]) >= threshold
}

// DKGResetCount returns the reset count for DKG of given round.
func (s *State) DKGResetCount(round uint64) uint64 {
	s.lock.RLock()
	defer s.lock.RUnlock()
	return s.dkgResetCount[round]
}