defmodule GitSync.Runs do
  @moduledoc """
  The pub/sub feed of sync progress, one topic per source.
  """

  import Ecto.Query

  alias GitSync.Repo
  alias GitSync.Run
  alias Phoenix.PubSub

  @limit 20

  @doc """
  The most recent runs of one source, newest first, each with what every
  destination made of it.
  """
  def list(source_id, limit \\ @limit) do
    from(r in Run,
      where: r.source_id == ^source_id,
      order_by: [desc: r.started_at, desc: r.id],
      limit: ^limit
    )
    |> Repo.all()
    |> Repo.preload(targets: [destination: :connection])
  end

  def latest_by_source do
    ranked =
      from r in Run,
        select: %{
          id: r.id,
          rank:
            row_number()
            |> over(partition_by: r.source_id, order_by: [desc: r.started_at, desc: r.id])
        }

    from(r in Run,
      join: latest in subquery(ranked),
      on: latest.id == r.id and latest.rank == 1,
      select: struct(r, [:source_id, :status, :started_at])
    )
    |> Repo.all()
    |> Map.new(&{&1.source_id, &1})
  end

  def recent(limit \\ 10) do
    from(r in Run,
      order_by: [desc: r.started_at, desc: r.id],
      limit: ^limit,
      select: struct(r, [:id, :source_id, :status, :started_at, :finished_at])
    )
    |> Repo.all()
    |> Repo.preload(source: :connection)
  end

  def get(id), do: Run |> Repo.get(id) |> Repo.preload(targets: [destination: :connection])

  @doc """
  Fails every run still marked running, and returns how many there were. A run
  is only ever finished by the process that started it, so one left running is
  one whose process is gone: call this as the server comes up, before anything
  starts a run of its own.
  """
  def abandon_running do
    now = DateTime.utc_now() |> DateTime.truncate(:second)

    {count, _} =
      Repo.update_all(from(r in Run, where: r.status == :running),
        set: [
          status: :failure,
          finished_at: now,
          updated_at: now,
          log: "abandoned: the server stopped before this run finished"
        ]
      )

    count
  end

  def subscribe(source_id), do: PubSub.subscribe(GitSync.PubSub, topic(source_id))

  def broadcast(%Run{} = run),
    do: PubSub.broadcast(GitSync.PubSub, topic(run.source_id), {:run, run})

  defp topic(source_id), do: "runs:#{source_id}"
end