neiam /gitgud

defmodule GitGud.Workflows.ConcurrencyTest do
  @moduledoc """
  `concurrency:` — at most one run (or job) of a group in progress and one
  waiting, as on GitHub.

  Workflow-level groups are driven through real pushes (`dispatch/2` reads
  the workflow out of a commit on disk); job-level ones through `rerun/1`
  of a stored document, which creates jobs by the same path.
  """
  use GitGud.DataCase, async: false

  import ExUnit.CaptureLog
  import GitGud.ForgeFixtures

  alias GitGud.Repositories.Repository
  alias GitGud.Workflows
  alias GitGud.Workflows.WorkflowJob
  alias GitGud.Workflows.WorkflowRun
  alias GitGud.Workflows.Yaml

  setup do
    {_user, repo} = repository_fixture()
    {:ok, repo: repo}
  end

  defp workflow(concurrency) do
    """
    name: deploy
    on: push
    concurrency: #{concurrency}
    jobs:
      deploy:
        runs-on: self-hosted
        steps:
          - run: ./deploy
    """
  end

  # One commit holding the workflow; every push of it to `ref` makes a run.
  defp commit!(%Repository{disk_path: path}, yaml) do
    dir = Path.join(System.tmp_dir!(), "concurrency-#{System.unique_integer([:positive])}")
    File.mkdir_p!(dir)
    file = Path.join(dir, "ci.yml")
    File.write!(file, yaml)
    env = [{"GIT_INDEX_FILE", Path.join(dir, "index")}]

    try do
      blob = git!(path, ["hash-object", "-w", file])

      git!(
        path,
        ["update-index", "--add", "--cacheinfo", "100644,#{blob},.forgejo/workflows/ci.yml"],
        env
      )

      tree = git!(path, ["write-tree"], env)
      git!(path, ["commit-tree", tree, "-m", "ci"])
    after
      File.rm_rf!(dir)
    end
  end

  defp git!(path, args, env \\ []) do
    {out, 0} = System.cmd("git", args, cd: path, stderr_to_stdout: true, env: env)
    String.trim(out)
  end

  defp push!(repo, sha, ref \\ "refs/heads/main") do
    git!(repo.disk_path, ["update-ref", ref, sha])
    {:ok, head} = GitGud.Git.from_hex(sha)
    {:ok, [run]} = Workflows.dispatch(repo, %{trigger: "push", head_sha: head, ref: ref})
    run
  end

  defp jobs(run) do
    from(j in WorkflowJob, where: j.workflow_run_id == ^run.id, order_by: j.id) |> Repo.all()
  end

  defp job(run), do: run |> jobs() |> hd()
  defp run(run), do: Repo.get!(WorkflowRun, run.id)

  describe "workflow level" do
    test "one in progress, one waiting; a newer run replaces the waiting one", %{repo: repo} do
      sha = commit!(repo, workflow("{group: deploy, cancel-in-progress: false}"))

      first = push!(repo, sha)
      assert run(first).concurrency_group == "deploy"
      assert job(first).status == "queued"

      second = push!(repo, sha)
      assert %WorkflowJob{status: "waiting", status_reason: "Waiting for deploy"} = job(second)
      assert run(second).status == "queued"

      third = push!(repo, sha)
      assert run(second).status == "cancelled"
      assert %WorkflowJob{status: "cancelled", status_reason: reason} = job(second)
      assert reason =~ "deploy"
      assert job(third).status == "waiting"
      assert job(first).status == "queued"

      # The first finishing lets the third go.
      {:ok, _} = Workflows.update_job_status(job(first), "running")
      {:ok, _} = Workflows.update_job_status(job(first), "success")
      assert run(first).status == "success"
      assert %WorkflowJob{status: "queued", status_reason: nil} = job(third)
    end

    test "a cancelled run lets the next one go too", %{repo: repo} do
      sha = commit!(repo, workflow("deploy"))

      first = push!(repo, sha)
      second = push!(repo, sha)
      assert job(second).status == "waiting"

      {:ok, _} = Workflows.cancel(run(first))
      assert job(second).status == "queued"
    end

    test "cancel-in-progress cancels the running run, and the new one starts", %{repo: repo} do
      sha = commit!(repo, workflow("{group: deploy, cancel-in-progress: true}"))

      first = push!(repo, sha)
      {:ok, _} = Workflows.update_job_status(job(first), "running")

      second = push!(repo, sha)
      assert run(first).status == "cancelled"
      assert job(first).status == "cancelled"
      assert job(second).status == "queued"
    end

    test "a group per ref: pushes to two branches don't interact", %{repo: repo} do
      sha = commit!(repo, workflow("${{ github.workflow }}-${{ github.ref }}"))

      main = push!(repo, sha, "refs/heads/main")
      feature = push!(repo, sha, "refs/heads/feature")
      assert run(main).concurrency_group == "deploy-refs/heads/main"
      assert run(feature).concurrency_group == "deploy-refs/heads/feature"
      assert job(main).status == "queued"
      assert job(feature).status == "queued"

      again = push!(repo, sha, "refs/heads/main")
      assert job(again).status == "waiting"
      assert job(feature).status == "queued"
    end

    test "a group we can't evaluate runs without concurrency control", %{repo: repo} do
      sha = commit!(repo, workflow("${{ toJSON(github) }}"))

      log =
        capture_log(fn ->
          first = push!(repo, sha)
          second = push!(repo, sha)
          send(self(), {:runs, first, second})
        end)

      assert_received {:runs, first, second}
      assert log =~ "concurrency ignored"
      assert run(first).concurrency_group == nil

      for r <- [first, second] do
        assert %WorkflowJob{status: "queued", status_reason: "Concurrency ignored: " <> _} =
                 job(r)
      end
    end
  end

  # `rerun/1` of a stored document creates jobs exactly as a dispatch does.
  defp run_workflow(repo, yaml) do
    {:ok, doc} = Yaml.parse(yaml)

    seed =
      %WorkflowRun{repository_id: repo.id}
      |> WorkflowRun.create_changeset(%{
        "number" => System.unique_integer([:positive]),
        "workflow_path" => ".forgejo/workflows/ci.yml",
        "workflow_name" => doc.name,
        "trigger" => "push",
        "head_ref" => "refs/heads/main",
        "head_sha" => :crypto.strong_rand_bytes(20),
        "workflow_doc" => doc.raw
      })
      |> Repo.insert!()

    {:ok, run} = Workflows.rerun(seed)
    run
  end

  describe "job level" do
    test "legs in the same group serialize, legs in different groups don't", %{repo: repo} do
      run =
        run_workflow(repo, """
        name: ci
        on: push
        jobs:
          deploy:
            strategy:
              matrix:
                target: [a, b]
                n: [1, 2]
            concurrency: deploy-${{ matrix.target }}
            runs-on: self-hosted
            steps:
              - run: ./deploy
        """)

      legs = Map.new(jobs(run), &{{&1.matrix["target"], &1.matrix["n"]}, &1})
      assert legs[{"a", 1}].status == "queued"

      assert %WorkflowJob{status: "waiting", status_reason: "Waiting for deploy-a"} =
               legs[{"a", 2}]

      assert legs[{"b", 1}].status == "queued"
      assert legs[{"b", 2}].status == "waiting"

      {:ok, _} = Workflows.update_job_status(legs[{"a", 1}], "success")
      assert Repo.get!(WorkflowJob, legs[{"a", 2}].id).status == "queued"
      assert Repo.get!(WorkflowJob, legs[{"b", 2}].id).status == "waiting"
      assert Repo.get!(WorkflowRun, run.id).status == "queued"
    end

    test "decided when the job is queued, so it can read needs", %{repo: repo} do
      run =
        run_workflow(repo, """
        name: ci
        on: push
        jobs:
          build:
            runs-on: self-hosted
            steps:
              - run: true
          deploy:
            needs: build
            concurrency: deploy-${{ needs.build.result }}
            runs-on: self-hosted
            steps:
              - run: ./deploy
        """)

      [build, deploy] = jobs(run)
      assert deploy.status == "blocked"
      assert deploy.concurrency_group == nil

      {:ok, _} = Workflows.update_job_status(build, "success")

      assert %WorkflowJob{status: "queued", concurrency_group: "deploy-success"} =
               Repo.get!(WorkflowJob, deploy.id)
    end

    test "cancel-in-progress across runs: the older job is superseded", %{repo: repo} do
      yaml = """
      name: ci
      on: push
      jobs:
        deploy:
          concurrency: {group: deploy, cancel-in-progress: true}
          runs-on: self-hosted
          steps:
            - run: ./deploy
      """

      first = run_workflow(repo, yaml)
      {:ok, _} = Workflows.update_job_status(job(first), "running")

      second = run_workflow(repo, yaml)
      assert job(second).status == "queued"

      assert %WorkflowJob{
               status: "cancelled",
               status_reason: "Cancelled by a newer job in deploy"
             } =
               job(first)

      # Superseded, not failed.
      assert run(first).status == "cancelled"
    end

    test "cancelling the run of the job in progress lets the waiting one go", %{repo: repo} do
      yaml = """
      name: ci
      on: push
      jobs:
        deploy:
          concurrency: deploy
          runs-on: self-hosted
          steps:
            - run: ./deploy
      """

      first = run_workflow(repo, yaml)
      second = run_workflow(repo, yaml)
      assert job(second).status == "waiting"

      {:ok, _} = Workflows.cancel(run(first))
      assert %WorkflowJob{status: "queued", status_reason: nil} = job(second)
    end

    test "re-running a job puts it back through its group", %{repo: repo} do
      yaml = """
      name: ci
      on: push
      jobs:
        deploy:
          concurrency: deploy
          runs-on: self-hosted
          steps:
            - run: ./deploy
      """

      first = run_workflow(repo, yaml)
      {:ok, _} = Workflows.update_job_status(job(first), "failure")

      second = run_workflow(repo, yaml)
      assert job(second).status == "queued"

      {:ok, _} = Workflows.rerun_job(run(first), "deploy")
      assert %WorkflowJob{status: "waiting", attempt: 2} = job(first)
    end
  end
end