`28e7461`

billing: IsBillingEventStaleForPrincipal + TouchBillingLastEventAtForPrincipal

Authored by mfwolffe <wolffemf@dukes.jmu.edu> 2 hours ago

SHA: 28e746195ef35e3093edcb73a6c44dbc7c25a4b5
Parents: 24bdba9
Tree: b83950e

4 changed files

Status	File	+	-
M	`internal/billing/billing.go`	81	0
M	`internal/billing/queries/billing.sql`	31	0
M	`internal/billing/sqlc/billing.sql.go`	131	26
M	`internal/billing/sqlc/querier.go`	15	0

internal/billing/billing.gomodified

  	})
+ }
 +// IsBillingEventStaleForPrincipal reports whether an incoming Stripe
 +// event's timestamp is older than the last event we've already applied
 +// for this principal. PRO08 D4: the handler refuses stale events so
 +// reverse-ordered retries can't regress state (e.g., a stale
 +// subscription.updated[active] arriving after a fresh
 +// subscription.updated[canceled] re-activating the principal).
 +//
 +// Returns false when there's no prior event on file (the first event
 +// is never stale) or when the row simply doesn't exist (defaults to
 +// allow; the caller's own ErrNoRows path handles missing-state).
 +func IsBillingEventStaleForPrincipal(ctx context.Context, deps Deps, p Principal, eventAt time.Time) (bool, error) {
 +	if err := validateDeps(deps); err != nil {
 +		return false, err
 +	}
 +	if err := p.Validate(); err != nil {
 +		return false, err
 +	}
 +	if eventAt.IsZero() {
 +		return false, nil
 +	}
 +	q := billingdb.New()
 +	switch p.Kind {
 +	case SubjectKindOrg:
 +		stale, err := q.IsOrgBillingEventStale(ctx, deps.Pool, billingdb.IsOrgBillingEventStaleParams{
 +			OrgID:   p.ID,
 +			EventAt: pgTime(eventAt),
 +		})
 +		if err != nil {
 +			if errors.Is(err, pgx.ErrNoRows) {
 +				return false, nil
 +			}
 +			return false, err
 +		}
 +		return stale, nil
 +	case SubjectKindUser:
 +		stale, err := q.IsUserBillingEventStale(ctx, deps.Pool, billingdb.IsUserBillingEventStaleParams{
 +			UserID:  p.ID,
 +			EventAt: pgTime(eventAt),
 +		})
 +		if err != nil {
 +			if errors.Is(err, pgx.ErrNoRows) {
 +				return false, nil
 +			}
 +			return false, err
 +		}
 +		return stale, nil
 +	}
 +	return false, nil
 +}
++
 +// TouchBillingLastEventAtForPrincipal bumps last_event_at on the
 +// principal's billing-state row. PRO08 D4: called after a successful
 +// apply so subsequent staleness checks have a baseline. The query
 +// uses GREATEST(prev, incoming) so an out-of-order-but-recent retry
 +// doesn't regress the timestamp.
 +func TouchBillingLastEventAtForPrincipal(ctx context.Context, deps Deps, p Principal, eventAt time.Time) error {
 +	if err := validateDeps(deps); err != nil {
 +		return err
 +	}
 +	if err := p.Validate(); err != nil {
 +		return err
 +	}
 +	if eventAt.IsZero() {
 +		return nil
 +	}
 +	q := billingdb.New()
 +	switch p.Kind {
 +	case SubjectKindOrg:
 +		return q.TouchOrgBillingLastEventAt(ctx, deps.Pool, billingdb.TouchOrgBillingLastEventAtParams{
 +			OrgID:   p.ID,
 +			EventAt: pgTime(eventAt),
 +		})
 +	case SubjectKindUser:
 +		return q.TouchUserBillingLastEventAt(ctx, deps.Pool, billingdb.TouchUserBillingLastEventAtParams{
 +			UserID:  p.ID,
 +			EventAt: pgTime(eventAt),
 +		})
 +	}
 +	return nil
 +}
++
  // 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

