feat: add event for voting finished (#82)

* feat: add event for voting finished

* chore: fixed linting

* chore: add marshal helper and tests

* chore: add state manager test

* chore: fixed linting

assets/valid-state.json

@@ -0,0 +1,8 @@
+{
+  "ChainInfos": {
+    "cosmos": {
+      "Chain": {},
+      "ProposalVotes": {}
+    }
+  }
+}

pkg/events/finished_voting.go

@@ -0,0 +1,18 @@
+package events
+
+import (
+	"main/pkg/types"
+)
+
+type FinishedVotingEvent struct {
+	Chain    *types.Chain
+	Proposal types.Proposal
+}
+
+func (e FinishedVotingEvent) Name() string {
+	return "finished_voting"
+}
+
+func (e FinishedVotingEvent) IsAlert() bool {
+	return false
+}

pkg/mutes/manager.go

@@ -3,6 +3,7 @@ package mutesmanager
 import (
 	"encoding/json"
 	"main/pkg/fs"
+	"main/pkg/utils"
 
 	"github.com/rs/zerolog"
 )
@@ -50,13 +51,9 @@ func (m *Manager) Save() {
 		return
 	}
 
-	content, err := json.Marshal(m.Mutes)
-	if err != nil {
-		m.Logger.Warn().Err(err).Msg("Could not marshal mutes")
-		return
-	}
+	content := utils.MustMarshal(m.Mutes)
 
-	if err = m.Filesystem.WriteFile(m.MutesPath, content, 0o600); err != nil {
+	if err := m.Filesystem.WriteFile(m.MutesPath, content, 0o600); err != nil {
 		m.Logger.Warn().Err(err).Msg("Could not save mutes")
 		return
 	}

pkg/report/generator.go

@@ -46,24 +46,53 @@ func (g *Generator) GenerateReport(oldState, newState state.State) reporters.Rep
 		}
 
 		for proposalID, proposalVotes := range chainInfo.ProposalVotes {
-			if !proposalVotes.Proposal.IsInVoting() {
-				g.Logger.Trace().
-					Str("name", chainName).
-					Str("proposal", proposalID).
-					Str("status", string(proposalVotes.Proposal.Status)).
-					Msg("Proposal is not in voting period - not generating reports for it")
-				continue
-			}
-
 			for wallet := range proposalVotes.Votes {
 				g.Logger.Trace().
 					Str("name", chainName).
 					Str("proposal", proposalID).
 					Str("wallet", wallet).
 					Msg("Generating report for a wallet vote")
 
-				oldVote, _, _ := oldState.GetVoteAndProposal(chainName, proposalID, wallet)
-				newVote, proposal, _ := newState.GetVoteAndProposal(chainName, proposalID, wallet)
+				oldVote, _ := oldState.GetVote(chainName, proposalID, wallet)
+				newVote, _ := newState.GetVote(chainName, proposalID, wallet)
+
+				oldProposal, oldProposalFound := oldState.GetProposal(chainName, proposalID)
+				newProposal := proposalVotes.Proposal
+
+				if oldProposalFound {
+					// There can be the following cases:
+					// 1) old proposal not found - this is a new proposal, no custom logic needed.
+					// 2) old proposal is found, it's not in voting and the current one is not in voting
+					// (like deposit -> deposit) - we don't need to report it at all.
+					// 3) old proposal is found, it's not in voting but the current one is in voting
+					// (like deposit -> deposit) - no custom logic needed.
+					// 4) old proposal is found, it's in voting and the current one is in voting
+					// (like voting -> voting) - no custom logic needed.
+					// 5) old proposal is found, it's in voting and the current one is not in voting
+					// (like voting -> passed) -> send a new event that the voting has finished.
+
+					// case 1, 3 and 4 is handled outside of this if case
+					if oldProposal.IsInVoting() && !newProposal.IsInVoting() { // case 5
+						g.Logger.Debug().
+							Str("chain", chainName).
+							Str("proposal", proposalID).
+							Str("wallet", wallet).
+							Msg("Previously proposal was in voting, but it's not now - sending an alert")
+
+						entries = append(entries, events.FinishedVotingEvent{
+							Chain:    g.Chains.FindByName(chainName),
+							Proposal: newProposal,
+						})
+						continue
+					} else if !oldProposal.IsInVoting() && !newProposal.IsInVoting() { // case 2
+						g.Logger.Debug().
+							Str("chain", chainName).
+							Str("proposal", proposalID).
+							Str("wallet", wallet).
+							Msg("Previously proposal was and is not in voting period - ignoring.")
+						continue
+					}
+				}
 
 				// Error querying for vote - need to notify via Telegram.
 				if newVote.IsError() {
@@ -74,7 +103,7 @@ func (g *Generator) GenerateReport(oldState, newState state.State) reporters.Rep
 						Msg("Error querying for vote - sending an alert")
 					entries = append(entries, events.VoteQueryError{
 						Chain:    g.Chains.FindByName(chainName),
-						Proposal: proposal,
+						Proposal: newProposal,
 						Error:    newVote.Error,
 					})
 
@@ -91,7 +120,7 @@ func (g *Generator) GenerateReport(oldState, newState state.State) reporters.Rep
 					entries = append(entries, events.NotVotedEvent{
 						Chain:    g.Chains.FindByName(chainName),
 						Wallet:   newVote.Wallet,
-						Proposal: proposal,
+						Proposal: newProposal,
 					})
 					continue
 				}
@@ -100,14 +129,14 @@ func (g *Generator) GenerateReport(oldState, newState state.State) reporters.Rep
 				if newVote.HasVoted() && !oldVote.HasVoted() {
 					g.Logger.Debug().
 						Str("chain", chainName).
-						Str("proposal", proposal.ID).
+						Str("proposal", newProposal.ID).
 						Str("wallet", wallet).
 						Msg("Wallet hasn't voted before but voted now - closing an alert")
 
 					entries = append(entries, events.VotedEvent{
 						Chain:    g.Chains.FindByName(chainName),
 						Wallet:   newVote.Wallet,
-						Proposal: proposal,
+						Proposal: newProposal,
 						Vote:     newVote.Vote,
 					})
 					continue
@@ -117,14 +146,14 @@ func (g *Generator) GenerateReport(oldState, newState state.State) reporters.Rep
 				if newVote.HasVoted() && oldVote.HasVoted() && !newVote.Vote.VotesEquals(oldVote.Vote) {
 					g.Logger.Debug().
 						Str("chain", chainName).
-						Str("proposal", proposal.ID).
+						Str("proposal", newProposal.ID).
 						Str("wallet", wallet).
 						Msg("Wallet changed its vote - sending an alert")
 
 					entries = append(entries, events.RevotedEvent{
 						Chain:    g.Chains.FindByName(chainName),
 						Wallet:   newVote.Wallet,
-						Proposal: proposal,
+						Proposal: newProposal,
 						Vote:     newVote.Vote,
 						OldVote:  oldVote.Vote,
 					})

pkg/report/generator_test.go

@@ -284,3 +284,55 @@ func TestReportGeneratorWithRevoted(t *testing.T) {
 	assert.True(t, ok, "Expected to have revoted type!")
 	assert.Equal(t, "proposal", entry.Proposal.ID, "Proposal ID mismatch!")
 }
+
+func TestReportGeneratorWithFinishedVoting(t *testing.T) {
+	t.Parallel()
+
+	stateManager := state.NewStateManager("./state.json", &fs.TestFS{}, logger.GetNopLogger())
+
+	oldState := state.State{
+		ChainInfos: map[string]*state.ChainInfo{
+			"chain": {
+				ProposalVotes: map[string]state.WalletVotes{
+					"proposal": {
+						Proposal: types.Proposal{
+							ID:     "proposal",
+							Status: types.ProposalStatusVoting,
+						},
+						Votes: map[string]state.ProposalVote{
+							"wallet": {},
+						},
+					},
+				},
+			},
+		},
+	}
+	newState := state.State{
+		ChainInfos: map[string]*state.ChainInfo{
+			"chain": {
+				ProposalVotes: map[string]state.WalletVotes{
+					"proposal": {
+						Proposal: types.Proposal{
+							ID:     "proposal",
+							Status: types.ProposalStatusPassed,
+						},
+						Votes: map[string]state.ProposalVote{
+							"wallet": {},
+						},
+					},
+				},
+			},
+		},
+	}
+
+	generator := NewReportGenerator(stateManager, logger.GetNopLogger(), types.Chains{
+		&types.Chain{Name: "chain"},
+	})
+
+	report := generator.GenerateReport(oldState, newState)
+	assert.Len(t, report.Entries, 1, "Expected to have 1 entry!")
+
+	entry, ok := report.Entries[0].(events.FinishedVotingEvent)
+	assert.True(t, ok, "Expected to have not voted type!")
+	assert.Equal(t, "proposal", entry.Proposal.ID, "Proposal ID mismatch!")
+}

pkg/state/generator.go

@@ -81,7 +81,9 @@ func (g *Generator) ProcessChain(
 		Int64("height", proposalsHeight).
 		Msg("Got proposals")
 
+	g.Mutex.Lock()
 	state.SetChainProposalsHeight(chain, proposalsHeight)
+	g.Mutex.Unlock()
 
 	var wg sync.WaitGroup
 
@@ -114,7 +116,9 @@ func (g *Generator) ProcessProposal(
 			Str("name", chain.Name).
 			Str("proposal", proposal.ID).
 			Msg("Proposal is not in voting period - not processing it")
+		g.Mutex.Lock()
 		state.SetProposal(chain, proposal)
+		g.Mutex.Unlock()
 		return
 	}
 
@@ -145,7 +149,7 @@ func (g *Generator) ProcessProposalAndWallet(
 	state State,
 	oldState State,
 ) {
-	oldVote, _, found := oldState.GetVoteAndProposal(chain.Name, proposal.ID, wallet.Address)
+	oldVote, found := oldState.GetVote(chain.Name, proposal.ID, wallet.Address)
 	vote, voteHeight, err := fetcher.GetVote(proposal.ID, wallet.Address, oldVote.Height)
 
 	proposalVote := ProposalVote{

pkg/state/manager.go

@@ -3,6 +3,7 @@ package state
 import (
 	"encoding/json"
 	"main/pkg/fs"
+	"main/pkg/utils"
 
 	"github.com/rs/zerolog"
 )
@@ -41,13 +42,9 @@ func (m *Manager) Load() {
 }
 
 func (m *Manager) Save() {
-	content, err := json.Marshal(m.State)
-	if err != nil {
-		m.Logger.Warn().Err(err).Msg("Could not marshal state")
-		return
-	}
+	content := utils.MustMarshal(m.State)
 
-	if err = m.Filesystem.WriteFile(m.StatePath, content, 0o600); err != nil {
+	if err := m.Filesystem.WriteFile(m.StatePath, content, 0o600); err != nil {
 		m.Logger.Warn().Err(err).Msg("Could not save state")
 		return
 	}

pkg/state/manager_test.go

@@ -0,0 +1,89 @@
+package state
+
+import (
+	"main/pkg/fs"
+	"main/pkg/logger"
+	"main/pkg/types"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestStateManagerLoadNotExisting(t *testing.T) {
+	t.Parallel()
+
+	log := logger.GetNopLogger()
+	filesystem := &fs.TestFS{}
+
+	manager := NewStateManager("non-existing.json", filesystem, log)
+	manager.Load()
+
+	assert.Empty(t, manager.State.ChainInfos)
+}
+
+func TestStateManagerLoadFailedUnmarshaling(t *testing.T) {
+	t.Parallel()
+
+	log := logger.GetNopLogger()
+	filesystem := &fs.TestFS{}
+
+	manager := NewStateManager("invalid-json.json", filesystem, log)
+	manager.Load()
+
+	assert.Empty(t, manager.State.ChainInfos)
+}
+
+func TestStateManagerLoadSuccess(t *testing.T) {
+	t.Parallel()
+
+	log := logger.GetNopLogger()
+	filesystem := &fs.TestFS{}
+
+	manager := NewStateManager("valid-state.json", filesystem, log)
+	manager.Load()
+
+	assert.NotEmpty(t, manager.State.ChainInfos)
+}
+
+func TestManagerSaveWithError(t *testing.T) {
+	t.Parallel()
+
+	log := logger.GetNopLogger()
+	filesystem := &fs.TestFS{WithWriteError: true}
+
+	manager := NewStateManager("out.json", filesystem, log)
+	manager.Load()
+	manager.Save()
+
+	assert.Empty(t, manager.State.ChainInfos)
+}
+
+func TestManagerSaveWithoutError(t *testing.T) {
+	t.Parallel()
+
+	log := logger.GetNopLogger()
+	filesystem := &fs.TestFS{}
+
+	manager := NewStateManager("out.json", filesystem, log)
+	manager.Load()
+	manager.Save()
+
+	assert.Empty(t, manager.State.ChainInfos)
+}
+
+func TestManagerCommitState(t *testing.T) {
+	t.Parallel()
+
+	log := logger.GetNopLogger()
+	filesystem := &fs.TestFS{}
+
+	state := NewState()
+	state.SetProposal(&types.Chain{Name: "chain"}, types.Proposal{ID: "id"})
+
+	manager := NewStateManager("out.json", filesystem, log)
+	manager.Load()
+	assert.Empty(t, manager.State.ChainInfos)
+
+	manager.CommitState(state)
+	assert.NotEmpty(t, manager.State.ChainInfos)
+}