Fix: split scheduled transfer transactions into chunks of 10

app/domain/repository/transfer.go

@@ -27,6 +27,7 @@ type Transfer interface {
 	// Returns Transfer with preloaded Fee table. Returns nil if not found
 	GetWithFee(txId string) (*entity.Transfer, error)
 	GetWithPreloads(txId string) (*entity.Transfer, error)
+	UpdateFee(txId string, fee string) error
 
 	Create(ct *transfer.Transfer) (*entity.Transfer, error)
 	UpdateStatusCompleted(txId string) error

app/domain/service/read-only.go

@@ -18,8 +18,10 @@ package service
 
 import (
 	mirror_node "github.com/limechain/hedera-eth-bridge-validator/app/clients/hedera/mirror-node/model"
+	model "github.com/limechain/hedera-eth-bridge-validator/app/model/transfer"
 )
 
 type ReadOnly interface {
 	FindTransfer(transferID string, fetch func() (*mirror_node.Response, error), save func(transactionID, scheduleID, status string) error)
+	FindAssetTransfer(transferID string, asset string, transfers []model.Hedera, fetch func() (*mirror_node.Response, error), save func(transactionID, scheduleID, status string) error)
 }

app/helper/fee/fee.go

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2021 LimeChain Ltd.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package fee
+
+import (
+	"github.com/hashgraph/hedera-sdk-go/v2"
+	model "github.com/limechain/hedera-eth-bridge-validator/app/model/transfer"
+	"strconv"
+)
+
+func GetTotalFeeFromTransfers(transfers []model.Hedera, receiver hedera.AccountID) string {
+	result := int64(0)
+	for _, transfer := range transfers {
+		if transfer.Amount < 0 {
+			continue
+		}
+		if transfer.AccountID == receiver {
+			continue
+		}
+		result += transfer.Amount
+	}
+
+	return strconv.FormatInt(result, 10)
+}

app/persistence/entity/transfer.go

@@ -28,9 +28,10 @@ type Transfer struct {
 	NativeAsset   string
 	Receiver      string
 	Amount        string
+	Fee           string
 	Status        string
 	Messages      []Message  `gorm:"foreignKey:TransferID"`
-	Fee           Fee        `gorm:"foreignKey:TransferID"`
+	Fees          []Fee      `gorm:"foreignKey:TransferID"`
 	Schedules     []Schedule `gorm:"foreignKey:TransferID"`
 }
 

app/persistence/transfer/transfer.go

@@ -58,7 +58,7 @@ func (tr Repository) GetByTransactionId(txId string) (*entity.Transfer, error) {
 func (tr Repository) GetWithPreloads(txId string) (*entity.Transfer, error) {
 	tx := &entity.Transfer{}
 	result := tr.dbClient.
-		Preload("Fee").
+		Preload("Fees").
 		Preload("Messages").
 		Model(entity.Transfer{}).
 		Where("transaction_id = ?", txId).
@@ -78,7 +78,7 @@ func (tr Repository) GetWithPreloads(txId string) (*entity.Transfer, error) {
 func (tr Repository) GetWithFee(txId string) (*entity.Transfer, error) {
 	tx := &entity.Transfer{}
 	result := tr.dbClient.
-		Preload("Fee").
+		Preload("Fees").
 		Model(entity.Transfer{}).
 		Where("transaction_id = ?", txId).
 		First(tx)
@@ -102,6 +102,18 @@ func (tr Repository) Save(tx *entity.Transfer) error {
 	return tr.dbClient.Save(tx).Error
 }
 
+func (tr Repository) UpdateFee(txId string, fee string) error {
+	err := tr.dbClient.
+		Model(entity.Transfer{}).
+		Where("transaction_id = ?", txId).
+		UpdateColumn("fee", fee).
+		Error
+	if err == nil {
+		tr.logger.Debugf("Updated Fee of TX [%s] to [%s]", txId, fee)
+	}
+	return err
+}
+
 func (tr Repository) UpdateStatusCompleted(txId string) error {
 	return tr.updateStatus(txId, status.Completed)
 }

app/process/handler/message-submission/handler_test.go

@@ -59,7 +59,7 @@ var (
 		Amount:        tr.Amount,
 		Status:        status.Initial,
 		Messages:      nil,
-		Fee:           entity.Fee{},
+		Fees:          []entity.Fee{},
 		Schedules:     nil,
 	}
 

app/process/handler/read-only/fee-transfer/handler.go

@@ -23,17 +23,20 @@ import (
 	"github.com/limechain/hedera-eth-bridge-validator/app/domain/client"
 	"github.com/limechain/hedera-eth-bridge-validator/app/domain/repository"
 	"github.com/limechain/hedera-eth-bridge-validator/app/domain/service"
+	util "github.com/limechain/hedera-eth-bridge-validator/app/helper/fee"
 	model "github.com/limechain/hedera-eth-bridge-validator/app/model/transfer"
 	"github.com/limechain/hedera-eth-bridge-validator/app/persistence/entity"
 	"github.com/limechain/hedera-eth-bridge-validator/app/persistence/entity/schedule"
 	"github.com/limechain/hedera-eth-bridge-validator/app/persistence/entity/status"
+	"github.com/limechain/hedera-eth-bridge-validator/app/services/fee/distributor"
 	"github.com/limechain/hedera-eth-bridge-validator/config"
 	log "github.com/sirupsen/logrus"
 	"strconv"
 )
 
 // Handler is transfers event handler
 type Handler struct {
+	transferRepository repository.Transfer
 	feeRepository      repository.Fee
 	scheduleRepository repository.Schedule
 	mirrorNode         client.MirrorNode
@@ -46,6 +49,7 @@ type Handler struct {
 }
 
 func NewHandler(
+	transferRepository repository.Transfer,
 	feeRepository repository.Fee,
 	scheduleRepository repository.Schedule,
 	mirrorNode client.MirrorNode,
@@ -59,6 +63,7 @@ func NewHandler(
 		log.Fatalf("Invalid account id [%s]. Error: [%s]", bridgeAccount, err)
 	}
 	return &Handler{
+		transferRepository: transferRepository,
 		feeRepository:      feeRepository,
 		scheduleRepository: scheduleRepository,
 		mirrorNode:         mirrorNode,
@@ -78,6 +83,12 @@ func (fmh Handler) Handle(payload interface{}) {
 		return
 	}
 
+	receiver, err := hedera.AccountIDFromString(transferMsg.Receiver)
+	if err != nil {
+		fmh.logger.Errorf("[%s] - Failed to parse event account [%s]. Error [%s].", transferMsg.TransactionId, transferMsg.Receiver, err)
+		return
+	}
+
 	transactionRecord, err := fmh.transfersService.InitiateNewTransfer(*transferMsg)
 	if err != nil {
 		fmh.logger.Errorf("[%s] - Error occurred while initiating processing. Error: [%s]", transferMsg.TransactionId, err)
@@ -102,36 +113,59 @@ func (fmh Handler) Handle(payload interface{}) {
 		remainder += calculatedFee - validFee
 	}
 
-	fmh.readOnlyService.FindTransfer(transferMsg.TransactionId, func() (*mirror_node.Response, error) {
-		return fmh.mirrorNode.GetAccountDebitTransactionsAfterTimestampString(fmh.bridgeAccount, transferMsg.Timestamp)
-	}, func(transactionID, scheduleID, status string) error {
-		err := fmh.scheduleRepository.Create(&entity.Schedule{
-			TransactionID: transactionID,
-			ScheduleID:    scheduleID,
-			Operation:     schedule.TRANSFER,
-			Status:        status,
-			TransferID: sql.NullString{
-				String: transferMsg.TransactionId,
-				Valid:  true,
-			},
+	err = fmh.transferRepository.UpdateFee(transferMsg.TransactionId, strconv.FormatInt(validFee, 10))
+	if err != nil {
+		fmh.logger.Errorf("[%s] - Failed to update fee [%d]. Error: [%s]", transferMsg.TransactionId, validFee, err)
+		return
+	}
+
+	transfers, err := fmh.distributorService.CalculateMemberDistribution(validFee)
+	transfers = append(transfers,
+		model.Hedera{
+			AccountID: receiver,
+			Amount:    remainder,
+		})
+
+	splitTransfers := distributor.SplitAccountAmounts(transfers,
+		model.Hedera{
+			AccountID: fmh.bridgeAccount,
+			Amount:    -intAmount,
 		})
-		if err != nil {
-			fmh.logger.Errorf("[%s] - Failed to create scheduled entity [%s]. Error: [%s]", transferMsg.TransactionId, scheduleID, err)
+
+	for _, splitTransfer := range splitTransfers {
+		feeAmount := util.GetTotalFeeFromTransfers(splitTransfer, receiver)
+
+		fmh.readOnlyService.FindAssetTransfer(transferMsg.TransactionId, transferMsg.TargetAsset, splitTransfer, func() (*mirror_node.Response, error) {
+			return fmh.mirrorNode.GetAccountDebitTransactionsAfterTimestampString(fmh.bridgeAccount, transferMsg.Timestamp)
+		}, func(transactionID, scheduleID, status string) error {
+			err := fmh.scheduleRepository.Create(&entity.Schedule{
+				TransactionID: transactionID,
+				ScheduleID:    scheduleID,
+				Operation:     schedule.TRANSFER,
+				Status:        status,
+				TransferID: sql.NullString{
+					String: transferMsg.TransactionId,
+					Valid:  true,
+				},
+			})
+			if err != nil {
+				fmh.logger.Errorf("[%s] - Failed to create scheduled entity [%s]. Error: [%s]", transferMsg.TransactionId, scheduleID, err)
+				return err
+			}
+			err = fmh.feeRepository.Create(&entity.Fee{
+				TransactionID: transactionID,
+				ScheduleID:    scheduleID,
+				Amount:        feeAmount,
+				Status:        status,
+				TransferID: sql.NullString{
+					String: transferMsg.TransactionId,
+					Valid:  true,
+				},
+			})
+			if err != nil {
+				fmh.logger.Errorf("[%s] - Failed to create fee  entity [%s]. Error: [%s]", transferMsg.TransactionId, scheduleID, err)
+			}
 			return err
-		}
-		err = fmh.feeRepository.Create(&entity.Fee{
-			TransactionID: transactionID,
-			ScheduleID:    scheduleID,
-			Amount:        strconv.FormatInt(validFee, 10),
-			Status:        status,
-			TransferID: sql.NullString{
-				String: transferMsg.TransactionId,
-				Valid:  true,
-			},
 		})
-		if err != nil {
-			fmh.logger.Errorf("[%s] - Failed to create fee  entity [%s]. Error: [%s]", transferMsg.TransactionId, scheduleID, err)
-		}
-		return err
-	})
+	}
 }

app/process/handler/read-only/fee-transfer/handler_test.go

@@ -53,7 +53,16 @@ var (
 
 func Test_NewHandler(t *testing.T) {
 	setup()
-	assert.Equal(t, h, NewHandler(mocks.MFeeRepository, mocks.MScheduleRepository, mocks.MHederaMirrorClient, accountId.String(), mocks.MDistributorService, mocks.MFeeService, mocks.MTransferService, mocks.MReadOnlyService))
+	assert.Equal(t, h, NewHandler(
+		mocks.MTransferRepository,
+		mocks.MFeeRepository,
+		mocks.MScheduleRepository,
+		mocks.MHederaMirrorClient,
+		accountId.String(),
+		mocks.MDistributorService,
+		mocks.MFeeService,
+		mocks.MTransferService,
+		mocks.MReadOnlyService))
 }
 
 func Test_Handle(t *testing.T) {
@@ -70,13 +79,13 @@ func Test_Handle(t *testing.T) {
 		Amount:        "100",
 		Status:        status.Initial,
 		Messages:      nil,
-		Fee:           entity.Fee{},
+		Fees:          []entity.Fee{},
 		Schedules:     nil,
 	}
 	mocks.MTransferService.On("InitiateNewTransfer", *tr).Return(tr, nil)
 	mocks.MFeeService.On("CalculateFee", tr.TargetAsset, int64(100)).Return(int64(10), int64(0))
 	mocks.MDistributorService.On("ValidAmount", 10).Return(int64(3))
-	mocks.MReadOnlyService.On("FindTransfer", mock.Anything, mock.Anything, mock.Anything)
+	mocks.MReadOnlyService.On("FindAssetTransfer", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything)
 	h.Handle(tr)
 }
 
@@ -85,7 +94,9 @@ func Test_Handle_FindTransfer(t *testing.T) {
 	mocks.MTransferService.On("InitiateNewTransfer", *tr).Return(&entity.Transfer{Status: status.Initial}, nil)
 	mocks.MFeeService.On("CalculateFee", tr.TargetAsset, int64(100)).Return(int64(10), int64(0))
 	mocks.MDistributorService.On("ValidAmount", int64(10)).Return(int64(3))
-	mocks.MReadOnlyService.On("FindTransfer", mock.Anything, mock.Anything, mock.Anything)
+	mocks.MTransferRepository.On("UpdateFee", tr.TransactionId, "3").Return(nil)
+	mocks.MDistributorService.On("CalculateMemberDistribution", int64(3)).Return([]model.Hedera{})
+	mocks.MReadOnlyService.On("FindAssetTransfer", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything)
 	h.Handle(tr)
 }
 
@@ -119,6 +130,7 @@ func Test_Handle_InitiateNewTransferFails(t *testing.T) {
 func setup() {
 	mocks.Setup()
 	h = &Handler{
+		transferRepository: mocks.MTransferRepository,
 		feeRepository:      mocks.MFeeRepository,
 		scheduleRepository: mocks.MScheduleRepository,
 		mirrorNode:         mocks.MHederaMirrorClient,