ash_dispatch

defmodule AshDispatch.Notifier.CounterHandler do
  @moduledoc """
  Side-effect orchestration for counter broadcasts — extracted from the
  prior `AshDispatch.Changes.BroadcastCounterUpdate` module.

  ## Why this exists separately from the notifier

  `AshDispatch.Notifier` is a thin adapter that pattern-matches the
  notification's action and looks up per-action config. The actual
  counter logic — recipient resolution (relationship-based, filter-
  based, MFA-based), query execution with scoping/filtering, and the
  configured `counter_broadcast_fn` invocation — lives here. This
  isolates the "what to broadcast" decisions (Notifier) from the "how
  to broadcast" orchestration (CounterHandler).

  ## Preserved verbatim from the prior change module

  The recipient resolution, query construction, scope-expression
  application, filter-by-record, and broadcast invocation logic is
  carried over unchanged from `Changes.BroadcastCounterUpdate`. The
  ONLY behavioural change is **when** this code runs: it used to fire
  via `Ash.Changeset.after_action/2` synchronously inside the action's
  transaction; it now fires from `Ash.Notifier.notify/1` post-commit
  (or is dropped on rollback).
  """

  alias AshDispatch.Config
  alias AshDispatch.Helpers.RecordReader
  alias AshDispatch.Helpers.ResourceIntrospection

  require Ash.Query
  require Logger

  @doc """
  Broadcast a single counter for a notification.

  Takes the notification (an `%Ash.Notifier.Notification{}`) and the
  per-action `counter_config` persisted by `InjectCounterBroadcasts`.
  Equivalent to the prior `broadcast_counter/2` private function
  called from the change's `after_action` hook.
  """
  @spec broadcast(Ash.Notifier.Notification.t(), keyword() | map()) :: :ok
  def broadcast(%Ash.Notifier.Notification{} = notification, config) do
    record = notification.data
    opts = config_to_opts(config)
    broadcast_counter(record, opts)
  end

  defp config_to_opts(config) when is_list(config), do: config

  defp config_to_opts(config) when is_map(config) do
    config
    |> Map.to_list()
    |> Enum.reject(fn {k, _} -> k == :__struct__ end)
  end

  # ── broadcast_counter — preserved verbatim from BroadcastCounterUpdate ──

  defp broadcast_counter(record, opts) do
    counter_name = Keyword.fetch!(opts, :counter_name)
    resource = Keyword.fetch!(opts, :resource)
    query_filter = Keyword.fetch!(opts, :query_filter)
    audience = Keyword.fetch!(opts, :audience)
    invalidates = Keyword.get(opts, :invalidates, [])
    filter_by_record = Keyword.get(opts, :filter_by_record)

    recipients = resolve_recipients_for_counter(record, audience, opts)

    authorize? = Keyword.get(opts, :authorize?, true)
    scope = Keyword.get(opts, :scope)

    user_id_path =
      ResourceIntrospection.resolve_user_id_path_for_scoping(resource,
        authorize?: authorize?,
        scope: scope,
        user_id_path: Keyword.get(opts, :user_id_path),
        audience: audience
      )

    Enum.each(recipients, fn recipient ->
      count =
        try do
          query =
            resource
            |> Ash.Query.new()

          query = apply_query_filter(query, query_filter)

          query =
            cond do
              scope && !filter_by_record ->
                apply_scope_expression(query, scope, recipient)

              user_id_path && !filter_by_record ->
                user_filter = ResourceIntrospection.build_user_filter(user_id_path, recipient.id)
                Ash.Query.do_filter(query, user_filter)

              true ->
                query
            end

          query = apply_filter_by_record(query, record, filter_by_record)

          Ash.count!(query, authorize?: authorize?, actor: recipient)
        rescue
          e ->
            Logger.error(
              "[AshDispatch.Notifier.CounterHandler] Failed to count #{counter_name}: #{inspect(e)}\n" <>
                "Resource: #{inspect(resource)}, Query filter: #{inspect(query_filter)}"
            )

            0
        end

      broadcast_to_user(recipient.id, counter_name, count, invalidates, audience)
    end)
  end

  # ── Recipient resolution — preserved verbatim ──────────────────

  defp resolve_recipients_for_counter(record, audience, opts) do
    if ResourceIntrospection.is_relationship_audience?(audience) do
      case extract_user_from_record(record, audience, opts) do
        nil -> []
        user -> [user]
      end
    else
      context = %{data: %{record: record}}
      channel = %{audience: audience}
      AshDispatch.Event.Helpers.resolve_recipients_for_audience(context, channel)
    end
  end

  defp extract_user_from_record(record, audience, opts) do
    user_id = extract_user_id(record, audience, opts)

    if user_id do
      %{id: user_id}
    else
      nil
    end
  end

  defp extract_user_id(record, audience, opts) do
    case Keyword.get(opts, :user_id_path) do
      nil ->
        relationship_name = ResourceIntrospection.get_audience_relationship(audience)

        user_id_field =
          Keyword.get(opts, :user_id_field) ||
            (relationship_name && String.to_atom("#{relationship_name}_id")) ||
            :user_id

        RecordReader.safe_get(record, user_id_field)

      path when is_list(path) ->
        case load_and_traverse_path(record, path) do
          {:ok, user_id} ->
            user_id

          {:error, reason} ->
            Logger.warning(
              "[AshDispatch.Notifier.CounterHandler] Failed to resolve user_id via path #{inspect(path)}: #{inspect(reason)}"
            )

            nil
        end
    end
  end

  defp load_and_traverse_path(record, path) do
    {relationships, [_final_field]} = Enum.split(path, -1)

    case Ash.load(record, relationships, authorize?: false) do
      {:ok, loaded_record} ->
        value = get_nested_value(loaded_record, path)
        {:ok, value}

      {:error, reason} ->
        {:error, reason}
    end
  end

  defp get_nested_value(record, [field]), do: RecordReader.safe_get(record, field)

  defp get_nested_value(record, [field | rest]) do
    case RecordReader.safe_get(record, field) do
      nil -> nil
      nested_record -> get_nested_value(nested_record, rest)
    end
  end

  # ── Query filter / scope / filter_by_record — preserved verbatim ──

  defp apply_query_filter(query, nil), do: query
  defp apply_query_filter(query, []), do: query

  defp apply_query_filter(query, query_filter) when is_list(query_filter) do
    if Keyword.keyword?(query_filter) do
      Enum.reduce(query_filter, query, fn {field, value}, acc_query ->
        import Ash.Query
        import Ash.Expr

        if is_list(value) do
          filter(acc_query, ^ref(field) in ^value)
        else
          filter(acc_query, ^ref(field) == ^value)
        end
      end)
    else
      Logger.warning(
        "[AshDispatch.Notifier.CounterHandler] Unknown query_filter format (non-keyword list): #{inspect(query_filter)}"
      )

      query
    end
  end

  defp apply_query_filter(query, query_filter) do
    Ash.Query.filter(query, ^query_filter)
  rescue
    e ->
      Logger.warning(
        "[AshDispatch.Notifier.CounterHandler] Failed to apply query_filter: #{inspect(e)}, filter: #{inspect(query_filter)}"
      )

      query
  end

  defp apply_scope_expression(query, scope_expr, recipient) do
    query
    |> Ash.Query.set_context(%{actor: recipient})
    |> Ash.Query.filter(^scope_expr)
  end

  defp apply_filter_by_record(query, _record, nil), do: query

  defp apply_filter_by_record(query, record, filter_config) do
    filter_field = get_config_value(filter_config, :field)
    record_field = get_config_value(filter_config, :record_field, :id)

    if filter_field do
      filter_value = RecordReader.safe_get(record, record_field)

      if filter_value do
        import Ash.Query
        import Ash.Expr

        filter(query, ^ref(filter_field) == ^filter_value)
      else
        Logger.warning(
          "[AshDispatch.Notifier.CounterHandler] Could not extract #{record_field} from record for filtering"
        )

        query
      end
    else
      query
    end
  end

  defp get_config_value(config, key, default \\ nil) do
    cond do
      is_list(config) -> Keyword.get(config, key, default)
      is_map(config) -> Map.get(config, key, default)
      true -> default
    end
  end

  # ── Broadcast — preserved verbatim ─────────────────────────────

  defp broadcast_to_user(user_id, counter_name, count, invalidates, _audience) do
    case Config.counter_broadcast_fn() do
      nil ->
        Logger.warning(
          "[AshDispatch.Notifier.CounterHandler] No counter_broadcast_fn configured, skipping broadcast"
        )

      broadcast_fn when is_function(broadcast_fn, 4) ->
        metadata = %{invalidate_queries: invalidates}
        broadcast_fn.(user_id, counter_name, count, metadata: metadata)

      {module, function} when is_atom(module) and is_atom(function) ->
        metadata = %{invalidate_queries: invalidates}
        apply(module, function, [user_id, counter_name, count, [metadata: metadata]])

      other ->
        Logger.error(
          "[AshDispatch.Notifier.CounterHandler] Invalid counter_broadcast_fn config: #{inspect(other)}"
        )
    end
  end
end