Show failing messages (#127)

* Modify message controller

* Create get_failing_messages method

* Show messages error in template

* Remove a lot of warnings

* Show messages in menu

* Change keys of logging

* Improve menu name

* Create pagination

* Add remaining tests

* Refactor

* Render buttons in templates

Co-authored-by: Emilio Carrión <[email protected]>

lib/postoffice/cache.ex

@@ -57,7 +57,7 @@ defmodule Postoffice.Cache do
   end
 
 
-  defp add_publishers(disabled_publishers_ids, state) when disabled_publishers_ids == [], do: :ok
+  defp add_publishers(disabled_publishers_ids, _state) when disabled_publishers_ids == [], do: :ok
 
   defp add_publishers(disabled_publishers_ids, state) do
     ids_tuples = Enum.map(disabled_publishers_ids, fn id -> {id, state} end)

lib/postoffice/messaging.ex

@@ -321,4 +321,10 @@ defmodule Postoffice.Messaging do
     from(p in Publisher, where: p.deleted == true)
     |> Repo.all()
   end
+
+  def get_failing_messages(%{"page"=> _page, "page_size"=> _page_size} = params) do
+    from(job in Oban.Job, where: job.state=="retryable")
+    |> Repo.paginate(params)
+    |> Map.from_struct
+  end
 end

lib/postoffice/repo.ex

@@ -2,4 +2,5 @@ defmodule Postoffice.Repo do
   use Ecto.Repo,
     otp_app: :postoffice,
     adapter: Ecto.Adapters.Postgres
+  use Scrivener
 end

lib/postoffice/workers/http.ex

@@ -40,7 +40,7 @@ defmodule Postoffice.Workers.Http do
       {:ok, %HTTPoison.Response{status_code: status_code, body: _body}}
       when status_code in 200..299 ->
         Logger.info("Succesfully sent http message",
-          message_id: id,
+          postoffice_message_id: id,
           target: target
         )
 
@@ -60,7 +60,7 @@ defmodule Postoffice.Workers.Http do
             response.status_code
           }"
 
