Feature: transfer indexing (#44)

cmd/api/handler/responses/stats.go

@@ -116,3 +116,17 @@ func NewFeeSummary(summary storage.FeeSummary) FeeSummary {
 		FeeCount:  summary.FeeCount,
 	}
 }
+
+type TokenTransferDistributionItem struct {
+	Asset          string `example:"nria"    format:"string"  json:"asset"`
+	Amount         string `example:"1000000" format:"integer" json:"amount"`
+	TransfersCount int64  `example:"1000000" format:"integer" json:"transfers_count"`
+}
+
+func NewTokenTransferDistributionItem(summary storage.TokenTransferDistributionItem) TokenTransferDistributionItem {
+	return TokenTransferDistributionItem{
+		Asset:          summary.Asset,
+		Amount:         summary.Amount,
+		TransfersCount: summary.TransfersCount,
+	}
+}

cmd/api/handler/stats.go

@@ -195,3 +195,42 @@ func (sh StatsHandler) FeeSummary(c echo.Context) error {
 	}
 	return c.JSON(http.StatusOK, response)
 }
+
+type tokenTransferDistributionRequest struct {
+	Limit uint64 `query:"limit" validate:"omitempty,min=1,max=100"`
+}
+
+func (p *tokenTransferDistributionRequest) SetDefault() {
+	if p.Limit == 0 {
+		p.Limit = 10
+	}
+}
+
+// TokenTransferDistribution godoc
+//
+//	@Summary		Token transfer distribution
+//	@Description	Token transfer distribution
+//	@Tags			stats
+//	@ID				stats-token-transfer-distribution
+//	@Param			limit	query	integer	false	"Count of requested entities"	mininum(1)	maximum(100)
+//	@Produce		json
+//	@Success		200	{array}		responses.TokenTransferDistributionItem
+//	@Failure		500	{object}	Error
+//	@Router			/v1/stats/token/transfer_distribution [get]
+func (sh StatsHandler) TokenTransferDistribution(c echo.Context) error {
+	req, err := bindAndValidate[tokenTransferDistributionRequest](c)
+	if err != nil {
+		return badRequestError(c, err)
+	}
+	req.SetDefault()
+
+	items, err := sh.repo.TokenTransferDistribution(c.Request().Context(), int(req.Limit))
+	if err != nil {
+		return handleError(c, err, sh.rollups)
+	}
+	response := make([]responses.TokenTransferDistributionItem, len(items))
+	for i := range items {
+		response[i] = responses.NewTokenTransferDistributionItem(items[i])
+	}
+	return c.JSON(http.StatusOK, response)
+}

cmd/api/handler/stats_test.go

@@ -11,6 +11,7 @@ import (
 	"testing"
 
 	"github.com/celenium-io/astria-indexer/cmd/api/handler/responses"
+	"github.com/celenium-io/astria-indexer/internal/currency"
 	"github.com/celenium-io/astria-indexer/internal/storage"
 	"github.com/celenium-io/astria-indexer/internal/storage/mock"
 	"github.com/labstack/echo/v4"
@@ -243,3 +244,63 @@ func (s *StatsTestSuite) TestSummaryTimeframe() {
 		s.Require().EqualValues(7, summary.BytesInBlockPct)
 	}
 }
