make transfers idempotent (#56)

Author: zcesur

lib/algora/contracts/contracts.ex

@@ -482,11 +482,14 @@ defmodule Algora.Contracts do
   defp transfer_funds(contract, %Transaction{type: :transfer} = transaction) when transaction.status != :succeeded do
     with {:ok, account} <- Repo.fetch_by(Account, user_id: transaction.user_id),
          {:ok, stripe_transfer} <-
-           Algora.PSP.Transfer.create(%{
-             amount: MoneyUtils.to_minor_units(transaction.net_amount),
-             currency: to_string(transaction.net_amount.currency),
-             destination: account.provider_id
-           }) do
+           Algora.PSP.Transfer.create(
+             %{
+               amount: MoneyUtils.to_minor_units(transaction.net_amount),
+               currency: to_string(transaction.net_amount.currency),
+               destination: account.provider_id
+             },
+             %{idempotency_key: transaction.id}
+           ) do
       update_transaction_status(transaction, stripe_transfer, :succeeded)
       mark_contract_as_paid(contract)
       {:ok, stripe_transfer}

lib/algora/payments/payments.ex

@@ -411,7 +411,7 @@ defmodule Algora.Payments do
     end
   end
 
-  defp initialize_transfer(%Transaction{} = credit) do
+  def initialize_transfer(%Transaction{} = credit) do
     %Transaction{}
     |> change(%{
       id: Nanoid.generate(),
@@ -438,8 +438,8 @@ defmodule Algora.Payments do
     |> Repo.insert()
   end
 
-  defp execute_transfer(%Transaction{} = transaction, account) do
-    charge = Repo.get_by(Transaction, type: :credit, status: :succeeded, group_id: transaction.group_id)
+  def execute_transfer(%Transaction{} = transaction, account) do
+    charge = Repo.get_by(Transaction, type: :charge, status: :succeeded, group_id: transaction.group_id)
 
     transfer_params =
       %{
@@ -451,23 +451,21 @@ defmodule Algora.Payments do
       |> Map.merge(if transaction.group_id, do: %{transfer_group: transaction.group_id}, else: %{})
       |> Map.merge(if charge && charge.provider_id, do: %{source_transaction: charge.provider_id}, else: %{})
 
-    # TODO: provide idempotency key
-    case PSP.Transfer.create(transfer_params) do
+    case PSP.Transfer.create(transfer_params, %{idempotency_key: transaction.id}) do
       {:ok, transfer} ->
-        # it's fine if this fails since we'll receive a webhook
         transaction
         |> change(%{
           status: :succeeded,
           succeeded_at: DateTime.utc_now(),
           provider_id: transfer.id,
+          provider_transfer_id: transfer.id,
           provider_meta: Util.normalize_struct(transfer)
         })
         |> Repo.update()
 
         {:ok, transfer}
 
       {:error, error} ->
-        # TODO: inconsistent state if this fails
         transaction
         |> change(%{status: :failed})
         |> Repo.update()

lib/algora/psp/psp.ex

@@ -47,7 +47,21 @@ defmodule Algora.PSP do
   defmodule Transfer do
     @moduledoc false
 
-    def create(params), do: Algora.PSP.client(__MODULE__).create(params)
+    @spec create(params, options) :: {:ok, Algora.PSP.transfer()} | {:error, Algora.PSP.error()}
+          when params: %{
+                 :amount => pos_integer,
+                 :currency => String.t(),
+                 :destination => String.t(),
+                 optional(:metadata) => Stripe.Types.metadata(),
+                 optional(:source_transaction) => String.t(),
+                 optional(:transfer_group) => String.t(),
+                 optional(:description) => String.t(),
+                 optional(:source_type) => String.t()
+               },
+               options: %{
+                 :idempotency_key => String.t()
+               }
+    def create(params, opts), do: Algora.PSP.client(__MODULE__).create(params, Keyword.new(opts))
   end
 
   @type session :: Stripe.Session.t()

test/algora/payments_test.exs

@@ -10,14 +10,14 @@ defmodule Algora.PaymentsTest do
   alias Algora.Payments.Transaction
   alias Algora.Repo
 
-  describe "execute_pending_transfer/1" do
-    setup do
-      user = insert(:user)
-      account = insert(:account, user: user)
+  setup do
+    user = insert(:user)
+    account = insert(:account, user: user)
 
-      {:ok, user: user, account: account}
-    end
+    {:ok, user: user, account: account}
+  end
 
+  describe "execute_pending_transfer/1" do
     test "executes transfer when user has positive balance", %{user: user, account: account} do
       credit =
         insert(:transaction,
@@ -106,6 +106,68 @@ defmodule Algora.PaymentsTest do
       transfer_tx = Repo.one(from t in Transaction, where: t.type == :transfer)
       assert transfer_tx.status == :failed
     end
+
+    test "prevents duplicate transfers", %{user: user} do
+      credit_tx =
+        insert(:transaction,
+          type: :credit,
+          status: :succeeded,
+          net_amount: Money.new(1, :USD),
+          group_id: Nanoid.generate(),
+          user: user
+        )
+
+      {:ok, transfer} = Payments.execute_pending_transfer(credit_tx.id)
+      {:ok, transfer_tx} = Repo.fetch_by(Transaction, type: :transfer, provider_id: transfer.id)
+
+      assert Money.equal?(transfer_tx.net_amount, Money.new(1, :USD))
+      assert transfer_tx.status == :succeeded
+      assert transfer_tx.succeeded_at != nil
+      assert transfer_tx.group_id == credit_tx.group_id
+      assert transfer_tx.user_id == credit_tx.user_id
+      assert transfer_tx.provider_id == transfer.id
+
+      {result, _log} = with_log(fn -> Payments.execute_pending_transfer(credit_tx.id) end)
+      assert {:error, :duplicate_transfer_attempt} = result
+
+      transfer_tx |> change(status: :succeeded) |> Repo.update!()
+      {result, _log} = with_log(fn -> Payments.execute_pending_transfer(credit_tx.id) end)
+      assert {:error, :duplicate_transfer_attempt} = result
+
+      transfer_tx |> change(status: :initialized) |> Repo.update!()
+      {result, _log} = with_log(fn -> Payments.execute_pending_transfer(credit_tx.id) end)
+      assert {:error, :duplicate_transfer_attempt} = result
+
+      transfer_tx |> change(status: :processing) |> Repo.update!()
+      {result, _log} = with_log(fn -> Payments.execute_pending_transfer(credit_tx.id) end)
+      assert {:error, :duplicate_transfer_attempt} = result
+    end
+
+    test "allows retrying failed/canceled transfers", %{user: user, account: account} do
+      credit_tx =
+        insert(:transaction,
+          type: :credit,
+          status: :succeeded,
+          net_amount: Money.new(1, :USD),
+          group_id: Nanoid.generate(),
+          user: user
+        )
+
+      Account |> Repo.one!() |> change(%{provider_id: "acct_invalid"}) |> Repo.update!()
+      {result, _log} = with_log(fn -> Payments.execute_pending_transfer(credit_tx.id) end)
+      assert {:error, %Stripe.Error{code: :invalid_request_error}} = result
+
+      Account |> Repo.one!() |> change(%{provider_id: account.provider_id}) |> Repo.update!()
+      {:ok, transfer} = Payments.execute_pending_transfer(credit_tx.id)
+      {:ok, transfer_tx} = Repo.fetch_by(Transaction, type: :transfer, status: :succeeded)
+
+      assert Money.equal?(transfer_tx.net_amount, Money.new(1, :USD))
+      assert transfer_tx.status == :succeeded
+      assert transfer_tx.succeeded_at != nil
+      assert transfer_tx.group_id == credit_tx.group_id
+      assert transfer_tx.user_id == credit_tx.user_id
+      assert transfer_tx.provider_id == transfer.id
+    end
   end
 
   describe "enqueue_pending_transfers/1" do
@@ -191,4 +253,41 @@ defmodule Algora.PaymentsTest do
       refute_enqueued(worker: ExecutePendingTransfer, args: %{credit_id: credit.id})
     end
   end
+
+  describe "execute_transfer/2" do
+    test "executes transfer idempotently with same transfer ID and transaction ID", %{account: account} do
+      credit_tx =
+        insert(:transaction,
+          type: :credit,
+          status: :succeeded,
+          net_amount: Money.new(1, :USD),
+          group_id: Nanoid.generate()
+        )
+
+      {:ok, transfer_tx} = Payments.initialize_transfer(credit_tx)
+
+      {:ok, transfer1} = Payments.execute_transfer(transfer_tx, account)
+      {:ok, transfer_tx1} = Repo.fetch_by(Transaction, type: :transfer)
+
+      assert Money.equal?(transfer_tx1.net_amount, Money.new(1, :USD))
+      assert transfer_tx1.status == :succeeded
+      assert transfer_tx1.succeeded_at != nil
+      assert transfer_tx1.group_id == credit_tx.group_id
+      assert transfer_tx1.user_id == credit_tx.user_id
+      assert transfer_tx1.provider_id == transfer1.id
+
+      {:ok, transfer2} = Payments.execute_transfer(transfer_tx, account)
+      {:ok, transfer_tx2} = Repo.fetch_by(Transaction, type: :transfer)
+
+      assert Money.equal?(transfer_tx2.net_amount, Money.new(1, :USD))
+      assert transfer_tx2.status == :succeeded
+      assert transfer_tx2.succeeded_at != nil
+      assert transfer_tx2.group_id == credit_tx.group_id
+      assert transfer_tx2.user_id == credit_tx.user_id
+      assert transfer_tx2.provider_id == transfer2.id
+
+      assert transfer1.id == transfer2.id
+      assert transfer_tx1.id == transfer_tx2.id
+    end
+  end
 end

test/support/stripe_mock.ex

@@ -31,7 +31,7 @@ defmodule Algora.Support.StripeMock do
   defmodule Transfer do
     @moduledoc false
 
-    def create(%{destination: "acct_invalid"}) do
+    def create(%{destination: "acct_invalid"}, _opts) do
       {:error,
        %Stripe.Error{
          source: :stripe,
@@ -40,10 +40,10 @@ defmodule Algora.Support.StripeMock do
        }}
     end
 
-    def create(%{amount: amount, currency: currency, destination: destination}) do
+    def create(%{amount: amount, currency: currency, destination: destination}, opts) do
       {:ok,
        %Stripe.Transfer{
-         id: "tr_#{Algora.Util.random_int()}",
+         id: "tr_#{:erlang.phash2(opts[:idempotency_key])}",
          amount: amount,
          currency: currency,
          destination: destination