CLS 99 add endpoint for getting all messages for account

.gitignore

@@ -26,3 +26,4 @@ go.work
 
 # IDE and other 3rd party tools
 .idea
+*.private.*

.run/requests/address.http

@@ -0,0 +1,16 @@
+################################################ Local
+
+### GET All indexed addresses
+http://{{host}}/v1/address
+
+### Get count of all indexed addresses
+http://{{host}}/v1/address/count
+
+### Get specific address data
+http://{{host}}/v1/address/celestia1z4909eqzfzngegw43tr2vle7d69fhww4hmmusw
+
+### Get specific address transactions
+http://{{host}}/v1/address/celestia1z4909eqzfzngegw43tr2vle7d69fhww4hmmusw/txs
+
+### Get specific address messages
+http://{{host}}/v1/address/celestia1z4909eqzfzngegw43tr2vle7d69fhww4hmmusw/messages

.run/requests/namespace.http

@@ -0,0 +1,6 @@
+### GET namespace messages
+http://{{host}}/v1/namespace/00000000000000000000000000000000000042690c204d39600fddd3/0/messages?limit=2
+
+### GET namespace messages
+http://{{host}}/v1/namespace/00000000000000000000000000000000000042690c204d39600fddd3/0/messages?limit=2
+

cmd/api/handler/address.go

