From 6e72f8776e7d06883b3bd079175290ce23cead8e Mon Sep 17 00:00:00 2001
From: Tsvetozar Penov <tsvetozar.penov@gmail.com>
Date: Thu, 6 Aug 2026 16:40:24 +0200
Subject: [PATCH] Institutional plan fixes

---
 docs/composable-api-plans-handover.md         |  24 ++
 lib/sanbase/billing/plan/bundle/lifecycle.ex  |   8 +
 .../billing/subscription/purchase_lock.ex     | 123 +++++++++
 .../billing/subscription/subscription.ex      |  27 ++
 .../subscription/purchase_lock_test.exs       | 257 ++++++++++++++++++
 5 files changed, 439 insertions(+)
 create mode 100644 lib/sanbase/billing/subscription/purchase_lock.ex
 create mode 100644 test/sanbase/billing/subscription/purchase_lock_test.exs

diff --git a/docs/composable-api-plans-handover.md b/docs/composable-api-plans-handover.md
index fb7341bcc..fa7481694 100644
--- a/docs/composable-api-plans-handover.md
+++ b/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"]`
diff --git a/lib/sanbase/billing/plan/bundle/lifecycle.ex b/lib/sanbase/billing/plan/bundle/lifecycle.ex
index 2a3146b68..0c50a70f0 100644
--- a/lib/sanbase/billing/plan/bundle/lifecycle.ex
+++ b/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()}
diff --git a/lib/sanbase/billing/subscription/purchase_lock.ex b/lib/sanbase/billing/subscription/purchase_lock.ex
new file mode 100644
index 000000000..4aabd70af
--- /dev/null
+++ b/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
diff --git a/lib/sanbase/billing/subscription/subscription.ex b/lib/sanbase/billing/subscription/subscription.ex
index 3e050350d..c501bbab7 100644
--- a/lib/sanbase/billing/subscription/subscription.ex
+++ b/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 ->
diff --git a/test/sanbase/billing/subscription/purchase_lock_test.exs b/test/sanbase/billing/subscription/purchase_lock_test.exs
new file mode 100644
index 000000000..156281b62
--- /dev/null
+++ b/test/sanbase/billing/subscription/purchase_lock_test.exs
@@ -0,0 +1,257 @@
+defmodule Sanbase.Billing.Subscription.PurchaseLockTest do
+  @moduledoc ~s"""
+  Proves that two concurrent SanAPI new-offering purchases cannot both charge.
+
+  ## Why the competing holder is a raw Postgrex connection
+
+  A session-level advisory lock belongs to a *connection*, and it is re-entrant
+  within one: the same session can take the same lock twice and succeed both times.
+  Ecto's SQL sandbox hands every process in a test the same connection, so two
+  `Task`s racing here would share one session, both acquire, and the test would
+  pass whether or not the lock worked at all.
+
+  So the second party is a real connection of its own, opened directly with
+  Postgrex against the same database. That is the only arrangement in which
+  `pg_try_advisory_lock` can actually answer `false`, and therefore the only one in
+  which this test means anything.
+  """
+
+  use Sanbase.DataCase, async: false
+
+  import Mock
+  import Sanbase.Factory
+
+  alias Sanbase.Billing.Plan
+  alias Sanbase.Billing.Subscription
+  alias Sanbase.Billing.Subscription.PurchaseLock
+  alias Sanbase.Repo
+  alias Sanbase.StripeApi
+
+  # Mirrors PurchaseLock's own namespace. Duplicated rather than exposed: the
+  # constant is an implementation detail, and a test that had to be handed it could
+  # not notice it changing.
+  @namespace 8412
+
+  setup context do
+    insert(:role_san_team)
+
+    plan =
+      insert(:plan_pro,
+        id: 9901,
+        name: "INSTITUTIONAL",
+        product_id: context.product_api.id,
+        interval: "month",
+        amount: 79_900,
+        is_private: false,
+        is_deprecated: false,
+        stripe_id: "plan_institutional_month_" <> Ecto.UUID.generate()
+      )
+
+    user = insert(:user, stripe_customer_id: "cus_lock_" <> Ecto.UUID.generate())
+
+    %{plan: plan, user: user}
+  end
+
+  describe "with_lock/2" do
+    test "runs the function and returns its value", %{user: user} do
+      assert {:ok, :ran} = PurchaseLock.with_lock(user.id, fn -> {:ok, :ran} end)
+    end
+
+    test "refuses while another connection holds the same user's lock", %{user: user} do
+      with_foreign_lock(user.id, fn ->
+        assert {:error, message} = PurchaseLock.with_lock(user.id, fn -> flunk("ran anyway") end)
+        assert message == PurchaseLock.busy_message()
+      end)
+    end
+
+    test "a lock on one user does not block another", %{user: user} do
+      other = insert(:user)
+
+      with_foreign_lock(other.id, fn ->
+        assert {:ok, :ran} = PurchaseLock.with_lock(user.id, fn -> {:ok, :ran} end)
+      end)
+    end
+
+    test "releases the lock when the function returns", %{user: user} do
+      # Asserted from another connection on purpose. A second `with_lock/2` in this
+      # process would succeed even if nothing were released, because an advisory
+      # lock is re-entrant within the session that holds it - so that assertion
+      # would prove nothing at all.
+      assert {:ok, :done} = PurchaseLock.with_lock(user.id, fn -> {:ok, :done} end)
+
+      assert foreign_can_lock?(user.id)
+    end
+
+    test "releases the lock when the function raises", %{user: user} do
+      # A crash mid-purchase must not leave the customer permanently unable to buy.
+      assert_raise RuntimeError, fn ->
+        PurchaseLock.with_lock(user.id, fn -> raise "boom" end)
+      end
+
+      assert foreign_can_lock?(user.id)
+    end
+
+    test "holds the lock while the function runs", %{user: user} do
+      # The other half of the above: proves the lock is genuinely taken, not that
+      # `with_lock/2` merely runs things and returns true from a stub.
+      PurchaseLock.with_lock(user.id, fn ->
+        refute foreign_can_lock?(user.id)
+      end)
+    end
+  end
+
+  describe "the Institutional purchase flow" do
+    test "a second concurrent purchase creates no Stripe subscription at all", context do
+      %{user: user, plan: plan} = context
+
+      with_mocks([
+        {StripeApi, [:passthrough],
+         [
+           update_customer_card: fn _, _ ->
+             {:ok, %Stripe.Customer{id: "cus_should_not_happen"}}
+           end,
+           create_subscription: fn _ ->
+             {:ok, %Stripe.Subscription{id: "sub_should_not_happen"}}
+           end
+         ]}
+      ]) do
+        with_foreign_lock(user.id, fn ->
+          assert {:error, message} = Subscription.subscribe(user, plan, "card_token")
+          assert message == PurchaseLock.busy_message()
+
+          # The whole point: the loser never reaches Stripe, so there is no second
+          # subscription to cancel and nothing to refund. `update_customer_card/2`
+          # is the first Stripe call the flow would make - asserting on it proves
+          # the refusal happens before *any* of them, not just before the charge.
+          assert_not_called(StripeApi.create_subscription(:_))
+          assert_not_called(StripeApi.update_customer_card(:_, :_))
+        end)
+      end
+
+      assert institutional_subscriptions(user) == []
+    end
+
+    test "only one Institutional subscription remains billable after a serialized pair",
+         context do
+      %{user: user, plan: plan} = context
+
+      # The winner and the loser, run in the order the lock imposes on them. The
+      # second call is the one that matters: by the time it runs the first has
+      # committed, so `ensure_plan_is_for_sale/2` can finally see it and refuses -
+      # which is exactly what the lock exists to guarantee it can do.
+      winner_id =
+        with_mocks(stripe_mocks()) do
+          assert {:ok, first} = Subscription.subscribe(user, plan, "card_token")
+          assert first.status == :active
+
+          assert {:error, %Subscription.Error{message: message}} =
+                   Subscription.subscribe(user, plan, "card_token")
+
+          # `has_active_subscriptions/2` catches this one, because both calls name
+          # the same plan id and it runs first. The offering check is what catches a
+          # *different* new-offering plan id - the yearly row, or a bundle - and that
+          # case is covered in Sanbase.Billing.Plan.InstitutionalTest.
+          assert message == "You are already subscribed to Sanapi by Santiment / INSTITUTIONAL"
+
+          first.id
+        end
+
+      assert [%Subscription{id: ^winner_id}] = institutional_subscriptions(user)
+    end
+
+    test "the lock does not stand in the way of a legitimate purchase", context do
+      %{user: user, plan: plan} = context
+
+      with_mocks(stripe_mocks()) do
+        assert {:ok, subscription} = Subscription.subscribe(user, plan, "card_token")
+        assert subscription.plan.name == "INSTITUTIONAL"
+      end
+
+      assert length(institutional_subscriptions(user)) == 1
+    end
+  end
+
+  # --- helpers ---
+
+  defp stripe_mocks do
+    [
+      {StripeApi, [:passthrough],
+       [
+         create_product: fn _ -> Sanbase.StripeApiTestResponse.create_product_resp() end,
+         create_plan: fn _ -> Sanbase.StripeApiTestResponse.create_plan_resp() end,
+         create_customer_with_card: fn _, _ ->
+           Sanbase.StripeApiTestResponse.create_or_update_customer_resp()
+         end,
+         # The one that actually fires here: the user already has a
+         # stripe_customer_id, so the card token updates the customer rather than
+         # creating one.
+         update_customer_card: fn _, _ ->
+           Sanbase.StripeApiTestResponse.create_or_update_customer_resp()
+         end,
+         create_coupon: fn _ -> Sanbase.StripeApiTestResponse.create_coupon_resp() end,
+         retrieve_coupon: fn coupon -> {:ok, %Stripe.Coupon{id: coupon, percent_off: 20}} end,
+         create_subscription: fn _ ->
+           Sanbase.StripeApiTestResponse.create_subscription_resp()
+         end
+       ]},
+      {Sanbase.Messaging.Discord, [:passthrough], [send_notification: fn _, _, _ -> :ok end]},
+      {Sanbase.TemplateMailer, [:passthrough],
+       send: fn _, _, _ -> {:ok, %{"status" => "sent"}} end}
+    ]
+  end
+
+  defp institutional_subscriptions(user) do
+    import Ecto.Query
+
+    from(s in Subscription,
+      join: p in Plan,
+      on: p.id == s.plan_id,
+      where: s.user_id == ^user.id and p.name == "INSTITUTIONAL",
+      where: s.status in [:active, :past_due, :trialing],
+      order_by: [asc: s.id]
+    )
+    |> Repo.all()
+  end
+
+  # Holds the user's advisory lock on a connection of its own for the duration of
+  # `fun`, then releases it and disconnects.
+  defp with_foreign_lock(user_id, fun) do
+    {:ok, conn} = start_foreign_connection()
+
+    %Postgrex.Result{rows: [[true]]} =
+      Postgrex.query!(conn, "SELECT pg_try_advisory_lock($1, $2)", [@namespace, user_id])
+
+    try do
+      fun.()
+    after
+      Postgrex.query!(conn, "SELECT pg_advisory_unlock($1, $2)", [@namespace, user_id])
+      GenServer.stop(conn)
+    end
+  end
+
+  defp foreign_can_lock?(user_id) do
+    {:ok, conn} = start_foreign_connection()
+
+    try do
+      %Postgrex.Result{rows: [[acquired?]]} =
+        Postgrex.query!(conn, "SELECT pg_try_advisory_lock($1, $2)", [@namespace, user_id])
+
+      if acquired? do
+        Postgrex.query!(conn, "SELECT pg_advisory_unlock($1, $2)", [@namespace, user_id])
+      end
+
+      acquired?
+    after
+      GenServer.stop(conn)
+    end
+  end
+
+  # `Repo.config/0` rather than `Application.get_env/2`: the connection details come
+  # from DATABASE_URL at runtime, so the compile-time config names a role that does
+  # not exist on a developer machine. Only the resolved config is usable.
+  defp start_foreign_connection do
+    Repo.config()
+    |> Keyword.take([:username, :password, :hostname, :port, :database])
+    |> Postgrex.start_link()
+  end
+end