feat(http): add IdempotencyKey in header if present (#70)

Author: paul-nicolas

pkg/attempt.go

@@ -35,7 +35,7 @@ type Attempt struct {
 	NextRetryAfter time.Time `json:"nextRetryAfter,omitempty" bun:"next_retry_after,nullzero"`
 }
 
-func MakeAttempt(ctx context.Context, httpClient *http.Client, retryPolicy BackoffPolicy, id, webhookID string, attemptNb int, cfg Config, payload []byte, isTest bool) (Attempt, error) {
+func MakeAttempt(ctx context.Context, httpClient *http.Client, retryPolicy BackoffPolicy, id, webhookID string, attemptNb int, cfg Config, idempotencyKey string, payload []byte, isTest bool) (Attempt, error) {
 	req, err := http.NewRequestWithContext(ctx, http.MethodPost, cfg.Endpoint, bytes.NewBuffer(payload))
 	if err != nil {
 		return Attempt{}, errors.Wrap(err, "http.NewRequestWithContext")
@@ -54,6 +54,9 @@ func MakeAttempt(ctx context.Context, httpClient *http.Client, retryPolicy Backo
 	req.Header.Set("formance-webhook-timestamp", fmt.Sprintf("%d", timestamp))
 	req.Header.Set("formance-webhook-signature", signature)
 	req.Header.Set("formance-webhook-test", fmt.Sprintf("%v", isTest))
+	if idempotencyKey != "" {
+		req.Header.Set("formance-webhook-idempotency-key", idempotencyKey)
+	}
 
 	resp, err := httpClient.Do(req)
 	if err != nil {

pkg/server/test.go

@@ -27,7 +27,7 @@ func (h *serverHandler) testOneConfigHandle(w http.ResponseWriter, r *http.Reque
 		logging.FromContext(r.Context()).Debugf("GET %s/%s%s", PathConfigs, id, PathTest)
 		retryPolicy := backoff.NewNoRetry()
 		attempt, err := webhooks.MakeAttempt(r.Context(), h.httpClient, retryPolicy, uuid.NewString(),
-			uuid.NewString(), 0, cfgs[0], []byte(`{"data":"test"}`), true)
+			uuid.NewString(), 0, cfgs[0], "ik", []byte(`{"data":"test"}`), true)
 		if err != nil {
 			logging.FromContext(r.Context()).Errorf("GET %s/%s%s: %s", PathConfigs, id, PathTest, err)
 			apierrors.ResponseError(w, r, err)

pkg/worker/module.go

@@ -84,7 +84,6 @@ func configureMessageRouter(r *message.Router, subscriber message.Subscriber, to
 func processMessages(store storage.Store, httpClient *http.Client, retryPolicy webhooks.BackoffPolicy, pool *pond.WorkerPool) func(msg *message.Message) error {
 	return func(msg *message.Message) error {
 		pool.Submit(func() {
-
 			var ev *publish.EventMessage
 			span, ev, err := publish.UnmarshalMessage(msg)
 			if err != nil {
@@ -131,16 +130,16 @@ func processMessages(store storage.Store, httpClient *http.Client, retryPolicy w
 				return
 			}
 
+			data, err := json.Marshal(ev)
+			if err != nil {
+				logging.FromContext(ctx).Error(err)
+				return
+			}
 			for _, cfg := range cfgs {
 				logging.FromContext(ctx).Debugf("found one config: %+v", cfg)
-				data, err := json.Marshal(ev)
-				if err != nil {
-					logging.FromContext(ctx).Error(err)
-					return
-				}
 
 				attempt, err := webhooks.MakeAttempt(ctx, httpClient, retryPolicy, uuid.NewString(),
-					uuid.NewString(), 0, cfg, data, false)
+					uuid.NewString(), 0, cfg, ev.IdempotencyKey, data, false)
 				if err != nil {
 					logging.FromContext(ctx).Error(err)
 					return

pkg/worker/worker.go

@@ -2,11 +2,13 @@ package worker
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"net/http"
 	"time"
 
 	"github.com/formancehq/go-libs/v2/logging"
+	"github.com/formancehq/go-libs/v2/publish"
 	webhooks "github.com/formancehq/webhooks/pkg"
 	"github.com/formancehq/webhooks/pkg/storage"
 	"github.com/google/uuid"
@@ -103,9 +105,16 @@ func (w *Retrier) attemptRetries(ctx context.Context, errChan chan error) {
 					continue
 				}
 
+				var ev publish.EventMessage
+				err = json.Unmarshal([]byte(atts[0].Payload), &ev)
+				if err != nil {
+					errChan <- errors.Wrap(err, "json.Unmarshal")
+					continue
+				}
+
 				newAttemptNb := atts[0].RetryAttempt + 1
 				attempt, err := webhooks.MakeAttempt(ctx, w.httpClient, w.retryPolicy, uuid.NewString(),
-					webhookID, newAttemptNb, atts[0].Config, []byte(atts[0].Payload), false)
+					webhookID, newAttemptNb, atts[0].Config, ev.IdempotencyKey, []byte(atts[0].Payload), false)
 				if err != nil {
 					errChan <- errors.Wrap(err, "webhooks.MakeAttempt")
 					continue