defmodule GitGudWeb.WorkflowLive.Show do
use GitGudWeb, :live_view
alias GitGud.Git
alias GitGud.Repositories
alias GitGud.Repositories.Storage
alias GitGud.Workflows
@impl true
def mount(%{"owner" => owner, "name" => name, "number" => num}, _session, socket) do
repo =
Repositories.get_repository_by_path!(owner, name)
|> GitGud.Repo.preload([:owner, :organization])
run = Workflows.get_run!(repo, String.to_integer(num))
if connected?(socket),
do: Phoenix.PubSub.subscribe(GitGud.PubSub, Workflows.topic_for_run(run.id))
user = socket.assigns.current_scope && socket.assigns.current_scope.user
socket =
socket
|> assign(:repo, repo)
|> assign(:handle, Storage.repo_handle(repo))
|> GitGudWeb.RepoLive.Header.assign_chrome(repo)
|> assign(:run, run)
|> assign(:artifacts, Workflows.list_artifacts(run))
|> assign(:expanded_step, nil)
|> assign(:can_admin?, user && admin?(user, repo))
|> assign(:page_title, "Run ##{run.number}")
|> assign(:now, DateTime.utc_now())
|> assign(:ticking?, false)
|> seed_step_streams(run)
|> ensure_ticking()
{:ok, socket}
end
# Drive a 1-second @now tick while the run is active so the elapsed
# counters on running jobs/steps update live. Self-perpetuates until the
# run is terminal; re-armed from the run-updating handlers (e.g. a
# per-job re-run flips the run back to `running`).
defp ensure_ticking(socket) do
if connected?(socket) and run_active?(socket.assigns.run) and not socket.assigns.ticking? do
Process.send_after(self(), :tick, 1000)
assign(socket, :ticking?, true)
else
socket
end
end
defp run_active?(%{status: s}), do: s in ~w(queued running)
defp admin?(user, %{owner_id: uid, organization_id: nil}), do: user.id == uid
defp admin?(user, %{organization: %{} = org}),
do: GitGud.Organizations.org_admin?(org, user)
defp admin?(_, _), do: false
defp seed_step_streams(socket, run) do
Enum.reduce(run.jobs, socket, fn job, acc ->
Enum.reduce(job.steps, acc, fn step, acc2 ->
stream(acc2, stream_name(step.id), [], dom_id: &"log-#{&1.seq}")
end)
end)
end
defp stream_name(step_id), do: :"step_log_#{step_id}"
defp find_step(run, id) do
Enum.find_value(run.jobs, fn job -> Enum.find(job.steps, &(&1.id == id)) end)
end
# Group jobs into dependency stages from their `needs` DAG: jobs in the
# same stage have no dependency between them, so they run in parallel.
# Matrix legs share a `job_id` with no `needs`, so they all land in
# stage 0 together.
defp job_stages(jobs) do
needs_by_id = Map.new(jobs, fn j -> {j.job_id, j.needs || []} end)
jobs
|> Enum.group_by(&job_depth(&1.job_id, needs_by_id, MapSet.new()))
|> Enum.sort_by(fn {stage, _} -> stage end)
|> Enum.map(fn {_stage, stage_jobs} -> stage_jobs end)
end
defp job_depth(id, needs_by_id, seen) do
if MapSet.member?(seen, id) do
0
else
case Map.get(needs_by_id, id, []) do
[] ->
0
needs ->
seen = MapSet.put(seen, id)
1 + (needs |> Enum.map(&job_depth(&1, needs_by_id, seen)) |> Enum.max(fn -> -1 end))
end
end
end
@impl true
def handle_info({:workflow_log, step_id, seq, body}, socket) do
if has_step?(socket, step_id) do
{:noreply, stream_insert(socket, stream_name(step_id), %{seq: seq, body: body}, at: -1)}
else
{:noreply, socket}
end
end
def handle_info({:workflow_job_status, _job}, socket) do
run = Workflows.get_run!(socket.assigns.repo, socket.assigns.run.number)
{:noreply,
socket
|> assign(:run, run)
|> assign(:artifacts, Workflows.list_artifacts(run))
|> ensure_ticking()}
end
# A step changed state (runner reported a new StepState). Re-read the
# run so the step + job badges reflect it. Without this clause the
# unmatched message crashes the LiveView, which then remounts and
# crashes again on the next step update — the "something went wrong"
# thrash.
def handle_info({:workflow_step_status, _step}, socket) do
run = Workflows.get_run!(socket.assigns.repo, socket.assigns.run.number)
{:noreply, socket |> assign(:run, run) |> ensure_ticking()}
end
# The run reached a terminal state — re-read so the status badge flips
# and the Re-run button appears without a manual reload.
def handle_info({:workflow_run_status, _run}, socket) do
run = Workflows.get_run!(socket.assigns.repo, socket.assigns.run.number)
{:noreply,
socket
|> assign(:run, run)
|> assign(:artifacts, Workflows.list_artifacts(run))
|> ensure_ticking()}
end
def handle_info(:tick, socket) do
socket = assign(socket, :now, DateTime.utc_now())
if run_active?(socket.assigns.run) do
Process.send_after(self(), :tick, 1000)
{:noreply, socket}
else
{:noreply, assign(socket, :ticking?, false)}
end
end
# Defensive catch-all: ignore any other broadcast on the run topic
# rather than letting an unhandled message take the whole view down.
def handle_info(_msg, socket), do: {:noreply, socket}
defp has_step?(socket, step_id) do
Enum.any?(socket.assigns.run.jobs, fn job ->
Enum.any?(job.steps, &(&1.id == step_id))
end)
end
@impl true
def handle_event("toggle_step", %{"id" => id}, socket) do
id = String.to_integer(id)
if socket.assigns.expanded_step == id do
{:noreply, assign(socket, :expanded_step, nil)}
else
# Hydrate the log view from persisted rows. The stream is seeded
# empty at mount and only ever fed by live PubSub appends, so a
# finished (or reloaded) run shows nothing until we load them here.
# `reset: true` replaces the stream with the authoritative set;
# dom_id keys on `seq` so any in-flight live appends dedupe.
logs =
case find_step(socket.assigns.run, id) do
%{} = step -> Workflows.list_step_logs(step)
_ -> []
end
{:noreply,
socket
|> assign(:expanded_step, id)
|> stream(stream_name(id), logs, reset: true, dom_id: &"log-#{&1.seq}")}
end
end
def handle_event("rerun", _params, socket) do
cond do
not socket.assigns.can_admin? ->
{:noreply, put_flash(socket, :error, "Only repo admins can re-run workflows.")}
true ->
case Workflows.rerun(socket.assigns.run) do
{:ok, new_run} ->
{:noreply,
socket
|> put_flash(:info, "Workflow re-run started as ##{new_run.number}.")
|> push_navigate(
to: ~p"/r/#{socket.assigns.handle}/#{socket.assigns.repo.name}/actions/#{new_run.number}"
)}
{:error, reason} ->
{:noreply, put_flash(socket, :error, "Could not re-run: #{inspect(reason)}")}
end
end
end
def handle_event("rerun_job", %{"job" => job_id}, socket) do
cond do
not socket.assigns.can_admin? ->
{:noreply, put_flash(socket, :error, "Only repo admins can re-run jobs.")}
true ->
case Workflows.rerun_job(socket.assigns.run, job_id) do
{:ok, _run} ->
run = Workflows.get_run!(socket.assigns.repo, socket.assigns.run.number)
{:noreply,
socket
|> put_flash(:info, "Re-running job #{job_id}.")
|> assign(:run, run)
|> ensure_ticking()}
{:error, reason} ->
{:noreply, put_flash(socket, :error, "Could not re-run job: #{inspect(reason)}")}
end
end
end
def handle_event("cancel", _params, socket) do
cond do
not socket.assigns.can_admin? ->
{:noreply, put_flash(socket, :error, "Only repo admins can cancel runs.")}
true ->
case Workflows.cancel(socket.assigns.run) do
{:ok, updated} ->
updated_full = Workflows.get_run!(socket.assigns.repo, updated.number)
{:noreply,
socket
|> assign(:run, updated_full)
|> put_flash(:info, "Run cancelled.")}
{:error, reason} ->
{:noreply, put_flash(socket, :error, "Could not cancel: #{inspect(reason)}")}
end
end
end
@impl true
def render(assigns) do
~H"""
<Layouts.app flash={@flash} current_scope={@current_scope}>
<div class="space-y-6">
<GitGudWeb.RepoLive.Header.header
repo={@repo}
handle={@handle}
current_scope={@current_scope}
has_packages?={@has_packages?}
can_admin?={@can_admin?}
latest_run={@latest_run}
/>
<header class="flex items-start justify-between gap-3">
<div>
<p class="text-xs opacity-60">
<.link navigate={~p"/r/#{@handle}/#{@repo.name}/actions"} class="link link-hover">
Back to actions
</.link>
</p>
<h1 class="text-2xl font-semibold">
#{@run.number} — {@run.workflow_name || @run.workflow_path}
</h1>
<p class="text-xs opacity-60 font-mono mt-1">
<span class={["badge", status_class(@run.status)]}>{@run.status}</span>
{@run.trigger} · {@run.head_ref} · {Git.short(@run.head_sha)}
</p>
<p class="text-xs opacity-60 mt-1">
<span :if={@run.started_at}>Started {format_dt(@run.started_at)}</span>
<span :if={@run.completed_at} class="ml-2">
· Finished {format_dt(@run.completed_at)}
</span>
<span :if={@run.started_at} class="ml-2">
· {duration(@run.started_at, @run.completed_at, @now)}
</span>
</p>
</div>
<div :if={@can_admin?} class="flex gap-2 shrink-0">
<button
:if={@run.status in ["queued", "running"]}
type="button"
phx-click="cancel"
data-confirm="Cancel this run? Running steps will terminate at the next heartbeat."
class="btn btn-sm btn-ghost text-error"
>
Cancel
</button>
<button
:if={@run.status in ["success", "failure", "cancelled"]}
type="button"
phx-click="rerun"
class="btn btn-sm"
>
<.icon name="hero-arrow-path" class="size-3" /> Re-run
</button>
</div>
</header>
<section class="space-y-6">
<div :for={stage <- job_stages(@run.jobs)} class="space-y-2">
<div
:if={length(stage) > 1}
class="text-xs uppercase tracking-wide opacity-50 flex items-center gap-1.5"
>
<.icon name="hero-bolt" class="size-3" /> {length(stage)} jobs run in parallel
</div>
<div class={["grid gap-4", length(stage) > 1 && "lg:grid-cols-2"]}>
<article :for={job <- stage} class="border border-base-300 rounded p-4">
<header class="flex items-baseline justify-between gap-3">
<h2 class="font-semibold font-mono">{job.name || job.job_id}</h2>
<div class="flex items-baseline gap-2 text-xs opacity-60">
<span :if={job.started_at}>{duration(job.started_at, job.completed_at, @now)}</span>
<span class={["badge badge-sm", status_class(job.status)]}>{job.status}</span>
<button
:if={@can_admin? and job.status in ["failure", "cancelled", "success", "skipped"]}
type="button"
phx-click="rerun_job"
phx-value-job={job.job_id}
class="btn btn-ghost btn-xs"
title={"Re-run only #{job.job_id} (keeps upstream jobs)"}
>
<.icon name="hero-arrow-path" class="size-3" /> Re-run job
</button>
</div>
</header>
<ol class="mt-3 space-y-2 text-sm">
<li :for={step <- job.steps} class="border border-base-200 rounded">
<button
type="button"
phx-click="toggle_step"
phx-value-id={step.id}
class="w-full flex items-baseline gap-2 px-3 py-2 hover:bg-base-200"
>
<span class="opacity-60 text-xs w-6">{step.number}</span>
<span class="flex-1 text-left font-mono">{step.name || "(unnamed)"}</span>
<span :if={step.started_at} class="text-xs opacity-50">
{duration(step.started_at, step.completed_at, @now)}
</span>
<span class={["badge badge-xs", status_class(step.status)]}>{step.status}</span>
<.icon
name={
if @expanded_step == step.id, do: "hero-chevron-up", else: "hero-chevron-down"
}
class="size-4 opacity-50"
/>
</button>
<pre
:if={@expanded_step == step.id}
id={"step-log-#{step.id}"}
phx-update="stream"
class="text-xs font-mono bg-black text-green-300 p-3 overflow-auto max-h-96"
phx-no-curly-interpolation
><div :for={{dom_id, chunk} <- @streams[stream_name(step.id)]} id={dom_id}><%= chunk.body %></div></pre>
</li>
</ol>
</article>
</div>
</div>
</section>
<section :if={@artifacts != []} class="space-y-2">
<h2 class="font-semibold">Artifacts</h2>
<ul class="divide-y divide-base-300 border border-base-300 rounded">
<li :for={a <- @artifacts} class="flex items-baseline justify-between px-3 py-2 text-sm">
<div class="flex flex-col">
<span class="font-mono">{a.name}</span>
<span class="text-xs opacity-60">
{format_size(a.size)}
<span :if={a.content_type}> · {a.content_type}</span>
<span :if={a.inserted_at}> · {format_dt(a.inserted_at)}</span>
</span>
</div>
<a
href={~p"/r/#{@handle}/#{@repo.name}/actions/#{@run.number}/artifacts/#{a.id}"}
class="btn btn-xs"
download
>
<.icon name="hero-arrow-down-tray" class="size-3" /> Download
</a>
</li>
</ul>
</section>
</div>
</Layouts.app>
"""
end
defp format_dt(%DateTime{} = dt), do: Calendar.strftime(dt, "%Y-%m-%d %H:%M")
defp format_dt(_), do: ""
defp format_size(nil), do: ""
defp format_size(b) when b < 1024, do: "#{b} B"
defp format_size(b) when b < 1_048_576, do: "#{Float.round(b / 1024, 1)} KB"
defp format_size(b), do: "#{Float.round(b / 1_048_576, 1)} MB"
# `now` is the ticking @now assign so LiveView re-evaluates the elapsed
# counter each second while a job/step is still running.
defp duration(nil, _finish, _now), do: ""
defp duration(start, nil, now),
do: "for #{seconds_to_human(DateTime.diff(now, start))}"
defp duration(start, %DateTime{} = finish, _now),
do: "in #{seconds_to_human(DateTime.diff(finish, start))}"
defp seconds_to_human(secs) when secs < 0, do: "0s"
defp seconds_to_human(secs) when secs < 60, do: "#{secs}s"
defp seconds_to_human(secs) when secs < 3600, do: "#{div(secs, 60)}m #{rem(secs, 60)}s"
defp seconds_to_human(secs), do: "#{div(secs, 3600)}h #{div(rem(secs, 3600), 60)}m"
defp status_class("success"), do: "badge-success"
defp status_class("running"), do: "badge-primary"
defp status_class("queued"), do: "badge-warning"
defp status_class("pending"), do: "badge-warning"
# `blocked` = waiting on upstream `needs:`. Distinct from `queued`
# so operators can tell pickup-eligible vs gated jobs.
defp status_class("blocked"), do: "badge-ghost"
defp status_class("failure"), do: "badge-error"
defp status_class("cancelled"), do: "badge-error"
defp status_class("skipped"), do: "badge-ghost"
defp status_class(_), do: "badge-neutral"
end
16.0 KiB · text
5af55d9