neiam /gitgud

defmodule GitGud.Workflows.MatrixDispatchTest do
  @moduledoc """
  Matrix builds end to end through the orchestration: a workflow with a
  matrix becomes several jobs, each routed to a runner that carries the
  labels that leg asked for.

  The motivating case is two architectures — an arm64 machine and an
  amd64 one, both feeding one workflow — which is also the case that
  exercises every moving part: expansion, `runs-on` substitution, and
  label-set claiming.
  """
  use GitGud.DataCase, async: false

  alias GitGud.Workflows
  alias GitGud.Workflows.Runners
  alias GitGud.Workflows.WorkflowJob
  alias GitGud.Workflows.WorkflowRun
  alias GitGud.Workflows.Yaml

  import GitGud.ForgeFixtures

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

  # `dispatch/2` wants a real git tree to read workflows out of, and the
  # tree isn't what's under test. `rerun/1` seeds jobs from a stored
  # `workflow_doc` through exactly the same creation path, so a seed row
  # plus a re-run gets the real thing without a repository on disk.
  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

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

  defp runner(repo, labels) do
    {:ok, runner, _uuid, _token} =
      Runners.create_offline(
        %{name: "r-#{System.unique_integer([:positive])}", labels: %{"labels" => labels}},
        repo
      )

    runner
  end

  describe "expansion" do
    test "a two-architecture matrix becomes two jobs", %{repo: repo} do
      run =
        run_workflow(repo, """
        name: ci
        on: push
        jobs:
          build:
            strategy:
              matrix:
                arch: [amd64, arm64]
            runs-on: [self-hosted, "${{ matrix.arch }}"]
            steps:
              - run: echo building for ${{ matrix.arch }}
        """)

      assert [amd, arm] = jobs(run)

      # One YAML job id, two rows — `needs: build` still names the group.
      assert amd.job_id == "build"
      assert arm.job_id == "build"

      assert amd.name == "build (amd64)"
      assert arm.name == "build (arm64)"

      assert amd.matrix == %{"arch" => "amd64"}
      assert arm.matrix == %{"arch" => "arm64"}
    end

    test "`runs-on` is resolved per leg, because it decides the runner", %{repo: repo} do
      run =
        run_workflow(repo, """
        name: ci
        on: push
        jobs:
          build:
            strategy:
              matrix:
                arch: [amd64, arm64]
            runs-on: [self-hosted, "${{ matrix.arch }}"]
            steps:
              - run: true
        """)

      assert [amd, arm] = jobs(run)

      assert amd.runs_on == %{"labels" => ["self-hosted", "amd64"]}
      assert arm.runs_on == %{"labels" => ["self-hosted", "arm64"]}
    end

    test "each leg's stored definition pins the matrix to one value", %{repo: repo} do
      run =
        run_workflow(repo, """
        name: ci
        on: push
        jobs:
          build:
            strategy:
              matrix:
                arch: [amd64, arm64]
            runs-on: self-hosted
            steps:
              - run: echo ${{ matrix.arch }}
        """)

      assert [amd, _arm] = jobs(run)

      # One value, so the runner's own engine expands it to exactly one
      # job and `${{ matrix.arch }}` in the step still resolves. Left
      # whole, the runner would fan out again inside this one job.
      assert amd.definition["strategy"] == %{"matrix" => %{"arch" => ["amd64"]}}
    end

    test "a job with no matrix is still one job, and keeps its name", %{repo: repo} do
      run =
        run_workflow(repo, """
        name: ci
        on: push
        jobs:
          test:
            name: unit tests
            runs-on: ubuntu-latest
            steps:
              - run: true
        """)

      assert [job] = jobs(run)
      assert job.name == "unit tests"
      assert job.matrix == nil
      assert job.runs_on == %{"labels" => ["ubuntu-latest"]}
      refute Map.has_key?(job.definition, "strategy")
    end

    test "steps are created for every leg", %{repo: repo} do
      run =
        run_workflow(repo, """
        name: ci
        on: push
        jobs:
          build:
            strategy:
              matrix:
                arch: [amd64, arm64]
            runs-on: self-hosted
            steps:
              - run: one
              - run: two
        """)

      for job <- jobs(run) do
        assert length(Repo.preload(job, :steps).steps) == 2
      end
    end
  end

  describe "routing to the right architecture" do
    setup %{repo: repo} do
      run =
        run_workflow(repo, """
        name: ci
        on: push
        jobs:
          build:
            strategy:
              matrix:
                arch: [amd64, arm64]
            runs-on: [self-hosted, "${{ matrix.arch }}"]
            steps:
              - run: true
        """)

      {:ok, run: run}
    end

    test "an arm64 runner claims the arm64 leg and only that one", %{repo: repo, run: _run} do
      arm = runner(repo, ["self-hosted", "arm64"])

      assert %WorkflowJob{} = claimed = Workflows.claim_next_job(arm)
      assert claimed.matrix == %{"arch" => "arm64"}

      # Nothing else here is for arm64.
      assert Workflows.claim_next_job(arm) == nil
    end

    test "two runners of different architectures take one leg each", %{repo: repo} do
      amd = runner(repo, ["self-hosted", "amd64"])
      arm = runner(repo, ["self-hosted", "arm64"])

      assert amd_job = Workflows.claim_next_job(amd)
      assert arm_job = Workflows.claim_next_job(arm)

      assert amd_job.matrix == %{"arch" => "amd64"}
      assert arm_job.matrix == %{"arch" => "arm64"}
      refute amd_job.id == arm_job.id

      # Both legs are spoken for.
      assert Workflows.claim_next_job(amd) == nil
      assert Workflows.claim_next_job(arm) == nil
    end

    test "a runner missing one of the required labels claims nothing", %{repo: repo} do
      # Carries `self-hosted` but neither architecture.
      idle = runner(repo, ["self-hosted", "riscv64"])

      assert Workflows.claim_next_job(idle) == nil
    end

    test "a runner with extra labels still qualifies — containment, not equality", %{repo: repo} do
      beefy = runner(repo, ["self-hosted", "arm64", "linux", "gpu"])

      assert claimed = Workflows.claim_next_job(beefy)
      assert claimed.matrix == %{"arch" => "arm64"}
    end
  end

  describe "routing without a matrix" do
    # Its own block: the matrix setup above queues work for both
    # architectures, which is exactly what a "this runner should find
    # nothing" assertion cannot have sitting around.
    test "a scalar runs-on still routes, which is every pre-existing workflow", %{repo: repo} do
      run =
        run_workflow(repo, """
        name: ci
        on: push
        jobs:
          plain:
            runs-on: ubuntu-latest
            steps:
              - run: true
        """)

      wrong = runner(repo, ["self-hosted", "arm64"])
      right = runner(repo, ["ubuntu-latest"])

      assert Workflows.claim_next_job(wrong) == nil

      assert claimed = Workflows.claim_next_job(right)
      assert claimed.workflow_run_id == run.id
    end

    test "`runs-on: self-hosted` matches any self-registered runner", %{repo: repo} do
      run_workflow(repo, """
      name: ci
      on: push
      jobs:
        anywhere:
          runs-on: self-hosted
          steps:
            - run: true
      """)

      # Doesn't carry the `self-hosted` label itself; it's implied.
      any = runner(repo, ["arm64"])

      assert claimed = Workflows.claim_next_job(any)
      assert claimed.job_id == "anywhere"
    end
  end

  describe "routing by architecture" do
    defp aarch64_runner(repo, labels) do
      {:ok, runner} = repo |> runner(labels) |> Runners.set_architecture("aarch64")
      runner
    end

    defp self_hosted_job(repo) do
      run_workflow(repo, """
      name: ci
      on: push
      jobs:
        anywhere:
          runs-on: self-hosted
          steps:
            - run: true
      """)
    end

    test "a runner is x86_64 unless set otherwise", %{repo: repo} do
      assert runner(repo, ["ubuntu-latest"]).architecture == "x86_64"
    end

    test "an aarch64 runner doesn't take `runs-on: self-hosted` work", %{repo: repo} do
      self_hosted_job(repo)

      arm = aarch64_runner(repo, ["ubuntu-24.04-arm"])
      assert Workflows.claim_next_job(arm) == nil

      # It's still there for an x86_64 runner.
      assert %WorkflowJob{job_id: "anywhere"} =
               Workflows.claim_next_job(runner(repo, ["ubuntu-latest"]))
    end

    test "an aarch64 runner takes the work its labels name", %{repo: repo} do
      run_workflow(repo, """
      name: ci
      on: push
      jobs:
        arm:
          runs-on: ubuntu-24.04-arm
          steps:
            - run: true
      """)

      arm = aarch64_runner(repo, ["ubuntu-24.04-arm"])
      assert %WorkflowJob{job_id: "arm"} = Workflows.claim_next_job(arm)
    end

    test "an aarch64 runner labelled self-hosted still leaves bare self-hosted work", %{
      repo: repo
    } do
      self_hosted_job(repo)

      arm = aarch64_runner(repo, ["self-hosted", "runner-aarch64"])
      assert Workflows.claim_next_job(arm) == nil

      run_workflow(repo, """
      name: arm
      on: push
      jobs:
        arm:
          runs-on: [self-hosted, runner-aarch64]
          steps:
            - run: true
      """)

      assert %WorkflowJob{job_id: "arm"} = Workflows.claim_next_job(arm)

      assert %WorkflowJob{job_id: "anywhere"} =
               Workflows.claim_next_job(runner(repo, ["ubuntu-latest"]))
    end

    test "the architecture is a label", %{repo: repo} do
      run_workflow(repo, """
      name: ci
      on: push
      jobs:
        build:
          strategy:
            matrix:
              arch: [x86_64, aarch64]
          runs-on: [self-hosted, "${{ matrix.arch }}"]
          steps:
            - run: true
      """)

      x86 = runner(repo, ["ubuntu-latest"])
      arm = aarch64_runner(repo, ["self-hosted"])

      assert Workflows.claim_next_job(x86).matrix == %{"arch" => "x86_64"}
      assert Workflows.claim_next_job(arm).matrix == %{"arch" => "aarch64"}
      assert Workflows.claim_next_job(x86) == nil
    end

    test "an unknown architecture is refused", %{repo: repo} do
      assert {:error, cs} =
               repo |> runner(["ubuntu-latest"]) |> Runners.set_architecture("riscv64")

      assert %{architecture: [_]} = errors_on(cs)
    end
  end

  describe "dependencies across a matrix" do
    setup %{repo: repo} do
      run =
        run_workflow(repo, """
        name: ci
        on: push
        jobs:
          build:
            strategy:
              matrix:
                arch: [amd64, arm64]
            runs-on: [self-hosted, "${{ matrix.arch }}"]
            steps:
              - run: true
          release:
            needs: build
            runs-on: self-hosted
            steps:
              - run: true
        """)

      {:ok, run: run}
    end

    test "a downstream job waits for every leg, not the first", %{run: run} do
      [amd, arm, release] = jobs(run)

      assert release.status == "blocked"

      {:ok, _} = Workflows.update_job_status(amd, "success")
      assert Repo.get!(WorkflowJob, release.id).status == "blocked"

      {:ok, _} = Workflows.update_job_status(arm, "success")
      assert Repo.get!(WorkflowJob, release.id).status == "queued"
    end

    test "one failing leg skips the downstream job", %{run: run} do
      [amd, arm, release] = jobs(run)

      {:ok, _} = Workflows.update_job_status(amd, "success")
      {:ok, _} = Workflows.update_job_status(arm, "failure")

      assert Repo.get!(WorkflowJob, release.id).status == "skipped"
    end

    test "the snapshot collapses the legs into one name", %{run: run} do
      [amd, arm, release] = jobs(run)

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

      # Downstream `${{ needs.build }}` is one name however many
      # machines produced it.
      assert %{"build" => %{"status" => "success"}} =
               Repo.get!(WorkflowJob, release.id).needs_snapshot
    end
  end

  describe "fail-fast" do
    test "a failing leg cancels its queued siblings", %{repo: repo} do
      run =
        run_workflow(repo, """
        name: ci
        on: push
        jobs:
          build:
            strategy:
              matrix:
                arch: [amd64, arm64, riscv64]
            runs-on: [self-hosted, "${{ matrix.arch }}"]
            steps:
              - run: true
        """)

      [first | rest] = jobs(run)

      {:ok, _} = Workflows.update_job_status(first, "failure")

      for sibling <- rest do
        assert Repo.get!(WorkflowJob, sibling.id).status == "cancelled"
      end
    end

    test "`fail-fast: false` lets the others finish", %{repo: repo} do
      run =
        run_workflow(repo, """
        name: ci
        on: push
        jobs:
          build:
            strategy:
              fail-fast: false
              matrix:
                arch: [amd64, arm64, riscv64]
            runs-on: [self-hosted, "${{ matrix.arch }}"]
            steps:
              - run: true
        """)

      [first | rest] = jobs(run)

      {:ok, _} = Workflows.update_job_status(first, "failure")

      for sibling <- rest do
        assert Repo.get!(WorkflowJob, sibling.id).status == "queued"
      end
    end

    test "a leg already running is cancelled too", %{repo: repo} do
      run =
        run_workflow(repo, """
        name: ci
        on: push
        jobs:
          build:
            strategy:
              matrix:
                arch: [amd64, arm64]
            runs-on: self-hosted
            steps:
              - run: true
        """)

      [first, second] = jobs(run)

      {:ok, _} = Workflows.update_job_status(second, "running")
      {:ok, _} = Workflows.update_job_status(first, "failure")

      # As on GitHub. Its runner hears it on the next UpdateTask, and
      # still uploads the log it has.
      assert Repo.get!(WorkflowJob, second.id).status == "cancelled"
    end

    test "a job with no matrix never fail-fasts its unrelated siblings", %{repo: repo} do
      run =
        run_workflow(repo, """
        name: ci
        on: push
        jobs:
          a:
            runs-on: self-hosted
            steps:
              - run: true
          b:
            runs-on: self-hosted
            steps:
              - run: true
        """)

      [a, b] = jobs(run)

      {:ok, _} = Workflows.update_job_status(a, "failure")

      assert Repo.get!(WorkflowJob, b.id).status == "queued"
    end
  end

  describe "re-running" do
    test "re-running a matrix job re-runs every leg, not the first one found", %{repo: repo} do
      run =
        run_workflow(repo, """
        name: ci
        on: push
        jobs:
          build:
            strategy:
              fail-fast: false
              matrix:
                arch: [amd64, arm64]
            runs-on: self-hosted
            steps:
              - run: true
        """)

      [amd, arm] = jobs(run)
      {:ok, _} = Workflows.update_job_status(amd, "failure")
      {:ok, _} = Workflows.update_job_status(arm, "success")

      assert {:ok, _} = Workflows.rerun_job(run, "build")

      assert Repo.get!(WorkflowJob, amd.id).status == "queued"
      assert Repo.get!(WorkflowJob, arm.id).status == "queued"
    end

    test "an unknown job id is still not found", %{repo: repo} do
      run =
        run_workflow(repo, """
        name: ci
        on: push
        jobs:
          build:
            runs-on: self-hosted
            steps:
              - run: true
        """)

      assert {:error, :not_found} = Workflows.rerun_job(run, "nope")
    end
  end

  describe "the run as a whole" do
    test "completes only once every leg is terminal", %{repo: repo} do
      run =
        run_workflow(repo, """
        name: ci
        on: push
        jobs:
          build:
            strategy:
              fail-fast: false
              matrix:
                arch: [amd64, arm64]
            runs-on: self-hosted
            steps:
              - run: true
        """)

      [amd, arm] = jobs(run)

      {:ok, _} = Workflows.update_job_status(amd, "success")
      assert Repo.get!(WorkflowRun, run.id).status in ~w(queued running)

      {:ok, _} = Workflows.update_job_status(arm, "success")
      assert Repo.get!(WorkflowRun, run.id).status == "success"
    end
  end

  describe "a matrix computed by an earlier job" do
    # The failures log a warning, as they should.
    @describetag :capture_log

    setup %{repo: repo} do
      run =
        run_workflow(repo, """
        name: ci
        on: push
        jobs:
          setup:
            runs-on: self-hosted
            steps:
              - run: true
          build:
            needs: setup
            strategy:
              matrix: ${{ fromJSON(needs.setup.outputs.m) }}
            runs-on: self-hosted
            steps:
              - run: echo ${{ matrix.os }}
          deploy:
            needs: build
            runs-on: self-hosted
            steps:
              - run: true
        """)

      {:ok, run: run}
    end

    defp finish_setup(run, outputs) do
      setup = Enum.find(jobs(run), &(&1.job_id == "setup"))
      {:ok, setup} = Workflows.set_outputs(setup, outputs)
      {:ok, _} = Workflows.update_job_status(setup, "success")
    end

    test "waits as one placeholder, then becomes a leg per combination", %{run: run} do
      assert [placeholder] = Enum.filter(jobs(run), &(&1.job_id == "build"))
      assert placeholder.name == "build"
      assert placeholder.status == "blocked"
      assert placeholder.matrix == nil

      finish_setup(run, %{"m" => ~s({"os":["a","b"]})})

      refute Repo.get(WorkflowJob, placeholder.id)

      legs = Enum.filter(jobs(run), &(&1.job_id == "build"))
      assert Enum.map(legs, & &1.matrix) == [%{"os" => "a"}, %{"os" => "b"}]
      assert Enum.map(legs, & &1.name) == ["build (a)", "build (b)"]
      assert Enum.all?(legs, &(&1.status == "queued"))
      assert Enum.all?(legs, &(&1.needs_snapshot["setup"]["status"] == "success"))

      assert Enum.all?(
               legs,
               &(&1.definition["strategy"]["matrix"] == %{"os" => [&1.matrix["os"]]})
             )

      assert Enum.all?(legs, &(length(Repo.preload(&1, :steps).steps) == 1))

      # The placeholder's pending status went with it; each leg has its own.
      contexts =
        run.repository_id
        |> then(
          &Workflows.list_commit_statuses(%GitGud.Repositories.Repository{id: &1}, run.head_sha)
        )
        |> Enum.map(& &1.context)

      refute "ci / build (push)" in contexts
      assert "ci / build (a) (push)" in contexts
    end

    test "a job needing it waits for every leg", %{run: run} do
      finish_setup(run, %{"m" => ~s({"os":["a","b"]})})

      [a, b] = Enum.filter(jobs(run), &(&1.job_id == "build"))
      deploy = Enum.find(jobs(run), &(&1.job_id == "deploy"))

      {:ok, _} = Workflows.update_job_status(a, "success")
      assert Repo.get!(WorkflowJob, deploy.id).status == "blocked"

      {:ok, _} = Workflows.update_job_status(b, "success")
      assert Repo.get!(WorkflowJob, deploy.id).status == "queued"
    end

    test "an output that isn't a matrix fails the job, saying why", %{run: run} do
      finish_setup(run, %{"m" => "nope"})

      [build] = Enum.filter(jobs(run), &(&1.job_id == "build"))
      assert build.status == "failure"
      assert build.status_reason =~ "could not be evaluated"

      [step | _] =
        build |> Repo.preload(:steps) |> Map.fetch!(:steps) |> Enum.sort_by(& &1.number)

      assert [%{body: "Error: " <> _}] = Workflows.list_step_logs(step)

      assert Enum.find(jobs(run), &(&1.job_id == "deploy")).status == "skipped"
      assert Repo.get!(WorkflowRun, run.id).status == "failure"
    end

    test "re-running the failed job expands it again", %{run: run} do
      finish_setup(run, %{"m" => "nope"})

      setup = Enum.find(jobs(run), &(&1.job_id == "setup"))
      {:ok, _} = Workflows.set_outputs(setup, %{"m" => ~s({"os":["a"]})})
      {:ok, _} = Workflows.rerun_job(Repo.get!(WorkflowRun, run.id), "build")

      assert [%{matrix: %{"os" => "a"}, status: "queued"}] =
               Enum.filter(jobs(run), &(&1.job_id == "build"))

      assert Repo.get!(WorkflowRun, run.id).status == "running"
    end
  end

  describe "a matrix computed without needs" do
    test "is expanded as the run is created", %{repo: repo} do
      run =
        run_workflow(repo, """
        name: ci
        on: push
        jobs:
          build:
            strategy:
              matrix:
                os: ${{ fromJSON('["a", "b"]') }}
                ref: ["${{ github.ref_name }}"]
            runs-on: self-hosted
            steps:
              - run: true
        """)

      assert Enum.map(jobs(run), & &1.matrix) == [
               %{"os" => "a", "ref" => "main"},
               %{"os" => "b", "ref" => "main"}
             ]
    end

    @tag :capture_log
    test "one that excludes everything fails the run", %{repo: repo} do
      run =
        run_workflow(repo, """
        name: ci
        on: push
        jobs:
          build:
            strategy:
              matrix:
                os: [a]
                exclude:
                  - os: a
            runs-on: self-hosted
            steps:
              - run: true
        """)

      assert [%{status: "failure", status_reason: reason}] = jobs(run)
      assert reason =~ "no combinations"
      assert Repo.get!(WorkflowRun, run.id).status == "failure"
    end

    @tag :capture_log
    test "re-running one that excludes everything fails it again, unclaimed", %{repo: repo} do
      run =
        run_workflow(repo, """
        name: ci
        on: push
        jobs:
          build:
            strategy:
              matrix:
                os: [a]
                exclude:
                  - os: a
            runs-on: self-hosted
            steps:
              - run: true
        """)

      {:ok, _} = Workflows.rerun_job(Repo.get!(WorkflowRun, run.id), "build")

      assert [%{status: "failure", attempt: 2, status_reason: reason}] = jobs(run)
      assert reason =~ "no combinations"
      assert Workflows.claim_next_job(runner(repo, ["self-hosted"])) == nil
      assert Repo.get!(WorkflowRun, run.id).status == "failure"
    end
  end

  describe "max-parallel" do
    test "caps the legs of a job running at once", %{repo: repo} do
      run =
        run_workflow(repo, """
        name: ci
        on: push
        jobs:
          deploy:
            strategy:
              max-parallel: 1
              matrix:
                env: [a, b, c]
            runs-on: self-hosted
            steps:
              - run: true
        """)

      [r1, r2, r3] = for _ <- 1..3, do: runner(repo, ["self-hosted"])

      assert %WorkflowJob{} = first = Workflows.claim_next_job(r1)
      assert Workflows.claim_next_job(r2) == nil
      assert Workflows.claim_jobs_for(r3, 3) == []

      [_, second, _] = jobs(run)
      assert Workflows.claim_job_by_handle(r2, Integer.to_string(second.id)) == nil

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

      assert %WorkflowJob{} = Workflows.claim_next_job(r2)
      assert Workflows.claim_next_job(r3) == nil
    end

    test "another job's legs are still claimable meanwhile", %{repo: repo} do
      run_workflow(repo, """
      name: ci
      on: push
      jobs:
        deploy:
          strategy:
            max-parallel: 1
            matrix:
              env: [a, b]
          runs-on: self-hosted
          steps:
            - run: true
        test:
          runs-on: self-hosted
          steps:
            - run: true
      """)

      r = runner(repo, ["self-hosted"])
      claimed = Workflows.claim_jobs_for(r, 3)

      assert claimed |> Enum.map(& &1.job_id) |> Enum.sort() == ["deploy", "test"]
    end
  end
end