Make Square webhook dedup restart-safe via database

The in-memory dedup is now only a fast path; the square_webhook_events INSERT ... ON CONFLICT DO NOTHING is the source of truth, so replays across restarts and after FIFO eviction are skipped. A DB failure fails closed with 503 so Square retries. Empty event_ids are rejected with 400 (no dispatch, no dedup row). Ordering trade-off (insert-before-dispatch) documented for when handlers mutate state.

backend/handlers/webhooks/square.go

@@ -10,6 +10,8 @@ import (
 	"net/http"
 	"os"
 	"sync"
+
+	"crussell/db"
 )
 
 type SquareWebhookEvent struct {
@@ -21,9 +23,10 @@ type SquareWebhookEvent struct {
 }
 
 // squareWebhookDedup is a bounded, mutex-guarded set of recently handled
-// event IDs. Square redelivers signed webhooks on retries (or a replay); once
-// the handlers mutate state a duplicate delivery would double-apply, so drop
-// replays while keeping the set bounded.
+// event IDs. It is a FAST-PATH cache only: the persistent source of truth is
+// the square_webhook_events table (see HandleSquareWebhook). It lets a replayed
+// delivery be dropped without a DB round-trip, but a restart clears it — the DB
+// row is what guarantees at-most-once processing across restarts.
 type squareWebhookDedup struct {
 	mu    sync.Mutex
 	seen  map[string]struct{}
@@ -39,10 +42,19 @@ func newSquareWebhookDedup(max int) *squareWebhookDedup {
 	}
 }
 
-// register reports whether id was already handled: false on first occurrence
-// (recording id, evicting the oldest once the cap is reached), true on a
-// replay (set untouched, preserving insertion order). Mutex-guarded — the
-// handler may be hit concurrently.
+// has reports whether id is in the set WITHOUT recording it. Used as the
+// fast-path short-circuit before the DB dedup insert.
+func (d *squareWebhookDedup) has(id string) bool {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	_, ok := d.seen[id]
+	return ok
+}
+
+// register records id, reporting whether it was already present (set untouched
+// on a replay, preserving insertion order). Mutex-guarded — the handler may be
+// hit concurrently. Called only after the DB insert has confirmed the event's
+// fate, so a failed DB write never leaves a stale entry that would drop a retry.
 func (d *squareWebhookDedup) register(id string) bool {
 	d.mu.Lock()
 	defer d.mu.Unlock()
@@ -59,9 +71,17 @@ func (d *squareWebhookDedup) register(id string) bool {
 	return false
 }
 
-// 1000 IDs far exceeds Square's redelivery window while capping memory.
-var squareWebhookEventsSeen = newSquareWebhookDedup(1000)
+// 500 IDs far exceeds the latency payoff of the fast-path cache; the DB row
+// (square_webhook_events) is the unbounded, restart-safe source of truth.
+var squareWebhookEventsSeen = newSquareWebhookDedup(500)
 
+// HandleSquareWebhook verifies and dispatches Square webhook events.
+//
+// Fail-closed chain: 503 when the signing key is unset, 403 on a missing/bad
+// signature, 400 on malformed JSON or an empty event_id (which cannot be
+// deduplicated — Square always sends one, so this is defensive). A correctly
+// signed, well-formed event is deduplicated by event_id before dispatch and
+// acknowledged 200.
 func HandleSquareWebhook(w http.ResponseWriter, r *http.Request) {
 	r.Body = http.MaxBytesReader(w, r.Body, 512*1024)
 	body, err := io.ReadAll(r.Body)
@@ -110,15 +130,63 @@ func HandleSquareWebhook(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	// Dedup BEFORE dispatch: a correctly signed replay of a handled event
-	// must not re-enter the handlers (which will mutate state once wired).
-	// Returns 200 to acknowledge delivery without processing.
-	if event.EventID != "" && squareWebhookEventsSeen.register(event.EventID) {
+	// An empty event_id cannot be deduplicated. Square always sends event_id,
+	// so this is defensive — but once handlers mutate state, a duplicate
+	// empty-ID event would double-apply. Reject with 400 (fail-safe, no
+	// dispatch, no dedup row): Square's retry policy retries on 5xx/timeouts
+	// but treats 4xx as non-retryable, so the malformed event is dropped
+	// without side effects.
+	if event.EventID == "" {
+		log.Printf("[SQUARE-WEBHOOK] Rejecting event with empty event_id (400)")
+		http.Error(w, "Invalid event", http.StatusBadRequest)
+		return
+	}
+
+	// Dedup BEFORE dispatch: a correctly signed replay of a handled event must
+	// not re-enter the handlers (which will mutate state once wired). The
+	// in-memory fast-path drops recent replays without a DB round-trip; the
+	// square_webhook_events INSERT ... ON CONFLICT DO NOTHING is the source of
+	// truth — 0 rows affected means the event was already handled (persisted
+	// from before a restart, or a concurrent duplicate) and dispatch is
+	// skipped. Returns 200 to acknowledge delivery without processing.
+	//
+	// ORDERING NOTE: the dedup row is committed before dispatch. If the process
+	// crashes between the insert and dispatch, the event is dropped (Square's
+	// retry is 200-skipped). This is acceptable while dispatch is log-only;
+	// when handlers mutate state, switch to dispatch-then-record or make
+	// dispatch idempotent.
+	if event.EventID != "" {
+		if squareWebhookEventsSeen.has(event.EventID) {
 			log.Printf("[SQUARE-WEBHOOK] Duplicate event_id %s; skipping (already processed)", event.EventID)
 			w.WriteHeader(http.StatusOK)
 			_, _ = w.Write([]byte("ok"))
 			return
 		}
+		if db.Conn == nil {
+			log.Printf("[SQUARE-WEBHOOK] DB unavailable — rejecting event_id %s (fail-closed)", event.EventID)
+			http.Error(w, "webhook processing unavailable", http.StatusServiceUnavailable)
+			return
+		}
+		tag, err := db.Conn.Exec(r.Context(),
+			"INSERT INTO square_webhook_events (event_id) VALUES ($1) ON CONFLICT (event_id) DO NOTHING", event.EventID)
+		if err != nil {
+			// Fail closed: without a successful dedup write we cannot prove this
+			// event hasn't been handled before, so reject and let Square retry
+			// later. event_id is not PII, so logging it is safe.
+			log.Printf("[SQUARE-WEBHOOK] Failed to record event_id %s (dedup write failed): %v", event.EventID, err)
+			http.Error(w, "webhook processing unavailable", http.StatusServiceUnavailable)
+			return
+		}
+		// Record in the fast-path cache only after the DB write succeeds, so a
+		// failed write never leaves a stale entry that would drop a retry.
+		squareWebhookEventsSeen.register(event.EventID)
+		if tag.RowsAffected() == 0 {
+			log.Printf("[SQUARE-WEBHOOK] Duplicate event_id %s; skipping (already processed)", event.EventID)
+			w.WriteHeader(http.StatusOK)
+			_, _ = w.Write([]byte("ok"))
+			return
+		}
+	}
 
 	log.Printf("[SQUARE-WEBHOOK] Received event: %s", event.Type)
 

backend/handlers/webhooks/testmain_test.go

@@ -0,0 +1,20 @@
+//go:build test
+
+package webhooks
+
+import (
+	"os"
+	"testing"
+
+	"crussell/db"
+	"crussell/testutils/testdb"
+)
+
+func TestMain(m *testing.M) {
+	pool := testdb.CreateTestDatabase("crussell_test_handlers_webhooks")
+	db.Conn = db.NewPoolProxy(pool)
+	testdb.SeedBaseline(pool)
+	code := m.Run()
+	testdb.DestroyTestDatabase(pool, "crussell_test_handlers_webhooks")
+	os.Exit(code)
+}

backend/handlers/webhooks/webhooks_test.go

@@ -13,8 +13,12 @@ import (
 	"log"
 	"net/http"
 	"net/http/httptest"
+	"os"
 	"strings"
 	"testing"
+
+	"crussell/db"
+	"github.com/jackc/pgx/v5/pgxpool"
 )
 
 // =============================================================================
@@ -126,6 +130,31 @@ func webhookTestEnv(t *testing.T, body []byte) (signature string) {
 	return base64.StdEncoding.EncodeToString(mac.Sum(nil))
 }
 
+// countWebhookEvents returns how many dedup rows exist for an event_id. The
+// handler commits its dedup insert to the pool (no per-test transaction), so a
+// fresh query sees it.
+func countWebhookEvents(t *testing.T, eventID string) int {
+	t.Helper()
+	var n int
+	if err := db.Conn.QueryRow(context.Background(),
+		"SELECT COUNT(*) FROM square_webhook_events WHERE event_id = $1", eventID).Scan(&n); err != nil {
+		t.Fatalf("failed to count square_webhook_events rows: %v", err)
+	}
+	return n
+}
+
+// testDBHost mirrors testdb.dbHost so the broken-pool test can reach the same
+// Postgres instance without importing testdb internals.
+func testDBHost() string {
+	if h := os.Getenv("TEST_DB_HOST"); h != "" {
+		return h
+	}
+	if h := os.Getenv("POSTGRES_HOST"); h != "" {
+		return h
+	}
+	return "localhost"
+}
+
 func TestHandleSquareWebhook_PaymentUpdated(t *testing.T) {
 	event := SquareWebhookEvent{
 		Type:      "payment.updated",
@@ -198,6 +227,38 @@ func TestHandleSquareWebhook_InvalidJSON(t *testing.T) {
 	}
 }
 
+// TestHandleSquareWebhook_EmptyEventID_Rejected verifies a signed event with an
+// empty event_id is rejected with 400 — fail-safe: no dispatch, no dedup row.
+// Square's retry policy treats 4xx as non-retryable, so the malformed event is
+// dropped without side effects.
+func TestHandleSquareWebhook_EmptyEventID_Rejected(t *testing.T) {
+	body := []byte(`{"type":"payment.updated","event_id":"","data":{"id":"payment_empty_id"}}`)
+	sig := webhookTestEnv(t, body)
+
+	var buf bytes.Buffer
+	oldOutput := log.Writer()
+	log.SetOutput(&buf)
+	defer log.SetOutput(oldOutput)
+
+	w := makeWebhookRequest(body, sig, context.Background())
+	if w.Code != http.StatusBadRequest {
+		t.Fatalf("expected 400 for empty event_id, got %d. body: %s", w.Code, w.Body.String())
+	}
+
+	out := buf.String()
+	if !strings.Contains(out, "Rejecting event with empty event_id (400)") {
+		t.Errorf("expected empty event_id rejection log, got:\n%s", out)
+	}
+	if strings.Contains(out, "Received event: payment.updated") {
+		t.Errorf("expected empty event_id to skip dispatch, got:\n%s", out)
+	}
+
+	// No dedup row may be written for an empty event_id.
+	if n := countWebhookEvents(t, ""); n != 0 {
+		t.Errorf("expected no dedup row for empty event_id, got %d", n)
+	}
+}
+
 func TestHandleSquareWebhook_BodyTooLarge(t *testing.T) {
 	// 600KB body exceeds the 512KB limit
 	largeBody := []byte(strings.Repeat("a", 600*1024))
@@ -328,6 +389,11 @@ func TestHandleSquareWebhook_DuplicateEventID(t *testing.T) {
 	if w2.Body.String() != "ok" {
 		t.Errorf("expected replay body 'ok', got %q", w2.Body.String())
 	}
+
+	// The persistent dedup row is written exactly once despite both deliveries.
+	if n := countWebhookEvents(t, event.EventID); n != 1 {
+		t.Errorf("expected exactly 1 persisted dedup row, got %d", n)
+	}
 }
 
 func TestHandleSquareWebhook_DistinctEventIDs(t *testing.T) {
@@ -379,3 +445,144 @@ func TestSquareWebhookDedup_CapEvictsOldest(t *testing.T) {
 		t.Errorf("expected evt_cap_0 to be evicted and treated as new")
 	}
 }
+
+func TestSquareWebhookDedup_HasDoesNotRecord(t *testing.T) {
+	d := newSquareWebhookDedup(3)
+	if d.has("evt_has_1") {
+		t.Error("expected has() on empty set to return false")
+	}
+	if d.register("evt_has_1") {
+		t.Error("expected first register to report new")
+	}
+	if !d.has("evt_has_1") {
+		t.Error("expected has() to observe a registered id")
+	}
+}
+
+// TestHandleSquareWebhook_DedupDispatchOnce verifies the handler body runs
+// exactly once across a delivery + replay, with a single persisted dedup row.
+func TestHandleSquareWebhook_DedupDispatchOnce(t *testing.T) {
+	event := SquareWebhookEvent{
+		Type:      "payment.updated",
+		EventID:   "evt_dispatch_once_1",
+		CreatedAt: "2025-01-01T00:00:00Z",
+		Data:      json.RawMessage(`{"id":"payment_dispatch_once_1"}`),
+	}
+	body, _ := json.Marshal(event)
+	sig := webhookTestEnv(t, body)
+
+	var buf bytes.Buffer
+	oldOutput := log.Writer()
+	log.SetOutput(&buf)
+	defer log.SetOutput(oldOutput)
+
+	w1 := makeWebhookRequest(body, sig, context.Background())
+	if w1.Code != http.StatusOK {
+		t.Fatalf("expected first delivery 200, got %d. body: %s", w1.Code, w1.Body.String())
+	}
+	w2 := makeWebhookRequest(body, sig, context.Background())
+	if w2.Code != http.StatusOK {
+		t.Fatalf("expected replay 200, got %d. body: %s", w2.Code, w2.Body.String())
+	}
+	if w2.Body.String() != "ok" {
+		t.Errorf("expected replay body 'ok', got %q", w2.Body.String())
+	}
+
+	out := buf.String()
+	if got := strings.Count(out, "Received event: payment.updated"); got != 1 {
+		t.Errorf("expected dispatch to run exactly once, saw %d 'Received event' log lines:\n%s", got, out)
+	}
+	if n := countWebhookEvents(t, event.EventID); n != 1 {
+		t.Errorf("expected exactly 1 persisted dedup row, got %d", n)
+	}
+}
+
+// TestHandleSquareWebhook_DedupPersistsAcrossRestart simulates a restart: the
+// event was handled by a previous process whose in-memory cache is gone, but
+// the dedup row survived in the DB — the replayed delivery must be skipped.
+func TestHandleSquareWebhook_DedupPersistsAcrossRestart(t *testing.T) {
+	event := SquareWebhookEvent{
+		Type:      "payment.updated",
+		EventID:   "evt_restart_1",
+		CreatedAt: "2025-01-01T00:00:00Z",
+		Data:      json.RawMessage(`{"id":"payment_restart_1"}`),
+	}
+	body, _ := json.Marshal(event)
+	sig := webhookTestEnv(t, body)
+
+	if _, err := db.Conn.Exec(context.Background(),
+		"INSERT INTO square_webhook_events (event_id) VALUES ($1) ON CONFLICT (event_id) DO NOTHING", event.EventID); err != nil {
+		t.Fatalf("failed to seed dedup row: %v", err)
+	}
+
+	var buf bytes.Buffer
+	oldOutput := log.Writer()
+	log.SetOutput(&buf)
+	defer log.SetOutput(oldOutput)
+
+	w := makeWebhookRequest(body, sig, context.Background())
+	if w.Code != http.StatusOK {
+		t.Fatalf("expected 200 for replayed event, got %d. body: %s", w.Code, w.Body.String())
+	}
+	if w.Body.String() != "ok" {
+		t.Errorf("expected body 'ok', got %q", w.Body.String())
+	}
+	if out := buf.String(); strings.Contains(out, "Received event: payment.updated") {
+		t.Errorf("expected replayed event to skip dispatch, got:\n%s", out)
+	}
+	if n := countWebhookEvents(t, event.EventID); n != 1 {
+		t.Errorf("expected still exactly 1 dedup row, got %d", n)
+	}
+}
+
+// TestHandleSquareWebhook_DedupInsertFails_FailsClosed verifies the handler
+// rejects (503) when the dedup INSERT cannot be persisted, so Square retries.
+func TestHandleSquareWebhook_DedupInsertFails_FailsClosed(t *testing.T) {
+	event := SquareWebhookEvent{
+		Type:      "payment.updated",
+		EventID:   "evt_db_down_1",
+		CreatedAt: "2025-01-01T00:00:00Z",
+		Data:      json.RawMessage(`{"id":"payment_db_down_1"}`),
+	}
+	body, _ := json.Marshal(event)
+	sig := webhookTestEnv(t, body)
+
+	orig := db.Conn
+	defer func() { db.Conn = orig }()
+	// A pool whose database does not exist fails every Exec — simulating a DB
+	// that is unreachable or down.
+	badDSN := fmt.Sprintf("postgres://myuser:mypassword@%s:5432/crussell_test_webhooks_nonexistent?sslmode=disable", testDBHost())
+	badPool, err := pgxpool.New(context.Background(), badDSN)
+	if err != nil {
+		t.Fatalf("failed to create broken pool: %v", err)
+	}
+	defer badPool.Close()
+	db.Conn = db.NewPoolProxy(badPool)
+
+	w := makeWebhookRequest(body, sig, context.Background())
+	if w.Code != http.StatusServiceUnavailable {
+		t.Fatalf("expected 503 when dedup write fails, got %d. body: %s", w.Code, w.Body.String())
+	}
+}
+
+// TestHandleSquareWebhook_DedupNilConn_FailsClosed verifies the handler also
+// rejects (503) when the DB was never wired at all (defensive fail-closed).
+func TestHandleSquareWebhook_DedupNilConn_FailsClosed(t *testing.T) {
+	event := SquareWebhookEvent{
+		Type:      "payment.updated",
+		EventID:   "evt_nil_conn_1",
+		CreatedAt: "2025-01-01T00:00:00Z",
+		Data:      json.RawMessage(`{"id":"payment_nil_conn_1"}`),
+	}
+	body, _ := json.Marshal(event)
+	sig := webhookTestEnv(t, body)
+
+	orig := db.Conn
+	db.Conn = nil
+	defer func() { db.Conn = orig }()
+
+	w := makeWebhookRequest(body, sig, context.Background())
+	if w.Code != http.StatusServiceUnavailable {
+		t.Fatalf("expected 503 when DB is not wired, got %d. body: %s", w.Code, w.Body.String())
+	}
+}