Has anyone written a credo rule to only allow scheduling with time interval tuples?

Did you ever sort this out? I’ve been having fun writing credo rules recently and would be happy to make one. My company could likely use this as well.

Fooled with AI agent in Zed editor feeding the prompt -

Write a credo rule that will forbid calling `schedule_in` with a raw number?

i.e. instead of:

`new(arguments, schedule_in: 60 * 1_000)`

I want only this to be allowed:

`new(arguments, schedule_in: {60, :hours})`

After a few iterations, AI agent came up with this code -

defmodule CredoChecks.NoRawScheduleIn do
  @moduledoc """
  Enforces the use of tuple-based time specifications for `schedule_in` instead of raw numbers.

  For example, this is bad:
      new(arguments, schedule_in: 60_000)
      new(arguments, schedule_in: 60 * 1_000)

  This is good:
      new(arguments, schedule_in: {60, :seconds})
      new(arguments, schedule_in: {2, :hours})
  """

  use Credo.Check,
    base_priority: :high,
    category: :consistency,
    explanations: [
      check: """
      Use `schedule_in` with a tuple like `{60, :seconds}` rather than raw numbers
      such as `60_000` or arithmetic expressions like `60 * 1_000`. This improves
      clarity and consistency across the codebase.
      """
    ]

  @doc false
  @impl true
  def run(%SourceFile{} = source_file, params \\ []) do
    issue_meta = IssueMeta.for(source_file, params)
    Credo.Code.prewalk(source_file, &traverse(&1, &2, issue_meta))
  end

  # Match function calls with keyword list arguments
  defp traverse(
         {{:., _, [{:__aliases__, _, _}, :new]}, meta, [_, opts]} = ast,
         issues,
         issue_meta
       )
       when is_list(opts) do
    case Keyword.get(opts, :schedule_in) do
      {number, unit} when is_integer(number) and is_atom(unit) ->
        {ast, issues}

      {:*, _, _} = expr ->
        {ast, [issue_for(issue_meta, meta, expr) | issues]}

      val when is_integer(val) ->
        {ast, [issue_for(issue_meta, meta, val) | issues]}

      _ ->
        {ast, issues}
    end
  end

  defp traverse(ast, issues, _), do: {ast, issues}

  defp issue_for(issue_meta, meta, trigger) do
    line_no = Keyword.get(meta, :line, nil)

    format_issue(
      issue_meta,
      message:
        "Avoid using raw numbers for `schedule_in`. Use a tuple like `{60, :seconds}` instead.",
      trigger: Macro.to_string(trigger),
      line_no: line_no
    )
  end
end

On this test module -

defmodule Test do
  @moduledoc false

  def new(_args, opts \\ []), do: opts
end

Test.new([123], schedule_in: 60_000)
Test.new([123], schedule_in: 60 * 1_000)
Test.new([123], schedule_in: {60, :hours})
Test.new([123], schedule_in: {1, :minute})

It seems to product correct warnings -

image1200×853 51.8 KB

Well that’s no fun

But if you’re going to keep going… this solution is Good Enough but not perfect as it checks any new function.

Ideally you want to ensure that new belongs to a worker. One idea is to include an option to specify a worker namespace or suffix, for example {MyCheck.ObanSchdeduleIn, namespace: [:MyApp, :Workers]}.

I haven’t sorted it out yet. I’ll give @meraj_enigma’s version a try. And I wonder how we could construct a bunch of test cases to ensure that there aren’t any false positives. Maybe we could run it against a bunch of projects from GitHub/hex.pm with some that do and some that don’t use Oban.

Like @sodapopcan mentioned, you will need some adjustment in the version I posted as it will check any new function. I did not realize it’s in Oban worker context.

Ok, here’s my pass.

This is assuming you want to pass the tuple syntax so that is what it checks for and anything else is considered and error.

It defaults to looking for modules with Worker suffixes though you can also specify a custom suffix, a namespace, or both. So use like so in your credo config:

{MyChecks, worker_suffix: "Job"}

This does NOT handle the case of importing new/2 because that’s just kinda evil Happy to add it, though.

If you specify both :worker_suffix and :worker_namespace they both must match. IE you could (for some reason) have MyApp.Workers.AWorkerThisIsNot… though not sure why you would want to do that.

EDIT: Oops left some temp stuff in for reporting… sorta fixed it, sorta (ie, "Some trigger")

