package backend

import (
	"crypto/ecdsa"
	"errors"
	"math"
	"reflect"
	"sync"
	"sync/atomic"
	"time"

	"github.com/autonity/autonity/consensus/tendermint/helpers"

	ring "github.com/zfjagann/golang-ring"

	"github.com/autonity/autonity/accounts/abi"
	"github.com/autonity/autonity/common"
	"github.com/autonity/autonity/common/fixsizecache"
	"github.com/autonity/autonity/consensus"
	tendermintCore "github.com/autonity/autonity/consensus/tendermint/core"
	"github.com/autonity/autonity/consensus/tendermint/core/constants"
	"github.com/autonity/autonity/consensus/tendermint/core/interfaces"
	"github.com/autonity/autonity/consensus/tendermint/core/message"
	"github.com/autonity/autonity/consensus/tendermint/events"
	"github.com/autonity/autonity/consensus/tendermint/router"
	"github.com/autonity/autonity/core"
	"github.com/autonity/autonity/core/types"
	"github.com/autonity/autonity/core/vm"
	"github.com/autonity/autonity/crypto"
	"github.com/autonity/autonity/crypto/blst"
	"github.com/autonity/autonity/ethdb"
	"github.com/autonity/autonity/event"
	"github.com/autonity/autonity/log"
)

const (
	// fetcherID is the ID indicates the block is from BFT engine
	fetcherID = "tendermint"
	// ring buffer to be able to handle at maximum 10 rounds, 100 committee and 3 messages types
	ringCapacity = 10 * 100 * 3
	// maximum number of future height messages
	maxFutureMsgs = 10 * 100 * 3
	// number of buckets to allocate in the fixed cache
	numBuckets = 5987
	// max number of entries in each packet
	numEntries = 10
	// message
	aggregatorMessageQueue = 5000
)

var (
	// ErrStoppedEngine is returned if the engine is stopped
	ErrStoppedEngine = errors.New("stopped engine")
)

// New creates an Ethereum Backend for BFT core engine.
func New(
	database ethdb.Database,
	nodeKey *ecdsa.PrivateKey,
	consensusKey blst.SecretKey,
	vmConfig *vm.Config,
	services *interfaces.Services,
	evMux *event.TypeMux,
	ms *tendermintCore.MsgStore,
	afdDispatchCh chan<- events.MessageEventer,
	log log.Logger,
) *Backend {

	knownMessages := fixsizecache.New[common.Hash, bool](numBuckets, numEntries, fixsizecache.HashKey[common.Hash])

	backend := &Backend{
		database:            database,
		eventMux:            event.NewTypeMuxSilent(evMux, log),
		nodeKey:             nodeKey,
		consensusKey:        consensusKey,
		address:             crypto.PubkeyToAddress(nodeKey.PublicKey),
		logger:              log,
		knownMessages:       knownMessages,
		vmConfig:            vmConfig,
		MsgStore:            ms, //TODO: we use this only in tests, to easily reach the msg store when having a reference to the backend. It would be better to just have the `accountability` module as a part of the backend object.
		askSyncRateLimiter:  helpers.NewTimeWindowLimiter(constants.AskSyncInterval, 2),
		aggregatorMessageCh: make(chan events.UnverifiedMessageEvent, aggregatorMessageQueue),
		jailingCh:           make(chan common.Address, 1000),
		afdDispatchCh:       afdDispatchCh, // to FD
		jailed: jailed{
			validators: make(map[common.Address]uint64),
		},
		future: future{
			messages:  make(map[uint64][]*events.UnverifiedMessageEvent),
			minHeight: math.MaxUint64,
		},
	}

	backend.pendingMessages.SetCapacity(ringCapacity)

	backend.router = router.Setup(nodeKey, backend.address, backend.logger)

	backend.gossiper = NewGossiper(
		backend.knownMessages,
		backend.address,
		backend.logger,
		backend.stopped,
		backend.router,
	)
	// apply custom services if needed (used for tests)
	if services != nil {
		backend.gossiper = services.Gossiper(backend)
		backend.router.SetPinger(services.Pinger(backend.router))
		backend.router.SetSelector(services.Selector(backend.router))
	}

	consensusCore := tendermintCore.New(backend, services, backend.address, log)
	backend.core = consensusCore
	backend.coreEventDispatcher = consensusCore

	backend.aggregator = newAggregator(backend, consensusCore, log)

	return backend
}

// ----------------------------------------------------------------------------

