member-console/internal/integrations/stripe/web/webhook_db_test.go

// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial
// SPDX-FileCopyrightText: 2025-2026 Christian Galo

// DB-backed webhook tests live in an external test package: an in-package
// test importing internal/migrate would close the test-binary-only cycle
// web(test) → migrate → integrations → stripe adapter → web
// (docs/building-an-integration.md §8).
package web_test

import (
	"bytes"
	"context"
	"database/sql"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"net/http/httptest"
	"os"
	"testing"
	"time"

	"git.coopcloud.tech/wiki-cafe/member-console/internal/db"
	"git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/stripe/web"
	"git.coopcloud.tech/wiki-cafe/member-console/internal/migrate"
	"github.com/google/uuid"
	_ "github.com/jackc/pgx/v5/stdlib"
	"github.com/stripe/stripe-go/v81/webhook"
)

const testSecret = "whsec_test_dedupe"

func testDB(t *testing.T) *sql.DB {
	t.Helper()
	dsn := os.Getenv("TEST_DATABASE_URL")
	if dsn == "" {
		t.Skip("TEST_DATABASE_URL not set, skipping integration test")
	}
	database, err := sql.Open("pgx", dsn)
	if err != nil {
		t.Fatalf("open database: %v", err)
	}
	if err := db.RunMigrations(database, migrate.Sources()); err != nil {
		t.Fatalf("run migrations: %v", err)
	}
	t.Cleanup(func() { database.Close() })
	return database
}

// deliver POSTs payload to the handler with a valid Stripe signature.
func deliver(t *testing.T, h *web.StripeWebhookHandler, payload []byte) int {
	t.Helper()
	now := time.Now()
	sig := webhook.ComputeSignature(now, payload, testSecret)
	req := httptest.NewRequest(http.MethodPost, "/webhooks/stripe", bytes.NewReader(payload))
	req.Header.Set("Stripe-Signature", fmt.Sprintf("t=%d,v1=%x", now.Unix(), sig))
	rec := httptest.NewRecorder()
	h.ServeHTTP(rec, req)
	return rec.Code
}

// Scenario: Stripe redelivers the same event at a later instant (its retry
// behavior on timeouts). The recording must dedupe on
// (provider, provider_event_id) across time — the partitioned table's
// ON CONFLICT target includes received_at and only collapses same-instant
// duplicates, which is how a redelivery used to insert (and reprocess) a
// second row.
func TestWebhookDuplicateDeliveryAcrossTime(t *testing.T) {
	database := testDB(t)
	h := &web.StripeWebhookHandler{DB: database, WebhookSecret: testSecret, Logger: slog.Default()}

	eventID := "evt_test_" + uuid.New().String()[:12]
	t.Cleanup(func() {
		_, _ = database.Exec(
			`DELETE FROM core.webhook_events WHERE provider = 'stripe' AND provider_event_id = $1`, eventID)
	})
	payload := []byte(fmt.Sprintf(
		`{"id": %q, "type": "customer.created", "data": {"object": {"id": "cus_test", "email": "dup@example.com"}}}`,
		eventID))

	if code := deliver(t, h, payload); code != http.StatusOK {
		t.Fatalf("first delivery = %d, want 200", code)
	}
	time.Sleep(20 * time.Millisecond) // distinct received_at — the case ON CONFLICT misses
	if code := deliver(t, h, payload); code != http.StatusOK {
		t.Fatalf("second delivery = %d, want 200", code)
	}

	var rows int
	if err := database.QueryRow(
		`SELECT COUNT(*) FROM core.webhook_events WHERE provider = 'stripe' AND provider_event_id = $1`,
		eventID).Scan(&rows); err != nil {
		t.Fatalf("count rows: %v", err)
	}
	if rows != 1 {
		t.Fatalf("duplicate delivery recorded %d rows, want 1", rows)
	}

	// The recorded payload must be the scrubbed form, not the raw body.
	var stored string
	if err := database.QueryRow(
		`SELECT payload::text FROM core.webhook_events WHERE provider = 'stripe' AND provider_event_id = $1`,
		eventID).Scan(&stored); err != nil {
		t.Fatalf("read payload: %v", err)
	}
	if bytes.Contains([]byte(stored), []byte("dup@example.com")) {
		t.Errorf("stored payload retains PII email: %s", stored)
	}
}

// recordingStarter stands in for the per-event workflow start: it records
// what the endpoint asked to start and can refuse, the way a Temporal
// outage would.
type recordingStarter struct {
	calls []web.QueuedEvent
	err   error
}

