strom

defmodule Strom.Composite do
  @moduledoc """
  Runs a set of components and is a component itself, meaning that a composite has the same interface - it accepts flow as input and returns a modified flow.

      ## Example
      iex> alias Strom.{Composite, Transformer, Splitter, Source, Sink}
      iex> transformer = Transformer.new(:s, &(&1 + 1))
      iex> splitter = Splitter.new(:s, %{odd: &(rem(&1, 2) == 1), even: &(rem(&1, 2) == 0)})
      iex> composite = [transformer, splitter] |> Composite.new() |> Composite.start()
      iex> source = :s |> Source.new([1, 2, 3]) |> Source.start()
      iex> %{odd: odd, even: even} = %{} |> Source.call(source) |> Composite.call(composite)
      iex> {Enum.to_list(odd), Enum.to_list(even)}
      {[3], [2, 4]}

      ## Composites can be created from other composites
      iex> alias Strom.{Composite, Transformer, Splitter, Source, Sink}
      iex> transformer = Transformer.new(:s, &(&1 + 1))
      iex> splitter = Splitter.new(:s, %{odd: &(rem(&1, 2) == 1), even: &(rem(&1, 2) == 0)})
      iex> c1 = Composite.new([transformer])
      iex> c2 = Composite.new([splitter])
      iex> source = Source.new(:s, [1, 2, 3])
      iex> composite = [source, c1, c2] |> Composite.new() |> Composite.start()
      iex> %{odd: odd, even: even} = %{} |> Composite.call(composite)
      iex> {Enum.to_list(odd), Enum.to_list(even)}
      {[3], [2, 4]}
  """

  defstruct pid: nil,
            name: nil,
            components: []

  use GenServer
  alias Strom.Renamer

  @type t :: %__MODULE__{}

  @spec new([struct()]) :: __MODULE__.t()
  def new(components, name \\ nil) when is_list(components) do
    components =
      components
      |> List.flatten()
      |> Enum.flat_map(fn
        %__MODULE__{components: components} -> components
        component -> [component]
      end)

    name = if name, do: name, else: generate_name(components)

    %__MODULE__{name: name, components: components}
  end

  @spec start(__MODULE__.t()) :: __MODULE__.t()
  def start(%__MODULE__{} = composite) do
    supervisor_name = :"Supervisor_#{composite.name}"
    component_supervisor_name = :"ComponentSupervisor_#{composite.name}"
    task_supervisor_name = :"TaskSupervisor_#{composite.name}"
    registry_name = :"Registry_#{composite.name}"

    {:ok, _supervisor_pid} =
      DynamicSupervisor.start_child(
        Strom.DynamicSupervisor,
        %{
          id: supervisor_name,
          start: {DynamicSupervisor, :start_link, [[name: supervisor_name]]},
          restart: :temporary
        }
      )

    {:ok, _supervisor_pid} =
      DynamicSupervisor.start_child(
        supervisor_name,
        %{
          id: component_supervisor_name,
          start: {DynamicSupervisor, :start_link, [[name: component_supervisor_name]]},
          restart: :temporary
        }
      )

    {:ok, _task_supervisor_pid} =
      DynamicSupervisor.start_child(
        supervisor_name,
        %{
          id: task_supervisor_name,
          start: {DynamicSupervisor, :start_link, [[name: task_supervisor_name]]},
          restart: :temporary
        }
      )

    {:ok, _registry_pid} =
      DynamicSupervisor.start_child(
        supervisor_name,
        %{
          id: registry_name,
          start: {Registry, :start_link, [[keys: :unique, name: registry_name]]},
          restart: :temporary
        }
      )

    {:ok, pid} =
      DynamicSupervisor.start_child(
        Strom.DynamicSupervisor,
        %{id: __MODULE__, start: {__MODULE__, :start_link, [composite]}, restart: :temporary}
      )

    Process.link(pid)
    %{composite | pid: pid}
  end

  def component_supervisor_name(name) do
    String.to_existing_atom("ComponentSupervisor_#{name}")
  end

  def task_supervisor_name(name) do
    String.to_existing_atom("TaskSupervisor_#{name}")
  end

  def registry_name(name) do
    String.to_existing_atom("Registry_#{name}")
  end

  def start_link(%__MODULE__{name: name} = composite) do
    GenServer.start_link(__MODULE__, composite, name: name)
  end

  @impl true
  def init(%__MODULE__{} = composite) do
    {:ok, %{composite | pid: self()}, {:continue, :start_components}}
  end

  @impl true
  def handle_continue(
        :start_components,
        %__MODULE__{name: name, components: components} = composite
      ) do
    {:noreply, %{composite | components: start_components(components, name)}}
  end

  def components(%__MODULE__{name: name}) do
    GenServer.call(name, :components)
  end

  @spec call(Strom.flow(), __MODULE__.t() | atom()) :: Strom.flow()
  def call(flow, %__MODULE__{name: name}),
    do: GenServer.call(name, {:call, flow}, :infinity)

  def call(flow, name) when is_atom(name),
    do: GenServer.call(name, {:call, flow}, :infinity)

  @spec stop(__MODULE__.t()) :: :ok
  def stop(%__MODULE__{name: name}) do
    pid = Process.whereis(name)
    Process.unlink(pid)
    GenServer.call(name, :stop_components)
    GenServer.call(name, :stop)
  end

  def start_components(components, name) do
    components
    |> Enum.reduce([], fn
      %{__struct__: Renamer} = component, acc ->
        [Renamer.start(component) | acc]

      %{__struct__: module} = component, acc ->
        component = %{component | composite: {name, make_ref()}}
        component = module.start(component)
        Process.monitor(component.pid)
        [component | acc]
    end)
    |> Enum.reverse()
  end

  def delete(composite, index) do
    delete(composite, index, index)
  end

  def delete(composite, index_from, index_to) do
    GenServer.call(composite.name, {:delete, index_from, index_to})
  end

  def insert(composite, index, new_components) do
    GenServer.call(composite.name, {:insert, index, new_components})
  end

  @impl true
  def handle_call({:call, init_flow}, _from, %__MODULE__{} = composite) do
    flow = reduce_flow(composite.components, init_flow)
    {:reply, flow, composite}
  end

  def handle_call(:components, _from, %__MODULE__{components: components} = composite) do
    {:reply, components, composite}
  end

  def handle_call(:stop_components, _from, %__MODULE__{components: components} = composite) do
    stop_components(components)
    {:reply, :ok, composite}
  end

  def handle_call(:stop, _from, %__MODULE__{} = composite) do
    {:stop, :normal, :ok, composite}
  end

  def handle_call(
        {:delete, index_from, index_to},
        _from,
        %__MODULE__{components: components} = composite
      ) do
    component = Enum.at(components, index_from)
    input_streams = Strom.GenMix.state(component.pid).input_streams

    {new_components, _} =
      Enum.reduce(components, {[], 0}, fn component, {acc, index} ->
        if index >= index_from and index <= index_to do
          :ok = component.__struct__.stop(component)
          {acc, index + 1}
        else
          {[component | acc], index + 1}
        end
      end)

    next_component = Enum.at(components, index_to + 1)
    GenServer.call(next_component.pid, {:restart, component.composite, input_streams})

    {:reply, composite, %{composite | components: Enum.reverse(new_components)}}
  end

  def handle_call(
        {:insert, index, new_components},
        _from,
        %__MODULE__{components: components, name: name} = composite
      )
      when is_list(new_components) do
    component_after = Enum.at(components, index)
    gm_after = Strom.GenMix.state(component_after.pid)

    new_components = start_components(new_components, name)
    flow = reduce_flow(new_components, gm_after.input_streams)

    GenServer.call(component_after.pid, {:restart, gm_after.composite, flow})

    components =
      components
      |> List.insert_at(index, new_components)
      |> List.flatten()

    {:reply, composite, %{composite | components: components}}
  end

  @impl true
  def handle_info({:DOWN, _ref, :process, _pid, :normal}, composite) do
    # component stopped normally
    {:noreply, composite}
  end

  def handle_info(
        {:DOWN, _ref, :process, pid, _not_normal},
        %__MODULE__{components: components} = composite
      ) do
    component = Enum.find(components, fn %{pid: ^pid} -> true end)
    Enum.each(components, & &1.__struct__.stop(&1))
    {:stop, {:component_crashed, component}, composite}
  end

  defp reduce_flow(components, init_flow) do
    Enum.reduce(components, init_flow, fn %{__struct__: module} = component, flow ->
      module.call(flow, component)
    end)
  end

  defp stop_components(components) do
    Enum.each(components, fn %{__struct__: module} = component ->
      module.stop(component)
    end)
  end

  defp generate_name(components) do
    components
    |> Enum.map(fn %{__struct__: struct} -> to_string(struct) end)
    |> Enum.map_join("", &String.at(&1, 13))
    |> String.slice(0..15)
    |> then(&(&1 <> "_" <> timestamp_postfix()))
    |> String.to_atom()
  end

  defp timestamp_postfix do
    :erlang.system_time()
    |> rem(round(1.0e9))
    |> to_string()
  end
end