Institutional plan fixes

docs/composable-api-plans-handover.md

@@ -1098,6 +1098,30 @@ $70. Belongs to task **TR**, still not started.
 subscription's 2027 renewal already shows `Applied balance -$3,500.00`, amount due $0.00.
 Product decision needed: is interval switching free, once per period, or renewal-only?
 
+**10. Two concurrent purchases can both charge — fixed for Institutional, still open for bundles.**
+`classify_active_sanapi/1` (`lifecycle.ex:597`) reads the customer's live SanAPI subscriptions
+with a plain `Repo.all` — no lock, no transaction — and the Stripe subscription is created some
+hundreds of milliseconds later. Before the first purchase there is no row to lock, so two
+overlapping requests for one user both see nothing, both pass, and both charge.
+`Subscription.has_active_subscriptions/2` is no defence: it is the same unlocked read, keyed on
+plan id. A double-clicked Buy button is enough, and afterwards nothing treats the pair as wrong —
+`classify_active_sanapi/1` would refuse a *third*, and the hourly replacement job only cancels
+legacy plans. Result: two live subscriptions billing in parallel at $1,050 or $799 a month.
+
+`Sanbase.Billing.Subscription.PurchaseLock` closes it: a per-user session-level advisory lock
+held on one checked-out connection for the whole check-and-create, `pg_try_advisory_lock` so the
+second request is refused at once rather than queueing behind an HTTP call it will fail anyway.
+Deliberately not a `Repo.transaction` — `Subscription.create/2` emits `:create_subscription`, and
+`BillingEventSubscriber` recomputes the quota on its own connection, so inside an uncommitted
+transaction it would read no subscription and write the wrong `api_call_limits` row.
+
+**Applied to the Institutional flow only.** Wrapping `Bundle.Lifecycle.subscribe/2` changes how a
+path that is already in production uses the connection pool — it would hold a connection across
+the Stripe call and the proration cancel — and that deserves its own deploy rather than riding
+along with the Institutional release. The change is one line at `lifecycle.ex:79`:
+`Subscription.PurchaseLock.with_lock(user.id, fn -> ...end)` around the existing body. Do it once
+Institutional is settled on production.
+
 **Not an abuse surface, but found in the same run:** the `/admin/bundle_subscriptions` "make a
 real GraphQL call" card posts to `SanbaseWeb.Endpoint.url()`, i.e. the admin pod's own
 endpoint. `:graphql_cache` is only started when `container_type() in ["web", "all"]`

lib/sanbase/billing/plan/bundle/lifecycle.ex

@@ -69,6 +69,14 @@ defmodule Sanbase.Billing.Plan.Bundle.Lifecycle do
   paid and is refused every request is worse off than one who was not charged. A
   legacy SanAPI subscription that the bundle replaces is canceled with proration
   once the new one is actually live - see `cancel_replaceable_if_live/2`.
