neiam
/apollos_crib
It's Apollo's Crib
public · Issues · Pulls 1 · Labels · Forks · Compare · Actions failure · Badges · Packages
⭐
Log in to mark this repository.
Add /cfg/offerings/work, the import queue across every feed
448b406 · Gabriel Morell · 2026-09-01 20:40
Message
{commit_body(@commit)}
Files changed
modified
apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/index.html.heex
+3
−0
@@ -15,6 +15,9 @@
| 15 | 15 | <% end %> |
| 16 | 16 | |
| 17 | 17 | <.itb icon="fa-upload" to={Routes.source_index_path(@socket, :import)} what="Import Offering"></.itb> |
| 18 | + <%!-- One feed's progress bar lives on its own page; this is the queue | |
| 19 | + behind all of them. --%> | |
| 20 | + <.itb icon="fa-list-check" to={Routes.source_work_path(@socket, :index)} what="Import Queue"></.itb> | |
| 18 | 21 | </div> |
| 19 | 22 | </div> |
| 20 | 23 |
added
apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/work.ex
+253
−0
@@ -0,0 +1,253 @@
| 1 | +defmodule RoomSanctumWeb.SourceLive.Work do | |
| 2 | + @moduledoc """ | |
| 3 | + What the GTFS import queue is doing, across every source at once. | |
| 4 | + | |
| 5 | + The source page shows one feed's progress bar, which is the right view when | |
| 6 | + you are looking at one feed. It is the wrong view for the question this | |
| 7 | + answers: a queue with a concurrency of one, several feeds behind it, and no | |
| 8 | + way to tell whether yours is running, waiting, or failed three times an hour | |
| 9 | + ago without opening each source in turn. | |
| 10 | + | |
| 11 | + Two things are stitched together here, because neither is sufficient alone: | |
| 12 | + | |
| 13 | + * The Oban rows say what is queued, running, done or discarded, and are the | |
| 14 | + only place a failure and its error text survive. They do not move on | |
| 15 | + their own, so they are re-read on a timer. | |
| 16 | + | |
| 17 | + * The `"gtfs"` PubSub topic carries live progress for whichever import is | |
| 18 | + running -- the same broadcasts the source page draws its bar from -- and | |
| 19 | + arrives many times a second while a feed loads. Oban knows a job is | |
| 20 | + executing; only this knows it is four files in. | |
| 21 | + """ | |
| 22 | + use RoomSanctumWeb, :live_view_a | |
| 23 | + | |
| 24 | + import Ecto.Query | |
| 25 | + | |
| 26 | + alias RoomSanctum.{Configuration, Repo} | |
| 27 | + | |
| 28 | + # Oban rows change without telling anyone, so they are re-read. Slow enough | |
| 29 | + # not to matter against a database this has spent the afternoon apologising | |
| 30 | + # to, quick enough that a job finishing is noticed before you wonder. | |
| 31 | + @reload_ms 3_000 | |
| 32 | + | |
| 33 | + @impl true | |
| 34 | + def mount(_params, _session, socket) do | |
| 35 | + if connected?(socket) do | |
| 36 | + Phoenix.PubSub.subscribe(RoomSanctum.PubSub, "gtfs") | |
| 37 | + Process.send_after(self(), :reload, @reload_ms) | |
| 38 | + end | |
| 39 | + | |
| 40 | + {:ok, | |
| 41 | + socket | |
| 42 | + |> assign(:page_title, "Import Queue") | |
| 43 | + |> assign(:progress, %{}) | |
| 44 | + |> load()} | |
| 45 | + end | |
| 46 | + | |
| 47 | + @impl true | |
| 48 | + def handle_info(:reload, socket) do | |
| 49 | + Process.send_after(self(), :reload, @reload_ms) | |
| 50 | + {:noreply, load(socket)} | |
| 51 | + end | |
| 52 | + | |
| 53 | + # Progress for one feed, as the source page reads it. Held per source rather | |
| 54 | + # than as a single "current" bar: the queue runs one at a time today, and a | |
| 55 | + # page that assumes that would quietly show the wrong feed's bar the day the | |
| 56 | + # concurrency is raised. | |
| 57 | + @impl true | |
| 58 | + def handle_info({:gtfs, id, :done}, socket) do | |
| 59 | + progress = Map.delete(socket.assigns.progress, to_string(id)) | |
| 60 | + | |
| 61 | + {:noreply, socket |> assign(:progress, progress) |> load()} | |
| 62 | + end | |
| 63 | + | |
| 64 | + @impl true | |
| 65 | + def handle_info({:gtfs, _id, :disabled}, socket), do: {:noreply, socket} | |
| 66 | + | |
| 67 | + @impl true | |
| 68 | + def handle_info({:gtfs, id, file, complete, total}, socket) do | |
| 69 | + # The "gtfs" topic carries every source in the installation, not just this | |
| 70 | + # user's. Without this, somebody else's import shows up here as the thing | |
| 71 | + # now running -- under a name this page cannot resolve, because it only | |
| 72 | + # loaded its own feeds. | |
| 73 | + if mine?(socket, id) do | |
| 74 | + entry = %{step: file, value: percent(complete, total), at: DateTime.utc_now()} | |
| 75 | + progress = Map.put(socket.assigns.progress, to_string(id), entry) | |
| 76 | + | |
| 77 | + {:noreply, assign(socket, :progress, progress)} | |
| 78 | + else | |
| 79 | + {:noreply, socket} | |
| 80 | + end | |
| 81 | + end | |
| 82 | + | |
| 83 | + defp mine?(socket, id) do | |
| 84 | + case Integer.parse(to_string(id)) do | |
| 85 | + {id, ""} -> Map.has_key?(socket.assigns.names, id) | |
| 86 | + _otherwise -> false | |
| 87 | + end | |
| 88 | + end | |
| 89 | + | |
| 90 | + @impl true | |
| 91 | + def handle_info(_msg, socket), do: {:noreply, socket} | |
| 92 | + | |
| 93 | + @impl true | |
| 94 | + def handle_event("requeue", %{"id" => id}, socket) do | |
| 95 | + case RoomGtfs.Worker.update_static_data(String.to_integer(id)) do | |
| 96 | + {:ok, %{conflict?: true}} -> | |
| 97 | + {:noreply, socket |> put_flash(:info, "That feed already has an import outstanding")} | |
| 98 | + | |
| 99 | + {:ok, _job} -> | |
| 100 | + {:noreply, socket |> put_flash(:info, "Queued") |> load()} | |
| 101 | + | |
| 102 | + _otherwise -> | |
| 103 | + {:noreply, put_flash(socket, :error, "Could not queue that import")} | |
| 104 | + end | |
| 105 | + end | |
| 106 | + | |
| 107 | + defp load(socket) do | |
| 108 | + sources = | |
| 109 | + Configuration.list_cfg_sources({:type, :gtfs}) | |
| 110 | + |> Enum.filter(&(&1.user_id == socket.assigns.current_user.id)) | |
| 111 | + | |
| 112 | + names = Map.new(sources, &{&1.id, &1.name}) | |
| 113 | + | |
| 114 | + socket | |
| 115 | + |> assign(:sources, Enum.sort_by(sources, & &1.name)) | |
| 116 | + |> assign(:names, names) | |
| 117 | + |> assign(:jobs, jobs()) | |
| 118 | + end | |
| 119 | + | |
| 120 | + # The queue's own history. Fifty is enough to cover a night's run of every | |
| 121 | + # feed and its retries without turning the page into a log. | |
| 122 | + defp jobs do | |
| 123 | + from(j in "oban_jobs", | |
| 124 | + where: j.queue == "gtfs_import", | |
| 125 | + order_by: [desc: j.id], | |
| 126 | + limit: 50, | |
| 127 | + select: %{ | |
| 128 | + id: j.id, | |
| 129 | + state: j.state, | |
| 130 | + args: j.args, | |
| 131 | + attempt: j.attempt, | |
| 132 | + max_attempts: j.max_attempts, | |
| 133 | + inserted_at: j.inserted_at, | |
| 134 | + scheduled_at: j.scheduled_at, | |
| 135 | + attempted_at: j.attempted_at, | |
| 136 | + completed_at: j.completed_at, | |
| 137 | + discarded_at: j.discarded_at, | |
| 138 | + errors: j.errors | |
| 139 | + } | |
| 140 | + ) | |
| 141 | + |> Repo.all() | |
| 142 | + rescue | |
| 143 | + # The table is Oban's, not ours, and a page that cannot read it should say | |
| 144 | + # so rather than take the whole view down. | |
| 145 | + _ -> [] | |
| 146 | + end | |
| 147 | + | |
| 148 | + @doc """ | |
| 149 | + The import that is actually running, if one is. | |
| 150 | + | |
| 151 | + Both halves are needed and neither is sufficient. Oban says which job is | |
| 152 | + executing, which is the authoritative answer to *which* feed -- but its rows | |
| 153 | + are re-read on a timer, so it lags the start of an import by seconds. The | |
| 154 | + progress broadcasts are immediate and say how far in it is, but arrive keyed | |
| 155 | + by a source id with no name attached and keep arriving from a run that has | |
| 156 | + since died. | |
| 157 | + | |
| 158 | + So: the freshest broadcast wins, because it is the one that moves; the | |
| 159 | + executing job is the fallback for the moment after a job is picked up and | |
| 160 | + before it has said anything. | |
| 161 | + """ | |
| 162 | + def active(jobs, progress, names) do | |
| 163 | + id = | |
| 164 | + newest_progress(progress) || | |
| 165 | + Enum.find_value(jobs, fn job -> if job.state == "executing", do: source_id(job) end) | |
| 166 | + | |
| 167 | + case id do | |
| 168 | + nil -> nil | |
| 169 | + id -> %{source_id: id, name: Map.get(names, id), progress: Map.get(progress, to_string(id))} | |
| 170 | + end | |
| 171 | + end | |
| 172 | + | |
| 173 | + # An import that stopped saying anything a minute ago is not running; it is | |
| 174 | + # a bar left behind by one that died, and leaving it up would be a page | |
| 175 | + # confidently reporting work that is not happening. | |
| 176 | + @stale_after_s 60 | |
| 177 | + | |
| 178 | + defp newest_progress(progress) do | |
| 179 | + progress | |
| 180 | + |> Enum.reject(fn {_id, p} -> DateTime.diff(DateTime.utc_now(), p.at) > @stale_after_s end) | |
| 181 | + |> Enum.max_by(fn {_id, p} -> DateTime.to_unix(p.at, :millisecond) end, fn -> nil end) | |
| 182 | + |> case do | |
| 183 | + {id, _p} -> String.to_integer(id) | |
| 184 | + nil -> nil | |
| 185 | + end | |
| 186 | + end | |
| 187 | + | |
| 188 | + @doc false | |
| 189 | + def source_id(%{args: %{"source_id" => id}}) when is_integer(id), do: id | |
| 190 | + def source_id(%{args: %{"source_id" => id}}) when is_binary(id), do: String.to_integer(id) | |
| 191 | + def source_id(_job), do: nil | |
| 192 | + | |
| 193 | + @doc """ | |
| 194 | + What a job's state should look like. | |
| 195 | + | |
| 196 | + Oban's vocabulary is not a reader's: `available` means waiting, `discarded` | |
| 197 | + means it gave up. The colours follow the meaning rather than the word. | |
| 198 | + """ | |
| 199 | + def state_label("available"), do: {"waiting", "badge-ghost"} | |
| 200 | + def state_label("scheduled"), do: {"scheduled", "badge-ghost"} | |
| 201 | + def state_label("executing"), do: {"running", "badge-info"} | |
| 202 | + def state_label("retryable"), do: {"retrying", "badge-warning"} | |
| 203 | + def state_label("completed"), do: {"done", "badge-success"} | |
| 204 | + def state_label("discarded"), do: {"gave up", "badge-error"} | |
| 205 | + def state_label("cancelled"), do: {"cancelled", "badge-ghost"} | |
| 206 | + def state_label(other), do: {other, "badge-ghost"} | |
| 207 | + | |
| 208 | + @doc """ | |
| 209 | + The last error a job recorded, if it recorded one. | |
| 210 | + | |
| 211 | + Only the message: the stacktrace Oban keeps alongside it is worth having in | |
| 212 | + the database and is not worth a table cell. | |
| 213 | + """ | |
| 214 | + def last_error([_ | _] = errors) do | |
| 215 | + errors | |
| 216 | + |> List.last() | |
| 217 | + |> Map.get("error", "") | |
| 218 | + |> to_string() | |
| 219 | + |> String.split("\n") | |
| 220 | + |> List.first() | |
| 221 | + |> String.slice(0, 160) | |
| 222 | + end | |
| 223 | + | |
| 224 | + def last_error(_errors), do: nil | |
| 225 | + | |
| 226 | + @doc false | |
| 227 | + def ago(nil), do: "--" | |
| 228 | + | |
| 229 | + def ago(%NaiveDateTime{} = at) do | |
| 230 | + at |> DateTime.from_naive!("Etc/UTC") |> ago() | |
| 231 | + end | |
| 232 | + | |
| 233 | + def ago(%DateTime{} = at) do | |
| 234 | + case DateTime.diff(DateTime.utc_now(), at) do | |
| 235 | + s when s < 60 -> "#{s}s ago" | |
| 236 | + s when s < 3600 -> "#{div(s, 60)}m ago" | |
| 237 | + s when s < 86_400 -> "#{div(s, 3600)}h ago" | |
| 238 | + s -> "#{div(s, 86_400)}d ago" | |
| 239 | + end | |
| 240 | + end | |
| 241 | + | |
| 242 | + # The step names the worker broadcasts, said the way the source page says | |
| 243 | + # them. | |
| 244 | + @doc false | |
| 245 | + def step_label(:downloading), do: "retrieving bundle" | |
| 246 | + def step_label(:parsing), do: "parsing bundle" | |
| 247 | + def step_label(:extracting), do: "extracting bundle" | |
| 248 | + def step_label(:error), do: "error" | |
| 249 | + def step_label(step), do: step |> to_string() |> String.replace("_", " ") | |
| 250 | + | |
| 251 | + defp percent(_complete, total) when total in [nil, 0], do: 0 | |
| 252 | + defp percent(complete, total), do: round(complete / total * 100) | |
| 253 | +end |
added
apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/work.html.heex
+143
−0
@@ -0,0 +1,143 @@
| 1 | +<.itb icon="fa-chevron-left" to={Routes.source_index_path(@socket, :index)} what="Back"></.itb> | |
| 2 | + | |
| 3 | +<h1 class="text-3xl font-bold text-accent mx-4 py-2">Import Queue</h1> | |
| 4 | + | |
| 5 | +<div class="mx-4 py-2"> | |
| 6 | + <p class="text-sm opacity-70 mb-2"> | |
| 7 | + <%!-- The concurrency is the whole mechanism, so it is worth saying rather | |
| 8 | + than leaving to be inferred from feeds appearing to stall. --%> | |
| 9 | + GTFS static imports run one at a time. A feed that is waiting is waiting for | |
| 10 | + the one above it, not stuck. | |
| 11 | + </p> | |
| 12 | + | |
| 13 | + <%!-- The running import gets the room, because it is the one thing on this | |
| 14 | + page that is changing while you watch it. --%> | |
| 15 | + <% running = active(@jobs, @progress, @names) %> | |
| 16 | + | |
| 17 | + <div class="card card-compact w-full bg-base-100 shadow-xl my-2"> | |
| 18 | + <div class="card-body"> | |
| 19 | + <h3 class="card-title">Now running</h3> | |
| 20 | + | |
| 21 | + <%= if running do %> | |
| 22 | + <div class="flex items-baseline gap-3"> | |
| 23 | + <.link | |
| 24 | + navigate={Routes.source_show_path(@socket, :show, running.source_id)} | |
| 25 | + class="text-2xl font-bold text-accent" | |
| 26 | + > | |
| 27 | + <%= running.name || "source #{running.source_id}" %> | |
| 28 | + </.link> | |
| 29 | + <span class="opacity-70"> | |
| 30 | + <%= if running.progress, do: step_label(running.progress.step), else: "starting" %> | |
| 31 | + </span> | |
| 32 | + </div> | |
| 33 | + | |
| 34 | + <progress | |
| 35 | + class="progress progress-secondary w-full" | |
| 36 | + value={if running.progress, do: running.progress.value, else: 0} | |
| 37 | + max="100" | |
| 38 | + > | |
| 39 | + </progress> | |
| 40 | + | |
| 41 | + <p class="text-sm opacity-70"> | |
| 42 | + <%= if running.progress, do: "#{running.progress.value}%", else: "picked up, no progress reported yet" %> | |
| 43 | + </p> | |
| 44 | + <% else %> | |
| 45 | + <p class="text-sm opacity-70">Nothing is importing right now.</p> | |
| 46 | + <% end %> | |
| 47 | + </div> | |
| 48 | + </div> | |
| 49 | + | |
| 50 | + <div class="card card-compact w-full bg-base-100 shadow-xl my-2"> | |
| 51 | + <div class="card-body"> | |
| 52 | + <h3 class="card-title">Feeds</h3> | |
| 53 | + | |
| 54 | + <div class="flex flex-wrap gap-2"> | |
| 55 | + <%= for source <- @sources do %> | |
| 56 | + <div class="flex items-center gap-2 border border-base-300 rounded px-2 py-1"> | |
| 57 | + <.link | |
| 58 | + navigate={Routes.source_show_path(@socket, :show, source.id)} | |
| 59 | + class="font-bold text-accent" | |
| 60 | + > | |
| 61 | + <%= source.name %> | |
| 62 | + </.link> | |
| 63 | + | |
| 64 | + <%!-- Live progress for whichever feed is loading, from the same | |
| 65 | + broadcasts the source page draws its bar from. Oban knows a | |
| 66 | + job is executing; only this knows it is four files in. --%> | |
| 67 | + <%!-- The bar lives in the card above; here a feed only has to say | |
| 68 | + whether it is the one running. --%> | |
| 69 | + <%= if running && running.source_id == source.id do %> | |
| 70 | + <span class="badge badge-info badge-sm">running</span> | |
| 71 | + <% else %> | |
| 72 | + <span class="text-xs opacity-50">idle</span> | |
| 73 | + <% end %> | |
| 74 | + | |
| 75 | + <button | |
| 76 | + class="btn btn-xs btn-ghost" | |
| 77 | + phx-click="requeue" | |
| 78 | + phx-value-id={source.id} | |
| 79 | + title="Queue a static import for this feed" | |
| 80 | + > | |
| 81 | + <i class="fa-solid fa-rotate"></i> | |
| 82 | + </button> | |
| 83 | + </div> | |
| 84 | + <% end %> | |
| 85 | + </div> | |
| 86 | + </div> | |
| 87 | + </div> | |
| 88 | + | |
| 89 | + <div class="card card-compact w-full bg-base-100 shadow-xl my-2"> | |
| 90 | + <div class="card-body"> | |
| 91 | + <h3 class="card-title">Jobs</h3> | |
| 92 | + | |
| 93 | + <p :if={@jobs == []} class="text-sm opacity-70"> | |
| 94 | + Nothing has been queued yet. | |
| 95 | + </p> | |
| 96 | + | |
| 97 | + <div :if={@jobs != []} class="overflow-x-auto"> | |
| 98 | + <table class="table table-compact w-full"> | |
| 99 | + <thead> | |
| 100 | + <tr> | |
| 101 | + <th>Job</th> | |
| 102 | + <th>Feed</th> | |
| 103 | + <th>State</th> | |
| 104 | + <th>Try</th> | |
| 105 | + <th>Queued</th> | |
| 106 | + <th>Started</th> | |
| 107 | + <th>Finished</th> | |
| 108 | + <th>Error</th> | |
| 109 | + </tr> | |
| 110 | + </thead> | |
| 111 | + <tbody> | |
| 112 | + <%= for job <- @jobs do %> | |
| 113 | + <% {label, class} = state_label(job.state) %> | |
| 114 | + <% sid = source_id(job) %> | |
| 115 | + <tr> | |
| 116 | + <td class="font-mono text-xs opacity-60"><%= job.id %></td> | |
| 117 | + <td> | |
| 118 | + <%= if sid && Map.get(@names, sid) do %> | |
| 119 | + <.link navigate={Routes.source_show_path(@socket, :show, sid)} class="text-accent"> | |
| 120 | + <%= Map.get(@names, sid) %> | |
| 121 | + </.link> | |
| 122 | + <% else %> | |
| 123 | + <%!-- A job whose source has since been deleted still has a | |
| 124 | + history worth reading. --%> | |
| 125 | + <span class="opacity-60">source <%= sid || "?" %></span> | |
| 126 | + <% end %> | |
| 127 | + </td> | |
| 128 | + <td><span class={"badge badge-sm #{class}"}><%= label %></span></td> | |
| 129 | + <td class="text-xs opacity-70"><%= job.attempt %>/<%= job.max_attempts %></td> | |
| 130 | + <td class="text-xs opacity-70"><%= ago(job.inserted_at) %></td> | |
| 131 | + <td class="text-xs opacity-70"><%= ago(job.attempted_at) %></td> | |
| 132 | + <td class="text-xs opacity-70"> | |
| 133 | + <%= ago(job.completed_at || job.discarded_at) %> | |
| 134 | + </td> | |
| 135 | + <td class="text-xs text-error"><%= last_error(job.errors) %></td> | |
| 136 | + </tr> | |
| 137 | + <% end %> | |
| 138 | + </tbody> | |
| 139 | + </table> | |
| 140 | + </div> | |
| 141 | + </div> | |
| 142 | + </div> | |
| 143 | +</div> |
modified
apps/room_sanctum/lib/room_sanctum_web/router.ex
+3
−0
@@ -65,6 +65,9 @@ defmodule RoomSanctumWeb.Router do
| 65 | 65 | live "/cfg/offerings", SourceLive.Index, :index |
| 66 | 66 | live "/cfg/offerings/new", SourceLive.Index, :new |
| 67 | 67 | live "/cfg/offerings/import", SourceLive.Index, :import |
| 68 | + # What the import queue is doing across every feed, rather than one feed's | |
| 69 | + # bar. Ahead of the `:id` routes for the same reason "map" is. | |
| 70 | + live "/cfg/offerings/work", SourceLive.Work, :index | |
| 68 | 71 | # Ahead of the `:id` routes. Nothing below actually collides -- none of them |
| 69 | 72 | # is three segments ending in a free value -- but "map" reading as an id is |
| 70 | 73 | # exactly the kind of thing a later route would introduce quietly. |
added
apps/room_sanctum/test/room_sanctum_web/live/source_work_test.exs
+198
−0
@@ -0,0 +1,198 @@
| 1 | +defmodule RoomSanctumWeb.SourceWorkTest do | |
| 2 | + @moduledoc """ | |
| 3 | + The import queue, across every feed at once. | |
| 4 | + | |
| 5 | + The source page shows one feed's bar, which cannot answer "is mine running or | |
| 6 | + waiting behind three others" without opening each source in turn. This | |
| 7 | + stitches Oban's rows -- the only place a failure survives -- to the live | |
| 8 | + progress broadcasts the source page already draws from. | |
| 9 | + """ | |
| 10 | + use RoomSanctumWeb.ConnCase | |
| 11 | + | |
| 12 | + import Phoenix.LiveViewTest | |
| 13 | + | |
| 14 | + alias RoomSanctum.{Accounts, Configuration, Repo} | |
| 15 | + | |
| 16 | + setup %{conn: conn} do | |
| 17 | + {:ok, user} = | |
| 18 | + Accounts.register_user(%{ | |
| 19 | + email: "work#{System.unique_integer([:positive])}@example.com", | |
| 20 | + password: "hello world!hello world!" | |
| 21 | + }) | |
| 22 | + | |
| 23 | + {:ok, source} = | |
| 24 | + Configuration.create_source(%{ | |
| 25 | + name: "Bus Feed", notes: "", type: :gtfs, enabled: true, user_id: user.id, | |
| 26 | + config: %{"__type__" => "gtfs", "url" => "https://e.test/g.zip", "tz" => "UTC"} | |
| 27 | + }) | |
| 28 | + | |
| 29 | + %{conn: log_in_user(conn, user), user: user, source: source} | |
| 30 | + end | |
| 31 | + | |
| 32 | + defp job(source_id, state, attrs \\ %{}) do | |
| 33 | + now = NaiveDateTime.utc_now() |> NaiveDateTime.truncate(:second) | |
| 34 | + | |
| 35 | + row = | |
| 36 | + Map.merge( | |
| 37 | + %{ | |
| 38 | + state: state, queue: "gtfs_import", worker: "RoomGtfs.ImportJob", | |
| 39 | + args: %{"source_id" => source_id}, errors: [], tags: [], meta: %{}, | |
| 40 | + attempt: 0, max_attempts: 3, priority: 0, | |
| 41 | + inserted_at: now, scheduled_at: now | |
| 42 | + }, | |
| 43 | + attrs | |
| 44 | + ) | |
| 45 | + | |
| 46 | + {1, [%{id: id}]} = Repo.insert_all("oban_jobs", [row], returning: [:id]) | |
| 47 | + id | |
| 48 | + end | |
| 49 | + | |
| 50 | + test "the page lists this user's feeds", %{conn: conn, source: source} do | |
| 51 | + {:ok, _live, html} = live(conn, "/cfg/offerings/work") | |
| 52 | + | |
| 53 | + assert html =~ source.name | |
| 54 | + assert html =~ "Import Queue" | |
| 55 | + end | |
| 56 | + | |
| 57 | + test "it says the queue runs one at a time, so a waiting feed is not a stuck one", %{conn: conn} do | |
| 58 | + {:ok, _live, html} = live(conn, "/cfg/offerings/work") | |
| 59 | + | |
| 60 | + assert html =~ "one at a time" | |
| 61 | + end | |
| 62 | + | |
| 63 | + test "nothing queued says so rather than showing an empty table", %{conn: conn} do | |
| 64 | + {:ok, _live, html} = live(conn, "/cfg/offerings/work") | |
| 65 | + | |
| 66 | + assert html =~ "Nothing has been queued yet" | |
| 67 | + end | |
| 68 | + | |
| 69 | + describe "job states" do | |
| 70 | + test "Oban's vocabulary is translated into a reader's", %{conn: conn, source: source} do | |
| 71 | + job(source.id, "available") | |
| 72 | + | |
| 73 | + {:ok, _live, html} = live(conn, "/cfg/offerings/work") | |
| 74 | + | |
| 75 | + # "available" means waiting for the job ahead of it, which is not what | |
| 76 | + # the word suggests. | |
| 77 | + assert html =~ "waiting" | |
| 78 | + refute html =~ ">available<" | |
| 79 | + end | |
| 80 | + | |
| 81 | + test "a job that gave up shows its error", %{conn: conn, source: source} do | |
| 82 | + job(source.id, "discarded", %{ | |
| 83 | + attempt: 3, | |
| 84 | + errors: [%{"error" => "** (MatchError) no match of right hand side\n at foo.ex:1"}], | |
| 85 | + discarded_at: NaiveDateTime.utc_now() |> NaiveDateTime.truncate(:second) | |
| 86 | + }) | |
| 87 | + | |
| 88 | + {:ok, _live, html} = live(conn, "/cfg/offerings/work") | |
| 89 | + | |
| 90 | + assert html =~ "gave up" | |
| 91 | + assert html =~ "MatchError" | |
| 92 | + # The stacktrace belongs in the database, not in a table cell. | |
| 93 | + refute html =~ "at foo.ex:1" | |
| 94 | + end | |
| 95 | + | |
| 96 | + test "a running job reads as running", %{conn: conn, source: source} do | |
| 97 | + job(source.id, "executing", %{attempt: 1}) | |
| 98 | + | |
| 99 | + {:ok, _live, html} = live(conn, "/cfg/offerings/work") | |
| 100 | + | |
| 101 | + assert html =~ "running" | |
| 102 | + assert html =~ "1/3" | |
| 103 | + end | |
| 104 | + end | |
| 105 | + | |
| 106 | + describe "live progress" do | |
| 107 | + test "an idle queue says so rather than showing an empty bar", %{conn: conn} do | |
| 108 | + {:ok, _live, html} = live(conn, "/cfg/offerings/work") | |
| 109 | + | |
| 110 | + assert html =~ "Nothing is importing right now" | |
| 111 | + assert html =~ "idle" | |
| 112 | + end | |
| 113 | + | |
| 114 | + test "a progress broadcast draws that feed's bar", %{conn: conn, source: source} do | |
| 115 | + {:ok, live, _html} = live(conn, "/cfg/offerings/work") | |
| 116 | + | |
| 117 | + send(live.pid, {:gtfs, to_string(source.id), :stop_times, 5, 10}) | |
| 118 | + html = render(live) | |
| 119 | + | |
| 120 | + assert html =~ "stop times" | |
| 121 | + assert html =~ ~s(value="50") | |
| 122 | + end | |
| 123 | + | |
| 124 | + test "the bundle steps are named the way the source page names them", %{conn: conn, source: source} do | |
| 125 | + {:ok, live, _html} = live(conn, "/cfg/offerings/work") | |
| 126 | + | |
| 127 | + send(live.pid, {:gtfs, to_string(source.id), :downloading, 1, 4}) | |
| 128 | + | |
| 129 | + assert render(live) =~ "retrieving bundle" | |
| 130 | + end | |
| 131 | + | |
| 132 | + test "finishing clears the bar rather than leaving it at its last value", %{conn: conn, source: source} do | |
| 133 | + {:ok, live, _html} = live(conn, "/cfg/offerings/work") | |
| 134 | + | |
| 135 | + send(live.pid, {:gtfs, to_string(source.id), :stop_times, 9, 10}) | |
| 136 | + assert render(live) =~ ~s(value="90") | |
| 137 | + | |
| 138 | + send(live.pid, {:gtfs, to_string(source.id), :done}) | |
| 139 | + html = render(live) | |
| 140 | + | |
| 141 | + refute html =~ ~s(value="90") | |
| 142 | + assert html =~ "Nothing is importing right now" | |
| 143 | + end | |
| 144 | + | |
| 145 | + test "a feed this user does not own is not shown as running", %{conn: conn, source: source} do | |
| 146 | + # The "gtfs" topic carries every source in the installation. Somebody | |
| 147 | + # else's import must not appear here, least of all under a name this page | |
| 148 | + # cannot resolve. | |
| 149 | + {:ok, live, _html} = live(conn, "/cfg/offerings/work") | |
| 150 | + | |
| 151 | + send(live.pid, {:gtfs, to_string(source.id + 9999), :stop_times, 5, 10}) | |
| 152 | + html = render(live) | |
| 153 | + | |
| 154 | + assert html =~ "Nothing is importing right now" | |
| 155 | + refute html =~ ~s(value="50") | |
| 156 | + end | |
| 157 | + | |
| 158 | + test "the running feed is named in its own card, and marked in the list", %{ | |
| 159 | + conn: conn, | |
| 160 | + source: source | |
| 161 | + } do | |
| 162 | + {:ok, live, _html} = live(conn, "/cfg/offerings/work") | |
| 163 | + | |
| 164 | + send(live.pid, {:gtfs, to_string(source.id), :stop_times, 5, 10}) | |
| 165 | + html = render(live) | |
| 166 | + | |
| 167 | + assert html =~ "Now running" | |
| 168 | + assert html =~ "running" | |
| 169 | + refute html =~ "Nothing is importing right now" | |
| 170 | + end | |
| 171 | + | |
| 172 | + test "a disabled source does not take the page down", %{conn: conn, source: source} do | |
| 173 | + {:ok, live, _html} = live(conn, "/cfg/offerings/work") | |
| 174 | + | |
| 175 | + send(live.pid, {:gtfs, to_string(source.id), :disabled}) | |
| 176 | + | |
| 177 | + assert render(live) =~ source.name | |
| 178 | + end | |
| 179 | + end | |
| 180 | + | |
| 181 | + test "another user's feeds are not listed", %{conn: conn} do | |
| 182 | + {:ok, stranger} = | |
| 183 | + Accounts.register_user(%{ | |
| 184 | + email: "other#{System.unique_integer([:positive])}@example.com", | |
| 185 | + password: "hello world!hello world!" | |
| 186 | + }) | |
| 187 | + | |
| 188 | + {:ok, theirs} = | |
| 189 | + Configuration.create_source(%{ | |
| 190 | + name: "Not Yours", notes: "", type: :gtfs, enabled: true, user_id: stranger.id, | |
| 191 | + config: %{"__type__" => "gtfs", "url" => "https://e.test/u.zip", "tz" => "UTC"} | |
| 192 | + }) | |
| 193 | + | |
| 194 | + {:ok, _live, html} = live(conn, "/cfg/offerings/work") | |
| 195 | + | |
| 196 | + refute html =~ theirs.name | |
| 197 | + end | |
| 198 | +end |
Parents: 8c2ccfc