defmodule MyChecks.ObanNew do
  use Credo.Check

  @opts [:worker_suffix, :worker_namespace]

  @impl true
  def run(source_file, params \\ []) do
    params =
      if not Enum.any?(@opts, &Keyword.has_key?(params, &1)) do
        Keyword.merge(params, worker_suffix: "Worker")
      else
        params
      end

    issue_meta = IssueMeta.for(source_file, params)

    Credo.Code.prewalk(source_file, &traverse(&1, &2, issue_meta, params))
  end

  defp traverse(
         {{:., _, [{_, _, aliases}, :new]}, meta, [_, opts]} = ast,
         issues,
         issue_meta,
         params
       ) do
    if worker?(aliases, params) do
      case opts[:schedule_in] do
        nil -> {ast, issues}
        {_, _} -> {ast, issues}
        _ -> {ast, issues ++ [issue_for("Some trigger", meta[:line], issue_meta)]}
      end
    else
      {ast, issues}
    end
  end

  defp traverse(ast, issues, _issue_meta, _params) do
    {ast, issues}
  end

  defp worker?(pieces, params) do
    pieces = Enum.map(pieces, &Atom.to_string/1)

    if params[:worker_namespace] != nil and params[:worker_suffix] != nil do
      Enum.any?(pieces, &(&1 == params[:worker_namespace])) &&
        Enum.any?(pieces, &String.ends_with?(&1, params[:worker_suffix]))
    else
      Enum.any?(pieces, fn piece ->
        cond do
          params[:worker_namespace] != nil ->
            piece == params[:worker_namespace]

          params[:worker_suffix] != nil ->
            String.ends_with?(piece, params[:worker_suffix])

          true ->
            false
        end
      end)
    end
  end

  defp issue_for(trigger, line_no, issue_meta) do
    format_issue(
      issue_meta,
      message: "Please use tuple syntax for `shedule_in` option",
      trigger: trigger,
      line_no: line_no
    )
  end
end

and tests:

defmodule MyChecks.ObanNewTest do
  use Credo.Test.Case

  alias MyChecks.ObanNew

  test "schedule_in must use tuple syntax" do
    """
    defmodule CredoSampleModule do
      def foo do
        MyApp.FooWorker.new("something", schedule_in: {60, :hours})
      end
    end
    """
    |> to_source_file()
    |> run_check(ObanNew)
    |> refute_issues()
  end

  test "schedule_in errors if using integer" do
    """
    defmodule CredoSampleModule do
      def foo do
        MyApp.FooWorker.new("something", schedule_in: 400 * 1000)
      end
    end
    """
    |> to_source_file()
    |> run_check(ObanNew)
    |> assert_issue()
  end

  test "ignores modules not ending with `Worker`" do
    """
    defmodule CredoSampleModule do
      def foo do
        NonWorkerModule.new("something", schedule_in: 400 * 1000)
      end
    end
    """
    |> to_source_file()
    |> run_check(ObanNew)
    |> refute_issues()
  end

  test "allows specifying custom suffix" do
    """
    defmodule CredoSampleModule do
      def foo do
        MyJob.new("something", schedule_in: 400 * 1000)
      end
    end
    """
    |> to_source_file()
    |> run_check(ObanNew, worker_suffix: "Job")
    |> assert_issue()
  end

  test "allow specifying a namespace" do
    """
    defmodule CredoSampleModule do
      def foo do
        MyJobs.Doit.new("something", schedule_in: 400 * 1000)
      end
    end
    """
    |> to_source_file()
    |> run_check(ObanNew, worker_namespace: "MyJobs")
    |> assert_issue()
  end

  test "allow specifying namespace and suffix" do
    """
    defmodule CredoSampleModule do
      def foo do
        MyJobs.SomeWorker.new("something", schedule_in: 400 * 1000)
      end
    end
    """
    |> to_source_file()
    |> run_check(ObanNew, worker_namespace: "MyJobs", worker_suffix: "Worker")
    |> assert_issue()

    """
    defmodule CredoSampleModule do
      def foo do
        MyJobs.Unrelated.new("something", schedule_in: 400 * 1000)
      end
    end
    """
    |> to_source_file()
    |> run_check(ObanNew, worker_namespace: "MyJobs", worker_suffix: "Worker")
    |> refute_issues()

    """
    defmodule CredoSampleModule do
      def foo do
        NonObanModuleEndingInWorker.new("something", schedule_in: 400 * 1000)
      end
    end
    """
    |> to_source_file()
    |> run_check(ObanNew, worker_namespace: "MyJobs", worker_suffix: "Worker")
    |> refute_issues()
  end
end

Thank you for this! Although after testing both of these, they don’t include this common pattern we often use where we call new as a non-qualified module call from within the worker module itself.

