member-console/internal/integrations/stripe/workflows/webhook.go

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

package workflows

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"strings"
	"time"

	internalstripe "git.coopcloud.tech/wiki-cafe/member-console/internal/integrations/stripe/store"
	"git.coopcloud.tech/wiki-cafe/member-console/internal/workflows/common"
	enumspb "go.temporal.io/api/enums/v1"
	"go.temporal.io/sdk/activity"
	"go.temporal.io/sdk/client"
	"go.temporal.io/sdk/temporal"
	"go.temporal.io/sdk/workflow"
)

// WebhookEvent is the handle a workflow carries for one core.webhook_events
// row: enough to name the row and the event in the Temporal UI, never the
// payload (activities read that from the row).
type WebhookEvent struct {
	ID              int64  `db:"id"`
	Provider        string `db:"provider"`
	ProviderEventID string `db:"provider_event_id"`
	EventType       string `db:"event_type"`
	// Status and RetryCount are the row's state when the handle was built.
	// Nothing in the workflow reads them: Temporal counts the attempts and
	// the row is re-read by every activity. They stay for the tests that
	// build handles from rows.
	Status     string `db:"status"`
	RetryCount int    `db:"retry_count"`
}

// WebhookEventWorkflowID is the Workflow ID for one Stripe event: Stripe's
// own event id, which is unique per event, so a redelivery that reaches the
// endpoint and starts the workflow again lands on the running execution
// (WorkflowIDConflictPolicy USE_EXISTING) instead of a second one.
func WebhookEventWorkflowID(providerEventID string) string {
	return "stripe-webhook-" + providerEventID
}

// The retry budget for processing one event. Temporal owns the schedule:
// the activity is retried from one second, doubling to ten minutes, for as
// long as the budget allows; a handler that declares its error
// non-retryable (checkedInt32's AmountOutOfRange) ends it at once. When the
// budget is spent the workflow dead-letters the row for an operator (the
// Stripe provider page lists them) and fails, which is what the Temporal UI
// should show. A day covers a database outage, a worker that needs a
// deploy, and an invoice.paid that arrived before its invoice.finalized
// was projected. (2026-09 security audit, remediation design D8.)
const (
	webhookRetryBudget      = 24 * time.Hour
	webhookRetryMaxInterval = 10 * time.Minute
	webhookAttemptTimeout   = time.Minute
)

// ProcessStripeWebhookEvent is one Stripe event's whole life after the
// endpoint recorded it: process it, retrying on Temporal's schedule, and
// either mark it done or dead-letter it. One execution per event, started
// by the webhook endpoint (and by the boot sweep for rows left unfinished).
func ProcessStripeWebhookEvent(ctx workflow.Context, evt WebhookEvent) error {
	var acts *WebhookActivities

	processCtx := workflow.WithActivityOptions(ctx, workflow.ActivityOptions{
		StartToCloseTimeout:    webhookAttemptTimeout,
		ScheduleToCloseTimeout: webhookRetryBudget,
		RetryPolicy: &temporal.RetryPolicy{
			InitialInterval:    time.Second,
			BackoffCoefficient: 2.0,
			MaximumInterval:    webhookRetryMaxInterval,
			// MaximumAttempts stays 0: the budget, not a count, bounds it.
		},
	})
	err := workflow.ExecuteActivity(processCtx, acts.ProcessWebhookEvent, evt).Get(ctx, nil)
	if err == nil {
		return nil
	}

	workflow.GetLogger(ctx).Error("stripe webhook event exhausted its retries; dead-lettering",
		"event_id", evt.ProviderEventID, "event_type", evt.EventType, "error", err)
	markCtx := workflow.WithActivityOptions(ctx, common.DefaultActivityOptions())
	if markErr := workflow.ExecuteActivity(markCtx, acts.MarkEventDeadLetter, evt.ID, err.Error()).Get(ctx, nil); markErr != nil {
		workflow.GetLogger(ctx).Error("could not dead-letter the event row", "event_id", evt.ProviderEventID, "error", markErr)
	}
	return err
}

