defmodule GitGud.Workflows do
@moduledoc """
CI workflow orchestration.
Push triggers `dispatch/2`, which scans `.github/workflows/*.yml` and
`.forgejo/workflows/*.yml` in the pushed tree, parses each, and
enqueues a `WorkflowRun` per workflow that matches the event.
Runners pull jobs via `claim_next_job/2`. The runner protocol layer
(Forgejo Actions Twirp) wraps this with token auth and protobuf
framing — but the orchestration core lives here.
"""
import Ecto.Query, warn: false
alias GitGud.Git
alias GitGud.Repo
alias GitGud.Repositories
alias GitGud.Repositories.Repository
alias GitGud.Workflows.Runner
alias GitGud.Workflows.WorkflowJob
alias GitGud.Workflows.WorkflowRun
alias GitGud.Workflows.WorkflowStep
alias GitGud.Workflows.Yaml
@workflow_dirs ~w(.github/workflows .forgejo/workflows)
# ── dispatch ─────────────────────────────────────────────────────────
@doc """
Inspect a push and create the workflow runs/jobs/steps it triggers.
`event` is `%{trigger: "push" | "pull_request", head_sha: <20 bytes>,
ref: "refs/heads/main"}`.
Returns `{:ok, [run]}`.
"""
def dispatch(%Repository{} = repo, event) do
%{trigger: trigger, head_sha: head_sha, ref: ref} = event
triggered_by_id = Map.get(event, :triggered_by_id)
branch = branch_from_ref(ref)
ctx = %{branch: branch}
workflows = discover_workflows(repo, head_sha)
runs =
for {path, doc} <- workflows, Yaml.matches?(doc, trigger, ctx) do
{:ok, run} =
create_run(repo, %{
workflow_path: path,
workflow_name: doc.name,
trigger: trigger,
head_sha: head_sha,
head_ref: ref,
workflow_doc: doc.raw,
triggered_by_id: triggered_by_id,
jobs: doc.jobs
})
run
end
{:ok, runs}
end
defp branch_from_ref("refs/heads/" <> b), do: b
defp branch_from_ref(_), do: nil
@doc """
Walk the tree at `head_sha`, returning `[{workflow_path, parsed_doc}]`.
Workflows that fail to parse are dropped silently for now; in a later
pass we'll surface them as failed runs with a parse-error annotation.
"""
def discover_workflows(%Repository{disk_path: path} = repo, head_sha) do
with {:ok, commit} <- Repositories.get_commit(repo, head_sha) do
@workflow_dirs
|> Enum.flat_map(&list_yaml_files(path, commit.tree_sha, &1))
|> Enum.flat_map(fn {workflow_path, blob_sha} ->
case Git.blob_bytes(path, blob_sha) do
{:ok, bytes} ->
case Yaml.parse(bytes) do
{:ok, doc} -> [{workflow_path, doc}]
_ -> []
end
_ ->
[]
end
end)
else
_ -> []
end
end
defp list_yaml_files(repo_path, tree_sha, dir) do
case Git.tree_entry_at(repo_path, tree_sha, dir) do
{:ok, %{type: "tree", sha: subtree}} ->
case Git.tree(repo_path, subtree) do
{:ok, entries} ->
entries
|> Enum.filter(&yaml_file?/1)
|> Enum.map(fn e -> {Path.join(dir, e.name), e.sha} end)
_ ->
[]
end
_ ->
[]
end
end
defp yaml_file?(%{type: "blob", name: name}),
do: String.ends_with?(name, ".yml") or String.ends_with?(name, ".yaml")
defp yaml_file?(_), do: false
# ── run + jobs + steps creation ──────────────────────────────────────
defp create_run(%Repository{id: rid}, %{jobs: jobs} = attrs) do
Repo.transaction(fn ->
number = next_run_number(rid)
run_attrs =
attrs
|> Map.delete(:jobs)
|> Map.put(:number, number)
run =
%WorkflowRun{repository_id: rid}
|> WorkflowRun.create_changeset(run_attrs)
|> Repo.insert!()
for job <- jobs, do: create_job(run, job)
# Initial pending commit status so the SCM view shows the run
# is in flight before the first claim lands.
update_run_commit_status(run)
run
end)
end
defp next_run_number(rid) do
(from(r in WorkflowRun,
where: r.repository_id == ^rid,
select: max(r.number)
)
|> Repo.one() || 0) + 1
end
defp create_job(%WorkflowRun{id: rid}, %{} = job) do
# Jobs with unresolved `needs:` start in `blocked` so claim_next_job
# can never pick them up before their prerequisites complete.
# Promotion to `queued` happens in `resolve_dependents/1`.
initial_status = if job.needs == [], do: "queued", else: "blocked"
job_row =
%WorkflowJob{workflow_run_id: rid}
|> WorkflowJob.create_changeset(%{
job_id: job.id,
name: job.name,
runs_on: %{"value" => job.runs_on},
needs: job.needs,
definition: job.raw
})
|> Ecto.Changeset.change(status: initial_status)
|> Repo.insert!()
job.steps
|> Enum.with_index(1)
|> Enum.each(fn {step, n} ->
%WorkflowStep{workflow_job_id: job_row.id}
|> WorkflowStep.create_changeset(%{
number: n,
name: Map.get(step, "name") || infer_step_name(step)
})
|> Repo.insert!()
end)
job_row
end
defp infer_step_name(%{"run" => cmd}) when is_binary(cmd) do
cmd |> String.split("\n", parts: 2) |> hd() |> String.slice(0, 80)
end
defp infer_step_name(%{"uses" => uses}) when is_binary(uses), do: uses
defp infer_step_name(_), do: nil
# ── queries ──────────────────────────────────────────────────────────
@doc """
Return the most-recent commit status row per `context` for a given
`(repository, sha)`. Multiple workflows on the same commit produce
separate rows; we return them all, ordered by context.
`sha` may be either a 20-byte binary OR a hex string — the column
is binary, so we coerce.
"""
def list_commit_statuses(%Repository{id: rid}, sha) when is_binary(sha) do
bin =
cond do
byte_size(sha) == 20 -> sha
byte_size(sha) == 40 -> Base.decode16!(sha, case: :mixed)
true -> sha
end
Repo.all(
from c in GitGud.Workflows.CommitStatus,
where: c.repository_id == ^rid and c.sha == ^bin,
order_by: [asc: c.context]
)
end
def list_commit_statuses(_, _), do: []
def list_runs(%Repository{id: rid}, opts \\ []) do
limit = Keyword.get(opts, :limit, 50)
WorkflowRun
|> where([r], r.repository_id == ^rid)
|> order_by([r], desc: r.inserted_at)
|> limit(^limit)
|> Repo.all()
end
@active_statuses ~w(queued running)
@doc "Statuses considered 'in flight' for the dashboards."
def active_statuses, do: @active_statuses
@doc """
Runs across a set of repositories — the org CI "single pane of glass".
`repo_ids` is the caller-computed set of repositories the viewer is
allowed to see (visibility policy lives in the caller, not here).
Opts:
* `:statuses` — restrict to these run statuses (default: all)
* `:limit` — cap rows (default: 100)
Each run is returned with `:repository` (and its `:owner` +
`:organization`) and `:triggered_by` preloaded for rendering.
"""
def list_runs_for_repos(repo_ids, opts \\ []) when is_list(repo_ids) do
if repo_ids == [] do
[]
else
base_runs_query(opts)
|> where([r], r.repository_id in ^repo_ids)
|> Repo.all()
end
end
@doc """
Runs a given user cares about: any run in a repository they can reach
(`repo_ids`, computed by the caller from ownership + org membership +
team grants) plus any run they personally triggered, even if the repo
later moved out of their reach.
Same opts + preloads as `list_runs_for_repos/2`.
"""
def list_runs_for_user(user_id, repo_ids, opts \\ [])
when is_integer(user_id) and is_list(repo_ids) do
base_runs_query(opts)
|> where([r], r.repository_id in ^repo_ids or r.triggered_by_id == ^user_id)
|> Repo.all()
end
defp base_runs_query(opts) do
limit = Keyword.get(opts, :limit, 100)
jobs_query = from(j in WorkflowJob, order_by: j.id)
query =
WorkflowRun
|> order_by([r], desc: r.inserted_at)
|> limit(^limit)
|> preload([:triggered_by, jobs: ^jobs_query, repository: [:owner, :organization]])
case Keyword.get(opts, :statuses) do
[_ | _] = statuses -> where(query, [r], r.status in ^statuses)
_ -> query
end
end
def get_run!(%Repository{id: rid}, number) do
WorkflowRun
|> where([r], r.repository_id == ^rid and r.number == ^number)
|> preload(jobs: :steps)
|> Repo.one!()
end
def get_job!(id), do: Repo.get!(WorkflowJob, id) |> Repo.preload(:steps)
@doc "Most recent run for a repo, or `nil` if there are none."
def latest_run(%Repository{id: rid}) do
WorkflowRun
|> where([r], r.repository_id == ^rid)
|> order_by([r], desc: r.inserted_at)
|> limit(1)
|> Repo.one()
end
@doc """
Latest CI run status for each of `repo_ids`, as
`%{repository_id => status}`. One query — for repo-list views that want
a CI status glance without an N+1. Repos with no runs are absent.
"""
def latest_run_status_by_repo(repo_ids) when is_list(repo_ids) do
from(r in WorkflowRun,
where: r.repository_id in ^repo_ids,
order_by: [asc: r.repository_id, desc: r.inserted_at],
distinct: r.repository_id,
select: {r.repository_id, r.status}
)
|> Repo.all()
|> Map.new()
end
@doc """
List artifacts attached to a run. Returns a list of
`%WorkflowArtifact{}` ordered by upload time.
"""
def list_artifacts(%WorkflowRun{id: rid}) do
GitGud.Workflows.WorkflowArtifact
|> where([a], a.workflow_run_id == ^rid)
|> order_by([a], asc: a.inserted_at)
|> Repo.all()
end
# ── runner protocol surface ──────────────────────────────────────────
@doc """
Atomically claim the next queued job whose `runs_on` label intersects
the runner's advertised labels. Returns `nil` if nothing is available.
Uses `SELECT … FOR UPDATE SKIP LOCKED` to make concurrent runner polls
cheap and correct.
"""
def claim_next_job(%Runner{} = runner) do
case effective_labels(runner) do
[] -> nil
labels -> do_claim_next_job(runner, labels)
end
end
def claim_next_job(_runner), do: nil
# Labels used for job matching = the operator-assigned `labels` UNION
# the runner-declared `agent_labels`. Without the union, a label added
# to the runner's config only lands in `agent_labels` (via Declare on
# restart) and never matches — you'd have to fully re-register. The
# union makes "edit config + restart" take effect as operators expect.
defp effective_labels(%Runner{labels: l, agent_labels: a}) do
Enum.uniq(label_list(l) ++ label_list(a))
end
defp label_list(%{"labels" => ls}) when is_list(ls), do: ls
defp label_list(_), do: []
@doc """
Claim up to `n` jobs in one batch — for runners that advertise
`task_capacity > 1` in their FetchTaskRequest. Returns a (possibly
empty) list, in claim order.
Each job is claimed in its own transaction (cheap rollback if mid-
loop encoding later fails); they all become `running` on success.
"""
def claim_jobs_for(%Runner{} = runner, n) when is_integer(n) and n > 0 do
claim_jobs_for(runner, n, [])
end
def claim_jobs_for(_runner, _n), do: []
defp claim_jobs_for(_runner, 0, acc), do: Enum.reverse(acc)
defp claim_jobs_for(runner, remaining, acc) do
case claim_next_job(runner) do
%WorkflowJob{} = job -> claim_jobs_for(runner, remaining - 1, [job | acc])
_ -> Enum.reverse(acc)
end
end
@doc """
Claim a specific job identified by `handle` — typically the
workflow_job_id encoded as a string. Used by `FetchSingleTask` for
targeted retrieval. Returns nil if no queued job matches.
We accept either a numeric string (interpreted as the WorkflowJob
pk) or a `<run_id>:<job_id>` form (yaml-level routing). Falls
back to normal `claim_next_job/1` for empty / unparseable handles.
"""
def claim_job_by_handle(%Runner{} = runner, handle)
when is_binary(handle) and handle != "" do
target_id =
case Integer.parse(handle) do
{n, ""} -> n
_ -> nil
end
cond do
is_integer(target_id) ->
do_claim_specific_job(runner, target_id)
true ->
# Unsupported handle shape; fall through to a normal claim.
claim_next_job(runner)
end
end
def claim_job_by_handle(runner, _), do: claim_next_job(runner)
defp do_claim_specific_job(%Runner{id: rid} = runner, job_id) do
labels = effective_labels(runner)
match_labels = if "self-hosted" in labels, do: labels, else: ["self-hosted" | labels]
Repo.transaction(fn ->
job =
scoped_job_query(runner)
|> where([j], j.id == ^job_id)
|> where([j], j.status == "queued")
|> where([j], fragment("(?->>'value') = ANY(?)", j.runs_on, ^match_labels))
|> lock("FOR UPDATE SKIP LOCKED")
|> Repo.one()
case job do
nil ->
nil
job ->
{:ok, claimed} = job |> WorkflowJob.claim_changeset(rid) |> Repo.update()
run = Repo.get!(WorkflowRun, claimed.workflow_run_id)
if run.status == "queued" do
run |> WorkflowRun.status_changeset("running") |> Repo.update!()
broadcast_ci_change(run.id)
end
claimed |> Repo.preload(:steps)
end
end)
|> case do
{:ok, result} -> result
_ -> nil
end
end
defp do_claim_next_job(%Runner{id: rid} = runner, labels) do
# GitHub-Actions semantics: `runs-on: self-hosted` matches any
# self-registered runner regardless of OS label. So if the runner
# was minted via `Runners.create_offline/2` we implicitly add
# `self-hosted` to the match set for the query. Operators who want
# strict label matching can still register a runner without the
# `self-hosted` label — but everything created via this codebase
# is by definition self-hosted, so the implicit wildcard is safe.
match_labels = if "self-hosted" in labels, do: labels, else: ["self-hosted" | labels]
Repo.transaction(fn ->
job =
scoped_job_query(runner)
|> where([j], j.status == "queued")
|> where([j], fragment("(?->>'value') = ANY(?)", j.runs_on, ^match_labels))
|> order_by([j], asc: j.id)
|> limit(1)
|> lock("FOR UPDATE SKIP LOCKED")
|> Repo.one()
case job do
nil ->
nil
job ->
{:ok, claimed} = job |> WorkflowJob.claim_changeset(rid) |> Repo.update()
# Promote the parent run to "running" on first claim.
run = Repo.get!(WorkflowRun, claimed.workflow_run_id)
if run.status == "queued" do
run |> WorkflowRun.status_changeset("running") |> Repo.update!()
broadcast_ci_change(run.id)
end
claimed |> Repo.preload(:steps)
end
end)
|> case do
{:ok, result} -> result
err -> err
end
end
# A runner only claims jobs in its scope:
# * repo-scoped runner → jobs in that repo
# * org-scoped runner → jobs in any repo owned by that org
# * global runner → any job
defp scoped_job_query(%Runner{repository_id: rid}) when is_integer(rid) do
from j in WorkflowJob,
join: r in WorkflowRun,
on: r.id == j.workflow_run_id,
where: r.repository_id == ^rid
end
defp scoped_job_query(%Runner{organization_id: oid}) when is_integer(oid) do
from j in WorkflowJob,
join: r in WorkflowRun,
on: r.id == j.workflow_run_id,
join: repo in Repository,
on: repo.id == r.repository_id,
where: repo.organization_id == ^oid
end
defp scoped_job_query(_global) do
from j in WorkflowJob
end
@doc """
Append log rows to a step at the given starting sequence number.
Forgejo's `UpdateLogRequest` carries an `index` field — the runner's
view of where this batch of rows starts in the step's total log. We
use that as `seq` for the first row, `index + 1` for the second, etc.
Retries from the runner with the same `index` re-send identical
rows; the `ON CONFLICT (workflow_step_id, seq) DO NOTHING` makes that
no-op (idempotent at-least-once).
`rows` is the list of `%LogRow{}` protobufs from the request.
"""
def append_log(%WorkflowStep{id: step_id}, start_seq, rows) when is_integer(start_seq) and is_list(rows) do
now = DateTime.utc_now(:second)
entries =
rows
|> Enum.with_index(start_seq)
|> Enum.map(fn {row, seq} ->
%{
workflow_step_id: step_id,
seq: seq,
body: row.content || "",
inserted_at: now
}
end)
if entries != [] do
Repo.insert_all(
"workflow_step_logs",
entries,
on_conflict: :nothing,
conflict_target: [:workflow_step_id, :seq]
)
Enum.each(entries, fn %{seq: seq, body: body} ->
broadcast({:log, step_id, seq, body})
end)
end
:ok
end
# Compatibility shim for callers that still use the old single-body
# signature (none in tree at the moment, but keeps the contract
# explicit). Computes a fresh sequence number and appends.
def append_log(%WorkflowStep{id: step_id} = step, body) when is_binary(body) do
seq = next_log_seq(step_id)
append_log(step, seq, [%{content: body}])
end
@doc """
Persisted log rows for a step, ascending by sequence. Hydrates the log
view when a step is opened (live appends arrive separately via PubSub).
Returns `%{seq, body}` maps — the shape the log stream renders.
The runner streams one flat log per job (stored on the first step, see
`ensure_log_step/1`); each step owns the slice `[log_index, log_index +
log_length)` of it. When those offsets are known we return that slice;
otherwise we fall back to rows stored directly under this step (the
whole stream for the holder step, empty for the rest — the old
behavior, e.g. while a step is still running).
"""
def list_step_logs(%WorkflowStep{workflow_job_id: jid, log_index: idx, log_length: len})
when is_integer(idx) and is_integer(len) and len > 0 do
holder_id = log_holder_id(jid)
upper = idx + len
from(l in "workflow_step_logs",
where: l.workflow_step_id == ^holder_id and l.seq >= ^idx and l.seq < ^upper,
order_by: [asc: l.seq],
select: %{seq: l.seq, body: l.body}
)
|> Repo.all()
end
def list_step_logs(%WorkflowStep{id: id}) do
from(l in "workflow_step_logs",
where: l.workflow_step_id == ^id,
order_by: [asc: l.seq],
select: %{seq: l.seq, body: l.body}
)
|> Repo.all()
end
# Step that physically holds the job's flat log stream — the lowest
# `number`, matching `ensure_log_step/1`.
defp log_holder_id(job_id) do
Repo.one(
from s in WorkflowStep,
where: s.workflow_job_id == ^job_id,
order_by: [asc: s.number],
limit: 1,
select: s.id
)
end
defp next_log_seq(step_id) do
max_seq =
Repo.one(
from(l in "workflow_step_logs",
where: l.workflow_step_id == ^step_id,
select: max(l.seq)
)
) || -1
max_seq + 1
end
@doc """
Look up a previously-claimed job by `(runner_id, request_key)`. If
the runner retries FetchTask with the same idempotency key (because
our response was dropped in flight), we re-return the same job
instead of orphaning the prior claim.
Returns the WorkflowJob row if a non-expired recovery row exists
AND the job is still `running` and assigned to this runner.
Otherwise nil.
"""
def recover_claimed_job(%Runner{id: rid}, key) when is_binary(key) and key != "" do
now = DateTime.utc_now(:second)
Repo.one(
from rec in "runner_request_recovery",
join: j in WorkflowJob,
on: j.id == rec.workflow_job_id,
where:
rec.runner_id == ^rid and rec.request_key == ^key and rec.expires_at > ^now and
j.runner_id == ^rid and j.status == "running",
select: j.id
)
|> case do
nil -> nil
jid -> Repo.get(WorkflowJob, jid) |> Repo.preload(:steps)
end
end
def recover_claimed_job(_, _), do: nil
@doc """
Stamp the (runner_id, request_key) -> job mapping so a retry of the
same FetchTask gets the same job back.
`ttl_seconds` defaults to 5 minutes — long enough for any
intermittent network blip, short enough that stale rows don't
accumulate.
"""
def stamp_request_recovery(runner, key, job, ttl_seconds \\ 300)
def stamp_request_recovery(%Runner{id: rid}, key, %WorkflowJob{id: jid}, ttl_seconds)
when is_binary(key) and key != "" do
expires_at = DateTime.utc_now(:second) |> DateTime.add(ttl_seconds, :second)
Repo.insert_all(
"runner_request_recovery",
[
%{
runner_id: rid,
request_key: key,
workflow_job_id: jid,
expires_at: expires_at
}
],
on_conflict: {:replace, [:workflow_job_id, :expires_at]},
conflict_target: [:runner_id, :request_key]
)
:ok
end
def stamp_request_recovery(_runner, _key, _job, _ttl_seconds), do: :ok
@doc """
The WorkflowJob currently being executed by the given runner.
Returns nil if there's no running claim.
"""
def current_job_for_runner(%Runner{id: rid}) do
Repo.one(
from j in WorkflowJob,
where: j.runner_id == ^rid and j.status == "running",
order_by: [desc: j.started_at],
limit: 1
)
end
def current_job_for_runner(_), do: nil
@doc """
Pick the canonical WorkflowStep for log streaming for the given job.
Forgejo-runner's `UpdateLog` doesn't carry per-step routing — all log
rows for a job arrive on one stream. We funnel them into the first
step (`number == 0`) of the job. If the job has no steps yet (e.g.,
the workflow YAML didn't materialize them at parse time), create a
synthetic step so logs land somewhere queryable.
"""
def ensure_log_step(%WorkflowJob{id: jid}) do
case Repo.one(
from s in WorkflowStep,
where: s.workflow_job_id == ^jid,
order_by: [asc: s.number],
limit: 1
) do
%WorkflowStep{} = step ->
step
nil ->
%WorkflowStep{workflow_job_id: jid}
|> WorkflowStep.create_changeset(%{number: 0, name: "log"})
|> Repo.insert!()
end
end
@doc """
Merge `new_outputs` (a `%{string => string}` map) into the job's
stored outputs. Upstream forgejo enforces 255-char key / 1MB value
limits; we apply the same so a hostile workflow can't unbounded-grow
the JSONB column.
"""
def set_outputs(%WorkflowJob{} = job, new_outputs) when is_map(new_outputs) do
sanitized =
new_outputs
|> Enum.filter(fn
{k, v} when is_binary(k) and is_binary(v) ->
byte_size(k) <= 255 and byte_size(v) <= 1_048_576
_ ->
false
end)
|> Map.new()
merged = Map.merge(job.outputs || %{}, sanitized)
job
|> Ecto.Changeset.change(outputs: merged)
|> Repo.update()
end
@doc """
Apply runner-reported step states to our WorkflowStep rows.
The runner sends a `StepState` per step in `TaskState.steps` —
identified by a 0-based `id` (its execution-order index). We
create steps with `number = 1..n` at dispatch time, so the
mapping is `runner_id + 1 == our number`. (Some runner versions
use 1-based; we also try a direct `number == runner_id` match.)
Per-step we update status (from the `Result` enum), conclusion,
and timings — and broadcast each change so the workflow LiveView
can light up steps in real time as the runner progresses.
"""
def apply_step_states(%WorkflowJob{id: job_id}, step_states) when is_list(step_states) do
steps =
from(s in WorkflowStep, where: s.workflow_job_id == ^job_id)
|> Repo.all()
by_number = Map.new(steps, fn s -> {s.number, s} end)
Enum.each(step_states, fn st ->
target = Map.get(by_number, st.id + 1) || Map.get(by_number, st.id)
case target do
nil ->
:ok
%WorkflowStep{} = step ->
status = step_result_to_status(st.result)
changes =
%{status: status}
|> maybe_put(:conclusion, step_status_conclusion(status))
|> maybe_put(:started_at, timestamp_to_dt(st.started_at))
|> maybe_put(:completed_at, timestamp_to_dt(st.stopped_at))
|> maybe_put_log_offsets(st)
{:ok, updated} =
step
|> Ecto.Changeset.change(changes)
|> Repo.update()
broadcast({:step_status, job_id, updated})
end
end)
:ok
end
def apply_step_states(_job, _), do: :ok
# The runner reports each step's slice of the job's flat log as
# [log_index, log_index + log_length). Only persist when the step
# actually has lines, so a 0-length intermediate update doesn't clobber
# offsets set by an earlier one.
defp maybe_put_log_offsets(changes, %{log_length: len, log_index: idx})
when is_integer(len) and len > 0 and is_integer(idx) do
changes |> Map.put(:log_index, idx) |> Map.put(:log_length, len)
end
defp maybe_put_log_offsets(changes, _), do: changes
defp step_result_to_status(:RESULT_SUCCESS), do: "success"
defp step_result_to_status(:RESULT_FAILURE), do: "failure"
defp step_result_to_status(:RESULT_CANCELLED), do: "cancelled"
defp step_result_to_status(:RESULT_SKIPPED), do: "skipped"
# `RESULT_UNSPECIFIED` with a `started_at` timestamp means the
# step is currently running.
defp step_result_to_status(_), do: "running"
defp step_status_conclusion(status) when status in ~w(success failure cancelled skipped),
do: status
defp step_status_conclusion(_), do: nil
defp maybe_put(map, _key, nil), do: map
defp maybe_put(map, key, value), do: Map.put(map, key, value)
defp timestamp_to_dt(nil), do: nil
defp timestamp_to_dt(%Google.Protobuf.Timestamp{seconds: 0, nanos: 0}), do: nil
defp timestamp_to_dt(%Google.Protobuf.Timestamp{seconds: secs}) when is_integer(secs) do
DateTime.from_unix!(secs) |> DateTime.truncate(:second)
end
defp timestamp_to_dt(_), do: nil
def update_job_status(%WorkflowJob{} = job, status, opts \\ []) do
{:ok, updated} =
job |> WorkflowJob.status_changeset(status, opts) |> Repo.update()
broadcast({:job_status, job.workflow_run_id, updated})
if WorkflowJob.terminal?(updated.status) do
finalize_steps(updated)
resolve_dependents(updated)
end
maybe_complete_run(updated)
{:ok, updated}
end
# When a job reaches a terminal state the runner doesn't reliably send
# a final per-step report — a failed job skips its remaining steps, and
# the terminal UpdateTask often carries only the job-level result — so
# `apply_step_states/2` never moves those rows off `pending`/`running`.
# Sweep the stragglers here so a finished job never shows in-flight
# steps:
# * job succeeded → step succeeded
# * step never started → skipped
# * step was in flight → inherit the job's terminal status
defp finalize_steps(%WorkflowJob{id: jid, status: job_status}) do
now = DateTime.utc_now() |> DateTime.truncate(:second)
from(s in WorkflowStep,
where: s.workflow_job_id == ^jid and s.status in ["pending", "running"]
)
|> Repo.all()
|> Enum.each(fn step ->
status =
cond do
job_status == "success" -> "success"
step.status == "pending" -> "skipped"
job_status in ["failure", "cancelled"] -> job_status
true -> "failure"
end
changes =
%{status: status}
|> maybe_put(:conclusion, step_status_conclusion(status))
|> maybe_put(:completed_at, step.completed_at || now)
{:ok, updated} = step |> Ecto.Changeset.change(changes) |> Repo.update()
broadcast({:step_status, jid, updated})
end)
end
@doc """
After a job reaches a terminal state, walk every sibling job in the
same run whose `needs:` array contains this job's `job_id` and
promote them from `blocked`:
* all listed needs are now `success`/`skipped` → `queued`
* any listed need is `failure`/`cancelled` → `cancelled`
(cascading failure, matches GitHub Actions default)
* mixed-but-not-failed (still some pending needs) → leave `blocked`
Public so tests can drive it directly.
"""
def resolve_dependents(%WorkflowJob{workflow_run_id: rid, job_id: completed_jid}) do
# All sibling jobs in this run, indexed by their job_id (yaml id).
siblings =
from(j in WorkflowJob, where: j.workflow_run_id == ^rid)
|> Repo.all()
by_jid = Map.new(siblings, fn j -> {j.job_id, j} end)
dependents = Enum.filter(siblings, fn j -> completed_jid in j.needs and j.status == "blocked" end)
Enum.each(dependents, fn dep ->
needs_states = Enum.map(dep.needs, fn jid -> Map.get(by_jid, jid) end)
cond do
Enum.any?(needs_states, &is_nil/1) ->
# A referenced need doesn't exist as a sibling — workflow
# author error. Cancel rather than block forever.
{:ok, _} = update_job_status(dep, "cancelled")
Enum.any?(needs_states, fn n -> n.status in ~w(failure cancelled) end) ->
{:ok, _} = update_job_status(dep, "cancelled")
Enum.all?(needs_states, fn n -> WorkflowJob.satisfies_dependency?(n.status) end) ->
# Snapshot upstream results/outputs now, while they're terminal.
# build_needs/1 reads this instead of re-reading upstream status
# at fetch time (which raced and could yield RESULT_UNSPECIFIED,
# making act skip the job on `if: success()`).
snapshot =
Map.new(dep.needs, fn jid ->
n = Map.get(by_jid, jid)
{jid, %{"status" => n.status, "outputs" => n.outputs || %{}}}
end)
{:ok, _} =
dep
|> Ecto.Changeset.change(status: "queued", needs_snapshot: snapshot)
|> Repo.update()
broadcast({:job_status, rid, Repo.get!(WorkflowJob, dep.id)})
true ->
# Some upstream is still pending. Leave blocked.
:ok
end
end)
:ok
end
defp maybe_complete_run(%WorkflowJob{workflow_run_id: rid}) do
statuses =
from(j in WorkflowJob, where: j.workflow_run_id == ^rid, select: j.status)
|> Repo.all()
cond do
# `blocked` is still pending — its upstream is in flight.
Enum.any?(statuses, &(&1 in ~w(queued running blocked))) ->
run = Repo.get!(WorkflowRun, rid)
update_run_commit_status(run)
:ok
Enum.any?(statuses, &(&1 in ~w(failure cancelled))) ->
run =
Repo.get!(WorkflowRun, rid)
|> WorkflowRun.status_changeset("failure")
|> Repo.update!()
update_run_commit_status(run)
broadcast_run_status(run)
true ->
run =
Repo.get!(WorkflowRun, rid)
|> WorkflowRun.status_changeset("success")
|> Repo.update!()
update_run_commit_status(run)
notify_run_webhook(run)
broadcast_run_status(run)
run
end
end
# Notify the run's LiveView that the run reached a terminal state so it
# can flip the status badge + show the re-run button without a reload.
# (The per-job broadcast races ahead of `maybe_complete_run`, so the
# run isn't terminal yet when that fires.)
defp broadcast_run_status(%WorkflowRun{id: rid} = run) do
Phoenix.PubSub.broadcast(GitGud.PubSub, topic_for_run(rid), {:workflow_run_status, run})
run
end
# Fire the `workflow_run` webhook for a run that just reached a
# terminal state. Loads the repo so the fan-out can reach org hooks.
defp notify_run_webhook(%WorkflowRun{} = run) do
repo = Repositories.get_repository!(run.repository_id)
_ = GitGud.Webhooks.notify(repo, "workflow_run", run_webhook_payload(repo, run))
:ok
rescue
# Webhooks are best-effort — never let a delivery enqueue failure
# roll back the run's terminal transition.
_ -> :ok
end
defp run_webhook_payload(%Repository{} = repo, %WorkflowRun{} = run) do
%{
"action" => "completed",
"workflow_run" => %{
"id" => run.id,
"run_number" => run.number,
"name" => run.workflow_name,
"path" => run.workflow_path,
"event" => run.trigger,
"status" => "completed",
"conclusion" => run.conclusion || run.status,
"head_sha" => run.head_sha && Git.to_hex(run.head_sha),
"head_branch" => head_branch(run.head_ref),
"html_url" =>
GitGudWeb.Endpoint.url() <>
"/r/repo/#{run.repository_id}/actions/runs/#{run.id}",
"run_started_at" => run.started_at,
"updated_at" => run.completed_at
},
"repository" => %{
"id" => repo.id,
"name" => repo.name,
"default_branch" => repo.default_branch
}
}
end
defp head_branch("refs/heads/" <> b), do: b
defp head_branch(other), do: other
@doc """
Upsert the commit status row for the given workflow run. Runs on
every status transition (queued → running → success/failure) so the
SCM views see ✓/✗ as soon as the run finishes.
`state` mapping:
queued, running -> "pending"
success -> "success"
failure, cancelled -> "failure"
other (error) -> "error"
"""
def update_run_commit_status(%WorkflowRun{} = run) do
state =
case run.status do
s when s in ~w(queued running) -> "pending"
"success" -> "success"
s when s in ~w(failure cancelled) -> "failure"
_ -> "error"
end
context = "gitgud-actions/" <> (run.workflow_name || run.workflow_path || "workflow")
target_url =
GitGudWeb.Endpoint.url() <>
"/r/repo/" <> Integer.to_string(run.repository_id) <>
"/actions/runs/" <> Integer.to_string(run.id)
description =
case run.status do
"success" -> "All checks passed"
"failure" -> "Some checks failed"
"cancelled" -> "Cancelled"
_ -> "Run #{run.number} #{run.status}"
end
attrs = %{
repository_id: run.repository_id,
sha: run.head_sha,
context: context,
state: state,
description: description,
target_url: target_url
}
result =
%GitGud.Workflows.CommitStatus{}
|> GitGud.Workflows.CommitStatus.changeset(attrs)
|> Repo.insert(
on_conflict:
{:replace, [:state, :description, :target_url, :updated_at]},
conflict_target: [:repository_id, :sha, :context]
)
# Every run transition (queued → running → terminal) flows through
# here, so it's the single chokepoint for waking the CI dashboards.
broadcast_ci_change(run.id)
result
end
# ── pubsub ───────────────────────────────────────────────────────────
def topic_for_run(run_id), do: "workflow_run:#{run_id}"
@doc """
PubSub topic the org + user CI dashboards subscribe to. A single
instance-wide topic keeps broadcasting cheap; subscribers re-query
their own (visibility-scoped) list on each event, so there's no risk
of leaking runs across visibility boundaries.
"""
def ci_topic, do: "ci:runs"
@doc "Wake the CI dashboards after a run's status changes."
def broadcast_ci_change(run_id) do
Phoenix.PubSub.broadcast(GitGud.PubSub, ci_topic(), {:ci_run_changed, run_id})
end
# ── re-run + cancel ──────────────────────────────────────────────────
@doc """
Re-run a previous workflow run. Creates a fresh `WorkflowRun` with
the same workflow_doc + head_sha + head_ref; jobs + steps are
re-seeded from the stored `workflow_doc` so they start in the
`queued` state for runners to pick up.
"""
def rerun(%WorkflowRun{} = old) do
repo = Repositories.get_repository!(old.repository_id)
doc = Yaml.normalize_doc(old.workflow_doc || %{})
create_run(repo, %{
workflow_path: old.workflow_path,
workflow_name: old.workflow_name,
trigger: old.trigger,
head_sha: old.head_sha,
head_ref: old.head_ref,
workflow_doc: old.workflow_doc,
jobs: doc.jobs
})
end
@doc """
Re-run a single job *in place* (e.g. retry a failed `image` without
re-running the 10-minute upstream `test`). Resets the job and its steps
to pending, clears its logs, re-queues it (`queued` when its `needs:`
are already satisfied, else `blocked`), and flips the run back to
`running` so the runner re-claims it. Upstream jobs are untouched.
"""
def rerun_job(%WorkflowRun{} = run, job_id) when is_binary(job_id) do
siblings = Repo.all(from j in WorkflowJob, where: j.workflow_run_id == ^run.id)
case Enum.find(siblings, &(&1.job_id == job_id)) do
nil ->
{:error, :not_found}
job ->
by_jid = Map.new(siblings, &{&1.job_id, &1})
needs_ok? =
job.needs == [] or
Enum.all?(job.needs, fn jid ->
case Map.get(by_jid, jid) do
%WorkflowJob{status: s} -> WorkflowJob.satisfies_dependency?(s)
_ -> false
end
end)
Repo.transaction(fn ->
# clean slate for this job's logs + steps. Logs are keyed by
# workflow_step_id, so scope by this job's step ids.
step_ids_q = from(s in WorkflowStep, where: s.workflow_job_id == ^job.id, select: s.id)
Repo.delete_all(from l in "workflow_step_logs", where: l.workflow_step_id in subquery(step_ids_q))
Repo.update_all(
from(s in WorkflowStep, where: s.workflow_job_id == ^job.id),
set: [
status: "pending",
conclusion: nil,
started_at: nil,
completed_at: nil,
log_index: nil,
log_length: nil
]
)
{:ok, updated} =
job
|> Ecto.Changeset.change(
status: if(needs_ok?, do: "queued", else: "blocked"),
conclusion: nil,
started_at: nil,
completed_at: nil,
runner_id: nil
)
|> Repo.update()
# Flip the run non-terminal so it isn't "complete" and the
# runner polls it. maybe_complete_run re-settles it on finish.
run |> WorkflowRun.status_changeset("running") |> Repo.update!()
updated
end)
|> case do
{:ok, updated} ->
broadcast({:job_status, run.id, updated})
broadcast_ci_change(run.id)
{:ok, Repo.get!(WorkflowRun, run.id)}
{:error, _} = err ->
err
end
end
end
@doc """
Cancel a queued or running workflow run. Flips the run + every
un-finished job + every pending/running step to `cancelled`. Runners
mid-step on a cancelled job see the status change on their next
heartbeat and self-terminate — we don't kill containers out-of-band.
No-op when the run is already terminal.
"""
def cancel(%WorkflowRun{status: status} = run) when status in ~w(success failure cancelled) do
{:ok, run}
end
def cancel(%WorkflowRun{} = run) do
Repo.transaction(fn ->
now = DateTime.utc_now(:second)
Repo.update_all(
from(j in WorkflowJob,
where: j.workflow_run_id == ^run.id and j.status in ["queued", "running"]
),
set: [status: "cancelled", conclusion: "cancelled", completed_at: now]
)
Repo.update_all(
from(s in WorkflowStep,
join: j in WorkflowJob,
on: j.id == s.workflow_job_id,
where: j.workflow_run_id == ^run.id and s.status in ["pending", "running"]
),
set: [status: "cancelled", conclusion: "cancelled", completed_at: now]
)
updated =
run
|> WorkflowRun.status_changeset("cancelled")
|> Repo.update!()
Phoenix.PubSub.broadcast(
GitGud.PubSub,
topic_for_run(run.id),
{:workflow_job_status, %WorkflowJob{workflow_run_id: run.id}}
)
updated
end)
|> case do
{:ok, updated} ->
update_run_commit_status(updated)
notify_run_webhook(updated)
{:ok, updated}
err ->
err
end
end
defp broadcast({:log, step_id, seq, body}) do
step = Repo.get!(WorkflowStep, step_id) |> Repo.preload(:workflow_job)
Phoenix.PubSub.broadcast(
GitGud.PubSub,
topic_for_run(step.workflow_job.workflow_run_id),
{:workflow_log, step_id, seq, body}
)
end
defp broadcast({:job_status, run_id, job}) do
Phoenix.PubSub.broadcast(
GitGud.PubSub,
topic_for_run(run_id),
{:workflow_job_status, job}
)
end
defp broadcast({:step_status, job_id, step}) do
# Re-derive the parent run id so subscribers (the workflow LV
# which subscribes per-run) get the update on the right topic.
case Repo.get(WorkflowJob, job_id) do
%WorkflowJob{workflow_run_id: rid} ->
Phoenix.PubSub.broadcast(
GitGud.PubSub,
topic_for_run(rid),
{:workflow_step_status, step}
)
_ ->
:ok
end
end
end
41.1 KiB · text
5af55d9