fix: Send system metadata to destrabbitmq & destawssqs (#297)

Author: alexluong

internal/destregistry/basepublisher.go

@@ -1,8 +1,12 @@
 package destregistry
 
 import (
+	"fmt"
 	"sync"
 	"sync/atomic"
+	"time"
+
+	"github.com/hookdeck/outpost/internal/models"
 )
 
 // BasePublisher provides common publisher functionality
@@ -30,3 +34,19 @@ func (p *BasePublisher) StartClose() {
 	p.closed.Store(true)
 	p.active.Wait()
 }
+
+func (p *BasePublisher) MakeMetadata(event *models.Event, timestamp time.Time) map[string]string {
+	systemMetadata := map[string]string{
+		"timestamp": fmt.Sprintf("%d", timestamp.Unix()),
+		"event-id":  event.ID,
+		"topic":     event.Topic,
+	}
+	metadata := make(map[string]string)
+	for k, v := range systemMetadata {
+		metadata[k] = v
+	}
+	for k, v := range event.Metadata {
+		metadata[k] = v
+	}
+	return metadata
+}

internal/destregistry/providers/destawssqs/destawssqs.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"net/url"
 	"strings"
+	"time"
 
 	awssdk "github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/config"
@@ -139,18 +140,21 @@ func (p *AWSSQSPublisher) Format(ctx context.Context, event *models.Event) (*sqs
 		return nil, err
 	}
 
-	attrs := make(map[string]types.MessageAttributeValue)
-	for k, v := range event.Metadata {
-		attrs[k] = types.MessageAttributeValue{
-			DataType:    aws.String("String"),
-			StringValue: aws.String(v),
-		}
+	metadata := p.BasePublisher.MakeMetadata(event, time.Now())
+	metadataBytes, err := json.Marshal(metadata)
+	if err != nil {
+		return nil, err
 	}
 
 	return &sqs.SendMessageInput{
-		QueueUrl:          awssdk.String(p.queueURL),
-		MessageBody:       awssdk.String(string(dataBytes)),
-		MessageAttributes: attrs,
+		QueueUrl:    awssdk.String(p.queueURL),
+		MessageBody: awssdk.String(string(dataBytes)),
+		MessageAttributes: map[string]types.MessageAttributeValue{
+			"metadata": {
+				DataType:    aws.String("String"),
+				StringValue: aws.String(string(metadataBytes)),
+			},
+		},
 	}, nil
 }
 

internal/destregistry/providers/destawssqs/destawssqs_publish_test.go

@@ -2,15 +2,18 @@ package destawssqs_test
 
 import (
 	"context"
+	"encoding/json"
 	"testing"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/sqs"
 	"github.com/hookdeck/outpost/internal/destregistry/providers/destawssqs"
 	testsuite "github.com/hookdeck/outpost/internal/destregistry/testing"
+	"github.com/hookdeck/outpost/internal/models"
 	"github.com/hookdeck/outpost/internal/util/awsutil"
 	"github.com/hookdeck/outpost/internal/util/testinfra"
 	"github.com/hookdeck/outpost/internal/util/testutil"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 )
@@ -52,8 +55,10 @@ func (c *SQSConsumer) consume() {
 
 			for _, msg := range result.Messages {
 				metadata := make(map[string]string)
-				for k, v := range msg.MessageAttributes {
-					metadata[k] = *v.StringValue
+				if metaAttr, ok := msg.MessageAttributes["metadata"]; ok {
+					if err := json.Unmarshal([]byte(*metaAttr.StringValue), &metadata); err != nil {
+						continue
+					}
 				}
 
 				c.msgChan <- testsuite.Message{
@@ -81,6 +86,23 @@ func (c *SQSConsumer) Close() error {
 	return nil
 }
 
+type SQSAsserter struct{}
+
+func (a *SQSAsserter) AssertMessage(t testsuite.TestingT, msg testsuite.Message, event models.Event) {
+	// Metadata is already parsed in the consumer
+	metadata := msg.Metadata
+
+	// Verify system metadata
+	assert.NotEmpty(t, metadata["timestamp"], "timestamp should be present")
+	assert.Equal(t, event.ID, metadata["event-id"], "event-id should match")
+	assert.Equal(t, event.Topic, metadata["topic"], "topic should match")
+
+	// Verify custom metadata
+	for k, v := range event.Metadata {
+		assert.Equal(t, v, metadata[k], "metadata key %s should match expected value", k)
+	}
+}
+
 type AWSSQSSuite struct {
 	testsuite.PublisherSuite
 	consumer *SQSConsumer
@@ -122,11 +144,12 @@ func (s *AWSSQSSuite) SetupSuite() {
 		}),
 	)
 
-	// Initialize publisher suite
+	// Initialize publisher suite with custom asserter
 	cfg := testsuite.Config{
 		Provider: provider,
 		Dest:     &destination,
 		Consumer: s.consumer,
+		Asserter: &SQSAsserter{},
 	}
 	s.InitSuite(cfg)
 }

internal/destregistry/providers/destrabbitmq/destrabbitmq.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"strings"
 	"sync"
+	"time"
 
 	"github.com/hookdeck/outpost/internal/destregistry"
 	"github.com/hookdeck/outpost/internal/destregistry/metadata"
@@ -150,7 +151,8 @@ func (p *RabbitMQPublisher) Publish(ctx context.Context, event *models.Event) (*
 	}
 
 	headers := make(amqp091.Table)
-	for k, v := range event.Metadata {
+	metadata := p.BasePublisher.MakeMetadata(event, time.Now())
+	for k, v := range metadata {
 		headers[k] = v
 	}
 

internal/destregistry/providers/destrabbitmq/destrabbitmq_publish_test.go

@@ -143,14 +143,17 @@ func (a *RabbitMQAsserter) AssertMessage(t testsuite.TestingT, msg testsuite.Mes
 	// Assert RabbitMQ-specific properties
 	assert.Equal(t, "application/json", delivery.ContentType)
 	assert.Equal(t, event.Topic, delivery.RoutingKey, "routing key should match event topic")
-	// assert.NotEmpty(t, delivery.MessageId)
-	// assert.NotEmpty(t, delivery.Timestamp)
-
-	// Could add more RabbitMQ-specific assertions:
-	// - Exchange routing
-	// - Message persistence
-	// - Priority
-	// - etc.
+
+	// Verify system metadata
+	metadata := msg.Metadata
+	assert.NotEmpty(t, metadata["timestamp"], "timestamp should be present")
+	assert.Equal(t, event.ID, metadata["event-id"], "event-id should match")
+	assert.Equal(t, event.Topic, metadata["topic"], "topic should match")
+
+	// Verify custom metadata
+	for k, v := range event.Metadata {
+		assert.Equal(t, v, metadata[k], "metadata key %s should match expected value", k)
+	}
 }
 
 // RabbitMQPublishSuite reimplements the publish tests using the shared test suite