path: root/internal/quorum/quorum.go

package quorum

import (
	"context"
	"fmt"
	"log"
	"os"
	"sort"
	"strconv"
	"strings"
	"time"

	"codeberg.org/snonux/gorum/internal/config"
	"codeberg.org/snonux/gorum/internal/vote"
)

type Quorum struct {
	conf config.Config
	// My own vote, sending to the partners
	myVote vote.Vote

	// From partners received votes
	voteCh chan vote.Vote
	votes  map[string]vote.Vote
}

type Score struct {
	ID    string
	Value int
}

func (s Score) Is(other Score) bool {
	return s.ID == other.ID && s.Value == other.Value
}

type Scores []Score

func (s Scores) Winner() (Score, error) {
	if len(s) == 0 {
		return Score{}, fmt.Errorf("emtpy score, no winner")
	}

	return s[0], nil
}

func New(conf config.Config) Quorum {
	return Quorum{
		conf:   conf,
		votes:  make(map[string]vote.Vote),
		voteCh: make(chan vote.Vote),
	}
}

func (quo Quorum) Start(ctx context.Context) <-chan vote.Vote {
	ch := make(chan vote.Vote)

	interval := time.Second * time.Duration(quo.conf.LoopIntervalS)
	if vote.Expiry <= interval {
		log.Fatal("quorum: LoopIntervalS ", quo.conf.LoopIntervalS,
			" should be less than the vote expiry of ", vote.Expiry)
	}

	go func() {
		defer close(ch)

		var (
			myVote  vote.Vote
			changed bool
		)

		for {
			select {
			case <-time.After(interval):
				myVote, _ = quo.makeMyVote()
				log.Println("quorum: made my vote:", myVote)
				ch <- myVote
			case v := <-quo.voteCh:
				quo.vote(v)
				if myVote, changed = quo.makeMyVote(); changed {
					log.Println("quorum: changed my vote:", myVote)
					ch <- myVote
				}
			case <-ctx.Done():
				return
			}

			if err := quo.persist(changed); err != nil {
				log.Println("quorum:", err)
				notifyError(quo.conf, err)
			}
		}
	}()

	return ch
}

func (quo Quorum) Vote(v vote.Vote) {
	log.Printf("quorum: queing vote %v", v)
	quo.voteCh <- v
}

func (quo Quorum) vote(v vote.Vote) {
	log.Printf("quorum: adding vote %v", v)
	quo.votes[v.FromID] = v
}

func (quo Quorum) scores() (scores Scores) {
	scoreMap := make(map[string]int)

	for _, vote := range quo.votes {
		if vote.Expired() {
			continue
		}
		for _, id := range vote.IDs {
			score := scoreMap[id]
			nodeNumber := quo.conf.NodeNumber(id)
			scoreMap[id] = score + 100 + (len(quo.conf.Nodes) - nodeNumber)
		}
	}

	for id, score_ := range scoreMap {
		scores = append(scores, Score{id, score_})
	}

	sort.Slice(scores, func(i, j int) bool {
		return scores[i].Value > scores[j].Value
	})

	return
}

func (quo *Quorum) strs() (string, string) {
	scores := quo.scores()
	log.Println("quorum scores:", scores)
	winner, err := scores.Winner()
	if err != nil {
		log.Println("the winner is", winner.ID)
	}

	var sb strings.Builder
	winnerStr := ""

	for i, score := range scores {
		sb.WriteString("At position ")
		sb.WriteString(strconv.Itoa(i + 1))

		if score.ID == quo.conf.MyID {
			sb.WriteString(" is current node ")
			if score.Is(winner) {
				winnerStr = fmt.Sprintf("Yeah, I, %s, am the winner!", winner.ID)
			}
		} else {
			sb.WriteString(" is partner node ")
		}
		sb.WriteString(score.ID)

		sb.WriteString(" (node #")
		sb.WriteString(strconv.Itoa(quo.conf.NodeNumber(score.ID)))

		sb.WriteString(") with total score of ")
		sb.WriteString(strconv.Itoa(score.Value))

		sb.WriteString("\n")
	}

	return winnerStr, sb.String()
}

func (quo *Quorum) persist(changed bool) error {
	winnerStr, scoresStr := quo.strs()

	if changed {
		if err := notify(quo.conf, "GORUM: Quorum changed", scoresStr); err != nil {
			return err
		}
	}

	if _, err := os.Stat(quo.conf.StateDir); os.IsNotExist(err) {
		if err := os.MkdirAll(quo.conf.StateDir, 0755); err != nil {
			return err
		}
	}

	if err := quo.persistWinner(winnerStr); err != nil {
		return err
	}

	return writeFileViaTmp(fmt.Sprintf("%s/%s", quo.conf.StateDir, quo.conf.ScoreFile), scoresStr)
}

func (quo *Quorum) persistWinner(winnerStr string) error {
	winnerFile := fmt.Sprintf("%s/%s", quo.conf.StateDir, quo.conf.WinnerFile)
	if winnerStr == "" {
		if _, err := os.Stat(winnerFile); err != nil {
			return os.Remove(winnerFile)
		}
		return nil
	}
	return writeFileViaTmp(winnerFile, winnerStr)
}

func (quo *Quorum) makeMyVote() (vote.Vote, bool) {
	newVote, err := quo.expireOldVotes()
	if err != nil {
		log.Println("quorum:", err)
		return quo.myVote, false
	}

	if quo.myVote.Equal(newVote) {
		return quo.myVote, false
	}

	quo.myVote = newVote
	return quo.myVote, true
}

func (quo Quorum) expireOldVotes() (vote.Vote, error) {
	var expired []string
	var live []string

	for fromNode, vote := range quo.votes {
		if vote.Expired() {
			log.Println("quorum: vote", vote, "from node", fromNode, "expired!")
			expired = append(expired, fromNode)
			continue
		}
		live = append(live, fromNode)
	}

	for _, e := range expired {
		delete(quo.votes, e)
	}

	return vote.New(quo.conf, live...)
}

// Create tmp file first, and then, once written, rename it.
func writeFileViaTmp(filePath, content string) error {
	tmpFilePath := fmt.Sprintf("%s.tmp", filePath)

	fd, err := os.Create(tmpFilePath)
	if err != nil {
		return err
	}
	defer fd.Close()

	if _, err := fd.WriteString(content); err != nil {
		return err
	}

	if err := fd.Sync(); err != nil {
		return err
	}

	return os.Rename(tmpFilePath, filePath)
}