-        Logger.info(error_reason, message_id: id)
+        Logger.info(error_reason, postoffice_message_id: id)
 
         {:ok, _data} =
           HistoricalData.create_failed_messages(%{
@@ -75,7 +75,7 @@ defmodule Postoffice.Workers.Http do
 
       {:error, %HTTPoison.Error{reason: reason}} ->
         error_reason = "Error trying to process message from HttpConsumer: #{reason}"
-        Logger.info(error_reason, message_id: id)
+        Logger.info(error_reason, postoffice_message_id: id)
 
         {:ok, _data} =
           HistoricalData.create_failed_messages(%{

lib/postoffice/workers/pubsub.ex

@@ -52,7 +52,7 @@ defmodule Postoffice.Workers.Pubsub do
 
       {:error, error} ->
         error_reason = "Error trying to process message from PubsubConsumer: #{error}"
-        Logger.info(error_reason, message_id: id)
+        Logger.info(error_reason, postoffice_message_id: id)
 
         {:ok, _data} =
           HistoricalData.create_failed_messages(%{

lib/postoffice_web/controllers/message_controller.ex

@@ -4,17 +4,19 @@ defmodule PostofficeWeb.MessageController do
   alias Postoffice.Messaging
   alias Postoffice.HistoricalData
 
-  def index(conn, %{"id" => id}) do
-    case Messaging.get_message!(id) do
-      nil ->
-        conn
-        |> put_flash(:info, "Message not found")
-        |> redirect(to: Routes.dashboard_path(conn, :index))
+  def index(conn, %{"page" => page, "page_size" => page_size} = params) do
+    messages = Messaging.get_failing_messages(params)
 
-      message ->
-        conn
-        |> redirect(to: Routes.message_path(conn, :show, message.id))
-    end
+    render(conn, "index.html",
+      page_name: "Messages",
+      messages: messages.entries,
+      page_number: messages.page_number,
+      total_pages: messages.total_pages
+    )
+  end
+
+  def index(conn, %{}) do
+    index(conn, %{"page" => 1, "page_size" => 100})
   end
 
   def show(conn, %{"id" => id}) do

lib/postoffice_web/router.ex

@@ -33,7 +33,7 @@ defmodule PostofficeWeb.Router do
 
     resources "/publishers", PublisherController, only: [:index, :new, :create, :edit, :update, :delete]
 
-    resources "/messages", MessageController, only: [:index, :show]
+    resources "/messages", MessageController, only: [:index]
 
     live_dashboard "/dashboard"
   end

lib/postoffice_web/templates/layout/app.html.eex

@@ -39,6 +39,12 @@
               <p>Publishers</p>
             </a>
           </li>
+          <li class="nav-item <%= if @page_name == "Messages" do %>active <% end %>">
+            <a class="nav-link" href="<%= Routes.message_path(@conn, :index) %>">
+              <i class="material-icons">mail_outline</i>
+              <p>Failing Messages</p>
+            </a>
+          </li>
         </ul>
       </div>
     </div>

lib/postoffice_web/templates/message/index.html.eex

@@ -0,0 +1,45 @@
+<div class="row">
+  <div class="col-md-12">
+    <%= if @page_number >= 10 do%>
+      <span>
+        <%= link "Prev Page",
+        class: "btn btn-primary btn-round",
+        to: Routes.message_path(@conn, :index, page: @page_number - 1, page_size: 100) %>
+      </span>
+    <% end %>
+    <%= if @page_number < @total_pages do%>
+      <span>
+        <%= link "Next Page",
+        class: "btn btn-primary btn-round",
+        to: Routes.message_path(@conn, :index, page: @page_number + 1, page_size: 100) %>
+      </span>
+    <% end %>
+    <span>
+      <p class="font-weight-bold">Pages: <%= @page_number %>/<%= @total_pages %></p>
+    </span>
+    <table class="table">
+      <thead>
+        <tr>
+          <th scope="col">#</th>
+          <th scope="col">Attemps</th>
+          <th scope="col">Max attempts</th>
+          <th scope="col">Last error</th>
+          <th scope="col">Scheduled at</th>
+          <th scope="col">Args</th>
+        </tr>
+      </thead>
+      <tbody>
+        <%= for message <- @messages do %>
+        <tr>
+          <td><%= message.id %></td>
+          <td><%= message.attempt %></td>
+          <td><%= message.max_attempts %></td>
+          <td><%= message.attempted_at%></td>
+          <td><%= message.scheduled_at%></td>
+          <td><%= Poison.encode!(message.args)%></td>
+        </tr>
+        <% end %>
+      </tbody>
+    </table>
+  </div>
+</div>

mix.exs

@@ -80,7 +80,8 @@ defmodule Postoffice.MixProject do
       {:cachex, "~> 3.3"},
       {:number, "~> 1.0.1"},
       {:oban, "2.5.0"},
-      {:prom_ex, "~> 1.0.0"}
+      {:prom_ex, "~> 1.0.0"},
+      {:scrivener_ecto, "~> 2.0"}
     ]
   end
 

mix.lock

@@ -61,6 +61,8 @@
   "postgrex": {:hex, :postgrex, "0.15.8", "f5e782bbe5e8fa178d5e3cd1999c857dc48eda95f0a4d7f7bd92a50e84a0d491", [:mix], [{:connection, "~> 1.0", [hex: :connection, repo: "hexpm", optional: false]}, {:db_connection, "~> 2.1", [hex: :db_connection, repo: "hexpm", optional: false]}, {:decimal, "~> 1.5 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}], "hexpm", "698fbfacea34c4cf22c8281abeb5cf68d99628d541874f085520ab3b53d356fe"},
   "prom_ex": {:hex, :prom_ex, "1.0.0", "512b438e07b499c6cf2db4d68b9f3ad0959aea3314420484bff268f0d402eafb", [:mix], [{:ecto, ">= 3.5.0", [hex: :ecto, repo: "hexpm", optional: true]}, {:finch, "~> 0.5.2", [hex: :finch, repo: "hexpm", optional: false]}, {:jason, "~> 1.2", [hex: :jason, repo: "hexpm", optional: false]}, {:oban, ">= 2.4.0", [hex: :oban, repo: "hexpm", optional: true]}, {:phoenix, ">= 1.5.0", [hex: :phoenix, repo: "hexpm", optional: true]}, {:phoenix_live_view, ">= 0.14.0", [hex: :phoenix_live_view, repo: "hexpm", optional: true]}, {:plug, ">= 1.10.0", [hex: :plug, repo: "hexpm", optional: true]}, {:plug_cowboy, "~> 2.1", [hex: :plug_cowboy, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.2", [hex: :telemetry, repo: "hexpm", optional: false]}, {:telemetry_metrics, "~> 0.6.0", [hex: :telemetry_metrics, repo: "hexpm", optional: false]}, {:telemetry_poller, "~> 0.5.1", [hex: :telemetry_poller, repo: "hexpm", optional: false]}], "hexpm", "a4dcebac0f92032cc80b27fb19e5dc3cb05fb1197a7840cb2e3ba553373ba418"},
   "ranch": {:hex, :ranch, "1.7.1", "6b1fab51b49196860b733a49c07604465a47bdb78aa10c1c16a3d199f7f8c881", [:rebar3], [], "hexpm", "451d8527787df716d99dc36162fca05934915db0b6141bbdac2ea8d3c7afc7d7"},
+  "scrivener": {:hex, :scrivener, "2.7.0", "fa94cdea21fad0649921d8066b1833d18d296217bfdf4a5389a2f45ee857b773", [:mix], [], "hexpm", "30da36a427f2519cf75993271fb7c5aad1759682a70f90d880a85c3d743d2c57"},
+  "scrivener_ecto": {:hex, :scrivener_ecto, "2.7.0", "cf64b8cb8a96cd131cdbcecf64e7fd395e21aaa1cb0236c42a7c2e34b0dca580", [:mix], [{:ecto, "~> 3.3", [hex: :ecto, repo: "hexpm", optional: false]}, {:scrivener, "~> 2.4", [hex: :scrivener, repo: "hexpm", optional: false]}], "hexpm", "e809f171687806b0031129034352f5ae44849720c48dd839200adeaf0ac3e260"},
   "sleeplocks": {:hex, :sleeplocks, "1.1.1", "3d462a0639a6ef36cc75d6038b7393ae537ab394641beb59830a1b8271faeed3", [:rebar3], [], "hexpm", "84ee37aeff4d0d92b290fff986d6a95ac5eedf9b383fadfd1d88e9b84a1c02e1"},
   "ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.6", "cf344f5692c82d2cd7554f5ec8fd961548d4fd09e7d22f5b62482e5aeaebd4b0", [:make, :mix, :rebar3], [], "hexpm", "bdb0d2471f453c88ff3908e7686f86f9be327d065cc1ec16fa4540197ea04680"},
   "swarm": {:hex, :swarm, "3.4.0", "64f8b30055d74640d2186c66354b33b999438692a91be275bb89cdc7e401f448", [:mix], [{:gen_state_machine, "~> 2.0", [hex: :gen_state_machine, repo: "hexpm", optional: false]}, {:libring, "~> 1.0", [hex: :libring, repo: "hexpm", optional: false]}], "hexpm", "94884f84783fc1ba027aba8fe8a7dae4aad78c98e9f9c76667ec3471585c08c6"},

test/postoffice/http_worker_test.exs

@@ -35,7 +35,7 @@ defmodule Postoffice.HttpWorkerTest do
         {:ok, %HTTPoison.Response{status_code: 201}}
       end)
 
-      assert {:ok, sent} = perform_job(HttpWorker, args)
+      assert {:ok, _sent} = perform_job(HttpWorker, args)
     end
 
     test "historical data is created when message is sent" do
@@ -147,7 +147,7 @@ defmodule Postoffice.HttpWorkerTest do
       {:ok, %HTTPoison.Response{status_code: 201}}
     end)
 
-    assert {:ok, sent} = perform_job(HttpWorker, args, attempt: 100)
+    assert {:ok, _sent} = perform_job(HttpWorker, args, attempt: 100)
   end
 
   test "not sent message when worker is deleted" do

test/postoffice/messaging_test.exs

@@ -154,7 +154,7 @@ defmodule Postoffice.MessagingTest do
     test "list_no_deleted_publishers/0 returns all existing no deleted publishers" do
       topic = Fixtures.create_topic()
       publisher = Fixtures.create_publisher(topic)
-      disabled_publisher = Fixtures.create_publisher(topic, @deleted_publisher_attrs)
+      Fixtures.create_publisher(topic, @deleted_publisher_attrs)
 
       publishers = Messaging.list_no_deleted_publishers()
       assert length(publishers) == 1
@@ -230,5 +230,43 @@ defmodule Postoffice.MessagingTest do
     test "count_failing_jobs/0 returns 0 if no retryable job exists" do
       assert Messaging.count_failing_jobs() == 0
     end
+
+    test "count_failing_jobs/0 returns failing job existents" do
+      Fixtures.create_failing_message(%{id: 1, user_id: 2})
+      Fixtures.create_failing_message(%{id: 2, user_id: 3})
+
+      assert Messaging.count_failing_jobs() == 2
+    end
+
+    test "count_failing_jobs/0 no returns retryable jobs when no exists" do
+      failing_messages = %{"page"=> 1, "page_size"=> 4}
+      |> Messaging.get_failing_messages
+
+      assert failing_messages == %{entries: [], page_number: 1, page_size: 4, total_entries: 0, total_pages: 1}
+    end
+
+    test "get_failing_messages/1 returns retryable jobs" do
+      first_failing_job = Fixtures.create_failing_message(%{id: 1, user_id: 2})
+      second_failing_job = Fixtures.create_failing_message(%{id: 2, user_id: 3})
+
+      failing_messages = %{"page"=> 1, "page_size"=> 4}
+      |> Messaging.get_failing_messages
+
+      assert failing_messages ==  %{entries: [first_failing_job, second_failing_job], page_number: 1, page_size: 4, total_entries: 2, total_pages: 1}
+    end
+
+    test "get_failing_messages/1 returns retryable jobs paginating" do
+      first_failing_job = Fixtures.create_failing_message(%{id: 1, user_id: 2})
+      second_failing_job = Fixtures.create_failing_message(%{id: 2, user_id: 3})
+
+      failing_messages = %{"page"=> 1, "page_size"=> 1}
+      |> Messaging.get_failing_messages
+      assert failing_messages ==  %{entries: [first_failing_job], page_number: 1, page_size: 1, total_entries: 2, total_pages: 2}
+
+      failing_messages = %{"page"=> 2, "page_size"=> 1}
+      |> Messaging.get_failing_messages
+      assert failing_messages ==  %{entries: [second_failing_job], page_number: 2, page_size: 1, total_entries: 2, total_pages: 2}
+    end
+
   end
 end

test/postoffice/pubsub_worker_test.exs

@@ -28,7 +28,7 @@ defmodule Postoffice.PubsubWorkerTest do
         {:ok, %PublishResponse{}}
       end)
 
-      assert {:ok, sent} = perform_job(PubsubWorker, args)
+      assert {:ok, _sent} = perform_job(PubsubWorker, args)
     end
 
     test "historical data is created if message is successfully sent" do
@@ -65,7 +65,7 @@ defmodule Postoffice.PubsubWorkerTest do
         {:ok, %PublishResponse{}}
       end)
 
-      assert {:ok, sent} = perform_job(PubsubWorker, args)
+      assert {:ok, _sent} = perform_job(PubsubWorker, args)
     end
 
     test "message is not send if there is any error on the request" do
@@ -120,7 +120,7 @@ defmodule Postoffice.PubsubWorkerTest do
         {:ok, %PublishResponse{}}
       end)
 
-      assert {:ok, sent} = perform_job(PubsubWorker, args, attempt: 100)
+      assert {:ok, _sent} = perform_job(PubsubWorker, args, attempt: 100)
     end
   end
 end

test/postoffice_web/controllers/api/publisher_controller_test.exs

@@ -160,9 +160,9 @@ defmodule PostofficeWeb.Api.PublisherControllerTest do
         Fixtures.create_topic()
         |> Fixtures.create_publisher()
 
-      conn = delete(conn, Routes.api_publisher_path(conn, :delete, publisher))
+      delete(conn, Routes.api_publisher_path(conn, :delete, publisher))
 
-      assert_receive {:publisher_deleted, publisher}
+      assert_receive {:publisher_deleted, _publisher}
     end
 
     test "Returns 400 when can not delete publisher", %{conn: conn} do

test/postoffice_web/controllers/message_controller_test.exs

@@ -0,0 +1,28 @@
+defmodule PostofficeWeb.MessageControllerTest do
+  use PostofficeWeb.ConnCase, async: true
+
+  alias Postoffice.Fixtures
+
+  setup do
+    {:ok, conn: Phoenix.ConnTest.build_conn()}
+  end
+
+  describe "list failed message" do
+    test "can access to messages list", %{conn: conn} do
+      failing_job = Fixtures.create_failing_message(%{id: 1, user_id: 2})
+      second_failing_job = Fixtures.create_failing_message(%{id: 23, user_id: 2})
+
+      response = conn
+      |> get(Routes.message_path(conn, :index, %{page: 1, page_size: 1}))
+
+      assert html_response(response, 200) =~ to_string(failing_job.id)
+      refute html_response(response, 200) =~ to_string(second_failing_job.id)
+
+      response = conn
+      |> get(Routes.message_path(conn, :index, %{page: 2, page_size: 1}))
+
+      refute html_response(response, 200) =~ to_string(failing_job.id)
+      assert html_response(response, 200) =~ to_string(second_failing_job.id)
+    end
+  end
+end

test/postoffice_web/controllers/publisher_controller_test.exs

@@ -123,8 +123,8 @@ defmodule PostofficeWeb.PublisherControllerTest do
 
   describe "Delete publisher" do
     test "delete a publisher", %{conn: conn} do
+      Fixtures.create_topic(%{name: "test2", origin_host: "example2.com"})
       topic = Fixtures.create_topic()
-      second_topic = Fixtures.create_topic(%{name: "test2", origin_host: "example2.com"})
       publisher = Fixtures.create_publisher(topic)
 
       assert conn
@@ -142,9 +142,9 @@ defmodule PostofficeWeb.PublisherControllerTest do
         Fixtures.create_topic()
         |> Fixtures.create_publisher()
 
-      conn = delete(conn, Routes.publisher_path(conn, :delete, publisher))
+      delete(conn, Routes.publisher_path(conn, :delete, publisher))
 
-      assert_receive {:publisher_deleted, publisher}
+      assert_receive {:publisher_deleted, _publisher}
     end
 
   end

test/support/fixtures.ex

@@ -4,6 +4,8 @@ defmodule Postoffice.Fixtures do
   """
   alias Postoffice
   alias Postoffice.Messaging
+  alias Postoffice.Repo
+  import Ecto.Changeset
 
   @topic_attrs %{
     name: "test",
@@ -38,4 +40,18 @@ defmodule Postoffice.Fixtures do
     {:ok, publisher} = Messaging.create_publisher(Map.put(attrs, :topic_id, topic.id))
     publisher
   end
+
+  def create_failing_message(data) do
+    {:ok, oban} =
+      data
+      |> Oban.Job.new(queue: :http, worker: Postoffice.SomeFakeWorker)
+      |> Repo.insert()
+
+    {:ok, updated_job} =
+      Repo.get(Oban.Job, oban.id)
+      |> change(%{state: "retryable"})
+      |> Repo.update()
+
+    updated_job
+  end
 end