defmodule MvWeb.MemberLive.Index.OverviewQuery do
  @moduledoc """
  Builds the Ash query for the member overview from the LiveView's filter/sort
  state, so every filter and sort resolves in PostgreSQL via the `:overview`
  keyset-paginated read action.

  All previously in-memory passes (cycle status, boolean/date custom fields,
  group sort) are expressed here as DB filters/sorts. A unique `id` tie-breaker
  is always appended to the sort so keyset pages never skip or duplicate rows.

  `build/1` returns an unread `Ash.Query`; the caller supplies `page:`/`actor:`
  options to `Ash.read/2`.
  """

  import Ash.Expr

  alias Mv.Membership.Member
  alias MvWeb.MemberLive.Index.DateFilter

  require Ash.Query
  require Ash.Sort

  @type opts :: %{optional(atom()) => term()}

  @doc """
  Builds the `:overview` query from the given filter/sort options.

  Recognised keys (all optional):

    * `:search` — full-text search string
    * `:group_filters` / `:groups` — `%{group_id => :in | :not_in}` and the valid groups
    * `:fee_type_filters` / `:fee_types` — `%{fee_type_id => :in | :not_in}` and valid fee types
    * `:date_filters` — built-in join/exit date filter map (see `DateFilter`)
    * `:sort_field` / `:sort_order` — sort key and direction
    * `:today` — reference date for cycle math (defaults to `Date.utc_today/0`)
  """
  @spec build(opts()) :: Ash.Query.t()
  def build(opts \\ %{}) do
    Member
    |> Ash.Query.for_read(:overview)
    |> apply_search(opts[:search])
    |> apply_group_filters(opts[:group_filters], opts[:groups])
    |> apply_fee_type_filters(opts[:fee_type_filters], opts[:fee_types])
    |> apply_boolean_custom_field_filters(
      opts[:boolean_custom_field_filters],
      opts[:boolean_custom_fields]
    )
    |> apply_date_filters(opts[:date_filters])
    |> apply_stichtag_filter(opts[:stichtag])
    |> apply_custom_date_filters(opts[:date_filters], opts[:date_custom_fields])
    |> apply_payment_filter(opts[:payment_filter], payment_period(opts))
    |> apply_suspended_filter(opts[:suspended], payment_period(opts))
    |> apply_sort(
      opts[:sort_field],
      opts[:sort_order],
      opts[:custom_fields] || [],
      payment_period(opts)
    )
  end

  # ---------------------------------------------------------------------------
  # Search
  # ---------------------------------------------------------------------------

  defp apply_search(query, search) when is_binary(search),
    do: Member.apply_overview_search(query, search)

  defp apply_search(query, _), do: query

  # ---------------------------------------------------------------------------
  # Group filters (:in OR across selected groups; :not_in AND-of-not-exists)
  #
  # Multiple "is" selections bundle into a single membership `exists` over the
  # union of group ids, so a member matching *any* selected group is kept (OR,
  # §1.29/§2.7). Each "is not" selection stays an independent not-exists, so a
  # member is excluded when it belongs to *any* of them.
  # ---------------------------------------------------------------------------

  defp apply_group_filters(query, group_filters, _groups)
       when group_filters in [nil, %{}],
       do: query

  defp apply_group_filters(query, group_filters, groups) do
    valid_ids = valid_id_set(groups)

    {in_filters, not_in_filters} =
      group_filters
      |> Enum.filter(fn {id_str, _} -> MapSet.member?(valid_ids, id_str) end)
      |> Enum.split_with(fn {_, value} -> value == :in end)

    query
    |> apply_group_in_filter(Enum.map(in_filters, fn {id_str, _} -> id_str end))
    |> apply_group_not_in_filters(Enum.map(not_in_filters, fn {id_str, _} -> id_str end))
  end

  defp apply_group_in_filter(query, id_strs) do
    case cast_uuids(id_strs) do
      [] -> query
      uuids -> Ash.Query.filter(query, expr(exists(member_groups, group_id in ^uuids)))
    end
  end

  defp apply_group_not_in_filters(query, id_strs) do
    Enum.reduce(id_strs, query, fn id_str, q ->
      case Ecto.UUID.cast(id_str) do
        {:ok, uuid} -> Ash.Query.filter(q, expr(not exists(member_groups, group_id == ^uuid)))
        _ -> q
      end
    end)
  end

  # ---------------------------------------------------------------------------
  # Fee-type filters (:in OR; :not_in AND)
  # ---------------------------------------------------------------------------

  defp apply_fee_type_filters(query, fee_type_filters, _fee_types)
       when fee_type_filters in [nil, %{}],
       do: query

  defp apply_fee_type_filters(query, fee_type_filters, fee_types) do
    valid_ids = valid_id_set(fee_types)

    {in_filters, not_in_filters} =
      fee_type_filters
      |> Enum.filter(fn {id_str, _} -> MapSet.member?(valid_ids, id_str) end)
      |> Enum.split_with(fn {_, value} -> value == :in end)

    in_uuids = cast_uuids(Enum.map(in_filters, fn {id_str, _} -> id_str end))

    query =
      if in_uuids == [] do
        query
      else
        Ash.Query.filter(query, expr(membership_fee_type_id in ^in_uuids))
      end

    Enum.reduce(not_in_filters, query, fn {id_str, _}, q ->
      case Ecto.UUID.cast(id_str) do
        {:ok, uuid} ->
          Ash.Query.filter(
            q,
            expr(membership_fee_type_id != ^uuid or is_nil(membership_fee_type_id))
          )

        _ ->
          q
      end
    end)
  end

  # ---------------------------------------------------------------------------
  # Boolean custom-field filters (AND across selected fields)
  #
  # Mirrors the previous in-memory behaviour: a member matches a `{field => bool}`
  # filter only if it has a stored value row for that field whose boolean value
  # equals `bool`. A member with no value row is excluded. Uses a JSONB
  # containment predicate (`value @> '{"value": <bool>}'`), which the GIN index on
  # `custom_field_values.value` can serve.
  # ---------------------------------------------------------------------------

  defp apply_boolean_custom_field_filters(query, filters, _boolean_fields)
       when filters in [nil, %{}],
       do: query

  defp apply_boolean_custom_field_filters(query, filters, boolean_fields) do
    valid_ids = valid_id_set(boolean_fields)

    filters
    |> Enum.filter(fn {id_str, _} -> MapSet.member?(valid_ids, id_str) end)
    |> Enum.reduce(query, fn {id_str, bool}, q ->
      case Ecto.UUID.cast(id_str) do
        {:ok, uuid} -> apply_one_boolean_filter(q, uuid, bool)
        _ -> q
      end
    end)
  end

  defp apply_one_boolean_filter(query, uuid, bool) when is_boolean(bool) do
    Ash.Query.filter(
      query,
      expr(
        exists(
          custom_field_values,
          custom_field_id == ^uuid and
            fragment("? @> jsonb_build_object('value', ?::boolean)", value, ^bool)
        )
      )
    )
  end

  defp apply_one_boolean_filter(query, _uuid, _bool), do: query

  # ---------------------------------------------------------------------------
  # Payment aging filter (period-scoped unpaid-cycle count)
  #
  # `fully_paid` keeps members with zero unpaid cycles whose cycle_end lies in
  # the active period; `{:has_unpaid, n}` keeps members with at least n such
  # cycles. Both push to SQL via the `unpaid_cycle_count` calculation (a
  # correlated aggregate over `membership_fee_cycles`), so no member set is
  # loaded into memory. Suspended/paid cycles are excluded by the calculation.
  # ---------------------------------------------------------------------------

  defp payment_period(opts) do
    case opts[:payment_period] do
      %{from: _, to: _} = period -> period
      _ -> %{from: nil, to: nil}
    end
  end

  defp apply_payment_filter(query, :fully_paid, %{from: from, to: to}) do
    Ash.Query.filter(
      query,
      expr(unpaid_cycle_count(period_from: ^from, period_to: ^to) == 0)
    )
  end

  defp apply_payment_filter(query, {:has_unpaid, n}, %{from: from, to: to})
       when is_integer(n) and n > 0 do
    Ash.Query.filter(
      query,
      expr(unpaid_cycle_count(period_from: ^from, period_to: ^to) >= ^n)
    )
  end

  # Two-sided open-cycle range (§1.22/§3.14): keeps members whose in-period
  # unpaid count is within [min, max]. `max` nil means unbounded above, so
  # `{:unpaid_range, 1, nil}` is the default "has open" (>= 1). Both bounds push
  # to SQL through the same `unpaid_cycle_count` calculation.
  defp apply_payment_filter(query, {:unpaid_range, min, max}, %{from: from, to: to})
       when is_integer(min) and min > 0 do
    query
    |> Ash.Query.filter(unpaid_cycle_count(period_from: ^from, period_to: ^to) >= ^min)
    |> apply_unpaid_upper_bound(max, from, to)
  end

  defp apply_payment_filter(query, _filter, _period), do: query

  defp apply_unpaid_upper_bound(query, max, from, to) when is_integer(max) and max > 0 do
    Ash.Query.filter(
      query,
      expr(unpaid_cycle_count(period_from: ^from, period_to: ^to) <= ^max)
    )
  end

  defp apply_unpaid_upper_bound(query, _max, _from, _to), do: query

  # Suspended-status filter (§1.23): keeps members with at least one suspended
  # cycle whose cycle_end falls in the active period. Independent of the unpaid
  # count, which excludes suspended (§2.5). A correlated `exists` pushes to SQL;
  # the period bounds are composed conditionally so an unbounded (nil) side adds
  # no `cycle_end` comparison rather than comparing against a literal nil.
  defp apply_suspended_filter(query, true, %{from: from, to: to}) do
    Ash.Query.filter(
      query,
      expr(exists(membership_fee_cycles, ^suspended_period_predicate(from, to)))
    )
  end

  defp apply_suspended_filter(query, _suspended, _period), do: query

  defp suspended_period_predicate(nil, nil), do: expr(status == :suspended)

  defp suspended_period_predicate(from, nil),
    do: expr(status == :suspended and cycle_end >= ^from)

  defp suspended_period_predicate(nil, to),
    do: expr(status == :suspended and cycle_end <= ^to)

  defp suspended_period_predicate(from, to),
    do: expr(status == :suspended and cycle_end >= ^from and cycle_end <= ^to)

  # ---------------------------------------------------------------------------
  # Built-in date filters (join/exit)
  # ---------------------------------------------------------------------------

  defp apply_date_filters(query, nil), do: query

  defp apply_date_filters(query, filters) when is_map(filters),
    do: DateFilter.apply_ash_filter(query, filters)

  defp apply_custom_date_filters(query, filters, custom_fields)
       when is_map(filters) and is_list(custom_fields),
       do: DateFilter.apply_custom_date_ash_filter(query, filters, custom_fields)

  defp apply_custom_date_filters(query, _filters, _custom_fields), do: query

  # ---------------------------------------------------------------------------
  # Point-in-time membership ("Stichtag"): a member counts as a member on the
  # reference date X iff it had joined by X and had not yet left as of X
  # (§1.24/§2.6). "Left as of X" means a non-nil exit_date at or before X, so an
  # exit_date strictly after X (or nil) still counts as a member on X.
  # ---------------------------------------------------------------------------

  defp apply_stichtag_filter(query, %Date{} = x) do
    Ash.Query.filter(
      query,
      expr(join_date <= ^x and (is_nil(exit_date) or exit_date > ^x))
    )
  end

  defp apply_stichtag_filter(query, _), do: query

  # ---------------------------------------------------------------------------
  # Sort (always append the unique id tie-breaker for keyset stability)
  # ---------------------------------------------------------------------------

  defp apply_sort(query, nil, _order, _custom_fields, _period),
    do: Ash.Query.sort(query, id: :asc)

  defp apply_sort(query, _field, nil, _custom_fields, _period),
    do: Ash.Query.sort(query, id: :asc)

  # Composite "Member" column: sort by last name then first name. The email that
  # shares the cell is display-only and does not affect the sort key (§1.21).
  defp apply_sort(query, field, order, _custom_fields, _period) when field in [:name, "name"] do
    Ash.Query.sort(query, [{:last_name, order}, {:first_name, order}, {:id, :asc}])
  end

  # Composite "Address" column: sort by city, then postal code, then street.
  defp apply_sort(query, field, order, _custom_fields, _period)
       when field in [:address, "address"] do
    Ash.Query.sort(query, [
      {:city, order},
      {:postal_code, order},
      {:street, order},
      {:id, :asc}
    ])
  end

  # "Beiträge" (payment) column: sort DB-side by the period-scoped unpaid-cycle
  # count (§1.27). The count calculation takes the active period as arguments so
  # the sort key matches the displayed value; the id tie-breaker keeps keyset
  # pages stable across equal counts.
  defp apply_sort(query, field, order, _custom_fields, %{from: from, to: to})
       when field in [:payment, "payment"] do
    Ash.Query.sort(query, [
      {Ash.Sort.expr_sort(unpaid_cycle_count(period_from: ^from, period_to: ^to), :integer),
       order},
      {:id, :asc}
    ])
  end

  defp apply_sort(query, field, order, custom_fields, _period) do
    case sort_key(field) do
      nil -> apply_custom_field_sort(query, field, order, custom_fields)
      key -> Ash.Query.sort(query, [{key, order}, {:id, :asc}])
    end
  end

  # Sorts by a custom field's JSONB value DB-side. The value is read from the
  # first matching `custom_field_values` row for the field, cast per the field's
  # value type so integers sort numerically and dates chronologically. Empty
  # strings and members without a value row sort last in both directions
  # (NULLS LAST), mirroring the previous in-memory `CustomFieldSort` behaviour.
  defp apply_custom_field_sort(query, field, order, custom_fields) do
    with id_str when is_binary(id_str) <- custom_field_id(field),
         %{} = cf <- Enum.find(custom_fields, &(to_string(&1.id) == id_str)) do
      nils_order = if order == :desc, do: :desc_nils_last, else: :asc_nils_last

      Ash.Query.sort(query, [
        {custom_field_sort_calc(cf.value_type, cf.id), nils_order},
        {:id, :asc}
      ])
    else
      _ -> Ash.Query.sort(query, id: :asc)
    end
  end

  defp custom_field_id(field) when is_atom(field), do: custom_field_id(Atom.to_string(field))

  defp custom_field_id(field) when is_binary(field) do
    case String.split(field, Mv.Constants.custom_field_prefix()) do
      ["", id_str] -> id_str
      _ -> nil
    end
  end

  defp custom_field_id(_), do: nil

  # Maps the field's value type to the first matching custom-field value's typed
  # sort key (scoped to the field), backed by the `sort_*` calculations on
  # `CustomFieldValue` (a raw fragment is not allowed as an aggregate `field:`).
  defp custom_field_sort_calc(:integer, id),
    do:
      Ash.Sort.expr_sort(
        first(custom_field_values,
          field: :sort_numeric,
          query: [filter: expr(custom_field_id == ^id)]
        ),
        :decimal
      )

  defp custom_field_sort_calc(:date, id),
    do:
      Ash.Sort.expr_sort(
        first(custom_field_values,
          field: :sort_date,
          query: [filter: expr(custom_field_id == ^id)]
        ),
        :date
      )

  defp custom_field_sort_calc(:boolean, id),
    do:
      Ash.Sort.expr_sort(
        first(custom_field_values,
          field: :sort_boolean,
          query: [filter: expr(custom_field_id == ^id)]
        ),
        :boolean
      )

  defp custom_field_sort_calc(_type, id),
    do:
      Ash.Sort.expr_sort(
        first(custom_field_values,
          field: :sort_text,
          query: [filter: expr(custom_field_id == ^id)]
        ),
        :string
      )

  # Resolves a sort field (atom or string) to a DB sort key, or nil if it is a
  # computed field handled elsewhere / not sortable.
  defp sort_key(field) when field in [:membership_fee_type, "membership_fee_type"],
    do: "membership_fee_type.name"

  defp sort_key(field) when field in [:groups, "groups"], do: :first_group_name

  defp sort_key(field) when is_atom(field) do
    if field in member_sort_fields(), do: field, else: nil
  end

  defp sort_key(field) when is_binary(field) do
    allowed = MapSet.new(member_sort_fields(), &Atom.to_string/1)
    if MapSet.member?(allowed, field), do: String.to_existing_atom(field), else: nil
  end

  defp sort_key(_), do: nil

  defp member_sort_fields do
    Mv.Constants.member_fields() -- [:notes]
  end

  # ---------------------------------------------------------------------------
  # Helpers
  # ---------------------------------------------------------------------------

  defp valid_id_set(records) when is_list(records) do
    records
    |> Enum.map(&to_string(&1.id))
    |> Enum.map(&normalize_uuid/1)
    |> Enum.reject(&is_nil/1)
    |> MapSet.new()
  end

  defp valid_id_set(_), do: MapSet.new()

  defp normalize_uuid(raw) when is_binary(raw) do
    case Ecto.UUID.cast(String.trim(raw)) do
      {:ok, uuid} -> to_string(uuid)
      _ -> nil
    end
  end

  defp normalize_uuid(_), do: nil

  defp cast_uuids(id_strs) do
    id_strs
    |> Enum.map(&Ecto.UUID.cast/1)
    |> Enum.filter(&match?({:ok, _}, &1))
    |> Enum.map(fn {:ok, uuid} -> uuid end)
  end
end