func (s *recordingStarter) start(_ context.Context, e web.QueuedEvent) error {
	s.calls = append(s.calls, e)
	return s.err
}

// The endpoint records the event, then starts its workflow, and answers 200
// only when both happened (stripe-integration-infrastructure "Stripe webhook
// endpoint ingests events asynchronously", 2026-09 audit remediation D8).
func TestWebhookStartsTheEventWorkflow(t *testing.T) {
	database := testDB(t)
	starter := &recordingStarter{}
	h := &web.StripeWebhookHandler{DB: database, WebhookSecret: testSecret, Logger: slog.Default(), StartProcessing: starter.start}

	eventID := "evt_start_" + uuid.New().String()[:12]
	t.Cleanup(func() {
		_, _ = database.Exec(`DELETE FROM core.webhook_events WHERE provider = 'stripe' AND provider_event_id = $1`, eventID)
	})
	payload := []byte(fmt.Sprintf(`{"id": %q, "type": "customer.created", "created": 1757400000, "data": {"object": {"id": "cus_start"}}}`, eventID))

	if code := deliver(t, h, payload); code != http.StatusOK {
		t.Fatalf("delivery = %d, want 200", code)
	}
	var rowID int64
	var eventAt time.Time
	if err := database.QueryRow(`SELECT id, provider_event_at FROM core.webhook_events WHERE provider = 'stripe' AND provider_event_id = $1`, eventID).Scan(&rowID, &eventAt); err != nil {
		t.Fatalf("read row: %v", err)
	}
	if eventAt.Unix() != 1757400000 {
		t.Errorf("the event's own time must be recorded, got %v", eventAt)
	}
	if len(starter.calls) != 1 || starter.calls[0] != (web.QueuedEvent{ID: rowID, ProviderEventID: eventID, EventType: "customer.created"}) {
		t.Fatalf("the workflow must be started once with the recorded row, got %+v", starter.calls)
	}

	// A redelivery of an unfinished event starts the workflow again (a
	// no-op when it is running), which is what recovers a process that died
	// between recording and starting.
	if code := deliver(t, h, payload); code != http.StatusOK {
		t.Fatalf("redelivery = %d, want 200", code)
	}
	if len(starter.calls) != 2 {
		t.Errorf("a redelivery of an unfinished event must re-issue the start, got %d calls", len(starter.calls))
	}

	// A redelivery of a finished event is acknowledged and starts nothing.
	if _, err := database.Exec(`UPDATE core.webhook_events SET status = 'completed' WHERE id = $1`, rowID); err != nil {
		t.Fatal(err)
	}
	if code := deliver(t, h, payload); code != http.StatusOK {
		t.Fatalf("redelivery of a finished event = %d, want 200", code)
	}
	if len(starter.calls) != 2 {
		t.Errorf("a finished event must not be started again, got %d calls", len(starter.calls))
	}
}

// A start that fails leaves the row recorded and answers 500, so Stripe
// redelivers; the redelivery, with Temporal back, starts the workflow.
func TestWebhookAnswers500WhenTheWorkflowCannotStart(t *testing.T) {
	database := testDB(t)
	starter := &recordingStarter{err: errors.New("temporal unavailable")}
	h := &web.StripeWebhookHandler{DB: database, WebhookSecret: testSecret, Logger: slog.Default(), StartProcessing: starter.start}

	eventID := "evt_nostart_" + uuid.New().String()[:12]
	t.Cleanup(func() {
		_, _ = database.Exec(`DELETE FROM core.webhook_events WHERE provider = 'stripe' AND provider_event_id = $1`, eventID)
	})
	payload := []byte(fmt.Sprintf(`{"id": %q, "type": "customer.created", "data": {"object": {"id": "cus_nostart"}}}`, eventID))

	if code := deliver(t, h, payload); code != http.StatusInternalServerError {
		t.Fatalf("delivery with no Temporal = %d, want 500", code)
	}
	var rows int
	if err := database.QueryRow(`SELECT COUNT(*) FROM core.webhook_events WHERE provider = 'stripe' AND provider_event_id = $1`, eventID).Scan(&rows); err != nil {
		t.Fatal(err)
	}
	if rows != 1 {
		t.Fatalf("the event must stay recorded, got %d rows", rows)
	}

	starter.err = nil
	if code := deliver(t, h, payload); code != http.StatusOK {
		t.Fatalf("redelivery with Temporal back = %d, want 200", code)
	}
	if len(starter.calls) != 2 {
		t.Errorf("the redelivery must start the workflow, got %d calls", len(starter.calls))
	}
}