// StartWebhookEvent starts the event's workflow, idempotently: a start for
// an execution that is already running is not an error. The endpoint calls
// this after recording the row; the boot sweep calls it for rows an earlier
// process left unfinished.
func StartWebhookEvent(ctx context.Context, c client.Client, taskQueue string, evt WebhookEvent) error {
	_, err := c.ExecuteWorkflow(ctx, client.StartWorkflowOptions{
		ID:                       WebhookEventWorkflowID(evt.ProviderEventID),
		TaskQueue:                taskQueue,
		WorkflowIDConflictPolicy: enumspb.WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING,
	}, ProcessStripeWebhookEvent, evt)
	if err != nil && temporal.IsWorkflowExecutionAlreadyStartedError(err) {
		return nil
	}
	return err
}

// unfinishedStripeEventsSQL selects the rows the boot sweep hands to their
// workflows: Stripe's, in any state short of finished.
const unfinishedStripeEventsSQL = `SELECT id, provider, provider_event_id, event_type
	 FROM core.webhook_events
	 WHERE provider = 'stripe' AND status IN ('received', 'processing', 'failed')
	 ORDER BY received_at ASC`

// SweepUnfinishedWebhookEvents starts a workflow for every Stripe row that
// is not finished (`received`, `processing`, `failed`). Run once at boot: it
// picks up rows recorded by a process that died before starting their
// workflow, and, on the release that replaced the polling processor, the
// rows that processor left behind. Each start is idempotent, so a row whose
// workflow is already running costs one no-op call.
func SweepUnfinishedWebhookEvents(ctx context.Context, c client.Client, taskQueue string, db *sql.DB, logger *slog.Logger) error {
	rows, err := db.QueryContext(ctx, unfinishedStripeEventsSQL)
	if err != nil {
		return fmt.Errorf("list unfinished webhook events: %w", err)
	}
	defer rows.Close()
	var started, failed int
	for rows.Next() {
		var e WebhookEvent
		if err := rows.Scan(&e.ID, &e.Provider, &e.ProviderEventID, &e.EventType); err != nil {
			return fmt.Errorf("scan unfinished webhook event: %w", err)
		}
		if err := StartWebhookEvent(ctx, c, taskQueue, e); err != nil {
			failed++
			logger.Warn("boot sweep: could not start the event's workflow",
				slog.String("event_id", e.ProviderEventID), slog.Any("error", err))
			continue
		}
		started++
	}
	if err := rows.Err(); err != nil {
		return fmt.Errorf("unfinished webhook events: %w", err)
	}
	if started+failed > 0 {
		logger.Info("boot sweep: unfinished stripe webhook events handed to their workflows",
			slog.Int("started", started), slog.Int("failed", failed))
	}
	return nil
}

// WebhookActivities holds dependencies for webhook processing activities.
type WebhookActivities struct {
	DB     *sql.DB
	Logger *slog.Logger
}

// NewWebhookActivities creates a new WebhookActivities instance.
func NewWebhookActivities(db *sql.DB, logger *slog.Logger) *WebhookActivities {
	return &WebhookActivities{DB: db, Logger: logger}
}