internal/billing/queries/billing.sqlmodified

   WHERE provider = 'stripe'
     AND provider_event_id = sqlc.arg(provider_event_id)::text;
 +-- name: IsOrgBillingEventStale :one
 +-- PRO08 D4: returns true when an incoming Stripe event's timestamp
 +-- is older than the last event we've already applied for this org.
 +-- Stripe doesn't guarantee delivery order across retries; without
 +-- this guard a stale `subscription.updated[active]` could re-activate
 +-- a canceled subscription. Returns false when no prior event has
 +-- been recorded (last_event_at IS NULL) — the first event is never
 +-- stale.
 +SELECT COALESCE(last_event_at > sqlc.arg(event_at)::timestamptz, false)::boolean AS stale
 +  FROM org_billing_states
 + WHERE org_id = sqlc.arg(org_id)::bigint;
++
 +-- name: IsUserBillingEventStale :one
 +SELECT COALESCE(last_event_at > sqlc.arg(event_at)::timestamptz, false)::boolean AS stale
 +  FROM user_billing_states
 + WHERE user_id = sqlc.arg(user_id)::bigint;
++
 +-- name: TouchOrgBillingLastEventAt :exec
 +-- PRO08 D4: bump last_event_at on successful apply. Conditional so
 +-- a fresh apply driven by an out-of-order-but-recent retry doesn't
 +-- regress the timestamp (GREATEST). NULL last_event_at acquires the
 +-- incoming value.
 +UPDATE org_billing_states
 +   SET last_event_at = GREATEST(COALESCE(last_event_at, sqlc.arg(event_at)::timestamptz), sqlc.arg(event_at)::timestamptz)
 + WHERE org_id = sqlc.arg(org_id)::bigint;
++
 +-- name: TouchUserBillingLastEventAt :exec
 +UPDATE user_billing_states
 +   SET last_event_at = GREATEST(COALESCE(last_event_at, sqlc.arg(event_at)::timestamptz), sqlc.arg(event_at)::timestamptz)
 + WHERE user_id = sqlc.arg(user_id)::bigint;
++
  -- name: TryAcquireWebhookEventLock :one
  -- PRO08 A3: transaction-scoped advisory lock keyed on the hash of
  -- the provider_event_id. Two concurrent webhook deliveries for the

internal/billing/sqlc/billing.sql.gomodified

                 ELSE org_billing_states.grace_until
             END,
             updated_at = now()
 -    RETURNING org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at
 +    RETURNING org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at
  ), org_update AS (
      UPDATE orgs
         SET plan = $2::org_plan,
       WHERE id = $1::bigint
      RETURNING id
+ )
 -SELECT org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at FROM state
 +SELECT org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at FROM state
+ `
  type ApplySubscriptionSnapshotParams struct {
  	LastWebhookEventID       string
  	CreatedAt                pgtype.Timestamptz
  	UpdatedAt                pgtype.Timestamptz
 +	LastEventAt              pgtype.Timestamptz
+ }
  func (q *Queries) ApplySubscriptionSnapshot(ctx context.Context, db DBTX, arg ApplySubscriptionSnapshotParams) (ApplySubscriptionSnapshotRow, error) {
  		&i.LastWebhookEventID,
  		&i.CreatedAt,
  		&i.UpdatedAt,
 +		&i.LastEventAt,
+ 	)
  	return i, err
+ }
                 ELSE user_billing_states.grace_until
             END,
             updated_at = now()
 -    RETURNING user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at
 +    RETURNING user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at
  ), user_update AS (
      UPDATE users
         SET plan = $2::user_plan,
       WHERE id = $1::bigint
      RETURNING id
+ )
 -SELECT user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at FROM state
 +SELECT user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at FROM state
+ `
  type ApplyUserSubscriptionSnapshotParams struct {
  	LastWebhookEventID       string
  	CreatedAt                pgtype.Timestamptz
  	UpdatedAt                pgtype.Timestamptz
 +	LastEventAt              pgtype.Timestamptz
+ }
  // Mirrors ApplySubscriptionSnapshot for orgs minus the seat columns
  		&i.LastWebhookEventID,
  		&i.CreatedAt,
  		&i.UpdatedAt,
 +		&i.LastEventAt,
+ 	)
  	return i, err
