Merge pull request #495 from hookdeck/delivery-timestamp

fix: delivery timestamp metadata in unix seconds

Author: leggetter

internal/destregistry/basepublisher.go

@@ -37,7 +37,7 @@ func (p *BasePublisher) StartClose() {
 
 func (p *BasePublisher) MakeMetadata(event *models.Event, timestamp time.Time) map[string]string {
 	systemMetadata := map[string]string{
-		"timestamp": fmt.Sprintf("%d", timestamp.UnixMilli()),
+		"timestamp": fmt.Sprintf("%d", timestamp.Unix()),
 		"event-id":  event.ID,
 		"topic":     event.Topic,
 	}

internal/destregistry/providers/destawskinesis/destawskinesis_publish_test.go

@@ -204,6 +204,7 @@ func (a *KinesisAsserter) AssertMessage(t testsuite.TestingT, msg testsuite.Mess
 
 	// Verify system metadata
 	assert.NotEmpty(t, metadata["timestamp"], "timestamp should be present")
+	testsuite.AssertTimestampIsUnixSeconds(t, metadata["timestamp"])
 	assert.Equal(t, event.ID, metadata["event-id"], "event-id should match")
 	assert.Equal(t, event.Topic, metadata["topic"], "topic should match")
 

internal/destregistry/providers/destawss3/destawss3_publish_test.go

@@ -140,6 +140,7 @@ func (a *S3Asserter) AssertMessage(t testsuite.TestingT, msg testsuite.Message,
 	// 2. Assert system metadata is present
 	metadata := msg.Metadata
 	assert.NotEmpty(t, metadata["timestamp"], "timestamp should be present")
+	testsuite.AssertTimestampIsUnixSeconds(t, metadata["timestamp"])
 	assert.Equal(t, event.ID, metadata["event-id"], "event-id should match")
 	assert.Equal(t, event.Topic, metadata["topic"], "topic should match")
 

internal/destregistry/providers/destawssqs/destawssqs_publish_test.go

@@ -94,6 +94,7 @@ func (a *SQSAsserter) AssertMessage(t testsuite.TestingT, msg testsuite.Message,
 
 	// Verify system metadata
 	assert.NotEmpty(t, metadata["timestamp"], "timestamp should be present")
+	testsuite.AssertTimestampIsUnixSeconds(t, metadata["timestamp"])
 	assert.Equal(t, event.ID, metadata["event-id"], "event-id should match")
 	assert.Equal(t, event.Topic, metadata["topic"], "topic should match")
 

internal/destregistry/providers/destazureservicebus/destazureservicebus_publish_test.go

@@ -119,6 +119,7 @@ func (a *AzureServiceBusAsserter) AssertMessage(t testsuite.TestingT, msg testsu
 
 	// Verify system metadata
 	assert.NotEmpty(t, metadata["timestamp"], "timestamp should be present")
+	testsuite.AssertTimestampIsUnixSeconds(t, metadata["timestamp"])
 	assert.Equal(t, event.ID, metadata["event-id"], "event-id should match")
 	assert.Equal(t, event.Topic, metadata["topic"], "topic should match")
 

internal/destregistry/providers/destrabbitmq/destrabbitmq_publish_test.go

@@ -147,6 +147,7 @@ func (a *RabbitMQAsserter) AssertMessage(t testsuite.TestingT, msg testsuite.Mes
 	// Verify system metadata
 	metadata := msg.Metadata
 	assert.NotEmpty(t, metadata["timestamp"], "timestamp should be present")
+	testsuite.AssertTimestampIsUnixSeconds(t, metadata["timestamp"])
 	assert.Equal(t, event.ID, metadata["event-id"], "event-id should match")
 	assert.Equal(t, event.Topic, metadata["topic"], "topic should match")
 

internal/destregistry/providers/destwebhook/destwebhook.go

@@ -557,16 +557,31 @@ func (p *WebhookPublisher) Format(ctx context.Context, event *models.Event) (*ht
 
 	req.Header.Set("Content-Type", "application/json")
 
-	// Add default headers unless disabled
-	if !p.disableTimestampHeader {
-		req.Header.Set(p.headerPrefix+"timestamp", fmt.Sprintf("%d", now.UnixMilli()))
-	}
-	if !p.disableEventIDHeader {
-		req.Header.Set(p.headerPrefix+"event-id", event.ID)
-	}
-	if !p.disableTopicHeader {
-		req.Header.Set(p.headerPrefix+"topic", event.Topic)
+	// Get merged metadata (system + event metadata) using BasePublisher
+	metadata := p.BasePublisher.MakeMetadata(event, now)
+
+	// Add headers from metadata, respecting disable flags
+	for key, value := range metadata {
+		// Check if this specific system header should be disabled
+		switch key {
+		case "timestamp":
+			if p.disableTimestampHeader {
+				continue
+			}
+		case "event-id":
+			if p.disableEventIDHeader {
+				continue
+			}
+		case "topic":
+			if p.disableTopicHeader {
+				continue
+			}
+		}
+		// Add the header with the appropriate prefix
+		req.Header.Set(p.headerPrefix+key, value)
 	}
+
+	// Add signature header if not disabled
 	if !p.disableSignatureHeader {
 		signatureHeader := p.sm.GenerateSignatureHeader(SignaturePayload{
 			EventID:   event.ID,
@@ -579,11 +594,6 @@ func (p *WebhookPublisher) Format(ctx context.Context, event *models.Event) (*ht
 		}
 	}
 
-	// Add metadata headers with the specified prefix
-	for key, value := range event.Metadata {
-		req.Header.Set(p.headerPrefix+strings.ToLower(key), value)
-	}
-
 	return req, nil
 }
 

internal/destregistry/providers/destwebhook/destwebhook_publish_test.go

@@ -91,7 +91,12 @@ func (a *WebhookAsserter) AssertMessage(t testsuite.TestingT, msg testsuite.Mess
 	assert.Equal(t, "application/json", req.Header.Get("Content-Type"))
 
 	// Verify default headers
-	assert.NotEmpty(t, req.Header.Get(a.headerPrefix+"timestamp"), "timestamp header should be present")
+	timestampHeader := req.Header.Get(a.headerPrefix + "timestamp")
+	assert.NotEmpty(t, timestampHeader, "timestamp header should be present")
+
+	// Verify timestamp is in Unix seconds (not milliseconds)
+	testsuite.AssertTimestampIsUnixSeconds(t, timestampHeader)
+
 	assert.Equal(t, event.ID, req.Header.Get(a.headerPrefix+"event-id"), "event-id header should match")
 	assert.Equal(t, event.Topic, req.Header.Get(a.headerPrefix+"topic"), "topic header should match")
 
@@ -105,6 +110,13 @@ func (a *WebhookAsserter) AssertMessage(t testsuite.TestingT, msg testsuite.Mess
 		signatureHeader := req.Header.Get(a.headerPrefix + "signature")
 		assertSignatureFormat(t, signatureHeader, a.expectedSignatures)
 
+		// Verify timestamp in signature header matches the timestamp header
+		signatureParts := strings.SplitN(signatureHeader, ",", 2)
+		if len(signatureParts) >= 2 {
+			signatureTimestampStr := strings.TrimPrefix(signatureParts[0], "t=")
+			assert.Equal(t, timestampHeader, signatureTimestampStr, "timestamp in signature header should match timestamp header")
+		}
+
 		// Verify each expected signature
 		for _, secret := range a.secrets {
 			assertValidSignature(t, secret, msg.Data, signatureHeader)

internal/destregistry/testing/publisher_suite.go

@@ -4,13 +4,15 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"strconv"
 	"sync"
 	"sync/atomic"
 	"time"
 
 	"github.com/hookdeck/outpost/internal/destregistry"
 	"github.com/hookdeck/outpost/internal/models"
 	"github.com/hookdeck/outpost/internal/util/testutil"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/suite"
 )
 
@@ -35,6 +37,39 @@ type TestingT interface {
 	Helper()
 }
 
+// AssertTimestampIsUnixSeconds verifies that a timestamp string is in Unix seconds format (not milliseconds).
+// It checks if the timestamp is within a reasonable range for Unix seconds (between year 2000 and 2100).
+func AssertTimestampIsUnixSeconds(t TestingT, timestampStr string, msgAndArgs ...interface{}) {
+	t.Helper()
+
+	timestampInt, err := strconv.ParseInt(timestampStr, 10, 64)
+	assert.NoError(t, err, "timestamp should be a valid integer")
+
+	// Check if timestamp is in a reasonable range for Unix seconds
+	// Year 2000: ~946,684,800
+	// Year 2100: ~4,102,444,800
+	// Current time in seconds: ~1,700,000,000 (2023-2024)
+	// Current time in millis:  ~1,700,000,000,000
+
+	minUnixSeconds := int64(946684800)     // Jan 1, 2000
+	maxUnixSeconds := int64(4102444800)    // Jan 1, 2100
+
+	if timestampInt < minUnixSeconds || timestampInt > maxUnixSeconds {
+		// Likely milliseconds - check if dividing by 1000 gives a reasonable timestamp
+		possibleSeconds := timestampInt / 1000
+		if possibleSeconds >= minUnixSeconds && possibleSeconds <= maxUnixSeconds {
+			assert.Fail(t, "timestamp appears to be in milliseconds, expected Unix seconds",
+				"timestamp %d is likely in milliseconds (would be %s if converted to seconds), expected Unix seconds (around %s)",
+				timestampInt,
+				time.Unix(possibleSeconds, 0).Format(time.RFC3339),
+				time.Now().Format(time.RFC3339))
+		} else {
+			assert.Fail(t, "timestamp is out of reasonable range",
+				"timestamp %d is not within reasonable Unix seconds range (year 2000-2100)", timestampInt)
+		}
+	}
+}
+
 // MessageConsumer is the interface that providers must implement
 type MessageConsumer interface {
 	// Consume returns a channel that receives messages