From 48833f15be5579070a0b93eef34c401f2b97adbe Mon Sep 17 00:00:00 2001 From: Graham McIntire Date: Fri, 17 Apr 2026 09:24:57 -0500 Subject: [PATCH] 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:' 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. --- config/config.exs | 3 +- config/dev.exs | 3 +- config/runtime.exs | 3 +- lib/microwaveprop/radio/csv_import.ex | 163 ++++++++++++- lib/microwaveprop/radio/import_run.ex | 58 +++++ .../workers/contact_import_worker.ex | 151 ++++++++++++ lib/microwaveprop_web/live/submit_live.ex | 8 +- .../20260417141734_create_import_runs.exs | 25 ++ test/microwaveprop/radio/csv_import_test.exs | 88 +++++++ test/microwaveprop/radio/import_run_test.exs | 75 ++++++ .../workers/contact_import_worker_test.exs | 222 ++++++++++++++++++ 11 files changed, 788 insertions(+), 11 deletions(-) create mode 100644 lib/microwaveprop/radio/import_run.ex create mode 100644 lib/microwaveprop/workers/contact_import_worker.ex create mode 100644 priv/repo/migrations/20260417141734_create_import_runs.exs create mode 100644 test/microwaveprop/radio/import_run_test.exs create mode 100644 test/microwaveprop/workers/contact_import_worker_test.exs diff --git a/config/config.exs b/config/config.exs index b095dd43..6c2dc46d 100644 --- a/config/config.exs +++ b/config/config.exs @@ -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}, diff --git a/config/dev.exs b/config/dev.exs index f11f72ba..65f19dd1 100644 --- a/config/dev.exs +++ b/config/dev.exs @@ -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}, diff --git a/config/runtime.exs b/config/runtime.exs index b0884096..f667f991 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -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}, diff --git a/lib/microwaveprop/radio/csv_import.ex b/lib/microwaveprop/radio/csv_import.ex index a3fbb780..6adb101c 100644 --- a/lib/microwaveprop/radio/csv_import.ex +++ b/lib/microwaveprop/radio/csv_import.ex @@ -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:"` 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 diff --git a/lib/microwaveprop/radio/import_run.ex b/lib/microwaveprop/radio/import_run.ex new file mode 100644 index 00000000..fcd6a87f --- /dev/null +++ b/lib/microwaveprop/radio/import_run.ex @@ -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 diff --git a/lib/microwaveprop/workers/contact_import_worker.ex b/lib/microwaveprop/workers/contact_import_worker.ex new file mode 100644 index 00000000..81dd8fa9 --- /dev/null +++ b/lib/microwaveprop/workers/contact_import_worker.ex @@ -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:"` 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 diff --git a/lib/microwaveprop_web/live/submit_live.ex b/lib/microwaveprop_web/live/submit_live.ex index 40e3f893..a76c80f3 100644 --- a/lib/microwaveprop_web/live/submit_live.ex +++ b/lib/microwaveprop_web/live/submit_live.ex @@ -246,7 +246,7 @@ defmodule MicrowavepropWeb.SubmitLive do

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.

@@ -513,9 +513,9 @@ defmodule MicrowavepropWeb.SubmitLive do and FREQ or BAND -
  • Only microwave contacts (902 MHz and up) will be imported
    • -
  • Sub-microwave contacts are silently skipped
    • -
  • Frequencies are fuzzy-matched to the nearest microwave band
    • +
  • Contacts on amateur bands from 50 MHz and up will be imported
    • +
  • HF contacts (below 50 MHz) are silently skipped
    • +
  • Frequencies are fuzzy-matched to the nearest amateur band
    • diff --git a/priv/repo/migrations/20260417141734_create_import_runs.exs b/priv/repo/migrations/20260417141734_create_import_runs.exs new file mode 100644 index 00000000..cfb91ea0 --- /dev/null +++ b/priv/repo/migrations/20260417141734_create_import_runs.exs @@ -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 diff --git a/test/microwaveprop/radio/csv_import_test.exs b/test/microwaveprop/radio/csv_import_test.exs index 106ce5bd..fc6508ac 100644 --- a/test/microwaveprop/radio/csv_import_test.exs +++ b/test/microwaveprop/radio/csv_import_test.exs @@ -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 diff --git a/test/microwaveprop/radio/import_run_test.exs b/test/microwaveprop/radio/import_run_test.exs new file mode 100644 index 00000000..6a755ad8 --- /dev/null +++ b/test/microwaveprop/radio/import_run_test.exs @@ -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 diff --git a/test/microwaveprop/workers/contact_import_worker_test.exs b/test/microwaveprop/workers/contact_import_worker_test.exs new file mode 100644 index 00000000..6bbc0814 --- /dev/null +++ b/test/microwaveprop/workers/contact_import_worker_test.exs @@ -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: 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