feat(import): async CSV import with progress-tracking schema
For large CSVs (e.g. the 34k-row ARRL dump), the synchronous commit
path blocked for minutes and a crash mid-commit left no audit trail.
Refactor to run inserts + enrichment enqueues via Oban.
- New import_runs table: stores preview rows + running counters (total,
processed, imported, refined, error) + status + errors map + timing.
- New :contact_import Oban queue (limit 4) + ContactImportWorker.
Args: {import_run_id, offset, limit}. Each chunk (100 rows) inserts
its slice via CsvImport.commit_rows/1, atomically increments counters
in one update_all, flips status, and broadcasts progress on
'csv_import:<id>' PubSub topic.
- CsvImport.commit_rows/1 extracted from commit/1 (back-compat preserved).
New CsvImport.enqueue/2 persists the run + dispatches N chunk jobs.
Also: submit page copy updated from '902 MHz and up' to '50 MHz and up'
matching the band allowlist expansion done earlier.
LiveView UI for /imports/:id is a separate commit.
This commit is contained in:
parent
f10cad9b42
commit
48833f15be
11 changed files with 788 additions and 11 deletions
|
|
@ -69,7 +69,8 @@ config :microwaveprop, Oban,
|
|||
admin: 1,
|
||||
nexrad: 2,
|
||||
ionosphere: 1,
|
||||
space_weather: 1
|
||||
space_weather: 1,
|
||||
contact_import: 4
|
||||
],
|
||||
plugins: [
|
||||
{Oban.Plugins.Pruner, max_age: 3600 * 24},
|
||||
|
|
|
|||
|
|
@ -84,7 +84,8 @@ config :microwaveprop, Oban,
|
|||
iemre: 10,
|
||||
narr: 6,
|
||||
backfill_enqueue: 1,
|
||||
exports: 1
|
||||
exports: 1,
|
||||
contact_import: 4
|
||||
],
|
||||
plugins: [
|
||||
{Oban.Plugins.Pruner, max_age: 3600 * 24},
|
||||
|
|
|
|||
|
|
@ -181,7 +181,8 @@ if config_env() == :prod do
|
|||
nexrad: 2,
|
||||
ionosphere: 1,
|
||||
space_weather: 1,
|
||||
exports: 1
|
||||
exports: 1,
|
||||
contact_import: 4
|
||||
],
|
||||
plugins: [
|
||||
{Oban.Plugins.Pruner, max_age: 3600 * 24},
|
||||
|
|
|
|||
|
|
@ -26,10 +26,13 @@ defmodule Microwaveprop.Radio.CsvImport do
|
|||
alias Microwaveprop.Radio.BandResolver
|
||||
alias Microwaveprop.Radio.Contact
|
||||
alias Microwaveprop.Radio.ImportMatcher
|
||||
alias Microwaveprop.Radio.ImportRun
|
||||
alias Microwaveprop.Repo
|
||||
alias Microwaveprop.Workers.ContactImportWorker
|
||||
alias Microwaveprop.Workers.ContactWeatherEnqueueWorker
|
||||
|
||||
@dedup_window_seconds 3600
|
||||
@chunk_size 100
|
||||
|
||||
# Header parsing is case-insensitive and whitespace-insensitive. Accepts
|
||||
# both the raw field names (station1, grid1, ...) and the human labels the
|
||||
|
|
@ -149,6 +152,24 @@ defmodule Microwaveprop.Radio.CsvImport do
|
|||
end
|
||||
|
||||
def commit(%{valid: valid_rows, refinements: refinements}) do
|
||||
{:ok, commit_rows(%{valid: valid_rows, refinements: refinements})}
|
||||
end
|
||||
|
||||
@doc """
|
||||
Synchronously run a batch of CSV rows through the insert / refinement
|
||||
pipeline without wrapping the result in an `{:ok, _}` tuple. This is
|
||||
the worker-facing variant of `commit/1` and is called per-chunk by
|
||||
`Microwaveprop.Workers.ContactImportWorker`.
|
||||
|
||||
Accepts the same shape as `commit/1`: either a list of valid rows
|
||||
(legacy) or a `%{valid: [...], refinements: [...]}` map.
|
||||
"""
|
||||
@spec commit_rows([map()] | %{valid: [map()], refinements: [map()]}) :: commit_result()
|
||||
def commit_rows(valid_rows) when is_list(valid_rows) do
|
||||
commit_rows(%{valid: valid_rows, refinements: []})
|
||||
end
|
||||
|
||||
def commit_rows(%{valid: valid_rows, refinements: refinements}) do
|
||||
init = %{imported: [], refined: [], errors: []}
|
||||
|
||||
result =
|
||||
|
|
@ -156,14 +177,148 @@ defmodule Microwaveprop.Radio.CsvImport do
|
|||
|> apply_valid_rows(valid_rows)
|
||||
|> apply_refinements(refinements)
|
||||
|
||||
{:ok,
|
||||
%{
|
||||
imported: Enum.reverse(result.imported),
|
||||
refined: Enum.reverse(result.refined),
|
||||
errors: Enum.reverse(result.errors)
|
||||
}
|
||||
end
|
||||
|
||||
@doc """
|
||||
Persist the preview result as an `ImportRun` and enqueue one
|
||||
`ContactImportWorker` job per chunk of rows. Returns
|
||||
`{:ok, import_run_id}`. The caller can then subscribe to
|
||||
`"csv_import:<id>"` on `Microwaveprop.PubSub` to track progress.
|
||||
|
||||
The submitted rows are stored once on the `ImportRun` row as a JSONB
|
||||
blob `%{"rows" => [...], "refinement_ids" => [...]}`. Valid-row entries
|
||||
come first and refinements are appended after; `refinement_ids` holds
|
||||
the `row_num` values of the refinements so the worker can route each
|
||||
row to the correct pipeline without re-classifying.
|
||||
"""
|
||||
@spec enqueue(preview_result() | %{valid: [map()], refinements: [map()]}, String.t()) ::
|
||||
{:ok, Ecto.UUID.t()}
|
||||
def enqueue(preview, submitter_email) do
|
||||
valid = Map.get(preview, :valid, [])
|
||||
refinements = Map.get(preview, :refinements, [])
|
||||
|
||||
serialized_rows = Enum.map(valid, &serialize_valid_row/1) ++ Enum.map(refinements, &serialize_refinement_row/1)
|
||||
refinement_ids = Enum.map(refinements, & &1.row_num)
|
||||
total_rows = length(serialized_rows)
|
||||
|
||||
attrs = %{
|
||||
submitter_email: submitter_email,
|
||||
rows: %{"rows" => serialized_rows, "refinement_ids" => refinement_ids},
|
||||
total_rows: total_rows
|
||||
}
|
||||
|
||||
{:ok, run} =
|
||||
%ImportRun{}
|
||||
|> ImportRun.changeset(attrs)
|
||||
|> Repo.insert()
|
||||
|
||||
enqueue_chunks(run.id, total_rows)
|
||||
|
||||
{:ok, run.id}
|
||||
end
|
||||
|
||||
defp enqueue_chunks(_run_id, 0), do: :ok
|
||||
|
||||
defp enqueue_chunks(run_id, total_rows) do
|
||||
0
|
||||
|> Stream.iterate(&(&1 + @chunk_size))
|
||||
|> Stream.take_while(&(&1 < total_rows))
|
||||
|> Enum.each(fn offset ->
|
||||
limit = min(@chunk_size, total_rows - offset)
|
||||
|
||||
%{"import_run_id" => run_id, "offset" => offset, "limit" => limit}
|
||||
|> ContactImportWorker.new()
|
||||
|> Oban.insert!()
|
||||
end)
|
||||
end
|
||||
|
||||
defp serialize_valid_row(row) do
|
||||
%{
|
||||
"kind" => "insert",
|
||||
"row_num" => row.row_num,
|
||||
"attrs" => row.attrs,
|
||||
"timestamp" => DateTime.to_iso8601(row.timestamp)
|
||||
}
|
||||
end
|
||||
|
||||
defp serialize_refinement_row(row) do
|
||||
%{
|
||||
"kind" => "refinement",
|
||||
"row_num" => row.row_num,
|
||||
"attrs" => row.attrs,
|
||||
"timestamp" => DateTime.to_iso8601(row.timestamp),
|
||||
"existing_id" => row.existing_id,
|
||||
"changes" => stringify_changes(row.changes)
|
||||
}
|
||||
end
|
||||
|
||||
# The refinement map comes out of ImportMatcher with atom keys
|
||||
# (:grid1/:grid2/:mode). JSONB serialization only round-trips string
|
||||
# keys so normalize here before handing the row to the worker.
|
||||
defp stringify_changes(changes) when is_map(changes) do
|
||||
Map.new(changes, fn {k, v} -> {to_string(k), v} end)
|
||||
end
|
||||
|
||||
@doc """
|
||||
Decode a row out of the JSONB blob stored on `ImportRun.rows` back
|
||||
into the shape `commit_rows/1` expects.
|
||||
|
||||
Returns `{:valid, row}` for insert rows and `{:refinement, row}` for
|
||||
refinements so the worker can route each row to the right pipeline.
|
||||
"""
|
||||
@spec decode_row(map()) :: {:valid, map()} | {:refinement, map()}
|
||||
def decode_row(%{"kind" => "insert"} = row) do
|
||||
{:valid,
|
||||
%{
|
||||
imported: Enum.reverse(result.imported),
|
||||
refined: Enum.reverse(result.refined),
|
||||
errors: Enum.reverse(result.errors)
|
||||
row_num: row["row_num"],
|
||||
attrs: row["attrs"],
|
||||
timestamp: decode_timestamp(row["timestamp"])
|
||||
}}
|
||||
end
|
||||
|
||||
def decode_row(%{"kind" => "refinement"} = row) do
|
||||
{:refinement,
|
||||
%{
|
||||
row_num: row["row_num"],
|
||||
attrs: row["attrs"],
|
||||
timestamp: decode_timestamp(row["timestamp"]),
|
||||
existing_id: row["existing_id"],
|
||||
changes: atomize_changes(row["changes"])
|
||||
}}
|
||||
end
|
||||
|
||||
defp decode_timestamp(nil), do: nil
|
||||
|
||||
defp decode_timestamp(iso) when is_binary(iso) do
|
||||
case DateTime.from_iso8601(iso) do
|
||||
{:ok, dt, _} -> dt
|
||||
_ -> nil
|
||||
end
|
||||
end
|
||||
|
||||
@refinement_change_keys ~w(grid1 grid2 mode band station1 station2 qso_timestamp)
|
||||
defp atomize_changes(changes) when is_map(changes) do
|
||||
Map.new(changes, fn {k, v} ->
|
||||
key_str = to_string(k)
|
||||
|
||||
if key_str in @refinement_change_keys do
|
||||
{String.to_existing_atom(key_str), v}
|
||||
else
|
||||
{key_str, v}
|
||||
end
|
||||
end)
|
||||
end
|
||||
|
||||
defp atomize_changes(_), do: %{}
|
||||
|
||||
@spec chunk_size() :: pos_integer()
|
||||
def chunk_size, do: @chunk_size
|
||||
|
||||
defp apply_valid_rows(acc, valid_rows) do
|
||||
Enum.reduce(valid_rows, acc, fn row, acc ->
|
||||
case Radio.create_contact(row.attrs) do
|
||||
|
|
|
|||
58
lib/microwaveprop/radio/import_run.ex
Normal file
58
lib/microwaveprop/radio/import_run.ex
Normal file
|
|
@ -0,0 +1,58 @@
|
|||
defmodule Microwaveprop.Radio.ImportRun do
|
||||
@moduledoc """
|
||||
Tracks an async CSV import: the full set of rows to process, running
|
||||
counters the `ContactImportWorker` increments as it processes chunks,
|
||||
and a status that moves from `"pending"` → `"running"` → `"complete"`
|
||||
(or `"failed"`).
|
||||
|
||||
The `rows` field is a JSONB blob shaped as `%{"rows" => [...], "refinement_ids" => [...]}`
|
||||
so workers can tell which rows are new-contact inserts vs refinements
|
||||
of an existing contact.
|
||||
"""
|
||||
use Ecto.Schema
|
||||
|
||||
import Ecto.Changeset
|
||||
|
||||
@primary_key {:id, :binary_id, autogenerate: true}
|
||||
@foreign_key_type :binary_id
|
||||
|
||||
@statuses ~w(pending running complete failed)
|
||||
|
||||
schema "import_runs" do
|
||||
field :submitter_email, :string
|
||||
field :rows, :map
|
||||
field :total_rows, :integer
|
||||
field :processed_rows, :integer, default: 0
|
||||
field :imported_count, :integer, default: 0
|
||||
field :refined_count, :integer, default: 0
|
||||
field :error_count, :integer, default: 0
|
||||
field :errors, :map, default: %{}
|
||||
field :status, :string, default: "pending"
|
||||
field :started_at, :utc_datetime_usec
|
||||
field :completed_at, :utc_datetime_usec
|
||||
|
||||
timestamps(type: :utc_datetime_usec)
|
||||
end
|
||||
|
||||
@type t :: %__MODULE__{}
|
||||
|
||||
@cast_fields ~w(submitter_email rows total_rows processed_rows imported_count
|
||||
refined_count error_count errors status started_at completed_at)a
|
||||
@required_fields ~w(submitter_email rows total_rows)a
|
||||
|
||||
@spec statuses() :: [String.t()]
|
||||
def statuses, do: @statuses
|
||||
|
||||
@spec changeset(t() | Ecto.Changeset.t(), map()) :: Ecto.Changeset.t()
|
||||
def changeset(run, attrs) do
|
||||
run
|
||||
|> cast(attrs, @cast_fields)
|
||||
|> validate_required(@required_fields)
|
||||
|> validate_inclusion(:status, @statuses)
|
||||
|> validate_number(:total_rows, greater_than_or_equal_to: 0)
|
||||
|> validate_number(:processed_rows, greater_than_or_equal_to: 0)
|
||||
|> validate_number(:imported_count, greater_than_or_equal_to: 0)
|
||||
|> validate_number(:refined_count, greater_than_or_equal_to: 0)
|
||||
|> validate_number(:error_count, greater_than_or_equal_to: 0)
|
||||
end
|
||||
end
|
||||
151
lib/microwaveprop/workers/contact_import_worker.ex
Normal file
151
lib/microwaveprop/workers/contact_import_worker.ex
Normal file
|
|
@ -0,0 +1,151 @@
|
|||
defmodule Microwaveprop.Workers.ContactImportWorker do
|
||||
@moduledoc """
|
||||
Processes one chunk of rows out of an `ImportRun` — the async half of
|
||||
the CSV import flow.
|
||||
|
||||
Each job receives an `import_run_id`, an `offset`, and a `limit`. The
|
||||
worker loads the run, takes that slice of rows, routes them through
|
||||
`CsvImport.commit_rows/1`, atomically bumps the per-run counters, and
|
||||
broadcasts the updated run on `"csv_import:<id>"` so the submitting
|
||||
LiveView can reflect progress.
|
||||
|
||||
Counters are updated with a single `UPDATE ... SET col = col + ?`
|
||||
query so concurrent workers finishing at the same time can't clobber
|
||||
each other. The worker that pushes `processed_rows` past `total_rows`
|
||||
flips `status` to `"complete"` and sets `completed_at`; any worker
|
||||
that sees `status = 'pending'` on its first increment flips it to
|
||||
`"running"` and sets `started_at`.
|
||||
"""
|
||||
use Oban.Worker, queue: :contact_import, max_attempts: 3
|
||||
|
||||
import Ecto.Query
|
||||
|
||||
alias Microwaveprop.Radio.CsvImport
|
||||
alias Microwaveprop.Radio.ImportRun
|
||||
alias Microwaveprop.Repo
|
||||
|
||||
require Logger
|
||||
|
||||
@pubsub Microwaveprop.PubSub
|
||||
|
||||
@impl Oban.Worker
|
||||
def perform(%Oban.Job{args: %{"import_run_id" => run_id, "offset" => offset, "limit" => limit}}) do
|
||||
case Repo.get(ImportRun, run_id) do
|
||||
nil ->
|
||||
Logger.warning("ContactImportWorker: import_run #{run_id} not found; discarding chunk")
|
||||
:ok
|
||||
|
||||
%ImportRun{} = run ->
|
||||
process_chunk(run, offset, limit)
|
||||
end
|
||||
end
|
||||
|
||||
defp process_chunk(run, offset, limit) do
|
||||
all_rows = get_in(run.rows, ["rows"]) || []
|
||||
chunk = Enum.slice(all_rows, offset, limit)
|
||||
|
||||
commit_input = build_commit_input(chunk)
|
||||
commit_result = CsvImport.commit_rows(commit_input)
|
||||
|
||||
updates = build_counter_updates(commit_result, length(chunk))
|
||||
updated_run = apply_chunk_updates(run.id, updates)
|
||||
|
||||
broadcast(run.id, updated_run)
|
||||
|
||||
:ok
|
||||
end
|
||||
|
||||
defp build_commit_input(chunk) do
|
||||
{valid, refinements} =
|
||||
Enum.reduce(chunk, {[], []}, fn row, {valid, refinements} ->
|
||||
case CsvImport.decode_row(row) do
|
||||
{:valid, decoded} -> {[decoded | valid], refinements}
|
||||
{:refinement, decoded} -> {valid, [decoded | refinements]}
|
||||
end
|
||||
end)
|
||||
|
||||
%{valid: Enum.reverse(valid), refinements: Enum.reverse(refinements)}
|
||||
end
|
||||
|
||||
defp build_counter_updates(commit_result, chunk_len) do
|
||||
errors_map =
|
||||
Map.new(commit_result.errors, fn {row_num, messages} ->
|
||||
{to_string(row_num), messages}
|
||||
end)
|
||||
|
||||
%{
|
||||
chunk_len: chunk_len,
|
||||
imported_count: length(commit_result.imported),
|
||||
refined_count: length(commit_result.refined),
|
||||
error_count: length(commit_result.errors),
|
||||
errors_map: errors_map
|
||||
}
|
||||
end
|
||||
|
||||
# Atomic per-chunk update. All counter fields are incremented in one
|
||||
# UPDATE so concurrent chunk-finishers can't clobber each other, and
|
||||
# the status / timestamp transitions are expressed as CASEs off the
|
||||
# pre-increment values.
|
||||
defp apply_chunk_updates(run_id, u) do
|
||||
now = DateTime.utc_now()
|
||||
chunk_len = u.chunk_len
|
||||
imported_count = u.imported_count
|
||||
refined_count = u.refined_count
|
||||
error_count = u.error_count
|
||||
errors_map = u.errors_map
|
||||
|
||||
query =
|
||||
from(r in ImportRun,
|
||||
where: r.id == ^run_id,
|
||||
update: [
|
||||
set: [
|
||||
updated_at: ^now,
|
||||
processed_rows: fragment("? + ?", r.processed_rows, ^chunk_len),
|
||||
imported_count: fragment("? + ?", r.imported_count, ^imported_count),
|
||||
refined_count: fragment("? + ?", r.refined_count, ^refined_count),
|
||||
error_count: fragment("? + ?", r.error_count, ^error_count),
|
||||
status:
|
||||
fragment(
|
||||
"""
|
||||
CASE
|
||||
WHEN ? + ? >= ? THEN 'complete'
|
||||
WHEN ? = 'pending' THEN 'running'
|
||||
ELSE ?
|
||||
END
|
||||
""",
|
||||
r.processed_rows,
|
||||
^chunk_len,
|
||||
r.total_rows,
|
||||
r.status,
|
||||
r.status
|
||||
),
|
||||
started_at:
|
||||
fragment(
|
||||
"CASE WHEN ? IS NULL THEN ? ELSE ? END",
|
||||
r.started_at,
|
||||
^now,
|
||||
r.started_at
|
||||
),
|
||||
completed_at:
|
||||
fragment(
|
||||
"CASE WHEN ? + ? >= ? THEN ? ELSE ? END",
|
||||
r.processed_rows,
|
||||
^chunk_len,
|
||||
r.total_rows,
|
||||
^now,
|
||||
r.completed_at
|
||||
),
|
||||
errors: fragment("? || ?::jsonb", r.errors, ^errors_map)
|
||||
]
|
||||
]
|
||||
)
|
||||
|
||||
{1, _} = Repo.update_all(query, [])
|
||||
|
||||
Repo.get!(ImportRun, run_id)
|
||||
end
|
||||
|
||||
defp broadcast(run_id, run) do
|
||||
Phoenix.PubSub.broadcast(@pubsub, "csv_import:#{run_id}", {:import_progress, run_id, run})
|
||||
end
|
||||
end
|
||||
|
|
@ -246,7 +246,7 @@ defmodule MicrowavepropWeb.SubmitLive do
|
|||
<p>
|
||||
Our propagation model improves with real-world data. Each verified contact
|
||||
you submit is matched against atmospheric conditions at the time of your contact,
|
||||
helping us calibrate predictions for all amateur bands from 902 MHz and up. The
|
||||
helping us calibrate predictions for all amateur bands from 50 MHz and up. The
|
||||
more contacts we have, the better our forecasts get for everyone.
|
||||
</p>
|
||||
</div>
|
||||
|
|
@ -513,9 +513,9 @@ defmodule MicrowavepropWeb.SubmitLive do
|
|||
and <code>FREQ</code>
|
||||
or <code>BAND</code>
|
||||
</li>
|
||||
<li>Only microwave contacts (902 MHz and up) will be imported</li>
|
||||
<li>Sub-microwave contacts are silently skipped</li>
|
||||
<li>Frequencies are fuzzy-matched to the nearest microwave band</li>
|
||||
<li>Contacts on amateur bands from 50 MHz and up will be imported</li>
|
||||
<li>HF contacts (below 50 MHz) are silently skipped</li>
|
||||
<li>Frequencies are fuzzy-matched to the nearest amateur band</li>
|
||||
</ul>
|
||||
</div>
|
||||
</div>
|
||||
|
|
|
|||
25
priv/repo/migrations/20260417141734_create_import_runs.exs
Normal file
25
priv/repo/migrations/20260417141734_create_import_runs.exs
Normal file
|
|
@ -0,0 +1,25 @@
|
|||
defmodule Microwaveprop.Repo.Migrations.CreateImportRuns do
|
||||
use Ecto.Migration
|
||||
|
||||
def change do
|
||||
create table(:import_runs, primary_key: false) do
|
||||
add :id, :binary_id, primary_key: true
|
||||
add :submitter_email, :string, null: false
|
||||
add :rows, :map, null: false
|
||||
add :total_rows, :integer, null: false
|
||||
add :processed_rows, :integer, null: false, default: 0
|
||||
add :imported_count, :integer, null: false, default: 0
|
||||
add :refined_count, :integer, null: false, default: 0
|
||||
add :error_count, :integer, null: false, default: 0
|
||||
add :errors, :map, null: false, default: %{}
|
||||
add :status, :string, null: false, default: "pending"
|
||||
add :started_at, :utc_datetime_usec
|
||||
add :completed_at, :utc_datetime_usec
|
||||
|
||||
timestamps(type: :utc_datetime_usec)
|
||||
end
|
||||
|
||||
create index(:import_runs, [:submitter_email])
|
||||
create index(:import_runs, [:status])
|
||||
end
|
||||
end
|
||||
|
|
@ -1,5 +1,6 @@
|
|||
defmodule Microwaveprop.Radio.CsvImportTest do
|
||||
use Microwaveprop.DataCase, async: true
|
||||
use Oban.Testing, repo: Microwaveprop.Repo
|
||||
|
||||
alias Microwaveprop.Radio.Contact
|
||||
alias Microwaveprop.Radio.CsvImport
|
||||
|
|
@ -863,4 +864,91 @@ defmodule Microwaveprop.Radio.CsvImportTest do
|
|||
assert result.refined == []
|
||||
end
|
||||
end
|
||||
|
||||
describe "enqueue/2" do
|
||||
alias Microwaveprop.Radio.ImportRun
|
||||
alias Microwaveprop.Workers.ContactImportWorker
|
||||
|
||||
# Uses testing: :inline — jobs run as they're inserted. We inspect
|
||||
# the post-run state (which exercises the worker too) and then
|
||||
# reset-and-re-count to verify the pre-run shape separately.
|
||||
test "creates an ImportRun with all rows serialized" do
|
||||
valid = [
|
||||
%{
|
||||
row_num: 2,
|
||||
attrs: %{
|
||||
"station1" => "W5XD",
|
||||
"station2" => "K5TR",
|
||||
"grid1" => "EM12",
|
||||
"grid2" => "EM00",
|
||||
"band" => "10000",
|
||||
"mode" => "CW",
|
||||
"qso_timestamp" => "2026-03-28T18:00:00Z",
|
||||
"submitter_email" => @submitter
|
||||
},
|
||||
timestamp: ~U[2026-03-28 18:00:00Z]
|
||||
}
|
||||
]
|
||||
|
||||
preview = %{valid: valid, refinements: []}
|
||||
|
||||
assert {:ok, run_id} = CsvImport.enqueue(preview, @submitter)
|
||||
run = Repo.get!(ImportRun, run_id)
|
||||
assert run.submitter_email == @submitter
|
||||
assert run.total_rows == 1
|
||||
assert [%{"kind" => "insert", "row_num" => 2}] = run.rows["rows"]
|
||||
assert run.rows["refinement_ids"] == []
|
||||
end
|
||||
|
||||
test "enqueues one ContactImportWorker job per chunk" do
|
||||
# 250 rows -> 3 jobs at chunk_size 100.
|
||||
valid =
|
||||
for i <- 0..249 do
|
||||
%{
|
||||
row_num: i + 2,
|
||||
attrs: %{
|
||||
"station1" => "W5XD",
|
||||
"station2" => "K5TR",
|
||||
"grid1" => "EM12",
|
||||
"grid2" => "EM00",
|
||||
"band" => "10000",
|
||||
"mode" => "CW",
|
||||
"qso_timestamp" => "2026-03-28T18:00:00Z",
|
||||
"submitter_email" => @submitter
|
||||
},
|
||||
timestamp: ~U[2026-03-28 18:00:00Z]
|
||||
}
|
||||
end
|
||||
|
||||
preview = %{valid: valid, refinements: []}
|
||||
|
||||
{:ok, _run_id} =
|
||||
Oban.Testing.with_testing_mode(:manual, fn ->
|
||||
CsvImport.enqueue(preview, @submitter)
|
||||
end)
|
||||
|
||||
# Three chunks: [0, 100), [100, 200), [200, 250).
|
||||
assert_enqueued(worker: ContactImportWorker, args: %{"offset" => 0, "limit" => 100})
|
||||
assert_enqueued(worker: ContactImportWorker, args: %{"offset" => 100, "limit" => 100})
|
||||
assert_enqueued(worker: ContactImportWorker, args: %{"offset" => 200, "limit" => 50})
|
||||
assert length(all_enqueued(worker: ContactImportWorker)) == 3
|
||||
end
|
||||
|
||||
test "returns pre-run counters and status on the persisted run" do
|
||||
# Fresh empty run so counters show their defaults regardless of
|
||||
# what the inline worker does.
|
||||
preview = %{valid: [], refinements: []}
|
||||
|
||||
assert {:ok, run_id} = CsvImport.enqueue(preview, @submitter)
|
||||
run = Repo.get!(ImportRun, run_id)
|
||||
assert run.total_rows == 0
|
||||
assert run.processed_rows == 0
|
||||
assert run.imported_count == 0
|
||||
assert run.refined_count == 0
|
||||
assert run.error_count == 0
|
||||
assert run.errors == %{}
|
||||
# No rows, no jobs enqueued — status stays pending.
|
||||
assert run.status == "pending"
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
75
test/microwaveprop/radio/import_run_test.exs
Normal file
75
test/microwaveprop/radio/import_run_test.exs
Normal file
|
|
@ -0,0 +1,75 @@
|
|||
defmodule Microwaveprop.Radio.ImportRunTest do
|
||||
use Microwaveprop.DataCase, async: true
|
||||
|
||||
alias Microwaveprop.Radio.ImportRun
|
||||
|
||||
@valid_attrs %{
|
||||
submitter_email: "test@example.com",
|
||||
rows: %{"rows" => [], "refinement_ids" => []},
|
||||
total_rows: 0
|
||||
}
|
||||
|
||||
describe "changeset/2" do
|
||||
test "accepts a minimal valid attrs map" do
|
||||
changeset = ImportRun.changeset(%ImportRun{}, @valid_attrs)
|
||||
assert changeset.valid?
|
||||
end
|
||||
|
||||
test "defaults status to pending via schema" do
|
||||
run = %ImportRun{}
|
||||
assert run.status == "pending"
|
||||
assert run.processed_rows == 0
|
||||
assert run.imported_count == 0
|
||||
assert run.refined_count == 0
|
||||
assert run.error_count == 0
|
||||
assert run.errors == %{}
|
||||
end
|
||||
|
||||
test "requires submitter_email" do
|
||||
changeset = ImportRun.changeset(%ImportRun{}, Map.delete(@valid_attrs, :submitter_email))
|
||||
refute changeset.valid?
|
||||
assert "can't be blank" in errors_on(changeset).submitter_email
|
||||
end
|
||||
|
||||
test "requires rows" do
|
||||
changeset = ImportRun.changeset(%ImportRun{}, Map.delete(@valid_attrs, :rows))
|
||||
refute changeset.valid?
|
||||
assert "can't be blank" in errors_on(changeset).rows
|
||||
end
|
||||
|
||||
test "requires total_rows" do
|
||||
changeset = ImportRun.changeset(%ImportRun{}, Map.delete(@valid_attrs, :total_rows))
|
||||
refute changeset.valid?
|
||||
assert "can't be blank" in errors_on(changeset).total_rows
|
||||
end
|
||||
|
||||
test "rejects a bad status" do
|
||||
changeset = ImportRun.changeset(%ImportRun{}, Map.put(@valid_attrs, :status, "weird"))
|
||||
refute changeset.valid?
|
||||
assert errors_on(changeset).status != []
|
||||
end
|
||||
|
||||
test "accepts every declared status" do
|
||||
for status <- ImportRun.statuses() do
|
||||
changeset = ImportRun.changeset(%ImportRun{}, Map.put(@valid_attrs, :status, status))
|
||||
assert changeset.valid?, "expected #{inspect(status)} to be valid"
|
||||
end
|
||||
end
|
||||
|
||||
test "round-trips through the database" do
|
||||
assert {:ok, run} =
|
||||
%ImportRun{}
|
||||
|> ImportRun.changeset(%{
|
||||
submitter_email: "a@example.com",
|
||||
rows: %{"rows" => [%{"row_num" => 2}], "refinement_ids" => []},
|
||||
total_rows: 1
|
||||
})
|
||||
|> Repo.insert()
|
||||
|
||||
reloaded = Repo.get!(ImportRun, run.id)
|
||||
assert reloaded.status == "pending"
|
||||
assert reloaded.total_rows == 1
|
||||
assert reloaded.rows == %{"rows" => [%{"row_num" => 2}], "refinement_ids" => []}
|
||||
end
|
||||
end
|
||||
end
|
||||
222
test/microwaveprop/workers/contact_import_worker_test.exs
Normal file
222
test/microwaveprop/workers/contact_import_worker_test.exs
Normal file
|
|
@ -0,0 +1,222 @@
|
|||
defmodule Microwaveprop.Workers.ContactImportWorkerTest do
|
||||
use Microwaveprop.DataCase, async: false
|
||||
|
||||
alias Microwaveprop.Radio.Contact
|
||||
alias Microwaveprop.Radio.CsvImport
|
||||
alias Microwaveprop.Radio.ImportRun
|
||||
alias Microwaveprop.Terrain.ElevationClient
|
||||
alias Microwaveprop.Weather.HrrrClient
|
||||
alias Microwaveprop.Weather.IemClient
|
||||
alias Microwaveprop.Workers.ContactImportWorker
|
||||
|
||||
@valid_header "station1,station2,grid1,grid2,band,mode,qso_timestamp"
|
||||
@submitter "test@example.com"
|
||||
|
||||
setup do
|
||||
Req.Test.stub(IemClient, fn conn ->
|
||||
case conn.request_path do
|
||||
"/cgi-bin/request/asos.py" -> Req.Test.text(conn, "#DEBUG,\nstation,valid,tmpf\n")
|
||||
_ -> Req.Test.json(conn, %{"profiles" => []})
|
||||
end
|
||||
end)
|
||||
|
||||
Req.Test.stub(HrrrClient, fn conn ->
|
||||
Plug.Conn.send_resp(conn, 404, "not found")
|
||||
end)
|
||||
|
||||
Req.Test.stub(ElevationClient, fn conn ->
|
||||
params = Plug.Conn.fetch_query_params(conn).query_params
|
||||
lat_count = params["latitude"] |> String.split(",") |> length()
|
||||
Req.Test.json(conn, %{"elevation" => List.duplicate(200.0, lat_count)})
|
||||
end)
|
||||
|
||||
:ok
|
||||
end
|
||||
|
||||
defp build_csv(rows) do
|
||||
Enum.join([@valid_header | rows], "\n")
|
||||
end
|
||||
|
||||
defp insert_preview(csv) do
|
||||
{:ok, preview} = CsvImport.preview(csv, @submitter)
|
||||
{:ok, run_id} = CsvImport.enqueue(%{valid: preview.valid, refinements: preview.refinements}, @submitter)
|
||||
# enqueue/2 also inserts jobs — but with testing: :inline those jobs already
|
||||
# ran. We want a fresh run for the unit test, so delete the existing one
|
||||
# and re-create with the same shape.
|
||||
:ok = reset_run(run_id, preview)
|
||||
run_id
|
||||
end
|
||||
|
||||
# Reset counters on the just-created run so we can drive the worker manually.
|
||||
defp reset_run(run_id, _preview) do
|
||||
{1, _} =
|
||||
Repo.update_all(from(r in ImportRun, where: r.id == ^run_id),
|
||||
set: [
|
||||
processed_rows: 0,
|
||||
imported_count: 0,
|
||||
refined_count: 0,
|
||||
error_count: 0,
|
||||
errors: %{},
|
||||
status: "pending",
|
||||
started_at: nil,
|
||||
completed_at: nil
|
||||
]
|
||||
)
|
||||
|
||||
Repo.delete_all(Contact)
|
||||
:ok
|
||||
end
|
||||
|
||||
describe "perform/1 happy path" do
|
||||
test "processes a chunk, inserts contacts, and bumps counters" do
|
||||
csv =
|
||||
build_csv([
|
||||
"W5XD,K5TR,EM12,EM00,10000,CW,2026-03-28T18:00:00Z",
|
||||
"N5AC,W5LUA,EM13,EM20,24000,SSB,2026-03-28T19:00:00Z"
|
||||
])
|
||||
|
||||
run_id = insert_preview(csv)
|
||||
|
||||
assert :ok =
|
||||
ContactImportWorker.perform(%Oban.Job{
|
||||
args: %{"import_run_id" => run_id, "offset" => 0, "limit" => 2}
|
||||
})
|
||||
|
||||
run = Repo.get!(ImportRun, run_id)
|
||||
assert run.processed_rows == 2
|
||||
assert run.imported_count == 2
|
||||
assert run.error_count == 0
|
||||
assert run.status == "complete"
|
||||
assert run.completed_at
|
||||
assert run.started_at
|
||||
|
||||
assert Repo.aggregate(Contact, :count) == 2
|
||||
end
|
||||
|
||||
test "flips status to running on first chunk, complete on last" do
|
||||
csv =
|
||||
build_csv([
|
||||
"W5XD,K5TR,EM12,EM00,10000,CW,2026-03-28T18:00:00Z",
|
||||
"N5AC,W5LUA,EM13,EM20,24000,SSB,2026-03-28T19:00:00Z",
|
||||
"W5XD,W5LUA,EM12,EM20,1296,CW,2026-03-28T20:00:00Z"
|
||||
])
|
||||
|
||||
run_id = insert_preview(csv)
|
||||
|
||||
# First chunk of 1: marks running.
|
||||
:ok =
|
||||
ContactImportWorker.perform(%Oban.Job{
|
||||
args: %{"import_run_id" => run_id, "offset" => 0, "limit" => 1}
|
||||
})
|
||||
|
||||
run = Repo.get!(ImportRun, run_id)
|
||||
assert run.status == "running"
|
||||
assert run.processed_rows == 1
|
||||
assert run.started_at
|
||||
refute run.completed_at
|
||||
|
||||
# Final chunk: marks complete.
|
||||
:ok =
|
||||
ContactImportWorker.perform(%Oban.Job{
|
||||
args: %{"import_run_id" => run_id, "offset" => 1, "limit" => 2}
|
||||
})
|
||||
|
||||
run = Repo.get!(ImportRun, run_id)
|
||||
assert run.status == "complete"
|
||||
assert run.processed_rows == 3
|
||||
assert run.completed_at
|
||||
end
|
||||
end
|
||||
|
||||
describe "perform/1 errors" do
|
||||
test "a failing row counts into error_count without blocking siblings" do
|
||||
# First row has an existing contact we will try to re-insert via a
|
||||
# forced-bad row path; simpler: create a unique DB constraint failure
|
||||
# by inserting a duplicate row manually. We force a changeset error
|
||||
# by stuffing an invalid grid into a row — preview() rejects it, so
|
||||
# skip preview and hand the worker a hand-rolled ImportRun.
|
||||
rows = [
|
||||
%{
|
||||
"kind" => "insert",
|
||||
"row_num" => 2,
|
||||
"attrs" => %{
|
||||
"station1" => "W5XD",
|
||||
"station2" => "K5TR",
|
||||
"grid1" => "ZZZZ",
|
||||
"grid2" => "EM00",
|
||||
"band" => "10000",
|
||||
"mode" => "CW",
|
||||
"qso_timestamp" => "2026-03-28T18:00:00Z",
|
||||
"submitter_email" => @submitter
|
||||
},
|
||||
"timestamp" => "2026-03-28T18:00:00Z"
|
||||
},
|
||||
%{
|
||||
"kind" => "insert",
|
||||
"row_num" => 3,
|
||||
"attrs" => %{
|
||||
"station1" => "N5AC",
|
||||
"station2" => "W5LUA",
|
||||
"grid1" => "EM13",
|
||||
"grid2" => "EM20",
|
||||
"band" => "24000",
|
||||
"mode" => "SSB",
|
||||
"qso_timestamp" => "2026-03-28T19:00:00Z",
|
||||
"submitter_email" => @submitter
|
||||
},
|
||||
"timestamp" => "2026-03-28T19:00:00Z"
|
||||
}
|
||||
]
|
||||
|
||||
{:ok, run} =
|
||||
%ImportRun{}
|
||||
|> ImportRun.changeset(%{
|
||||
submitter_email: @submitter,
|
||||
rows: %{"rows" => rows, "refinement_ids" => []},
|
||||
total_rows: 2
|
||||
})
|
||||
|> Repo.insert()
|
||||
|
||||
:ok =
|
||||
ContactImportWorker.perform(%Oban.Job{
|
||||
args: %{"import_run_id" => run.id, "offset" => 0, "limit" => 2}
|
||||
})
|
||||
|
||||
reloaded = Repo.get!(ImportRun, run.id)
|
||||
assert reloaded.processed_rows == 2
|
||||
assert reloaded.imported_count == 1
|
||||
assert reloaded.error_count == 1
|
||||
assert reloaded.status == "complete"
|
||||
assert Map.has_key?(reloaded.errors, "2")
|
||||
assert [msg | _] = reloaded.errors["2"]
|
||||
assert is_binary(msg)
|
||||
end
|
||||
end
|
||||
|
||||
describe "broadcast" do
|
||||
test "publishes progress to csv_import:<id> after each chunk" do
|
||||
csv = build_csv(["W5XD,K5TR,EM12,EM00,10000,CW,2026-03-28T18:00:00Z"])
|
||||
run_id = insert_preview(csv)
|
||||
|
||||
:ok = Phoenix.PubSub.subscribe(Microwaveprop.PubSub, "csv_import:#{run_id}")
|
||||
|
||||
:ok =
|
||||
ContactImportWorker.perform(%Oban.Job{
|
||||
args: %{"import_run_id" => run_id, "offset" => 0, "limit" => 1}
|
||||
})
|
||||
|
||||
assert_receive {:import_progress, ^run_id, %ImportRun{status: "complete", processed_rows: 1}}
|
||||
end
|
||||
end
|
||||
|
||||
describe "missing run" do
|
||||
test "returns :ok for a deleted import run" do
|
||||
run_id = Ecto.UUID.generate()
|
||||
|
||||
assert :ok =
|
||||
ContactImportWorker.perform(%Oban.Job{
|
||||
args: %{"import_run_id" => run_id, "offset" => 0, "limit" => 1}
|
||||
})
|
||||
end
|
||||
end
|
||||
end
|
||||
Loading…
Add table
Reference in a new issue