type Backend struct {
	eventMux     *event.TypeMuxSilent
	nodeKey      *ecdsa.PrivateKey
	consensusKey blst.SecretKey
	address      common.Address
	logger       log.Logger
	blockchain   *core.BlockChain
	database     ethdb.Database
	currentBlock func() *types.Block
	hasBadBlock  func(hash common.Hash) bool

	// the channels for tendermint engine notifications
	proposalVerifiedCh  chan<- *types.Block
	commitCh            chan<- *types.Block
	aggregatorMessageCh chan events.UnverifiedMessageEvent // to send events to the aggregator
	afdDispatchCh       chan<- events.MessageEventer       // to send events to the fault detector
	jailingCh           chan common.Address
	proposedBlockHash   common.Hash
	coreStarting        atomic.Bool
	coreRunning         atomic.Bool
	core                interfaces.Core
	coreEventDispatcher interfaces.EventDispatcher
	stopped             chan struct{}
	wg                  sync.WaitGroup

	// used to save consensus messages while core is stopped
	pendingMessages ring.Ring

	// interface to find peers
	Broadcaster consensus.Broadcaster
	// interface to enqueue blocks to fetcher
	Enqueuer consensus.Enqueuer
	// interface to gossip consensus messages
	gossiper interfaces.Gossiper

	knownMessages *fixsizecache.Cache[common.Hash, bool] // the cache of self messages
	vmConfig      *vm.Config

	// MsgStore contains recent consensus messages for accountability and peer state recovery.
	MsgStore           *tendermintCore.MsgStore
	askSyncRateLimiter *helpers.TimeWindowLimiter
	cleanupTicker      *time.Ticker

	router interfaces.Router

	aggregator *aggregator

	jailed jailed // metadata for p2p jailed validators
	future future // buffer for future height events and related metadata
}

type future struct {
	messages            map[uint64][]*events.UnverifiedMessageEvent // UnverifiedMessageEvent is used slightly inappropriately here, as the messages height messages still need to pass the checks in `handleDecodedMsg` before being posted to the aggregator.
	minHeight           uint64                                      // lowest future height buffered in the message store
	maxHeight           uint64                                      // highest future height buffered in the message store
	size                uint64                                      // number of messages in the message store
	lastProcessedHeight uint64                                      // last future height processed by Core. It is needed to avoid race condition which could delay a message.
	sync.RWMutex
}

type jailed struct {
	validators map[common.Address]uint64 // maps validator address to the epoch in which he was jailed in
	sync.RWMutex
}

func (sb *Backend) BlockChain() *core.BlockChain {
	return sb.blockchain
}

func (sb *Backend) EpochByHeight(height uint64) (*types.EpochInfo, error) {
	return sb.BlockChain().EpochByHeight(height)
}

func (sb *Backend) CommitteeByHeight(height uint64) (*types.Committee, error) {
	return sb.BlockChain().CommitteeByHeight(height)
}

func (sb *Backend) MinNonExpiredHeight(coreHeight uint64) (uint64, error) {
	heightRange, err := sb.blockchain.AccountabilityParamsByHeight(coreHeight)
	if err != nil {
		return 0, err
	}
	return helpers.MinNonExpiredHeight(coreHeight, heightRange.Range.Uint64()), nil
}

func (sb *Backend) MessageCh() <-chan events.UnverifiedMessageEvent {
	return sb.aggregatorMessageCh
}

// Address implements tendermint.Backend.Address
func (sb *Backend) Address() common.Address {
	return sb.address
}

// Broadcast implements tendermint.Backend.Broadcast
func (sb *Backend) Broadcast(committee *types.Committee, message message.Msg) {
	// send to self (directly to Core and FD, no need to verify local messages)
	go sb.gossiper.Gossip(committee, message, sb.Address())
	// a goroutine is required here to avoid creating a deadlock, broadcast can be called from the messageEventHandler itself
	go sb.DispatchToCore(events.NewMessageEvent(message, nil, sb.Address(), time.Now(), true)) // core
	sb.DispatchToFD(events.NewMessageEvent(message, nil, sb.Address(), time.Now(), true))      // FD
}

func (sb *Backend) AskSync(committee *types.Committee, syncMsg *message.AskSyncMsg) error {
	clusteringThreshold := router.DefaultScaleThresholdForClustering
	if threshold, err := sb.blockchain.ClusteringThresholdByHeight(syncMsg.Height); err == nil {
		clusteringThreshold = threshold.Int64()
	} else {
		sb.logger.Error("Failed to get clustering threshold, using default", "err", err, "height", syncMsg.Height)
	}
	return sb.gossiper.AskSync(committee, syncMsg, clusteringThreshold)
}

