Expand Oban usage for logging, rerror reporting, and periodic jobs

config/config.exs

@@ -40,8 +40,8 @@ config :changelog,
 
 config :changelog, Oban,
   repo: Changelog.Repo,
-  plugins: [Oban.Plugins.Pruner, Oban.Plugins.Stager],
-  queues: [comment_notifier: 10]
+  queues: [comment_notifier: 10, scheduled: 5],
+  plugins: [Oban.Plugins.Pruner, Oban.Plugins.Stager]
 
 config :changelog, Changelog.Mailer, adapter: Bamboo.LocalAdapter
 

config/prod.exs

@@ -50,13 +50,17 @@ config :changelog, Changelog.Mailer,
 
 config :elixir, :time_zone_database, Tzdata.TimeZoneDatabase
 
-config :changelog, Changelog.Scheduler,
-  global: true,
-  timezone: "US/Central",
-  jobs: [
-    {"0 4 * * *", {Changelog.Stats, :process, []}},
-    {"0 3 * * *", {Changelog.Slack.Tasks, :import_member_ids, []}},
-    {"* * * * *", {Changelog.NewsQueue, :publish, []}}
+config :changelog, Oban,
+  plugins: [
+    Oban.Plugins.Pruner,
+    Oban.Plugins.Stager,
+    {Oban.Plugins.Cron,
+     timezone: "US/Central",
+     crons: [
+       {"0 4 * * *", Changelog.ObanWorkers.StatsProcessor},
+       {"0 3 * * *", Changelog.ObanWorkers.SlackImporter},
+       {"* * * * *", Changelog.ObanWorkers.NewsPublisher}
+     ]}
   ]
 
 config :changelog, Changelog.PromEx,

config/test.exs

@@ -22,6 +22,5 @@ config :changelog, Changelog.Repo,
   username: System.get_env("DB_USER", "postgres")
 
 config :changelog, Oban,
-  crontab: false,
   queues: false,
   plugins: false

lib/changelog/application.ex

@@ -24,16 +24,19 @@ defmodule Changelog.Application do
         global_ttl: :timer.minutes(5),
         touch_on_read: false
       ),
-      Changelog.Scheduler,
       Changelog.EpisodeTracker,
       Changelog.Metacasts.Filterer.Cache,
       {Oban, oban_config()}
     ]
 
-    # See https://hexdocs.pm/elixir/Supervisor.html
-    # for other strategies and supported options
-    opts = [strategy: :one_for_one, name: Changelog.Supervisor]
-    Supervisor.start_link(children, opts)
+    # Only attach the telemetry logger when we aren't in an IEx shell
+    unless Code.ensure_loaded?(IEx) && IEx.started?() do
+      Oban.Telemetry.attach_default_logger(:info)
+
+      Changelog.ObanReporter.attach()
+    end
+
+    Supervisor.start_link(children, strategy: :one_for_one, name: Changelog.Supervisor)
   end
 
   defp oban_config do

lib/changelog/oban_reporter.ex

@@ -0,0 +1,23 @@
+defmodule Changelog.ObanReporter do
+  @moduledoc false
+
+  def attach do
+    :telemetry.attach(
+      "oban-errors",
+      [:oban, :job, :exception],
+      &handle_event/4,
+      []
+    )
+  end
+
+  def handle_event([:oban, :job, _], measure, meta, _) do
+    extra =
+      meta.job
+      |> Map.take([:id, :args, :meta, :queue, :worker, :attempt, :max_attempts])
+      |> Map.merge(measure)
+
+    Sentry.capture_exception(meta.error, stacktrace: meta.stacktrace, extra: extra)
+  end
+
+  def handle_event(_event, _measure, _meta, _opts), do: :ok
+end

lib/changelog/oban_workers/comment_notifier.ex

@@ -7,8 +7,6 @@ defmodule Changelog.ObanWorkers.CommentNotifier do
 
   alias Changelog.{NewsItemComment, Notifier, Repo}
 
-  @five_mins 60 * 5
-
   @impl Oban.Worker
   def perform(%Oban.Job{args: %{"comment_id" => comment_id}}) do
     comment = Repo.get(NewsItemComment, comment_id)
