enable committer message compression per record

Author: nischitpra

internal/storage/kafka_publisher.go

@@ -10,6 +10,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/DataDog/zstd"
 	"github.com/rs/zerolog/log"
 	config "github.com/thirdweb-dev/indexer/configs"
 	"github.com/thirdweb-dev/indexer/internal/common"
@@ -306,19 +307,27 @@ func (p *KafkaPublisher) createBlockRevertMessage(chainId uint64, blockNumber ui
 }
 
 func (p *KafkaPublisher) createRecord(msgType MessageType, chainId uint64, blockNumber uint64, timestamp time.Time, msgJson []byte) (*kgo.Record, error) {
-	// Create headers with metadata
+	// Compress the JSON data using zstd
+	compressedData, err := zstd.Compress(nil, msgJson)
+	if err != nil {
+		return nil, fmt.Errorf("failed to compress message data: %v", err)
+	}
+
+	// Create headers with metadata including compression info
 	headers := []kgo.RecordHeader{
 		{Key: "chain_id", Value: []byte(fmt.Sprintf("%d", chainId))},
 		{Key: "block_number", Value: []byte(fmt.Sprintf("%d", blockNumber))},
 		{Key: "type", Value: []byte(fmt.Sprintf("%s", msgType))},
 		{Key: "timestamp", Value: []byte(timestamp.Format(time.RFC3339Nano))},
 		{Key: "schema_version", Value: []byte("1")},
+		{Key: "original_size", Value: []byte(fmt.Sprintf("%d", len(msgJson)))},
+		{Key: "content-encoding", Value: []byte("zstd")},
 	}
 
 	return &kgo.Record{
 		Topic:     fmt.Sprintf("insight.commit.blocks.%d", chainId),
 		Key:       []byte(fmt.Sprintf("%d:%s:%d", chainId, msgType, blockNumber)),
-		Value:     msgJson,
+		Value:     compressedData,
 		Headers:   headers,
 		Partition: 0,
 	}, nil