// ProcessWebhookEvent transitions an event to 'processing', dispatches it,
// and marks it 'completed' or 'skipped'. A failed attempt is recorded on the
// row (`failed`, the attempt number Temporal reports, the error) before the
// error goes back to Temporal to schedule the next one, so the operator
// page and the row agree with the Temporal UI about where an event stands.
func (a *WebhookActivities) ProcessWebhookEvent(ctx context.Context, evt WebhookEvent) error {
	// Transition to processing
	_, err := a.DB.ExecContext(ctx,
		`UPDATE core.webhook_events
		 SET status = 'processing', updated_at = NOW()
		 WHERE id = $1 AND received_at = (SELECT received_at FROM core.webhook_events WHERE id = $1)`,
		evt.ID,
	)
	if err != nil {
		return fmt.Errorf("mark processing: %w", err)
	}

	// Dispatch to event-type handler.
	finalStatus, err := a.dispatchEvent(ctx, evt)
	if err != nil {
		a.recordFailedAttempt(ctx, evt.ID, currentAttempt(ctx), err)
		// A non-retryable *temporal.ApplicationError (e.g. the checked money
		// conversions in webhook_invoice.go — schema-hardening D6) must reach
		// the Temporal worker as-is: the SDK's failure conversion type-switches
		// on the returned error's own dynamic type to decide whether to retry,
		// so wrapping it with fmt.Errorf's "%w" here would bury it inside a
		// plain *fmt.wrapError and the activity would retry-spin on an amount
		// that will never fit. Errors of any other shape keep the same
		// dispatch-context wrapping as before.
		var appErr *temporal.ApplicationError
		if errors.As(err, &appErr) {
			return appErr
		}
		return fmt.Errorf("dispatch %s: %w", evt.EventType, err)
	}

	// Mark final status
	_, err = a.DB.ExecContext(ctx,
		`UPDATE core.webhook_events
		 SET status = $1, processed_at = NOW(), updated_at = NOW()
		 WHERE id = $2 AND received_at = (SELECT received_at FROM core.webhook_events WHERE id = $2)`,
		finalStatus, evt.ID,
	)
	if err != nil {
		return fmt.Errorf("mark %s: %w", finalStatus, err)
	}

	return nil
}

// currentAttempt is Temporal's attempt number for this activity execution,
// or 1 when the activity runs outside Temporal (tests call it directly).
func currentAttempt(ctx context.Context) int32 {
	if activity.IsActivity(ctx) {
		return activity.GetInfo(ctx).Attempt
	}
	return 1
}

// recordFailedAttempt writes a failed attempt to the row. Best effort: the
// error that matters is the one going back to Temporal, and a row that
// could not be updated is usually the same outage.
func (a *WebhookActivities) recordFailedAttempt(ctx context.Context, eventID int64, attempt int32, procErr error) {
	if _, err := a.DB.ExecContext(ctx,
		`UPDATE core.webhook_events
		 SET status = 'failed', retry_count = $1, error_message = $2, updated_at = NOW()
		 WHERE id = $3 AND received_at = (SELECT received_at FROM core.webhook_events WHERE id = $3)`,
		attempt, procErr.Error(), eventID,
	); err != nil {
		a.Logger.Warn("could not record the failed attempt on the webhook event row",
			slog.Int64("event_id", eventID), slog.Any("error", err))
	}
}

// MarkEventDeadLetter records that an event's workflow gave up: the retry
// budget is spent or the failure was terminal. The row keeps the last
// attempt count and error; only an operator moves it from here.
func (a *WebhookActivities) MarkEventDeadLetter(ctx context.Context, eventID int64, errMsg string) error {
	if _, err := a.DB.ExecContext(ctx,
		`UPDATE core.webhook_events
		 SET status = 'dead_letter', error_message = $1, updated_at = NOW()
		 WHERE id = $2 AND received_at = (SELECT received_at FROM core.webhook_events WHERE id = $2)`,
		errMsg, eventID,
	); err != nil {
		return fmt.Errorf("mark dead_letter: %w", err)
	}
	a.Logger.Warn("webhook event moved to dead letter", slog.Int64("event_id", eventID), slog.String("error", errMsg))
	return nil
}

// supersededByNewerEvent reports whether a completed event for the same
// Stripe object carries a later event time than evt. Stripe does not
// guarantee delivery order, and the product, price and customer handlers
// project what an event says about its object (synced, deleted, active),
// so an older event applied after a newer one would put back a state Stripe
// has already moved past. The event time is Stripe's `created` on the
// event, recorded by the endpoint; rows from before it was recorded compare
// as never superseded.
func (a *WebhookActivities) supersededByNewerEvent(ctx context.Context, evt WebhookEvent, objectID string) (bool, error) {
	var superseded bool
	err := a.DB.QueryRowContext(ctx,
		`SELECT EXISTS (
		     SELECT 1 FROM core.webhook_events newer
		     WHERE newer.provider = 'stripe'
		       AND newer.status = 'completed'
		       AND newer.payload->>'id' = $1
		       AND newer.provider_event_at > (SELECT provider_event_at FROM core.webhook_events WHERE id = $2)
		 )`,
		objectID, evt.ID,
	).Scan(&superseded)
	if err != nil {
		return false, fmt.Errorf("check for a newer applied event: %w", err)
	}
	if superseded {
		a.Logger.Info("stale webhook event skipped: a newer event for the object was already applied",
			slog.String("event_id", evt.ProviderEventID), slog.String("event_type", evt.EventType), slog.String("object_id", objectID))
	}
	return superseded, nil
}

