Implement transfer threshold

cmd/process.go

@@ -7,6 +7,8 @@ import (
 	"time"
 
 	"cosmossdk.io/log"
+	"cosmossdk.io/math"
+	cctptypes "github.com/circlefin/noble-cctp/x/cctp/types"
 	"github.com/spf13/cobra"
 	"github.com/strangelove-ventures/noble-cctp-relayer/circle"
 	"github.com/strangelove-ventures/noble-cctp-relayer/types"
@@ -91,7 +93,8 @@ func StartProcessor(
 
 			// if a filter's condition is met, mark as filtered
 			if FilterDisabledCCTPRoutes(cfg, logger, msg) ||
-				filterInvalidDestinationCallers(registeredDomains, logger, msg) {
+				filterInvalidDestinationCallers(registeredDomains, logger, msg) ||
+				filterLowTransfers(cfg, logger, msg) {
 				msg.Status = types.Filtered
 			}
 
@@ -184,6 +187,35 @@ func filterInvalidDestinationCallers(registeredDomains map[types.Domain]types.Ch
 	return !chain.IsDestinationCaller(msg.DestinationCaller)
 }
 
+func filterLowTransfers(cfg *types.Config, logger log.Logger, msg *types.MessageState) bool {
+	bm, err := new(cctptypes.BurnMessage).Parse(msg.MsgBody)
+	if err != nil {
+		logger.Info("This is not a burn message", "err", err)
+		return true
+	}
+
+	if bm.Amount.LT(math.NewIntFromUint64(cfg.MinAmount)) {
+		logger.Info(
+			"Filtered tx because the transfer amount is less than the minimum allowed amount",
+			"source_domain", msg.SourceDomain,
+			"source_tx", msg.SourceTxHash,
+			"amount", bm.Amount,
+			"min_amount", cfg.MinAmount,
+		)
+		return true
+	}
+
+	logger.Info(
+		"Not filtering tx due to low transfer amount",
+		"source_domain", msg.SourceDomain,
+		"source_tx", msg.SourceTxHash,
+		"amount", bm.Amount.Uint64(),
+		"min_amount", cfg.MinAmount,
+	)
+
+	return false
+}
+
 func LookupKey(sourceTxHash string) string {
 	// return fmt.Sprintf("%s-%s", sourceTxHash, messageType)
 	return sourceTxHash

cmd/root.go

@@ -112,6 +112,7 @@ func Parse(file string) (*types.Config, error) {
 		ProcessorWorkerCount: cfg.ProcessorWorkerCount,
 		Api:                  cfg.Api,
 		Chains:               make(map[string]types.ChainConfig),
+		MinAmount:            cfg.MinAmount,
 	}
 
 	for name, chain := range cfg.Chains {

noble/message_state.go

@@ -66,6 +66,7 @@ func txToMessageState(tx *ctypes.ResultTx) ([]*types.MessageState, error) {
 						Nonce:             msg.Nonce,
 						SourceTxHash:      tx.Hash.String(),
 						MsgSentBytes:      rawMessageSentBytes,
+						MsgBody:           msg.MessageBody,
 						DestinationCaller: msg.DestinationCaller,
 						Created:           now,
 						Updated:           now,

types/config.go

@@ -12,6 +12,7 @@ type Config struct {
 	Api                  struct {
 		TrustedProxies []string `yaml:"trusted-proxies"`
 	} `yaml:"api"`
+	MinAmount uint64 `yaml:"min-amount"`
 }
 
 type ConfigWrapper struct {
@@ -26,6 +27,7 @@ type ConfigWrapper struct {
 	Api                  struct {
 		TrustedProxies []string `yaml:"trusted-proxies"`
 	} `yaml:"api"`
+	MinAmount uint64 `yaml:"min-amount"`
 }
 
 type ChainConfig interface {

types/message_state.go

@@ -41,6 +41,7 @@ type MessageState struct {
 	SourceTxHash      string
 	DestTxHash        string
 	MsgSentBytes      []byte // bytes of the MessageSent message transmitter event
+	MsgBody           []byte // bytes of the MessageBody
 	DestinationCaller []byte // address authorized to call transaction
 	Channel           string // "channel-%d" if a forward, empty if not a forward
 	Created           time.Time
@@ -68,6 +69,7 @@ func EvmLogToMessageState(abi abi.ABI, messageSent abi.Event, log *ethtypes.Log)
 		DestDomain:        Domain(message.DestinationDomain),
 		SourceTxHash:      log.TxHash.Hex(),
 		MsgSentBytes:      rawMessageSentBytes,
+		MsgBody:           message.MessageBody,
 		DestinationCaller: message.DestinationCaller,
 		Nonce:             message.Nonce,
 		Created:           time.Now(),