Store events to avoid duplication. Move event handling to async tasks.

lib/code_corps.ex

@@ -18,7 +18,7 @@ defmodule CodeCorps do
       supervisor(CodeCorps.Endpoint, []),
       # Start your own worker by calling: CodeCorps.Worker.start_link(arg1, arg2, arg3)
       # worker(CodeCorps.Worker, [arg1, arg2, arg3]),
-
+      supervisor(Task.Supervisor, [[name: :webhook_processor, restart: :transient]]),
       worker(Segment, [Application.get_env(:segment, :write_key)])
     ]
 

lib/code_corps/stripe_service/stripe_connect_customer.ex

@@ -19,7 +19,7 @@ defmodule CodeCorps.StripeService.StripeConnectCustomerService do
     end
   end
 
-  def update(%StripeConnectCustomer{id_from_stripe: id_from_stripe, stripe_connect_account: connect_account} = connect_customer, attributes) do
+  def update(%StripeConnectCustomer{id_from_stripe: id_from_stripe, stripe_connect_account: connect_account}, attributes) do
     @api.Customer.update(id_from_stripe, attributes, connect_account: connect_account.id_from_stripe)
   end
 

lib/code_corps/stripe_service/stripe_platform_card.ex

@@ -81,7 +81,7 @@ defmodule CodeCorps.StripeService.StripePlatformCardService do
     stripe_platform_card |> Map.take([:exp_month, :exp_year, :name])
   end
 
-  defp do_update_connect_cards(stripe_platform_card, attributes) when attributes == %{}, do: []
+  defp do_update_connect_cards(_stripe_platform_card, attributes) when attributes == %{}, do: []
   defp do_update_connect_cards(stripe_platform_card, attributes) do
     stripe_platform_card
     |> Repo.preload([stripe_connect_cards: [:stripe_connect_account, :stripe_platform_card]])

lib/code_corps/stripe_service/stripe_platform_customer.ex

