Graham McIntire
48833f15be
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.
58 lines
2 KiB
Elixir
58 lines
2 KiB
Elixir
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
|