+ }
             grace_until = NULL,
             updated_at = now()
       WHERE org_id = $1
 -    RETURNING org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at
 +    RETURNING org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at
  ), org_update AS (
      UPDATE orgs
         SET plan = state.plan,
       WHERE orgs.id = state.org_id
      RETURNING orgs.id
+ )
 -SELECT org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at FROM state
 +SELECT org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at FROM state
+ `
  type ClearBillingLockRow struct {
  	LastWebhookEventID       string
  	CreatedAt                pgtype.Timestamptz
  	UpdatedAt                pgtype.Timestamptz
 +	LastEventAt              pgtype.Timestamptz
+ }
  func (q *Queries) ClearBillingLock(ctx context.Context, db DBTX, orgID int64) (ClearBillingLockRow, error) {
  		&i.LastWebhookEventID,
  		&i.CreatedAt,
  		&i.UpdatedAt,
 +		&i.LastEventAt,
+ 	)
  	return i, err
+ }
             grace_until = NULL,
             updated_at = now()
       WHERE user_id = $1
 -    RETURNING user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at
 +    RETURNING user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at
  ), user_update AS (
      UPDATE users
         SET plan = state.plan,
       WHERE users.id = state.user_id
      RETURNING users.id
+ )
 -SELECT user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at FROM state
 +SELECT user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at FROM state
+ `
  type ClearUserBillingLockRow struct {
  	LastWebhookEventID       string
  	CreatedAt                pgtype.Timestamptz
  	UpdatedAt                pgtype.Timestamptz
 +	LastEventAt              pgtype.Timestamptz
+ }
  func (q *Queries) ClearUserBillingLock(ctx context.Context, db DBTX, userID int64) (ClearUserBillingLockRow, error) {
  		&i.LastWebhookEventID,
  		&i.CreatedAt,
  		&i.UpdatedAt,
 +		&i.LastEventAt,
+ 	)
  	return i, err
+ }
  const getOrgBillingState = `-- name: GetOrgBillingState :one
 -SELECT org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at FROM org_billing_states WHERE org_id = $1
 +SELECT org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at FROM org_billing_states WHERE org_id = $1
+ `
  // SPDX-License-Identifier: AGPL-3.0-or-later
  		&i.LastWebhookEventID,
  		&i.CreatedAt,
  		&i.UpdatedAt,
 +		&i.LastEventAt,
+ 	)
  	return i, err
+ }
  const getOrgBillingStateByStripeCustomer = `-- name: GetOrgBillingStateByStripeCustomer :one
 -SELECT org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at FROM org_billing_states
 +SELECT org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at FROM org_billing_states
  WHERE provider = 'stripe'
    AND stripe_customer_id = $1
+ `
  		&i.LastWebhookEventID,
  		&i.CreatedAt,
  		&i.UpdatedAt,
 +		&i.LastEventAt,
+ 	)
  	return i, err
+ }
  const getOrgBillingStateByStripeSubscription = `-- name: GetOrgBillingStateByStripeSubscription :one
 -SELECT org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at FROM org_billing_states
 +SELECT org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at FROM org_billing_states
  WHERE provider = 'stripe'
    AND stripe_subscription_id = $1
+ `
  		&i.LastWebhookEventID,
  		&i.CreatedAt,
  		&i.UpdatedAt,
 +		&i.LastEventAt,
+ 	)
  	return i, err
+ }
  const getUserBillingState = `-- name: GetUserBillingState :one
 -SELECT user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at FROM user_billing_states WHERE user_id = $1
 +SELECT user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at FROM user_billing_states WHERE user_id = $1
+ `
  // ─── user_billing_states (PRO03) ──────────────────────────────────
  		&i.LastWebhookEventID,
  		&i.CreatedAt,
  		&i.UpdatedAt,
 +		&i.LastEventAt,
+ 	)
  	return i, err
+ }
  const getUserBillingStateByStripeCustomer = `-- name: GetUserBillingStateByStripeCustomer :one
 -SELECT user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at FROM user_billing_states
 +SELECT user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at FROM user_billing_states
  WHERE provider = 'stripe'
    AND stripe_customer_id = $1
+ `
  		&i.LastWebhookEventID,
  		&i.CreatedAt,
  		&i.UpdatedAt,
 +		&i.LastEventAt,
