ash_dispatch

defmodule AshDispatch.Event.CustomTopic do
  @moduledoc """
  Lightweight `:custom_topic` transport for `AshDispatch.Event`.

  Some events are fire-and-forget per-record PubSub broadcasts to a
  fixed (or per-record) topic — no recipients, no content rendering,
  no `DeliveryReceipt` rows. The full `AshDispatch.Event` Spark DSL +
  `Dispatcher` pipeline is the wrong shape for that workload: it
  exists to fan a single event out across N channels with per-
  audience templates and receipts.

  `use AshDispatch.Event, transports: [custom_topic: [...]]` injects a
  thin helper layer that bypasses the dispatcher and calls
  `Phoenix.PubSub.broadcast/3` directly, with the safe-broadcast +
  telemetry wrapping that hand-rolled broadcasters were otherwise
  re-implementing.

  ## Usage

      defmodule MyApp.Events.SomethingHappened do
        use AshDispatch.Event,
          transports: [
            custom_topic: [
              pubsub: MyApp.PubSub,
              topic: "things:all",
              event_name: :something_happened
            ]
          ]

        def broadcast(record) do
          payload = %{id: record.id, foo: record.foo}
          safe_broadcast({event_name(), payload})
        end
      end

  ## Options

    * `:pubsub` (required) — the `Phoenix.PubSub` module to broadcast through.
    * `:topic` (required) — either a string topic OR an `{Module, :function}`
      tuple resolved at broadcast time with the record as the single arg.
      Per-record routing is the foundation for per-wave / per-crucible
      topic variants without changing the call sites.
    * `:event_name` (optional) — atom subscribers receive in the wire-shape
      `{event_name, payload}`. Defaults to the underscored last segment
      of the module name (e.g. `EngagementClaimed` → `:engagement_claimed`).

  ## Generated helpers

  Injected into the implementor module:

    * `topic/0` — static topic string (raises if `:topic` is an MFA tuple).
    * `topic/1` — resolves the topic for a given record (calls the MFA
      tuple or returns the static string).
    * `event_name/0` — the wire-event atom.
    * `safe_broadcast/1` — wraps `Phoenix.PubSub.broadcast/3` with rescue +
      `[:ash_dispatch, :custom_topic, :broadcast_failure]` telemetry +
      log. Always returns `:ok`. Takes the `{event_name, payload}` tuple.
    * `safe_broadcast/2` — same as `/1` but takes an explicit record for
      per-record topic resolution: `safe_broadcast(record, {evt, payload})`.

  All generated functions are `defoverridable` — implementors can replace
  any of them without losing the others' compile-time wiring.

  ## Why this lives next to `AshDispatch.Event`

  The `use` call is the same module name; the heavyweight DSL path
  (channels / recipients / content) remains the default when no
  `:transports` option is passed. The split happens at macro-expansion
  time: when `transports: [custom_topic: ...]` is present, this module's
  `inject/1` returns extra quoted AST appended to the existing DSL
  setup; when absent, behaviour is identical to pre-extension.
  """

  @doc false
  def inject(opts) do
    pubsub = Keyword.fetch!(opts, :pubsub)
    topic_spec = Keyword.fetch!(opts, :topic)
    event_name_override = Keyword.get(opts, :event_name)

    # Macro-time validation: topic must be a literal string OR a 2-tuple
    # `{Module, :function}`. Runtime values aren't supported because the
    # spec is captured at compile time for fast access.
    validate_topic_spec!(topic_spec)

    quote do
      require Logger

      @custom_topic_pubsub unquote(pubsub)
      @custom_topic_spec unquote(Macro.escape(topic_spec))
      @custom_topic_event_name unquote(event_name_override) ||
                                 AshDispatch.Event.CustomTopic.derive_event_name(__MODULE__)

      @doc "Static PubSub topic this event publishes to. Raises if topic is per-record."
      @spec topic() :: String.t()
      def topic do
        case @custom_topic_spec do
          topic when is_binary(topic) ->
            topic

          {_mod, _fun} ->
            raise ArgumentError,
                  "#{inspect(__MODULE__)} uses a per-record topic; call topic/1 with the record"
        end
      end

      @doc "Topic resolved for a given record (handles static + per-record MFA)."
      @spec topic(term()) :: String.t()
      def topic(record) do
        case @custom_topic_spec do
          topic when is_binary(topic) -> topic
          {mod, fun} -> apply(mod, fun, [record])
        end
      end

      @doc "Wire-event atom subscribers receive in `{event_name, payload}`."
      @spec event_name() :: atom()
      def event_name, do: @custom_topic_event_name

      @doc """
      Broadcast a `{event_name, payload}` tuple to the static topic.

      Failures are caught + logged + telemetry-emitted as
      `[:ash_dispatch, :custom_topic, :broadcast_failure]`. Always
      returns `:ok` — broadcasts are notification side-channels, not
      sources of truth.
      """
      @spec safe_broadcast({atom(), term()}) :: :ok
      def safe_broadcast({_evt, _payload} = message) do
        AshDispatch.Event.CustomTopic.do_broadcast(
          __MODULE__,
          @custom_topic_pubsub,
          topic(),
          message
        )
      end

      @doc "Broadcast for per-record topic routing — resolves topic from record."
      @spec safe_broadcast(term(), {atom(), term()}) :: :ok
      def safe_broadcast(record, {_evt, _payload} = message) do
        AshDispatch.Event.CustomTopic.do_broadcast(
          __MODULE__,
          @custom_topic_pubsub,
          topic(record),
          message
        )
      end

      defoverridable topic: 0, topic: 1, event_name: 0, safe_broadcast: 1, safe_broadcast: 2
    end
  end

  @doc false
  def do_broadcast(module, pubsub, topic, {event_name, _payload} = message) do
    Phoenix.PubSub.broadcast(pubsub, topic, message)
    :ok
  rescue
    e ->
      require Logger

      Logger.warning(
        "[#{inspect(module)}] custom_topic broadcast failed: #{Exception.message(e)}"
      )

      :telemetry.execute(
        [:ash_dispatch, :custom_topic, :broadcast_failure],
        %{count: 1},
        %{module: module, event: event_name, topic: topic, error: inspect(e)}
      )

      :ok
  end

  @doc false
  def derive_event_name(module) do
    module
    |> Module.split()
    |> List.last()
    |> Macro.underscore()
    |> String.to_atom()
  end

  defp validate_topic_spec!(topic) when is_binary(topic), do: :ok
  defp validate_topic_spec!({mod, fun}) when is_atom(mod) and is_atom(fun), do: :ok

  defp validate_topic_spec!(other) do
    raise ArgumentError, """
    Invalid :topic for AshDispatch.Event custom_topic transport.

    Expected a string topic OR a `{Module, :function}` 2-tuple for
    per-record routing. Got: #{inspect(other)}
    """
  end
end