defmodule Bonny.Axn do
  @moduledoc """
  Describes a resource action event.

  This is the token passed to all steps of your operator and controller
  pipeline.

  This module gets imported to your controllers where you should use the
  functions `register_descendant/3`, `update_status/2` and the ones to register
  events: `success_event/2`, `failure_event/2` and/or `register_event/6`. Note
  that these functions raise exceptions if those resources have already been
  applied to the cluster.

  The `register_before_*` functions can be used in `Pluggable` steps in order
  to register callbacks that are called before applying resources to the
  cluster. Have a look at `Bonny.Pluggable.Logger` for a use case.

  ## Action event fields

  These fields contain information on the action event that occurred.

    * `action` - the action that triggered this event
    * `resource` - the resource the action was applied to
    * `conn` - the connection to the cluster the event occurred
    * `operator` - the operator that discovered and dispatched the event
    * `controller` - the controller handling the event and its init opts

  ## Reaction fields

    * `descendants` - descending resources defined by the handling controller
    * `status` - the data to be applied to the status subresource
    * `events` - Kubernetes events regarding the resource to be applied to the cluster


  ## Pipeline fields

    * `halted` - the boolean status on whether the pipeline was halted
    * `assigns` - shared user data as a map
    * `private` - shared library data as a map
    * `states` - The states for status, events and descendants

  """

  @derive Pluggable.Token

  alias Bonny.Event
  alias Bonny.Resource

  import Bitwise
  require Logger

  @type assigns :: %{optional(atom) => any}

  @type states :: integer()

  @type t :: %__MODULE__{
          action: :add | :modify | :reconcile | :delete,
          conn: K8s.Conn.t(),
          descendants: %{{name :: binary(), namespace :: binary()} => {integer, Resource.t()}},
          events: list(Bonny.Event.t()),
          resource: Resource.t(),
          status: map() | nil,
          assigns: assigns(),
          private: assigns(),
          halted: boolean(),
          controller: {controller :: module(), init_opts :: keyword()} | nil,
          operator: module() | nil,
          states: states()
        }

  @enforce_keys [:conn, :resource, :action]
  defstruct [
    :action,
    :conn,
    :resource,
    :controller,
    status: nil,
    assigns: %{},
    private: %{},
    descendants: %{},
    events: [],
    halted: false,
    operator: nil,
    states: 0
  ]

  @status_applied 1
  @descendants_applied 1 <<< 1
  @events_emitted 1 <<< 2

  defguard is_status_applied(axn) when (axn.states &&& @status_applied) == @status_applied

  defguard are_descendants_applied(axn)
           when (axn.states &&& @descendants_applied) == @descendants_applied

  defguard are_events_emitted(axn) when (axn.states &&& @events_emitted) == @events_emitted

  @spec new!(Keyword.t()) :: t()
  def new!(fields), do: struct!(__MODULE__, fields)

  defmodule StatusAlreadyAppliedError do
    defexception message: "the status has already been applied"

    @moduledoc """
    Error raised when trying to update or apply an already applied status
    """
  end

  defmodule DescendantsAlreadyAppliedError do
    defexception message: "the descendants have already been applied"

    @moduledoc """
    Error raised when trying to register a descendant or apply the descendants
    when already applied.
    """
  end

  defmodule EventsAlreadyEmittedError do
    defexception message: "the events have already been emitted"

    @moduledoc """
    Error raised when trying to register an event or emit evnts when already
    emitted.
    """
  end

  @doc """
  Registers a Kubernetes event to the `%Axn{}` token to be emitted by Bonny.
  """
  @spec register_event(
          t(),
          Resource.t() | nil,
          Event.event_type(),
          binary(),
          binary(),
          binary()
        ) :: t()
  def register_event(
        axn,
        related \\ nil,
        event_type,
        reason,
        action,
        message
      ) do
    event = Bonny.Event.new!(axn.resource, related, event_type, reason, action, message)
    add_event(axn, event)
  end

  @doc """
  Registers a asuccess event to the `%Axn{}` token to be emitted by Bonny.
  """
  @spec success_event(t(), Keyword.t()) :: t()
  def success_event(axn, opts \\ []) do
    action_string = axn.action |> Atom.to_string() |> String.capitalize()

    event =
      [
        reason: "Successful #{action_string}",
        message: "Resource #{axn.action} was successful."
      ]
      |> Keyword.merge(opts)
      |> Keyword.merge(
        event_type: :Normal,
        regarding: axn.resource,
        action: Atom.to_string(axn.action)
      )
      |> Event.new!()

    add_event(axn, event)
  end

  @doc """
  Registers a failure event to the `%Axn{}` token to be emitted by Bonny.
  """
  @spec failure_event(t(), Keyword.t()) :: t()
  def failure_event(axn, opts \\ []) do
    action_string = axn.action |> Atom.to_string() |> String.capitalize()

    event =
      [
        reason: "Failed #{action_string}",
        message: "Resource #{axn.action} has failed, no reason as specified."
      ]
      |> Keyword.merge(opts)
      |> Keyword.merge(
        event_type: :Warning,
        regarding: axn.resource,
        action: Atom.to_string(axn.action)
      )
      |> Event.new!()

    add_event(axn, event)
  end

  defp add_event(axn, _) when are_events_emitted(axn) do
    raise EventsAlreadyEmittedError
  end

  defp add_event(axn, event), do: %__MODULE__{axn | events: [event | axn.events]}

  @doc """
  Empties the list of events without emitting them.
  """
  @spec clear_events(t()) :: t()
  def clear_events(axn) when are_events_emitted(axn) do
    raise EventsAlreadyEmittedError
  end

  def clear_events(axn), do: %{axn | events: []}

  @doc """
  Registers a decending object to be applied.
  Owner reference will be added automatically unless disabled through
  the option `omit_owner_ref` or if the resource is cluster-scoped.

  > #### No Owner References for Cluster-scoped Resources {: .info}
  >
  > Cluster-scoped resources (resources without a namespace) will
  > automatically have owner references omitted, regardless of the
  > `omit_owner_ref` option, as they should never have owner references
  > according to Kubernetes best practices.

  If you need some resources to be applied before others, use the `group`
  option to indicate which group a resourceis added. Groups are applied in
  ascending order. Resources within a group are applied simultaneously.

  ## Options

  * `omit_owner_ref` - when `true`, adding the owner reference is omitted.
    Default: `false`. Note: This is automatically set to `true` for
    cluster-scoped resources.
  * `group` - Integer (posittive or negative). Controls the order in which
    descendants are applied to the cluster. Default: `0`

  ## Examples

  The following code registers a namespace and a pod as descendants. The
  namespace is assigned to group `-1` as it needs to exist when the pod
  is created within it. The namespace is cluster-scoped so owner references
  are automatically omitted.

  ```
  ns = %{
    "apiVersion" => "v1",
    "kind" => "Namespace",
    "metadata" => %{"name" => "my_ns"}
  }
  pod = %{
    "apiVersion" => "v1",
    "kind" => "Pod",
    "metadata" => %{"namespace" => "my_ns", "name" => "my_pod"},
    ...
  }

  axn
  |> Bonny.Axn.register_descendant(resource, ns, group: -1)
  |> Bonny.Axn.register_descendant(resource, pod)

  # Cluster-scoped resource like PersistentVolume automatically omits owner ref
  pv = %{
    "apiVersion" => "v1",
    "kind" => "PersistentVolume",
    "metadata" => %{"name" => "my-pv"},
    "spec" => %{"capacity" => %{"storage" => "1Gi"}}
  }

  axn |> Bonny.Axn.register_descendant(resource, pv)  # No owner ref added
  ```
  """
  @spec register_descendant(t(), Resource.t(), Keyword.t()) :: t()
  def register_descendant(axn, descendant, opts \\ [])

  def register_descendant(axn, _, _) when are_descendants_applied(axn) do
    raise DescendantsAlreadyAppliedError
  end

  def register_descendant(axn, descendant, opts) do
    group = Keyword.get(opts, :group, 0)

    # Automatically omit owner references for cluster-scoped resources
    should_omit_owner_ref = opts[:omit_owner_ref] || cluster_scoped_resource?(descendant)

    descendant =
      if should_omit_owner_ref,
        do: descendant,
        else: Resource.add_owner_reference(descendant, axn.resource)

    key = Bonny.Resource.gvkn(descendant)

    %__MODULE__{axn | descendants: Map.put(axn.descendants, key, {group, descendant})}
  end

  # Determines if a resource is cluster-scoped by checking if it has no namespace
  @spec cluster_scoped_resource?(Resource.t()) :: boolean()
  defp cluster_scoped_resource?(resource) do
    namespace =
      get_in(resource, ["metadata", "namespace"]) || get_in(resource, [:metadata, :namespace])

    is_nil(namespace) || namespace == ""
  end

  @doc """
  Executes `fun` for the resource status and applies the new status
  subresource. This can be called multiple times.

  `fun` should be a function of arity 1. It will be passed the
  current status object and expected to return the updated one.

  If no current status exists, an empty map is passed to `fun`
  """
  @spec update_status(t(), (map() -> map())) :: t()

  def update_status(axn, _) when is_status_applied(axn) do
    raise StatusAlreadyAppliedError
  end

  def update_status(axn, fun) do
    current_status = axn.status || axn.resource["status"] || %{}
    new_status = fun.(current_status)
    struct!(axn, status: new_status)
  end

  @doc """
  Emits the events created for this Axn.
  """

  @spec emit_events(t()) :: t()
  def emit_events(axn) when are_events_emitted(axn) do
    raise EventsAlreadyEmittedError
  end

  def emit_events(%__MODULE__{events: events, conn: conn, operator: operator} = axn) do
    events
    |> List.wrap()
    |> Enum.map(&run_before_emit_event(&1, axn))
    |> Enum.map(fn event -> {Bonny.EventRecorder.emit(event, operator, conn), event} end)
    |> Enum.each(fn
      {{:ok, _}, _} ->
        :ok

      {{:error, error}, event} ->
        id = identifier(axn)
        message = emit_event_error_message(error)

        Logger.error("#{inspect(id)} - #{message}",
          library: :bonny,
          event: event,
          error: error
        )
    end)

    mark_events_emitted(axn)
  end

  @doc """
  Applies the status to the resource's status subresource in the cluster.
  If no status was specified, :noop is returned.
  """
  @spec apply_status(t(), Keyword.t()) :: t()
  def apply_status(axn, apply_opts \\ [])

  def apply_status(axn, _) when is_status_applied(axn) do
    raise StatusAlreadyAppliedError
  end

  def apply_status(%__MODULE__{status: nil} = axn, _) do
    mark_status_applied(axn)
  end

  def apply_status(%Bonny.Axn{resource: resource} = axn, apply_opts) do
    axn = put_in(axn.resource["status"], axn.status)

    result =
      axn.resource
      |> run_before_apply_status(axn)
      |> Resource.apply_status(axn.conn, apply_opts)

    case result do
      {:ok, _} ->
        mark_status_applied(axn)

      {:error, error} ->
        id = identifier(axn)
        message = apply_status_error_message(error)

        Logger.error("#{inspect(id)} - #{message}",
          library: :bonny,
          resource: resource,
          error: error
        )

        raise "#{inspect(id)} - #{message}"
    end
  end

  defp apply_error_message(%{message: message}), do: [" ", message]

  defp apply_error_message(_), do: []

  defp apply_status_error_message(%K8s.Discovery.Error{}) do
    [
      "Failed applying resource status.",
      " ",
      "The status subresource for this resource seems to be disabled."
    ]
  end

  defp apply_status_error_message(error) do
    ["Failed applying resource status." | apply_error_message(error)]
  end

  defp apply_descendant_error_message(error, descendant) do
    gvkn = Resource.gvkn(descendant)
    ["Failed applying descending (child) resource #{inspect(gvkn)}." | apply_error_message(error)]
  end

  defp emit_event_error_message(error) do
    ["Failed emitting event." | apply_error_message(error)]
  end

  @doc """
  Applies the dependants to the cluster in groups.
  If `:create_events` is true, will create an event for each successful apply.
  Always creates events upon failed applies.

  ## Options

  `:create_events` - Whether events should be created upon success. Defaults to `true`

  All further options are passed to `K8s.Client.apply/2`
  """
  @spec apply_descendants(t(), Keyword.t()) :: t()
  def apply_descendants(axn, opts \\ [])

  def apply_descendants(axn, _) when are_descendants_applied(axn) do
    raise DescendantsAlreadyAppliedError
  end

  def apply_descendants(axn, opts) do
    axn.descendants
    |> Map.values()
    |> Enum.group_by(
      fn {group_id, _descendant} -> group_id end,
      fn {_, descendant} -> descendant end
    )
    |> Enum.sort_by(fn {group_id, _descendants} -> group_id end)
    |> Enum.map(fn {_, descendant_group} -> descendant_group end)
    |> Enum.reduce(axn, &apply_descendant_group(&2, &1, opts))
    |> mark_descendants_applied()
  end

  @spec apply_descendant_group(t(), [map()], Keyword.t()) :: t()
  defp apply_descendant_group(axn, descendants, opts) do
    {create_events, apply_opts} = Keyword.pop(opts, :create_events, [])

    descendants
    |> run_before_apply_descendants(axn)
    |> Enum.map(&Bonny.Resource.drop_managed_fields/1)
    |> Resource.apply_async(axn.conn, apply_opts)
    |> Enum.reduce(axn, fn
      {_, {:ok, descendant}}, acc ->
        if create_events do
          acc
          |> success_event(
            reason: "Successfully applied descendant",
            message:
              "Successfully applied #{K8s.Resource.FieldAccessors.kind(descendant)} #{K8s.Resource.FieldAccessors.name(descendant)} to the cluster.",
            related: descendant
          )
        else
          acc
        end

      {descendant, {:error, error}}, _acc ->
        id = identifier(axn)
        message = apply_descendant_error_message(error, descendant)

        Logger.error("#{inspect(id)} - #{message}",
          library: :bonny,
          resource: descendant,
          error: error
        )

        raise "#{inspect(id)} - #{message}"
    end)
  end

  @doc ~S"""
  Registers a callback to be invoked before a status is applied to the
  status subresource.

  Callbacks are invoked in the reverse order they are defined (callbacks
  defined first are invoked last).

  ## Examples

  To log a message for the status being applied:

      require Logger
      Bonny.Axn.register_before_apply_status(axn, fn resource, axn ->
        Logger.info("Status of the #{resource["kind"]} named #{resource["metadata"]["name"]} is applied to namespace #{resource["metadata"]["namespace"]}")
        resource
      end)
  """
  @spec register_before_apply_status(t(), (Resource.t(), t() -> Resource.t())) :: t()
  def register_before_apply_status(%__MODULE__{private: private} = axn, callback)
      when is_function(callback, 2) do
    %{axn | private: update_in(private[:before_apply_status], &[callback | &1 || []])}
  end

  @doc ~S"""
  Registers a callback to be invoked before descendants are applied to the
  cluster.

  Callbacks are invoked in the reverse order they are defined (callbacks
  defined first are invoked last).

  ## Examples

  To log a message:

      require Logger
      Bonny.Axn.register_before_apply_status(axn, fn descendants, axn ->
        Enum.each(descendants, &Logger.info("Descending #{&1["kind"]} named #{&1["name"]} is applied to namespace #{&1["metadata"]["namespace"]}"))
        descendants
      end)
  """
  @spec register_before_apply_descendants(t(), (list(Resource.t()), t() -> list(Resource.t()))) ::
          t()
  def register_before_apply_descendants(%__MODULE__{private: private} = axn, callback)
      when is_function(callback, 2) do
    %{axn | private: update_in(private[:before_apply_descendants], &[callback | &1 || []])}
  end

  @doc ~S"""
  Registers a callback to be invoked before events are emitted to the
  cluster.

  Callbacks are invoked in the reverse order they are defined (callbacks
  defined first are invoked last).

  ## Examples

  To log a message:

      require Logger
      Bonny.Axn.register_before_apply_status(axn, fn events, axn ->
        Logger.info("Event of type #{event.event_type} is emitted")
        events
      end)
  """
  @spec register_before_emit_event(t(), (Bonny.Event.t(), t() -> Bonny.Event.t())) :: t()
  def register_before_emit_event(%__MODULE__{private: private} = axn, callback)
      when is_function(callback, 2) do
    %{axn | private: update_in(private[:before_emit_event], &[callback | &1 || []])}
  end

  @doc ~S"""
  Registers a callback to be invoked at the very end of an  action event's
  processing by the operator.

  Callbacks are invoked in the reverse order they are defined (callbacks
  defined first are invoked last).

  ## Examples

  To log a message after the an event was processed by the operator:

      Bonny.Axn.register_after_processed(axn, fn axn ->
        Logger.info("done")
        axn
      end)
  """
  @spec register_after_processed(t(), (t() -> t())) :: t()
  def register_after_processed(%__MODULE__{private: private} = axn, callback)
      when is_function(callback, 1) do
    %{axn | private: update_in(private[:after_processed], &[callback | &1 || []])}
  end

  defp run_before_apply_status(resource, %__MODULE__{private: private} = axn) do
    for callback <- List.wrap(private[:before_apply_status]), reduce: resource do
      resource -> callback.(resource, axn)
    end
  end

  defp run_before_apply_descendants(descendants, %__MODULE__{private: private} = axn) do
    for callback <- List.wrap(private[:before_apply_descendants]), reduce: descendants do
      descendants -> callback.(descendants, axn)
    end
  end

  defp run_before_emit_event(event, %__MODULE__{private: private} = axn) do
    for callback <- List.wrap(private[:before_emit_event]), reduce: event do
      event -> callback.(event, axn)
    end
  end

  @doc false
  @spec run_after_processed(t()) :: any()
  def run_after_processed(%__MODULE__{private: private} = axn) do
    for callback <- List.wrap(private[:after_processed]), reduce: axn do
      axn -> callback.(axn)
    end
  end

  defp mark_status_applied(axn), do: do_mark_state(axn, @status_applied)
  defp mark_descendants_applied(axn), do: do_mark_state(axn, @descendants_applied)
  defp mark_events_emitted(axn), do: do_mark_state(axn, @events_emitted)

  defp do_mark_state(%__MODULE__{states: states} = axn, bitmask) do
    %{axn | states: states ||| bitmask}
  end

  @doc """
  Returns an identifier of an action event (resource and action) as tuple.
  Can be used in logs and similar.
  """
  @spec identifier(t()) ::
          {namespace_name :: binary(), api_version :: binary(), kind_action :: binary()}
  def identifier(%__MODULE__{action: action, resource: resource}) do
    {ns_name, api_version, others} = Bonny.Resource.gvkn(resource)
    {ns_name, api_version, "#{others}, Action=#{inspect(action)}"}
  end

  @doc """
  Sets the condition in the resource status.

  The field `.status.conditions`, if configured in the CRD, nolds a list of
  conditions, their `status` with a `message` and two timestamps. On the
  resource this could look something like this (taken from a Pod):

  ```
  kind: Pod
  status:
    conditions:
      - lastTransitionTime: "2019-10-22T16:29:24Z"
        status: "True"
        type: PodScheduled
      - lastTransitionTime: "2019-10-22T16:29:24Z"
        status: "True"
        type: Initialized
      - lastTransitionTime: "2019-10-22T16:29:31Z"
        status: "True"
        type: ContainersReady
      - lastTransitionTime: "2019-10-22T16:29:31Z"
        status: "True"
        type: Ready
  ```
  """
  @spec set_condition(
          axn :: t(),
          type :: binary(),
          status :: boolean(),
          message :: binary() | nil
        ) :: t()
  def set_condition(axn, type, status, message \\ nil) do
    condition_status = if(status, do: "True", else: "False")
    now = DateTime.utc_now()

    condition =
      %{
        "type" => type,
        "status" => condition_status,
        "message" => message,
        "lastHeartbeatTime" => now,
        "lastTransitionTime" => now
      }
      |> Map.reject(&is_nil(elem(&1, 1)))

    update_status(axn, fn status ->
      next_conditions =
        status
        |> Map.get("conditions", [])
        |> Map.new(&{&1["type"], &1})
        |> Map.update(type, condition, fn
          %{"status" => ^condition_status} = old_condition ->
            Map.put(condition, "lastTransitionTime", old_condition["lastTransitionTime"])

          _old_condition ->
            condition
        end)
        |> Map.values()

      Map.put(status, "conditions", next_conditions)
    end)
  end
end
