altworx_runbox

defmodule Runbox.Runtime.Simple.TemplateCarrier do
  @moduledoc group: :internal
  @moduledoc """
  Template carrier for Simple scenario.

  GenStage handling Simple scenario template. It carries the state of the template and handles
  `init` and `handle_message` callbacks. It expects to consume messages from input streams and
  produces output-stream-ready events.
  """
  alias Runbox.RunStartContext
  alias Runbox.Runtime.OutputAction
  alias Runbox.Scenario.Simple
  alias Runbox.Scenario.Simple.Config
  alias Runbox.StateStore.Entity
  alias Toolbox.Message
  use GenStage
  require Logger

  defmodule State do
    @moduledoc false
    defstruct [:module, :state_entity, :start_from, :runbox_ctx, :run_id, :scenario_id]
  end

  @doc "Returns component name."
  def component_name do
    :template
  end

  @doc "Starts the GenStage."
  def start_link(args, runbox_ctx, start_ctx) do
    GenStage.start_link(
      __MODULE__,
      {args, runbox_ctx, start_ctx}
    )
  end

  @doc "Returns state entity of a running template carrier."
  def get_state_entity(pid) do
    GenStage.call(pid, :fetch_state_entity)
  end

  @impl true
  def init({args, runbox_ctx, start_ctx}) do
    %{
      run_id: run_id,
      config: %{
        module: module,
        subscribe_to: [sub_component],
        scenario_id: scenario_id,
        start_from: start_from,
        start_or_continue: start_or_continue
      },
      state_entity: state_entity
    } = args

    subscribe_to = {RunStartContext.component_pid(start_ctx, sub_component), []}

    case start_or_continue do
      :start -> send(self(), :init_state)
      :continue -> send(self(), :load_state)
    end

    Logger.metadata(run_id: run_id, scenario_id: scenario_id)

    {:producer_consumer,
     %State{
       module: module,
       state_entity: state_entity,
       start_from: start_from,
       runbox_ctx: runbox_ctx,
       run_id: run_id,
       scenario_id: scenario_id
     }, subscribe_to: [subscribe_to]}
  end

  @impl true
  def handle_info(:init_state, %State{} = state) do
    case Simple.init(state.module, %Config{start_from: state.start_from}) do
      {:ok, oas, template_state} ->
        entity = Entity.update_state(state.state_entity, state.start_from, template_state)

        {:noreply, process_oas(oas, state.start_from, state, :init, template_state),
         %State{state | state_entity: entity}}

      {:error, error} ->
        log_init_error(error)
        {:stop, :init_error, state}
    end
  end

  def handle_info(:load_state, %State{} = state) do
    entity = state.state_entity

    case Simple.set_state(state.module, Entity.state(entity)) do
      {:ok, template_state} ->
        entity = Entity.update_state(entity, Entity.timestamp(entity), template_state)
        {:noreply, [], %State{state | state_entity: entity}}

      {:error, error} ->
        log_state_load_error(error, Entity.state(entity))
        {:stop, :set_state_error, state}
    end
  end

  @impl true
  def handle_events(events, _from, %State{} = state) do
    result =
      Enum.reduce_while(
        events,
        {:ok, [], state.state_entity},
        fn msg, {:ok, outputs, state_entity} ->
          entity_save_result =
            Entity.ack_processed_time(
              state_entity,
              msg.timestamp,
              {Runbox, :save_entity, [state.runbox_ctx]},
              fn entity -> transform_entity_before_save(entity, state.module) end
            )

          case entity_save_result do
            {:ok, state_entity} ->
              # outputs are either output actions or tick messages
              {new_outputs, new_state_entity} = handle_message(msg, state, state_entity)

              {:cont, {:ok, outputs ++ new_outputs, new_state_entity}}

            {:error, _} = error ->
              {:halt, error}
          end
        end
      )

    case result do
      {:ok, outputs, state_entity} ->
        {:noreply, add_tick_if_empty(outputs, events), %State{state | state_entity: state_entity}}

      _error ->
        # error was already logged at this point
        {:stop, :handle_message_error, state}
    end
  end

  # use tick to advance time
  defp handle_message(%Message{type: :tick} = msg, _state, state_entity) do
    template_state = Entity.state(state_entity)
    {[msg], Entity.update_state(state_entity, msg.timestamp, template_state)}
  end

  defp handle_message(msg, %State{} = state, state_entity) do
    template_state = Entity.state(state_entity)

    case Simple.handle_message(state.module, msg, template_state) do
      {:ok, oas, template_state} ->
        {
          process_oas(oas, msg.timestamp, state, msg, template_state),
          Entity.update_state(state_entity, msg.timestamp, template_state)
        }

      {:error, error} ->
        log_message_error(error, msg, template_state)
        {[], Entity.update_state(state_entity, msg.timestamp, template_state)}
    end
  end

  @impl true
  def handle_call(:fetch_state_entity, _from, %State{} = state) do
    {:reply, state.state_entity, [], state}
  end

  defp process_oas(oas, ts, %State{} = state, origin, origin_state) do
    Enum.flat_map(oas, fn oa ->
      if OutputAction.is_oa_body?(oa) do
        [OutputAction.new(oa, ts, state.scenario_id, state.run_id)]
      else
        log_unknown_oa(oa, origin, origin_state)
        []
      end
    end)
  end

  defp transform_entity_before_save(state_entity, module) do
    case Simple.get_state(module, Entity.state(state_entity)) do
      {:ok, template_state} ->
        {:ok, Entity.update_state(state_entity, Entity.timestamp(state_entity), template_state)}

      {:error, reason} = error ->
        log_state_save_error(reason, Entity.state(state_entity))
        error
    end
  end

  # We need to ensure output is not empty. If template processes all events but doesn't generate any
  # output action, and there are no ticks coming in, Output Sink at the end would not get any
  # messages. Therefore it would not ack the time to StateStore thus preventing StateStore from
  # storing savepoints. We must always output something, so add tick if there is no output.
  defp add_tick_if_empty([], events) do
    ts = List.last(events).timestamp
    [%Message{timestamp: ts, type: :tick, body: %{}}]
  end

  defp add_tick_if_empty(output, _events) do
    output
  end

  defp log_init_error({:bad_return_value, value}) do
    Logger.warning("""
    Simple scenario has crashed during initialization because of unexpected return value.
    Value: #{inspect(value)}\
    """)
  end

  defp log_init_error({:exception, exception, stacktrace}) do
    info = String.trim(Exception.format(:error, exception, stacktrace))

    Logger.warning("""
    Simple scenario has crashed during initialization because of exception.
    Exception: #{info}\
    """)
  end

  defp log_state_load_error({:bad_return_value, value}, state) do
    Logger.warning("""
    Simple scenario has crashed during set_state callback because of unexpected return value.
    Value: #{inspect(value)}
    External state: #{inspect(state)}\
    """)
  end

  defp log_state_load_error({:exception, exception, stacktrace}, state) do
    info = String.trim(Exception.format(:error, exception, stacktrace))

    Logger.warning("""
    Simple scenario has crashed during set_state callback because of exception.
    Exception: #{info}
    External state: #{inspect(state)}\
    """)
  end

  defp log_state_save_error({:bad_return_value, value}, state) do
    Logger.warning("""
    Simple scenario has crashed during get_state callback because of bad return value.
    Value: #{inspect(value)}
    Internal state: #{inspect(state)}\
    """)
  end

  defp log_state_save_error({:exception, exception, stacktrace}, state) do
    info = String.trim(Exception.format(:error, exception, stacktrace))

    Logger.warning("""
    Simple scenario has crashed during get_state callback because of exception.
    Exception: #{info}
    Internal state: #{inspect(state)}\
    """)
  end

  defp log_message_error({:bad_return_value, value}, msg, state) do
    Logger.warning("""
    Simple scenario has ignored message because of unexpected return value.
    Value: #{inspect(value)}
    Message: #{inspect(msg)}
    State: #{inspect(state)}\
    """)
  end

  defp log_message_error({:exception, exception, stacktrace}, msg, state) do
    info = String.trim(Exception.format(:error, exception, stacktrace))

    Logger.warning("""
    Simple scenario has ignored message because of exception.
    Exception: #{info}
    Message: #{inspect(msg)}
    State: #{inspect(state)}\
    """)
  end

  defp log_unknown_oa(oa, origin, state) do
    Logger.warning("""
    Simple scenario has ignored unknown output action.
    Output action: #{inspect(oa)}
    Origin/message: #{inspect(origin)}
    State: #{inspect(state)}\
    """)
  end
end