feat(statistic): basic background-job & cache workflow

bg-job not work in test, see samsondav/rihanna#75

Author: mydearxym

config/config.exs

@@ -75,6 +75,11 @@ config :groupher_server, GroupherServer.Mailer,
   adapter: Bamboo.MailgunAdapter,
   domain: "mailer.coderplanets.com"
 
+# handle background jobs
+config :rihanna,
+  jobs_table_name: "background_jobs",
+  producer_postgres_connection: {Ecto, GroupherServer.Repo}
+
 import_config "#{Mix.env()}.exs"
 
 if File.exists?("config/#{Mix.env()}.secret.exs") do

lib/groupher_server/application.ex

@@ -30,7 +30,8 @@ defmodule GroupherServer.Application do
               options: []
             )
         ]
-      ])
+      ]),
+      {Rihanna.Supervisor, [postgrex: GroupherServer.Repo.config()]}
     ]
 
     # See https://hexdocs.pm/elixir/Supervisor.html

lib/groupher_server/statistics/delegates/contribute.ex

@@ -1,29 +1,35 @@
 defmodule GroupherServer.Statistics.Delegate.Contribute do
+  @moduledoc """
+  contribute statistics for user and community, record how many content
+  has been add to it
+  """
   import Ecto.Query, warn: false
   import Helper.Utils
   import ShortMaps
 
-  alias GroupherServer.Repo
-  alias GroupherServer.Accounts.User
-  alias GroupherServer.CMS.Community
-  alias GroupherServer.Statistics.{UserContribute, CommunityContribute}
-  alias Helper.{Cache, ORM, QueryBuilder}
+  alias GroupherServer.{Accounts, CMS, Repo, Statistics}
+
+  alias Accounts.User
+  alias CMS.Community
+  alias Statistics.{CommunityContribute, UserContribute}
+
+  alias Helper.{Cache, Later, ORM, QueryBuilder}
 
   @community_contribute_days get_config(:general, :community_contribute_days)
   @user_contribute_months get_config(:general, :user_contribute_months)
 
   @doc """
   update user's contributes record
   """
-  def make_contribute(%User{id: id}) do
+  def make_contribute(%User{id: id} = user) do
     today = Timex.today() |> Date.to_iso8601()
 
     case ORM.find_by(UserContribute, user_id: id, date: today) do
       {:ok, contribute} ->
-        contribute |> inc_contribute_count(:user) |> done
+        update_contribute_record(contribute)
 
       {:error, _} ->
-        UserContribute |> ORM.create(%{user_id: id, date: today, count: 1})
+        insert_contribute_record(user)
     end
   end
 
@@ -35,20 +41,10 @@ defmodule GroupherServer.Statistics.Delegate.Contribute do
 
     case ORM.find_by(CommunityContribute, community_id: id, date: today) do
       {:ok, contribute} ->
-        result = contribute |> inc_contribute_count(:community) |> done
-
-        # TODO:  move to background job
-        list_contributes_digest(%Community{id: id})
-        result
+        update_contribute_record(contribute)
 
       {:error, _} ->
-        result =
-          CommunityContribute
-          |> ORM.create(%{community_id: id, date: today, count: 1})
-
-        # TODO:  move to background job
-        list_contributes_digest(%Community{id: id})
-        result
+        insert_contribute_record(%Community{id: id})
     end
   end
 
@@ -78,26 +74,52 @@ defmodule GroupherServer.Statistics.Delegate.Contribute do
         {:ok, result}
 
       {:error, _} ->
-        %Community{id: id}
-        |> get_contributes()
-        |> to_counts_digest(days: @community_contribute_days)
-        |> done
-        |> cache_result(scope)
+        get_contributes_then_cache(%Community{id: id})
     end
   end
 
-  # TODO:  mv to helper/cache, also 规范一下 scope
-  defp cache_result({:ok, result}, scope) do
-    Cache.put(scope, result)
-    {:ok, result}
+  # NOTE*  must be public, cause it will be exec by background job
+  def get_contributes_then_cache(%Community{id: id}) do
+    scope = Cache.get_scope(:community_contributes, id)
+
+    %Community{id: id}
+    |> do_get_contributes()
+    |> to_counts_digest(days: @community_contribute_days)
+    |> done_and_cache(scope)
+  end
+
+  defp update_contribute_record(%UserContribute{} = contribute) do
+    contribute |> inc_contribute_count(:user) |> done
+  end
+
+  defp insert_contribute_record(%User{id: id}) do
+    today = Timex.today() |> Date.to_iso8601()
+
+    UserContribute |> ORM.create(%{user_id: id, date: today, count: 1})
+  end
+
+  defp update_contribute_record(%CommunityContribute{community_id: community_id} = contribute) do
+    with {:ok, result} <- inc_contribute_count(contribute, :community) do
+      cache_contribute_later(%Community{id: community_id})
+      {:ok, result}
+    end
+  end
+
+  defp insert_contribute_record(%Community{id: id}) do
+    today = Timex.today() |> Date.to_iso8601()
+
+    with {:ok, result} <-
+           ORM.create(CommunityContribute, %{community_id: id, date: today, count: 1}) do
+      cache_contribute_later(%Community{id: id})
+      {:ok, result}
+    end
   end
 