+
+func (s *StatsTestSuite) TestFeeSummary() {
+	req := httptest.NewRequest(http.MethodGet, "/", nil)
+	rec := httptest.NewRecorder()
+	c := s.echo.NewContext(req, rec)
+	c.SetPath("/v1/stats/fee/summary")
+
+	s.stats.EXPECT().
+		FeeSummary(gomock.Any()).
+		Return([]storage.FeeSummary{
+			{
+				Asset:    currency.DefaultCurrency,
+				Amount:   "1000",
+				FeeCount: 100,
+			},
+		}, nil)
+
+	s.Require().NoError(s.handler.FeeSummary(c))
+	s.Require().Equal(http.StatusOK, rec.Code)
+
+	var result []responses.FeeSummary
+	err := json.NewDecoder(rec.Body).Decode(&result)
+	s.Require().NoError(err)
+	s.Require().Len(result, 1)
+
+	summary := result[0]
+	s.Require().EqualValues("1000", summary.Amount)
+	s.Require().EqualValues(100, summary.FeeCount)
+	s.Require().EqualValues(currency.DefaultCurrency, summary.Asset)
+}
+
+func (s *StatsTestSuite) TestTokenTransferDistribution() {
+	req := httptest.NewRequest(http.MethodGet, "/", nil)
+	rec := httptest.NewRecorder()
+	c := s.echo.NewContext(req, rec)
+	c.SetPath("/v1/stats/token/transfer_distribution")
+
+	s.stats.EXPECT().
+		TokenTransferDistribution(gomock.Any(), 10).
+		Return([]storage.TokenTransferDistributionItem{
+			{
+				Asset:          currency.DefaultCurrency,
+				Amount:         "1000",
+				TransfersCount: 100,
+			},
+		}, nil)
+
+	s.Require().NoError(s.handler.TokenTransferDistribution(c))
+	s.Require().Equal(http.StatusOK, rec.Code)
+
+	var result []responses.TokenTransferDistributionItem
+	err := json.NewDecoder(rec.Body).Decode(&result)
+	s.Require().NoError(err)
+	s.Require().Len(result, 1)
+
+	summary := result[0]
+	s.Require().EqualValues("1000", summary.Amount)
+	s.Require().EqualValues(100, summary.TransfersCount)
+	s.Require().EqualValues(currency.DefaultCurrency, summary.Asset)
+}

cmd/api/init.go

@@ -354,6 +354,11 @@ func initHandlers(ctx context.Context, e *echo.Echo, cfg Config, db postgres.Sto
 		{
 			fee.GET("/summary", statsHandler.FeeSummary)
 		}
+
+		token := stats.Group("/token")
+		{
+			token.GET("/transfer_distribution", statsHandler.TokenTransferDistribution)
+		}
 	}
 
 	if cfg.ApiConfig.Prometheus {

database/views/09_transfer_stats_by_hour.sql

@@ -0,0 +1,12 @@
+CREATE MATERIALIZED VIEW IF NOT EXISTS transfer_stats_by_hour
+WITH (timescaledb.continuous, timescaledb.materialized_only=false) AS
+	select 
+		time_bucket('1 hour'::interval, time) AS ts,
+		transfer.asset as asset,
+		count(*) as transfers_count,
+		sum(amount) as amount
+	from transfer
+	group by 1, 2
+	order by 1 desc;
+
+CALL add_view_refresh_job('transfer_stats_by_hour', INTERVAL '1 minute', INTERVAL '1 minute');

database/views/10_transfer_stats_by_day.sql

@@ -0,0 +1,12 @@
+CREATE MATERIALIZED VIEW IF NOT EXISTS transfer_stats_by_day
+WITH (timescaledb.continuous, timescaledb.materialized_only=false) AS
+	select 
+		time_bucket('1 day'::interval, transfer_stats_by_hour.ts) AS ts,
+		transfer_stats_by_hour.asset as asset,
+		sum(transfers_count) as transfers_count,
+		sum(amount) as amount
+	from transfer_stats_by_hour
+	group by 1, 2
+	order by 1 desc;
+
+CALL add_view_refresh_job('transfer_stats_by_day', INTERVAL '1 minute', INTERVAL '1 minute');

database/views/11_transfer_stats_by_month.sql

@@ -0,0 +1,12 @@
+CREATE MATERIALIZED VIEW IF NOT EXISTS transfer_stats_by_month
+WITH (timescaledb.continuous, timescaledb.materialized_only=false) AS
+	select 
+		time_bucket('1 month'::interval, transfer_stats_by_day.ts) AS ts,
+		transfer_stats_by_day.asset as asset,
+		sum(transfers_count) as transfers_count,
+		sum(amount) as amount
+	from transfer_stats_by_day
+	group by 1, 2
+	order by 1 desc;
+
+CALL add_view_refresh_job('transfer_stats_by_month', INTERVAL '1 minute', INTERVAL '1 hour');

internal/storage/block.go

@@ -58,6 +58,7 @@ type Block struct {
 	BlockSignatures []BlockSignature          `bun:"-"` // internal field for saving block signatures
 	Constants       []*Constant               `bun:"-"` // internal field for updating constants
 	Bridges         []*Bridge                 `bun:"-"` // internal field for saving bridges
+	Transfers       []*Transfer               `bun:"-"` // internal field for saving transfers
 
 	Txs      []*Tx       `bun:"rel:has-many"`
 	Stats    *BlockStats `bun:"rel:has-one,join:height=height"`