Carry the event's receipt time into every delivery (closes #257)

internal/delivery/engine.go

@@ -440,7 +440,7 @@ func (e *Engine) processNewTask(
 
 	event := buildEventFromTask(task)
 
-	event, err = e.resolveEventBody(
+	event, err = e.hydrateEvent(
 		webhookDB, event, task,
 	)
 	if err != nil {
@@ -512,7 +512,7 @@ func (e *Engine) processRetryTask(
 
 	event := buildEventFromTask(task)
 
-	event, err = e.resolveEventBody(
+	event, err = e.hydrateEvent(
 		webhookDB, event, task,
 	)
 	if err != nil {
@@ -1547,6 +1547,11 @@ func truncate(s string, maxLen int) string {
 
 // --- Helper functions ---
 
+// buildEventFromTask reconstructs the event a Task describes, as far
+// as the Task itself goes. The fields it cannot fill — the body when
+// it was too large to inline, and the receipt time, which no Task
+// carries — come from the stored row in hydrateEvent, which every
+// caller of this function runs next.
 func buildEventFromTask(task *Task) database.Event {
 	event := database.Event{
 		EntrypointID: task.EntrypointID,
@@ -1574,29 +1579,67 @@ func buildTargetFromTask(task *Task) database.Target {
 	return target
 }
 
-func (e *Engine) resolveEventBody(
+// hydrateEvent fills in the event fields a Task does not carry, by
+// reading the stored event row.
+//
+// CreatedAt is the event's receipt time and lives only in that row.
+// The Slack target renders it into every message it sends, so an
+// unhydrated event puts the zero time in front of a human on every
+// notification the product delivers. See
+// https://git.eeqj.de/sneak/webhooker/issues/257.
+//
+// The body comes from the same row when the Task did not inline it,
+// which is the case for a body at or above MaxInlineBodySize.
+//
+// A read failure is fatal to the delivery only when the body depended
+// on it. When the Task inlined the body, the delivery has everything
+// it needs to be sent and goes ahead with the timestamp unset: the row
+// can be gone under a retention reap while a queued delivery still
+// holds its body, and dropping a deliverable event to protect one
+// metadata field would be a worse failure than the one it prevents.
+func (e *Engine) hydrateEvent(
 	webhookDB *gorm.DB,
 	event database.Event,
 	task *Task,
 ) (database.Event, error) {
-	if task.Body != nil {
+	columns := []string{"created_at"}
+
+	if task.Body == nil {
+		columns = append(columns, "body")
+	}
+
+	var dbEvent database.Event
+
+	err := webhookDB.Select(columns).
+		First(&dbEvent, "id = ?", task.EventID).Error
+	if err != nil {
+		if task.Body == nil {
+			return event, fmt.Errorf(
+				"fetching event body: %w", err,
+			)
+		}
+
+		e.log.Warn(
+			"could not read the stored event; delivering "+
+				"the inlined body without its receipt time",
+			"event_id", task.EventID,
+			"delivery_id", task.DeliveryID,
+			"error", err,
+		)
+
 		event.Body = *task.Body
 
 		return event, nil
 	}
 
-	var dbEvent database.Event
+	event.CreatedAt = dbEvent.CreatedAt
 
-	err := webhookDB.Select("body").
-		First(&dbEvent, "id = ?", task.EventID).Error
-	if err != nil {
-		return event, fmt.Errorf(
-			"fetching event body: %w", err,
-		)
+	if task.Body != nil {
+		event.Body = *task.Body
+	} else {
+		event.Body = dbEvent.Body
 	}
 
-	event.Body = dbEvent.Body
-
 	return event, nil
 }
 

internal/delivery/event_timestamp_test.go

@@ -0,0 +1,407 @@
+package delivery_test
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gorm.io/gorm"
+	"sneak.berlin/go/webhooker/internal/database"
+	"sneak.berlin/go/webhooker/internal/delivery"
+)
+
+// tsEventCreatedAt is the receipt time seeded on the events these
+// tests deliver. It is far enough from both the zero time and from
+// now that neither can be mistaken for it.
+func tsEventCreatedAt() time.Time {
+	return time.Date(
+		2026, time.March, 4, 5, 6, 7, 0, time.UTC,
+	)
+}
+
+// tsZeroStamp is what a Slack message renders when the event handed
+// to FormatSlackMessage carries no CreatedAt.
+const tsZeroStamp = "*Timestamp:* `0001-01-01T00:00:00Z`"
+
+// tsEventBody is the body seeded on every event in this file. It is
+// small enough that a Task can inline it.
+const tsEventBody = `{"hello":"world"}`
+
+// tsUndeliverableHook stands in for a Slack incoming webhook on the
+// tests that never send: the config parser requires a URL, but no
+// request is made.
+const tsUndeliverableHook = "https://hooks.slack.com/services/T/B/x"
+
+// tsSink is a stand-in Slack incoming webhook that records the raw
+// body posted to it.
+type tsSink struct {
+	*httptest.Server
+
+	bodies chan []byte
+}
+
+func newTSSink(t *testing.T) *tsSink {
+	t.Helper()
+
+	s := &tsSink{bodies: make(chan []byte, 8)}
+
+	s.Server = httptest.NewServer(http.HandlerFunc(
+		func(w http.ResponseWriter, r *http.Request) {
+			body, _ := io.ReadAll(r.Body)
+
+			select {
+			case s.bodies <- body:
+			default:
+			}
+
+			w.WriteHeader(http.StatusOK)
+		},
+	))
+
+	t.Cleanup(s.Close)
+
+	return s
+}
+
+// text returns the Slack message text from the single payload the
+// sink received.
+func (s *tsSink) text(t *testing.T) string {
+	t.Helper()
+
+	select {
+	case raw := <-s.bodies:
+		t.Logf("raw slack payload: %s", raw)
+
+		var payload struct {
+			Text string `json:"text"`
+		}
+
+		require.NoError(t, json.Unmarshal(raw, &payload))
+
+		return payload.Text
+	case <-time.After(5 * time.Second):
+		t.Fatal("slack sink received no payload")
+
+		return ""
+	}
+}
+
+func tsSlackConfig(t *testing.T, url string) string {
+	t.Helper()
+
+	data, err := json.Marshal(
+		delivery.SlackTargetConfig{WebhookURL: url},
+	)
+	require.NoError(t, err)
+
+	return string(data)
+}
+
+// tsSeedEvent writes an event whose CreatedAt is tsEventCreatedAt
+// rather than the write time, so an assertion on the rendered
+// timestamp cannot pass by accident against "roughly now".
+func tsSeedEvent(
+	t *testing.T, db *gorm.DB, webhookID string,
+) database.Event {
+	t.Helper()
+
+	event := database.Event{
+		WebhookID:    webhookID,
+		EntrypointID: uuid.New().String(),
+		Method:       http.MethodPost,
+		Headers:      `{}`,
+		Body:         tsEventBody,
+		ContentType:  "application/json",
+	}
+	event.ID = uuid.New().String()
+	event.CreatedAt = tsEventCreatedAt()
+	event.UpdatedAt = tsEventCreatedAt()
+
+	require.NoError(t, db.Create(&event).Error)
+
+	var stored database.Event
+
+	require.NoError(t,
+		db.First(&stored, "id = ?", event.ID).Error,
+	)
+	require.Equal(t,
+		tsEventCreatedAt().UTC(), stored.CreatedAt.UTC(),
+		"seeded created_at did not round-trip",
+	)
+
+	return event
+}
+
+// tsSeedTarget writes the slack target row into the main database.
+// The retry path confirms the target still exists before sending.
+func tsSeedTarget(
+	t *testing.T, mainDB *gorm.DB, webhookID, config string,
+) database.Target {
+	t.Helper()
+
+	target := database.Target{
+		WebhookID: webhookID,
+		Name:      "slack-sink",
+		Type:      database.TargetTypeSlack,
+		Config:    config,
+		Active:    true,
+	}
+
+	require.NoError(t, mainDB.Create(&target).Error)
+
+	return target
+}
+
+func tsTask(
+	d database.Delivery,
+	event database.Event,
+	webhookID string,
+	target database.Target,
+	attemptNum int,
+	body *string,
+) delivery.Task {
+	return delivery.Task{
+		DeliveryID:   d.ID,
+		EventID:      event.ID,
+		WebhookID:    webhookID,
+		EntrypointID: event.EntrypointID,
+		TargetID:     target.ID,
+		TargetName:   target.Name,
+		TargetType:   database.TargetTypeSlack,
+		TargetConfig: target.Config,
+		MaxRetries:   0,
+		Method:       event.Method,
+		Headers:      event.Headers,
+		ContentType:  event.ContentType,
+		Body:         body,
+		AttemptNum:   attemptNum,
+	}
+}
+
+func tsAssertRealTimestamp(t *testing.T, text string) {
+	t.Helper()
+
+	assert.NotContains(t, text, tsZeroStamp,
+		"slack message carries the zero timestamp",
+	)
+	assert.Contains(t, text,
+		"*Timestamp:* `"+
+			tsEventCreatedAt().UTC().Format(time.RFC3339)+"`",
+		"slack message does not carry the event's receipt time",
+	)
+}
+
+// tsCase is one end-to-end delivery of a seeded event to a slack
+// sink, over whichever engine path `process` names.
+type tsCase struct {
+	// status is the delivery row's status before the engine runs.
+	// The retry path refuses a delivery that is not retrying.
+	status database.DeliveryStatus
+
+	// inlineBody mirrors a Task built for a body under
+	// MaxInlineBodySize. When false the engine reads the body back
+	// from the stored row.
+	inlineBody bool
+
+	attemptNum int
+
+	process func(
+		ctx context.Context, e *delivery.Engine, task *delivery.Task,
+	)
+}
+
+// run delivers one event through the named path and returns the
+// Slack message text the sink received.
+func (c tsCase) run(t *testing.T) (iSetup, database.Delivery, string) {
+	t.Helper()
+
+	s := newISetup(t)
+	sink := newTSSink(t)
+
+	cfg := tsSlackConfig(t, sink.URL)
+	target := tsSeedTarget(t, s.MainDB, s.WebhookID, cfg)
+	event := tsSeedEvent(t, s.WebhookDB, s.WebhookID)
+
+	d := iSeedDelivery(
+		t, s.WebhookDB, event.ID, target.ID, c.status,
+	)
+
+	var body *string
+
+	if c.inlineBody {
+		bodyStr := event.Body
+		body = &bodyStr
+	}
+
+	task := tsTask(
+		d, event, s.WebhookID, target, c.attemptNum, body,
+	)
+
+	c.process(context.TODO(), s.Engine, &task)
+
+	return s, d, sink.text(t)
+}
+
+// TestSlackFirstAttemptCarriesEventTimestamp covers the path an
+// event takes on its first delivery: the task comes from the
+// receiver and the engine reconstructs the event from it.
+func TestSlackFirstAttemptCarriesEventTimestamp(t *testing.T) {
+	t.Parallel()
+
+	s, d, text := tsCase{
+		status:     database.DeliveryStatusPending,
+		inlineBody: true,
+		attemptNum: 1,
+		process: func(
+			ctx context.Context,
+			e *delivery.Engine,
+			task *delivery.Task,
+		) {
+			e.ExportProcessNewTask(ctx, task)
+		},
+	}.run(t)
+
+	tsAssertRealTimestamp(t, text)
+
+	iAssertStatus(t, s.WebhookDB, d.ID,
+		database.DeliveryStatusDelivered,
+	)
+}
+
+// TestSlackFirstAttemptLargeBodyCarriesEventTimestamp covers the
+// first-attempt path for an event whose body exceeded
+// MaxInlineBodySize, so the task carries no body and the engine
+// reads it back from the stored row.
+func TestSlackFirstAttemptLargeBodyCarriesEventTimestamp(
+	t *testing.T,
+) {
+	t.Parallel()
+
+	_, _, text := tsCase{
+		status:     database.DeliveryStatusPending,
+		inlineBody: false,
+		attemptNum: 1,
+		process: func(
+			ctx context.Context,
+			e *delivery.Engine,
+			task *delivery.Task,
+		) {
+			e.ExportProcessNewTask(ctx, task)
+		},
+	}.run(t)
+
+	tsAssertRealTimestamp(t, text)
+}
+
+// TestSlackRetryCarriesEventTimestamp covers the retry path, which
+// reconstructs the event from the same task the first attempt used.
+func TestSlackRetryCarriesEventTimestamp(t *testing.T) {
+	t.Parallel()
+
+	s, d, text := tsCase{
+		status:     database.DeliveryStatusRetrying,
+		inlineBody: true,
+		attemptNum: 2,
+		process: func(
+			ctx context.Context,
+			e *delivery.Engine,
+			task *delivery.Task,
+		) {
+			e.ExportProcessRetryTask(ctx, task)
+		},
+	}.run(t)
+
+	tsAssertRealTimestamp(t, text)
+
+	iAssertStatus(t, s.WebhookDB, d.ID,
+		database.DeliveryStatusDelivered,
+	)
+}
+
+// TestFormatSlackMessageOverTaskReconstructedEvent asserts on the
+// formatted message directly, over the event the delivery paths
+// reconstruct from a Task. It is the unit-level guard under the
+// end-to-end tests: revert the CreatedAt population in hydrateEvent
+// and this fails on the zero timestamp.
+func TestFormatSlackMessageOverTaskReconstructedEvent(
+	t *testing.T,
+) {
+	t.Parallel()
+
+	s := newISetup(t)
+
+	cfg := tsSlackConfig(t, tsUndeliverableHook)
+	target := tsSeedTarget(t, s.MainDB, s.WebhookID, cfg)
+	event := tsSeedEvent(t, s.WebhookDB, s.WebhookID)
+
+	d := iSeedDelivery(
+		t, s.WebhookDB, event.ID, target.ID,
+		database.DeliveryStatusPending,
+	)
+
+	bodyStr := event.Body
+	task := tsTask(d, event, s.WebhookID, target, 1, &bodyStr)
+
+	rebuilt, err := s.Engine.ExportEventForTask(
+		s.WebhookDB, &task,
+	)
+	require.NoError(t, err)
+	assert.False(t, rebuilt.CreatedAt.IsZero(),
+		"reconstructed event carries the zero time",
+	)
+	assert.Equal(t,
+		tsEventCreatedAt().UTC(), rebuilt.CreatedAt.UTC(),
+	)
+
+	tsAssertRealTimestamp(
+		t, delivery.FormatSlackMessage(&rebuilt),
+	)
+}
+
+// TestEventReconstructionSurvivesAReapedRow pins the fallback: an
+// event row reaped by retention while its delivery still holds the
+// body inline is still delivered, with the receipt time unset,
+// rather than dropped.
+func TestEventReconstructionSurvivesAReapedRow(t *testing.T) {
+	t.Parallel()
+
+	s := newISetup(t)
+
+	cfg := tsSlackConfig(t, tsUndeliverableHook)
+	target := tsSeedTarget(t, s.MainDB, s.WebhookID, cfg)
+	event := tsSeedEvent(t, s.WebhookDB, s.WebhookID)
+
+	d := iSeedDelivery(
+		t, s.WebhookDB, event.ID, target.ID,
+		database.DeliveryStatusPending,
+	)
+
+	bodyStr := event.Body
+	task := tsTask(d, event, s.WebhookID, target, 1, &bodyStr)
+
+	require.NoError(t, s.WebhookDB.Unscoped().Delete(
+		&database.Event{}, "id = ?", event.ID,
+	).Error)
+
+	rebuilt, err := s.Engine.ExportEventForTask(
+		s.WebhookDB, &task,
+	)
+	require.NoError(t, err)
+	assert.Equal(t, bodyStr, rebuilt.Body)
+	assert.True(t, rebuilt.CreatedAt.IsZero())
+
+	// A task with no inlined body has nothing left to deliver, so
+	// the same reaped row is an error there.
+	noBody := task
+	noBody.Body = nil
+
+	_, err = s.Engine.ExportEventForTask(s.WebhookDB, &noBody)
+	require.Error(t, err)
+}

internal/delivery/export_test.go

@@ -151,6 +151,16 @@ func (e *Engine) ExportProcessRetryTask(
 	e.processRetryTask(ctx, task)
 }
 
+// ExportEventForTask exposes the event reconstruction the delivery
+// paths run: buildEventFromTask followed by hydrateEvent.
+func (e *Engine) ExportEventForTask(
+	webhookDB *gorm.DB, task *Task,
+) (database.Event, error) {
+	return e.hydrateEvent(
+		webhookDB, buildEventFromTask(task), task,
+	)
+}
+
 // ExportProcessDelivery exposes processDelivery.
 func (e *Engine) ExportProcessDelivery(
 	ctx context.Context,