ex_ihdl_subscription_base

defmodule ExIhdlSubscriptionBase do
  @moduledoc """
  Thin implementation of the ExGcloudPubsubPuller.PullController behaviour.

  To make use of this module, `use` it, and implement the callbacks.

  ## Examples

      defmodule MySub do
        use ExIhdlSubscriptionBase

        @impl true
        def moap_module(), do: "hm"

        @impl true
        def moap_key(), do: "rssi"

        @impl true
        def registry_topic(), do: "some-events-topic"

        @impl true
        def subscription_suffix, do: "some-service"

        @impl true
        def process_message(%PubsubMessage{data: data}) do
          log_info("process_message: \#{inspect(data)}")

          cond do
            do_something(data) ->
              :ok
            true ->
              :error
          end
        end
      end
  """

  @doc """
  Should return the MOAP object under which to look for the MOAP key.

  ## Examples

      def moap_module(), do: "hm"
  """
  @callback moap_module() :: String.t()

  @doc """
  Should return the MOAP key for the desired data.

  ## Examples

      def moap_key(), do: "rssi"
  """
  @callback moap_key() :: String.t()

  @doc """
  Should return the topic from which all events are sourced.

  ## Examples

      def registry_topic(), do: "staging-eon-registry-events"
  """
  @callback registry_topic() :: String.t()

  @doc """
  Should return the suffix to append to the generated subscription, recommended to be unique to this application (e.g. it's name).

  ## Examples

      def subscription_suffix(), do: "my-app"
  """
  @callback subscription_suffix() :: String.t()

  @doc """
  Gets called with each message received from the ingest pipe. Should return `:ok` or `:error`.

  ## Examples

      def process_message(msg), do: something(msg)
  """
  @callback process_message(GoogleApi.PubSub.V1.Model.PubsubMessage.t()) :: :ok | :error

  defmacro __using__(_) do
    quote do
      @behaviour ExGcloudPubsubPuller.PullController
      @behaviour ExIhdlSubscriptionBase
      alias ExIhdlSubscriptionBase.IngestPipe
      alias ExIhdlSubscriptionBase.MemoryStore
      alias GoogleApi.PubSub.V1.Model.PubsubMessage
      require Logger

      defp log_prefixed(msg), do: [log_prefix(), msg] |> Enum.join()
      defp log_prefix(), do: ["[", moap_module(), ".", moap_key(), "] "] |> Enum.join()
      defp log_error(msg), do: msg |> log_prefixed() |> Logger.error()
      defp log_warn(msg), do: msg |> log_prefixed() |> Logger.warn()
      defp log_info(msg), do: msg |> log_prefixed() |> Logger.info()

      defp memory_store_key(), do: [moap_module(), moap_key()] |> Enum.join()
      defp schedule_subscription_refresh(), do: MemoryStore.save(memory_store_key(), nil)

      @impl true
      def subscription_id() do
        log_info("subscription_id: getting..")

        case MemoryStore.load(memory_store_key()) do
          nil ->
            log_info("subscription_id: getting from ingest pipe api..")

            pipe_args = %{
              moap_module: moap_module(),
              moap_key: moap_key(),
              registry_topic: registry_topic(),
              subscription_suffix: subscription_suffix()
            }

            {:ok, %{"topic" => %{"subscriptions" => subscriptions}}} =
              IngestPipe.ensure_pipe(pipe_args)

            {:ok, %{"name" => subscription_id}} =
              IngestPipe.find_subscription(subscriptions, pipe_args)

            MemoryStore.save(memory_store_key(), %{
              subscription_id: subscription_id
            })

            log_info("subscription_id: got from ingest pipe api #{subscription_id}")
            subscription_id

          %{subscription_id: subscription_id} ->
            log_info("subscription_id: got from memory store #{subscription_id}")
            subscription_id
        end
      end

      @impl true
      def handle_message(%PubsubMessage{data: nil}) do
        log_error("handle_message: received empty message!")
        :error
      end

      @impl true
      def handle_message(%PubsubMessage{data: data} = msg) do
        log_info("handle_message: #{inspect(data)}")
        process_message(msg |> Map.from_struct())
      end

      @impl true
      def handle_stagnant() do
        log_warn("handle_stagnant: subscription is stagnant!")

        schedule_subscription_refresh()
      end

      @impl true
      def handle_pull_error(%Tesla.Env{status: 404}) do
        log_error("handle_pull_error: subscription doesn't exist!")

        schedule_subscription_refresh()
      end

      @impl true
      def handle_pull_error(%Tesla.Env{status: 403}),
        do: log_error("handle_pull_error: I am not authorized!")

      @impl true
      def handle_pull_error(msg), do: log_error("handle_pull_error: #{inspect(msg)}")

      @impl true
      def handle_ack_error(%Tesla.Env{status: 404}),
        do: log_error("handle_ack_error: message doesn't exist!")

      @impl true
      def handle_ack_error(%Tesla.Env{status: 403}),
        do: log_error("handle_ack_error: I am not authorized!")

      @impl true
      def handle_ack_error(msg), do: log_error("handle_ack_error: #{inspect(msg)}")
    end
  end
end