@@ -22,7 +20,7 @@ defmodule Changelog.ObanWorkers.CommentNotifier do
   """
   def schedule_notification(%NewsItemComment{id: id}) do
     %{comment_id: id}
-    |> __MODULE__.new(schedule_in: @five_mins)
+    |> new(schedule_in: {5, :minutes})
     |> Oban.insert()
   end
 end

lib/changelog/oban_workers/news_publisher.ex

@@ -0,0 +1,10 @@
+defmodule Changelog.ObanWorkers.NewsPublisher do
+  use Oban.Worker, queue: :scheduled
+
+  alias Changelog.NewsQueue
+
+  @impl Oban.Worker
+  def perform(_job) do
+    {:ok, NewsQueue.publish()}
+  end
+end

lib/changelog/oban_workers/slack_importer.ex

@@ -0,0 +1,28 @@
+defmodule Changelog.ObanWorkers.SlackImporter do
+  use Oban.Worker, queue: :scheduled
+
+  alias Changelog.{Person, Repo}
+  alias Changelog.Slack.Client
+
+  import Ecto.Query
+
+  @impl Oban.Worker
+  def perform(_job) do
+    %{"members" => members} = Client.list()
+
+    for %{"id" => id, "profile" => profile} <- members do
+      email = Map.get(profile, "email", "")
+
+      import_member_id(id, email)
+    end
+
+    :ok
+  end
+
+  def import_member_id(id, email) do
+    Person
+    |> where([p], p.email == ^email)
+    |> where([p], is_nil(p.slack_id) or p.slack_id == "pending")
+    |> Repo.update_all(set: [slack_id: id])
+  end
+end

lib/changelog/stats/stats.ex → lib/changelog/oban_workers/stats_processor.ex

@@ -1,59 +1,57 @@
-defmodule Changelog.Stats do
-  import Ecto
-  import Ecto.Changeset
+defmodule Changelog.ObanWorkers.StatsProcessor do
+  use Oban.Worker, queue: :scheduled, unique: [period: 600]
+
+  import Ecto.Query, only: [select: 3]
 
   alias Changelog.{Cache, Podcast, Repo, Episode, EpisodeStat}
   alias Changelog.Stats.{Analyzer, Parser, S3}
+  alias Ecto.Changeset
 
   require Logger
 
-  def process do
-    end_date = Timex.today()
-    start_date = Timex.shift(end_date, days: -2)
+  @impl Oban.Worker
+  def perform(%Job{args: %{"date" => date, "podcast_id" => podcast_id}}) do
+    date = Date.from_iso8601!(date)
+    podcast = Repo.get!(Podcast, podcast_id)
 
-    for time <- Timex.Interval.new(from: start_date, until: end_date) do
-      time |> Timex.to_date() |> process()
-    end
-  end
-
-  def process(date) do
-    Logger.info("Stats: Start processing for #{date}")
-    podcasts = Repo.all(Podcast.public())
-    process(date, podcasts)
-    Cache.delete_prefix("stats-")
-    Logger.info("Stats: Finished processing for #{date}")
-  end
+    processed =
+      date
+      |> S3.get_logs(podcast.slug)
+      |> Parser.parse()
+      |> Enum.group_by(& &1.episode)
+      |> Enum.map(fn {slug, entries} -> process_episode(date, podcast, slug, entries) end)
 
-  def process(date, podcast) when not is_list(podcast), do: process(date, [podcast])
+    Podcast.update_stat_counts(podcast)
 
-  def process(date, podcasts) do
-    podcasts
-    |> Enum.map(&Task.async(fn -> process_podcast(date, &1) end))
-    # 10 minutes
-    |> Enum.flat_map(&Task.await(&1, 600_000))
+    {:ok, processed}
   end
 
-  defp process_podcast(date, podcast) do
-    Logger.info("Stats: Processing #{podcast.name}")
+  def perform(_job) do
+    today = Date.utc_today()
+    range = Date.range(Date.add(today, -2), Date.add(today, -1))
 
-    processed =
-      S3.get_logs(date, podcast.slug)
-      |> Parser.parse()
-      |> Enum.group_by(& &1.episode)
-      |> Enum.map(fn {slug, entries} ->
-        process_episode(date, podcast, slug, entries)
-      end)
+    podcast_ids =
+      Podcast.public()
+      |> select([p], p.id)
+      |> Repo.all()
 
-    Podcast.update_stat_counts(podcast)
-    Logger.info("Stats: Finished Processing #{podcast.name}")
+    jobs =
+      for(date <- range, pid <- podcast_ids, do: %{date: date, podcast_id: pid})
+      |> Enum.map(&new/1)
+      |> Oban.insert_all()
+
+    Cache.delete_prefix("stats-")
 
-    processed
+    {:ok, jobs}
   end
 
+  @impl Oban.Worker
+  def timeout(_job), do: 600_000
+
   defp process_episode(date, podcast, slug, entries) do
-    if episode = Repo.get_by(assoc(podcast, :episodes), slug: slug) do
+    if episode = Repo.get_by(Ecto.assoc(podcast, :episodes), slug: slug) do
       stat =
-        case Repo.get_by(assoc(episode, :episode_stats), date: date) do
+        case Repo.get_by(Ecto.assoc(episode, :episode_stats), date: date) do
           nil ->
             %EpisodeStat{
               episode_id: episode.id,
@@ -67,7 +65,7 @@ defmodule Changelog.Stats do
         end
 
       stat =
-        change(stat, %{
+        Changeset.change(stat, %{
           total_bytes: Analyzer.bytes(entries),
           downloads: Analyzer.downloads(entries, stat.episode_bytes),
           uniques: Analyzer.uniques_count(entries),

lib/changelog_web/controllers/slack_controller.ex

@@ -2,7 +2,8 @@ defmodule ChangelogWeb.SlackController do
   use ChangelogWeb, :controller
 
   alias Changelog.Episode
-  alias Changelog.Slack.{Client, Countdown, Messages, Tasks}
+  alias Changelog.Slack.{Client, Countdown, Messages}
+  alias Changelog.ObanWorkers.SlackImporter
 
   require Logger
 
@@ -32,7 +33,7 @@ defmodule ChangelogWeb.SlackController do
     id = Map.get(member, "id")
     email = get_in(member, ["profile", "email"]) || ""
     Client.im(id, Messages.welcome())
-    Tasks.import_member_id(id, email)
+    SlackImporter.import_member_id(id, email)
     json(conn, %{})
   end
 

lib/mix/tasks/slack.ex

@@ -5,6 +5,6 @@ defmodule Mix.Tasks.Changelog.Slack do
 
   def run(_) do
     Mix.Task.run("app.start")
-    Changelog.Slack.Tasks.import_member_ids()
+    Changelog.ObanWorkers.SlackImporter.perform(%Oban.Job{})
   end
 end

lib/mix/tasks/stats/process.ex

@@ -1,6 +1,9 @@
 defmodule Mix.Tasks.Changelog.Stats.Process do
   use Mix.Task
 
+  alias Changelog.{Podcast, Repo}
+  alias Changelog.ObanWorkers.StatsProcessor
+
   @shortdoc "Processes stats for given date, or all missing dates"
 
   def run(args) when is_nil(args), do: run([])
@@ -9,8 +12,22 @@ defmodule Mix.Tasks.Changelog.Stats.Process do
     Mix.Task.run("app.start")
 
     case Timex.parse(List.first(args), "{YYYY}-{0M}-{D}") do
-      {:ok, time} -> Changelog.Stats.process(Timex.to_date(time))
-      {:error, _message} -> Changelog.Stats.process()
+      {:ok, time} ->
+        date = Timex.to_date(time)
+
+        Podcast.public()
+        |> Repo.all()
+        |> Enum.map(&StatsProcessor.new(%{date: date, podcast_id: &1.id}))
+        |> Oban.insert_all()
+
+      {:error, _message} ->
+        %{}
+        |> StatsProcessor.new()
+        |> Oban.insert!()
     end
+
+    results = Oban.drain_queue(queue: :scheduled, with_recursion: true, with_safety: false)
+
+    Mix.shell().info("Stats processed for #{results.success - 1} dates/podcasts")
   end
 end

mix.exs

@@ -41,7 +41,7 @@ defmodule Changelog.Mixfile do
       {:plug_cowboy, "~> 2.0"},
       # Leaving this here for future dev loops with @akoutmos
       # {:prom_ex, github: "akoutmos/prom_ex", branch: "master"},
-      {:oban, "~> 2.7.2"},
+      {:oban, "~> 2.8"},
       {:prom_ex, "~> 1.3.0"},
       {:unplug, "~> 0.2.1"},
       {:postgrex, ">= 0.0.0"},
@@ -66,7 +66,6 @@ defmodule Changelog.Mixfile do
       {:nimble_csv, "~> 1.1.0"},
       {:sweet_xml, "~> 0.6"},
       {:user_agent_parser, "~> 1.0"},
-      {:quantum, ">= 2.1.0"},
       {:oauth, github: "tim/erlang-oauth"},
       {:ueberauth_github, "~> 0.4"},
       {:ueberauth_twitter, github: "jerodsanto/ueberauth_twitter"},