Bound Stripe webhook dedupe retention

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
 }

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)
+		}
+	}
+}