prop/lib/microwaveprop/radio/csv_import.ex

defmodule Microwaveprop.Radio.CsvImport do
  @moduledoc """
  Parse a CSV of contacts into a preview (validated + de-duplicated) and then
  commit those rows into the database.

  Upload flow:

      {:ok, preview} = CsvImport.preview(csv_string, submitter_email)
      # show preview.valid / preview.invalid / preview.duplicates to the user
      {:ok, result} = CsvImport.commit(preview.valid)

  Two rows count as duplicates of each other if they share:

    * the same pair of callsigns (direction-agnostic),
    * the same grids for those callsigns,
    * the same band, and
    * timestamps within 60 minutes of each other.

  Dedup runs both against existing DB contacts and against earlier rows in
  the same CSV (first row wins, subsequent matches are flagged).
  """

  import Ecto.Query, only: [from: 2]

  alias Microwaveprop.Radio
  alias Microwaveprop.Radio.Contact
  alias Microwaveprop.Repo
  alias Microwaveprop.Workers.ContactWeatherEnqueueWorker

  @dedup_window_seconds 3600

  @columns_with_mode ~w(station1 station2 grid1 grid2 band mode qso_timestamp)
  @columns_without_mode ~w(station1 station2 grid1 grid2 band qso_timestamp)

  @doc """
  Parse-and-insert in one shot. Kept for backward compatibility — does NOT
  perform duplicate detection. New code should use `preview/2` + `commit/1`.
  """
  @type preview_result :: %{
          valid: [map()],
          invalid: [map()],
          duplicates: [map()],
          total_rows: non_neg_integer(),
          submitter_email: String.t()
        }

  @type import_result :: %{
          imported: [Contact.t()],
          errors: [{pos_integer(), [String.t()]}]
        }

  @type commit_result :: %{
          imported: [Contact.t()],
          errors: [{pos_integer(), [String.t()]}]
        }

  @spec import(String.t(), String.t()) ::
          {:ok, import_result()} | {:error, :empty_csv} | {:error, :no_data_rows}
  def import(csv_string, submitter_email) do
    with {:ok, rows} <- parse_csv(csv_string, submitter_email) do
      {valid, invalid} = split_parsed(rows)
      {:ok, commit_result} = commit(valid)
      invalid_errors = Enum.map(invalid, fn %{row_num: n, messages: m} -> {n, m} end)

      {:ok,
       %{
         imported: commit_result.imported,
         errors: invalid_errors ++ commit_result.errors
       }}
    end
  end

  @doc """
  Parses the CSV, validates each row with the submission changeset, and
  classifies every row as valid / invalid / duplicate without inserting.

  Returns:

    * `{:ok, preview}` where `preview` is a map with keys
      `:valid`, `:invalid`, `:duplicates`, `:total_rows`, `:submitter_email`.
    * `{:error, :empty_csv}` if there is nothing to process.
    * `{:error, :no_data_rows}` if only a header is present.

  Valid rows are maps of the form `%{row_num, attrs, timestamp}`. Duplicate
  rows add a `:reason` field indicating whether the conflict was found in
  the database (`:existing_contact`) or earlier in the uploaded file
  (`:earlier_in_upload`).
  """
  @spec preview(String.t(), String.t()) ::
          {:ok, preview_result()} | {:error, :empty_csv} | {:error, :no_data_rows}
  def preview(csv_string, submitter_email) do
    with {:ok, rows} <- parse_csv(csv_string, submitter_email) do
      {valid, invalid} = split_parsed(rows)
      existing = load_existing_for_dedup(valid)
      {unique, duplicates} = dedupe(valid, existing)

      {:ok,
       %{
         valid: unique,
         invalid: invalid,
         duplicates: duplicates,
         total_rows: length(rows),
         submitter_email: submitter_email
       }}
    end
  end

  @doc """
  Inserts the rows returned by `preview/2` in `:valid`. Each is passed to
  `Radio.create_contact/1` and its enrichment is enqueued on success.
  Returns `{:ok, %{imported: [...], errors: [{row_num, messages}, ...]}}`.
  """
  @spec commit([map()]) :: {:ok, commit_result()}
  def commit(valid_rows) when is_list(valid_rows) do
    result =
      Enum.reduce(valid_rows, %{imported: [], errors: []}, fn row, acc ->
        case Radio.create_contact(row.attrs) do
          {:ok, contact} ->
            ContactWeatherEnqueueWorker.enqueue_for_contact(contact)
            %{acc | imported: [contact | acc.imported]}

          {:error, %Ecto.Changeset{} = changeset} ->
            %{acc | errors: [{row.row_num, changeset_error_strings(changeset)} | acc.errors]}
        end
      end)

    {:ok, %{imported: Enum.reverse(result.imported), errors: Enum.reverse(result.errors)}}
  end

  # -- parsing ---------------------------------------------------------------

  defp parse_csv(csv_string, submitter_email) do
    numbered_lines =
      csv_string
      |> String.replace("\r\n", "\n")
      |> String.split("\n")
      |> Enum.with_index(1)

    non_blank = Enum.reject(numbered_lines, fn {line, _num} -> blank?(line) end)

    case non_blank do
      [] ->
        {:error, :empty_csv}

      [{_header, _}] ->
        {:error, :no_data_rows}

      [{_header, _} | data_lines] ->
        rows = Enum.map(data_lines, fn {line, row_num} -> parse_row(line, row_num, submitter_email) end)
        {:ok, rows}
    end
  end

  defp blank?(line), do: String.trim(line) == ""

  defp parse_row(line, row_num, submitter_email) do
    fields = parse_csv_fields(line)

    case columns_for(length(fields)) do
      nil ->
        {:invalid,
         %{
           row_num: row_num,
           messages: ["expected 6 columns (no mode) or 7 columns (with mode), got #{length(fields)}"]
         }}

      columns ->
        attrs =
          columns
          |> Enum.zip(fields)
          |> Map.new()
          |> Map.put("submitter_email", submitter_email)

        parse_row_with_timestamp(attrs, row_num)
    end
  end

  defp columns_for(7), do: @columns_with_mode
  defp columns_for(6), do: @columns_without_mode
  defp columns_for(_), do: nil

  defp parse_row_with_timestamp(attrs, row_num) do
    case normalize_timestamp(attrs["qso_timestamp"]) do
      {:ok, iso_timestamp} ->
        validate_parsed_row(attrs, row_num, iso_timestamp)

      {:error, msg} ->
        {:invalid, %{row_num: row_num, messages: [msg]}}
    end
  end

  defp validate_parsed_row(attrs, row_num, iso_timestamp) do
    attrs = Map.put(attrs, "qso_timestamp", iso_timestamp)
    changeset = Contact.submission_changeset(%Contact{}, attrs)

    if changeset.valid? do
      {:ok, datetime} = parse_iso_datetime(iso_timestamp)
      {:parsed, %{row_num: row_num, attrs: attrs, timestamp: datetime}}
    else
      {:invalid, %{row_num: row_num, messages: changeset_error_strings(changeset)}}
    end
  end

  defp split_parsed(rows) do
    rows
    |> Enum.reduce({[], []}, fn
      {:parsed, row}, {valid, invalid} -> {[row | valid], invalid}
      {:invalid, row}, {valid, invalid} -> {valid, [row | invalid]}
    end)
    |> then(fn {valid, invalid} -> {Enum.reverse(valid), Enum.reverse(invalid)} end)
  end

  # -- dedup -----------------------------------------------------------------

  defp load_existing_for_dedup([]), do: %{}

  defp load_existing_for_dedup(valid_rows) do
    bands = valid_rows |> Enum.map(& &1.attrs["band"]) |> Enum.uniq() |> Enum.map(&to_decimal/1)
    timestamps = Enum.map(valid_rows, & &1.timestamp)
    min_ts = timestamps |> Enum.min(DateTime) |> DateTime.add(-@dedup_window_seconds, :second)
    max_ts = timestamps |> Enum.max(DateTime) |> DateTime.add(@dedup_window_seconds, :second)

    from(c in Contact,
      where:
        c.band in ^bands and c.qso_timestamp >= ^min_ts and c.qso_timestamp <= ^max_ts and
          c.flagged_invalid == false,
      select: %{
        station1: c.station1,
        station2: c.station2,
        grid1: c.grid1,
        grid2: c.grid2,
        band: c.band,
        qso_timestamp: c.qso_timestamp
      }
    )
    |> Repo.all()
    |> Enum.reduce(%{}, fn row, acc ->
      key = dedup_key(row.station1, row.grid1, row.station2, row.grid2, row.band)
      Map.update(acc, key, [row.qso_timestamp], &[row.qso_timestamp | &1])
    end)
  end

  defp dedupe(rows, db_map) do
    # in_batch_map accumulates keys from rows already accepted in this CSV so
    # we can distinguish DB duplicates from earlier-row duplicates when we
    # report them.
    {accepted, _in_batch, duplicates} =
      Enum.reduce(rows, {[], %{}, []}, fn row, {accepted, in_batch, dups} ->
        key = dedup_key_from_attrs(row.attrs)

        cond do
          conflict_in_map?(db_map, key, row.timestamp) ->
            {accepted, in_batch, [Map.put(row, :reason, :existing_contact) | dups]}

          conflict_in_map?(in_batch, key, row.timestamp) ->
            {accepted, in_batch, [Map.put(row, :reason, :earlier_in_upload) | dups]}

          true ->
            new_batch = Map.update(in_batch, key, [row.timestamp], &[row.timestamp | &1])
            {[row | accepted], new_batch, dups}
        end
      end)

    {Enum.reverse(accepted), Enum.reverse(duplicates)}
  end

  defp conflict_in_map?(map, key, timestamp) do
    case Map.fetch(map, key) do
      :error -> false
      {:ok, timestamps} -> Enum.any?(timestamps, &within_window?(&1, timestamp))
    end
  end

  defp within_window?(t1, t2) do
    abs(DateTime.diff(t1, t2, :second)) <= @dedup_window_seconds
  end

  defp dedup_key_from_attrs(attrs) do
    dedup_key(attrs["station1"], attrs["grid1"], attrs["station2"], attrs["grid2"], attrs["band"])
  end

  defp dedup_key(s1, g1, s2, g2, band) do
    a = normalize_callsite(s1, g1)
    b = normalize_callsite(s2, g2)
    [c1, c2] = Enum.sort([a, b])
    {c1, c2, normalize_band(band)}
  end

  defp normalize_callsite(call, grid) do
    {upcase_trim(call), upcase_trim(grid)}
  end

  defp upcase_trim(nil), do: ""
  defp upcase_trim(value), do: value |> to_string() |> String.trim() |> String.upcase()

  defp normalize_band(%Decimal{} = b), do: Decimal.to_integer(b)
  defp normalize_band(b) when is_integer(b), do: b
  defp normalize_band(b) when is_binary(b), do: b |> String.trim() |> String.to_integer()

  defp to_decimal(%Decimal{} = d), do: d
  defp to_decimal(v) when is_integer(v), do: Decimal.new(v)
  defp to_decimal(v) when is_binary(v), do: v |> String.trim() |> Decimal.new()

  defp parse_iso_datetime(iso) do
    case DateTime.from_iso8601(iso) do
      {:ok, dt, _} -> {:ok, dt}
      _ -> :error
    end
  end

  # -- timestamps ------------------------------------------------------------

  # Regex patterns for timestamp parsing, ordered most-specific first.
  # Accepts many hand-entered formats. All times assumed UTC.
  # Each returns {year, month, day, hour, minute, second} or nil.
  defp timestamp_parsers do
    [
      # ISO-ish: 2024-06-15T14:30:00Z, 2024-06-15 14:30, 2024/06/15 14:30:00, etc.
      # Accepts T or space separator, optional seconds, optional timezone suffix
      {~r"^(\d{4})[-/](\d{1,2})[-/](\d{1,2})[T\s]+(\d{1,2}):(\d{2})(?::(\d{2}))?(?:\.\d+)?(?:Z|[+-]\d{2}:?\d{2})?$"i,
       fn
         [_, y, m, d, h, mi, s] when byte_size(s) > 0 -> {y, m, d, h, mi, s}
         [_, y, m, d, h, mi | _] -> {y, m, d, h, mi, "00"}
       end},
      # Date only: 2024-06-15, 2024/06/15
      {~r"^(\d{4})[-/](\d{1,2})[-/](\d{1,2})$", fn [_, y, m, d] -> {y, m, d, "00", "00", "00"} end},
      # US with AM/PM: 6/15/2024 2:30 PM, 6-15-2024 2:30PM, 06.15.2024 02:30 am
      {~r"^(\d{1,2})[-/.](\d{1,2})[-/.](\d{4})\s+(\d{1,2}):(\d{2})(?::(\d{2}))?\s*(AM|PM)$"i,
       fn
         [_, m, d, y, h, mi, s, ampm] when byte_size(s) > 0 ->
           {y, m, d, to_string(parse_12h(String.to_integer(h), String.upcase(ampm))), mi, s}

         [_, m, d, y, h, mi, _, ampm] ->
           {y, m, d, to_string(parse_12h(String.to_integer(h), String.upcase(ampm))), mi, "00"}
       end},
      # US 24h: 6/15/2024 14:30, 06-15-2024 14:30:00, 6.15.2024 14:30
      {~r"^(\d{1,2})[-/.](\d{1,2})[-/.](\d{4})\s+(\d{1,2}):(\d{2})(?::(\d{2}))?$",
       fn
         [_, m, d, y, h, mi, s] when byte_size(s) > 0 -> {y, m, d, h, mi, s}
         [_, m, d, y, h, mi | _] -> {y, m, d, h, mi, "00"}
       end},
      # US date only: 6/15/2024, 06-15-2024
      {~r"^(\d{1,2})[-/.](\d{1,2})[-/.](\d{4})$", fn [_, m, d, y] -> {y, m, d, "00", "00", "00"} end},
      # Compact: 20240615T143000, 20240615 1430, 20240615T1430
      {~r"^(\d{4})(\d{2})(\d{2})[T\s]?(\d{2})(\d{2})(\d{2})?$",
       fn
         [_, y, m, d, h, mi, s] when byte_size(s) > 0 -> {y, m, d, h, mi, s}
         [_, y, m, d, h, mi | _] -> {y, m, d, h, mi, "00"}
       end}
    ]
  end

  @doc """
  Parses various date/time formats into ISO 8601 UTC strings.
  All times are assumed UTC.
  """
  @spec normalize_timestamp(String.t()) :: {:ok, String.t()} | {:error, String.t()}
  def normalize_timestamp(raw) do
    trimmed = String.trim(raw)

    # Try ISO 8601 with Elixir's built-in parser first
    case DateTime.from_iso8601(trimmed) do
      {:ok, dt, _} ->
        {:ok, DateTime.to_iso8601(dt)}

      _ ->
        try_regex_parsers(trimmed)
    end
  end

  defp try_regex_parsers(trimmed) do
    timestamp_parsers()
    |> Enum.find_value(fn {regex, extractor} ->
      case Regex.run(regex, trimmed) do
        nil -> nil
        match -> format_iso(extractor.(match))
      end
    end)
    |> case do
      {:ok, _} = ok -> ok
      nil -> {:error, "qso_timestamp could not be parsed: #{trimmed}"}
    end
  end

  defp parse_12h(12, "AM"), do: 0
  defp parse_12h(12, "PM"), do: 12
  defp parse_12h(h, "PM"), do: h + 12
  defp parse_12h(h, "AM"), do: h

  defp format_iso({y, m, d, h, mi, s}) do
    y = String.pad_leading(to_string(y), 4, "0")
    m = String.pad_leading(to_string(m), 2, "0")
    d = String.pad_leading(to_string(d), 2, "0")
    h = String.pad_leading(to_string(h), 2, "0")
    mi = String.pad_leading(to_string(mi), 2, "0")
    s = String.pad_leading(to_string(s), 2, "0")
    {:ok, "#{y}-#{m}-#{d}T#{h}:#{mi}:#{s}Z"}
  end

  # -- csv field parser ------------------------------------------------------

  # RFC 4180 CSV field parser. Handles quoted fields with commas and escaped quotes.
  defp parse_csv_fields(line) do
    line
    |> String.trim()
    |> do_parse_fields([], "")
    |> Enum.reverse()
    |> Enum.map(&String.trim/1)
  end

  defp do_parse_fields("", acc, current), do: [current | acc]

  defp do_parse_fields(<<"\"", rest::binary>>, acc, "") do
    parse_quoted_field(rest, acc, "")
  end

  defp do_parse_fields(<<",", rest::binary>>, acc, current) do
    do_parse_fields(rest, [current | acc], "")
  end

  defp do_parse_fields(<<c, rest::binary>>, acc, current) do
    do_parse_fields(rest, acc, current <> <<c>>)
  end

  defp parse_quoted_field(<<"\"\"", rest::binary>>, acc, current) do
    parse_quoted_field(rest, acc, current <> "\"")
  end

  defp parse_quoted_field(<<"\"", rest::binary>>, acc, current) do
    # End of quoted field — skip to next comma or end
    case rest do
      <<",", rest2::binary>> -> do_parse_fields(rest2, [current | acc], "")
      "" -> [current | acc]
      _ -> do_parse_fields(rest, [current | acc], "")
    end
  end

  defp parse_quoted_field(<<c, rest::binary>>, acc, current) do
    parse_quoted_field(rest, acc, current <> <<c>>)
  end

  defp parse_quoted_field("", acc, current), do: [current | acc]

  defp changeset_error_strings(changeset) do
    changeset
    |> Ecto.Changeset.traverse_errors(fn {message, opts} ->
      Regex.replace(~r"%{(\w+)}", message, fn _, key ->
        opts |> Keyword.get(String.to_existing_atom(key), key) |> to_string()
      end)
    end)
    |> Enum.flat_map(fn {field, messages} ->
      Enum.map(messages, &"#{field} #{&1}")
    end)
  end
end