+
+  ⚠️ Two overlapping calls for the same user can both pass `classify_for_subscribe/1`
+  and both charge, leaving two bundles billing in parallel - a double-clicked Buy
+  button is enough. `Sanbase.Billing.Subscription.PurchaseLock` is the fix and is
+  already written and applied to the Institutional flow; it is deliberately *not*
+  applied here yet, because doing so changes how a live production path uses the
+  connection pool and that wants its own deploy. See §10.1 of
+  docs/composable-api-plans-handover.md.
   """
   @spec subscribe(User.t(), subscribe_opts()) ::
           {:ok, Subscription.t()} | {:error, term()}

lib/sanbase/billing/subscription/purchase_lock.ex

@@ -0,0 +1,123 @@
+defmodule Sanbase.Billing.Subscription.PurchaseLock do
+  @moduledoc ~s"""
+  One SanAPI new-offering purchase at a time, per user.
+
+  ## The race this closes
+
+  Both purchase flows for the new offering - `Bundle.Lifecycle.subscribe/2` and
+  the Institutional branch of `Subscription.subscribe/4` - check that the customer
+  has no live SanAPI subscription and then, some hundreds of milliseconds later,
+  create one in Stripe. The check is a plain read: `classify_active_sanapi/1` runs
+  `Repo.all` with no lock, because before the first purchase there is no row to
+  lock.
+
+  So two overlapping requests for the same user both see nothing, both pass, and
+  both charge. `Subscription.has_active_subscriptions/2` does not help - it is the
+  same unlocked read, keyed on plan id. The customer ends up with two live
+  subscriptions billing in parallel at $799 or $1,050 a month, and nothing in the
+  system treats that as wrong afterwards: `classify_active_sanapi/1` would refuse a
+  *third*, and the hourly replacement job only cancels legacy plans.
+
+  A double-clicked Buy button is enough.
+
+  ## Why an advisory lock rather than a transaction or a claim row
+
+  **Not `Repo.transaction`.** The critical section has to contain the Stripe call,
+  and `Subscription.create/2` emits `:create_subscription` on the event bus.
+  `BillingEventSubscriber` handles it by recomputing the customer's quota, on its
+  own connection - so inside an uncommitted transaction it would read no
+  subscription and write the wrong `api_call_limits` row.
+
+  **Not post-create winner selection.** It works, but it means deliberately
+  charging the customer twice and then refunding one, which is worse for them than
+  being told to try again.
+
+  **Not a claim row.** It would need a table, a unique index, and a rule for
+  clearing claims left behind by a process that died mid-Stripe. An advisory lock
+  is released by Postgres when the connection goes, which is the same guarantee
+  without the bookkeeping.
+
+  `pg_try_advisory_lock` is used rather than `pg_advisory_lock`: the second request
+  is a double-submit, not work waiting to be done, so it should be told so
+  immediately rather than queue behind an HTTP call to Stripe and then fail the
+  check anyway.
+
+  ## Connection affinity
+
+  A session-level advisory lock belongs to the connection that took it, so the
+  lock, the work and the unlock all have to run on one connection - hence
+  `Repo.checkout/2`. Every query the wrapped function makes uses that same
+  connection, and it is held for as long as Stripe takes to answer. That is the
+  unavoidable cost of making check-and-create atomic; it is bounded by
+  `@timeout`, and it applies only to purchases, which are rare.
+  """
+
+  require Logger
+
+  alias Sanbase.Repo
+
+  # Arbitrary but stable. Advisory locks share one namespace across the database,
+  # so the first argument keeps these from colliding with any other use.
+  @namespace 8412
+
+  # Generous, because a Stripe call sits inside. If it is ever hit, the customer
+  # sees a failed purchase rather than a duplicate one.
+  @timeout :timer.seconds(60)
+
+  @busy_message "A subscription purchase for this account is already in progress. " <>
+                  "Please wait for it to finish before trying again."
+
+  @doc ~s"""
+  Run `fun` while holding this user's purchase lock.
+
+  Returns whatever `fun` returns. If another request already holds the lock,
+  `fun` is **not** run and `{:error, message}` is returned.
+
+  The lock is released whether `fun` returns, raises or throws - and by Postgres
+  itself if the connection dies, so a crash mid-purchase cannot leave a user
+  permanently unable to buy.
+  """
+  @spec with_lock(pos_integer(), (-> result)) :: result | {:error, String.t()}
+        when result: term()
+  def with_lock(user_id, fun) when is_integer(user_id) and is_function(fun, 0) do
+    Repo.checkout(
+      fn ->
+        if acquire(user_id) do
+          try do
+            fun.()
+          after
+            release(user_id)
+          end
+        else
+          Logger.info(
+            "[PurchaseLock] Refused a concurrent SanAPI purchase for user #{user_id} - " <>
+              "another one is already in progress."
+          )
+
+          {:error, @busy_message}
+        end
+      end,
+      timeout: @timeout
+    )
+  end
+
+  @doc ~s"""
+  The message a caller gets when someone else holds the lock.
+
+  Exposed so tests can assert on it without restating it.
+  """
+  @spec busy_message() :: String.t()
+  def busy_message, do: @busy_message
+
+  defp acquire(user_id) do
+    %{rows: [[acquired?]]} =
+      Repo.query!("SELECT pg_try_advisory_lock($1, $2)", [@namespace, user_id])
+
+    acquired?
+  end
+
+  defp release(user_id) do
+    Repo.query!("SELECT pg_advisory_unlock($1, $2)", [@namespace, user_id])
+    :ok
+  end
+end

lib/sanbase/billing/subscription/subscription.ex

@@ -294,6 +294,12 @@ defmodule Sanbase.Billing.Subscription do
   @spec subscribe(%User{}, %Plan{}, string_or_nil, string_or_nil) ::
           {:ok, %__MODULE__{}} | {:error, %Stripe.Error{} | String.t()}
   def subscribe(user, plan, card_token \\ nil, coupon \\ nil) do
+    serialize_new_offering_purchase(user, plan, fn ->
+      do_subscribe(user, plan, card_token, coupon)
+    end)
+  end
+
+  defp do_subscribe(user, plan, card_token, coupon) do
     with {:ok, coupon} <- maybe_validate_san_holder_coupon(user, coupon),
          :ok <- has_active_subscriptions(user, plan),
          :ok <- ensure_plan_is_for_sale(user, plan),
@@ -312,6 +318,12 @@ defmodule Sanbase.Billing.Subscription do
   Subscribe user with payment_method_id to a plan.
   """
   def subscribe2(user, plan, payment_method_id, coupon \\ nil) do
+    serialize_new_offering_purchase(user, plan, fn ->
+      do_subscribe2(user, plan, payment_method_id, coupon)
+    end)
+  end
+
+  defp do_subscribe2(user, plan, payment_method_id, coupon) do
     with {:ok, coupon} <- maybe_validate_san_holder_coupon(user, coupon),
          :ok <- has_active_subscriptions(user, plan),
          :ok <- ensure_plan_is_for_sale(user, plan),
@@ -326,6 +338,21 @@ defmodule Sanbase.Billing.Subscription do
     end
   end
 
+  # Only the new offering is serialized. `ensure_plan_is_for_sale/2` reads the
+  # customer's live SanAPI subscriptions and then charges, so without a lock two
+  # overlapping requests both pass that read and both create a billable Stripe
+  # subscription - see `Subscription.PurchaseLock`.
+  #
+  # Every other plan keeps the exact path it had before, lock and all: their only
+  # coexistence rule is `has_active_subscriptions/2`, which is unchanged, and
+  # holding a connection across Stripe for the whole existing catalogue is a cost
+  # with nothing to buy.
+  defp serialize_new_offering_purchase(user, %Plan{name: "INSTITUTIONAL" <> _}, fun) do
+    __MODULE__.PurchaseLock.with_lock(user.id, fun)
+  end
+
+  defp serialize_new_offering_purchase(_user, _plan, fun), do: fun.()
+
   # Cancel asynchronously to avoid blocking the request. If it fails it is ok but capture the error in sentry
   def maybe_cancel_async(user_id, plan) do
     run = fn ->