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.
Sources: the import queue only lists the user's own jobs
503bcb9 · Gabe Morell · 2026-11-04 22:51
Message
{commit_body(@commit)}
Files changed
modified
apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/work.ex
+11
−6
@@ -1,6 +1,7 @@
| 1 | 1 | defmodule RoomSanctumWeb.SourceLive.Work do |
| 2 | 2 | @moduledoc """ |
| 3 | − What the GTFS import queue is doing, across every source at once. | |
| 3 | + What the GTFS import queue is doing, across every one of the user's sources | |
| 4 | + at once. | |
| 4 | 5 | |
| 5 | 6 | The source page shows one feed's progress bar, which is the right view when |
| 6 | 7 | you are looking at one feed. It is the wrong view for the question this |
@@ -182,14 +183,18 @@ defmodule RoomSanctumWeb.SourceLive.Work do
| 182 | 183 | socket |
| 183 | 184 | |> assign(:sources, Enum.sort_by(sources, & &1.name)) |
| 184 | 185 | |> assign(:names, names) |
| 185 | − |> assign(:jobs, jobs()) | |
| 186 | + |> assign(:jobs, jobs(Map.keys(names))) | |
| 186 | 187 | end |
| 187 | 188 | |
| 188 | − # The queue's own history. Fifty is enough to cover a night's run of every | |
| 189 | − # feed and its retries without turning the page into a log. | |
| 190 | − defp jobs do | |
| 189 | + # The queue's own history, for the user's own feeds only: the queue is | |
| 190 | + # shared, and its rows carry each feed's error text. Fifty is enough to cover | |
| 191 | + # a night's run of every feed and its retries without turning the page into | |
| 192 | + # a log. Compared as text, since a job's source_id may be stored either way. | |
| 193 | + defp jobs(source_ids) do | |
| 194 | + ids = Enum.map(source_ids, &to_string/1) | |
| 195 | + | |
| 191 | 196 | from(j in "oban_jobs", |
| 192 | − where: j.queue == "gtfs_import", | |
| 197 | + where: j.queue == "gtfs_import" and fragment("?->>'source_id'", j.args) in ^ids, | |
| 193 | 198 | order_by: [desc: j.id], |
| 194 | 199 | limit: 50, |
| 195 | 200 | select: %{ |
modified
apps/room_sanctum/lib/room_sanctum_web/live/cfg_source_live/work.html.heex
+1
−1
@@ -142,7 +142,7 @@
| 142 | 142 | <%= for job <- @jobs do %> |
| 143 | 143 | <% {label, class} = state_label(job.state) %> |
| 144 | 144 | <% sid = source_id(job) %> |
| 145 | − <tr> | |
| 145 | + <tr id={"job-#{job.id}"}> | |
| 146 | 146 | <td class="font-mono text-xs opacity-60"><%= job.id %></td> |
| 147 | 147 | <td> |
| 148 | 148 | <%= if sid && Map.get(@names, sid) do %> |
modified
apps/room_sanctum/test/room_sanctum_web/live/ownership_test.exs
+41
−24
@@ -188,6 +188,35 @@ defmodule RoomSanctumWeb.OwnershipTest do
| 188 | 188 | |> Enum.filter(&(is_nil(state) or &1 == state)) |
| 189 | 189 | end |
| 190 | 190 | |
| 191 | + defp import_job!(source, errors \\ []) do | |
| 192 | + now = NaiveDateTime.utc_now() |> NaiveDateTime.truncate(:second) | |
| 193 | + | |
| 194 | + {1, [%{id: job_id}]} = | |
| 195 | + Repo.insert_all( | |
| 196 | + "oban_jobs", | |
| 197 | + [ | |
| 198 | + %{ | |
| 199 | + state: "executing", | |
| 200 | + queue: "gtfs_import", | |
| 201 | + worker: "RoomGtfs.ImportJob", | |
| 202 | + args: %{"source_id" => source.id}, | |
| 203 | + errors: errors, | |
| 204 | + tags: [], | |
| 205 | + meta: %{}, | |
| 206 | + attempt: 1, | |
| 207 | + max_attempts: 3, | |
| 208 | + priority: 0, | |
| 209 | + inserted_at: now, | |
| 210 | + scheduled_at: now, | |
| 211 | + attempted_at: now | |
| 212 | + } | |
| 213 | + ], | |
| 214 | + returning: [:id] | |
| 215 | + ) | |
| 216 | + | |
| 217 | + job_id | |
| 218 | + end | |
| 219 | + | |
| 191 | 220 | describe "the source pages" do |
| 192 | 221 | test "will not show another user's source", ctx do |
| 193 | 222 | theirs = source_for(ctx.them, "Their markets") |
@@ -228,36 +257,24 @@ defmodule RoomSanctumWeb.OwnershipTest do
| 228 | 257 | |
| 229 | 258 | test "the import queue will not clear another user's job", ctx do |
| 230 | 259 | theirs = gtfs_source_for(ctx.them) |
| 231 | − now = NaiveDateTime.utc_now() |> NaiveDateTime.truncate(:second) | |
| 232 | − | |
| 233 | − {1, [%{id: job_id}]} = | |
| 234 | − Repo.insert_all( | |
| 235 | − "oban_jobs", | |
| 236 | − [ | |
| 237 | − %{ | |
| 238 | − state: "executing", | |
| 239 | − queue: "gtfs_import", | |
| 240 | − worker: "RoomGtfs.ImportJob", | |
| 241 | − args: %{"source_id" => theirs.id}, | |
| 242 | − errors: [], | |
| 243 | − tags: [], | |
| 244 | − meta: %{}, | |
| 245 | − attempt: 1, | |
| 246 | − max_attempts: 3, | |
| 247 | − priority: 0, | |
| 248 | − inserted_at: now, | |
| 249 | − scheduled_at: now, | |
| 250 | − attempted_at: now | |
| 251 | − } | |
| 252 | − ], | |
| 253 | − returning: [:id] | |
| 254 | − ) | |
| 260 | + job_id = import_job!(theirs) | |
| 255 | 261 | |
| 256 | 262 | {:ok, live, _html} = live(ctx.conn, "/cfg/offerings/work") |
| 257 | 263 | render_click(live, "clear", %{"id" => to_string(job_id)}) |
| 258 | 264 | |
| 259 | 265 | assert import_jobs(theirs.id) == ["executing"] |
| 260 | 266 | end |
| 267 | + | |
| 268 | + test "the import queue lists only the user's own jobs", ctx do | |
| 269 | + mine = import_job!(gtfs_source_for(ctx.me)) | |
| 270 | + theirs = import_job!(gtfs_source_for(ctx.them), [%{"error" => "their secret failure"}]) | |
| 271 | + | |
| 272 | + {:ok, live, _html} = live(ctx.conn, "/cfg/offerings/work") | |
| 273 | + | |
| 274 | + assert has_element?(live, "#job-#{mine}") | |
| 275 | + refute has_element?(live, "#job-#{theirs}") | |
| 276 | + refute render(live) =~ "their secret failure" | |
| 277 | + end | |
| 261 | 278 | end |
| 262 | 279 | |
| 263 | 280 | defp foci_for(user) do |
Parents: ac95475