defmodule Ash.Engine do
  @moduledoc """
  The Ash engine handles the parallelization/running of requests to Ash.

  Much of the complexity of this doesn't come into play for simple requests.
  The way it works is that it accepts a list of `Ash.Engine.Request` structs.
  Some of values on those structs will be instances of `Ash.Engine.Request.UnresolvedField`.
  These unresolved fields can express a dependence on the field values from other requests.
  This allows the engine to wait on executing some code until it has its required inputs,
  or if all of its dependencies are met, it can execute it immediately. The engine's job is
  to resolve its unresolved fields in the proper order, potentially in parallel.
  It also has knowledge baked in about certain special fields, like `data` which is the
  field we are ultimately trying to resolve, and `query` which is the field that drives authorization
  for read requests. Authorization is done on a *per engine request* basis.

  As the complexity of a system grows, it becomes very difficult to write code that
  is both imperative and performant. This is especially true of a framework that is
  designed to be configurable. What exactly is done, as well as the order it is done in,
  and whether or not is can be parallelized, varies wildly based on factors like how
  the resources are configured and what capabilities the data layer has. By implementing
  a generic "parallel engine", we can let the engine solve that problem. We simply
  have to express the various operations that must happen, and what other pieces of data
  they need in order to happen, and the engine handles the rest.

  There are various tradeoffs in the current design. The original version of the engine started a process
  for each request. While this had the least constrained performance characteristics of all the designs,
  it was problematic for various reasons. The primary reason being that it could deadlock without any
  reasonable way to debug said deadlock because the various states were distributed. The second version
  of the engine introduced a central `Engine` process that helped with some of these issues, but ultimately
  had the same problem. The third (and current) version of the engine is reworked instead to be drastically
  simpler, potentially at the expense of performance for some requests. Instead of starting a process per
  request, it opts to only parallelize the `data` field resolution of fields that are marked as `async?: true`,
  (unlike the previous versions which started a process for the whole request.) Although it does its best
  to prioritize starting any async tasks, it is possible that if some mix of async/sync requests are passed in
  a potentially long running sync task could prevent it from starting an async task, giving this potentially worse
  performance characteristics. In practice, this doesn't really matter because the robust data layers support running
  asynchronously, unless they are in a transaction in which case everything runs serially anyway.

  The current version of the engine can be seen as an event loop that will async some events and yield them. It also
  has support for a concurrency limit (per engine invocation, not globally, although that could now be added much more
  easily). This limit defaults to `2 * schedulers_online`.

  Check out the docs for `Ash.Engine.Request` for some more information. This is a private
  interface at the moment, though, so this documentation is just here to explain how it works
  it is not intended to give you enough information to use the engine directly.
  """

  defstruct [
    :ref,
    :id,
    :resolved_fields,
    :authorize?,
    :actor,
    :verbose?,
    :async?,
    :concurrency_limit,
    # There are no other failure modes, but this is there
    # to express the intent for there to eventually be.
    failure_mode: :complete,
    return_notifications?: false,
    opts: [],
    requests: [],
    data: %{},
    unsent_dependencies: [],
    dependencies_seen: MapSet.new(),
    dependencies: %{},
    reverse_dependencies: %{},
    resource_notifications: [],
    tasks: [],
    errors: [],
    notifications: MapSet.new(),
    pending_tasks: []
  ]

  alias Ash.Engine.Request
  require Logger

  def run(requests, opts \\ []) do
    cond do
      opts[:transaction?] && !opts[:resource] ->
        raise "Engine invoked with `transaction?: true` but no resource, so no transaction could be started."

      opts[:transaction?] && Ash.DataLayer.data_layer_can?(opts[:resource], :transact) ->
        Ash.DataLayer.transaction(
          opts[:resource],
          fn ->
            case do_run(requests, opts) do
              {:ok, result} ->
                result

              {:error, error} ->
                Ash.DataLayer.rollback(opts[:resource], error)
            end
          end,
          opts[:timeout] || :infinity
        )

      true ->
        if !Application.get_env(:ash, :disable_async?) &&
             (is_nil(opts[:resource]) ||
                Ash.DataLayer.data_layer_can?(opts[:resource], :async_engine)) && opts[:timeout] &&
             opts[:timeout] != :infinity && !Ash.DataLayer.in_transaction?(opts[:resource]) do
          task =
            async(fn ->
              do_run(requests, opts)
            end)

          try do
            Task.await(task, opts[:timeout])
          catch
            :exit, {:timeout, {Task, :await, [^task, timeout]}} ->
              {:error, Ash.Error.Invalid.Timeout.exception(timeout: timeout, name: opts[:name])}
          end
        else
          do_run(requests, opts)
        end
    end
    |> case do
      {:ok, %{resource_notifications: resource_notifications} = result} ->
        notifications =
          if opts[:return_notifications?] do
            resource_notifications
          else
            Ash.Notifier.notify(resource_notifications)
          end

        {:ok, %{result | resource_notifications: notifications}}

      {:error, :timeout} ->
        {:error, Ash.Error.Invalid.Timeout.exception(timeout: opts[:timeout], name: opts[:name])}

      other ->
        other
    end
  end

  def do_run(requests, opts \\ []) do
    state =
      %__MODULE__{
        ref: make_ref(),
        id: System.unique_integer([:positive, :monotonic]),
        resolved_fields: %{},
        actor: opts[:actor],
        return_notifications?: opts[:return_notifications?],
        concurrency_limit: System.schedulers_online() * 2,
        authorize?: opts[:authorize?] || false,
        verbose?: opts[:verbose?] || false,
        async?: !Application.get_env(:ash, :disable_async?),
        opts: opts
      }
      |> add_requests(requests)

    log(state, fn ->
      "Engine Starting - #{Enum.map_join(state.requests, ", ", & &1.name)}"
    end)

    case run_to_completion(state) do
      %__MODULE__{errors: []} = result ->
        {:ok,
         result.requests
         |> Enum.reduce(result, &add_data(&2, &1.path, &1.data))}

      state ->
        {:error, state}
    end
  end

  defp add_data(state, path, data) do
    %{
      state
      | data: put_nested_key(state.data, path, data)
    }
  end

  defp set_async(requests) do
    if Application.get_env(:ash, :disable_async?) do
      Enum.map(requests, &Map.put(&1, :async?, false))
    else
      Enum.map(requests, fn request ->
        if must_be_local?(request) do
          %{request | async?: false}
        else
          request
        end
      end)
    end
  end

  @doc false
  def must_be_local?(%{async?: false}), do: true

  def must_be_local?(request) do
    [request.resource | request.touches_resources || []]
    |> Enum.filter(& &1)
    |> Enum.any?(fn resource ->
      (Ash.DataLayer.data_layer_can?(resource, :transact) &&
         Ash.DataLayer.in_transaction?(resource)) ||
        not Ash.DataLayer.data_layer_can?(resource, :async_engine)
    end)
  end

  defp run_to_completion(state) do
    if Enum.all?(state.requests, &(&1.state in [:complete, :error])) do
      log(state, "Engine Complete")
      state
    else
      state
      |> run_iteration()
      |> case do
        ^state ->
          if state.tasks == [] && state.pending_tasks == [] do
            detect_deadlocks(state)

            raise """
            Engine Deadlock! No async tasks and state is the same after iteration.
            #{long_breakdown(state)}
            """
          else
            state
            |> run_iteration()
            |> run_to_completion()
          end

        new_state ->
          run_to_completion(new_state)
      end
    end
  end

  defp errors(state) do
    if state.errors == [] do
      ""
    else
      """
      Errors:
      #{Enum.map_join(state.errors, "\n", fn error -> Exception.format(:error, Ash.Error.to_ash_error(error), stacktrace(error)) end)}
      """
    end
  end

  defp stacktrace(%{stacktraces?: true, stacktrace: %{stacktrace: stacktrace}})
       when not is_nil(stacktrace) do
    stacktrace
  end

  defp stacktrace(_), do: nil

  defp completed_requests(state) do
    state.requests
    |> Enum.filter(&(&1.state == :complete))
    |> case do
      [] ->
        ""

      requests ->
        """
        Complete Requests:
        #{Enum.map_join(requests, "\n\n", fn request -> Request.summarize(request) end)}
        """
    end
  end

  defp errored_requests(state) do
    state.requests
    |> Enum.filter(&(&1.state == :error))
    |> case do
      [] ->
        ""

      requests ->
        """
        Errored Requests:
        #{Enum.map_join(requests, "\n\n", fn request -> Request.summarize(request) end)}
        """
    end
  end

  defp pending_requests(state) do
    state.requests
    |> Enum.reject(&(&1.state in [:complete, :error]))
    |> case do
      [] ->
        ""

      requests ->
        """
        Requests:
        #{Enum.map_join(requests, "\n\n", fn request -> Request.summarize(request) <> "\n  " <> depends_on_summary(request, state) end)}
        """
    end
  end

  defp depends_on_summary(request, state) do
    dependencies = state.dependencies[request.path] || []

    if Enum.empty?(dependencies) do
      " state: #{request.state}"
    else
      " state: #{request.state} | depends on #{Enum.map_join(dependencies, ", ", &name_of(&1, state))}"
    end
  end

  defp name_of({path, dep}, state) do
    "#{inspect(Enum.find(state.requests, &(&1.path == path)).name)}.#{dep}"
  end

  defp run_iteration(
         %__MODULE__{unsent_dependencies: [{request_path, dep} | remaining_unsent_dependencies]} =
           state
       ) do
    state = %{state | unsent_dependencies: remaining_unsent_dependencies}
    request = Enum.find(state.requests, &(&1.path == request_path))
    path = :lists.droplast(dep)
    field = List.last(dep)

    depended_on_request = Enum.find(state.requests, &(&1.path == path))

    cond do
      request.state == :error ->
        state

      depended_on_request.state == :error ->
        case Request.wont_receive(request, path, field) do
          {:stop, :dependency_failed, new_request} ->
            new_request = %{new_request | state: :error}

            state
            |> replace_request(new_request)
            |> notify_error(new_request)
        end

      true ->
        {state, notifications, dependencies} =
          notify_local_request(
            state,
            depended_on_request,
            request,
            field
          )

        state
        |> notify(notifications)
        |> store_dependencies(dependencies)
    end
  end

  defp run_iteration(state) do
    state = start_pending_tasks(state)

    state =
      state.tasks
      |> Enum.reduce(state, fn task, state ->
        case Task.yield(task, 0) do
          {:ok, {request_path, result}} ->
            state = %{state | tasks: state.tasks -- [task]}
            request = Enum.find(state.requests, &(&1.path == request_path))

            new_request = %{request | async_fetch_state: {:fetched, result}}

            replace_request(state, new_request)

          nil ->
            state
        end
      end)

    case Enum.find(state.requests, &match?({:requested, _}, &1.async_fetch_state)) do
      nil ->
        {async, sync} =
          state.requests
          |> Enum.filter(fn request ->
            Enum.empty?(state.dependencies[request.path] || [])
          end)
          |> Enum.split_with(& &1.async?)

        {state, do_sync?} =
          Enum.reduce(async, {state, true}, fn
            request, {state, false} ->
              {do_run_iteration(state, request), false}

            request, {state, true} ->
              new_state = do_run_iteration(state, request)
              # We only want to process synchronous requests once all asynchronous requests
              # have done all of their work.
              {new_state, state == new_state}
          end)

        if do_sync? do
          Enum.reduce(sync, state, &do_run_iteration(&2, &1))
        else
          state
        end

      request ->
        {_, resolver_context} = request.async_fetch_state
        request = %{request | async_fetch_state: :fetching}

        if Enum.count(state.tasks) >= state.concurrency_limit do
          pending_task = fn ->
            {request.path, request.data.resolver.(resolver_context)}
          end

          replace_request(
            %{state | pending_tasks: [pending_task | state.pending_tasks]},
            request
          )
        else
          task =
            async(fn ->
              {request.path, request.data.resolver.(resolver_context)}
            end)

          replace_request(%{state | tasks: [task | state.tasks]}, request)
        end
    end
  end

  defp async(func) do
    ash_context = Ash.get_context_for_transfer()

    Task.async(fn ->
      Ash.transfer_context(ash_context)

      func.()
    end)
  end

  defp do_run_iteration(state, request) do
    log(state, fn -> breakdown(state) end)
    {state, notifications, dependencies} = fully_advance_request(state, request)

    state
    |> notify(notifications)
    |> store_dependencies(dependencies)
  end

  defp detect_deadlocks(state) do
    state.dependencies
    |> Enum.each(fn {path, deps} ->
      case Enum.find_value(deps, &depends_on(state, &1, path)) do
        nil ->
          :ok

        {circular_dep, path} ->
          raise "Deadlock detected! #{inspect(path)} and #{circular_dep} depend on each other via #{Enum.map_join(path, " -> ", &inspect/1)}"

        circular_dep ->
          raise "Deadlock detected! #{inspect(path)} and #{circular_dep} depend on each other"
      end
    end)

    state
  end

  defp depends_on(state, source, destination, trail \\ []) do
    deps =
      state.dependencies[source]
      |> Kernel.||([])

    if Enum.any?(deps, fn {path, _} -> path == destination end) do
      source
    else
      deps
      |> Enum.reject(&(&1 in trail))
      |> Enum.find(&depends_on(state, &1, destination, [&1 | []]))
      |> case do
        nil ->
          false

        {circular_dep, path} ->
          {source, path ++ circular_dep}

        circular_dep ->
          {source, [circular_dep]}
      end
    end
  end

  defp breakdown(state) do
    """
    State breakdown:
    #{Enum.map_join(state.requests, "\n", &"#{&1.name}: #{&1.state}")}
    """
  end

  def long_breakdown(state) do
    """
    #{errors(state)}
    #{completed_requests(state)}
    #{errored_requests(state)}
    #{pending_requests(state)}
    """
  end

  defp fully_advance_request(state, request) do
    case advance_request(request, state) do
      {:ok, new_request, notifications, dependencies, resource_notification} ->
        new_state =
          state
          |> replace_request(new_request)
          |> add_resource_notification(resource_notification)

        new_dependencies = build_dependencies(new_request, dependencies)

        {new_state, notifications, new_dependencies}

      {:ok, new_request, notifications, dependencies} ->
        new_state = replace_request(state, new_request)

        new_dependencies = build_dependencies(new_request, dependencies)

        {new_state, notifications, new_dependencies}

      {:error, error, new_request} ->
        new_request = %{new_request | state: :error}

        new_state =
          state
          |> add_error(new_request.path, error)
          |> replace_request(new_request)
          |> notify_error(new_request)

        {new_state, [], []}
    end
  end

  defp notify_error(state, new_request) do
    state.reverse_dependencies[new_request.path]
    |> Kernel.||([])
    |> Enum.reduce(state, fn {request_path_that_wont_receive, dep_they_wont_receive}, state ->
      request = Enum.find(state.requests, &(&1.path == request_path_that_wont_receive))

      if request.state == :error do
        state
      else
        case Request.wont_receive(
               request,
               new_request.path,
               dep_they_wont_receive
             ) do
          {:stop, :dependency_failed, new_request} ->
            new_request = %{new_request | state: :error}

            state
            |> notify_error(new_request)
            |> replace_request(new_request)
        end
      end
    end)
  end

  defp start_pending_tasks(%{pending_tasks: []} = state), do: state

  defp start_pending_tasks(state) do
    available_tasks = state.concurrency_limit - Enum.count(state.tasks)

    {to_start, remaining} = Enum.split(state.pending_tasks, available_tasks)

    state = %{state | pending_tasks: remaining}

    new_tasks = Enum.map(to_start, &Task.async/1)

    %{state | tasks: state.tasks ++ new_tasks}
  end

  defp advance_request(%{state: state} = request, _state) when state in [:error] do
    {:ok, request, [], []}
  end

  defp advance_request(request, state) do
    case Request.next(request) do
      {:already_complete, new_request, new_notifications, new_dependencies} ->
        {:ok, new_request, new_notifications, new_dependencies}

      {:complete, new_request, new_notifications, new_dependencies} ->
        if new_request.resource && new_request.notify? do
          resource_notification =
            sanitize_notification(Request.resource_notification(new_request), state)

          {:ok, new_request, new_notifications, new_dependencies, resource_notification}
        else
          {:ok, new_request, new_notifications, new_dependencies}
        end

      {:continue, new_request, new_notifications} ->
        {:ok, new_request, new_notifications, []}

      {:error, error, new_request} ->
        {:error, error, new_request}

      {:wait, new_request, new_notifications, new_dependencies} ->
        {:ok, new_request, new_notifications, new_dependencies}
    end
  end

  defp notify(state, notifications) do
    notifications =
      notifications
      |> Enum.uniq()
      |> Enum.reject(&MapSet.member?(state.notifications, &1))

    new_state_notifications = Enum.reduce(notifications, state.notifications, &MapSet.put(&2, &1))
    state = %{state | notifications: new_state_notifications}

    notifications
    |> Request.sort_and_clean_notifications()
    |> Enum.reduce(state, fn
      {:requests, requests}, state ->
        state
        |> add_requests(requests)

      {:set_extra_data, key, value}, state ->
        %{state | data: Map.put(state.data, key, value)}

      %Ash.Notifier.Notification{} = resource_notification, state ->
        resource_notification = sanitize_notification(resource_notification, state)

        add_resource_notification(state, resource_notification)

      {receiver_path, request_path, field, value}, state ->
        receiver_request = Enum.find(state.requests, &(&1.path == receiver_path))

        {:continue, new_request} =
          Request.receive_field(receiver_request, request_path, field, value)

        state
        |> Map.update!(:dependencies, fn dependencies ->
          Map.update!(dependencies, receiver_request.path, fn deps ->
            MapSet.delete(deps, {request_path, field})
          end)
        end)
        |> Map.update!(:reverse_dependencies, fn dependencies ->
          Map.update!(dependencies, request_path, fn deps ->
            MapSet.delete(deps, {receiver_request.path, field})
          end)
        end)
        |> replace_request(new_request)
    end)
  end

  defp sanitize_notification(resource_notification, state) do
    %{
      resource_notification
      | from: self(),
        metadata:
          Map.merge(
            resource_notification.metadata || %{},
            state.opts[:notification_metadata] || %{}
          )
    }
  end

  defp store_dependencies(state, dependencies) do
    dependencies
    |> Enum.uniq()
    |> Enum.reject(fn {request_path, dep} = path_and_field ->
      MapSet.member?(state.dependencies_seen, path_and_field) ||
        dep in (state.dependencies[request_path] || [])
    end)
    |> Enum.reduce(
      state,
      fn {request_path, dep} = seen_dep, state ->
        dep_path = :lists.droplast(dep)
        dep_field = List.last(dep)

        depended_on_request = Enum.find(state.requests, &(&1.path == dep_path))

        if !depended_on_request do
          raise "Engine Error in request #{inspect(request_path)}: No request found with path #{inspect(dep_path)}. Available paths:\n #{Enum.map_join(state.requests, "\n", &inspect(&1.path))}"
        end

        # we want to send things from non async requests
        # after we've sent all info to async requests
        unsent_dependencies =
          if depended_on_request.async? do
            state.unsent_dependencies ++ [{request_path, dep}]
          else
            [{request_path, dep} | state.unsent_dependencies]
          end

        %{
          state
          | dependencies:
              state.dependencies
              |> Map.put_new_lazy(request_path, fn -> MapSet.new() end)
              |> Map.update!(request_path, &MapSet.put(&1, {dep_path, dep_field})),
            reverse_dependencies:
              state.reverse_dependencies
              |> Map.put_new_lazy(dep_path, fn -> MapSet.new() end)
              |> Map.update!(dep_path, &MapSet.put(&1, {request_path, dep_field})),
            dependencies_seen: MapSet.put(state.dependencies_seen, seen_dep),
            unsent_dependencies: unsent_dependencies
        }
      end
    )
  end

  defp notify_local_request(
         state,
         depended_on_request,
         request,
         field
       ) do
    case Request.send_field(depended_on_request, request.path, field) do
      {:ok, new_request, new_notifications} ->
        state = replace_request(state, new_request)
        {state, new_notifications, []}

      {:waiting, new_request, new_notifications, new_dependencies} ->
        new_dependencies = build_dependencies(new_request, new_dependencies)

        new_state = replace_request(state, new_request)

        {new_state, new_notifications, new_dependencies}

      {:error, error, new_request} ->
        new_state =
          state
          |> replace_request(%{new_request | state: :error})
          |> add_error(new_request.path, error)

        {new_state, [], []}
    end
  end

  defp add_error(state, path, errors) when is_list(errors) do
    Enum.reduce(errors, state, &add_error(&2, path, &1))
  end

  defp add_error(state, _path, error) do
    error = Ash.Error.to_ash_error(error)

    if error in state.errors do
      state
    else
      %{state | errors: [error | state.errors]}
    end
  end

  defp replace_request(state, request) do
    %{
      state
      | requests:
          Enum.map(state.requests, fn existing_request ->
            if existing_request.path == request.path do
              request
            else
              existing_request
            end
          end)
    }
  end

  defp add_resource_notification(state, resource_notification) do
    if Ash.DataLayer.in_transaction?(resource_notification.resource) ||
         state.return_notifications? do
      %{state | resource_notifications: [resource_notification | state.resource_notifications]}
    else
      unsent = Ash.Notifier.notify(resource_notification)

      %{state | resource_notifications: unsent ++ state.resource_notifications}
    end
  end

  def add_requests(state, requests) do
    requests =
      Enum.reject(requests, fn new_request ->
        Enum.find(state.requests, &(&1.path == new_request.path))
      end)

    {async, non_async} =
      requests
      |> Enum.map(fn request ->
        authorize? = request.authorize? and state.authorize?

        %{
          request
          | authorize?: authorize?,
            authorized?: !authorize?,
            actor: request.actor || state.actor,
            verbose?: request.verbose? || state.verbose?,
            async?: request.async? and state.async?
        }
      end)
      |> Enum.map(&Request.add_initial_authorizer_state/1)
      |> set_async()
      |> Enum.split_with(& &1.async?)

    %{state | requests: async ++ state.requests ++ non_async}
  end

  defp build_dependencies(request, dependencies) do
    Enum.map(dependencies, fn dep ->
      {request.path, dep}
    end)
  end

  defp log(state, message, level \\ :debug)

  defp log(%{verbose?: true, id: id}, message, level) when is_function(message) do
    Logger.log(level, fn -> ["#{id}: ", message.()] end)
  end

  defp log(%{verbose?: true, id: id}, message, level) do
    Logger.log(level, fn -> ["#{id}: ", message] end)
  end

  defp log(_, _, _) do
    false
  end

  def put_nested_key(state, [key], value) do
    Map.put(state, key, value)
  end

  def put_nested_key(state, [key | rest], value) do
    case Map.fetch(state, key) do
      {:ok, nested_state} when is_map(nested_state) ->
        Map.put(state, key, put_nested_key(nested_state, rest, value))

      _ ->
        Map.put(state, key, put_nested_key(%{}, rest, value))
    end
  end

  def put_nested_key(state, key, value) do
    Map.put(state, key, value)
  end
end