+ 	)
  	return i, err
+ }
  const getUserBillingStateByStripeSubscription = `-- name: GetUserBillingStateByStripeSubscription :one
 -SELECT user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at FROM user_billing_states
 +SELECT user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at FROM user_billing_states
  WHERE provider = 'stripe'
    AND stripe_subscription_id = $1
+ `
  		&i.LastWebhookEventID,
  		&i.CreatedAt,
  		&i.UpdatedAt,
 +		&i.LastEventAt,
+ 	)
  	return i, err
+ }
  	return i, err
+ }
 +const isOrgBillingEventStale = `-- name: IsOrgBillingEventStale :one
 +SELECT COALESCE(last_event_at > $1::timestamptz, false)::boolean AS stale
 +  FROM org_billing_states
 + WHERE org_id = $2::bigint
 +`
++
 +type IsOrgBillingEventStaleParams struct {
 +	EventAt pgtype.Timestamptz
 +	OrgID   int64
 +}
++
 +// PRO08 D4: returns true when an incoming Stripe event's timestamp
 +// is older than the last event we've already applied for this org.
 +// Stripe doesn't guarantee delivery order across retries; without
 +// this guard a stale `subscription.updated[active]` could re-activate
 +// a canceled subscription. Returns false when no prior event has
 +// been recorded (last_event_at IS NULL) — the first event is never
 +// stale.
 +func (q *Queries) IsOrgBillingEventStale(ctx context.Context, db DBTX, arg IsOrgBillingEventStaleParams) (bool, error) {
 +	row := db.QueryRow(ctx, isOrgBillingEventStale, arg.EventAt, arg.OrgID)
 +	var stale bool
 +	err := row.Scan(&stale)
 +	return stale, err
 +}
++
 +const isUserBillingEventStale = `-- name: IsUserBillingEventStale :one
 +SELECT COALESCE(last_event_at > $1::timestamptz, false)::boolean AS stale
 +  FROM user_billing_states
 + WHERE user_id = $2::bigint
 +`
++
 +type IsUserBillingEventStaleParams struct {
 +	EventAt pgtype.Timestamptz
 +	UserID  int64
 +}
++
 +func (q *Queries) IsUserBillingEventStale(ctx context.Context, db DBTX, arg IsUserBillingEventStaleParams) (bool, error) {
 +	row := db.QueryRow(ctx, isUserBillingEventStale, arg.EventAt, arg.UserID)
 +	var stale bool
 +	err := row.Scan(&stale)
 +	return stale, err
 +}
++
  const listFailedWebhookEvents = `-- name: ListFailedWebhookEvents :many
  SELECT id, provider, provider_event_id, event_type, api_version,
         received_at, processed_at, processing_attempts, process_error,
             last_webhook_event_id = $1::text,
             updated_at = now()
       WHERE org_id = $2::bigint
 -    RETURNING org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at
 +    RETURNING org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at
  ), org_update AS (
      UPDATE orgs
         SET plan = 'free',
       WHERE id = $2::bigint
      RETURNING id
+ )
 -SELECT org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at FROM state
 +SELECT org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at FROM state
+ `
  type MarkCanceledParams struct {
  	LastWebhookEventID       string
  	CreatedAt                pgtype.Timestamptz
  	UpdatedAt                pgtype.Timestamptz
 +	LastEventAt              pgtype.Timestamptz
+ }
  func (q *Queries) MarkCanceled(ctx context.Context, db DBTX, arg MarkCanceledParams) (MarkCanceledRow, error) {
  		&i.LastWebhookEventID,
  		&i.CreatedAt,
  		&i.UpdatedAt,
 +		&i.LastEventAt,
+ 	)
  	return i, err
+ }
         last_webhook_event_id = $2::text,
         updated_at = now()
   WHERE org_id = $3::bigint
 -RETURNING org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at
 +RETURNING org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at
