From 5ea89da2ad118b4a08ce44d2e24307265a3d05a8 Mon Sep 17 00:00:00 2001 From: Stephen Adamson Date: Mon, 3 Aug 2026 18:12:35 +0100 Subject: [PATCH] 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 | 100 ++++++++-- backend/handlers/webhooks/testmain_test.go | 20 ++ backend/handlers/webhooks/webhooks_test.go | 207 +++++++++++++++++++++ 3 files changed, 311 insertions(+), 16 deletions(-) create mode 100644 backend/handlers/webhooks/testmain_test.go diff --git a/backend/handlers/webhooks/square.go b/backend/handlers/webhooks/square.go index 4dfe2cd..0b0d705 100644 --- a/backend/handlers/webhooks/square.go +++ b/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,16 +130,64 @@ 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) { - log.Printf("[SQUARE-WEBHOOK] Duplicate event_id %s; skipping (already processed)", event.EventID) - w.WriteHeader(http.StatusOK) - _, _ = w.Write([]byte("ok")) + // 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) switch event.Type { diff --git a/backend/handlers/webhooks/testmain_test.go b/backend/handlers/webhooks/testmain_test.go new file mode 100644 index 0000000..01d80ee --- /dev/null +++ b/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) +} diff --git a/backend/handlers/webhooks/webhooks_test.go b/backend/handlers/webhooks/webhooks_test.go index d925d6a..92163e8 100644 --- a/backend/handlers/webhooks/webhooks_test.go +++ b/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()) + } +}