// dispatchEvent routes webhook events to type-specific handlers.
// Returns the final status string ("completed" or "skipped").
func (a *WebhookActivities) dispatchEvent(ctx context.Context, evt WebhookEvent) (string, error) {
	switch {
	case strings.HasPrefix(evt.EventType, "checkout.session."):
		return a.handleCheckoutSessionEvent(ctx, evt)
	case strings.HasPrefix(evt.EventType, "customer.subscription."):
		return a.handleSubscriptionEvent(ctx, evt)
	case strings.HasPrefix(evt.EventType, "customer."):
		return a.handleCustomerEvent(ctx, evt)
	case strings.HasPrefix(evt.EventType, "product."):
		return a.handleProductEvent(ctx, evt)
	case strings.HasPrefix(evt.EventType, "price."):
		return a.handlePriceEvent(ctx, evt)
	case strings.HasPrefix(evt.EventType, "invoice."):
		return a.handleInvoiceEvent(ctx, evt)
	case strings.HasPrefix(evt.EventType, "payment_method."):
		return a.handlePaymentMethodEvent(ctx, evt)
	default:
		a.Logger.Info("webhook event skipped (unhandled type)",
			slog.String("event_id", evt.ProviderEventID),
			slog.String("event_type", evt.EventType))
		return "skipped", nil
	}
}

// webhookCustomerPayload represents the scrubbed payload for customer events.
type webhookCustomerPayload struct {
	ID string `json:"id"` // Stripe Customer ID (cus_…)
	// Livemode is the object's own flag, the environment the id lives in
	// (stripe-environment-stamp D1). It is read from the object inside the
	// stored payload, not from the event envelope; every Stripe object
	// carries it, so a parsed payload always has a real answer here.
	Livemode bool `json:"livemode"`
}