@trace :debounced
def debounced(args) do
  timeout_seconds = ceil(Application.fetch_env!(:my_app, :thumbnail_timeout_ms) / 1000)

  new(args,
    schedule_in: timeout_seconds,
    meta: %{distributed_trace_headers: NewRelic.distributed_trace_headers(:other)},
    unique: [fields: [:args, :worker], period: timeout_seconds]
  )
end

And separately it doesn’t handle calls where the schedule_in is passed through another function first, e.g. like: start_my_worker(user.id, email, schedule_in: 5). Although that probably can’t be reasonably caught.

But they are helpful starting points!

Ya I don’t think this would be possible with Credo AFAIK. You’d have to do some custom parsing of imports. With that, it’s probably better then not to worry about module qualification (ie, @meraj_enigma’s version) and hope that another function new is not going to have a schedule_in option.

Otherwise, for unqualified calls, you can look for use Oban.Worker and track it in the accumulator along with issues (I do this for a couple of my checks and it works nicely):

def run(...) do
  # ...

  {issues, _worker?} =
    {Credo.Code.prewalk(source_file, &traverse(&1, &2, issue_meta, params), {[], false})

  issues
end

defp traverse({:use, _, [{_, _, [:Oban, :Worker]} | _]} = ast, {issues, _}, _issue_meta) do
  {ast, {issues, true}}
end

After learning your use cases, came up with this (several passes with AI agent here too!) that handles all cases except the one above ^^.

Here is the full code -

defmodule CredoChecks.NoRawScheduleIn do
  @moduledoc """
  Enforces the use of tuple-based time specifications for `schedule_in` instead of raw numbers
  in Oban workers only.

  This check works with both module-qualified and unqualified `new` calls within Oban worker
  modules, and also flags variables that might contain raw time values.

  For example, this is bad in an Oban worker:
      # Module-qualified calls
      MyWorker.new(arguments, schedule_in: 60_000)
      MyWorker.new(arguments, schedule_in: 60 * 1_000)

      # Unqualified calls
      new(arguments, schedule_in: 60_000)
      new(arguments, schedule_in: 60 * 1_000)

      # Variables containing raw time values
      timeout_seconds = 60
      new(arguments, schedule_in: timeout_seconds)

  This is good:
      new(arguments, schedule_in: {60, :seconds})
      new(arguments, schedule_in: {2, :hours})
  """

  use Credo.Check,
    base_priority: :high,
    category: :consistency,
    explanations: [
      check: """
      In Oban workers, use `schedule_in` with a tuple like `{60, :seconds}` rather than
      raw numbers such as `60_000`, arithmetic expressions like `60 * 1_000`, or variables
      that might contain raw time values. This improves clarity and consistency across
      the codebase by making time units explicit.
      """
    ]

  @doc false
  @impl true
  def run(%SourceFile{} = source_file, params \\ []) do
    issue_meta = IssueMeta.for(source_file, params)

    result = Credo.Code.prewalk(source_file, &traverse(&1, &2, issue_meta), {nil, false, []})

    case result do
      {_ast, {_, _, issues}} -> issues |> Enum.reverse()
      {_, _, issues} -> issues |> Enum.reverse()
      _ -> []
    end
  end

  # Track module definitions and detect Oban workers
  defp traverse(
         {:defmodule, _meta, [{:__aliases__, _, module_name} | _]} = ast,
         {_current_module, _is_oban_worker, issues},
         _issue_meta
       ) do
    {ast, {module_name, false, issues}}
  end

  # Detect Oban worker usage
  defp traverse(
         {:use, _, [{:__aliases__, _, [:Oban, :Worker]} | _]} = ast,
         {current_module, _is_oban_worker, issues},
         _issue_meta
       ) do
    {ast, {current_module, true, issues}}
  end

  # Match module-qualified function calls with keyword list arguments (only in Oban workers)
  defp traverse(
         {{:., _, [{:__aliases__, _, _}, :new]}, meta, [_, opts]} = ast,
         {current_module, true, issues},
         issue_meta
       )
       when is_list(opts) do
    new_issues = check_schedule_in_opts(opts, issues, issue_meta, meta)
    {ast, {current_module, true, new_issues}}
  end

  # Match unqualified function calls with keyword list arguments (only in Oban workers)
  defp traverse(
         {:new, meta, [_, opts]} = ast,
         {current_module, true, issues},
         issue_meta
       )
       when is_list(opts) do
    new_issues = check_schedule_in_opts(opts, issues, issue_meta, meta)
    {ast, {current_module, true, new_issues}}
  end

  # Default case - continue traversal
  defp traverse(ast, acc, _issue_meta), do: {ast, acc}

  defp check_schedule_in_opts(opts, issues, issue_meta, meta) do
    case Keyword.get(opts, :schedule_in) do
      {number, unit} when is_integer(number) and is_atom(unit) ->
        issues

      {:*, _, _} = expr ->
        [issue_for(issue_meta, meta, expr, :arithmetic) | issues]

      val when is_integer(val) ->
        [issue_for(issue_meta, meta, val, :raw_number) | issues]

      # Variable that might contain a raw time value
      {var_name, _, nil} when is_atom(var_name) ->
        [issue_for(issue_meta, meta, var_name, :variable) | issues]

      _ ->
        issues
    end
  end

  defp issue_for(issue_meta, meta, trigger, type) do
    line_no = Keyword.get(meta, :line, nil)

    message =
      case type do
        :arithmetic ->
          "In Oban workers, avoid using arithmetic expressions for `schedule_in`. Use a tuple like `{60, :seconds}` instead."

        :raw_number ->
          "In Oban workers, avoid using raw numbers for `schedule_in`. Use a tuple like `{60, :seconds}` instead."

        :variable ->
          "In Oban workers, avoid using variables that may contain raw time values for `schedule_in`. Use a tuple like `{60, :seconds}` instead."
      end

    format_issue(
      issue_meta,
      message: message,
      trigger: Macro.to_string(trigger),
      line_no: line_no
    )
  end
end

And here is the file I tested on -

defmodule MyObanWorker do
  use Oban.Worker, queue: :default

  @impl Oban.Worker
  def perform(%Oban.Job{args: args}) do
    # Implementation here
    :ok
  end

  # These should be flagged - bad patterns in Oban worker
  def schedule_work_bad_examples do
    # Raw numbers - should be flagged
    MyObanWorker.new(%{task: "example"}, schedule_in: 60_000)
    new(%{task: "example"}, schedule_in: 30_000)

    # Arithmetic expressions - should be flagged
    MyObanWorker.new(%{task: "example"}, schedule_in: 60 * 1_000)
    new(%{task: "example"}, schedule_in: 45 * 1_000)

    # Variables that might contain raw time values - should be flagged
    timeout_seconds = 60
    raw_timeout = 60_000
    calculated_timeout = ceil(30_000 / 1000)

    new(%{task: "example"}, schedule_in: timeout_seconds)
    MyObanWorker.new(%{task: "example"}, schedule_in: raw_timeout)
    new(%{task: "example"}, schedule_in: calculated_timeout)
  end

  # These should NOT be flagged - good patterns
  def schedule_work_good_examples do
    # Tuple format - should be OK
    MyObanWorker.new(%{task: "example"}, schedule_in: {60, :seconds})
    new(%{task: "example"}, schedule_in: {2, :hours})
    new(%{task: "example"}, schedule_in: {30, :minutes})
  end

  @trace :debounced
  def debounced(args) do
    timeout_seconds = ceil(Application.fetch_env!(:my_app, :thumbnail_timeout_ms) / 1000)

    # This should be flagged since it's in an Oban worker
    new(args,
      schedule_in: timeout_seconds,
      meta: %{distributed_trace_headers: NewRelic.distributed_trace_headers(:other)},
      unique: [fields: [:args, :worker], period: timeout_seconds]
    )
  end
end

defmodule AnotherObanWorker do
  use Oban.Worker, queue: :high_priority, max_attempts: 3

  @impl Oban.Worker
  def perform(%Oban.Job{args: args}) do
    :ok
  end

  # These should also be flagged since this is an Oban worker
  def bad_scheduling do
    new(%{data: "test"}, schedule_in: 120_000)
    AnotherObanWorker.new(%{data: "test"}, schedule_in: 5 * 60 * 1000)
  end
end

# This is NOT an Oban worker - none of these should be flagged
defmodule RegularModule do
  def new(_args, opts \\ []), do: opts

  def some_function do
    # These should NOT be flagged - not in an Oban worker
    RegularModule.new([123], schedule_in: 60_000)
    new([123], schedule_in: 60 * 1_000)

    timeout_var = 30_000
    new([123], schedule_in: timeout_var)

    # Even tuple format in non-Oban module
    new([123], schedule_in: {60, :seconds})
  end
end

# Another non-Oban module that might use similar patterns
defmodule SomeService do
  def start_timer(args) do
    # This looks similar but should NOT be flagged since it's not an Oban worker
    new(args, schedule_in: 5000)
    SomeService.new(args, schedule_in: 10 * 1000)
  end

  def new(_args, _opts), do: :ok
end

Output credo results -

image1200×1225 123 KB