defmodule Microwaveprop.Radio do
  @moduledoc false

  import Ecto.Query

  alias Microwaveprop.Accounts.Scope
  alias Microwaveprop.Accounts.User
  alias Microwaveprop.Cache
  alias Microwaveprop.Radio.BandResolver
  alias Microwaveprop.Radio.Contact
  alias Microwaveprop.Radio.ContactEdit
  alias Microwaveprop.Radio.Maidenhead
  alias Microwaveprop.Repo
  alias Microwaveprop.Workers.ContactWeatherEnqueueWorker

  @per_page 20
  @max_per_page 200
  @sortable_fields ~w(station1 station2 band mode distance_km qso_timestamp inserted_at)a
  @count_cache_key {__MODULE__, :total_contact_count}
  @count_cache_ttl_ms 30_000
  @map_payload_cache_key {__MODULE__, :contact_map_payload}
  @map_payload_ttl_ms 10 * 60 * 1_000

  @type contact_page :: %{
          entries: [Contact.t()],
          grouped_entries: [{Contact.t(), [Contact.t()]}],
          page: pos_integer(),
          total_pages: pos_integer(),
          total_entries: non_neg_integer()
        }

  @doc """
  Pre-encoded JSON payload for `/contacts/map`. Computing this requires loading
  every contact with coordinates, deriving formatted rows, deduplicating
  reciprocals, and JSON-encoding ~58k rows — ~150-300ms of work that would
  otherwise run on every page mount. Cached for 10 minutes, invalidated when
  a new contact is inserted.
  """
  @spec contact_map_payload() :: %{
          json: iodata(),
          count: non_neg_integer(),
          bands: [integer()]
        }
  def contact_map_payload do
    if Application.get_env(:microwaveprop, :cache_contact_map, true) do
      Cache.fetch_or_store(@map_payload_cache_key, @map_payload_ttl_ms, fn ->
        build_contact_map_payload()
      end)
    else
      build_contact_map_payload()
    end
  end

  defp build_contact_map_payload do
    contacts = load_contacts_for_map()
    bands = contacts |> Enum.map(fn [_, _, _, _, band | _] -> band end) |> Enum.uniq() |> Enum.sort()
    %{json: Jason.encode_to_iodata!(contacts), count: length(contacts), bands: bands}
  end

  defp load_contacts_for_map do
    from(c in Contact,
      where: not is_nil(c.pos1) and not is_nil(c.pos2) and c.private == false,
      select: %{
        id: c.id,
        pos1: c.pos1,
        pos2: c.pos2,
        band: c.band,
        station1: c.station1,
        station2: c.station2,
        mode: c.mode,
        distance_km: c.distance_km,
        qso_timestamp: c.qso_timestamp
      }
    )
    |> Repo.all()
    |> Enum.map(&format_map_contact/1)
    |> Enum.reject(&is_nil/1)
    |> dedup_map_reciprocals()
  end

  defp format_map_contact(c) do
    lat1 = c.pos1["lat"]
    lon1 = c.pos1["lon"]
    lat2 = c.pos2["lat"]
    lon2 = c.pos2["lon"]

    if lat1 && lon1 && lat2 && lon2 do
      [
        lat1,
        lon1,
        lat2,
        lon2,
        format_map_band(c.band),
        c.station1,
        c.station2,
        c.mode,
        format_map_distance(c.distance_km),
        format_map_timestamp(c.qso_timestamp),
        c.id
      ]
    end
  end

  defp format_map_band(nil), do: 0
  defp format_map_band(band), do: Decimal.to_integer(band)

  defp format_map_distance(nil), do: nil
  defp format_map_distance(d), do: Decimal.to_float(d)

  defp format_map_timestamp(nil), do: nil
  defp format_map_timestamp(ts), do: Calendar.strftime(ts, "%Y-%m-%d %H:%M")

  defp dedup_map_reciprocals(contacts) do
    Enum.uniq_by(contacts, fn [_lat1, _lon1, _lat2, _lon2, band, s1, s2, _mode, _dist, ts, _id] ->
      stations = Enum.sort([s1 || "", s2 || ""])
      hour = if ts, do: String.slice(ts, 0..12), else: ""
      {stations, band, hour}
    end)
  end

  @doc """
  Returns contacts submitted by the given user, newest first. Used by
  the `/u/:callsign` profile page. Private contacts are filtered out
  unless the viewer is the profile owner or an admin — anonymous or
  third-party viewers must not see private rows.
  """
  @spec list_contacts_for_user(User.t(), User.t() | nil) :: [Contact.t()]
  def list_contacts_for_user(%User{} = owner, viewer \\ nil) do
    Contact
    |> where([c], c.user_id == ^owner.id)
    |> filter_private_for_viewer(owner, viewer)
    |> order_by([c], desc: c.qso_timestamp, desc: c.id)
    |> preload(:user)
    |> Repo.all()
  end

  defp filter_private_for_viewer(query, %User{id: owner_id}, %User{id: viewer_id}) when owner_id == viewer_id, do: query

  defp filter_private_for_viewer(query, _owner, %User{is_admin: true}), do: query

  defp filter_private_for_viewer(query, _owner, _viewer) do
    where(query, [c], c.private == false)
  end

  @doc """
  Returns the 100 most recent contacts where the given callsign appears
  as either `station1` or `station2`, newest first. Matching is
  case-insensitive.

  When a `viewer` is provided, owner and admin viewers see private
  contacts; other viewers only see non-private contacts.
  """
  @spec list_contacts_involving_callsign(String.t() | nil, User.t() | nil) :: [Contact.t()]
  def list_contacts_involving_callsign(callsign, viewer \\ nil)

  def list_contacts_involving_callsign(callsign, _viewer) when callsign in [nil, ""], do: []

  def list_contacts_involving_callsign(callsign, viewer) when is_binary(callsign) do
    upcased = String.upcase(callsign)

    Contact
    |> where([c], fragment("upper(?)", c.station1) == ^upcased or fragment("upper(?)", c.station2) == ^upcased)
    |> filter_private_for_callsign_viewer(viewer)
    |> order_by([c], desc: c.qso_timestamp, desc: c.id)
    |> limit(100)
    |> preload(:user)
    |> Repo.all()
  end

  defp filter_private_for_callsign_viewer(query, nil), do: where(query, [c], c.private == false)

  defp filter_private_for_callsign_viewer(query, %User{is_admin: true}), do: query

  defp filter_private_for_callsign_viewer(query, %User{callsign: callsign}) do
    # If the viewer is a co-station on the contact, let them see it even if private
    upcased = String.upcase(callsign)

    where(
      query,
      [c],
      c.private == false or
        (c.private == true and
           (fragment("upper(?)", c.station1) == ^upcased or fragment("upper(?)", c.station2) == ^upcased))
    )
  end

  @spec list_contacts(keyword()) :: contact_page()
  def list_contacts(opts \\ []) do
    page = max(Keyword.get(opts, :page, 1), 1)
    per_page = clamp_per_page(Keyword.get(opts, :per_page, @per_page))
    offset = (page - 1) * per_page
    search = Keyword.get(opts, :search)
    scope = Keyword.get(opts, :scope)

    {sort_field, sort_dir} = sort_opts(opts)

    base_query =
      Contact
      |> maybe_search(search)
      |> filter_private(scope)

    total_entries = total_entries_for(search, scope, base_query)
    total_pages = max(ceil(total_entries / per_page), 1)

    entries =
      base_query
      |> order_by([q], [{^sort_dir, field(q, ^sort_field)}, {^sort_dir, q.id}])
      |> limit(^per_page)
      |> offset(^offset)
      |> Repo.all()
      |> Repo.preload(:user)

    %{
      entries: entries,
      grouped_entries: group_reciprocals(entries),
      page: page,
      total_pages: total_pages,
      total_entries: total_entries
    }
  end

  defp clamp_per_page(value) when is_integer(value) and value >= 1, do: min(value, @max_per_page)

  defp clamp_per_page(_), do: @per_page

  @doc """
  Groups reciprocal contacts (same station pair swapped, same band, same timestamp).
  Returns a list of `{primary, reciprocals}` tuples where reciprocals is
  a (possibly empty) list of matching contacts.
  """
  @spec group_reciprocals([Contact.t()]) :: [{Contact.t(), [Contact.t()]}]
  def group_reciprocals(contacts) do
    contacts
    |> Enum.group_by(&reciprocal_key/1)
    |> Enum.map(fn {_key, [primary | rest]} -> {primary, rest} end)
  end

  defp reciprocal_key(contact) do
    stations = Enum.sort([contact.station1, contact.station2])
    {stations, contact.band, reciprocal_hour_key(contact.qso_timestamp)}
  end

  defp reciprocal_hour_key(nil), do: nil

  defp reciprocal_hour_key(%_{} = ts) do
    Map.take(ts, [:year, :month, :day, :hour])
  end

  # Only cache the unscoped, unsearched total — scope-dependent counts
  # differ per-viewer and must not share a cache.
  defp total_entries_for(search, nil, query) when search in [nil, ""] do
    if Application.get_env(:microwaveprop, :cache_contact_count, true) do
      Cache.fetch_or_store(@count_cache_key, @count_cache_ttl_ms, fn ->
        Repo.aggregate(query, :count)
      end)
    else
      Repo.aggregate(query, :count)
    end
  end

  defp total_entries_for(_search, _scope, query), do: Repo.aggregate(query, :count)

  # Visibility filter for scope-aware queries. Admins see everything;
  # logged-in users see non-private + their own private; others see only
  # non-private.
  defp filter_private(query, %Scope{user: %User{is_admin: true}}), do: query

  defp filter_private(query, %Scope{user: %User{id: user_id}}) do
    where(query, [c], c.private == false or c.user_id == ^user_id)
  end

  defp filter_private(query, _), do: where(query, [c], c.private == false)

  defp maybe_search(query, nil), do: query
  defp maybe_search(query, ""), do: query

  defp maybe_search(query, search) do
    terms = search |> String.split() |> Enum.map(&String.trim/1) |> Enum.reject(&(&1 == ""))

    case terms do
      [a, b] ->
        pa = "%" <> String.upcase(a) <> "%"
        pb = "%" <> String.upcase(b) <> "%"

        where(
          query,
          [q],
          (ilike(q.station1, ^pa) and ilike(q.station2, ^pb)) or
            (ilike(q.station1, ^pb) and ilike(q.station2, ^pa))
        )

      _ ->
        pattern = "%" <> String.upcase(search) <> "%"
        where(query, [q], ilike(q.station1, ^pattern) or ilike(q.station2, ^pattern))
    end
  end

  defp sort_opts(opts) do
    sort_by = Keyword.get(opts, :sort_by, :qso_timestamp)
    sort_order = Keyword.get(opts, :sort_order, :desc)

    field = if sort_by in @sortable_fields, do: sort_by, else: :qso_timestamp
    dir = if sort_order in [:asc, :desc], do: sort_order, else: :desc

    {field, dir}
  end

  # ── Enrichment status queries ────────────────────────────────────

  @enrichable_statuses [:pending, :failed]

  @spec contacts_needing_enrichment(
          atom(),
          non_neg_integer(),
          (Ecto.Query.t() -> Ecto.Query.t()) | nil
        ) :: [Contact.t()]
  def contacts_needing_enrichment(field, limit \\ 500, extra_filter \\ nil) do
    query =
      Contact
      |> where([q], field(q, ^field) in ^@enrichable_statuses and not is_nil(q.pos1))
      |> order_by([q], asc: q.qso_timestamp)
      |> limit(^limit)

    query = if extra_filter, do: extra_filter.(query), else: query
    Repo.all(query)
  end

  @spec unprocessed_contacts(non_neg_integer()) :: [Contact.t()]
  def unprocessed_contacts(limit \\ 500), do: contacts_needing_enrichment(:weather_status, limit)

  @spec unprocessed_hrrr_contacts(non_neg_integer()) :: [Contact.t()]
  def unprocessed_hrrr_contacts(limit \\ 500), do: contacts_needing_enrichment(:hrrr_status, limit)

  @spec unprocessed_terrain_contacts(non_neg_integer()) :: [Contact.t()]
  def unprocessed_terrain_contacts(limit \\ 500),
    do: contacts_needing_enrichment(:terrain_status, limit, &where(&1, [q], not is_nil(q.pos2)))

  @spec unprocessed_iemre_contacts(non_neg_integer()) :: [Contact.t()]
  def unprocessed_iemre_contacts(limit \\ 500), do: contacts_needing_enrichment(:iemre_status, limit)

  @spec unprocessed_radar_contacts(non_neg_integer()) :: [Contact.t()]
  def unprocessed_radar_contacts(limit \\ 500),
    do: contacts_needing_enrichment(:radar_status, limit, &where(&1, [q], not is_nil(q.pos2)))

  @spec set_enrichment_status!([Ecto.UUID.t()], atom(), atom()) :: {non_neg_integer(), nil | [any()]}
  def set_enrichment_status!(ids, field, status) do
    Contact
    |> where([q], q.id in ^ids and field(q, ^field) != ^status)
    |> Repo.update_all(set: [{field, status}])
  end

  @spec mark_weather_queued!([Ecto.UUID.t()]) :: {non_neg_integer(), nil | [any()]}
  def mark_weather_queued!(ids), do: set_enrichment_status!(ids, :weather_status, :queued)

  @spec mark_hrrr_queued!([Ecto.UUID.t()]) :: {non_neg_integer(), nil | [any()]}
  def mark_hrrr_queued!(ids), do: set_enrichment_status!(ids, :hrrr_status, :queued)

  @spec mark_terrain_queued!([Ecto.UUID.t()]) :: {non_neg_integer(), nil | [any()]}
  def mark_terrain_queued!(ids), do: set_enrichment_status!(ids, :terrain_status, :queued)

  @spec mark_iemre_queued!([Ecto.UUID.t()]) :: {non_neg_integer(), nil | [any()]}
  def mark_iemre_queued!(ids), do: set_enrichment_status!(ids, :iemre_status, :queued)

  @doc """
  Returns a list of {lat, lon} points along the contact path: pos1, midpoint, pos2.
  Returns [pos1] if pos2 is nil, or [] if pos1 is nil.
  """
  @spec contact_path_points(Contact.t()) :: [{float(), float()}]
  def contact_path_points(%{pos1: nil}), do: []

  def contact_path_points(%{pos1: pos1, pos2: nil}) do
    lat = pos1["lat"]
    lon = pos1["lon"]
    if lat && lon, do: [{lat, lon}], else: []
  end

  def contact_path_points(%{pos1: pos1, pos2: pos2}) do
    build_path_points(extract_latlon(pos1), extract_latlon(pos2))
  end

  defp extract_latlon(%{"lat" => lat} = pos) when is_number(lat) do
    lon = pos["lon"]
    if lon, do: {lat, lon}
  end

  defp extract_latlon(_), do: nil

  defp build_path_points({lat1, lon1}, {lat2, lon2}) do
    {mid_lat, mid_lon} = great_circle_midpoint(lat1, lon1, lat2, lon2)
    [{lat1, lon1}, {mid_lat, mid_lon}, {lat2, lon2}]
  end

  defp build_path_points({lat1, lon1}, _), do: [{lat1, lon1}]
  defp build_path_points(_, _), do: []

  # Spherical-vector midpoint between two lat/lon points. The arithmetic
  # mean of two longitudes folds across the anti-meridian (e.g.
  # `(179 + -179) / 2 == 0` back at Greenwich), so paths that cross the
  # date line otherwise get a bogus center.
  defp great_circle_midpoint(lat1, lon1, lat2, lon2) do
    rlat1 = deg_to_rad(lat1)
    rlat2 = deg_to_rad(lat2)
    rlon1 = deg_to_rad(lon1)
    dlon = deg_to_rad(lon2 - lon1)

    bx = :math.cos(rlat2) * :math.cos(dlon)
    by = :math.cos(rlat2) * :math.sin(dlon)

    mid_lat =
      :math.atan2(
        :math.sin(rlat1) + :math.sin(rlat2),
        :math.sqrt((:math.cos(rlat1) + bx) ** 2 + by ** 2)
      )

    mid_lon = rlon1 + :math.atan2(by, :math.cos(rlat1) + bx)

    {rad_to_deg(mid_lat), normalize_lon(rad_to_deg(mid_lon))}
  end

  defp rad_to_deg(rad), do: rad * 180.0 / :math.pi()

  defp normalize_lon(lon) when lon > 180.0, do: lon - 360.0
  defp normalize_lon(lon) when lon < -180.0, do: lon + 360.0
  defp normalize_lon(lon), do: lon

  @spec haversine_km(number(), number(), number(), number()) :: float()
  defdelegate haversine_km(lat1, lon1, lat2, lon2), to: Microwaveprop.Geo

  @spec backfill_distances([Contact.t()]) :: Postgrex.Result.t() | nil
  def backfill_distances(contacts) do
    updates =
      for contact <- contacts,
          is_nil(contact.distance_km) && contact.pos1 && contact.pos2,
          lat1 = contact.pos1["lat"],
          lon1 = contact.pos1["lon"],
          lat2 = contact.pos2["lat"],
          lon2 = contact.pos2["lon"],
          lat1 && lon1 && lat2 && lon2 do
        dist = lat1 |> haversine_km(lon1, lat2, lon2) |> round()
        {contact.id, dist}
      end

    execute_distance_updates(updates)
  end

  defp execute_distance_updates([]), do: nil

  defp execute_distance_updates(updates) do
    # Build a single UPDATE ... FROM (VALUES ...) statement
    # Ecto.UUID.dump!/1 converts string UUIDs to the 16-byte binary Postgrex expects
    {values_sql, params} =
      updates
      |> Enum.with_index()
      |> Enum.reduce({"", []}, fn {{id, dist}, i}, {sql, params} ->
        frag = "($#{i * 2 + 1}::uuid, $#{i * 2 + 2}::numeric)"
        sep = if sql == "", do: "", else: ", "
        {sql <> sep <> frag, params ++ [Ecto.UUID.dump!(id), dist]}
      end)

    Repo.query!(
      "UPDATE contacts AS c SET distance_km = v.dist FROM (VALUES #{values_sql}) AS v(id, dist) WHERE c.id = v.id",
      params
    )
  end

  defp deg_to_rad(deg), do: deg * :math.pi() / 180

  @dialyzer {:no_match, get_contact!: 1}
  @spec get_contact!(term()) :: Contact.t()
  def get_contact!(id) do
    Contact
    |> Repo.get!(id)
    |> Repo.preload(:flagged_by_user)
  end

  @doc """
  True when the viewer (per `Scope`) is allowed to see `contact`.
  Public contacts are always viewable; private contacts only by the
  submitter or an admin.
  """
  @spec can_view?(Contact.t(), Scope.t() | nil) :: boolean()
  def can_view?(%Contact{private: false}, _scope), do: true
  def can_view?(%Contact{}, %Scope{user: %User{is_admin: true}}), do: true
  def can_view?(%Contact{user_id: user_id}, %Scope{user: %User{id: user_id}}) when not is_nil(user_id), do: true
  def can_view?(%Contact{}, _scope), do: false

  @doc """
  Hard-delete a contact and everything attached to it. All FK
  references (`terrain_profiles`, `contact_edits`,
  `contact_common_volume_radar`) declare `on_delete: :delete_all`,
  so a single `Repo.delete!` cascades through.

  Admin-only operation — callers must gate with `is_admin` before
  invoking. Irreversible: prefer `toggle_flagged_invalid!/2` when
  the data should be hidden but kept in the audit trail.
  """
  @spec delete_contact!(Contact.t()) :: Contact.t()
  def delete_contact!(%Contact{} = contact), do: Repo.delete!(contact)

  @spec toggle_flagged_invalid!(Contact.t(), User.t() | nil) :: Contact.t()
  def toggle_flagged_invalid!(contact, flagger \\ nil) do
    new_state = !contact.flagged_invalid

    attrs =
      if new_state do
        %{
          flagged_invalid: true,
          flagged_by_user_id: flagger && flagger.id,
          flagged_at: DateTime.truncate(DateTime.utc_now(), :second)
        }
      else
        # Clear attribution on unflag — the next flag (potentially
        # by a different admin) gets a fresh record, and the badge
        # stops claiming a stale flagger after the contact was
        # restored to valid.
        %{flagged_invalid: false, flagged_by_user_id: nil, flagged_at: nil}
      end

    contact
    |> Ecto.Changeset.change(attrs)
    |> Repo.update!()
  end

  @doc """
  Returns all flagged-invalid contacts, newest first. Admin-only
  viewer — the admin contact-edits page uses this to surface flagged
  QSOs for review alongside pending edit requests.
  """
  @spec list_flagged_contacts() :: [Contact.t()]
  def list_flagged_contacts do
    Contact
    |> where([c], c.flagged_invalid == true)
    |> order_by([c], desc: c.inserted_at, desc: c.id)
    |> Repo.all()
  end

  @spec change_contact(Contact.t(), map()) :: Ecto.Changeset.t()
  def change_contact(contact, attrs \\ %{}) do
    Contact.submission_changeset(contact, attrs)
  end

  @doc """
  Computes pos1/pos2/distance_km from grid squares if missing.
  Returns the updated contact, or the original if already computed or grids are missing.
  """
  @spec ensure_positions!(Contact.t()) :: Contact.t()
  def ensure_positions!(contact) do
    needs_pos1 = is_nil(contact.pos1) && contact.grid1
    needs_pos2 = is_nil(contact.pos2) && contact.grid2

    if needs_pos1 || needs_pos2 do
      do_ensure_positions(contact)
    else
      contact
    end
  end

  defp do_ensure_positions(contact) do
    pos1 = contact.pos1 || latlon_from_grid(contact.grid1)
    pos2 = contact.pos2 || latlon_from_grid(contact.grid2)
    distance = compute_distance(pos1, pos2, contact.distance_km)

    changes = build_position_changes(contact, pos1, pos2, distance)

    apply_position_changes(contact, changes)
  end

  defp compute_distance(pos1, pos2, _existing_distance) when not is_nil(pos1) and not is_nil(pos2) do
    lon1 = pos1["lon"]
    lon2 = pos2["lon"]

    pos1["lat"]
    |> haversine_km(lon1, pos2["lat"], lon2)
    |> round()
    |> Decimal.new()
  end

  defp compute_distance(_pos1, _pos2, existing_distance), do: existing_distance

  defp build_position_changes(contact, pos1, pos2, distance) do
    Enum.reduce([pos1: pos1, pos2: pos2, distance_km: distance], %{}, fn {key, val}, acc ->
      if val && is_nil(Map.get(contact, key)), do: Map.put(acc, key, val), else: acc
    end)
  end

  defp apply_position_changes(contact, changes) when changes == %{}, do: contact

  defp apply_position_changes(contact, changes) do
    changes = reset_enrichment_statuses(contact, changes)

    contact
    |> Ecto.Changeset.change(changes)
    |> Repo.update!()
  end

  defp reset_enrichment_statuses(contact, changes) do
    new_pos1 = Map.get(changes, :pos1)
    new_pos2 = Map.get(changes, :pos2)

    changes =
      if new_pos1 do
        changes
        |> maybe_reset_status(:hrrr_status, contact.hrrr_status)
        |> maybe_reset_status(:weather_status, contact.weather_status)
        |> maybe_reset_status(:iemre_status, contact.iemre_status)
      else
        changes
      end

    if new_pos1 && (new_pos2 || contact.pos2) do
      maybe_reset_status(changes, :terrain_status, contact.terrain_status)
    else
      changes
    end
  end

  defp maybe_reset_status(changes, field, :unavailable), do: Map.put(changes, field, :pending)
  defp maybe_reset_status(changes, field, :complete), do: Map.put(changes, field, :pending)
  defp maybe_reset_status(changes, _field, _status), do: changes

  defp latlon_from_grid(nil), do: nil

  defp latlon_from_grid(grid) do
    case Maidenhead.to_latlon(grid) do
      {:ok, {lat, lon}} ->
        %{"lat" => lat, "lon" => lon}

      _ ->
        # Try truncating to nearest valid even length (e.g. "FN03c" → "FN03")
        truncated = String.slice(grid, 0, div(String.length(grid), 2) * 2)

        case Maidenhead.to_latlon(truncated) do
          {:ok, {lat, lon}} -> %{"lat" => lat, "lon" => lon}
          _ -> nil
        end
    end
  end

  @dedup_window_seconds 3600

  @spec create_contact(map(), String.t() | nil) ::
          {:ok, Contact.t()} | {:error, Ecto.Changeset.t()} | {:error, :duplicate, Contact.t()}
  def create_contact(attrs, user_id \\ nil) do
    # Place user_id on the struct BEFORE running submission_changeset so
    # validate_user_or_email/1 can see it. Otherwise an authenticated
    # caller that omits submitter_email gets a spurious "can't be blank"
    # error even though user_id satisfies the ownership invariant.
    changeset = Contact.submission_changeset(%Contact{user_id: user_id}, normalize_band_attr(attrs))

    if changeset.valid? do
      insert_validated_contact(changeset)
    else
      {:error, %{changeset | action: :insert}}
    end
  end

  # Translate ADIF wavelength labels ("33cm") and raw frequency strings
  # ("903.100") into our canonical MHz integer string so every create
  # path — manual form, CSV commit, ADIF commit, API, console — agrees
  # on what "33cm" means before the changeset's validate_inclusion runs.
  # Atom and string keys are both supported so this works regardless of
  # where the map came from.
  defp normalize_band_attr(attrs) do
    cond do
      raw = Map.get(attrs, "band") ->
        case BandResolver.resolve_as_string(raw) do
          nil -> attrs
          canonical -> Map.put(attrs, "band", canonical)
        end

      raw = Map.get(attrs, :band) ->
        case BandResolver.resolve_as_string(raw) do
          nil -> attrs
          canonical -> Map.put(attrs, :band, canonical)
        end

      true ->
        attrs
    end
  end

  defp insert_validated_contact(changeset) do
    if existing = find_duplicate_contact(changeset) do
      {:error, :duplicate, existing}
    else
      resolve_grids_and_insert(changeset)
    end
  end

  defp resolve_grids_and_insert(changeset) do
    grid1 = Ecto.Changeset.get_change(changeset, :grid1)
    grid2 = Ecto.Changeset.get_change(changeset, :grid2)

    case {Maidenhead.to_latlon(grid1), Maidenhead.to_latlon(grid2)} do
      {{:ok, {lat1, lon1}}, {:ok, {lat2, lon2}}} ->
        distance = lat1 |> haversine_km(lon1, lat2, lon2) |> round() |> Decimal.new()

        result =
          changeset
          |> Ecto.Changeset.put_change(:pos1, %{"lat" => lat1, "lon" => lon1})
          |> Ecto.Changeset.put_change(:pos2, %{"lat" => lat2, "lon" => lon2})
          |> Ecto.Changeset.put_change(:distance_km, distance)
          |> Ecto.Changeset.put_change(:user_submitted, true)
          |> Repo.insert()

        with {:ok, _} <- result do
          invalidate_contact_map_caches()
        end

        result

      _ ->
        changeset = Ecto.Changeset.add_error(changeset, :grid1, "could not resolve grid to coordinates")
        {:error, %{changeset | action: :insert}}
    end
  rescue
    _e in Ecto.ConstraintError ->
      # Unique constraint violation from a concurrent insert — the
      # dedup check at `insert_validated_contact` raced with another
      # process. Re-check and return the existing contact.
      case find_duplicate_contact(changeset) do
        nil -> {:error, :constraint_error}
        existing -> {:error, :duplicate, existing}
      end
  end

  defp find_duplicate_contact(changeset) do
    s1 = changeset |> Ecto.Changeset.get_field(:station1) |> to_string() |> String.upcase()
    s2 = changeset |> Ecto.Changeset.get_field(:station2) |> to_string() |> String.upcase()
    g1 = changeset |> Ecto.Changeset.get_field(:grid1) |> to_string() |> String.upcase()
    g2 = changeset |> Ecto.Changeset.get_field(:grid2) |> to_string() |> String.upcase()
    band = Ecto.Changeset.get_field(changeset, :band)
    ts = Ecto.Changeset.get_field(changeset, :qso_timestamp)

    band_decimal = if is_binary(band), do: Decimal.new(band), else: band
    min_ts = DateTime.add(ts, -@dedup_window_seconds, :second)
    max_ts = DateTime.add(ts, @dedup_window_seconds, :second)

    Repo.one(
      from(c in Contact,
        where:
          c.band == ^band_decimal and c.qso_timestamp >= ^min_ts and c.qso_timestamp <= ^max_ts and
            c.flagged_invalid == false and fragment("LEAST(UPPER(station1), UPPER(station2)) = ?", ^min(s1, s2)) and
            fragment("GREATEST(UPPER(station1), UPPER(station2)) = ?", ^max(s1, s2)) and
            fragment("LEAST(UPPER(grid1), UPPER(grid2)) = ?", ^min(g1, g2)) and
            fragment("GREATEST(UPPER(grid1), UPPER(grid2)) = ?", ^max(g1, g2))
      )
    )
  end

  @refinement_fields ~w(grid1 grid2 mode)a
  @refinement_allowed_modes ~w(CW SSB FM FT8 FT4 Q65)

  @doc """
  Apply a grid/mode refinement from an import row to an existing contact.

  `changes` may contain any of `:grid1`, `:grid2`, or `:mode`. Grid changes
  trigger recomputation of the corresponding `pos` and the `distance_km`,
  and reset all four enrichment status fields to `:pending` so the caller
  can re-enqueue the enrichment pipeline. Mode-only changes do not touch
  positions or enrichment status.
  """
  @spec apply_contact_refinement(Contact.t(), map()) ::
          {:ok, Contact.t()} | {:error, Ecto.Changeset.t()}
  def apply_contact_refinement(%Contact{} = contact, changes) when is_map(changes) do
    cleaned = Map.take(changes, @refinement_fields)

    if cleaned == %{} do
      {:ok, contact}
    else
      do_apply_contact_refinement(contact, cleaned)
    end
  end

  defp do_apply_contact_refinement(contact, changes) do
    changeset =
      contact
      |> Ecto.Changeset.cast(changes, @refinement_fields)
      |> validate_refinement_mode()
      |> maybe_recompute_refinement_positions(contact, changes)

    if changeset.valid? do
      Repo.update(changeset)
    else
      {:error, %{changeset | action: :update}}
    end
  end

  defp validate_refinement_mode(changeset) do
    Ecto.Changeset.validate_inclusion(changeset, :mode, @refinement_allowed_modes)
  end

  defp maybe_recompute_refinement_positions(changeset, contact, changes) do
    new_grid1 = Map.get(changes, :grid1)
    new_grid2 = Map.get(changes, :grid2)

    cond do
      new_grid1 && new_grid2 ->
        put_pos_changes(changeset, new_grid1, new_grid2, :both)

      new_grid1 ->
        put_pos_changes(changeset, new_grid1, contact.grid2, :side1)

      new_grid2 ->
        put_pos_changes(changeset, contact.grid1, new_grid2, :side2)

      true ->
        changeset
    end
  end

  defp put_pos_changes(changeset, grid1, grid2, sides) do
    case {Maidenhead.to_latlon(grid1), Maidenhead.to_latlon(grid2)} do
      {{:ok, {lat1, lon1}}, {:ok, {lat2, lon2}}} ->
        distance = lat1 |> haversine_km(lon1, lat2, lon2) |> round() |> Decimal.new()

        changeset
        |> maybe_put_pos_for_side(:pos1, %{"lat" => lat1, "lon" => lon1}, sides)
        |> maybe_put_pos_for_side(:pos2, %{"lat" => lat2, "lon" => lon2}, sides)
        |> Ecto.Changeset.put_change(:distance_km, distance)
        |> reset_all_enrichment_statuses()

      _ ->
        Ecto.Changeset.add_error(changeset, :grid1, "could not resolve grid to coordinates")
    end
  end

  defp maybe_put_pos_for_side(changeset, :pos1, pos, sides) when sides in [:side1, :both] do
    Ecto.Changeset.put_change(changeset, :pos1, pos)
  end

  defp maybe_put_pos_for_side(changeset, :pos2, pos, sides) when sides in [:side2, :both] do
    Ecto.Changeset.put_change(changeset, :pos2, pos)
  end

  defp maybe_put_pos_for_side(changeset, _field, _pos, _sides), do: changeset

  defp reset_all_enrichment_statuses(changeset) do
    changeset
    |> Ecto.Changeset.put_change(:hrrr_status, :pending)
    |> Ecto.Changeset.put_change(:weather_status, :pending)
    |> Ecto.Changeset.put_change(:terrain_status, :pending)
    |> Ecto.Changeset.put_change(:iemre_status, :pending)
  end

  # ── Contact Edits ───────────────────────────────────────────────

  @doc """
  Create a proposed edit for a contact. Only stores fields that actually
  differ from the contact's current values. Normalizes callsigns and grids
  to uppercase.
  """
  @spec create_contact_edit(Contact.t(), User.t(), map()) ::
          {:ok, ContactEdit.t()} | {:error, Ecto.Changeset.t()}
  def create_contact_edit(%Contact{} = contact, user, proposed_changes) when is_map(proposed_changes) do
    # Normalize and diff against current values
    normalized = normalize_proposed(proposed_changes)
    diffed = diff_against_contact(contact, normalized)

    attrs = %{
      contact_id: contact.id,
      user_id: user.id,
      proposed_changes: diffed
    }

    %ContactEdit{}
    |> ContactEdit.changeset(attrs)
    |> Repo.insert()
  end

  # Whitelist of keys an owner / pending-edit submitter is allowed
  # to propose. Anything not in this set is dropped before the
  # changes hit `String.to_existing_atom` + `Ecto.Changeset.change`
  # downstream — without this filter a crafted form could mass-assign
  # `user_id`, `flagged_invalid`, `inserted_at`, etc. on a contact the
  # caller owns. `band` is in the list because admins edit it through
  # this same path; `ContactEdit.changeset` rejects `band` for
  # non-admin pending edits via its own validation.
  @editable_proposed_keys ~w(station1 station2 grid1 grid2 band mode qso_timestamp height1_ft height2_ft private)

  @doc false
  @spec normalize_proposed(map()) :: map()
  def normalize_proposed(changes) do
    changes
    |> Map.take(@editable_proposed_keys)
    |> normalize_string_field("station1")
    |> normalize_string_field("station2")
    |> normalize_string_field("grid1")
    |> normalize_string_field("grid2")
    |> normalize_string_field("mode")
    |> normalize_integer_field("height1_ft")
    |> normalize_integer_field("height2_ft")
    |> normalize_boolean_field("private")
    |> normalize_timestamp_field("qso_timestamp")
  end

  # The /contacts/:id edit form pre-fills qso_timestamp as
  # `"YYYY-MM-DD HH:MM"` via Calendar.strftime, then submits whatever
  # the user typed. Coerce to %DateTime{} (UTC) so the diff matches
  # against the stored DateTime instead of always reading as a change,
  # and so the apply path doesn't crash on a non-ISO string. Drop the
  # field when empty or unparseable so a bogus value is treated as
  # "no change" rather than raising.
  defp normalize_timestamp_field(map, key) do
    case Map.get(map, key, :not_provided) do
      :not_provided -> map
      nil -> Map.delete(map, key)
      "" -> Map.delete(map, key)
      %DateTime{} = dt -> Map.put(map, key, dt)
      val when is_binary(val) -> put_or_drop_timestamp(map, key, parse_timestamp(val))
      _ -> Map.delete(map, key)
    end
  end

  defp put_or_drop_timestamp(map, key, nil), do: Map.delete(map, key)
  defp put_or_drop_timestamp(map, key, %DateTime{} = dt), do: Map.put(map, key, dt)

  defp parse_timestamp(val) do
    trimmed = String.trim(val)

    # Accept `"YYYY-MM-DD HH:MM[:SS][Z]"` (form pre-fill, space sep) and
    # the strict ISO `"YYYY-MM-DDTHH:MM:SSZ"` form. NaiveDateTime handles
    # both separators and an optional seconds field.
    iso = trimmed |> String.replace(" ", "T") |> ensure_seconds() |> String.trim_trailing("Z")

    case NaiveDateTime.from_iso8601(iso) do
      {:ok, ndt} -> DateTime.from_naive!(ndt, "Etc/UTC")
      _ -> nil
    end
  end

  defp ensure_seconds(s) do
    # Add ":00" before any trailing "Z" / end-of-string when only HH:MM
    # is present, since NaiveDateTime.from_iso8601 requires seconds.
    case Regex.run(~r/^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2})(Z?)$/, s) do
      [_, prefix, tail] -> prefix <> ":00" <> tail
      _ -> s
    end
  end

  # HTML checkboxes arrive as "true"/"false" strings; Ecto won't coerce
  # those inside a validate_inclusion path, so normalize at the boundary.
  defp normalize_boolean_field(map, key) do
    case Map.get(map, key, :not_provided) do
      :not_provided -> map
      nil -> map
      val when is_boolean(val) -> Map.put(map, key, val)
      "true" -> Map.put(map, key, true)
      "false" -> Map.put(map, key, false)
      _ -> map
    end
  end

  defp normalize_string_field(map, key) do
    case Map.get(map, key) do
      nil -> map
      val when is_binary(val) -> Map.put(map, key, val |> String.trim() |> String.upcase())
      val -> Map.put(map, key, val)
    end
  end

  # Forms deliver integer-shaped inputs as strings. Drop the field entirely
  # on empty strings so the diff doesn't register "" != current_value.
  defp normalize_integer_field(map, key) do
    case Map.get(map, key, :not_provided) do
      :not_provided ->
        map

      nil ->
        map

      val when is_integer(val) ->
        map

      "" ->
        Map.delete(map, key)

      val when is_binary(val) ->
        case Integer.parse(String.trim(val)) do
          {int, ""} -> Map.put(map, key, int)
          _ -> map
        end

      _ ->
        map
    end
  end

  @doc false
  @spec diff_against_contact(Contact.t(), map()) :: map()
  def diff_against_contact(contact, proposed) do
    Enum.reduce(proposed, %{}, fn {key, new_val}, acc ->
      current = current_value(contact, key)

      if values_equal?(current, new_val) do
        acc
      else
        Map.put(acc, key, new_val)
      end
    end)
  end

  defp current_value(contact, "station1"), do: contact.station1
  defp current_value(contact, "station2"), do: contact.station2
  defp current_value(contact, "grid1"), do: contact.grid1
  defp current_value(contact, "grid2"), do: contact.grid2
  defp current_value(contact, "mode"), do: contact.mode

  defp current_value(contact, "band") do
    if contact.band, do: Decimal.to_integer(contact.band)
  end

  defp current_value(contact, "qso_timestamp"), do: contact.qso_timestamp
  defp current_value(contact, "height1_ft"), do: contact.height1_ft
  defp current_value(contact, "height2_ft"), do: contact.height2_ft
  defp current_value(contact, "private"), do: contact.private
  defp current_value(_contact, _key), do: nil

  defp values_equal?(a, b) when is_binary(a) and is_binary(b) do
    String.upcase(a) == String.upcase(b)
  end

  defp values_equal?(a, b) when is_integer(a), do: a == to_integer(b)
  defp values_equal?(a, b) when is_integer(b), do: to_integer(a) == b
  defp values_equal?(a, b), do: a == b

  defp to_integer(v) when is_integer(v), do: v
  defp to_integer(v) when is_binary(v), do: String.to_integer(v)
  defp to_integer(%Decimal{} = v), do: Decimal.to_integer(v)
  defp to_integer(v), do: v

  @spec list_pending_edits() :: [ContactEdit.t()]
  def list_pending_edits do
    ContactEdit
    |> where([e], e.status == :pending)
    |> order_by([e], desc: e.inserted_at)
    |> preload([:user, :contact])
    |> Repo.all()
  end

  @doc """
  Base query for pending contact edits with `:user` and `:contact`
  preloaded. Used as a `live_table` `data_provider` so sort/paginate
  can be applied on top.
  """
  @spec pending_edits_query() :: Ecto.Query.t()
  def pending_edits_query do
    from e in ContactEdit,
      as: :resource,
      where: e.status == :pending,
      preload: [:user, :contact]
  end

  @spec list_contact_edits(Ecto.UUID.t()) :: [ContactEdit.t()]
  def list_contact_edits(contact_id) do
    ContactEdit
    |> where([e], e.contact_id == ^contact_id)
    |> order_by([e], desc: e.inserted_at)
    |> preload([:user, :reviewed_by])
    |> Repo.all()
  end

  @spec pending_edit_count() :: non_neg_integer()
  def pending_edit_count do
    ContactEdit
    |> where([e], e.status == :pending)
    |> Repo.aggregate(:count)
  end

  @spec pending_edit_for_user(Ecto.UUID.t(), Ecto.UUID.t()) :: ContactEdit.t() | nil
  def pending_edit_for_user(contact_id, user_id) do
    ContactEdit
    |> where([e], e.contact_id == ^contact_id and e.user_id == ^user_id and e.status == :pending)
    |> preload([:user, :contact])
    |> Repo.one()
  end

  @spec get_contact_edit(Ecto.UUID.t()) :: ContactEdit.t() | nil
  def get_contact_edit(id) do
    ContactEdit
    |> preload([:user, :contact, :reviewed_by])
    |> Repo.get(id)
  end

  @spec approve_edit(ContactEdit.t(), User.t(), String.t() | nil) ::
          {:ok, ContactEdit.t()} | {:error, any()}
  def approve_edit(%ContactEdit{status: :pending} = edit, admin, note) do
    Repo.transaction(fn ->
      # Check the contact still exists before proceeding — the FK has
      # on_delete: :delete_all so if the contact is gone the edit row is
      # also gone, making any update to it a StaleEntryError.
      case Repo.get(Contact, edit.contact_id) do
        nil ->
          Repo.rollback(:contact_deleted)

        contact ->
          # Mark edit as approved
          {:ok, approved} =
            edit
            |> ContactEdit.review_changeset(%{
              status: :approved,
              admin_note: note,
              reviewed_by_id: admin.id,
              reviewed_at: DateTime.truncate(DateTime.utc_now(), :second)
            })
            |> Repo.update()

          # Apply changes to contact
          _ = apply_edit_to_contact(contact, edit.proposed_changes)
          Repo.preload(approved, [:user, :contact, :reviewed_by])
      end
    end)
  end

  @spec reject_edit(ContactEdit.t(), User.t(), String.t() | nil) ::
          {:ok, ContactEdit.t()} | {:error, Ecto.Changeset.t()}
  def reject_edit(%ContactEdit{status: :pending} = edit, admin, note) do
    edit
    |> ContactEdit.review_changeset(%{
      status: :rejected,
      admin_note: note,
      reviewed_by_id: admin.id,
      reviewed_at: DateTime.truncate(DateTime.utc_now(), :second)
    })
    |> Repo.update()
  end

  @doc """
  Whether the given user submitted (owns) this contact. Returns `false`
  for anonymous contacts (`user_id == nil`) and for a `nil` user.
  """
  @spec owner?(Contact.t(), User.t() | nil) :: boolean()
  def owner?(%Contact{user_id: nil}, _user), do: false
  def owner?(%Contact{}, nil), do: false
  def owner?(%Contact{user_id: uid}, %User{id: uid}), do: true
  def owner?(%Contact{}, %User{}), do: false

  @doc """
  Allow the original submitter of a contact to edit it directly without
  going through the admin review queue. Reuses the same normalization
  and diff logic as `create_contact_edit/3` so owners can only push
  meaningful changes (at least one field must actually differ).

  Returns `{:ok, updated_contact}` on success, `{:error, :not_owner}` if
  the user did not submit the contact, or `{:error, :no_changes}` if
  every proposed value already matches the current contact.
  """
  @spec apply_owner_edit(Contact.t(), User.t(), map()) ::
          {:ok, Contact.t()} | {:error, :not_owner | :no_changes}
  def apply_owner_edit(%Contact{} = contact, %User{} = user, proposed_changes) when is_map(proposed_changes) do
    if owner?(contact, user) do
      diffed = diff_against_contact(contact, normalize_proposed(proposed_changes))

      if diffed == %{} do
        {:error, :no_changes}
      else
        {:ok, apply_edit_to_contact(contact, diffed)}
      end
    else
      {:error, :not_owner}
    end
  end

  @doc "Apply proposed changes directly to a contact (admin use)."
  @spec apply_edit_to_contact(Contact.t(), map()) :: Contact.t()
  def apply_edit_to_contact(contact, proposed_changes) do
    changes = build_contact_changes(contact, proposed_changes)
    changes = maybe_recompute_positions(changes, contact, proposed_changes)
    changes = maybe_reset_terrain_for_heights(changes, proposed_changes)

    updated =
      contact
      |> Ecto.Changeset.change(changes)
      |> Repo.update!()

    maybe_enqueue_terrain_for_heights(updated, proposed_changes)

    # Public contact-map + count caches are built from `private == false`
    # rows. A grid correction, callsign change, flagged_invalid flip, or
    # a `private: false -> true` toggle on an already-mapped contact all
    # change what /api/contacts/map and /contacts/map should return —
    # private flips in particular are a privacy leak until the 10-minute
    # TTL expires. Invalidate on every direct update.
    invalidate_contact_map_caches()

    updated
  end

  defp invalidate_contact_map_caches do
    Cache.invalidate(@count_cache_key)
    Cache.invalidate(@map_payload_cache_key)
    Cache.invalidate(MicrowavepropWeb.ContactMapController.cache_key())
    :ok
  end

  defp maybe_enqueue_terrain_for_heights(contact, proposed) do
    if Map.has_key?(proposed, "height1_ft") or Map.has_key?(proposed, "height2_ft") do
      ContactWeatherEnqueueWorker.enqueue_for_contact(contact, [:terrain])
    end

    :ok
  end

  # Antenna heights feed directly into the elevation-profile terrain
  # analysis (diffraction loss, clearance verdict). Changing them
  # invalidates the stored terrain_profile, so reset the status to
  # :pending and let TerrainProfileWorker re-run.
  defp maybe_reset_terrain_for_heights(changes, proposed) do
    if Map.has_key?(proposed, "height1_ft") or Map.has_key?(proposed, "height2_ft") do
      Map.put(changes, :terrain_status, :pending)
    else
      changes
    end
  end

  defp maybe_recompute_positions(changes, contact, proposed) do
    needs_recompute =
      Map.has_key?(proposed, "grid1") or
        Map.has_key?(proposed, "grid2") or
        Map.has_key?(proposed, "band") or
        Map.has_key?(proposed, "qso_timestamp")

    if needs_recompute do
      recompute_positions(changes, contact, proposed)
    else
      changes
    end
  end

  defp recompute_positions(changes, contact, proposed) do
    pos1 = resolve_grid_position(Map.get(proposed, "grid1", contact.grid1)) || contact.pos1
    pos2 = resolve_grid_position(Map.get(proposed, "grid2", contact.grid2)) || contact.pos2
    distance = compute_distance(pos1, pos2)

    changes
    |> Map.put(:pos1, pos1)
    |> Map.put(:pos2, pos2)
    |> Map.put(:distance_km, distance)
    |> Map.put(:hrrr_status, :pending)
    |> Map.put(:weather_status, :pending)
    |> Map.put(:terrain_status, :pending)
    |> Map.put(:iemre_status, :pending)
  end

  defp compute_distance(pos1, pos2) when not is_nil(pos1) and not is_nil(pos2) do
    lon1 = pos1["lon"]
    lon2 = pos2["lon"]
    pos1["lat"] |> haversine_km(lon1, pos2["lat"], lon2) |> round() |> Decimal.new()
  end

  defp compute_distance(_, _), do: nil

  defp build_contact_changes(_contact, proposed) do
    Enum.reduce(proposed, %{}, fn
      {"band", val}, acc ->
        Map.put(acc, :band, Decimal.new(to_string(val)))

      {"qso_timestamp", val}, acc when is_binary(val) ->
        {:ok, dt, _} = DateTime.from_iso8601(val)
        Map.put(acc, :qso_timestamp, dt)

      {"qso_timestamp", %DateTime{} = dt}, acc ->
        Map.put(acc, :qso_timestamp, dt)

      {key, val}, acc ->
        Map.put(acc, String.to_existing_atom(key), val)
    end)
  end

  defp resolve_grid_position(nil), do: nil

  defp resolve_grid_position(grid) do
    case Maidenhead.to_latlon(grid) do
      {:ok, {lat, lon}} -> %{"lat" => lat, "lon" => lon}
      _ -> nil
    end
  end
end