fix(events): SNS universal target message attributes

Author: ctnnguyen

src/main/java/io/github/hectorvent/floci/services/scheduler/ScheduleInvoker.java

@@ -10,6 +10,7 @@
 import io.github.hectorvent.floci.services.scheduler.model.Target;
 import io.github.hectorvent.floci.services.sns.SnsService;
 import io.github.hectorvent.floci.services.sqs.SqsService;
+import io.github.hectorvent.floci.services.sqs.model.MessageAttributeValue;
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
 import org.jboss.logging.Logger;
@@ -114,8 +115,9 @@ private void invokeUniversalTarget(String serviceAction, String input, String re
                 String subject = text(params, "Subject");
                 String messageGroupId = text(params, "MessageGroupId");
                 String messageDeduplicationId = text(params, "MessageDeduplicationId");
+                Map<String, MessageAttributeValue> messageAttributes = parseMessageAttributes(params);
                 String snsRegion = extractRegion(topicArn != null ? topicArn : targetArn, region);
-                snsService.publish(topicArn, targetArn, null, message, subject, null,
+                snsService.publish(topicArn, targetArn, null, message, subject, messageAttributes,
                         messageGroupId, messageDeduplicationId, snsRegion);
                 LOG.debugv("Scheduler delivered to SNS (universal target): {0}", topicArn);
             }
@@ -131,6 +133,20 @@ private void invokeUniversalTarget(String serviceAction, String input, String re
         }
     }
 
+    private Map<String, MessageAttributeValue> parseMessageAttributes(JsonNode params) {
+        JsonNode attrsNode = params.path("MessageAttributes");
+        if (!attrsNode.isObject()) {
+            return null;
+        }
+        Map<String, MessageAttributeValue> attributes = new HashMap<>();
+        attrsNode.fields().forEachRemaining(entry -> {
+            String dataType = entry.getValue().path("DataType").asText("String");
+            String stringValue = entry.getValue().path("StringValue").asText(null);
+            attributes.put(entry.getKey(), new MessageAttributeValue(stringValue, dataType));
+        });
+        return attributes.isEmpty() ? null : attributes;
+    }
+
     private static String text(JsonNode node, String field) {
         JsonNode value = node.get(field);
         return value != null && !value.isNull() ? value.asText() : null;

src/test/java/io/github/hectorvent/floci/services/scheduler/ScheduleInvokerTest.java

@@ -8,18 +8,14 @@
 import io.github.hectorvent.floci.services.scheduler.model.Target;
 import io.github.hectorvent.floci.services.sns.SnsService;
 import io.github.hectorvent.floci.services.sqs.SqsService;
+import io.github.hectorvent.floci.services.sqs.model.MessageAttributeValue;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyInt;
-import static org.mockito.ArgumentMatchers.anyString;
-import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.ArgumentMatchers.isNull;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
+import java.util.Map;
+
+import static org.mockito.ArgumentMatchers.*;
+import static org.mockito.Mockito.*;
 
 class ScheduleInvokerTest {
 
@@ -43,6 +39,45 @@ void setUp() {
                 eventBridgeService, new ObjectMapper(), config);
     }
 
+    @Test
+    void universalSnsPublishForwardsMessageAttributes() {
+        Target target = new Target();
+        target.setArn("arn:aws:scheduler:::aws-sdk:sns:publish");
+        target.setRoleArn("arn:aws:iam::000000000000:role/x");
+        target.setInput("{\"TopicArn\":\"" + TOPIC_ARN + "\","
+                + "\"Message\":\"{}\","
+                + "\"Subject\":\"my-subject\","
+                + "\"MessageAttributes\":{"
+                + "\"EventName\":{\"DataType\":\"String\",\"StringValue\":\"my-subject\"}"
+                + "}}");
+
+        invoker.invoke(target, "us-east-1");
+
+        verify(snsService).publish(
+                eq(TOPIC_ARN), isNull(), isNull(),
+                eq("{}"), eq("my-subject"),
+                argThat((Map<String, MessageAttributeValue> attrs) ->
+                        attrs != null
+                        && attrs.containsKey("EventName")
+                        && "my-subject".equals(attrs.get("EventName").getStringValue())
+                        && "String".equals(attrs.get("EventName").getDataType())),
+                isNull(), isNull(), eq("us-east-1"));
+    }
+
+    @Test
+    void universalSnsPublishWithNoMessageAttributesPassesNull() {
+        Target target = new Target();
+        target.setArn("arn:aws:scheduler:::aws-sdk:sns:publish");
+        target.setRoleArn("arn:aws:iam::000000000000:role/x");
+        target.setInput("{\"TopicArn\":\"" + TOPIC_ARN + "\",\"Message\":\"hello\"}");
+
+        invoker.invoke(target, "us-east-1");
+
+        verify(snsService).publish(eq(TOPIC_ARN), isNull(), isNull(),
+                eq("hello"), isNull(), isNull(),
+                isNull(), isNull(), eq("us-east-1"));
+    }
+
     @Test
     void universalSnsPublishReadsTopicArnAndMessageFromInput() {
         Target target = new Target();