@@ -117,14 +117,15 @@ func (handler *AddressHandler) List(c echo.Context) error {
 //	@Description	Get address transactions
 //	@Tags			address
 //	@ID				address-transactions
-//	@Param			limit		query	integer	false	"Count of requested entities"			mininum(1)	maximum(100)
-//	@Param			offset		query	integer	false	"Offset"								mininum(1)
+//	@Param			hash	path	string	true	"Hash"	minlength(48)	maxlength(48)
+//	@Param			limit		query	integer	false	"Count of requested entities"			minimum(1)	maximum(100)
+//	@Param			offset		query	integer	false	"Offset"								minimum(1)
 //	@Param			sort		query	string	false	"Sort order"							Enums(asc, desc)
 //	@Param			status		query	types.Status	false	"Comma-separated status list"
 //	@Param			msg_type	query	types.MsgType	false	"Comma-separated message types list"
-//	@Param			from		query	integer	false	"Time from in unix timestamp"			mininum(1)
-//	@Param			to			query	integer	false	"Time to in unix timestamp"				mininum(1)
-//	@Param			height		query	integer	false	"Block number"							mininum(1)
+//	@Param			from		query	integer	false	"Time from in unix timestamp"			minimum(1)
+//	@Param			to			query	integer	false	"Time to in unix timestamp"				minimum(1)
+//	@Param			height		query	integer	false	"Block number"							minimum(1)
 //	@Produce		json
 //	@Success		200	{array}		responses.Tx
 //	@Failure		400	{object}	Error
@@ -172,6 +173,77 @@ func (handler *AddressHandler) Transactions(c echo.Context) error {
 	return returnArray(c, response)
 }
 
+type getAddressMessages struct {
+	Hash   string `param:"hash"   validate:"required,address"`
+	Limit  uint64 `query:"limit"  validate:"omitempty,min=1,max=100"`
+	Offset uint64 `query:"offset" validate:"omitempty,min=0"`
+	Sort   string `query:"sort"   validate:"omitempty,oneof=asc desc"`
+}
+
+func (p *getAddressMessages) SetDefault() {
+	if p.Limit == 0 {
+		p.Limit = 10
+	}
+	if p.Sort == "" {
+		p.Sort = asc
+	}
+}
+
+func (p *getAddressMessages) ToFilters() storage.AddressMsgsFilter {
+	return storage.AddressMsgsFilter{
+		Limit:  int(p.Limit),
+		Offset: int(p.Offset),
+		Sort:   pgSort(p.Sort),
+	}
+}
+
+// Messages godoc
+//
+//	@Summary		Get address messages
+//	@Description	Get address messages
+//	@Tags			address
+//	@ID				address-messages
+//	@Param			hash	    path	string	true	"Hash"	minlength(48)	maxlength(48)
+//	@Param			limit		query	integer	false	"Count of requested entities"			minimum(1)	maximum(100)
+//	@Param			offset		query	integer	false	"Offset"								minimum(1)
+//	@Param			sort		query	string	false	"Sort order"							Enums(asc, desc)
+//	@Produce		json
+//	@Success		200	{array}		responses.Message
+//	@Failure		400	{object}	Error
+//	@Failure		500	{object}	Error
+//	@Router			/v1/address/{hash}/messages [get]
+func (handler *AddressHandler) Messages(c echo.Context) error {
+	req, err := bindAndValidate[getAddressMessages](c)
+	if err != nil {
+		return badRequestError(c, err)
+	}
+
+	req.SetDefault()
+
+	_, hash, err := types.Address(req.Hash).Decode()
+	if err != nil {
+		return badRequestError(c, err)
+	}
+
+	address, err := handler.address.ByHash(c.Request().Context(), hash)
+	if err := handleError(c, err, handler.address); err != nil {
+		return err
+	}
+
+	filters := req.ToFilters()
+	msgs, err := handler.address.Messages(c.Request().Context(), address.Id, filters)
+	if err := handleError(c, err, handler.txs); err != nil {
+		return err
+	}
+
+	response := make([]responses.Message, len(msgs))
+	for i := range msgs {
+		response[i] = responses.NewMessageForAddress(msgs[i])
+	}
+
+	return returnArray(c, response)
+}
+
 // Count godoc
 //
 //	@Summary		Get count of addresses in network

cmd/api/handler/address_test.go

@@ -6,12 +6,6 @@ package handler
 import (
 	"context"
 	"encoding/json"
-	"net/http"
-	"net/http/httptest"
-	"net/url"
-	"strings"
-	"testing"
-
 	"github.com/dipdup-io/celestia-indexer/cmd/api/handler/responses"
 	"github.com/dipdup-io/celestia-indexer/internal/storage"
 	"github.com/dipdup-io/celestia-indexer/internal/storage/mock"
@@ -20,6 +14,11 @@ import (
 	"github.com/shopspring/decimal"
 	"github.com/stretchr/testify/suite"
 	"go.uber.org/mock/gomock"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"testing"
 )
 
 var (
@@ -230,6 +229,60 @@ func (s *AddressTestSuite) TestListHeight() {
 	s.Require().Equal(types.StatusSuccess, tx.Status)
 }
 
+func (s *AddressTestSuite) TestMessages() {
+	q := make(url.Values)
+	q.Set("limit", "10")
+	q.Set("offset", "0")
+	q.Set("sort", "desc")
+
+	req := httptest.NewRequest(http.MethodGet, "/?"+q.Encode(), nil)
+	rec := httptest.NewRecorder()
+	c := s.echo.NewContext(req, rec)
+	c.SetPath("/address/:hash/messages")
+	c.SetParamNames("hash")
+	c.SetParamValues(testAddress)
+
+	s.address.EXPECT().
+		ByHash(gomock.Any(), testHashAddress).
+		Return(storage.Address{
+			Id:      1,
+			Hash:    testHashAddress,
+			Address: testAddress,
+		}, nil)
+
+	s.address.EXPECT().
+		Messages(gomock.Any(), uint64(1), gomock.Any()).
+		Return([]storage.MsgAddress{
+			{
+				AddressId: 1,
+				MsgId:     1,
+				Type:      types.MsgAddressTypeDelegator,
+				Msg: &storage.Message{
+					Id:       1,
+					Height:   1000,
+					Position: 0,
+					Type:     types.MsgWithdrawDelegatorReward,
+					TxId:     1,
+					Data:     nil,
+				},
+			},
+		}, nil)
+
+	s.Require().NoError(s.handler.Messages(c))
+	s.Require().Equal(http.StatusOK, rec.Code)
+
+	var msgs []responses.Message
+	err := json.NewDecoder(rec.Body).Decode(&msgs)
+	s.Require().NoError(err)
+	s.Require().Len(msgs, 1)
+
+	msg := msgs[0]
+	s.Require().EqualValues(1, msg.Id)
+	s.Require().EqualValues(1000, msg.Height)
+	s.Require().Equal(int64(0), msg.Position)
+	s.Require().EqualValues(types.MsgWithdrawDelegatorReward, msg.Type)
+}
+
 func (s *AddressTestSuite) TestCount() {
 	req := httptest.NewRequest(http.MethodGet, "/", nil)
 	rec := httptest.NewRecorder()

cmd/api/handler/responses/message.go

@@ -34,3 +34,15 @@ func NewMessage(msg storage.Message) Message {
 		Data:     msg.Data,
 	}
 }
+
+func NewMessageForAddress(msg storage.MsgAddress) Message {
+	return Message{
+		Id:       msg.MsgId,
+		Height:   msg.Msg.Height,
+		Time:     msg.Msg.Time,
+		Position: msg.Msg.Position,
+		TxId:     msg.Msg.TxId,
+		Type:     msg.Msg.Type,
+		Data:     msg.Msg.Data,
+	}
+}

cmd/api/init.go

@@ -219,6 +219,7 @@ func initHandlers(ctx context.Context, e *echo.Echo, cfg Config, db postgres.Sto
 		addressGroup.GET("/count", addressHandlers.Count)
 		addressGroup.GET("/:hash", addressHandlers.Get)
 		addressGroup.GET("/:hash/txs", addressHandlers.Transactions)
+		addressGroup.GET("/:hash/messages", addressHandlers.Messages)
 	}
 
 	blockHandlers := handler.NewBlockHandler(db.Blocks, db.BlockStats, db.Event, db.Namespace, db.State, cfg.Indexer.Name)

internal/storage/address.go

@@ -18,12 +18,19 @@ type AddressListFilter struct {
 	Sort   storage.SortOrder
 }
 
+type AddressMsgsFilter struct {
+	Limit  int
+	Offset int
+	Sort   storage.SortOrder
+}
+
 //go:generate mockgen -source=$GOFILE -destination=mock/$GOFILE -package=mock -typed
 type IAddress interface {
 	storage.Table[*Address]
 
 	ByHash(ctx context.Context, hash []byte) (Address, error)
-	ListWithBalance(ctx context.Context, fltrs AddressListFilter) ([]Address, error)
+	ListWithBalance(ctx context.Context, filters AddressListFilter) ([]Address, error)
+	Messages(ctx context.Context, id uint64, filters AddressMsgsFilter) ([]MsgAddress, error)
 }
 
 // Address -

internal/storage/postgres/address.go

@@ -5,6 +5,7 @@ package postgres
 
 import (
 	"context"
+	"github.com/uptrace/bun"
 
 	"github.com/dipdup-io/celestia-indexer/internal/storage"
 	"github.com/dipdup-net/go-lib/database"
@@ -32,13 +33,31 @@ func (a *Address) ByHash(ctx context.Context, hash []byte) (address storage.Addr
 	return
 }
 
-func (a *Address) ListWithBalance(ctx context.Context, fltrs storage.AddressListFilter) (result []storage.Address, err error) {
+func (a *Address) ListWithBalance(ctx context.Context, filters storage.AddressListFilter) (result []storage.Address, err error) {
 	query := a.DB().NewSelect().Model(&result).
-		Offset(fltrs.Offset).
+		Offset(filters.Offset).
 		Relation("Balance")
 
-	query = addressListFilter(query, fltrs)
+	query = addressListFilter(query, filters)
 
 	err = query.Scan(ctx)
 	return
 }
+
+func (a *Address) Messages(ctx context.Context, id uint64, filters storage.AddressMsgsFilter) (msgs []storage.MsgAddress, err error) {
+	query := a.DB().NewSelect().Model(&msgs).
+		Where("address_id = ?", id).
+		Offset(filters.Offset).
+		Relation("Msg")
+
+	query = addressMsgsFilter(query, filters)
+
+	err = query.Scan(ctx)
+	return
+}
+
+func addressMsgsFilter(query *bun.SelectQuery, filters storage.AddressMsgsFilter) *bun.SelectQuery {
+	query = limitScope(query, filters.Limit)
+	query = sortScope(query, "msg_id", filters.Sort)
+	return query
+}

internal/storage/postgres/storage_test.go

@@ -292,6 +292,27 @@ func (s *StorageTestSuite) TestAddressList() {
 	s.Require().Equal("utia", addresses[1].Balance.Currency)
 }
 
+func (s *StorageTestSuite) TestAddressMessages() {
+	ctx, ctxCancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer ctxCancel()
+
+	messages, err := s.storage.Address.Messages(ctx, 1, storage.AddressMsgsFilter{
+		Limit:  10,
+		Offset: 0,
+		Sort:   sdk.SortOrderAsc,
+	})
+	s.Require().NoError(err)
+	s.Require().Len(messages, 1)
+
+	s.Require().EqualValues(types.MsgAddressTypeFromAddress, messages[0].Type)
+
+	msg := messages[0].Msg
+	s.Require().EqualValues(1, msg.Id)
+	s.Require().EqualValues(1000, msg.Height)
+	s.Require().EqualValues(0, msg.Position)
+	s.Require().Equal(types.MsgWithdrawDelegatorReward, msg.Type)
+}
+
 func (s *StorageTestSuite) TestEventByTxId() {
 	ctx, ctxCancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer ctxCancel()