+ `
  type MarkPastDueParams struct {
  		&i.LastWebhookEventID,
  		&i.CreatedAt,
  		&i.UpdatedAt,
 +		&i.LastEventAt,
+ 	)
  	return i, err
+ }
             last_webhook_event_id = $1::text,
             updated_at = now()
       WHERE org_id = $2::bigint
 -    RETURNING org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at
 +    RETURNING org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at
  ), org_update AS (
      UPDATE orgs
         SET plan = state.plan,
       WHERE orgs.id = state.org_id
      RETURNING orgs.id
+ )
 -SELECT org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at FROM state
 +SELECT org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at FROM state
+ `
  type MarkPaymentSucceededParams struct {
  	LastWebhookEventID       string
  	CreatedAt                pgtype.Timestamptz
  	UpdatedAt                pgtype.Timestamptz
 +	LastEventAt              pgtype.Timestamptz
+ }
  func (q *Queries) MarkPaymentSucceeded(ctx context.Context, db DBTX, arg MarkPaymentSucceededParams) (MarkPaymentSucceededRow, error) {
  		&i.LastWebhookEventID,
  		&i.CreatedAt,
  		&i.UpdatedAt,
 +		&i.LastEventAt,
+ 	)
  	return i, err
+ }
             last_webhook_event_id = $1::text,
             updated_at = now()
       WHERE user_id = $2::bigint
 -    RETURNING user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at
 +    RETURNING user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at
  ), user_update AS (
      UPDATE users
         SET plan = 'free',
       WHERE id = $2::bigint
      RETURNING id
+ )
 -SELECT user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at FROM state
 +SELECT user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at FROM state
+ `
  type MarkUserCanceledParams struct {
  	LastWebhookEventID       string
  	CreatedAt                pgtype.Timestamptz
  	UpdatedAt                pgtype.Timestamptz
 +	LastEventAt              pgtype.Timestamptz
+ }
  func (q *Queries) MarkUserCanceled(ctx context.Context, db DBTX, arg MarkUserCanceledParams) (MarkUserCanceledRow, error) {
  		&i.LastWebhookEventID,
  		&i.CreatedAt,
  		&i.UpdatedAt,
 +		&i.LastEventAt,
+ 	)
  	return i, err
+ }
         last_webhook_event_id = $2::text,
         updated_at = now()
   WHERE user_id = $3::bigint
 -RETURNING user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at
 +RETURNING user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at
+ `
  type MarkUserPastDueParams struct {
  		&i.LastWebhookEventID,
  		&i.CreatedAt,
  		&i.UpdatedAt,
 +		&i.LastEventAt,
+ 	)
  	return i, err
+ }
             last_webhook_event_id = $1::text,
             updated_at = now()
       WHERE user_id = $2::bigint
 -    RETURNING user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at
 +    RETURNING user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at
  ), user_update AS (
      UPDATE users
         SET plan = state.plan,
       WHERE users.id = state.user_id
      RETURNING users.id
+ )
 -SELECT user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at FROM state
 +SELECT user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at FROM state
+ `
  type MarkUserPaymentSucceededParams struct {
  	LastWebhookEventID       string
  	CreatedAt                pgtype.Timestamptz
  	UpdatedAt                pgtype.Timestamptz
 +	LastEventAt              pgtype.Timestamptz
+ }
  func (q *Queries) MarkUserPaymentSucceeded(ctx context.Context, db DBTX, arg MarkUserPaymentSucceededParams) (MarkUserPaymentSucceededRow, error) {
  		&i.LastWebhookEventID,
  		&i.CreatedAt,
  		&i.UpdatedAt,
 +		&i.LastEventAt,
+ 	)
  	return i, err
+ }
     SET stripe_customer_id = EXCLUDED.stripe_customer_id,
         provider = 'stripe',
         updated_at = now()
 -RETURNING org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at
 +RETURNING org_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, billable_seats, seat_snapshot_at, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at
+ `
  type SetStripeCustomerParams struct {
  		&i.LastWebhookEventID,
  		&i.CreatedAt,
  		&i.UpdatedAt,
 +		&i.LastEventAt,
+ 	)
  	return i, err