// Gossip implements tendermint.Backend.Gossip
func (sb *Backend) Gossip(committee *types.Committee, msg message.Msg, sender common.Address) {
	sb.gossiper.Gossip(committee, msg, sender)
}

func (sb *Backend) SlowGossip(committee *types.Committee, msg message.Msg, sender common.Address) {
	clusteringThreshold := router.DefaultScaleThresholdForClustering
	if threshold, err := sb.blockchain.ClusteringThresholdByHeight(msg.H()); err == nil {
		clusteringThreshold = threshold.Int64()
	} else {
		sb.logger.Error("Failed to get clustering threshold, using default", "err", err, "height", msg.H())
	}
	sb.gossiper.SlowGossip(committee, msg, sender, clusteringThreshold)
}

// UpdateStopChannel implements tendermint.Backend.Gossip
func (sb *Backend) UpdateStopChannel(stopCh chan struct{}) {
	sb.gossiper.UpdateStopChannel(stopCh)
}

// KnownMsgHash dumps the known messages in case of gossiping.
func (sb *Backend) KnownMsgHash() []common.Hash {
	return sb.knownMessages.Keys()
}

func (sb *Backend) Logger() log.Logger {
	return sb.logger
}

func (sb *Backend) Gossiper() interfaces.Gossiper {
	return sb.gossiper
}

func (sb *Backend) Router() interfaces.Router {
	return sb.router
}

// Commit implements tendermint.Backend.Commit
func (sb *Backend) Commit(proposal *types.Block, round int64, quorumCertificate *types.AggregateSignature) error {
	h := proposal.Header()

	// update block header with quorum certificate and commit round
	h.QuorumCertificate = quorumCertificate
	h.Round = uint64(round)
	proposal = proposal.WithSeal(h)
	sb.logger.Info("Quorum of Precommits received", "proposal", proposal.Hash(), "round", round, "height", proposal.Number().Uint64())

	// - if the proposed and committed blocks are the same, send the proposed hash
	//   to resultCh channel, which is being watched inside the worker.ResultLoop() function.
	// - otherwise, we try to insert the block.
	// -- if success, the ChainHeadEvent event will be broadcasted, try to build
	//    the next block and the previous Seal() will be stopped.
	// -- otherwise, a error will be returned and a round change event will be fired.
	if sb.proposedBlockHash == proposal.Hash() && !sb.isResultChanNil() {
		sb.sendResultChan(proposal)
		return nil
	}

	if sb.Enqueuer != nil {
		sb.Enqueuer.Enqueue(fetcherID, proposal)
	}
	return nil
}

func (sb *Backend) Post(ev any) {
	switch msg := ev.(type) {
	case events.CommitEvent:
		sb.coreEventDispatcher.Post(msg)
	case events.NewCandidateBlockEvent:
		sb.coreEventDispatcher.Post(msg)
	case events.UnverifiedMessageEvent:
		sb.aggregatorMessageCh <- msg
	default:
		sb.eventMux.Post(msg)
	}
}

func (sb *Backend) DispatchToCore(ev any) {
	switch ev := ev.(type) {
	case events.MessageEvent:
		sb.coreEventDispatcher.Post(ev)
	case events.OldMessageEvent:
		// ignore old height messages for Core
	default:
		panic("unknown event " + reflect.TypeOf(ev).String())
	}
	return
}

func (sb *Backend) DispatchToFD(ev any) {
	switch ev := ev.(type) {
	case events.MessageEventer:
		sb.afdDispatchCh <- ev
	}
	return
}

func (sb *Backend) Subscribe(types ...any) *event.TypeMuxSubscription {
	return sb.eventMux.Subscribe(types...)
}

