diff --git a/internal/delivery/engine.go b/internal/delivery/engine.go index e3415c3..e023918 100644 --- a/internal/delivery/engine.go +++ b/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 } diff --git a/internal/delivery/event_timestamp_test.go b/internal/delivery/event_timestamp_test.go new file mode 100644 index 0000000..87bd07f --- /dev/null +++ b/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) +} diff --git a/internal/delivery/export_test.go b/internal/delivery/export_test.go index e3b20c2..b71de96 100644 --- a/internal/delivery/export_test.go +++ b/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,