billing: record (subject_kind, subject_id) on webhook receipts + ListFailedWebhookEvents

Status	File	+
M	`internal/billing/billing.go`	39
M	`internal/billing/queries/billing.sql`	30
M	`internal/billing/sqlc/billing.sql.go`	89
M	`internal/billing/sqlc/querier.go`	12

internal/billing/billing.gomodified

  	return billingdb.New().GetWebhookEventReceipt(ctx, deps.Pool, providerEventID)
+ }
++// SetWebhookEventSubjectForPrincipal records the resolved subject on
++// the receipt row. Called after a successful subject-resolution step
++// in the webhook apply path (before guard + state mutation) so the
++// audit trail survives even if the apply later fails. Migration 0075's
++// CHECK constraint enforces both-or-neither — the helper rejects a
++// zero principal.
++func SetWebhookEventSubjectForPrincipal(ctx context.Context, deps Deps, providerEventID string, p Principal) error {
++	if err := validateDeps(deps); err != nil {
++		return err
++	}
++	providerEventID = strings.TrimSpace(providerEventID)
++	if providerEventID == "" {
++		return ErrWebhookEventID
++	}
++	if err := p.Validate(); err != nil {
++		return err
++	}
++	return billingdb.New().SetWebhookEventSubject(ctx, deps.Pool, billingdb.SetWebhookEventSubjectParams{
++		SubjectKind:     billingdb.BillingSubjectKind(p.Kind),
++		SubjectID:       p.ID,
++		ProviderEventID: providerEventID,
++	})
++}
++
++// ListFailedWebhookEvents is the operator query for "events we
++// received but failed to process." Returns rows whose process_error
++// is non-empty OR that have any processing_attempts but no
++// processed_at (in-flight failures). Returned in descending received_at
++// order; limit caps the result set.
++func ListFailedWebhookEvents(ctx context.Context, deps Deps, limit int32) ([]billingdb.ListFailedWebhookEventsRow, error) {
++	if err := validateDeps(deps); err != nil {
++		return nil, err
++	}
++	if limit <= 0 {
++		limit = 50
++	}
++	return billingdb.New().ListFailedWebhookEvents(ctx, deps.Pool, limit)
++}
++
  func UpsertInvoice(ctx context.Context, deps Deps, snap InvoiceSnapshot) (billingdb.BillingInvoice, error) {
  	if err := validateDeps(deps); err != nil {
  		return billingdb.BillingInvoice{}, err

internal/billing/queries/billing.sqlmodified

     AND provider_event_id = $1
  RETURNING *;
++-- name: SetWebhookEventSubject :exec
++-- Records the resolved subject on the receipt row after a successful
++-- subject-resolution step. Called from the apply path before guard +
++-- state mutation so the receipt carries the audit trail even if the
++-- subsequent apply fails. Migration 0075's CHECK constraint enforces
++-- both-or-neither; callers must pass a non-zero subject.
++UPDATE billing_webhook_events
++   SET subject_kind = sqlc.arg(subject_kind)::billing_subject_kind,
++       subject_id   = sqlc.arg(subject_id)::bigint
++ WHERE provider = 'stripe'
++   AND provider_event_id = sqlc.arg(provider_event_id)::text;
++
++-- name: ListFailedWebhookEvents :many
++-- Operator query for "events we received but failed to process."
++-- A row is "failed" when it has a non-empty process_error OR when
++-- it has never been processed (processed_at NULL) and has at least
++-- one processing attempt. Rows that are merely new and untouched
++-- (attempts=0, processed_at NULL, no error) are excluded.
++SELECT id, provider, provider_event_id, event_type, api_version,
++       received_at, processed_at, processing_attempts, process_error,
++       subject_kind, subject_id
++  FROM billing_webhook_events
++ WHERE provider = 'stripe'
++   AND (
++        process_error <> ''
++        OR (processed_at IS NULL AND processing_attempts > 0)
++       )
++ ORDER BY received_at DESC
++ LIMIT $1;
++
  -- ─── user_billing_states (PRO03) ──────────────────────────────────
  -- name: GetUserBillingState :one

internal/billing/sqlc/billing.sql.gomodified

  	return i, err
+ }
++const listFailedWebhookEvents = `-- name: ListFailedWebhookEvents :many
++SELECT id, provider, provider_event_id, event_type, api_version,
++       received_at, processed_at, processing_attempts, process_error,
++       subject_kind, subject_id
++  FROM billing_webhook_events
++ WHERE provider = 'stripe'
++   AND (
++        process_error <> ''
++        OR (processed_at IS NULL AND processing_attempts > 0)
++       )
++ ORDER BY received_at DESC
++ LIMIT $1
++`
++
++type ListFailedWebhookEventsRow struct {
++	ID                 int64
++	Provider           BillingProvider
++	ProviderEventID    string
++	EventType          string
++	ApiVersion         string
++	ReceivedAt         pgtype.Timestamptz
++	ProcessedAt        pgtype.Timestamptz
++	ProcessingAttempts int32
++	ProcessError       string
++	SubjectKind        NullBillingSubjectKind
++	SubjectID          pgtype.Int8
++}
++
++// Operator query for "events we received but failed to process."
++// A row is "failed" when it has a non-empty process_error OR when
++// it has never been processed (processed_at NULL) and has at least
++// one processing attempt. Rows that are merely new and untouched
++// (attempts=0, processed_at NULL, no error) are excluded.
++func (q *Queries) ListFailedWebhookEvents(ctx context.Context, db DBTX, limit int32) ([]ListFailedWebhookEventsRow, error) {
++	rows, err := db.Query(ctx, listFailedWebhookEvents, limit)
++	if err != nil {
++		return nil, err
++	}
++	defer rows.Close()
++	items := []ListFailedWebhookEventsRow{}
++	for rows.Next() {
++		var i ListFailedWebhookEventsRow
++		if err := rows.Scan(
++			&i.ID,
++			&i.Provider,
++			&i.ProviderEventID,
++			&i.EventType,
++			&i.ApiVersion,
++			&i.ReceivedAt,
++			&i.ProcessedAt,
++			&i.ProcessingAttempts,
++			&i.ProcessError,
++			&i.SubjectKind,
++			&i.SubjectID,
++		); err != nil {
++			return nil, err
++		}
++		items = append(items, i)
++	}
++	if err := rows.Err(); err != nil {
++		return nil, err
++	}
++	return items, nil
++}
++
  const listInvoicesForOrg = `-- name: ListInvoicesForOrg :many
  SELECT id, org_id, provider, stripe_invoice_id, stripe_customer_id, stripe_subscription_id, status, number, currency, amount_due_cents, amount_paid_cents, amount_remaining_cents, hosted_invoice_url, invoice_pdf_url, period_start, period_end, due_at, paid_at, voided_at, created_at, updated_at, subject_kind, subject_id FROM billing_invoices
  WHERE subject_kind = 'org' AND subject_id = $1
  	return i, err
+ }
++const setWebhookEventSubject = `-- name: SetWebhookEventSubject :exec
++UPDATE billing_webhook_events
++   SET subject_kind = $1::billing_subject_kind,
++       subject_id   = $2::bigint
++ WHERE provider = 'stripe'
++   AND provider_event_id = $3::text
++`
++
++type SetWebhookEventSubjectParams struct {
++	SubjectKind     BillingSubjectKind
++	SubjectID       int64
++	ProviderEventID string
++}
++
++// Records the resolved subject on the receipt row after a successful
++// subject-resolution step. Called from the apply path before guard +
++// state mutation so the receipt carries the audit trail even if the
++// subsequent apply fails. Migration 0075's CHECK constraint enforces
++// both-or-neither; callers must pass a non-zero subject.
++func (q *Queries) SetWebhookEventSubject(ctx context.Context, db DBTX, arg SetWebhookEventSubjectParams) error {
++	_, err := db.Exec(ctx, setWebhookEventSubject, arg.SubjectKind, arg.SubjectID, arg.ProviderEventID)
++	return err
++}
++
  const upsertInvoice = `-- name: UpsertInvoice :one
  INSERT INTO billing_invoices (

internal/billing/sqlc/querier.gomodified

  	GetUserBillingStateByStripeCustomer(ctx context.Context, db DBTX, stripeCustomerID pgtype.Text) (UserBillingState, error)
  	GetUserBillingStateByStripeSubscription(ctx context.Context, db DBTX, stripeSubscriptionID pgtype.Text) (UserBillingState, error)
  	GetWebhookEventReceipt(ctx context.Context, db DBTX, providerEventID string) (BillingWebhookEvent, error)
++	// Operator query for "events we received but failed to process."
++	// A row is "failed" when it has a non-empty process_error OR when
++	// it has never been processed (processed_at NULL) and has at least
++	// one processing attempt. Rows that are merely new and untouched
++	// (attempts=0, processed_at NULL, no error) are excluded.
++	ListFailedWebhookEvents(ctx context.Context, db DBTX, limit int32) ([]ListFailedWebhookEventsRow, error)
  	// PRO03: filters on the polymorphic subject columns so the index
  	// billing_invoices_subject_created_idx services this query. The
  	// legacy `org_id` column is kept populated by UpsertInvoice for the
  	MarkWebhookEventProcessed(ctx context.Context, db DBTX, providerEventID string) (BillingWebhookEvent, error)
  	SetStripeCustomer(ctx context.Context, db DBTX, arg SetStripeCustomerParams) (OrgBillingState, error)
  	SetUserStripeCustomer(ctx context.Context, db DBTX, arg SetUserStripeCustomerParams) (UserBillingState, error)
++	// Records the resolved subject on the receipt row after a successful
++	// subject-resolution step. Called from the apply path before guard +
++	// state mutation so the receipt carries the audit trail even if the
++	// subsequent apply fails. Migration 0075's CHECK constraint enforces
++	// both-or-neither; callers must pass a non-zero subject.
++	SetWebhookEventSubject(ctx context.Context, db DBTX, arg SetWebhookEventSubjectParams) error
  	// ─── billing_invoices ──────────────────────────────────────────────
  	// PRO03: writes both legacy `org_id` and polymorphic
  	// `(subject_kind, subject_id)`. Callers continue to bind org_id only;

tenseleyflow/shithub / `7296411`

4 changed files