@@ -56,7 +56,7 @@ defmodule CodeCorps.StripeService.StripePlatformCustomerService do
   - `{:error, :unhandled}` -if something unexpected went wrong
   """
   def update_from_stripe(id_from_stripe) do
-    with customer                                            <- Repo.get_by(StripePlatformCustomer, id_from_stripe: id_from_stripe),
+    with %StripePlatformCustomer{} = customer                <- Repo.get_by(StripePlatformCustomer, id_from_stripe: id_from_stripe),
          {:ok, %Stripe.Customer{} = stripe_customer}         <- @api.Customer.retrieve(id_from_stripe),
          {:ok, params}                                       <- StripePlatformCustomerAdapter.to_params(stripe_customer, %{}),
          {:ok, %StripePlatformCustomer{} = platform_customer, connect_customer_updates} <- perform_update(customer, params)

lib/code_corps/stripe_service/webhook_processing/connect_event_handler.ex

@@ -0,0 +1,22 @@
+defmodule CodeCorps.StripeService.WebhookProcessing.ConnectEventHandler do
+  @moduledoc """
+  In charge of handling Stripe Connect webhooks
+  """
+
+  alias CodeCorps.StripeService.Events
+
+  @doc """
+  Handles Stripe Connect webhooks
+
+  ## Returns
+  * The result of calling the specific handlers `handle/1` function. This result ought ot be a tupple,
+    in which the first member is `:ok`, followed by one or more other elements, usually modified records.
+  * `{:ok, :unhandled_event}` if the specific event is not supported yet or at all
+  """
+  def handle_event(%{"type" => type} = attributes), do: do_handle(type, attributes)
+
+  defp do_handle("account.updated", attributes), do: Events.AccountUpdated.handle(attributes)
+  defp do_handle("customer.subscription.deleted", attributes), do: Events.CustomerSubscriptionDeleted.handle(attributes)
+  defp do_handle("customer.subscription.updated", attributes), do: Events.CustomerSubscriptionUpdated.handle(attributes)
+  defp do_handle(_, _), do: {:ok, :unhandled_event}
+end

lib/code_corps/stripe_service/webhook_processing/platform_event_handler.ex

@@ -0,0 +1,21 @@
+defmodule CodeCorps.StripeService.WebhookProcessing.PlatformEventHandler do
+  @moduledoc """
+  In charge of handling Stripe Platform webhooks
+  """
+
+  alias CodeCorps.StripeService.Events
+
+  @doc """
+  Handles Stripe Platform webhooks
+
+  ## Returns
+  * The result of calling the specific handlers `handle/1` function. This result ought ot be a tupple,
+    in which the first member is `:ok`, followed by one or more other elements, usually modified records.
+  * `{:ok, :unhandled_event}` if the specific event is not supported yet or at all
+  """
+  def handle_event(%{"type" => type} = attributes), do: do_handle(type, attributes)
+
+  defp do_handle("customer.updated", attributes), do: Events.CustomerUpdated.handle(attributes)
+  defp do_handle("customer.source.updated", attributes), do: Events.CustomerSourceUpdated.handle(attributes)
+  defp do_handle(_, _), do: {:ok, :unhandled_event}
+end

lib/code_corps/stripe_service/webhook_processing/webhook_processor.ex

@@ -0,0 +1,84 @@
+defmodule CodeCorps.StripeService.WebhookProcessing.WebhookProcessor do
+  @moduledoc """
+  Used to process a Stripe webhook request.
+  """
+
+  alias CodeCorps.StripeEvent
+  alias CodeCorps.Repo
+
+  @doc """
+  Used to process a Stripe webhook event.
+
+  Receives the event json as the first parameter.
+  Since a webhook can be a platform or a connect webhook,
+  the function requires the handler module as the second parameter.
+
+  ## Returns
+
+  * `{:ok, :ignored_by_environment}` if the event was ignored due to environment mismatch
+  * `{:ok, :enqueued}` if the event will be handled
+
+  ## Note
+
+  Stripe events can have their `livemode` property set to `true` or `false`.
+  A livemode event should be handled by the production environment, while all other environments
+  handle non-livemode events.
+  """
+  def process_async(%{} = json, handler) do
+    case event_matches_environment?(json) do
+      true -> do_process_async(json, handler)
+      false -> {:ok, :ignored_by_environment}
+    end
+  end
+
+  defp do_process_async(json, handler) do
+    Task.Supervisor.start_child(:webhook_processor, fn -> do_process(json, handler) end)
+  end
+
+  defp event_matches_environment?(%{"livemode" => livemode}) do
+    case Application.get_env(:code_corps, :stripe_env) do
+      :prod -> livemode
+      _ -> !livemode
+    end
+  end
+
+  defp do_process(%{"id" => event_id, "type" => event_type} = json, handler) do
+    with {:ok, %StripeEvent{} = event} <- find_or_create_event(event_id, event_type) do
+      case handler.handle_event(json) |> Tuple.to_list do
+        [:ok, :unhandled_event] -> event |> set_unhandled
+        [:ok | _results]        -> event |> set_processed
+        [:error | _error]       -> event |> set_errored
+      end
+    else
+      {:error, :already_processing} -> nil
+    end
+  end
+
+  defp find_or_create_event(id_from_stripe, type) do
+    case find_event(id_from_stripe) do
+      %StripeEvent{status: "processing"} -> {:error, :already_processing}
+      %StripeEvent{} = event -> {:ok, event}
+      nil -> create_event(id_from_stripe, type)
+    end
+  end
+
+  defp find_event(id_from_stripe) do
+    Repo.get_by(StripeEvent, id_from_stripe: id_from_stripe)
+  end
+
+  defp create_event(id_from_stripe, type) do
+    %StripeEvent{} |> StripeEvent.create_changeset(%{id_from_stripe: id_from_stripe, type: type}) |> Repo.insert
+  end
+
+  defp set_processed(%StripeEvent{} = event) do
+    event |> StripeEvent.update_changeset(%{status: "processed"}) |> Repo.update
+  end
+
+  defp set_errored(%StripeEvent{} = event) do
+    event |> StripeEvent.update_changeset(%{status: "errored"}) |> Repo.update
+  end
+
+  defp set_unhandled(%StripeEvent{} = event) do
+    event |> StripeEvent.update_changeset(%{status: "unhandled"}) |> Repo.update
+  end
+end

priv/repo/migrations/20161207112519_add_stripe_events.exs

@@ -0,0 +1,15 @@
+defmodule CodeCorps.Repo.Migrations.AddStripeEvents do
+  use Ecto.Migration
+
+  def change do
+    create table(:stripe_events) do
+      add :id_from_stripe, :string, null: false
+      add :status, :string, default: "unprocessed"
+      add :type, :string, null: false
+
+      timestamps()
+    end
+
+    create unique_index(:stripe_events, [:id_from_stripe])
+  end
+end

test/controllers/stripe_connect_events_controller_test.exs

@@ -1,8 +1,7 @@
 defmodule CodeCorps.StripeConnectEventsControllerTest do
   use CodeCorps.ConnCase
 
-  alias CodeCorps.Project
-  alias CodeCorps.StripeConnectAccount
+  alias CodeCorps.{Project, StripeConnectAccount, StripeEvent}
 
   setup do
     conn =
@@ -40,8 +39,23 @@ defmodule CodeCorps.StripeConnectEventsControllerTest do
     }
   end
 
+  defp wait_for_supervisor, do: wait_for_children(:webhook_processor)
+
+  # used to have the test wait for or the children of a supervisor to exit
+
+  defp wait_for_children(supervisor_ref) do
+    Task.Supervisor.children(supervisor_ref)
+    |> Enum.each(&wait_for_child/1)
+  end
+
+  defp wait_for_child(pid) do
+    # Wait until the pid is dead
+    ref = Process.monitor(pid)
+    assert_receive {:DOWN, ^ref, _, _, _}
+  end
+
   describe "account.updated" do
-    test "returns 200 and updates account when one matches", %{conn: conn} do
+    test "updates account when one matches", %{conn: conn} do
       event = event_for(@account, "account.updated")
       stripe_id =  @account["id"]
 
@@ -53,20 +67,15 @@ defmodule CodeCorps.StripeConnectEventsControllerTest do
       path = stripe_connect_events_path(conn, :create)
       assert conn |> post(path, event) |> response(200)
 
+      wait_for_supervisor
+
       updated_account = Repo.get_by(StripeConnectAccount, id_from_stripe: stripe_id)
       assert updated_account.transfers_enabled
     end
-
-    test "returns 400 when doesn't match an existing account", %{conn: conn} do
-      event = event_for(@account, "account.updated")
-
-      path = stripe_connect_events_path(conn, :create)
-      assert conn |> post(path, event) |> response(400)
-    end
   end
 
   describe "customer.subscription.updated" do
-    test "returns 200 and updates subscription when one matches", %{conn: conn} do
+    test "updates subscription when one matches", %{conn: conn} do
       event = event_for(@subscription, "customer.subscription.updated")
       stripe_id =  @subscription["id"]
       connect_customer_id = @subscription["customer"]
@@ -89,13 +98,15 @@ defmodule CodeCorps.StripeConnectEventsControllerTest do
       path = stripe_connect_events_path(conn, :create)
       assert conn |> post(path, event) |> response(200)
 
+      wait_for_supervisor
+
       updated_project = Repo.get_by(Project, id: project.id)
       assert updated_project.total_monthly_donated == 0
     end
   end
 
   describe "customer.subscription.deleted" do
-    test "returns 200 and sets subscription to inactive when one matches", %{conn: conn} do
+    test "sets subscription to inactive when one matches", %{conn: conn} do
       event = event_for(@subscription, "customer.subscription.deleted")
       stripe_id =  @subscription["id"]
       connect_customer_id = @subscription["customer"]
@@ -118,16 +129,121 @@ defmodule CodeCorps.StripeConnectEventsControllerTest do
       path = stripe_connect_events_path(conn, :create)
       assert conn |> post(path, event) |> response(200)
 
+      wait_for_supervisor
+
       updated_project = Repo.get_by(Project, id: project.id)
       assert updated_project.total_monthly_donated == 0
     end
   end
 
-  describe "any other event" do
-    test "returns 200", %{conn: conn} do
-      event = event_for(%{}, "any.other")
+  describe "any event" do
+    test "returns 400, does nothing if event is livemode and env is not :prod", %{conn: conn} do
+      Application.put_env(:code_corps, :stripe_env, :other)
+
+      event = %{"id" => "evt_123", "livemode" => true, "type" => "any.event"}
+
+      path = conn |> stripe_connect_events_path(:create)
+      assert conn |> post(path, event) |> response(400)
+
+      wait_for_supervisor
+
+      assert StripeEvent |> Repo.aggregate(:count, :id) == 0
+
+      # put env back to original state
+      Application.put_env(:code_corps, :stripe_env, :test)
+    end
+
+    test "returns 400, does nothing if event is not livemode and env is :prod", %{conn: conn} do
+      Application.put_env(:code_corps, :stripe_env, :prod)
+
+      event = %{"id" => "evt_123", "livemode" => false, "type" => "any.event"}
+
+      path = conn |> stripe_connect_events_path(:create)
+      assert conn |> post(path, event) |> response(400)
+
+      wait_for_supervisor
+
+      assert StripeEvent |> Repo.aggregate(:count, :id) == 0
+
+      # put env back to original state
+      Application.put_env(:code_corps, :stripe_env, :test)
+    end
+
+    test "creates event if id is new", %{conn: conn} do
+      event = %{"id" => "evt_123", "livemode" => false, "type" => "any.event"}
+
+      path = conn |> stripe_connect_events_path(:create)
+      assert conn |> post(path, event) |> response(200)
+
+      wait_for_supervisor
+
+      assert StripeEvent |> Repo.aggregate(:count, :id) == 1
+    end
+
+    test "uses existing event if id exists", %{conn: conn} do
+      insert(:stripe_event, id_from_stripe: "evt_123")
+
+      event = %{"id" => "evt_123", "livemode" => false, "type" => "any.event"}
+
       path = conn |> stripe_connect_events_path(:create)
       assert conn |> post(path, event) |> response(200)
+
+      wait_for_supervisor
+
+      assert StripeEvent |> Repo.aggregate(:count, :id) == 1
+    end
+
+    test "sets event as unhandled if event is not handled", %{conn: conn} do
+      event = %{"id" => "evt_123", "livemode" => false, "type" => "any.event"}
+
+      path = conn |> stripe_connect_events_path(:create)
+      assert conn |> post(path, event) |> response(200)
+
+      wait_for_supervisor
+
+      record = StripeEvent |> Repo.one
+      assert record.status == "unhandled"
+    end
+
+    test "errors out event if handling fails", %{conn: conn} do
+      # we build the event, but do not make the account, causing it to error out
+      event = event_for(@account, "account.updated")
+
+      path = conn |> stripe_connect_events_path(:create)
+      assert conn |> post(path, event) |> response(200)
+
+      wait_for_supervisor
+
+      record = StripeEvent |> Repo.one
+      assert record.status == "errored"
+    end
+
+    test "marks event as processed if handling is done", %{conn: conn} do
+      # we build the event AND create the account, so it should process correctly
+      event = event_for(@account, "account.updated")
+      insert(:stripe_connect_account, id_from_stripe: @account["id"])
+
+      path = conn |> stripe_connect_events_path(:create)
+      assert conn |> post(path, event) |> response(200)
+
+      wait_for_supervisor
+
+      record = StripeEvent |> Repo.one
+      assert record.status == "processed"
+    end
+
+    test "leaves event alone if already processing", %{conn: conn} do
+      insert(:stripe_event, id_from_stripe: "evt_123", status: "processing")
+
+      event = %{"id" => "evt_123", "livemode" => false, "type" => "any.event"}
+
+      path = conn |> stripe_platform_events_path(:create)
+      assert conn |> post(path, event) |> response(200)
+
+      wait_for_supervisor
+
+      record = StripeEvent |> Repo.one
+      assert record.status == "processing"
     end
   end
 end