+ }
     SET stripe_customer_id = EXCLUDED.stripe_customer_id,
         provider = 'stripe',
         updated_at = now()
 -RETURNING user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at
 +RETURNING user_id, provider, stripe_customer_id, stripe_subscription_id, stripe_subscription_item_id, plan, subscription_status, current_period_start, current_period_end, cancel_at_period_end, trial_end, past_due_at, canceled_at, locked_at, lock_reason, grace_until, last_webhook_event_id, created_at, updated_at, last_event_at
+ `
  type SetUserStripeCustomerParams struct {
  		&i.LastWebhookEventID,
  		&i.CreatedAt,
  		&i.UpdatedAt,
 +		&i.LastEventAt,
+ 	)
  	return i, err
+ }
  	return err
+ }
 +const touchOrgBillingLastEventAt = `-- name: TouchOrgBillingLastEventAt :exec
 +UPDATE org_billing_states
 +   SET last_event_at = GREATEST(COALESCE(last_event_at, $1::timestamptz), $1::timestamptz)
 + WHERE org_id = $2::bigint
 +`
++
 +type TouchOrgBillingLastEventAtParams struct {
 +	EventAt pgtype.Timestamptz
 +	OrgID   int64
 +}
++
 +// PRO08 D4: bump last_event_at on successful apply. Conditional so
 +// a fresh apply driven by an out-of-order-but-recent retry doesn't
 +// regress the timestamp (GREATEST). NULL last_event_at acquires the
 +// incoming value.
 +func (q *Queries) TouchOrgBillingLastEventAt(ctx context.Context, db DBTX, arg TouchOrgBillingLastEventAtParams) error {
 +	_, err := db.Exec(ctx, touchOrgBillingLastEventAt, arg.EventAt, arg.OrgID)
 +	return err
 +}
++
 +const touchUserBillingLastEventAt = `-- name: TouchUserBillingLastEventAt :exec
 +UPDATE user_billing_states
 +   SET last_event_at = GREATEST(COALESCE(last_event_at, $1::timestamptz), $1::timestamptz)
 + WHERE user_id = $2::bigint
 +`
++
 +type TouchUserBillingLastEventAtParams struct {
 +	EventAt pgtype.Timestamptz
 +	UserID  int64
 +}
++
 +func (q *Queries) TouchUserBillingLastEventAt(ctx context.Context, db DBTX, arg TouchUserBillingLastEventAtParams) error {
 +	_, err := db.Exec(ctx, touchUserBillingLastEventAt, arg.EventAt, arg.UserID)
 +	return err
 +}
++
  const tryAcquireWebhookEventLock = `-- name: TryAcquireWebhookEventLock :one
  SELECT pg_try_advisory_xact_lock(hashtext($1)::bigint) AS acquired
+ `

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)
 +	// PRO08 D4: returns true when an incoming Stripe event's timestamp
 +	// is older than the last event we've already applied for this org.
 +	// Stripe doesn't guarantee delivery order across retries; without
 +	// this guard a stale `subscription.updated[active]` could re-activate
 +	// a canceled subscription. Returns false when no prior event has
 +	// been recorded (last_event_at IS NULL) — the first event is never
 +	// stale.
 +	IsOrgBillingEventStale(ctx context.Context, db DBTX, arg IsOrgBillingEventStaleParams) (bool, error)
 +	IsUserBillingEventStale(ctx context.Context, db DBTX, arg IsUserBillingEventStaleParams) (bool, 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
  	// 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
 +	// PRO08 D4: bump last_event_at on successful apply. Conditional so
 +	// a fresh apply driven by an out-of-order-but-recent retry doesn't
 +	// regress the timestamp (GREATEST). NULL last_event_at acquires the
 +	// incoming value.
 +	TouchOrgBillingLastEventAt(ctx context.Context, db DBTX, arg TouchOrgBillingLastEventAtParams) error
 +	TouchUserBillingLastEventAt(ctx context.Context, db DBTX, arg TouchUserBillingLastEventAtParams) error
  	// PRO08 A3: transaction-scoped advisory lock keyed on the hash of
  	// the provider_event_id. Two concurrent webhook deliveries for the
  	// same event_id race past CreateWebhookEventReceipt before either has