From d011d4086819fead904ae36b50f9c438dcf8b3e5 Mon Sep 17 00:00:00 2001
From: Kunal Gupta <39487888+iamKunalGupta@users.noreply.github.com>
Date: Thu, 7 Mar 2024 04:29:59 +0530
Subject: [PATCH 1/4] fix(telemtry): issue with aws sns

---
 flow/shared/telemetry/sns_message_sender.go | 11 +++++++++--
 1 file changed, 9 insertions(+), 2 deletions(-)

diff --git a/flow/shared/telemetry/sns_message_sender.go b/flow/shared/telemetry/sns_message_sender.go
index 42bdd026a..022889aee 100644
--- a/flow/shared/telemetry/sns_message_sender.go
+++ b/flow/shared/telemetry/sns_message_sender.go
@@ -5,6 +5,7 @@ import (
 	"crypto/sha256"
 	"encoding/hex"
 	"strings"
+	"unicode"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/config"
@@ -37,7 +38,13 @@ func (s *SNSMessageSenderImpl) SendMessage(ctx context.Context, subject string,
 	h := sha256.New()
 	h.Write([]byte(deduplicationString))
 	deduplicationHash := hex.EncodeToString(h.Sum(nil))
-
+	// AWS SNS Subject constraints
+	messageSubject := strings.TrimFunc(subject, func(r rune) bool {
+		return !unicode.IsPrint(r)
+	})
+	if len(messageSubject) > 100 {
+		messageSubject = messageSubject[:99]
+	}
 	publish, err := s.client.Publish(ctx, &sns.PublishInput{
 		Message: aws.String(body),
 		MessageAttributes: map[string]types.MessageAttributeValue{
@@ -66,7 +73,7 @@ func (s *SNSMessageSenderImpl) SendMessage(ctx context.Context, subject string,
 				StringValue: aws.String(deduplicationHash),
 			},
 		},
-		Subject:  aws.String(subject[:100]),
+		Subject:  aws.String(messageSubject),
 		TopicArn: aws.String(s.topic),
 	})
 	if err != nil {

From bc2f5dac269bdd2b249e00841608cd4671071953 Mon Sep 17 00:00:00 2001
From: Kunal Gupta <39487888+iamKunalGupta@users.noreply.github.com>
Date: Thu, 7 Mar 2024 04:30:58 +0530
Subject: [PATCH 2/4] chore: small fix

---
 flow/shared/telemetry/sns_message_sender.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/flow/shared/telemetry/sns_message_sender.go b/flow/shared/telemetry/sns_message_sender.go
index 022889aee..4289773f3 100644
--- a/flow/shared/telemetry/sns_message_sender.go
+++ b/flow/shared/telemetry/sns_message_sender.go
@@ -42,7 +42,7 @@ func (s *SNSMessageSenderImpl) SendMessage(ctx context.Context, subject string,
 	messageSubject := strings.TrimFunc(subject, func(r rune) bool {
 		return !unicode.IsPrint(r)
 	})
-	if len(messageSubject) > 100 {
+	if len(messageSubject) >= 100 {
 		messageSubject = messageSubject[:99]
 	}
 	publish, err := s.client.Publish(ctx, &sns.PublishInput{

From e85310b7d21a00bdf0a1922181f31291015f8dee Mon Sep 17 00:00:00 2001
From: Kunal Gupta <39487888+iamKunalGupta@users.noreply.github.com>
Date: Thu, 7 Mar 2024 04:40:16 +0530
Subject: [PATCH 3/4] fix: string cleaning

---
 flow/shared/telemetry/sns_message_sender.go | 14 +++++++++-----
 1 file changed, 9 insertions(+), 5 deletions(-)

diff --git a/flow/shared/telemetry/sns_message_sender.go b/flow/shared/telemetry/sns_message_sender.go
index 4289773f3..7cdd83e70 100644
--- a/flow/shared/telemetry/sns_message_sender.go
+++ b/flow/shared/telemetry/sns_message_sender.go
@@ -39,11 +39,15 @@ func (s *SNSMessageSenderImpl) SendMessage(ctx context.Context, subject string,
 	h.Write([]byte(deduplicationString))
 	deduplicationHash := hex.EncodeToString(h.Sum(nil))
 	// AWS SNS Subject constraints
-	messageSubject := strings.TrimFunc(subject, func(r rune) bool {
-		return !unicode.IsPrint(r)
-	})
-	if len(messageSubject) >= 100 {
-		messageSubject = messageSubject[:99]
+	messageSubject := ""
+	for _, char := range subject {
+		if unicode.IsPrint(char) {
+			messageSubject += string(char)
+		}
+	}
+	maxSubjectSize := 99
+	if len(messageSubject) > maxSubjectSize {
+		messageSubject = messageSubject[:maxSubjectSize]
 	}
 	publish, err := s.client.Publish(ctx, &sns.PublishInput{
 		Message: aws.String(body),

From 2361ed0b1215f8a3463414c98304970f3f6c8458 Mon Sep 17 00:00:00 2001
From: Kunal Gupta <39487888+iamKunalGupta@users.noreply.github.com>
Date: Thu, 7 Mar 2024 04:52:42 +0530
Subject: [PATCH 4/4] refactor: combine loop to prevent slicing

---
 flow/shared/telemetry/sns_message_sender.go | 18 ++++++++++--------
 1 file changed, 10 insertions(+), 8 deletions(-)

diff --git a/flow/shared/telemetry/sns_message_sender.go b/flow/shared/telemetry/sns_message_sender.go
index 7cdd83e70..67cf3eeba 100644
--- a/flow/shared/telemetry/sns_message_sender.go
+++ b/flow/shared/telemetry/sns_message_sender.go
@@ -39,15 +39,17 @@ func (s *SNSMessageSenderImpl) SendMessage(ctx context.Context, subject string,
 	h.Write([]byte(deduplicationString))
 	deduplicationHash := hex.EncodeToString(h.Sum(nil))
 	// AWS SNS Subject constraints
-	messageSubject := ""
-	for _, char := range subject {
+	var messageSubjectBuilder strings.Builder
+	maxSubjectSize := 99
+	for currentLength, char := range subject {
 		if unicode.IsPrint(char) {
-			messageSubject += string(char)
+			messageSubjectBuilder.WriteRune(char)
+		} else {
+			messageSubjectBuilder.WriteRune(' ')
+		}
+		if currentLength > maxSubjectSize {
+			break
 		}
-	}
-	maxSubjectSize := 99
-	if len(messageSubject) > maxSubjectSize {
-		messageSubject = messageSubject[:maxSubjectSize]
 	}
 	publish, err := s.client.Publish(ctx, &sns.PublishInput{
 		Message: aws.String(body),
@@ -77,7 +79,7 @@ func (s *SNSMessageSenderImpl) SendMessage(ctx context.Context, subject string,
 				StringValue: aws.String(deduplicationHash),
 			},
 		},
-		Subject:  aws.String(messageSubject),
+		Subject:  aws.String(messageSubjectBuilder.String()),
 		TopicArn: aws.String(s.topic),
 	})
 	if err != nil {