-  # TODO:  mv to helper/cache, also 规范一下 scope
-  defp cache_result({:error, result}, _scope) do
-    {:error, result}
+  defp cache_contribute_later(%Community{id: id}) do
+    Later.exec({__MODULE__, :get_contributes_then_cache, [%Community{id: id}]})
   end
 
-  defp get_contributes(%Community{id: id}) do
+  defp do_get_contributes(%Community{id: id}) do
     community_id = integerfy(id)
 
     CommunityContribute

lib/helper/later.ex

@@ -0,0 +1,14 @@
+defmodule Helper.Later do
+  @moduledoc """
+  background jobs support, currently using https://github.com/samphilipd/rihanna
+  """
+
+  @doc """
+  ## Example
+  iex> Later.exec({__MODULE__, :get_contributes_then_cache, [%Community{id: id}]})
+  {:ok, _}
+  """
+  def exec({mod, func, args}) do
+    Rihanna.enqueue({mod, func, args})
+  end
+end

lib/helper/utils.ex

@@ -6,6 +6,8 @@ defmodule Helper.Utils do
   import Helper.ErrorHandler
   import Helper.ErrorCode
 
+  alias Helper.Cache
+
   def get_config(section, key, app \\ :groupher_server)
 
   def get_config(section, :all, app) do
@@ -50,6 +52,13 @@ defmodule Helper.Utils do
   # def done({:error, error}), do: {:error, error}
   def done(result), do: {:ok, result}
 
+  def done_and_cache(result, scope) do
+    with {:ok, res} <- done(result) do
+      Cache.put(scope, res)
+      {:ok, res}
+    end
+  end
+
   @doc """
   see: https://hexdocs.pm/absinthe/errors.html#content for error format
   """

mix.exs

@@ -97,7 +97,8 @@ defmodule GroupherServer.Mixfile do
       {:recase, "~> 0.4.0"},
       {:nanoid, "~> 2.0.0"},
       {:bamboo, "1.2.0"},
-      {:cachex, "3.1.3"}
+      {:cachex, "3.1.3"},
+      {:rihanna, "1.3.5"}
     ]
   end
 

mix.lock

