Add stats server (#51)

* Add stats server

* App go version

* Fix tests

.github/workflows/golangci-lint.yml

@@ -18,7 +18,7 @@ jobs:
       - uses: actions/checkout@v4
       - uses: actions/setup-go@v5
         with:
-          go-version: '1.21'
+          go-version: '1.22'
           cache: false
       - name: golangci-lint
         uses: golangci/golangci-lint-action@v4

.github/workflows/unit-tests.yml

@@ -18,7 +18,7 @@ jobs:
       - uses: actions/checkout@v4
       - uses: actions/setup-go@v5
         with:
-          go-version: '1.21'
+          go-version: '1.22'
           cache: false
       - name: Get dependencies
         run: go mod download && go mod verify

CHANGELOG.md

@@ -7,6 +7,9 @@ to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
 
+### Added
+- Stats server
+
 ## [0.1.7] - 2024-03-20
 
 ### Changed

go.mod

@@ -10,7 +10,7 @@ require (
 	github.com/golang/protobuf v1.5.3
 	github.com/google/uuid v1.6.0
 	github.com/gorilla/mux v1.8.1
-	github.com/goverland-labs/goverland-core-storage/protocol v0.0.25
+	github.com/goverland-labs/goverland-core-storage/protocol v0.0.0
 	github.com/goverland-labs/goverland-datasource-snapshot/protocol v0.4.4
 	github.com/goverland-labs/goverland-helpers-ens-resolver/protocol v0.1.0
 	github.com/goverland-labs/goverland-platform-events v0.2.1
@@ -23,6 +23,7 @@ require (
 	github.com/s-larionov/process-manager v0.0.1
 	github.com/shopspring/decimal v1.3.1
 	github.com/stretchr/testify v1.8.4
+	golang.org/x/sync v0.6.0
 	google.golang.org/grpc v1.62.0
 	google.golang.org/protobuf v1.32.0
 	gorm.io/driver/postgres v1.5.2

go.sum

@@ -180,6 +180,8 @@ golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.6.0 h1:5BMeUDZ7vkXGfEr1x9B4bRcTH4lpkTkpdh0T/J+qjbQ=
+golang.org/x/sync v0.6.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=

internal/app.go

@@ -19,16 +19,18 @@ import (
 	"gorm.io/gorm"
 	"gorm.io/gorm/logger"
 
+	"github.com/goverland-labs/goverland-core-storage/protocol/storagepb"
+
 	"github.com/goverland-labs/goverland-core-storage/internal/config"
 	"github.com/goverland-labs/goverland-core-storage/internal/dao"
 	"github.com/goverland-labs/goverland-core-storage/internal/ensresolver"
 	"github.com/goverland-labs/goverland-core-storage/internal/events"
 	"github.com/goverland-labs/goverland-core-storage/internal/proposal"
+	"github.com/goverland-labs/goverland-core-storage/internal/stats"
 	"github.com/goverland-labs/goverland-core-storage/internal/vote"
 	"github.com/goverland-labs/goverland-core-storage/pkg/grpcsrv"
 	"github.com/goverland-labs/goverland-core-storage/pkg/health"
 	"github.com/goverland-labs/goverland-core-storage/pkg/prometheus"
-	"github.com/goverland-labs/goverland-core-storage/protocol/storagepb"
 )
 
 type Application struct {
@@ -54,6 +56,8 @@ type Application struct {
 	ensService *ensresolver.Service
 
 	eventsRepo *events.Repo
+
+	statsService *stats.Service
 }
 
 func NewApplication(cfg config.App) (*Application, error) {
@@ -171,6 +175,8 @@ func (a *Application) initServices() error {
 		return fmt.Errorf("init vote: %w", err)
 	}
 
+	a.initStats()
+
 	err = a.initAPI()
 	if err != nil {
 		return fmt.Errorf("init api: %w", err)
@@ -282,6 +288,13 @@ func (a *Application) initVote(nc *nats.Conn, pb *natsclient.Publisher) error {
 	return nil
 }
 
+func (a *Application) initStats() {
+	a.statsService = stats.NewService(a.daoRepo, a.proposalRepo)
+	cw := stats.NewCalcTotalsWorker(a.statsService)
+
+	a.manager.AddWorker(process.NewCallbackWorker("calc-totals", cw.Start))
+}
+
 func (a *Application) initAPI() error {
 	authInterceptor := grpcsrv.NewAuthInterceptor()
 	srv := grpcsrv.NewGrpcServer(
@@ -295,6 +308,7 @@ func (a *Application) initAPI() error {
 	storagepb.RegisterProposalServer(srv, proposal.NewServer(a.proposalService))
 	storagepb.RegisterVoteServer(srv, vote.NewServer(a.voteService))
 	storagepb.RegisterEnsServer(srv, ensresolver.NewServer(a.ensService))
+	storagepb.RegisterStatsServer(srv, stats.NewServer(a.statsService))
 
 	a.manager.AddWorker(grpcsrv.NewGrpcServerWorker("API", srv, a.cfg.InternalAPI.Bind))
 

internal/dao/filters.go

@@ -36,6 +36,18 @@ func (f CategoryFilter) Apply(db *gorm.DB) *gorm.DB {
 	return db.Where("categories @> ?", fmt.Sprintf("\"%s\"", f.Category))
 }
 
+type VerifiedFilter struct {
+}
+
+func (f VerifiedFilter) Apply(db *gorm.DB) *gorm.DB {
+	var (
+		dummy Dao
+		_     = dummy.Verified
+	)
+
+	return db.Where("verified is true")
+}
+
 type NotCategoryFilter struct {
 	Category string
 }

internal/dao/repo.go

@@ -96,6 +96,25 @@ func (r *Repo) GetByFilters(filters []Filter, count bool) (DaoList, error) {
 	}, nil
 }
 
+func (r *Repo) GetCountByFilters(filters []Filter) (int64, error) {
+	db := r.db.Model(&Dao{})
+	for _, f := range filters {
+		if _, ok := f.(PageFilter); ok {
+			continue
+		}
+
+		db = f.Apply(db)
+	}
+
+	var cnt int64
+	err := db.Count(&cnt).Error
+	if err != nil {
+		return 0, fmt.Errorf("db.Count: %w", err)
+	}
+
+	return cnt, nil
+}
+
 func (r *Repo) GetCategories() ([]string, error) {
 	var res []string
 	err := r.db.Raw(`SELECT distinct JSONB_ARRAY_ELEMENTS_TEXT(categories) AS category FROM daos ORDER BY category`).Scan(&res).Error

internal/dao/service_test.go

@@ -70,11 +70,13 @@ func TestUnitHandleDao(t *testing.T) {
 				m := NewMockDataProvider(ctrl)
 				m.EXPECT().GetByID(gomock.Any()).Times(1).Return(nil, gorm.ErrRecordNotFound)
 				m.EXPECT().Create(gomock.Any()).Times(1).Return(nil)
+				m.EXPECT().UpdateProposalCnt(gomock.Any()).Times(1).Return(nil)
+				m.EXPECT().UpdateActiveVotes(gomock.Any()).Times(1).Return(nil)
 				return m
 			},
 			p: func(ctrl *gomock.Controller) Publisher {
 				m := NewMockPublisher(ctrl)
-				m.EXPECT().PublishJSON(gomock.Any(), gomock.Any(), gomock.Any()).Times(3).Return(nil)
+				m.EXPECT().PublishJSON(gomock.Any(), gomock.Any(), gomock.Any()).Times(2).Return(nil)
 				return m
 			},
 			event:    Dao{ID: id1},
@@ -85,6 +87,8 @@ func TestUnitHandleDao(t *testing.T) {
 				m := NewMockDataProvider(ctrl)
 				m.EXPECT().GetByID(gomock.Any()).Times(1).Return(&Dao{ID: id1, Name: "updated"}, nil)
 				m.EXPECT().Update(gomock.Any()).Times(1).Return(nil)
+				m.EXPECT().UpdateProposalCnt(gomock.Any()).Times(1).Return(nil)
+				m.EXPECT().UpdateActiveVotes(gomock.Any()).Times(1).Return(nil)
 				return m
 			},
 			p: func(ctrl *gomock.Controller) Publisher {
@@ -138,11 +142,13 @@ func TestUnitHandleDao(t *testing.T) {
 				m := NewMockDataProvider(ctrl)
 				m.EXPECT().GetByID(gomock.Any()).Times(1).Return(nil, gorm.ErrRecordNotFound)
 				m.EXPECT().Create(gomock.Any()).Times(1).Return(nil)
+				m.EXPECT().UpdateProposalCnt(gomock.Any()).Times(1).Return(nil)
+				m.EXPECT().UpdateActiveVotes(gomock.Any()).Times(1).Return(nil)
 				return m
 			},
 			p: func(ctrl *gomock.Controller) Publisher {
 				m := NewMockPublisher(ctrl)
-				m.EXPECT().PublishJSON(gomock.Any(), gomock.Any(), gomock.Any()).Times(3).Return(errors.New("unexpected error"))
+				m.EXPECT().PublishJSON(gomock.Any(), gomock.Any(), gomock.Any()).Times(2).Return(errors.New("unexpected error"))
 				return m
 			},
 			event:    Dao{ID: id1},
@@ -166,6 +172,8 @@ func TestUnitHandleDao(t *testing.T) {
 				m := NewMockDataProvider(ctrl)
 				m.EXPECT().GetByID(gomock.Any()).Times(1).Return(&Dao{ID: id1, Name: "name"}, nil)
 				m.EXPECT().Update(gomock.Any()).Times(1).Return(nil)
+				m.EXPECT().UpdateProposalCnt(gomock.Any()).Times(1).Return(nil)
+				m.EXPECT().UpdateActiveVotes(gomock.Any()).Times(1).Return(nil)
 				return m
 			},
 			p: func(ctrl *gomock.Controller) Publisher {

internal/proposal/repo.go

@@ -84,6 +84,28 @@ func (r *Repo) GetByFilters(filters []Filter) (ProposalList, error) {
 	return getProposalList(db, filters, cnt)
 }
 
+func (r *Repo) GetCountByFilters(filters []Filter) (int64, error) {
+	db := r.db.
+		Model(&Proposal{}).
+		InnerJoins("inner join daos on daos.id = proposals.dao_id")
+
+	for _, f := range filters {
+		if _, ok := f.(PageFilter); ok {
+			continue
+		}
+
+		db = f.Apply(db)
+	}
+
+	var cnt int64
+	err := db.Count(&cnt).Error
+	if err != nil {
+		return 0, fmt.Errorf("db.Count: %w", err)
+	}
+
+	return cnt, nil
+}
+
 func (r *Repo) GetTop(filters []Filter) (ProposalList, error) {
 	db := getTopProposalOfDaoTable(r.db)
 	db = db.InnerJoins("inner join daos on daos.id = proposals.dao_id").Order("votes/(EXTRACT(EPOCH FROM CURRENT_TIMESTAMP)-start) desc")

internal/proposal/service_test.go

@@ -97,7 +97,7 @@ func TestUnitHandleProposal(t *testing.T) {
 			},
 			p: func(ctrl *gomock.Controller) Publisher {
 				m := NewMockPublisher(ctrl)
-				m.EXPECT().PublishJSON(gomock.Any(), gomock.Any(), gomock.Any()).Times(1).Return(nil)
+				m.EXPECT().PublishJSON(gomock.Any(), gomock.Any(), gomock.Any()).Times(2).Return(nil)
 				return m
 			},
 			event:    Proposal{ID: "id-1", Title: "name", Quorum: 50},
@@ -178,7 +178,7 @@ func TestUnitHandleProposal(t *testing.T) {
 			},
 			p: func(ctrl *gomock.Controller) Publisher {
 				m := NewMockPublisher(ctrl)
-				m.EXPECT().PublishJSON(gomock.Any(), gomock.Any(), gomock.Any()).Times(1).Return(errors.New("unexpected error"))
+				m.EXPECT().PublishJSON(gomock.Any(), gomock.Any(), gomock.Any()).Times(2).Return(errors.New("unexpected error"))
 				return m
 			},
 			event:    Proposal{ID: "id-1", Quorum: 50},
@@ -239,6 +239,7 @@ func TestUnitProcessAvailableForVoting(t *testing.T) {
 						End:       int(time.Now().Add(time.Hour * 24).Unix()),
 					},
 				}, nil)
+				m.EXPECT().Update(gomock.Any()).Times(1).Return(nil)
 				return m
 			},
 			er: func(ctrl *gomock.Controller) EventRegistered {
@@ -266,6 +267,7 @@ func TestUnitProcessAvailableForVoting(t *testing.T) {
 						End:       int(time.Now().Add(-time.Hour * 1).Unix()),
 					},
 				}, nil)
+				m.EXPECT().Update(gomock.Any()).Times(1).Return(nil)
 				return m
 			},
 			er: func(ctrl *gomock.Controller) EventRegistered {
@@ -293,6 +295,7 @@ func TestUnitProcessAvailableForVoting(t *testing.T) {
 						End:       int(time.Now().Add(time.Hour * 24 * 7).Unix()),
 					},
 				}, nil)
+				m.EXPECT().Update(gomock.Any()).Times(1).Return(nil)
 				return m
 			},
 			er: func(ctrl *gomock.Controller) EventRegistered {
@@ -320,6 +323,7 @@ func TestUnitProcessAvailableForVoting(t *testing.T) {
 						End:       int(time.Now().Add(time.Hour * 24).Unix()),
 					},
 				}, nil)
+				m.EXPECT().Update(gomock.Any()).Times(1).Return(nil)
 				return m
 			},
 			er: func(ctrl *gomock.Controller) EventRegistered {

internal/stats/calc_totals_worker.go

@@ -0,0 +1,36 @@
+package stats
+
+import (
+	"context"
+	"time"
+
+	"github.com/rs/zerolog/log"
+)
+
+const (
+	calcCheckDelay = time.Minute * 10
+)
+
+type CalcTotalsWorker struct {
+	service *Service
+}
+
+func NewCalcTotalsWorker(s *Service) *CalcTotalsWorker {
+	return &CalcTotalsWorker{
+		service: s,
+	}
+}
+
+func (w *CalcTotalsWorker) Start(ctx context.Context) error {
+	for {
+		if err := w.service.refreshTotals(); err != nil {
+			log.Error().Err(err).Msg("calc totals worker")
+		}
+
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-time.After(calcCheckDelay):
+		}
+	}
+}

internal/stats/models.go

@@ -0,0 +1,15 @@
+package stats
+
+type Dao struct {
+	Total         int64
+	TotalVerified int64
+}
+
+type Proposals struct {
+	Total int64
+}
+
+type Totals struct {
+	Dao       Dao
+	Proposals Proposals
+}

internal/stats/server.go

@@ -0,0 +1,33 @@
+package stats
+
+import (
+	"context"
+
+	pb "github.com/goverland-labs/goverland-core-storage/protocol/storagepb"
+)
+
+type Server struct {
+	pb.UnimplementedStatsServer
+
+	service *Service
+}
+
+func NewServer(s *Service) *Server {
+	return &Server{
+		service: s,
+	}
+}
+
+func (s *Server) GetTotals(_ context.Context, _ *pb.GetTotalsRequest) (*pb.GetTotalsResponse, error) {
+	totals := s.service.GetTotals()
+
+	return &pb.GetTotalsResponse{
+		Dao: &pb.DaoStats{
+			Total:         totals.Dao.Total,
+			TotalVerified: totals.Dao.TotalVerified,
+		},
+		Proposals: &pb.ProposalsStats{
+			Total: totals.Proposals.Total,
+		},
+	}, nil
+}