defmodule Actors.Actor.CallerConsumer do
  @moduledoc """
  An Elixir module representing a GenStage consumer responsible for handling
  events initiated by `CallerProducer` and interacting with actors in the system.
  """
  use GenStage
  use Retry

  require Logger
  require OpenTelemetry.Tracer, as: Tracer

  alias Actors.Actor.CallerProducer
  alias Actors.Config.PersistentTermConfig, as: Config
  alias Actors.Actor.Entity, as: ActorEntity
  alias Actors.Actor.Entity.Supervisor, as: ActorEntitySupervisor
  alias Actors.Actor.InvocationScheduler
  alias Actors.Actor.Pool, as: ActorPool

  alias Actors.Registry.{ActorRegistry, HostActor}

  alias Eigr.Functions.Protocol.Actors.{
    Actor,
    ActorId,
    Metadata,
    ActorSettings,
    ActorSystem,
    Registry
  }

  alias Eigr.Functions.Protocol.{
    InvocationRequest,
    ProxyInfo,
    RegistrationRequest,
    RegistrationResponse,
    RequestStatus,
    ServiceInfo,
    SpawnRequest,
    SpawnResponse
  }

  alias Sidecar.Measurements

  import Spawn.Utils.Common, only: [to_existing_atom_or_new: 1]

  @activate_actors_min_demand 0
  @activate_actors_max_demand 4

  @erpc_timeout 5_000

  @doc """
  Starts the consumer process and subscribes to the `CallerProducer` GenStage.
  """
  def start_link(opts \\ []) do
    id = Keyword.get(opts, :id, 1)
    GenStage.start_link(__MODULE__, opts, name: Module.concat(__MODULE__, "#{id}"))
  end

  @doc """
  Initializes the GenStage consumer.

  It subscribes to the `CallerProducer` GenStage with specified backpressure values.
  """
  @impl true
  def init(opts) do
    {min_demand, max_demand} = get_backpressure_values_allowed(opts)

    {:consumer, :ok,
     subscribe_to: [
       {CallerProducer, min_demand: min_demand, max_demand: max_demand}
     ]}
  end

  defp get_backpressure_values_allowed(opts) do
    index = Keyword.get(opts, :id, 0)
    actual_max_demand = Config.get(:actors_global_backpressure_max_demand)
    actual_min_demand = Config.get(:actors_global_backpressure_min_demand)
    backpressure_options = {actual_min_demand, actual_max_demand}

    Logger.debug(
      "Initialize Actor Event Consumer ID: #{index}. With Backpressure options: #{inspect(backpressure_options)}"
    )

    backpressure_options
  end

  @doc """
  Handles incoming events from the `CallerProducer` GenStage.

  Dispatches events to the appropriate functions for further processing.
  """
  @impl true
  def handle_events(events, _from, state) do
    if length(events) > 1,
      do: Logger.debug("Flushing the Event buffer. Buffer Size: #{inspect(length(events))}")

    Enum.each(events, &dispatch_to_actor/1)

    {:noreply, [], state}
  end

  defp dispatch_to_actor({from, {:register, event, opts}} = _producer_event) do
    reply_to_producer(from, register(event, opts))
  end

  defp dispatch_to_actor({from, {:get_state, event, _opts}} = _producer_event) do
    reply_to_producer(from, get_state(event))
  end

  defp dispatch_to_actor({from, {:spawn_actor, event, opts}} = _producer_event) do
    reply_to_producer(from, spawn_actor(event, opts))
  end

  defp dispatch_to_actor({from, {:invoke, request, opts}} = _producer_event) do
    if request.register_ref != "" and not is_nil(request.register_ref) do
      spawn_req = %SpawnRequest{
        actors: [%ActorId{request.actor.id | parent: request.register_ref}]
      }

      spawn_actor(spawn_req, opts)
    end

    reply_to_producer(from, invoke_with_span(request, opts))
  end

  defp reply_to_producer(:fake_from, response), do: response

  defp reply_to_producer(from, response) do
    GenStage.reply(from, response)
  end

  @doc """
  Registers actors in the system based on the provided registration request.

  Handles registration requests and ensures actors are properly registered.
  """
  def register(
        %RegistrationRequest{
          service_info: %ServiceInfo{} = _service_info,
          actor_system:
            %ActorSystem{name: _name, registry: %Registry{actors: actors} = _registry} =
              actor_system
        } = _registration,
        opts
      ) do
    if Sidecar.GracefulShutdown.running?() do
      actors
      |> Map.values()
      |> Enum.map(fn actor -> ActorPool.create_actor_host_pool(actor, opts) end)
      |> List.flatten()
      |> Enum.filter(&(&1.node == Node.self()))
      |> ActorRegistry.register()
      |> tap(fn _sts -> warmup_actors(actor_system, actors, opts) end)
      |> case do
        :ok ->
          status = %RequestStatus{status: :OK, message: "Accepted"}
          {:ok, %RegistrationResponse{proxy_info: get_proxy_info(), status: status}}

        _ ->
          status = %RequestStatus{
            status: :ERROR,
            message: "Failed to register one or more Actors"
          }

          {:error, %RegistrationResponse{proxy_info: get_proxy_info(), status: status}}
      end
    else
      status = %RequestStatus{
        status: :ERROR,
        message: "You can't register actors when node is stopping"
      }

      {:error, %RegistrationResponse{proxy_info: get_proxy_info(), status: status}}
    end
  end

  defp get_proxy_info() do
    %ProxyInfo{
      protocol_major_version: 1,
      protocol_minor_version: 2,
      proxy_name: "spawn",
      proxy_version: "1.0.0"
    }
  end

  @doc """
  Gets the state of the specified actor.

  This function attempts to retrieve the state of the actor identified by the given
  `ActorId`. It uses an exponential backoff strategy for retrying in case of errors
  and logs any failures.

  ## Parameters

  - `id` (%ActorId): The unique identifier of the actor.

  ## Returns

  The state of the actor if successful, otherwise an error is raised.

  ## Retry Strategy

  The function utilizes an exponential backoff strategy with randomized delays and
  a maximum expiry time of 30,000 milliseconds.

  ## Errors

  The function handles errors such as `:error`, `:exit`, `:noproc`, `:erpc`,
  `:noconnection`, and `:timeout`. It also rescues `ErlangError` exceptions and logs
  detailed error messages.

  """
  def get_state(%ActorId{name: actor_name, system: system_name} = id) do
    retry with: exponential_backoff() |> randomize |> expiry(30_000),
          atoms: [:error, :exit, :noproc, :erpc, :noconnection, :timeout],
          rescue_only: [ErlangError] do
      try do
        do_lookup_action(
          system_name,
          {false, system_name, actor_name, id},
          nil,
          fn actor_ref, _actor_ref_id ->
            ActorEntity.get_state(actor_ref)
          end
        )
      rescue
        e ->
          Logger.error("Failure to make a call to actor #{inspect(actor_name)} #{inspect(e)}")

          reraise e, __STACKTRACE__
      end
    after
      result -> result
    else
      error -> error
    end
  end

  @doc """
  Spawns an actor or a group of actors based on the provided `SpawnRequest`.

  This function is responsible for spawning actors based on the specified `SpawnRequest`.
  It retrieves the hosts associated with the provided actor IDs and registers the actors.
  Additionally, it handles cases where the system is in the process of draining or stopping.

  ## Parameters

  - `spawn` (%SpawnRequest): The request containing information about the actors to spawn.
  - `opts` (Keyword.t): Additional options for spawning the actors. Defaults to an empty keyword list.

  ## Returns

  If successful, it returns `{:ok, %SpawnResponse{status: %RequestStatus{status: :OK, message: "Accepted"}}}`.
  Otherwise, an error is raised.

  ## Actor Spawning Process

  - Retrieves actor hosts based on actor IDs from the `ActorRegistry`.
  - Filters the hosts based on the system's graceful shutdown status.
  - Registers the selected hosts in the `ActorRegistry`.
  - Returns a success response.

  ## Errors

  - Raises an `ArgumentError` if attempting to spawn an unnamed actor that has not been registered before.

  """
  def spawn_actor(spawn, opts \\ [])

  def spawn_actor(%SpawnRequest{actors: actors} = _spawn, opts) do
    hosts =
      Enum.map(actors, fn %ActorId{} = id ->
        case ActorRegistry.get_hosts_by_actor(id, parent: true) do
          {:ok, actor_hosts} ->
            to_spawn_hosts(id, actor_hosts, opts)
            |> then(fn hosts ->
              if Sidecar.GracefulShutdown.get_status() in [:draining, :stopping] do
                Enum.reject(hosts, &(&1.node == Node.self()))
              else
                hosts
              end
            end)

          error ->
            raise ArgumentError,
                  "You are trying to create an actor from an Unamed actor that has never been registered before. ActorId: #{inspect(id)}. Details. #{inspect(error)}"
        end
      end)
      |> List.flatten()
      |> Enum.filter(&(&1.node == Node.self()))

    ActorRegistry.register(hosts)

    status = %RequestStatus{status: :OK, message: "Accepted"}
    {:ok, %SpawnResponse{status: status}}
  end

  @doc """
  Invokes an actor action with distributed tracing using OpenTelemetry.

  This function performs an actor action invocation, incorporating distributed tracing
  with OpenTelemetry. It sets up the tracing context, adds relevant attributes,
  and handles asynchronous and synchronous invocations.

  ## Parameters

  - `request` (%InvocationRequest): The request containing information about the invocation.
  - `opts` (Keyword.t): Additional options for the invocation. Defaults to an empty keyword list.

  ## Returns

  A tuple containing the status and the result of the invocation.
  If the invocation is asynchronous, it returns `{:ok, :async}`.

  ## Tracing Context

  The function sets up the tracing context and adds attributes related to the invocation.
  It uses OpenTelemetry to trace the client invoke with the kind set to `:client`.

  ## Retry Mechanism

  The function incorporates a retry mechanism with backoff, randomization, and timeout
  to handle potential errors during the invocation.

  ## Error Handling

  In case of errors during the invocation, appropriate logging and tracing events are added,
  and the error is re-raised with a stack trace.

  """
  def invoke_with_span(
        %InvocationRequest{
          actor: %Actor{id: %ActorId{name: _name, system: _actor_id_system} = actor_id} = actor,
          system: %ActorSystem{} = system,
          action_name: action_name,
          async: async?,
          metadata: metadata,
          caller: caller,
          pooled: pooled?
        } = request,
        opts
      ) do
    {time, result} =
      :timer.tc(fn ->
        metadata_attributes =
          for {key, value} <- metadata,
              do: {to_existing_atom_or_new(key), value}

        metadata_attributes =
          metadata_attributes ++
            [
              {:async, async?},
              {"from", get_caller(caller)},
              {"target", actor_id.name}
            ]

        {_current, opts} =
          Keyword.get_and_update(opts, :span_ctx, fn span_ctx ->
            maybe_include_span(span_ctx)
          end)

        Tracer.with_span opts[:span_ctx], "client invoke", kind: :client do
          Tracer.set_attributes(metadata_attributes)

          # Instead of using Map.get/3, which performs a lookup twice, we use pattern matching
          timeout =
            case metadata["request-timeout"] do
              nil -> 10_000
              value -> value
            end

          retry_while with: exponential_backoff() |> randomize |> expiry(timeout) do
            try do
              Tracer.add_event("lookup", [{"target", actor.id.name}])

              actor_fqdn =
                if pooled? do
                  case ActorRegistry.get_hosts_by_actor(actor_id) do
                    {:ok, actor_hosts} ->
                      # Here the results are shuffled using Enum.shuffle/1 to introduce randomness.
                      # Then, the first shuffled result is chosen as the random choice.
                      # This approach is more efficient than choosing randomly from a complete list

                      # Shuffle the results to introduce randomness
                      shuffled_actor_hosts = Enum.shuffle(actor_hosts)

                      # Choose the first result (which is now a random result)
                      host = hd(shuffled_actor_hosts)

                      {pooled?, system.name, host.actor.id.parent, actor_id}

                    _ ->
                      fqdn =
                        {pooled?, system.name, "#{actor.id.name}-1",
                         %ActorId{actor_id | name: "#{actor.id.name}-1", parent: actor_id.name}}

                      fqdn
                  end
                else
                  {pooled?, system.name, actor_id.name, actor_id}
                end

              do_lookup_action(system.name, actor_fqdn, system, fn actor_ref, actor_ref_id ->
                %InvocationRequest{
                  actor: %Actor{} = actor
                } = request

                request_params = %InvocationRequest{
                  request
                  | actor: %Actor{actor | id: actor_ref_id}
                }

                if is_nil(request.scheduled_to) || request.scheduled_to == 0 do
                  maybe_invoke_async(async?, actor_ref, request_params, opts)
                else
                  InvocationScheduler.schedule_invoke(request_params)

                  {:ok, :async}
                end
              end)
            rescue
              e ->
                Logger.error(
                  "Failure to make a call to actor #{inspect(actor.id.name)} #{inspect(e)}"
                )

                reraise e, __STACKTRACE__
            end
            |> case do
              result = :error ->
                {:cont, result}

              result = {:error, msg} ->
                {:cont, result}

              result = {:error, :action_not_found, msg} ->
                {:halt, result}

              result ->
                {:halt, result}
            end
          end
        end
      end)

    Measurements.dispatch_invoke_duration(system.name, actor.id.name, action_name, time)
    result
  end

  defp to_spawn_hosts(id, actor_hosts, spawned_opts) do
    Enum.map(actor_hosts, fn %HostActor{
                               node: node,
                               actor: %Actor{} = unamed_actor,
                               opts: opts
                             } = _host ->
      spawned_actor = %Actor{unamed_actor | id: id}

      new_opts =
        if Keyword.has_key?(spawned_opts, :revision) do
          Keyword.put(opts, :revision, Keyword.get(spawned_opts, :revision, 0))
        else
          opts
        end

      %HostActor{node: node, actor: spawned_actor, opts: new_opts}
    end)
  end

  defp maybe_include_span(span_ctx) do
    if is_nil(span_ctx), do: {span_ctx, OpenTelemetry.Ctx.new()}, else: {span_ctx, span_ctx}
  end

  defp get_caller(nil), do: "external"
  defp get_caller(caller), do: caller.name

  @doc """
  Performs the action of looking up or creating an actor based on the given parameters.

  This function is responsible for looking up or creating an actor based on the specified actor
  fully-qualified domain name (FQDN). It incorporates distributed tracing with OpenTelemetry
  to capture relevant events and attributes during the lookup or creation process.

  ## Parameters

  - `system_name` (String): The name of the actor system.
  - `actor_fqdn` (tuple): A tuple representing the fully-qualified domain name (FQDN) of the actor.
  - `system` (%ActorSystem{}): The actor system.
  - `action_fun` (function): The function to be invoked once the actor is looked up or created.
    It receives the actor reference and actor reference ID as parameters.

  ## Tracing Context

  The function sets up a span with the name "actor-lookup" to trace the lookup or creation
  process. It adds relevant attributes, such as the actor FQDN, to the tracing context.

  ## Retry Mechanism

  The function incorporates a retry mechanism with backoff, randomization, and timeout
  to handle potential errors during the lookup or creation process.

  ## Returns

  The result of the `action_fun` function or an error tuple in case of failure.

  ## Error Handling

  In case of errors during the lookup or creation process, appropriate logging and tracing events
  are added, and the error is returned as part of the result tuple.

  """
  def do_lookup_action(
        system_name,
        {pooled, system_name, parent, %ActorId{name: actor_name} = actor_id} = actor_fqdn,
        system,
        action_fun
      ) do
    Tracer.with_span "actor-lookup" do
      Tracer.set_attributes([{:actor_fqdn, actor_fqdn}])

      case Spawn.Cluster.Node.Registry.lookup(Actors.Actor.Entity, parent) do
        [{actor_ref, actor_ref_id}] ->
          Tracer.add_event("actor-status", [{"alive", true}])
          Tracer.set_attributes([{"actor-pid", "#{inspect(actor_ref)}"}])
          Logger.debug("Lookup Actor #{actor_name}. PID: #{inspect(actor_ref)}")

          # Ensures that the name change will not affect the host function call
          if pooled do
            throw("Pooled Actors are not supported yet")
            # action_fun.(actor_ref, %ActorId{actor_ref_id | name: actor_name})
          else
            action_fun.(actor_ref, actor_ref_id)
          end

        _ ->
          Tracer.add_event("actor-status", [{"alive", false}])

          Tracer.with_span "actor-reactivation" do
            Tracer.set_attributes([{:system_name, system_name}])
            Tracer.set_attributes([{:actor_name, actor_name}])

            case ActorRegistry.lookup(actor_id,
                   filter_by_parent: pooled,
                   parent: parent
                 ) do
              {:ok, %HostActor{node: node, actor: actor, opts: opts}} ->
                do_call(
                  system,
                  node,
                  actor,
                  actor_fqdn,
                  action_fun,
                  opts
                )

              {:not_found, _} ->
                Logger.error("Actor #{actor_name} not found on ActorSystem #{system_name}")

                Tracer.add_event("reactivation-failure", [
                  {:cause, "not_found"}
                ])

                {:error, "Actor #{actor_name} not found on ActorSystem #{system_name}"}

              {:erpc, :timeout} ->
                Logger.error(
                  "Failed to invoke Actor #{actor_name} on ActorSystem #{system_name}: Node connection timeout"
                )

                Tracer.add_event("reactivation-failure", [
                  {:cause, "timeout"}
                ])

                {:error, "Node connection timeout"}

              {:error, reason} ->
                Logger.error(
                  "Failed to invoke Actor #{actor_name} on ActorSystem #{system_name}: #{inspect(reason)}"
                )

                Tracer.add_event("reactivation-failure", [
                  {:cause, "#{inspect(reason)}"}
                ])

                {:error, reason}

              _ ->
                Logger.error("Failed to invoke Actor #{actor_name} on ActorSystem #{system_name}")

                Tracer.add_event("reactivation-failure", [
                  {:cause, "unknown"}
                ])

                {:error, "Failed to invoke Actor #{actor_name} on ActorSystem #{system_name}"}
            end
          end
      end
    end
  end

  defp do_call(
         system,
         node,
         actor,
         {pooled, _system_name, _parent, actor_name} = _actor_fqdn,
         action_fun,
         opts
       ) do
    case :erpc.call(
           node,
           __MODULE__,
           :try_reactivate_actor,
           [system, actor, opts],
           @erpc_timeout
         ) do
      {:ok, actor_ref} ->
        Tracer.set_attributes([{"actor-pid", "#{inspect(actor_ref)}"}])

        Tracer.add_event("try-reactivate-actor", [
          {"reactivation-on-node", "#{inspect(node)}"}
        ])

        if pooled,
          # Ensures that the name change will not affect the host function call
          do: action_fun.(actor_ref, %ActorId{actor.id | name: actor_name.name}),
          else: action_fun.(actor_ref, actor.id)

      _ ->
        raise ErlangError
    end
  end

  defp maybe_invoke_async(true, actor_ref, request, opts) do
    ActorEntity.invoke_async(actor_ref, request, opts)
    {:ok, :async}
  end

  defp maybe_invoke_async(false, actor_ref, request, opts) do
    ActorEntity.invoke(actor_ref, request, opts)
  end

  @doc """
  Tries to reactivate an actor.

  Reactivation is attempted by looking up the actor in the registry
  or creating a new actor if not found.
  """
  @spec try_reactivate_actor(ActorSystem.t(), Actor.t(), any()) :: {:ok, any()} | {:error, any()}
  def try_reactivate_actor(system, actor, opts \\ [])

  def try_reactivate_actor(
        %ActorSystem{} = system,
        %Actor{id: %ActorId{name: name} = _id} = actor,
        opts
      ) do
    case ActorEntitySupervisor.lookup_or_create_actor(system, actor, opts) do
      {:ok, actor_ref} ->
        Logger.debug("Actor #{name} reactivated. ActorRef PID: #{inspect(actor_ref)}")
        {:ok, actor_ref}

      reason ->
        Logger.error("Failed to reactivate actor #{name}: #{inspect(reason)}")
        {:error, reason}
    end
  end

  # To lookup all actors
  def try_reactivate_actor(nil, %Actor{id: %ActorId{name: name} = _id} = actor, opts) do
    case ActorEntitySupervisor.lookup_or_create_actor(nil, actor, opts) do
      {:ok, actor_ref} ->
        Logger.debug("Actor #{name} reactivated. ActorRef PID: #{inspect(actor_ref)}")
        {:ok, actor_ref}

      reason ->
        Logger.error("Failed to reactivate actor #{name}: #{inspect(reason)}")
        {:error, reason}
    end
  end

  defp warmup_actors(actor_system, actors, opts) when is_map(actors) do
    spawn(fn ->
      actors
      |> Flow.from_enumerable(
        min_demand: @activate_actors_min_demand,
        max_demand: @activate_actors_max_demand
      )
      |> Flow.filter(&is_selectable?/1)
      |> Flow.map(fn {actor_name, actor} ->
        {time, result} =
          :timer.tc(&lookup_or_create_actor/4, [actor_system, actor_name, actor, opts])

        Logger.info(
          "Actor #{actor_name} Activated on Node #{inspect(Node.self())} in #{inspect(time)}ms"
        )

        result
      end)
      |> Flow.run()
    end)
  end

  @spec lookup_or_create_actor(ActorSystem.t(), String.t(), Actor.t(), any()) ::
          {:ok, pid()} | {:error, String.t()}
  defp lookup_or_create_actor(actor_system, actor_name, actor, opts) do
    case ActorEntitySupervisor.lookup_or_create_actor(actor_system, actor, opts) do
      {:ok, pid} ->
        {:ok, pid}

      _ ->
        Logger.debug("Failed to register Actor #{actor_name}")
        {:error, "Failed to register Actor #{actor_name}"}
    end
  end

  defp is_selectable?(
         {_actor_name,
          %Actor{
            metadata: %Metadata{channel_group: channel_group},
            settings: %ActorSettings{stateful: stateful, kind: kind}
          } = _actor}
       ) do
    cond do
      kind == :POOLED ->
        false

      match?(true, stateful) and kind != :UNAMED ->
        true

      not is_nil(channel_group) and length(channel_group) > 0 ->
        true

      true ->
        false
    end
  end

  defp is_selectable?({_actor_name, %Actor{} = _actor}),
    do: false
end