workflow_poc.ex

defmodule Flow do
  use GenServer

  defstruct input: nil, data: nil, steps: [], log: [], versioned_input: []

  # api

  def create(steps, input) when is_list(steps) do
    GenServer.start_link(__MODULE__, [steps, input])
  end

  def process_now(pid, input \\ nil) do
    GenServer.call(pid, {:process_now, input})
  end

  def debug(pid) do
    GenServer.call(pid, :debug)
  end

  # callbacks

  def init([steps, input]) do
    {:ok, %__MODULE__{steps: steps, input: input}}
  end

  def handle_call(
        {:process_now, input},
        _,
        %{input: old_input, versioned_input: versioned_input} = state
      ) do
    state =
      case input do
        nil -> state
        input -> %{state | input: input, versioned_input: [old_input | versioned_input]}
      end

    {res, state} = do_process(state)
    {:reply, res, state}
  end

  def handle_call(:debug, _, state) do
    {:reply, state, state}
  end

  # private

  defp do_process(%{steps: []} = state), do: {:ok, state}

  defp do_process(%{steps: [step | tail], input: input, data: data, log: log} = state) do
    case apply(step, :call, [input, data]) do
      {:ok, data} ->
        log_msg = {:ok, "#{step} success"}
        do_process(%{state | steps: tail, data: data, log: [log_msg | log]})

      {:error, msg} ->
        log_msg = {:error, "#{step} failure. Reason: #{msg}"}
        {:error, %{state | log: [log_msg | log]}}
    end
  end
end

defmodule FlowMonitor do
  use GenServer

  defstruct flows: %{}, counter: 1

  def create() do
    GenServer.start_link(__MODULE__, nil, name: FlowMonitor)
  end

  def flow_create(steps, input) do
    GenServer.call(__MODULE__, {:flow_create, steps, input})
  end

  def flow_process_now(id, input \\ nil) do
    GenServer.call(__MODULE__, {:flow_process_now, id, input})
  end

  # callback

  def init(_) do
    {:ok, %__MODULE__{}}
  end

  def handle_call({:flow_create, steps, input}, _, %{flows: flows, counter: id} = state) do
    {:ok, pid} = apply(Flow, :create, [steps, input])
    {:reply, {:ok, id, pid}, %{state | flows: Map.put(flows, id, pid), counter: id + 1}}
  end

  def handle_call({:flow_process_now, id, input}, _, %{flows: flows} = state) do
    res =
      case Map.get(flows, id) do
        nil -> {:error, :not_found}
        pid -> Flow.process_now(pid, input)
      end

    {:reply, res, state}
  end
end

defmodule Input1 do
  defstruct username: nil
end

defmodule Data1 do
  defstruct db_id: nil, service_id: nil
end

defmodule Task1 do
  def call(input, nil) do
    # initialize data for the process
    {:ok, %Data1{}}
  end
end

defmodule Task2 do
  def call(input, %Data1{} = data) do
    # do some fancyness
    {:ok, %{data | db_id: "#{input.username}_1"}}
  end
end

defmodule Task3 do
  def call(input, %Data1{} = data) do
    # OooOOOOOoooo
    {:error, "#{input.username} already exists in this fictitious service"}
    # {:ok, %{data | service_id: "#{input.username}_a"}}
  end
end

defmodule Task4 do
  def call(input, %Data1{} = data) do
    case input.username do
      "bob" ->
        {:ok, %{data | db_id: "#{input.username}_1"}}

      _ ->
        {:error, "username should be bob in order to pass"}
    end
  end
end

# FlowMonitor.create()
# {:ok, id, pid} = FlowMonitor.flow_create([Task1, Task2, Task3], %Input1{username: "jan"})
# FlowMonitor.flow_process_now(1)

# {:ok, id, pid} = FlowMonitor.flow_create([Task1, Task2, Task4], %Input1{username: "tom"})
# FlowMonitor.flow_process_now(2)
# FlowMonitor.flow_process_now(2, %Input1{username: "bob"})

# {:ok, pid} = Flow.create([Task1, Task2, Task3], %Input1{username: "jan"})
# Flow.process_now(pid)
# Flow.debug(pid)