csh2fjhyll_kvasir_agent

defmodule Kvasir.Agent do
  @moduledoc """
  Documentation for Kvasir.Agent.
  """

  defmacro __using__(opts \\ []) do
    source =
      Macro.expand(opts[:source], __CALLER__) || raise "Need to pass the Kvasir EventSource."

    topic = Macro.expand(opts[:topic], __CALLER__) || raise "Need to pass the Kafka topic."
    model = opts[:model] || raise "Need to pass a Kvasir model."
    {cache, cache_opts} = Macro.expand(Kvasir.Agent.Config.cache!(opts), __CALLER__)
    registry = Kvasir.Agent.Config.registry!(opts)
    partitions = source.__topics__()[topic].partitions

    warmup =
      if cache do
        quote do
          @doc false
          @spec warmup :: non_neg_integer
          def warmup do
            __MODULE__
            |> unquote(cache).stream()
            |> Stream.each(fn
              {id, offset, state, cache} ->
                Supervisor.preload(
                  unquote(registry),
                  __MODULE__,
                  @key.partition!(id, @partitions),
                  id,
                  offset,
                  state,
                  cache
                )

              {i, :corrupted_state} ->
                __MODULE__.open(i)
            end)
            |> Enum.count()
          end
        end
      else
        quote do
          @doc false
          @spec warmup :: non_neg_integer
          def warmup, do: 0
        end
      end

    restore =
      if cache do
        quote do
          @doc false
          @spec check_cache_integrity(restore :: boolean) :: non_neg_integer
          def check_cache_integrity(restore)

          def check_cache_integrity(true) do
            __MODULE__
            |> unquote(cache).stream()
            |> Stream.map(fn
              {i, :corrupted_state} -> __MODULE__.open(i)
              _ -> :skip
            end)
            |> Stream.reject(&(&1 == :skip))
            |> Enum.count()
          end

          def check_cache_integrity(false) do
            __MODULE__
            |> unquote(cache).stream()
            |> Stream.map(fn
              {_, :corrupted_state} -> 1
              _ -> 0
            end)
            |> Stream.reject(&(&1 == :skip))
            |> Enum.sum()
          end
        end
      else
        quote do
          @doc false
          @spec check_cache_integrity(restore :: boolean) :: non_neg_integer
          def check_cache_integrity(restore)
          def check_cache_integrity(_), do: 0
        end
      end

    # Disabled environments
    unless ApplicationX.mix_env() in (opts[:disable] || []) do
      quote do
        use Kvasir.Command.Dispatcher
        alias Kvasir.Agent
        alias Kvasir.Agent.{Manager, Supervisor}
        @compile {:inline, __agent__: 1}
        require unquote(source)

        @source unquote(source)
        @topic unquote(topic)
        @cache unquote({cache, Macro.escape(cache_opts)})
        @registry unquote(registry)
        @model unquote(model)
        @key @source.__topics__()[@topic].key
        @partitions unquote(partitions)

        @doc false
        @spec child_spec(Keyword.t()) :: map
        def child_spec(opts \\ []), do: Agent.child_spec(__agent__(:config), opts)

        @doc false
        @spec do_dispatch(Kvasir.Command.t()) :: {:ok, Kvasir.Command.t()} | {:error, atom}
        @impl Kvasir.Command.Dispatcher
        def do_dispatch(command)

        def do_dispatch(command = %{__meta__: meta = %{scope: {:instance, id}}}) do
          case __key_check__(id, :dispatch, command) do
            {:ok, i} ->
              command = %{command | __meta__: %{meta | scope: {:instance, i}}}
              Manager.dispatch(__MODULE__, @registry, command)

            {:redirect, target} ->
              target.dispatch(command)

            err ->
              err
          end
        end

        @doc ~S"""
        Dispatch a command, raises on failure.

        ## Examples

        ```elixir
        iex> dispatch!(<cmd>, instance: <id>)
        <cmd>
        ```
        """
        @spec dispatch!(Kvasir.Command.t(), Keyword.t()) :: Kvasir.Command.t() | no_return
        def dispatch!(command, opts \\ []) do
          case dispatch(command, opts) do
            {:ok, cmd} -> cmd
            {:error, err} -> raise "Command dispatch failed: #{inspect(err)}."
          end
        end

        @doc ~S"""
        Start and return the pid of an agent instance.

        ## Examples

        ```elixir
        iex> open(<id>)
        {:ok, #PID<0.109.0>}
        ```
        """
        @spec open(any) :: {:ok, pid} | {:error, atom}
        def open(id) do
          case __key_check__(id, :open) do
            {:ok, i} ->
              Supervisor.open(unquote(registry), __MODULE__, @key.partition!(i, @partitions), i)

            {:redirect, target} ->
              target.open(id)

            err ->
              err
          end
        end

        @doc ~S"""
        Rebuild the state of an agent based on the event source.

        ## Examples

        ```elixir
        iex> rebuild(<id>)
        :ok
        ```
        """
        @spec rebuild(any, Keyword.t()) :: :ok | {:error, atom}
        def rebuild(id, opts \\ []) do
          timeout = Keyword.get(opts, :timeout, 60_000)
          async? = Keyword.get(opts, :async, false)
          callback = Keyword.get(opts, :callback, fn _ -> nil end)

          with {:ok, instance} <- open(id) do
            if async? do
              GenServer.cast(instance, {:rebuild, callback})
            else
              GenServer.call(instance, :rebuild, timeout)
            end
          end
        end

        unquote(warmup)
        unquote(restore)

        @doc ~S"""
        Inspect the current state of an agent instance.

        The agent is opened based on the given id.

        ## Examples

        ```elixir
        iex> inspect(<id>)
        {:ok, <state>}
        ```
        """
        @spec inspect(any) :: {:ok, term} | {:error, atom}
        def inspect(id) do
          case __key_check__(id, :inspect) do
            {:ok, i} ->
              Manager.inspect(unquote(registry), __MODULE__, @key.partition!(i, @partitions), i)

            {:redirect, target} ->
              target.inspect(id)

            err ->
              err
          end
        end

        @doc ~S"""
        The current amount of active agent instances.

        ## Examples

        ```elixir
        iex> count()
        4
        ```
        """
        @spec count :: non_neg_integer
        def count, do: Supervisor.count(__MODULE__, @partitions)

        @doc ~S"""
        List the IDs of the currently active agent instances.

        ## Examples

        ```elixir
        iex> list()
        [<id>, <id>]
        ```
        """
        @spec list :: [term]
        def list, do: Supervisor.list(__MODULE__, @partitions)

        @doc ~S"""
        Get the agent instance process pid.

        ## Examples

        ```elixir
        iex> whereis(<id>)
        <pid>
        ```
        """
        @spec whereis(any) :: pid | nil
        def whereis(id) do
          case __key_check__(id, :whereis) do
            {:ok, i} -> Supervisor.whereis(__MODULE__, @key.partition!(i, @partitions), i)
            {:redirect, target} -> target.whereis(id)
            _ -> nil
          end
        end

        @doc ~S"""
        Checks whether a given agent instance is currently active.

        ## Examples

        ```elixir
        iex> alive?(<id>)
        true
        ```
        """
        @spec alive?(any) :: boolean
        def alive?(id) do
          case __key_check__(id, :alive?) do
            {:ok, i} -> Supervisor.alive?(__MODULE__, @key.partition!(i, @partitions), i)
            {:redirect, target} -> target.alive?(id)
            _ -> false
          end
        end

        @doc ~S"""
        Forces an agent instance to sleep.

        (Shuts down the process.)

        ## Examples

        ```elixir
        iex> sleep(<id>)
        :ok
        ```

        Optionally allows to pass a reason for sleep:
        ```elixir
        iex> sleep(<id>, :testing)
        :ok
        ```
        """
        @spec sleep(id :: any, reason :: atom) :: :ok
        def sleep(id, reason \\ :sleep) do
          if pid = whereis(id), do: send(pid, {:shutdown, reason})

          :ok
        end

        @doc ~S"""
        Dynamically configure specific components.

        ## Examples

        ```elixir
        iex> config(:source, [])
        {:ok, []}
        ```
        """
        @spec config(component :: atom, opts :: Keyword.t()) :: Keyword.t()
        def config(_component, opts), do: opts

        @doc false
        @spec key(term) :: :allow | {:redirect, module} | {:block, reason :: term}
        def key(key)
        def key(_), do: :allow

        @doc false
        @spec __key_check__(term, atom, term) ::
                {:ok, term} | {:redirect, module} | {:error, reason :: term}
        def __key_check__(key, action \\ nil, context \\ nil)

        def __key_check__(id, action, context) do
          require Logger

          with {:ok, k} <- @key.parse(id, []) do
            case apply(__MODULE__, :key, [k]) do
              :allow ->
                {:ok, k}

              redirect = {:redirect, target} ->
                Logger.warn(
                  fn ->
                    "Agent<#{inspect(__MODULE__)}>: Received key #{inspect(k)} redirect to #{inspect(target)}"
                  end,
                  agent: __MODULE__,
                  key: k,
                  target: target,
                  action: action,
                  context: context
                )

                redirect

              {:block, reason} ->
                Logger.error(
                  fn ->
                    "Agent<#{inspect(__MODULE__)}>: Received key #{inspect(k)} that is blocked because: #{inspect(reason)}"
                  end,
                  agent: __MODULE__,
                  key: k,
                  reason: reason,
                  action: action,
                  context: context
                )

                {:error, reason}
            end
          end
        end

        defoverridable key: 1, config: 2

        @doc false
        @spec __agent__(atom) :: term
        def __agent__(:config),
          do: %{
            agent: __MODULE__,
            cache: @cache,
            source: @source,
            model: @model,
            registry: @registry,
            topic: @topic,
            key: @key
          }

        def __agent__(:topic), do: @topic
        def __agent__(:key), do: @key

        @doc false
        @spec __supervisor__(non_neg_integer) :: module
        unquote(
          Enum.reduce(0..(partitions - 1), nil, fn i, acc ->
            quote do
              unquote(acc)
              def __supervisor__(unquote(i)), do: unquote(:"#{__CALLER__.module}.Supervisor#{i}")
            end
          end)
        )

        @doc false
        @spec __registry__(non_neg_integer) :: module
        unquote(
          Enum.reduce(0..(partitions - 1), nil, fn i, acc ->
            quote do
              unquote(acc)
              def __registry__(unquote(i)), do: unquote(:"#{__CALLER__.module}.Registry#{i}")
            end
          end)
        )

        @doc false
        @spec __manager__(non_neg_integer) :: module
        unquote(
          Enum.reduce(0..(partitions - 1), nil, fn i, acc ->
            quote do
              unquote(acc)
              def __manager__(unquote(i)), do: unquote(:"#{__CALLER__.module}.Manager#{i}")
            end
          end)
        )

        @doc false
        @spec __key__(term) :: {:ok, term, non_neg_integer} | {:error, term}
        def __key__(id) do
          with {:ok, k} <- @key.parse(id, []),
               {:ok, p} <- @key.partition(k, @partitions),
               do: {:ok, k, p}
        end

        @doc false
        @spec __partition__(term) :: {:ok, non_neg_integer} | {:error, term}
        def __partition__(id), do: @key.partition(id, @partitions)
      end
    end
  end

  def child_spec(config = %{agent: agent}, opts \\ []) do
    if Keyword.get(opts, :generate_command_registry, true),
      do: Kvasir.Command.RegistryGenerator.create()

    %{
      id: agent,
      start: {Kvasir.Agent.Supervisor, :start_link, [config]},
      type: :supervisor
    }
  end
end