implement transfers

Author: zcesur

config/config.exs

@@ -34,6 +34,7 @@ config :algora, Oban,
     notify_bounty: 1,
     notify_tip_intent: 1,
     notify_claim: 1,
+    transfers: 1,
     activity_notifier: 1,
     activity_mailer: 1
   ]

lib/algora/payments/jobs/execute_pending_transfers.ex

@@ -0,0 +1,84 @@
+defmodule Algora.Payments.Jobs.ExecutePendingTransfers do
+  @moduledoc false
+  use Oban.Worker, queue: :transfers
+
+  import Ecto.Changeset
+  import Ecto.Query
+
+  alias Algora.MoneyUtils
+  alias Algora.Payments.Account
+  alias Algora.Payments.Transaction
+  alias Algora.Repo
+  alias Algora.Util
+
+  require Logger
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{args: %{user_id: user_id, group_id: group_id}}) do
+    total_credits =
+      Repo.one(
+        from(t in Transaction,
+          where: t.user_id == ^user_id,
+          where: t.type == :credit,
+          where: t.status == :succeeded,
+          select: sum(t.net_amount)
+        )
+      ) || Money.zero(:USD)
+
+    total_transfers =
+      Repo.one(
+        from(t in Transaction,
+          where: t.user_id == ^user_id,
+          where: t.type == :transfer,
+          where: t.status == :succeeded or t.status == :processing or t.status == :initialized,
+          select: sum(t.net_amount)
+        )
+      ) || Money.zero(:USD)
+
+    pending_amount = Money.sub!(total_credits, total_transfers)
+
+    with {:ok, account} <- Repo.fetch_by(Account, user_id: user_id, provider: :stripe, payouts_enabled: true),
+         true <- Money.positive?(pending_amount) do
+      {:ok, transaction} =
+        %Transaction{}
+        |> change(%{
+          provider: "stripe",
+          type: :credit,
+          status: :initialized,
+          user_id: user_id,
+          gross_amount: pending_amount,
+          net_amount: pending_amount,
+          total_fee: Money.zero(:USD),
+          group_id: group_id
+        })
+        |> Algora.Validations.validate_positive(:gross_amount)
+        |> Algora.Validations.validate_positive(:net_amount)
+        |> foreign_key_constraint(:user_id)
+        |> Repo.insert()
+
+      Repo.transact(fn ->
+        # TODO: set other params
+        # TODO: provide idempotency key
+        {:ok, transfer} =
+          Stripe.Transfer.create(%{
+            amount: MoneyUtils.to_minor_units(pending_amount),
+            currency: to_string(pending_amount.currency),
+            destination: account.stripe_account_id
+          })
+
+        {:ok, transaction} =
+          transaction
+          |> change(%{
+            status: if(transfer.status == :succeeded, do: :succeeded, else: :failed),
+            provider_id: transfer.id,
+            provider_meta: Util.normalize_struct(transfer)
+          })
+          |> Repo.update()
+
+        {:ok, transaction}
+      end)
+    else
+      _ -> :ok
+    end
+  end
+end

lib/algora/payments/jobs/execute_transfer.ex

@@ -4,47 +4,67 @@ defmodule AlgoraWeb.Webhooks.StripeController do
   import Ecto.Query
 
   alias Algora.Payments
-  alias Algora.Payments.Jobs.ExecuteTransfer
+  alias Algora.Payments.Jobs.ExecutePendingTransfers
   alias Algora.Payments.Transaction
   alias Algora.Repo
 
   require Logger
 
+  @metadata_version "2"
+
   @impl true
   def handle_event(%Stripe.Event{
         type: "charge.succeeded",
-        data: %{object: %{metadata: %{"version" => "2", "group_id" => group_id}}}
+        data: %{object: %{metadata: %{"version" => @metadata_version, "group_id" => group_id}}}
       })
       when is_binary(group_id) do
-    {:ok, count} =
-      Repo.transact(fn ->
-        {count, _} =
-          Repo.update_all(from(t in Transaction, where: t.group_id == ^group_id),
-            set: [status: :succeeded, succeeded_at: DateTime.utc_now()]
-          )
-
-        # TODO: get pending transfers (recipient with active payout accounts)
-        transfers = []
-
-        Enum.map(transfers, fn %{transfer_id: transfer_id, user_id: user_id} ->
-          %{transfer_id: transfer_id, user_id: user_id}
-          |> ExecuteTransfer.new()
-          |> Oban.insert()
+    Repo.transact(fn ->
+      update_result =
+        Repo.update_all(from(t in Transaction, where: t.group_id == ^group_id),
+          set: [status: :succeeded, succeeded_at: DateTime.utc_now()]
+        )
+
+      # TODO: split into two groups:
+      #     - has active payout account -> execute pending transfers
+      #     - has no active payout account -> notify user to connect payout account
+      jobs_result =
+        from(t in Transaction,
+          where: t.group_id == ^group_id,
+          where: t.type == :credit,
+          where: t.status == :succeeded
+        )
+        |> Repo.all()
+        |> Enum.map(fn %{user_id: user_id} -> user_id end)
+        |> Enum.uniq()
+        |> Enum.reduce_while(:ok, fn user_id, :ok ->
+          case %{user_id: user_id, group_id: group_id}
+               |> ExecutePendingTransfers.new()
+               |> Oban.insert() do
+            {:ok, _job} -> {:cont, :ok}
+            error -> {:halt, error}
+          end
         end)
 
-        {:ok, count}
-      end)
+      with {count, _} when count > 0 <- update_result,
+           :ok <- jobs_result do
+        Payments.broadcast()
+      else
+        {:error, reason} ->
+          Logger.error("Failed to update transactions: #{inspect(reason)}")
+          {:error, :failed_to_update_transactions}
 
-    if count == 0 do
-      {:error, :no_transactions_found}
-    else
-      Payments.broadcast()
-      {:ok, nil}
-    end
+        _error ->
+          Logger.error("Failed to update transactions")
+          {:error, :failed_to_update_transactions}
+      end
+    end)
   end
 
   @impl true
   def handle_event(%Stripe.Event{type: "transfer.created"} = event) do
+    # TODO: update transaction
+    # TODO: broadcast
+    # TODO: notify user
     Logger.info("Stripe #{event.type} event: #{event.id}")
   end