Feature: add cumulative stats

cmd/api/handler/stats.go

@@ -229,6 +229,44 @@ func (sh StatsHandler) Series(c echo.Context) error {
 	return returnArray(c, response)
 }
 
+// SeriesCumulative godoc
+//
+//	@Summary		Get cumulative histogram with precomputed stats
+//	@Description	Get cumulative histogram with precomputed stats by series name and timeframe
+//	@Tags			stats
+//	@ID				stats-series-cumulative
+//	@Param			timeframe	path	string	true	"Timeframe"						Enums(day, week, month, year)
+//	@Param			name		path	string	true	"Series name"					Enums(blobs_size, blobs_count, fee, tx_count, gas_used, gas_limit, bytes_in_block)
+//	@Param			from		query	integer	false	"Time from in unix timestamp"	mininum(1)
+//	@Param			to			query	integer	false	"Time to in unix timestamp"		mininum(1)
+//	@Produce		json
+//	@Success		200	{array}		responses.SeriesItem
+//	@Failure		400	{object}	Error
+//	@Failure		500	{object}	Error
+//	@Router			/stats/series/{name}/{timeframe}/cumulative [get]
+func (sh StatsHandler) SeriesCumulative(c echo.Context) error {
+	req, err := bindAndValidate[seriesRequest](c)
+	if err != nil {
+		return badRequestError(c, err)
+	}
+
+	histogram, err := sh.repo.CumulativeSeries(
+		c.Request().Context(),
+		storage.Timeframe(req.Timeframe),
+		req.SeriesName,
+		storage.NewSeriesRequest(req.From, req.To),
+	)
+	if err != nil {
+		return handleError(c, err, sh.nsRepo)
+	}
+
+	response := make([]responses.SeriesItem, len(histogram))
+	for i := range histogram {
+		response[i] = responses.NewSeriesItem(histogram[i])
+	}
+	return returnArray(c, response)
+}
+
 type namespaceSeriesRequest struct {
 	Id         string `example:"0011223344" param:"id"        swaggertype:"string"  validate:"required,hexadecimal,len=56"`
 	Timeframe  string `example:"hour"       param:"timeframe" swaggertype:"string"  validate:"required,oneof=hour day week month year"`

cmd/api/handler/stats_test.go

@@ -236,6 +236,7 @@ func (s *StatsTestSuite) TestBlockStatsHistogram() {
 		storage.SeriesGasPrice,
 		storage.SeriesGasUsed,
 		storage.SeriesBytesInBlock,
+		storage.SeriesBlobsCount,
 	} {
 
 		for _, tf := range []storage.Timeframe{
@@ -279,6 +280,53 @@ func (s *StatsTestSuite) TestBlockStatsHistogram() {
 	}
 }
 
+func (s *StatsTestSuite) TestBlockCumulativeStatsHistogram() {
+	for _, name := range []string{
+		storage.SeriesBlobsSize,
+		storage.SeriesFee,
+		storage.SeriesTxCount,
+		storage.SeriesGasLimit,
+		storage.SeriesGasUsed,
+		storage.SeriesBytesInBlock,
+		storage.SeriesBlobsCount,
+	} {
+
+		for _, tf := range []storage.Timeframe{
+			storage.TimeframeDay,
+			storage.TimeframeWeek,
+			storage.TimeframeMonth,
+			storage.TimeframeYear,
+		} {
+			req := httptest.NewRequest(http.MethodGet, "/", nil)
+			rec := httptest.NewRecorder()
+			c := s.echo.NewContext(req, rec)
+			c.SetPath("/v1/stats/series/:name/:timeframe/cumulative")
+			c.SetParamNames("name", "timeframe")
+			c.SetParamValues(name, string(tf))
+
+			s.stats.EXPECT().
+				Series(gomock.Any(), tf, name, gomock.Any()).
+				Return([]storage.SeriesItem{
+					{
+						Time:  testTime,
+						Value: "11234",
+					},
+				}, nil)
+
+			s.Require().NoError(s.handler.Series(c))
+			s.Require().Equal(http.StatusOK, rec.Code)
+
+			var response []responses.SeriesItem
+			err := json.NewDecoder(rec.Body).Decode(&response)
+			s.Require().NoError(err)
+			s.Require().Len(response, 1)
+
+			item := response[0]
+			s.Require().Equal("11234", item.Value)
+		}
+	}
+}
+
 func (s *StatsTestSuite) TestNamespaceStatsHistogram() {
 	for _, name := range []string{
 		storage.SeriesNsPfbCount,

cmd/api/init.go

@@ -401,6 +401,7 @@ func initHandlers(ctx context.Context, e *echo.Echo, cfg Config, db postgres.Sto
 		series := stats.Group("/series")
 		{
 			series.GET("/:name/:timeframe", statsHandler.Series)
+			series.GET("/:name/:timeframe/cumulative", statsHandler.SeriesCumulative)
 		}
 	}
 

internal/storage/postgres/stats.go

@@ -210,6 +210,58 @@ func (s Stats) NamespaceSeries(ctx context.Context, timeframe storage.Timeframe,
 	return
 }
 
+func (s Stats) CumulativeSeries(ctx context.Context, timeframe storage.Timeframe, name string, req storage.SeriesRequest) (response []storage.SeriesItem, err error) {
+	query := s.db.DB().NewSelect()
+	switch timeframe {
+	case storage.TimeframeDay:
+		query.Table(storage.ViewBlockStatsByDay)
+	case storage.TimeframeWeek:
+		query.Table(storage.ViewBlockStatsByWeek)
+	case storage.TimeframeMonth:
+		query.Table(storage.ViewBlockStatsByMonth)
+	case storage.TimeframeYear:
+		query.Table(storage.ViewBlockStatsByYear)
+	default:
+		return nil, errors.Errorf("unexpected timeframe %s", timeframe)
+	}
+
+	switch name {
+	case storage.SeriesBlobsSize:
+		query.ColumnExpr("ts, sum(sum(blobs_size)) OVER(ORDER BY ts) as value")
+	case storage.SeriesFee:
+		query.ColumnExpr("ts, sum(sum(fee)) OVER(ORDER BY ts) as value")
+	case storage.SeriesTxCount:
+		query.ColumnExpr("ts, sum(sum(tx_count)) OVER(ORDER BY ts) as value")
+	case storage.SeriesGasLimit:
+		query.ColumnExpr("ts, sum(sum(gas_limit)) OVER(ORDER BY ts) as value")
+	case storage.SeriesGasUsed:
+		query.ColumnExpr("ts, sum(sum(gas_used)) OVER(ORDER BY ts) as value")
+	case storage.SeriesBytesInBlock:
+		query.ColumnExpr("ts, sum(sum(bytes_in_block)) OVER(ORDER BY ts) as value")
+	case storage.SeriesBlobsCount:
+		query.ColumnExpr("ts, sum(sum(blobs_count)) OVER(ORDER BY ts) as value")
+	default:
+		return nil, errors.Errorf("unexpected series name: %s", name)
+	}
+
+	withQuery := query.Group("ts")
+
+	q := s.db.DB().
+		NewSelect().
+		With("q", withQuery).
+		Table("q")
+
+	if !req.From.IsZero() {
+		q = q.Where("ts >= ?", req.From)
+	}
+	if !req.To.IsZero() {
+		q = q.Where("ts < ?", req.To)
+	}
+
+	err = q.Scan(ctx, &response)
+	return
+}
+
 func (s Stats) StakingSeries(ctx context.Context, timeframe storage.Timeframe, name string, validatorId uint64, req storage.SeriesRequest) (response []storage.SeriesItem, err error) {
 	var view string
 	switch timeframe {

internal/storage/postgres/stats_test.go

@@ -250,7 +250,27 @@ func (s *StatsTestSuite) TestSeriesWithFrom() {
 	ctx, ctxCancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer ctxCancel()
 
-	items, err := s.storage.Stats.Series(ctx, storage.TimeframeHour, storage.SeriesBlobsSize, storage.SeriesRequest{
+	items, err := s.storage.Stats.Series(ctx, storage.TimeframeDay, storage.SeriesBlobsSize, storage.SeriesRequest{
+		From: time.Unix(1701192801, 0).UTC(),
+	})
+	s.Require().NoError(err)
+	s.Require().Len(items, 0)
+}
+
+func (s *StatsTestSuite) TestCumulativeSeries() {
+	ctx, ctxCancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer ctxCancel()
+
+	items, err := s.storage.Stats.CumulativeSeries(ctx, storage.TimeframeDay, storage.SeriesBlobsSize, storage.SeriesRequest{})
+	s.Require().NoError(err)
+	s.Require().Len(items, 1)
+}
+
+func (s *StatsTestSuite) TestCumulativeSeriesWithFrom() {
+	ctx, ctxCancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer ctxCancel()
+
+	items, err := s.storage.Stats.CumulativeSeries(ctx, storage.TimeframeHour, storage.SeriesBlobsSize, storage.SeriesRequest{
 		From: time.Unix(1701192801, 0).UTC(),
 	})
 	s.Require().NoError(err)

internal/storage/stats.go

@@ -159,6 +159,7 @@ type IStats interface {
 	TPS(ctx context.Context) (TPS, error)
 	TxCountForLast24h(ctx context.Context) ([]TxCountForLast24hItem, error)
 	Series(ctx context.Context, timeframe Timeframe, name string, req SeriesRequest) ([]SeriesItem, error)
+	CumulativeSeries(ctx context.Context, timeframe Timeframe, name string, req SeriesRequest) ([]SeriesItem, error)
 	NamespaceSeries(ctx context.Context, timeframe Timeframe, name string, nsId uint64, req SeriesRequest) (response []SeriesItem, err error)
 	StakingSeries(ctx context.Context, timeframe Timeframe, name string, validatorId uint64, req SeriesRequest) (response []SeriesItem, err error)
 }

internal/storage/types/packed_bytes.go

@@ -1,3 +1,6 @@
+// SPDX-FileCopyrightText: 2024 PK Lab AG <contact@pklab.io>
+// SPDX-License-Identifier: MIT
+
 package types
 
 import (