// handleCustomerEvent processes customer.created, customer.updated, customer.deleted events.
func (a *WebhookActivities) handleCustomerEvent(ctx context.Context, evt WebhookEvent) (string, error) {
	// Read the scrubbed payload from the webhook_events table.
	var payloadBytes []byte
	err := a.DB.QueryRowContext(ctx,
		`SELECT payload FROM core.webhook_events
		 WHERE id = $1 AND received_at = (SELECT received_at FROM core.webhook_events WHERE id = $1)`,
		evt.ID,
	).Scan(&payloadBytes)
	if err != nil {
		return "", fmt.Errorf("read webhook payload: %w", err)
	}

	var payload webhookCustomerPayload
	if err := json.Unmarshal(payloadBytes, &payload); err != nil {
		return "", fmt.Errorf("parse customer webhook payload: %w", err)
	}

	stripeCustomerID := payload.ID
	if stripeCustomerID == "" {
		return "", fmt.Errorf("customer webhook payload missing id")
	}
	if stale, err := a.supersededByNewerEvent(ctx, evt, stripeCustomerID); err != nil {
		return "", err
	} else if stale {
		return "skipped", nil
	}

	q := internalstripe.New(a.DB)

	switch evt.EventType {
	case "customer.created":
		// Idempotent upsert: if mapping already exists (written by outbox), update timestamp.
		// If not, create with sync_status = 'synced'.
		// We need the billing_account_id — look up by stripe_customer_id first.
		existing, err := q.GetCustomerMappingByStripeCustomerID(ctx, sql.NullString{String: stripeCustomerID, Valid: true})
		if err == nil {
			// The upsert rather than UpdateCustomerMappingSyncStatus: same
			// row, same status, same touched updated_at, and it can carry
			// the environment the parsed customer reports
			// (stripe-environment-stamp D1). The status-only update has no
			// column for the flag, so a mapping whose first news of Stripe
			// is customer.created would stay unstamped.
			if _, err := q.UpsertCustomerMapping(ctx, internalstripe.UpsertCustomerMappingParams{
				BillingAccountID: existing.BillingAccountID,
				StripeCustomerID: sql.NullString{String: stripeCustomerID, Valid: true},
				SyncStatus:       "synced",
				Livemode:         sql.NullBool{Bool: payload.Livemode, Valid: true},
			}); err != nil {
				return "", fmt.Errorf("update existing mapping: %w", err)
			}
		} else if err == sql.ErrNoRows {
			// No mapping yet — the outbox hasn't run or this was created externally.
			// We can't create a mapping without a billing_account_id, so log and skip.
			a.Logger.Warn("customer.created webhook: no mapping found for stripe customer (created externally?)",
				slog.String("stripe_customer_id", stripeCustomerID))
			return "completed", nil
		} else {
			return "", fmt.Errorf("lookup mapping by stripe customer: %w", err)
		}

	case "customer.updated":
		// Touch updated_at on the mapping row.
		existing, err := q.GetCustomerMappingByStripeCustomerID(ctx, sql.NullString{String: stripeCustomerID, Valid: true})
		if err == sql.ErrNoRows {
			a.Logger.Warn("customer.updated webhook: no mapping found",
				slog.String("stripe_customer_id", stripeCustomerID))
			return "completed", nil
		} else if err != nil {
			return "", fmt.Errorf("lookup mapping: %w", err)
		}
		// The upsert in place of UpdateCustomerMappingSyncStatus: same row,
		// same status, same touched updated_at, and it can carry the
		// environment the parsed customer reports
		// (stripe-environment-stamp D1).
		if _, err := q.UpsertCustomerMapping(ctx, internalstripe.UpsertCustomerMappingParams{
			BillingAccountID: existing.BillingAccountID,
			StripeCustomerID: existing.StripeCustomerID,
			SyncStatus:       existing.SyncStatus,
			Livemode:         sql.NullBool{Bool: payload.Livemode, Valid: true},
		}); err != nil {
			return "", fmt.Errorf("update mapping timestamp: %w", err)
		}

	case "customer.deleted":
		existing, err := q.GetCustomerMappingByStripeCustomerID(ctx, sql.NullString{String: stripeCustomerID, Valid: true})
		if err == sql.ErrNoRows {
			a.Logger.Warn("customer.deleted webhook: no mapping found",
				slog.String("stripe_customer_id", stripeCustomerID))
			return "completed", nil
		} else if err != nil {
			return "", fmt.Errorf("lookup mapping: %w", err)
		}
		if err := q.UpdateCustomerMappingSyncStatus(ctx, internalstripe.UpdateCustomerMappingSyncStatusParams{
			BillingAccountID: existing.BillingAccountID,
			SyncStatus:       "deleted",
		}); err != nil {
			return "", fmt.Errorf("mark mapping deleted: %w", err)
		}

	default:
		a.Logger.Info("customer sub-event skipped",
			slog.String("event_type", evt.EventType))
		return "skipped", nil
	}

	a.Logger.Info("customer webhook processed",
		slog.String("event_type", evt.EventType),
		slog.String("stripe_customer_id", stripeCustomerID))
	return "completed", nil
}

// webhookProductPayload represents the scrubbed payload for product events.
type webhookProductPayload struct {
	ID string `json:"id"` // Stripe Product ID (prod_…)
	// Livemode is the object's own flag, the environment the id lives in
	// (stripe-environment-stamp D1). It is read from the object inside the
	// stored payload, not from the event envelope; every Stripe object
	// carries it, so a parsed payload always has a real answer here.
	Livemode bool `json:"livemode"`
}

