From e55013d14f1b43e2e17f48a76303fe7224267bee Mon Sep 17 00:00:00 2001 From: rcourtman Date: Wed, 22 Apr 2026 13:37:05 +0100 Subject: [PATCH] Bound Stripe webhook dedupe retention --- internal/api/payments_webhook_handlers.go | 112 +++++++++++++++-- ...stripe_webhook_handlers_additional_test.go | 118 ++++++++++++++++++ 2 files changed, 222 insertions(+), 8 deletions(-) diff --git a/internal/api/payments_webhook_handlers.go b/internal/api/payments_webhook_handlers.go index c8cded2a7f..a5311a86c7 100644 --- a/internal/api/payments_webhook_handlers.go +++ b/internal/api/payments_webhook_handlers.go @@ -13,6 +13,7 @@ import ( "os" "path/filepath" "strings" + "sync" "time" "github.com/rcourtman/pulse-go-rewrite/internal/config" @@ -21,7 +22,13 @@ import ( "github.com/stripe/stripe-go/v82/webhook" ) -const stripeWebhookBodyLimit = 1024 * 1024 // 1MiB +const ( + stripeWebhookBodyLimit = 1024 * 1024 // 1MiB + // Keep handled event markers long enough to cover Stripe's automatic retry + // window and operator-driven resends, but not forever. + stripeWebhookDoneRetention = 30 * 24 * time.Hour + stripeWebhookPruneInterval = time.Hour +) var errStripeWebhookEventInFlight = errors.New("stripe webhook event is in-flight") @@ -574,17 +581,23 @@ func resolvePulseDataDir(dataPath string) string { // stripeWebhookDeduper provides durable idempotency for Stripe webhook event IDs. // Stripe retries webhooks; without a persistent dedupe store, retries can provision duplicate tenants. type stripeWebhookDeduper struct { - dir string - lockTTL time.Duration - now func() time.Time - hashSalt []byte + dir string + lockTTL time.Duration + doneTTL time.Duration + pruneTTL time.Duration + now func() time.Time + hashSalt []byte + pruneMu sync.Mutex + lastPrune time.Time } func newStripeWebhookDeduper(dir string) *stripeWebhookDeduper { return &stripeWebhookDeduper{ - dir: dir, - lockTTL: 10 * time.Minute, - now: time.Now, + dir: dir, + lockTTL: 10 * time.Minute, + doneTTL: stripeWebhookDoneRetention, + pruneTTL: stripeWebhookPruneInterval, + now: time.Now, // Salt prevents event IDs from being used directly as filenames if they contain odd characters. // (Event IDs are normally safe, but this keeps the filesystem contract tight.) hashSalt: []byte("pulse-stripe-webhook-v1"), @@ -656,6 +669,9 @@ func (d *stripeWebhookDeduper) Do(eventID string, fn func() error) (already bool } return false, fmt.Errorf("commit dedupe: %w", err) } + if pruneErr := d.maybePruneArtifacts(); pruneErr != nil { + log.Warn().Err(pruneErr).Str("dir", d.dir).Msg("Stripe dedupe: failed to prune old webhook markers") + } return false, nil } @@ -729,6 +745,86 @@ func (d *stripeWebhookDeduper) filenameForID(id string) string { return hex.EncodeToString(mac.Sum(nil)) } +func (d *stripeWebhookDeduper) maybePruneArtifacts() error { + if d == nil || strings.TrimSpace(d.dir) == "" { + return nil + } + + now := d.now() + + d.pruneMu.Lock() + defer d.pruneMu.Unlock() + + if d.pruneTTL > 0 && !d.lastPrune.IsZero() && now.Sub(d.lastPrune) < d.pruneTTL { + return nil + } + + if err := d.pruneArtifacts(now); err != nil { + return err + } + + d.lastPrune = now + return nil +} + +func (d *stripeWebhookDeduper) pruneArtifacts(now time.Time) error { + entries, err := os.ReadDir(d.dir) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil + } + return fmt.Errorf("read dedupe dir: %w", err) + } + + doneCutoff := now.Add(-d.doneTTL) + lockCutoff := now.Add(-d.lockTTL) + + for _, entry := range entries { + if entry.IsDir() { + continue + } + + name := entry.Name() + path := filepath.Join(d.dir, name) + + switch { + case strings.HasSuffix(name, ".done"): + if d.doneTTL <= 0 { + continue + } + info, infoErr := entry.Info() + if infoErr != nil { + if errors.Is(infoErr, os.ErrNotExist) { + continue + } + return fmt.Errorf("stat dedupe marker %s: %w", path, infoErr) + } + if info.ModTime().After(doneCutoff) || info.ModTime().Equal(doneCutoff) { + continue + } + case strings.HasSuffix(name, ".lock"), strings.HasSuffix(name, ".tmp"): + info, infoErr := entry.Info() + if infoErr != nil { + if errors.Is(infoErr, os.ErrNotExist) { + continue + } + return fmt.Errorf("stat dedupe artifact %s: %w", path, infoErr) + } + if info.ModTime().After(lockCutoff) || info.ModTime().Equal(lockCutoff) { + continue + } + default: + continue + } + + if rmErr := os.Remove(path); rmErr != nil && !errors.Is(rmErr, os.ErrNotExist) { + return fmt.Errorf("remove dedupe artifact %s: %w", path, rmErr) + } + } + + return nil +} + type stripeCustomerOrgIndex struct { dir string } diff --git a/internal/api/stripe_webhook_handlers_additional_test.go b/internal/api/stripe_webhook_handlers_additional_test.go index 947edee0cd..8cdf0784b5 100644 --- a/internal/api/stripe_webhook_handlers_additional_test.go +++ b/internal/api/stripe_webhook_handlers_additional_test.go @@ -4,8 +4,10 @@ import ( "bytes" "context" "encoding/json" + "errors" "net/http" "net/http/httptest" + "os" "testing" "time" @@ -297,3 +299,119 @@ func TestStripeWebhook_HandleEvent_ErrorsAndUnhandled(t *testing.T) { t.Fatalf("unexpected error for unhandled event: %v", err) } } + +func TestStripeWebhookDeduperDoReturnsDuplicateAfterSuccessfulHandle(t *testing.T) { + t.Parallel() + + d := newStripeWebhookDeduper(t.TempDir()) + d.pruneTTL = 0 + + calls := 0 + already, err := d.Do("evt_first", func() error { + calls++ + return nil + }) + if err != nil { + t.Fatalf("first Do error: %v", err) + } + if already { + t.Fatal("first Do should not be treated as duplicate") + } + if calls != 1 { + t.Fatalf("handler calls = %d, want 1", calls) + } + if _, err := os.Stat(d.donePath("evt_first")); err != nil { + t.Fatalf("expected done marker to be written: %v", err) + } + + already, err = d.Do("evt_first", func() error { + calls++ + return nil + }) + if err != nil { + t.Fatalf("second Do error: %v", err) + } + if !already { + t.Fatal("second Do should be treated as duplicate") + } + if calls != 1 { + t.Fatalf("handler calls after duplicate = %d, want 1", calls) + } +} + +func TestStripeWebhookDeduperDoReturnsInFlightWhenLockExists(t *testing.T) { + t.Parallel() + + d := newStripeWebhookDeduper(t.TempDir()) + if err := os.WriteFile(d.lockPath("evt_inflight"), []byte("lock"), 0o600); err != nil { + t.Fatalf("write lock file: %v", err) + } + + calls := 0 + already, err := d.Do("evt_inflight", func() error { + calls++ + return nil + }) + if already { + t.Fatal("in-flight event should not be treated as completed duplicate") + } + if !errors.Is(err, errStripeWebhookEventInFlight) { + t.Fatalf("Do error = %v, want %v", err, errStripeWebhookEventInFlight) + } + if calls != 0 { + t.Fatalf("handler calls = %d, want 0", calls) + } +} + +func TestStripeWebhookDeduperDoPrunesExpiredArtifacts(t *testing.T) { + t.Parallel() + + now := time.Date(2026, time.April, 22, 12, 0, 0, 0, time.UTC) + d := newStripeWebhookDeduper(t.TempDir()) + d.now = func() time.Time { return now } + d.doneTTL = 24 * time.Hour + d.lockTTL = 10 * time.Minute + d.pruneTTL = 0 + + writeArtifact := func(path string, modTime time.Time) { + t.Helper() + if err := os.WriteFile(path, []byte("artifact"), 0o600); err != nil { + t.Fatalf("write artifact %s: %v", path, err) + } + if err := os.Chtimes(path, modTime, modTime); err != nil { + t.Fatalf("chtimes %s: %v", path, err) + } + } + + oldDone := d.donePath("evt_old_done") + recentDone := d.donePath("evt_recent_done") + oldLock := d.lockPath("evt_old_lock") + recentLock := d.lockPath("evt_recent_lock") + oldTmp := d.donePath("evt_old_tmp") + ".tmp" + + writeArtifact(oldDone, now.Add(-d.doneTTL-time.Minute)) + writeArtifact(recentDone, now.Add(-d.doneTTL+time.Minute)) + writeArtifact(oldLock, now.Add(-d.lockTTL-time.Minute)) + writeArtifact(recentLock, now.Add(-d.lockTTL+time.Minute)) + writeArtifact(oldTmp, now.Add(-d.lockTTL-time.Minute)) + + already, err := d.Do("evt_new", func() error { return nil }) + if err != nil { + t.Fatalf("Do error: %v", err) + } + if already { + t.Fatal("new event should not be treated as duplicate") + } + + for _, path := range []string{oldDone, oldLock, oldTmp} { + if _, err := os.Stat(path); !errors.Is(err, os.ErrNotExist) { + t.Fatalf("expected %s to be pruned, stat err=%v", path, err) + } + } + + for _, path := range []string{recentDone, recentLock, d.donePath("evt_new")} { + if _, err := os.Stat(path); err != nil { + t.Fatalf("expected %s to remain, stat err=%v", path, err) + } + } +}