Add message filtering capability

plugins/indexing/README.md

@@ -17,6 +17,9 @@ export S3_LARGE_MSG_BUCKET_NAME=""
 export PLUGIN_LOG_FILE=PATH_TO_DESIRED_LOG_FILE
 # Optionally you can also specify the log level, one of "trace", "debug", "info", "warn", "error"
 export PLUGIN_LOG_LEVEL="WARN"
+# Optionally you can specifiy a comma separated list of event types which are allowed to be published on the queue.
+# When omitted it will default to publishing all message types.
+export ALLOWED_MESSAGE_TYPES="block,account"
 ```
 
 Lastly, as we're using SQS and S3 the node needs access to a valid set of AWS credentials with permission to publish messages to the specified queue and upload access to the specified bucket.

plugins/indexing/pluginaws/message_filter.go

@@ -0,0 +1,30 @@
+package pluginaws
+
+import (
+	"slices"
+
+	"github.com/sedaprotocol/seda-chain/plugins/indexing/types"
+)
+
+func (sc *SqsClient) filterMessages(data []*types.Message) []*types.Message {
+	allowedMessages := make([]*types.Message, 0, len(data))
+
+	for _, message := range data {
+		if sc.isMessageAllowed(message) {
+			allowedMessages = append(allowedMessages, message)
+		} else {
+			sc.logger.Trace("skipping message", "type", message.Type)
+		}
+	}
+
+	return allowedMessages
+}
+
+func (sc *SqsClient) isMessageAllowed(event *types.Message) bool {
+	// When no allowlist is specified assume everything is allowed.
+	if len(sc.allowedMessages) == 0 {
+		return true
+	}
+
+	return slices.Contains(sc.allowedMessages, event.Type)
+}

plugins/indexing/pluginaws/sqs_client.go

@@ -16,7 +16,8 @@ const (
 func (sc *SqsClient) PublishToQueue(data []*types.Message) error {
 	sc.logger.Trace("publishing to queue", "size", len(data))
 
-	sizedBatchEntries, err := sc.createSizedBatchEntries(data)
+	allowedMessages := sc.filterMessages(data)
+	sizedBatchEntries, err := sc.createSizedBatchEntries(allowedMessages)
 	if err != nil {
 		return err
 	}

plugins/indexing/pluginaws/types.go

@@ -3,6 +3,7 @@ package pluginaws
 import (
 	"fmt"
 	"os"
+	"strings"
 
 	"github.com/aws/aws-sdk-go/service/s3"
 	"github.com/aws/aws-sdk-go/service/sqs"
@@ -11,16 +12,18 @@ import (
 )
 
 var (
-	queueURLEnvName  = "SQS_QUEUE_URL"
-	bucketURLEnvName = "S3_LARGE_MSG_BUCKET_NAME"
+	queueURLEnvName      = "SQS_QUEUE_URL"
+	bucketURLEnvName     = "S3_LARGE_MSG_BUCKET_NAME"
+	messageFilterEnvName = "ALLOWED_MESSAGE_TYPES"
 )
 
 type SqsClient struct {
-	bucketName string
-	queueURL   string
-	sqsClient  *sqs.SQS
-	s3Client   *s3.S3
-	logger     *log.Logger
+	bucketName      string
+	queueURL        string
+	allowedMessages []string
+	sqsClient       *sqs.SQS
+	s3Client        *s3.S3
+	logger          *log.Logger
 }
 
 func NewSqsClient(logger *log.Logger) (*SqsClient, error) {
@@ -34,6 +37,12 @@ func NewSqsClient(logger *log.Logger) (*SqsClient, error) {
 		return nil, fmt.Errorf("missing environment variable '%s'", bucketURLEnvName)
 	}
 
+	allowedMessages := make([]string, 0)
+	messageFilterString, found := os.LookupEnv(messageFilterEnvName)
+	if found {
+		allowedMessages = parseMessageFilterString((messageFilterString))
+	}
+
 	sess, err := NewSession()
 	if err != nil {
 		return nil, fmt.Errorf("failed to initialise session: %w", err)
@@ -47,10 +56,15 @@ func NewSqsClient(logger *log.Logger) (*SqsClient, error) {
 	awsS3Client := s3.New(sess, s3Config)
 
 	return &SqsClient{
-		queueURL:   queueURL,
-		bucketName: bucketName,
-		sqsClient:  awsSqsClient,
-		s3Client:   awsS3Client,
-		logger:     logger,
+		queueURL:        queueURL,
+		bucketName:      bucketName,
+		allowedMessages: allowedMessages,
+		sqsClient:       awsSqsClient,
+		s3Client:        awsS3Client,
+		logger:          logger,
 	}, nil
 }
+
+func parseMessageFilterString(data string) []string {
+	return strings.Split(data, ",")
+}