Implement transfer liqudity event handler

chains/evm/calls/events/listener.go

@@ -24,15 +24,18 @@ type ChainClient interface {
 }
 
 type Listener struct {
-	client ChainClient
-	abi    abi.ABI
+	client   ChainClient
+	abi      abi.ABI
+	adminAbi abi.ABI
 }
 
 func NewListener(client ChainClient) *Listener {
-	abi, _ := abi.JSON(strings.NewReader(consts.BridgeABI))
+	bridgeAbi, _ := abi.JSON(strings.NewReader(consts.BridgeABI))
+	adminAbi, _ := abi.JSON(strings.NewReader(consts.BridgeABI))
 	return &Listener{
-		client: client,
-		abi:    abi,
+		client:   client,
+		abi:      bridgeAbi,
+		adminAbi: adminAbi,
 	}
 }
 
@@ -128,6 +131,36 @@ func (l *Listener) FetchRetryEvents(ctx context.Context, contractAddress common.
 	return retryEvents, nil
 }
 
+func (l *Listener) FetchTransferLiqudityEvents(ctx context.Context, contractAddress common.Address, startBlock *big.Int, endBlock *big.Int) ([]*TransferLiquidity, error) {
+	logs, err := l.client.FetchEventLogs(ctx, contractAddress, string(TransferLiquiditySig), startBlock, endBlock)
+	if err != nil {
+		return nil, err
+	}
+
+	transferLiqudityEvents := make([]*TransferLiquidity, 0)
+	for _, re := range logs {
+		r, err := l.unpackTransferLiqudity(re.Data)
+		if err != nil {
+			log.Err(err).Msgf("failed unpacking refresh event log")
+			continue
+		}
+
+		r.TransactionHash = re.TxHash.Hex()
+		transferLiqudityEvents = append(transferLiqudityEvents, r)
+	}
+	return transferLiqudityEvents, nil
+}
+
+func (l *Listener) unpackTransferLiqudity(data []byte) (*TransferLiquidity, error) {
+	var tl TransferLiquidity
+	err := l.adminAbi.UnpackIntoInterface(&tl, "TransferLiquidity", data)
+	if err != nil {
+		return &TransferLiquidity{}, err
+	}
+
+	return &tl, nil
+}
+
 func (l *Listener) FetchKeygenEvents(ctx context.Context, contractAddress common.Address, startBlock *big.Int, endBlock *big.Int) ([]ethTypes.Log, error) {
 	logs, err := l.client.FetchEventLogs(ctx, contractAddress, string(StartKeygenSig), startBlock, endBlock)
 	if err != nil {

chains/evm/listener/eventHandlers/event-handler.go

@@ -12,6 +12,7 @@ import (
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 
+	"github.com/ChainSafe/sygma-relayer/chains"
 	"github.com/ChainSafe/sygma-relayer/chains/evm/calls/consts"
 	"github.com/ChainSafe/sygma-relayer/relayer/transfer"
 	"github.com/ChainSafe/sygma-relayer/store"
@@ -38,6 +39,7 @@ type EventListener interface {
 	FetchRetryEvents(ctx context.Context, contractAddress common.Address, startBlock *big.Int, endBlock *big.Int) ([]events.RetryEvent, error)
 	FetchRetryDepositEvents(event events.RetryEvent, bridgeAddress common.Address, blockConfirmations *big.Int) ([]events.Deposit, error)
 	FetchDeposits(ctx context.Context, address common.Address, startBlock *big.Int, endBlock *big.Int) ([]*events.Deposit, error)
+	FetchTransferLiqudityEvents(ctx context.Context, contractAddress common.Address, startBlock *big.Int, endBlock *big.Int) ([]*events.TransferLiquidity, error)
 }
 
 type PropStorer interface {
@@ -449,3 +451,74 @@ func (eh *DepositEventHandler) HandleEvents(startBlock *big.Int, endBlock *big.I
 
 	return nil
 }
+
+type TransferLiqudityEventHandler struct {
+	log           zerolog.Logger
+	adminAddress  common.Address
+	eventListener EventListener
+	domainID      uint8
+	msgChan       chan []*message.Message
+}
+
+func NewTransferLiquidityEventHandler(
+	logC zerolog.Context,
+	eventListener EventListener,
+	adminAddress common.Address,
+	domainID uint8,
+	msgChan chan []*message.Message,
+) *TransferLiqudityEventHandler {
+	return &TransferLiqudityEventHandler{
+		log:           logC.Logger(),
+		domainID:      domainID,
+		msgChan:       msgChan,
+		eventListener: eventListener,
+		adminAddress:  adminAddress,
+	}
+}
+
+func (eh *TransferLiqudityEventHandler) HandleEvents(
+	startBlock *big.Int,
+	endBlock *big.Int,
+) error {
+	transferLiqudityEvents, err := eh.eventListener.FetchTransferLiqudityEvents(
+		context.Background(), eh.adminAddress, startBlock, endBlock,
+	)
+	if err != nil {
+		return fmt.Errorf("unable to fetch transfer liqudity events because of: %+v", err)
+	}
+	if len(transferLiqudityEvents) == 0 {
+		return nil
+	}
+
+	domainMsgs := make(map[uint8][]*message.Message)
+	for _, e := range transferLiqudityEvents {
+		messageID := fmt.Sprintf("%d-%d-%d-%d-transfer-liqudity", eh.domainID, e.DomainID, startBlock, endBlock)
+		payload := []interface{}{
+			common.LeftPadBytes(e.Amount.Bytes(), 32),
+			e.DestinationAddress,
+		}
+
+		msg := message.NewMessage(
+			eh.domainID,
+			e.DomainID,
+			transfer.TransferMessageData{
+				DepositNonce: chains.CalculateNonce(endBlock, e.TransactionHash),
+				ResourceId:   e.ResourceID,
+				Metadata:     nil,
+				Payload:      payload,
+				Type:         transfer.FungibleTransfer,
+			},
+			messageID,
+			transfer.TransferMessageType)
+		domainMsgs[e.DomainID] = append(domainMsgs[e.DomainID], msg)
+
+		log.Debug().Str("messageID", msg.ID).Msgf("Resolved transfer liqudity message %+v in block range: %s-%s", msg, startBlock.String(), endBlock.String())
+	}
+
+	for _, msgs := range domainMsgs {
+		go func(msgs []*message.Message) {
+			eh.msgChan <- msgs
+		}(msgs)
+	}
+	return nil
+}

chains/evm/listener/eventHandlers/event-handler_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/golang/mock/gomock"
 	"github.com/stretchr/testify/suite"
 
+	"github.com/ChainSafe/sygma-relayer/chains"
 	"github.com/ChainSafe/sygma-relayer/chains/evm/calls/events"
 	"github.com/ChainSafe/sygma-relayer/chains/evm/listener/eventHandlers"
 	mock_listener "github.com/ChainSafe/sygma-relayer/chains/evm/listener/eventHandlers/mock"
@@ -501,3 +502,89 @@ func (s *DepositHandlerTestSuite) Test_SuccessfulHandleDeposit() {
 	s.Nil(err)
 	s.Equal(msgs, []*message.Message{{Data: transfer.TransferMessageData{DepositNonce: 1}}, {Data: transfer.TransferMessageData{DepositNonce: 2}}})
 }
+
+type TransferLiqudityHandlerTestSuite struct {
+	suite.Suite
+	transferLiqudityEventHandler *eventHandlers.TransferLiqudityEventHandler
+	mockEventListener            *mock_listener.MockEventListener
+	domainID                     uint8
+	msgChan                      chan []*message.Message
+}
+
+func TestRunTrasferLiqudityHandlerTestSuite(t *testing.T) {
+	suite.Run(t, new(TransferLiqudityHandlerTestSuite))
+}
+
+func (s *TransferLiqudityHandlerTestSuite) SetupTest() {
+	ctrl := gomock.NewController(s.T())
+	s.domainID = 1
+	s.mockEventListener = mock_listener.NewMockEventListener(ctrl)
+	s.msgChan = make(chan []*message.Message, 2)
+	s.transferLiqudityEventHandler = eventHandlers.NewTransferLiquidityEventHandler(log.Logger.With(), s.mockEventListener, common.Address{}, s.domainID, s.msgChan)
+}
+
+func (s *TransferLiqudityHandlerTestSuite) Test_FetchDepositFails() {
+	s.mockEventListener.EXPECT().FetchTransferLiqudityEvents(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return([]*events.TransferLiquidity{}, fmt.Errorf("error"))
+
+	err := s.transferLiqudityEventHandler.HandleEvents(big.NewInt(0), big.NewInt(5))
+
+	s.NotNil(err)
+	s.Equal(len(s.msgChan), 0)
+}
+
+func (s *TransferLiqudityHandlerTestSuite) Test_NoEvents() {
+	s.mockEventListener.EXPECT().FetchTransferLiqudityEvents(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return([]*events.TransferLiquidity{}, nil)
+
+	err := s.transferLiqudityEventHandler.HandleEvents(big.NewInt(0), big.NewInt(5))
+
+	s.Nil(err)
+	s.Equal(len(s.msgChan), 0)
+}
+
+func (s *TransferLiqudityHandlerTestSuite) Test_ValidEvents() {
+	e1 := &events.TransferLiquidity{
+		DomainID:           2,
+		ResourceID:         [32]byte{1},
+		Amount:             big.NewInt(100),
+		TransactionHash:    "hash1",
+		DestinationAddress: []byte("recipient1"),
+	}
+	e2 := &events.TransferLiquidity{
+		DomainID:           3,
+		ResourceID:         [32]byte{2},
+		Amount:             big.NewInt(200),
+		TransactionHash:    "hash2",
+		DestinationAddress: []byte("recipient2"),
+	}
+	s.mockEventListener.EXPECT().FetchTransferLiqudityEvents(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return([]*events.TransferLiquidity{e1, e2}, nil)
+
+	err := s.transferLiqudityEventHandler.HandleEvents(big.NewInt(0), big.NewInt(5))
+
+	s.Nil(err)
+	msg1 := <-s.msgChan
+	msg2 := <-s.msgChan
+
+	s.Equal(len(msg1), 1)
+	s.Equal(len(msg2), 1)
+	s.Equal(msg1[0].Data.(transfer.TransferMessageData), transfer.TransferMessageData{
+		DepositNonce: chains.CalculateNonce(big.NewInt(5), e1.TransactionHash),
+		ResourceId:   e1.ResourceID,
+		Metadata:     nil,
+		Payload: []interface{}{
+			common.LeftPadBytes(e1.Amount.Bytes(), 32),
+			e1.DestinationAddress,
+		},
+		Type: transfer.FungibleTransfer,
+	})
+	s.Equal(msg2[0].Data.(transfer.TransferMessageData), transfer.TransferMessageData{
+		DepositNonce: chains.CalculateNonce(big.NewInt(5), e2.TransactionHash),
+		ResourceId:   e2.ResourceID,
+		Metadata:     nil,
+		Payload: []interface{}{
+			common.LeftPadBytes(e2.Amount.Bytes(), 32),
+			e2.DestinationAddress,
+		},
+		Type: transfer.FungibleTransfer,
+	})
+
+}