fix observer cron (#1351)

lib/console/deployments/cron.ex

@@ -293,6 +293,7 @@ defmodule Console.Deployments.Cron do
   def run_observers() do
     Observer.runnable()
     |> Observer.ordered(asc: :id)
+    |> Repo.stream(method: :keyset)
     |> Console.throttle()
     |> Flow.from_enumerable()
     |> Flow.map(&Console.Deployments.Observer.Runner.run/1)

lib/console/schema/app_notification.ex

@@ -2,6 +2,9 @@ defmodule Console.Schema.AppNotification do
   use Piazza.Ecto.Schema
   alias Console.Schema.{User}
 
+  @expiry [days: -7]
+  @too_old [days: -30]
+
   defenum Priority, low: 0, medium: 1, high: 2
 
   schema "app_notifications" do
@@ -27,9 +30,12 @@ defmodule Console.Schema.AppNotification do
   end
 
   def expired(query \\ __MODULE__) do
-    expiry = Timex.now() |> Timex.shift(days: -7)
+    expiry  = Timex.now() |> Timex.shift(@expiry)
+    too_old = Timex.now() |> Timex.shift(@too_old)
 
-    from(n in query, where: not is_nil(n.read_at) and n.read_at < ^expiry)
+    from(n in query,
+      where: (not is_nil(n.read_at) and n.read_at < ^expiry) or n.inserted_at < ^too_old
+    )
   end
 
   def changeset(model, attrs \\ %{}) do

test/console/deployments/cron_test.exs

@@ -5,6 +5,7 @@ defmodule Console.Deployments.CronTest do
   alias Kazan.Apis.Core.V1, as: Core
   alias Console.Deployments.{Cron, Clusters, Services}
 
+
   describe "#prune_services/0" do
     test "it will wipe stale drained services" do
       svcs = insert_list(3, :service, deleted_at: Timex.now())
@@ -266,4 +267,39 @@ defmodule Console.Deployments.CronTest do
       end
     end
   end
+
+  describe "#prune_notifications/0" do
+    test "it will wipe old read or really old unread notifications" do
+      read = insert_list(2, :app_notification, read_at: Timex.now() |> Timex.shift(days: -8))
+      unread = insert_list(2, :app_notification, inserted_at: Timex.now() |> Timex.shift(days: -40))
+      ignore = insert_list(3, :app_notification)
+
+      Cron.prune_notifications()
+
+      for n <- read ++ unread,
+        do: refute refetch(n)
+
+      for n <- ignore,
+        do: assert refetch(n)
+    end
+  end
+end
+
+defmodule Console.Deployments.AsyncCronTest do
+  use Console.DataCase, async: false
+  use Mimic
+  alias Console.Deployments.{Cron}
+
+  setup :set_mimic_global
+
+  describe "#run_observers/0" do
+    test "it can execute pending observers" do
+      insert_list(2, :observer, next_run_at: Timex.now() |> Timex.shift(minutes: -1))
+      insert(:observer, next_run_at: Timex.now() |> Timex.shift(minutes: 5))
+
+      expect(Console.Deployments.Observer.Runner, :run, 2, fn _ -> :ok end)
+
+      :ok = Cron.run_observers()
+    end
+  end
 end

test/test_helper.exs

@@ -34,6 +34,7 @@ Mimic.copy(Tentacat.App.Installations)
 Mimic.copy(Tentacat.Organizations.Hooks)
 Mimic.copy(Console.Deployments.Pr.Git)
 Mimic.copy(Console.Deployments.Pr.Dispatcher)
+Mimic.copy(Console.Deployments.Observer.Runner)
 
 ExUnit.start()
 Ecto.Adapters.SQL.Sandbox.mode(Console.Repo, :manual)