altworx_runbox

defmodule Runbox.Runtime.Simple.StateLoader do
  @moduledoc """
  Handles run state load or initialization for simple scenario runtime.
  """

  alias Runbox.Runtime.Simple.TemplateCarrier
  alias Runbox.Runtime.Simple.Timezip
  alias Runbox.Scenario.StartParams
  alias Runbox.StateStore
  require Logger

  @behaviour Runbox.Runtime.StateLoader

  @template_id TemplateCarrier.component_name()
  @output_id :output_stream
  @timezip_id Timezip.component_name()

  @impl true
  def load_existing_state(%StartParams{} = start_params, instance_context, start_recipe) do
    with {:ok, ts, entities} <- StateStore.load_run_state(instance_context.state_store_pid) do
      {:ok, assign_states(start_recipe, start_params, ts, entities, :continue, instance_context)}
    end
  end

  @impl true
  def init_new_state(%StartParams{} = start_params, instance_context, start_recipe) do
    # Template Carrier initializes its state on its own
    entity_defs = [{@template_id, nil}, {@output_id, nil}]

    {:ok, entities} =
      StateStore.init_run_state(
        instance_context.state_store_pid,
        start_params.start_from,
        entity_defs
      )

    {:ok,
     assign_states(
       start_recipe,
       start_params,
       start_params.start_from,
       entities,
       :start,
       instance_context
     )}
  end

  defp assign_states(
         start_recipe,
         %StartParams{} = start_params,
         ts,
         entities,
         start_type,
         instance_context
       ) do
    entities = Map.new(entities, &{&1.id, &1})

    # Moving state loaders to Runbox introduced this shadow dependency of knowing scenario
    # components which are defined in Altworx.
    # This applies to input_stream, tick_timezip components.
    Enum.map(
      start_recipe,
      &assign_state(&1, start_params, ts, entities, start_type, instance_context)
    )
  end

  defp assign_state(
         %{id: @template_id, args: args} = def,
         %StartParams{} = start_params,
         ts,
         entities,
         type,
         _
       ) do
    entity = Map.get(entities, @template_id)

    config =
      args.config
      |> Map.put(:scenario_id, start_params.scenario_id)
      |> Map.put(:start_from, ts)
      |> Map.put(:start_or_continue, type)

    args =
      args
      |> Map.put(:run_id, start_params.run_id)
      |> Map.put(:config, config)
      |> Map.put(:state_entity, entity)

    %{def | args: args}
  end

  defp assign_state(
         %{id: {:input_stream, _}, args: args} = def,
         %StartParams{} = start_params,
         ts,
         _,
         type,
         ctx
       ) do
    topic = Map.get(start_params.input_topics, args.config.topic)

    # where should input stream start reading data
    ts =
      cond do
        type == :continue -> ts + 1
        args.config.type == :input_topic -> ts
        args.config.type == :load_topic -> :earliest
      end

    config =
      args.config
      |> Map.put(:topic, topic)
      |> Map.put(:start_from, ts)
      |> Map.put(:brod_client, ctx.brod_client)

    args =
      args
      |> Map.put(:run_id, start_params.run_id)
      |> Map.put(:config, config)

    %{def | args: args}
  end

  defp assign_state(
         %{id: @output_id, args: args} = def,
         %StartParams{} = start_params,
         ts,
         entities,
         _,
         _
       ) do
    entity = Map.get(entities, @output_id)

    config =
      args.config
      |> Map.put(:side_effects?, start_params.side_effects?)
      |> Map.put(:output_actions_from, start_params.output_actions_from)
      |> Map.put(:notifications_from, start_params.notifications_from)
      |> Map.put(:start_from, ts)
      |> Map.put(:scenario_id, start_params.scenario_id)

    args =
      args
      |> Map.put(:run_id, start_params.run_id)
      |> Map.put(:config, config)
      |> Map.put(:state_entity, entity)

    %{def | args: args}
  end

  defp assign_state(%{id: @timezip_id, args: args} = def, %StartParams{} = start_params, _, _, _, _) do
    config = Map.put(args.config, :scenario_id, start_params.scenario_id)

    args =
      args
      |> Map.put(:run_id, start_params.run_id)
      |> Map.put(:config, config)

    %{def | args: args}
  end

  defp assign_state(
         %{id: {:tick_timezip, logical_topic}, args: args} = def,
         %StartParams{} = start_params,
         start_from,
         _,
         _,
         _
       ) do
    {time, res} =
      :timer.tc(fn ->
        topic_map = Map.get(start_params.input_topics, logical_topic)
        physical_topic = topic_map.physical_topic

        config =
          args.config
          |> Map.put(:scenario_id, start_params.scenario_id)
          |> Map.put(:topic, physical_topic)
          |> Map.put(:start_from, start_from)

        args =
          args
          |> Map.put(:run_id, start_params.run_id)
          |> Map.put(:config, config)

        %{def | args: args}
      end)

    Logger.info(
      "Run #{start_params.run_id} fetch tick_timezip #{logical_topic} state took #{div(time, 1000)}ms."
    )

    res
  end
end