@@ -14,8 +14,8 @@
   "connection": {:hex, :connection, "1.0.4", "a1cae72211f0eef17705aaededacac3eb30e6625b04a6117c1b2db6ace7d5976", [:mix], [], "hexpm"},
   "cors_plug": {:hex, :cors_plug, "1.5.0", "6311ea6ac9fb78b987df52a7654136626a7a0c3b77f83da265f952a24f2fc1b0", [], [{:plug, "~> 1.4", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm"},
   "corsica": {:hex, :corsica, "1.1.2", "5ad8b9dcbeeda4762d78a57c0c8c2f88e1eef8741508517c98cb79e0db1f107d", [:mix], [{:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm"},
-  "cowboy": {:hex, :cowboy, "2.6.3", "99aa50e94e685557cad82e704457336a453d4abcb77839ad22dbe71f311fcc06", [], [{:cowlib, "~> 2.7.3", [hex: :cowlib, repo: "hexpm", optional: false]}, {:ranch, "~> 1.7.1", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm"},
-  "cowlib": {:hex, :cowlib, "2.7.3", "a7ffcd0917e6d50b4d5fb28e9e2085a0ceb3c97dea310505f7460ff5ed764ce9", [], [], "hexpm"},
+  "cowboy": {:hex, :cowboy, "2.6.3", "99aa50e94e685557cad82e704457336a453d4abcb77839ad22dbe71f311fcc06", [:rebar3], [{:cowlib, "~> 2.7.3", [hex: :cowlib, repo: "hexpm", optional: false]}, {:ranch, "~> 1.7.1", [hex: :ranch, repo: "hexpm", optional: false]}], "hexpm"},
+  "cowlib": {:hex, :cowlib, "2.7.3", "a7ffcd0917e6d50b4d5fb28e9e2085a0ceb3c97dea310505f7460ff5ed764ce9", [:rebar3], [], "hexpm"},
   "credo": {:hex, :credo, "1.1.0", "e0c07b2fd7e2109495f582430a1bc96b2c71b7d94c59dfad120529f65f19872f", [:mix], [{:bunt, "~> 0.2.0", [hex: :bunt, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm"},
   "dataloader": {:hex, :dataloader, "1.0.6", "fb724d6d3fb6acb87d27e3b32dea3a307936ad2d245faf9cf5221d1323d6a4ba", [:mix], [{:ecto, ">= 0.0.0", [hex: :ecto, repo: "hexpm", optional: true]}], "hexpm"},
   "db_connection": {:hex, :db_connection, "2.1.0", "122e2f62c4906bf2e49554f1e64db5030c19229aa40935f33088e7d543aa79d0", [:mix], [{:connection, "~> 1.0.2", [hex: :connection, repo: "hexpm", optional: false]}], "hexpm"},
@@ -59,6 +59,7 @@
   "pre_commit": {:hex, :pre_commit, "0.3.4", "e2850f80be8090d50ad8019ef2426039307ff5dfbe70c736ad0d4d401facf304", [:mix], [], "hexpm"},
   "ranch": {:hex, :ranch, "1.7.1", "6b1fab51b49196860b733a49c07604465a47bdb78aa10c1c16a3d199f7f8c881", [:rebar3], [], "hexpm"},
   "recase": {:hex, :recase, "0.4.0", "8fb52846f75948156385af2dfdc12f69e5ce27b022a9d1682c70a2fb3ed149c7", [:mix], [], "hexpm"},
+  "rihanna": {:hex, :rihanna, "1.3.5", "5f5e6c5b1e514978a29a6791f338f4bb963401959fc212bd18d4a2c92d79a7a4", [:mix], [{:ecto, ">= 2.0.0", [hex: :ecto, repo: "hexpm", optional: true]}, {:ecto_sql, ">= 3.0.0", [hex: :ecto_sql, repo: "hexpm", optional: true]}, {:postgrex, ">= 0.13.3", [hex: :postgrex, repo: "hexpm", optional: false]}], "hexpm"},
   "scrivener": {:hex, :scrivener, "2.5.0", "e1f78c62b6806d91cc9c4778deef1ea4e80aa9fadfce2c16831afe0468cc8a2c", [:mix], [], "hexpm"},
   "scrivener_ecto": {:git, "https://github.com/mastani-stack/scrivener_ecto", "a6ebbbe9585e9829219ed0f73a3afcb4edde3558", [branch: "dev"]},
   "sentry": {:hex, :sentry, "6.4.1", "882287f1f3167dc4794865124977e2d88878d51d19930c0d0e7cc3a663a4a181", [:mix], [{:hackney, "~> 1.8 or 1.6.5", [hex: :hackney, repo: "hexpm", optional: false]}, {:phoenix, "~> 1.3", [hex: :phoenix, repo: "hexpm", optional: true]}, {:plug, "~> 1.6", [hex: :plug, repo: "hexpm", optional: true]}, {:poison, "~> 1.5 or ~> 2.0 or ~> 3.0", [hex: :poison, repo: "hexpm", optional: false]}], "hexpm"},

priv/repo/migrations/20190710090445_create_background_jobs.exs

@@ -0,0 +1,3 @@
+defmodule GroupherServer.Repo.Migrations.CreateBackgroundJobs do
+  use Rihanna.Migration
+end

test/groupher_server/statistics/statistics_test.exs

@@ -3,7 +3,7 @@ defmodule GroupherServer.Test.Statistics do
 
   import Helper.Utils, only: [get_config: 2]
 
-  alias Helper.ORM
+  alias Helper.{Cache, Later, ORM}
   alias GroupherServer.Accounts.User
   alias GroupherServer.{CMS, Repo, Statistics}
 
@@ -148,7 +148,6 @@ defmodule GroupherServer.Test.Statistics do
       assert length(contributes) == @community_contribute_days + 1
     end
 
-    alias Helper.Cache
     @tag :wip
     test "the contributes data should be cached after first query", ~m(community)a do
       scope = Cache.get_scope(:community_contributes, community.id)
@@ -159,14 +158,24 @@ defmodule GroupherServer.Test.Statistics do
       assert {:ok, contributes} = Cache.get(scope)
     end
 
-    @tag :wip
+    @tag :wip2
+    test "Rihanna should work in test sandbox", ~m(community)a do
+      res = Rihanna.enqueue({IO, :puts, ["Work, work, work, work, work."]})
+      Process.sleep(1000)
+      # IO.inspect(res, label: "res")
+    end
+
     test "cache should be update after make contributes", ~m(community)a do
       scope = Cache.get_scope(:community_contributes, community.id)
       assert {:error, nil} = Cache.get(scope)
 
       Statistics.make_contribute(%Community{id: community.id})
 
-      assert {:ok, _} = Cache.get(scope)
+      # res = Later.exec({IO, :puts, ["Work, work, work, work, work."]})
+      # Process.sleep(1000)
+      # IO.inspect(res, label: "res")
+
+      # assert {:ok, _} = Cache.get(scope)
     end
   end
 end