add correlation

CHANGELOG.md

@@ -3,6 +3,27 @@
 Notable changes to this library, newest first, per release. Written for whoever hosts the library or
 receives its deliveries.
 
+## v0.2.0
+
+### Added — the causing act's correlation id travels with the delivery
+
+`Event.CorrelationID` is the correlation id of the request that caused the event, when the host knew
+one. It is stored with the event (`MemoryStore` keeps it; a relational `Store` gets the column from
+`sql/V2__webhook_event_correlation_id.sql`) and the worker sends it as the platform's `X-Correlation-ID`
+header on **every attempt** of every delivery of the event — a retry continues the same thread. An event
+with no correlation id (background work) is sent without the header, never with an empty one. The header
+name comes from `go-platform-kit`'s `propagation` package, which is now a dependency: the library carries
+the platform kit like the platform's other libraries rather than re-declaring a concern the kit owns.
+
+For receivers: a delivery may now carry `X-Correlation-ID`; quote it when asking the host about a
+delivery. Nothing else on the wire changes.
+
+### Fixed — the delivery header's documentation
+
+`Headers.Delivery` was documented as "unique per attempt". The worker has always sent the delivery's own
+id — one per event per endpoint, the same value on every attempt — and the README said so; the doc
+comment now says the same.
+
 ## v0.1.0
 
 Initial code.

README.md