// handleProductEvent processes product.created, product.updated, product.deleted events.
func (a *WebhookActivities) handleProductEvent(ctx context.Context, evt WebhookEvent) (string, error) {
	var payloadBytes []byte
	err := a.DB.QueryRowContext(ctx,
		`SELECT payload FROM core.webhook_events
		 WHERE id = $1 AND received_at = (SELECT received_at FROM core.webhook_events WHERE id = $1)`,
		evt.ID,
	).Scan(&payloadBytes)
	if err != nil {
		return "", fmt.Errorf("read webhook payload: %w", err)
	}

	var payload webhookProductPayload
	if err := json.Unmarshal(payloadBytes, &payload); err != nil {
		return "", fmt.Errorf("parse product webhook payload: %w", err)
	}

	stripeProductID := payload.ID
	if stripeProductID == "" {
		return "", fmt.Errorf("product webhook payload missing id")
	}
	if stale, err := a.supersededByNewerEvent(ctx, evt, stripeProductID); err != nil {
		return "", err
	} else if stale {
		return "skipped", nil
	}

	q := internalstripe.New(a.DB)

	switch evt.EventType {
	case "product.created":
		existing, err := q.GetProductMappingByStripeID(ctx, sql.NullString{String: stripeProductID, Valid: true})
		if err == nil {
			// The mapping exists, so re-upsert with the same status to
			// touch updated_at, and stamp the environment the parsed product
			// reports (stripe-environment-stamp D1), the same flag the
			// updated branch writes.
			_, err = q.UpsertProductMapping(ctx, internalstripe.UpsertProductMappingParams{
				ProductID:       existing.ProductID,
				StripeProductID: sql.NullString{String: stripeProductID, Valid: true},
				SyncStatus:      "synced",
				Livemode:        sql.NullBool{Bool: payload.Livemode, Valid: true},
			})
			if err != nil {
				return "", fmt.Errorf("update existing product mapping: %w", err)
			}
		} else if err == sql.ErrNoRows {
			// No mapping yet — webhook arrived before outbox. Can't create without product_id.
			a.Logger.Warn("product.created webhook: no mapping found for stripe product (created externally?)",
				slog.String("stripe_product_id", stripeProductID))
			return "completed", nil
		} else {
			return "", fmt.Errorf("lookup product mapping by stripe id: %w", err)
		}

	case "product.updated":
		existing, err := q.GetProductMappingByStripeID(ctx, sql.NullString{String: stripeProductID, Valid: true})
		if err == sql.ErrNoRows {
			a.Logger.Warn("product.updated webhook: no mapping found",
				slog.String("stripe_product_id", stripeProductID))
			return "completed", nil
		} else if err != nil {
			return "", fmt.Errorf("lookup product mapping: %w", err)
		}
		_, err = q.UpsertProductMapping(ctx, internalstripe.UpsertProductMappingParams{
			ProductID:       existing.ProductID,
			StripeProductID: existing.StripeProductID,
			SyncStatus:      existing.SyncStatus,
			Livemode:        sql.NullBool{Bool: payload.Livemode, Valid: true},
		})
		if err != nil {
			return "", fmt.Errorf("update product mapping timestamp: %w", err)
		}

	case "product.deleted":
		existing, err := q.GetProductMappingByStripeID(ctx, sql.NullString{String: stripeProductID, Valid: true})
		if err == sql.ErrNoRows {
			a.Logger.Warn("product.deleted webhook: no mapping found",
				slog.String("stripe_product_id", stripeProductID))
			return "completed", nil
		} else if err != nil {
			return "", fmt.Errorf("lookup product mapping: %w", err)
		}
		if err := q.MarkProductMappingDeleted(ctx, existing.ProductID); err != nil {
			return "", fmt.Errorf("mark product mapping deleted: %w", err)
		}

	default:
		a.Logger.Info("product sub-event skipped",
			slog.String("event_type", evt.EventType))
		return "skipped", nil
	}

	a.Logger.Info("product webhook processed",
		slog.String("event_type", evt.EventType),
		slog.String("stripe_product_id", stripeProductID))
	return "completed", nil
}

