neiam /gitgud

defmodule GitGud.Workflows.Schedules do
  @moduledoc """
  `on: schedule`: which crons a repository's workflows declare, and
  firing them.

  As on GitHub, a schedule is read from the **default branch's** latest
  commit and runs there, five-field POSIX cron in UTC (Oban's
  `@daily`-style nicknames also parse; `@reboot` is refused). A cron that
  can fire more often than every five minutes — GitHub's limit — is
  refused with a warning, as is one that doesn't parse.

  `sync/1` mirrors the declarations into `workflow_schedules` whenever the
  default branch moves or changes; `GitGud.Workflows.Workers.ScheduleTick`
  fires the due rows each minute.
  """

  import Ecto.Query

  alias GitGud.Repo
  alias GitGud.Repositories.Repository
  alias GitGud.Workflows
  alias GitGud.Workflows.WorkflowSchedule
  alias Oban.Cron.Expression

  require Logger

  # GitHub's shortest interval between two runs of one schedule.
  @min_interval_minutes 5

  @doc """
  Make the repository's `workflow_schedules` rows match the crons its
  default branch declares now: add new ones, drop the rest. A repository
  whose default branch doesn't exist has none. Returns `{:ok, count}`.
  """
  def sync(%Repository{id: rid} = repo) do
    wanted =
      case Workflows.default_branch_head(repo) do
        {:ok, _ref, sha} ->
          for {path, doc} <- Workflows.discover_workflows(repo, sha),
              cron <- crons(doc),
              valid?(cron, path, repo),
              uniq: true,
              do: {path, cron}

        :error ->
          []
      end

    now = DateTime.utc_now(:second)

    rows =
      for {path, cron} <- wanted,
          do: %{
            repository_id: rid,
            workflow_path: path,
            cron: cron,
            inserted_at: now,
            updated_at: now
          }

    Repo.transaction(fn ->
      _ =
        Repo.insert_all(WorkflowSchedule, rows,
          on_conflict: :nothing,
          conflict_target: [:repository_id, :workflow_path, :cron]
        )

      stale =
        from(s in WorkflowSchedule,
          where: s.repository_id == ^rid,
          select: {s.id, s.workflow_path, s.cron}
        )
        |> Repo.all()
        |> Enum.reject(fn {_id, path, cron} -> {path, cron} in wanted end)
        |> Enum.map(&elem(&1, 0))

      _ = Repo.delete_all(from s in WorkflowSchedule, where: s.id in ^stale)
    end)

    {:ok, length(wanted)}
  end

  @doc "The cron strings a workflow's `on.schedule` lists, in order."
  def crons(%{triggers: %{"schedule" => entries}}) when is_list(entries) do
    for %{"cron" => cron} <- entries, is_binary(cron), do: String.trim(cron)
  end

  def crons(_doc), do: []

  @doc """
  `{:ok, expression}` for a cron a schedule may use, else `{:error,
  reason}`: it must parse, mustn't be `@reboot`, and mustn't fire more
  often than every #{@min_interval_minutes} minutes.
  """
  def parse(cron) when is_binary(cron) do
    case Expression.parse(cron) do
      {:ok, %Expression{reboot?: true}} -> {:error, "@reboot is not a schedule"}
      {:ok, expr} -> check_interval(expr)
      {:error, e} -> {:error, Exception.message(e)}
    end
  end

  # The shortest gap between two firings, read off the expression's
  # minute and hour sets rather than by stepping `next_at/2` through a day
  # (exact, and no clock to pick). Within an hour it is
  # the closest two minutes; across an hour boundary, when two adjacent
  # hours both fire (23 and 0 count, taking any two days as adjacent),
  # the last minute of one to the first of the next.
  defp check_interval(expr) do
    minutes = Enum.sort(expr.minutes)
    hours = expr.hours

    within =
      minutes |> Enum.zip(Enum.drop(minutes, 1)) |> Enum.map(fn {a, b} -> b - a end)

    across =
      if Enum.any?(hours, &MapSet.member?(hours, rem(&1 + 1, 24))),
        do: [60 - List.last(minutes) + hd(minutes)],
        else: []

    case Enum.min(within ++ across, fn -> nil end) do
      gap when is_integer(gap) and gap < @min_interval_minutes ->
        {:error, "fires more often than every #{@min_interval_minutes} minutes"}

      _ ->
        {:ok, expr}
    end
  end

  defp valid?(cron, path, repo) do
    case parse(cron) do
      {:ok, _expr} ->
        true

      {:error, reason} ->
        Logger.warning(
          "ignoring schedule #{inspect(cron)} in #{path} of repository #{repo.id}: #{reason}"
        )

        false
    end
  end

  @doc """
  Fire every schedule due at `at`'s UTC minute, at most once per minute
  each, however many nodes tick. Archived repositories don't run.
  Returns the runs created.
  """
  def fire_due(%DateTime{} = at) do
    minute =
      at |> DateTime.shift_zone!("Etc/UTC") |> DateTime.truncate(:second) |> Map.put(:second, 0)

    from(s in WorkflowSchedule,
      join: r in Repository,
      on: r.id == s.repository_id,
      where: r.archived == false,
      select: s
    )
    |> Repo.all()
    |> Enum.filter(&due?(&1, minute))
    |> Enum.filter(&claim(&1, minute))
    |> Enum.flat_map(&fire/1)
  end

  defp due?(%WorkflowSchedule{cron: cron}, minute) do
    case parse(cron) do
      {:ok, expr} -> Expression.now?(expr, minute)
      _ -> false
    end
  end

  # The check and the stamp in one statement: of two nodes ticking the
  # same minute, one updates the row and the other finds nothing to.
  defp claim(%WorkflowSchedule{id: id}, minute) do
    {count, _} =
      from(s in WorkflowSchedule,
        where: s.id == ^id and (is_nil(s.last_fired_at) or s.last_fired_at != ^minute)
      )
      |> Repo.update_all(set: [last_fired_at: minute])

    count == 1
  end

  defp fire(%WorkflowSchedule{} = schedule) do
    repo = Repo.get!(Repository, schedule.repository_id)

    case Workflows.dispatch_schedule(repo, schedule.workflow_path, schedule.cron) do
      {:ok, run} ->
        [run]

      {:error, reason} ->
        Logger.info(
          "schedule #{inspect(schedule.cron)} of #{schedule.workflow_path} " <>
            "(repository #{repo.id}) didn't run: #{inspect(reason)}"
        )

        []
    end
  end
end