// VerifyProposal implements tendermint.Backend.VerifyProposal and verify if the proposal is valid
func (sb *Backend) VerifyProposal(proposalBlock *types.Block) (time.Duration, error) {
	// TODO: fix always false statement and check for non nil
	// TODO: use interface instead of type

	if sb.HasBadProposal(proposalBlock.Hash()) {
		return 0, core.ErrBannedHash
	}

	// verify if the proposal block is already included in the node's local chain.
	// This scenario can happen when we are processing a proposal, but in the meantime other peers already reached quorum on it,
	// therefore we already received the finalized block through p2p block propagation.
	// NOTE: this function execution is not atomic, the block could be not included at the time of this check
	// and become included right after we passed the check.
	if sb.blockchain.HasHeader(proposalBlock.Hash(), proposalBlock.NumberU64()) {
		return 0, constants.ErrAlreadyHaveBlock
	}

	// verify the header of proposed proposal
	err := sb.VerifyHeader(sb.blockchain, proposalBlock.Header(), false)
	// ignore errEmptyQuorumCertificate error because we don't have the quorum certificate yet
	if err == nil || errors.Is(err, errEmptyQuorumCertificate) {
		var (
			header         = proposalBlock.Header()
			proposalNumber = header.Number.Uint64()
			parent         = sb.blockchain.GetBlock(proposalBlock.ParentHash(), proposalBlock.NumberU64()-1)
		)

		// We need to process all the transaction to get the latest state to get the latest committee
		state, stateErr := sb.blockchain.StateAt(parent.Root())
		if stateErr != nil {
			return 0, stateErr
		}

		// Validate the body of the proposal
		if err = sb.blockchain.Validator().ValidateBody(proposalBlock); err != nil {
			return 0, err
		}

		receipts, _, usedGas, epochInfo, contractsConfig, err := sb.blockchain.Processor().Process(proposalBlock, state, *sb.vmConfig)
		if err != nil {
			sb.logger.Error("state processing failed", "error", err, "height", proposalNumber)
			return 0, err
		}
		//Validate the state of the proposal
		if err = sb.blockchain.Validator().ValidateState(proposalBlock, state, receipts, usedGas); err != nil {
			sb.logger.Error("proposal proposed, bad root state", "error", err)
			return 0, err
		}
		// As the epoch infos(committee, lastEpochBlock, nextEpochBlock) are saving in the contract state, thus the
		// consistency of them are checked by the state validation, moreover that since the epoch info in block header
		// is a factor to compute the block hash, thus they are immutable in the hash chain once there are quorum
		// certificates to finalize the block. Thus, the epoch boundary checking is not required anymore, however
		// the epoch info in the proposal's header should be equal to the epoch info dumped from the contract state.
		if !proposalBlock.Header().Epoch.Equal(epochInfo) {
			sb.logger.Error("inconsistent epoch info",
				"currentVerifier", sb.address.String(),
				"proposalNumber", proposalNumber,
				"headerEpoch", header.Epoch,
				"computedEpoch", epochInfo,
			)
			return 0, consensus.ErrInconsistentEpochInfo
		}

		// cache verified proposal state
		sb.blockchain.CacheProposalState(proposalBlock.Hash(), receipts, usedGas, state, contractsConfig)

		return 0, nil
	} else if errors.Is(err, consensus.ErrFutureTimestampBlock) {
		return time.Unix(int64(proposalBlock.Header().Time), 0).Sub(now()), consensus.ErrFutureTimestampBlock
	}

	// Here we are considering this proposal invalid because we pruned the parent's state
	// however this is our local node fault, not the remote proposer fault.
	if errors.Is(err, consensus.ErrPrunedAncestor) {
		sb.logger.Error("Rejecting a proposal because local node has pruned parent's state")
		sb.logger.Error("Please check your pruning settings")
	}
	return 0, err
}

// Sign implements tendermint.Backend.Sign
func (sb *Backend) Sign(data common.Hash) blst.Signature {
	signature := sb.consensusKey.Sign(data[:])
	return signature
}

func (sb *Backend) HeadBlock() *types.Block {
	return sb.currentBlock()
}

func (sb *Backend) HasBadProposal(hash common.Hash) bool {
	if sb.hasBadBlock == nil {
		return false
	}
	return sb.hasBadBlock(hash)
}

func (sb *Backend) GetContractABI() *abi.ABI {
	// after the contract is upgradable, call it from contract object rather than from conf.
	return sb.blockchain.ProtocolContracts().ABI()
}

func (sb *Backend) CoreState() interfaces.CoreState {
	return sb.core.CoreState()
}

// CommitteeEnodes retrieve the list of validators enodes for the current block
func (sb *Backend) CommitteeEnodes() []string {
	header := sb.blockchain.CurrentBlock().Header()
	stateDB, err := sb.blockchain.StateAt(header.Root)
	if err != nil {
		sb.logger.Error("Failed to get state", "err", err, "height", header.Number.Uint64())
		return nil
	}
	enodes, err := sb.blockchain.ProtocolContracts().CallGetCommitteeEnodes(stateDB, header, false)
	if err != nil {
		sb.logger.Error("Failed to get block committee", "err", err)
		return nil
	}
	return enodes.StrList
}

// called by tendermint core to dump core state
func (sb *Backend) FutureMsgs() []message.Msg {
	sb.future.RLock()
	defer sb.future.RUnlock()

	var msgs []message.Msg
	for _, evs := range sb.future.messages {
		for _, ev := range evs {
			msgs = append(msgs, ev.Message)
		}
	}

	return msgs
}