// webhookPricePayload represents the scrubbed payload for price events.
type webhookPricePayload struct {
	ID     string `json:"id"`     // Stripe Price ID (price_…)
	Active bool   `json:"active"` // Whether the price is active
	// Livemode is the object's own flag, the environment the id lives in
	// (stripe-environment-stamp D1). It is read from the object inside the
	// stored payload, not from the event envelope; every Stripe object
	// carries it, so a parsed payload always has a real answer here.
	Livemode bool `json:"livemode"`
}

// handlePriceEvent processes price.created and price.updated events.
func (a *WebhookActivities) handlePriceEvent(ctx context.Context, evt WebhookEvent) (string, error) {
	var payloadBytes []byte
	err := a.DB.QueryRowContext(ctx,
		`SELECT payload FROM core.webhook_events
		 WHERE id = $1 AND received_at = (SELECT received_at FROM core.webhook_events WHERE id = $1)`,
		evt.ID,
	).Scan(&payloadBytes)
	if err != nil {
		return "", fmt.Errorf("read webhook payload: %w", err)
	}

	var payload webhookPricePayload
	if err := json.Unmarshal(payloadBytes, &payload); err != nil {
		return "", fmt.Errorf("parse price webhook payload: %w", err)
	}

	stripePriceID := payload.ID
	if stripePriceID == "" {
		return "", fmt.Errorf("price webhook payload missing id")
	}
	if stale, err := a.supersededByNewerEvent(ctx, evt, stripePriceID); err != nil {
		return "", err
	} else if stale {
		return "skipped", nil
	}

	q := internalstripe.New(a.DB)

	switch evt.EventType {
	case "price.created":
		existing, err := q.GetPriceMappingByStripeID(ctx, sql.NullString{String: stripePriceID, Valid: true})
		if err == nil {
			// The mapping exists, so re-upsert with the same status to
			// touch updated_at, and stamp the environment the parsed price
			// reports (stripe-environment-stamp D1), the same flag the
			// updated branch writes.
			_, err = q.UpsertPriceMapping(ctx, internalstripe.UpsertPriceMappingParams{
				PriceID:       existing.PriceID,
				StripePriceID: sql.NullString{String: stripePriceID, Valid: true},
				SyncStatus:    "synced",
				Livemode:      sql.NullBool{Bool: payload.Livemode, Valid: true},
			})
			if err != nil {
				return "", fmt.Errorf("update existing price mapping: %w", err)
			}
		} else if err == sql.ErrNoRows {
			// No mapping yet — webhook arrived before outbox. Can't create without price_id.
			a.Logger.Warn("price.created webhook: no mapping found for stripe price (created externally?)",
				slog.String("stripe_price_id", stripePriceID))
			return "completed", nil
		} else {
			return "", fmt.Errorf("lookup price mapping by stripe id: %w", err)
		}

	case "price.updated":
		existing, err := q.GetPriceMappingByStripeID(ctx, sql.NullString{String: stripePriceID, Valid: true})
		if err == sql.ErrNoRows {
			a.Logger.Warn("price.updated webhook: no mapping found",
				slog.String("stripe_price_id", stripePriceID))
			return "completed", nil
		} else if err != nil {
			return "", fmt.Errorf("lookup price mapping: %w", err)
		}

		if !payload.Active {
			// Stripe archives prices rather than deleting them.
			if err := q.MarkPriceMappingDeleted(ctx, existing.PriceID); err != nil {
				return "", fmt.Errorf("mark price mapping deleted: %w", err)
			}
		} else {
			// Active price update — bump updated_at.
			_, err = q.UpsertPriceMapping(ctx, internalstripe.UpsertPriceMappingParams{
				PriceID:       existing.PriceID,
				StripePriceID: existing.StripePriceID,
				SyncStatus:    existing.SyncStatus,
				Livemode:      sql.NullBool{Bool: payload.Livemode, Valid: true},
			})
			if err != nil {
				return "", fmt.Errorf("update price mapping timestamp: %w", err)
			}
		}

	default:
		a.Logger.Info("price sub-event skipped",
			slog.String("event_type", evt.EventType))
		return "skipped", nil
	}

	a.Logger.Info("price webhook processed",
		slog.String("event_type", evt.EventType),
		slog.String("stripe_price_id", stripePriceID))
	return "completed", nil
}