273 lines
10 KiB
Go
273 lines
10 KiB
Go
// SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Commercial
|
|
// SPDX-FileCopyrightText: 2025-2026 Christian Galo
|
|
|
|
// Package web holds Stripe's HTTP handlers (the webhook receiver), moved
|
|
// from internal/server per openspec/changes/integration-extraction task 3.3.
|
|
package web
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"encoding/json"
|
|
"errors"
|
|
"io"
|
|
"log/slog"
|
|
"net/http"
|
|
|
|
internalstripe "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/stripe/store"
|
|
stripe "github.com/stripe/stripe-go/v81"
|
|
"github.com/stripe/stripe-go/v81/webhook"
|
|
)
|
|
|
|
// piiKeys lists JSON keys that are stripped from Stripe payloads before storage.
|
|
var piiKeys = map[string]struct{}{
|
|
"email": {},
|
|
"name": {},
|
|
"phone": {},
|
|
"phone_number": {},
|
|
"address": {},
|
|
"shipping": {},
|
|
"billing_details": {},
|
|
"owner": {},
|
|
// "card" and "bank_account" are intentionally NOT scrubbed: in Stripe webhook
|
|
// payloads these objects contain only safe descriptor fields (brand, last4,
|
|
// exp_month, exp_year, routing_number last4) — no raw card numbers or CVCs,
|
|
// which Stripe never sends over webhooks. Scrubbing them would prevent
|
|
// payment method capture for invoice/payment projection (Phase 1d).
|
|
"customer_email": {},
|
|
"customer_name": {},
|
|
"customer_phone": {},
|
|
"customer_address": {},
|
|
"receipt_email": {},
|
|
"account_holder_name": {},
|
|
}
|
|
|
|
// ScrubPII recursively removes known PII keys from a JSON-decoded value.
|
|
func ScrubPII(v interface{}) interface{} {
|
|
switch val := v.(type) {
|
|
case map[string]interface{}:
|
|
out := make(map[string]interface{}, len(val))
|
|
for k, child := range val {
|
|
if _, isPII := piiKeys[k]; isPII {
|
|
out[k] = "[REDACTED]"
|
|
continue
|
|
}
|
|
out[k] = ScrubPII(child)
|
|
}
|
|
return out
|
|
case []interface{}:
|
|
out := make([]interface{}, len(val))
|
|
for i, child := range val {
|
|
out[i] = ScrubPII(child)
|
|
}
|
|
return out
|
|
default:
|
|
return v
|
|
}
|
|
}
|
|
|
|
// QueuedEvent names a recorded webhook event for the workflow that will
|
|
// process it: the row id and what Stripe called it.
|
|
type QueuedEvent struct {
|
|
ID int64
|
|
ProviderEventID string
|
|
EventType string
|
|
}
|
|
|
|
// StripeWebhookHandler handles incoming Stripe webhook events.
|
|
type StripeWebhookHandler struct {
|
|
DB *sql.DB
|
|
WebhookSecret string
|
|
// KeyMode is the environment the configured API key is in ("live",
|
|
// "test", or "" when no key is configured), handed in beside the
|
|
// signing secret by the adapter that owns both. An event from the
|
|
// other environment is captured and refused against it
|
|
// (stripe-environment-stamp D6); an empty KeyMode refuses nothing,
|
|
// since there is no key to disagree with.
|
|
KeyMode string
|
|
Logger *slog.Logger
|
|
// StartProcessing starts the recorded event's workflow. It must be
|
|
// idempotent for an execution already running, since a redelivery of an
|
|
// unfinished event calls it again. Nil records events without starting
|
|
// anything, which only a handler wired without Temporal does.
|
|
StartProcessing func(ctx context.Context, e QueuedEvent) error
|
|
}
|
|
|
|
// unfinishedStatuses are the row states in which a redelivery re-issues the
|
|
// workflow start: the event was recorded but its workflow may never have
|
|
// started (the process died in between and Stripe got no 200). `refused`
|
|
// is deliberately absent: a refused event never had a workflow and never
|
|
// gets one, so its redelivery is answered and dropped
|
|
// (stripe-environment-stamp D6).
|
|
var unfinishedStatuses = map[string]bool{"received": true, "processing": true, "failed": true}
|
|
|
|
// statusRefused is the row state of an event from the other Stripe
|
|
// environment: captured, so an operator can count it on the provider page,
|
|
// and not processed, because its objects do not exist under the key in
|
|
// force.
|
|
const statusRefused = "refused"
|
|
|
|
// eventEnvironment turns the envelope's livemode into the word the
|
|
// provider_environment column holds. core.webhook_events is
|
|
// provider-neutral, so the column carries Stripe's own vocabulary the way
|
|
// event_type does (design D6).
|
|
func eventEnvironment(livemode bool) string {
|
|
if livemode {
|
|
return internalstripe.ModeLive
|
|
}
|
|
return internalstripe.ModeTest
|
|
}
|
|
|
|
// ServeHTTP verifies the Stripe signature, scrubs PII, inserts the event
|
|
// idempotently into core.webhook_events, and starts the event's workflow.
|
|
// It acknowledges with 2xx only once the event is durably recorded and its
|
|
// workflow started, or the event is a duplicate of a finished one; a failed
|
|
// insert or a failed start answers 5xx so Stripe redelivers instead of the
|
|
// event being silently dropped. The redelivery of an unfinished duplicate
|
|
// starts the workflow again, which is a no-op when it is already running.
|
|
//
|
|
// An event whose environment disagrees with the key's mode is recorded
|
|
// `refused` and answered 200 with no workflow started: the row is the
|
|
// evidence the provider page counts, and Stripe stops redelivering an
|
|
// event this deployment will never process (design D6).
|
|
func (h *StripeWebhookHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPost {
|
|
http.Error(w, "Method not allowed", http.StatusMethodNotAllowed)
|
|
return
|
|
}
|
|
|
|
// Read body (Stripe SDK needs the raw bytes for signature verification)
|
|
const maxBodyBytes = 65536
|
|
body, err := io.ReadAll(io.LimitReader(r.Body, maxBodyBytes))
|
|
if err != nil {
|
|
h.Logger.Error("failed to read webhook body", slog.Any("error", err))
|
|
http.Error(w, "Bad request", http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
// Verify signature with API version mismatch tolerance
|
|
// Stripe CLI and API may use different versions; we handle this gracefully
|
|
sigHeader := r.Header.Get("Stripe-Signature")
|
|
event, err := webhook.ConstructEventWithOptions(body, sigHeader, h.WebhookSecret, webhook.ConstructEventOptions{
|
|
IgnoreAPIVersionMismatch: true,
|
|
})
|
|
if err != nil {
|
|
h.Logger.Warn("invalid Stripe webhook signature", slog.Any("error", err))
|
|
http.Error(w, "Invalid signature", http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
// Log API version mismatches for monitoring
|
|
// The SDK has stripe.APIVersion, and event.APIVersion is the webhook's version
|
|
if event.APIVersion != stripe.APIVersion {
|
|
h.Logger.Warn("Stripe API version mismatch detected",
|
|
slog.String("event_id", event.ID),
|
|
slog.String("event_type", string(event.Type)),
|
|
slog.String("webhook_api_version", event.APIVersion),
|
|
slog.String("sdk_api_version", stripe.APIVersion),
|
|
slog.String("recommendation", "consider upgrading stripe-go SDK"))
|
|
}
|
|
|
|
// Parse payload for PII scrubbing
|
|
var raw interface{}
|
|
if err := json.Unmarshal(event.Data.Raw, &raw); err != nil {
|
|
h.Logger.Error("failed to parse webhook payload", slog.Any("error", err))
|
|
http.Error(w, "Bad payload", http.StatusBadRequest)
|
|
return
|
|
}
|
|
|
|
scrubbed := ScrubPII(raw)
|
|
scrubbedJSON, err := json.Marshal(scrubbed)
|
|
if err != nil {
|
|
h.Logger.Error("failed to marshal scrubbed payload", slog.Any("error", err))
|
|
http.Error(w, "Internal error", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
// Idempotent insert. NOT the table's ON CONFLICT: the partitioned
|
|
// unique key includes received_at, so ON CONFLICT only collapses
|
|
// same-instant duplicates — a Stripe redelivery seconds later would
|
|
// insert a second row and reprocess. WHERE NOT EXISTS dedupes on
|
|
// (provider, provider_event_id) across time; the remaining
|
|
// concurrent-delivery race is the same one ON CONFLICT left open.
|
|
// The envelope's world, which the handler used to discard, and whether
|
|
// it is the world the key can reach. A refused event is recorded like
|
|
// any other and processed like none: its objects exist somewhere the
|
|
// current key cannot read (design D6).
|
|
eventMode := eventEnvironment(event.Livemode)
|
|
refused := h.KeyMode != "" && eventMode != h.KeyMode
|
|
status := "received"
|
|
if refused {
|
|
status = statusRefused
|
|
}
|
|
|
|
var rowID int64
|
|
err = h.DB.QueryRowContext(r.Context(),
|
|
`INSERT INTO core.webhook_events
|
|
(provider, provider_event_id, event_type, payload, status, provider_event_at, provider_environment)
|
|
SELECT $1, $2, $3, $4, $6, CASE WHEN $5::bigint > 0 THEN to_timestamp($5::bigint) END, $7
|
|
WHERE NOT EXISTS (
|
|
SELECT 1 FROM core.webhook_events
|
|
WHERE provider = $1 AND provider_event_id = $2
|
|
)
|
|
RETURNING id`,
|
|
"stripe", event.ID, string(event.Type), scrubbedJSON, event.Created, status, eventMode,
|
|
).Scan(&rowID)
|
|
switch {
|
|
case err == nil && refused:
|
|
h.Logger.Warn("stripe: webhook event mode disagrees with the key; captured, not processed",
|
|
slog.String("event_id", event.ID),
|
|
slog.String("event_type", string(event.Type)),
|
|
slog.String("event_mode", eventMode),
|
|
slog.String("key_mode", h.KeyMode))
|
|
// 200, so Stripe stops redelivering an event nothing here will
|
|
// ever process.
|
|
w.WriteHeader(http.StatusOK)
|
|
return
|
|
case err == nil:
|
|
h.Logger.Info("webhook event received",
|
|
slog.String("event_id", event.ID),
|
|
slog.String("event_type", string(event.Type)))
|
|
case errors.Is(err, sql.ErrNoRows):
|
|
// A duplicate delivery. Finished events are acknowledged and left
|
|
// alone; an unfinished one gets its workflow (re)started, since
|
|
// this redelivery may be Stripe's answer to a process that died
|
|
// between recording the row and starting the workflow.
|
|
var status string
|
|
if err := h.DB.QueryRowContext(r.Context(),
|
|
`SELECT id, status FROM core.webhook_events WHERE provider = $1 AND provider_event_id = $2
|
|
ORDER BY received_at ASC LIMIT 1`,
|
|
"stripe", event.ID).Scan(&rowID, &status); err != nil {
|
|
h.Logger.Error("failed to read the recorded webhook event", slog.String("event_id", event.ID), slog.Any("error", err))
|
|
http.Error(w, "Internal error", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
if !unfinishedStatuses[status] {
|
|
w.WriteHeader(http.StatusOK)
|
|
return
|
|
}
|
|
default:
|
|
h.Logger.Error("failed to insert webhook event",
|
|
slog.String("event_id", event.ID),
|
|
slog.Any("error", err))
|
|
// Nothing was recorded: answer 500 so Stripe redelivers. Returning
|
|
// 200 here would silently drop the event, since Stripe only retries
|
|
// on non-2xx responses.
|
|
http.Error(w, "Internal error", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
|
|
if h.StartProcessing != nil {
|
|
if err := h.StartProcessing(r.Context(), QueuedEvent{ID: rowID, ProviderEventID: event.ID, EventType: string(event.Type)}); err != nil {
|
|
h.Logger.Error("failed to start the webhook event's workflow; the row stays recorded",
|
|
slog.String("event_id", event.ID), slog.Any("error", err))
|
|
// Recorded but not started: answer 500 so Stripe redelivers,
|
|
// and the redelivery starts it (the unfinished branch above).
|
|
http.Error(w, "Internal error", http.StatusInternalServerError)
|
|
return
|
|
}
|
|
}
|
|
|
|
w.WriteHeader(http.StatusOK)
|
|
}
|