package distworker

import (
	"sync"
	"time"
)

// consensusDecision is the outcome of one consensus round. The
// selfcheck module drives its alert state machine off Down/Up; NoQuorum
// is the "we don't know yet" verdict returned when too few voters
// reported for either side to be a majority.
type consensusDecision int

const (
	// consensusNoQuorum means we have fewer than the minimum voters
	// required to reach a majority. The selfcheck holds the
	// previous state instead of flipping.
	consensusNoQuorum consensusDecision = iota
	// consensusUp means a majority of (self + known peers) reported
	// the master API as up.
	consensusUp
	// consensusDown means a majority reported the master API as
	// down. The selfcheck starts (or continues) the down timer.
	consensusDown
)

// String makes the verdict easy to log without bespoke formatting.
func (d consensusDecision) String() string {
	switch d {
	case consensusUp:
		return "up"
	case consensusDown:
		return "down"
	default:
		return "no-quorum"
	}
}

// tallyConsensus counts how many of self+peers are up vs down. The
// returned sizes are useful for logging/debugging and let the
// selfcheck report "2 of 3 voters say down" in its log line.
//
// selfUp == nil means "self vote is unknown" (e.g. the first tick
// has not completed). The cluster size is the number of known votes
// (self, if known, plus each peer observation).
func tallyConsensus(selfUp *bool, peers []peerObservation) (up, down, total int) {
	if selfUp != nil {
		total++
		if *selfUp {
			up++
		} else {
			down++
		}
	}
	for i := range peers {
		total++
		if peers[i].Up {
			up++
		} else {
			down++
		}
	}
	return up, down, total
}

// decideConsensus applies the simple-majority rule: whichever side
// (up or down) has at least floor(total/2)+1 votes wins. When total
// is 0 (no votes at all) or neither side reaches that threshold the
// result is consensusNoQuorum.
//
// minVotes is the minimum number of fresh votes (self + peers) that
// must be present before any verdict is reported. It is the gate
// that turns simple-majority into a peer-backed cluster quorum: a
// multi-worker deployment (WorkerInit.Peers non-empty) passes
// minVotes=2 so a lone self vote — no fresh peer observations yet —
// is consensusNoQuorum, both for up and down. A single-worker /
// no-peer deployment passes minVotes=1 to preserve the prior
// single-node behavior. Values below 1 are clamped to 1.
//
// The "self" vote is required to reach quorum: a worker that has not
// produced its own first probe cannot make a down consensus call
// (its own vote would be missing). Pass nil for selfUp to model the
// pre-first-probe window.
func decideConsensus(selfUp *bool, peers []peerObservation, minVotes int) consensusDecision {
	if selfUp == nil {
		return consensusNoQuorum
	}
	if minVotes < 1 {
		minVotes = 1
	}
	up, down, total := tallyConsensus(selfUp, peers)
	if total < minVotes {
		return consensusNoQuorum
	}
	// Standard majority for a non-empty set: floor(total/2)+1.
	// For 1 voter (self only, no peers yet) this collapses to 1
	// which still requires unanimous agreement with self.
	majority := total/2 + 1
	if up >= majority {
		return consensusUp
	}
	if down >= majority {
		return consensusDown
	}
	return consensusNoQuorum
}

// consensusState tracks the cluster-level up/down verdict over time
// so the selfcheck module can fire alerts only after the verdict has
// held for selfcheckConsensusWait. This mirrors the "incident
// state machine" idea from
// docs/distributed/worker-to-worker-raft.md §9.1 in miniature: we
// only have two states (down / clear) and one wait threshold, but
// the structure is the same so the next slice can swap the rule for
// the full FSM without changing the alert site.
type consensusState struct {
	mu sync.Mutex
	// downSince records the wall-clock time the cluster verdict
	// first flipped to consensusDown. Cleared when the verdict
	// flips back to consensusUp. nil means "not currently down".
	downSince *time.Time
	// alertActive mirrors the prior selfcheckState.sent* flags. It
	// is true between the down-alert firing and the recovery alert
	// firing so a duplicate probe does not re-send the same
	// "master is down" notification.
	alertActive bool
	// lastVerdict keeps the most recent decision for the next tick
	// to compare against without recomputing from scratch.
	lastVerdict consensusDecision
	// notificationLeader is the last deterministic worker elected to
	// send system-contact notifications for this local view of the
	// cluster. The first non-empty leader only initializes the field;
	// later changes are alert-worthy.
	notificationLeader string
}

func (s *consensusState) notificationLeaderChanged(leader string) (old string, changed bool) {
	if s == nil || leader == "" {
		return "", false
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if s.notificationLeader == "" {
		s.notificationLeader = leader
		return "", false
	}
	if s.notificationLeader == leader {
		return "", false
	}
	old = s.notificationLeader
	s.notificationLeader = leader
	return old, true
}

// isDownConsensusHeld reports whether the supplied verdict means
// "down for at least selfcheckConsensusWait" given the current
// state. A fresh down verdict sets downSince; a subsequent down
// verdict keeps the original timestamp so the wait is measured from
// the first observation, not the most recent.
func (s *consensusState) isDownConsensusHeld(verdict consensusDecision, now time.Time) bool {
	if s == nil {
		return false
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	switch verdict {
	case consensusDown:
		if s.downSince == nil {
			t := now
			s.downSince = &t
		}
		s.lastVerdict = verdict
		return now.Sub(*s.downSince) >= selfcheckConsensusWait
	case consensusUp:
		s.downSince = nil
		s.alertActive = false
		s.lastVerdict = verdict
		return false
	default:
		// NoQuorum: hold the existing state. Do not reset
		// downSince (a transient blip should not extend the
		// timer, but it should not erase progress either).
		return false
	}
}

// markDownAlertFired records that the down alert has been emitted so
// the next tick does not fire it again. Idempotent.
func (s *consensusState) markDownAlertFired() {
	if s == nil {
		return
	}
	s.mu.Lock()
	s.alertActive = true
	s.mu.Unlock()
}

// shouldFireRecovery reports whether the cluster has been up long
// enough to fire a recovery alert. Recovery uses the same wait
// window as the down alert so a flapping verdict does not spam
// recovery notifications.
func (s *consensusState) shouldFireRecovery(verdict consensusDecision, now time.Time) bool {
	if s == nil {
		return false
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	if verdict != consensusUp {
		return false
	}
	if !s.alertActive {
		return false
	}
	if s.lastVerdict != consensusUp {
		// Just flipped from down to up; stamp the recovery timer.
		s.lastVerdict = verdict
		t := now
		s.downSince = &t
		return false
	}
	if s.downSince == nil {
		return false
	}
	return now.Sub(*s.downSince) >= selfcheckConsensusWait
}

// markRecoveryFired clears the alert-active flag so a future
// down verdict can fire the down alert again.
func (s *consensusState) markRecoveryFired() {
	if s == nil {
		return
	}
	s.mu.Lock()
	s.alertActive = false
	s.downSince = nil
	s.mu.Unlock()
}