@@ -9,8 +9,11 @@ and a bounded give-up so a dead endpoint never keeps a queue alive forever.
 go get github.com/gmb-lib/go-webhook
 ```
 
-Standard library only. The library fixes everything a **receiver** can observe and leaves to the
-**host** what only the host knows: the event body, where things are stored, when the worker runs.
+The library fixes everything a **receiver** can observe and leaves to the **host** what only the host
+knows: the event body, where things are stored, when the worker runs. It builds on
+[go-platform-kit](https://github.com/gmb-lib/go-platform-kit) for the one cross-cutting concern it
+shares with every other service and library of the platform — the correlation id and its header name —
+and on the standard library for everything else.
 Because the receiver-facing contract lives here and nowhere else, a host can later move delivery to
 another process — same package, another host — and no receiver sees a different call.
 
@@ -23,11 +26,13 @@ Every delivery is one HTTP `POST` to the registered endpoint URL:
 | `Content-Type` | `application/json` |
 | `Webhook-Signature` | `t=<unix seconds>,v1=<hex HMAC-SHA256>` — two `v1` values while a secret rotation is in progress |
 | `Webhook-Event` | the event type, so you can route before parsing the body |
-| `Webhook-Delivery` | the delivery id — the same for every attempt of one delivery |
+| `Webhook-Delivery` | the delivery id — one per event per endpoint, the same for every attempt of one delivery |
+| `X-Correlation-ID` | the correlation id of the act that caused the event (a request a person or another system made), when the host knew one — the same on every attempt; quote it when you ask the host about a delivery. Absent for an event no request caused |
 | `User-Agent` | the host's name, or `go-webhook/<version>` |
 
 The body is the host's event, byte for byte; the event id inside it is stable across attempts, so
 **de-duplicate on it**. A host may rename the three `Webhook-*` headers; its documentation then says so.
+`X-Correlation-ID` is the platform's header and is never renamed.
 
 **The signature** is the lowercase hex HMAC-SHA256, keyed with a secret the host gave you when you
 registered, over the bytes `<t> "." <raw body>`. `t` is bound into the signed bytes so a replayed
@@ -90,8 +95,10 @@ _ = disp.Subscribe(ctx, webhook.Subscription{
     EventTypes: []string{"order.completed"},     // empty = every type
 })
 
-// publishing — never waits on a receiver; it only writes deliveries
-_, _ = disp.Publish(ctx, webhook.Event{ID: eventID, ClientID: "acme", Type: "order.completed", Payload: body})
+// publishing — never waits on a receiver; it only writes deliveries. CorrelationID is the
+// causing request's (propagation.CorrelationID(ctx) in a handler); leave it empty for
+// background work and no header is sent.
+_, _ = disp.Publish(ctx, webhook.Event{ID: eventID, ClientID: "acme", Type: "order.completed", Payload: body, CorrelationID: correlationID})
 
 // delivering — one goroutine per process, or several processes against a claiming Store
 go worker.Run(ctx, 5*time.Second, 100, errs)
@@ -109,8 +116,9 @@ that a subscription disabled after an event was queued is not sent to.
 ### Storing it
 
 `Store` is five kinds of read and write over subscriptions, events and deliveries; `MemoryStore` is the
-complete reference implementation. For a database, [`sql/V1__webhook_tables.sql`](sql/V1__webhook_tables.sql)
-is the table shape — copy it into your migrations and map it, column for column, onto the Go types.
+complete reference implementation. For a database, [`sql/`](sql/) is the table shape — copy its
+migrations into yours and map them, column for column, onto the Go types (`V1` the three tables, `V2` the
+event's `correlation_id`).
 Secrets are stored as **references** into your secret store, never as values; your `Store` resolves
 them when the worker asks. With several workers, claim rows as you read them
 (`FOR UPDATE SKIP LOCKED`) so no delivery is sent twice.

dispatcher.go

@@ -8,6 +8,8 @@ import (
 	"io"
 	"net/http"
 	"time"
+
+	"github.com/gmb-lib/go-platform-kit/propagation"
 )
 
 // Dispatcher is the seam a host publishes through. Publish records an event and
@@ -254,6 +256,10 @@ func (w *Worker) send(ctx context.Context, sub Subscription, ev Event, d Deliver
 	body := []byte(ev.Payload)
 	ctx, cancel := context.WithTimeout(ctx, w.timeout())
 	defer cancel()
+	// The causing act's correlation id rides on the context as well as on the header, so
+	// a host-supplied Client whose transport reads the platform's context sees the same
+	// thread (an empty id leaves the context as it is).
+	ctx = propagation.WithCorrelationID(ctx, ev.CorrelationID)
 	req, err := http.NewRequestWithContext(ctx, http.MethodPost, sub.EndpointURL, bytes.NewReader(body))
 	if err != nil {
 		return 0, "build request: " + err.Error()
@@ -264,6 +270,11 @@ func (w *Worker) send(ctx context.Context, sub Subscription, ev Event, d Deliver
 	req.Header.Set(h.Signature, SignatureHeader(now, body, secrets...))
 	req.Header.Set(h.Event, ev.Type)
 	req.Header.Set(h.Delivery, d.ID)
+	// Every attempt of a delivery carries the same correlation id: a retry continues the
+	// thread the causing act started. An event that no request caused carries none.
+	if ev.CorrelationID != "" {
+		req.Header.Set(propagation.HeaderCorrelationID, ev.CorrelationID)
+	}
 
 	resp, err := w.client().Do(req)
 	if err != nil {

dispatcher_test.go

@@ -9,6 +9,8 @@ import (
 	"sync/atomic"
 	"testing"
 	"time"
+
+	"github.com/gmb-lib/go-platform-kit/propagation"
 )
 
 // fakeClock is a settable time source shared by the dispatcher and the worker.
@@ -118,7 +120,8 @@ func TestWorkerRetriesThenDeliversWithVerifiableSignature(t *testing.T) {
 		t.Fatal(err)
 	}
 	payload := []byte(`{"eventId":"e1","type":"signing-request.completed","sequence":3}`)
-	ds, err := disp.Publish(ctx, Event{ID: "e1", ClientID: "acme", Type: "signing-request.completed", Payload: payload})
+	const correlation = "01K5VB0E-the-causing-request"
+	ds, err := disp.Publish(ctx, Event{ID: "e1", ClientID: "acme", Type: "signing-request.completed", Payload: payload, CorrelationID: correlation})
 	if err != nil || len(ds) != 1 {
 		t.Fatalf("publish: %v %+v", err, ds)
 	}
@@ -193,6 +196,42 @@ func TestWorkerRetriesThenDeliversWithVerifiableSignature(t *testing.T) {
 	if err := Verify(sig, last.body, [][]byte{testSecret("stranger")}, sentAt, 5*time.Minute); err == nil {
 		t.Fatal("a stranger's secret must not verify")
 	}
+	// The correlation id of the act that caused the event travels on EVERY attempt, the
+	// same value each time — a retry is the same thread, not a new one.
+	for i, call := range rcv.calls {
+		if got := call.headers.Get(propagation.HeaderCorrelationID); got != correlation {
+			t.Fatalf("attempt %d: %s = %q, want %q", i+1, propagation.HeaderCorrelationID, got, correlation)
+		}
+	}
+}
+
+// An event that no request caused — background work — has no correlation id, and the
+// delivery then carries no header at all rather than an empty one.
+func TestWorkerSendsNoCorrelationHeaderWhenTheEventHasNone(t *testing.T) {
+	ctx := context.Background()
+	rcv := &receiver{}
+	srv := httptest.NewServer(rcv.handler(t))
+	defer srv.Close()
+	store := NewMemoryStore()
+	clock := &fakeClock{t: time.Unix(1757170123, 0)}
+	disp := &InProcess{Store: store, Clock: clock.Now}
+	_ = disp.Subscribe(ctx, Subscription{ID: "s", ClientID: "c", EndpointURL: srv.URL, Enabled: true, Secrets: []Secret{{Value: testSecret("k")}}})
+	ds, _ := disp.Publish(ctx, Event{ID: "e", ClientID: "c", Type: "t", Payload: []byte(`{}`)})
+	w := &Worker{Store: store, Clock: clock.Now, Jitter: NoJitter}
+	if n, err := w.RunOnce(ctx, 0); err != nil || n != 1 {
+		t.Fatalf("run: n=%d err=%v", n, err)
+	}
+	if d := mustDelivery(t, store, ds[0].ID); d.Status != StatusDelivered {
+		t.Fatalf("delivered expected: %+v", d)
+	}
+	rcv.mu.Lock()
+	defer rcv.mu.Unlock()
+	if len(rcv.calls) != 1 {
+		t.Fatalf("receiver saw %d calls", len(rcv.calls))
+	}
+	if vals := rcv.calls[0].headers.Values(propagation.HeaderCorrelationID); len(vals) != 0 {
+		t.Fatalf("no correlation id on the event, yet the header was sent: %q", vals)
+	}
 }
 
 func TestWorkerDropsOnClientErrorAndDeadLettersWhenExhausted(t *testing.T) {

go.mod

@@ -1,3 +1,5 @@
 module github.com/gmb-lib/go-webhook
 
 go 1.26.6
+
+require github.com/gmb-lib/go-platform-kit v1.11.1

go.sum

@@ -0,0 +1,12 @@
+github.com/gmb-lib/go-platform-kit v1.11.1 h1:UG2VTTLByAyH7X1q6cP/RECurFS6wGs9azPxIiUzZtY=
+github.com/gmb-lib/go-platform-kit v1.11.1/go.mod h1:/ZCrZUjDBF+CniHa5xah3IfG0P7o6TqMYXK5ojvNkwU=
+github.com/go-quicktest/qt v1.102.0 h1:HSQxCeh5YZH3EL3W39ixjtyaEhcWSXQHtHnMBzSs474=
+github.com/go-quicktest/qt v1.102.0/go.mod h1:p4lGIVX+8Wa6ZPNDvqcxq36XpUDLh42FLetFU7odllI=
+github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
+github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
+github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
+github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
+github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/rogpeppe/go-internal v1.16.0 h1:O9DK+vNMDVGLr2BeZqmpLeMjiMNkuXfcqntWbZV6S5g=
+github.com/rogpeppe/go-internal v1.16.0/go.mod h1:DrUVZyrJU+txYW5/1kwtXQSMFio52ZOxX7yM1VHvnxs=

sql/README.md

@@ -1,9 +1,12 @@
 # Table shape for a relational Store
 
 `V1__webhook_tables.sql` is the reference DDL behind the library's `Store` interface: three tables
-(`subscription`, `event`, `delivery`) whose columns match the Go types one for one. It is not applied
-by the library — copy it into your own migration set, put it in your own schema, and implement `Store`
-over it in whatever way your service talks to its database.
+(`subscription`, `event`, `delivery`) whose columns match the Go types one for one.
+`V2__webhook_event_correlation_id.sql` adds the event's `correlation_id` (the causing request's id,
+sent as `X-Correlation-ID` on every attempt; nullable). Neither is applied by the library — copy them
+into your own migration set in order, put them in your own schema, and implement `Store` over them in
+whatever way your service talks to its database. A host that copied `V1` before `V2` existed adds `V2`
+as its own next migration.
 
 Two things the shape decides on purpose:
 

sql/V2__webhook_event_correlation_id.sql

@@ -0,0 +1,11 @@
+-- V2: the event remembers the correlation id of the act that caused it.
+--
+-- A delivery carries the X-Correlation-ID header of the request that caused the event
+-- (a person's or another system's act), the same value on every attempt, so a receiver
+-- can quote one id and the host can find the whole thread across its services. For that
+-- the id has to survive with the event, not with the process: a worker that restarts
+-- between two attempts reads it back from here. NULL when no request caused the event
+-- (background work); the header is then not sent.
+--
+-- Nullable, no default — a metadata-only add, no table rewrite.
+ALTER TABLE webhook.event ADD COLUMN IF NOT EXISTS correlation_id text;

store_test.go

@@ -75,11 +75,16 @@ func TestMemoryStoreSubscriptionsEventsAndNotFound(t *testing.T) {
 		t.Fatalf("acme's subscriptions, sorted: %+v", subs)
 	}
 
-	_ = m.SaveEvent(ctx, Event{ID: "e1", ClientID: "acme", Type: "x", Payload: []byte(`{"a":1}`)})
+	_ = m.SaveEvent(ctx, Event{ID: "e1", ClientID: "acme", Type: "x", Payload: []byte(`{"a":1}`), CorrelationID: "corr-1"})
 	ev, err := m.Event(ctx, "e1")
 	if err != nil || string(ev.Payload) != `{"a":1}` {
 		t.Fatalf("event round-trip: %v %s", err, ev.Payload)
 	}
+	// The correlation id is part of the stored event: a store that dropped it would send
+	// the first attempt with the header and a retry without.
+	if ev.CorrelationID != "corr-1" {
+		t.Fatalf("correlation id not stored with the event: %+v", ev)
+	}
 
 	_ = m.Enqueue(ctx, []Delivery{{ID: "d2", EventID: "e1"}, {ID: "d1", EventID: "e1"}, {ID: "d9", EventID: "e9"}})
 	byEvent, _ := m.ByEvent(ctx, "e1")

webhook.go

@@ -30,7 +30,7 @@ import (
 )
 
 // Version is the library version, sent as the User-Agent when a host does not set one.
-const Version = "0.1.0"
+const Version = "0.2.0"
 
 // Event is one thing that happened. ID must be stable across delivery attempts: a
 // receiver de-duplicates on it. Payload is the exact body sent — this package neither
@@ -44,6 +44,14 @@ type Event struct {
 	// another's events.
 	ClientID string
 	Payload  json.RawMessage
+	// CorrelationID is the correlation id of the act that caused the event — the request
+	// a person or another system made — when that act had one. It is stored with the
+	// event and sent as the X-Correlation-ID header (the platform kit's
+	// propagation.HeaderCorrelationID — the one home of that header's name) on every
+	// attempt of every delivery of the event, so a receiver can quote it and
+	// the host can find the whole thread across its services. Empty for an event that
+	// no request caused (background work): the header is then not sent.
+	CorrelationID string
 }
 
 // Subscription is one registered endpoint. EventTypes empty means every type.
@@ -106,8 +114,9 @@ type Headers struct {
 	Signature string
 	// Event carries the event type, so a receiver can route before parsing the body.
 	Event string
-	// Delivery carries the delivery attempt id — unique per attempt, while the event id
-	// inside the body is the same across attempts.
+	// Delivery carries the delivery id — one per event per endpoint, the same value on
+	// every attempt of that delivery. The event id inside the body is the same across
+	// endpoints too; a receiver may de-duplicate on either.
 	Delivery string
 }