first draft insert_node_events

lib/radiator/outline/dispatch.ex

@@ -0,0 +1,19 @@
+defmodule Radiator.Outline.Dispatch do
+  @moduledoc false
+
+  alias Radiator.Outline.Event
+  alias Radiator.Outline.EventProducer
+
+  def insert_node(attributes, user_id, event_id) do
+    "insert_node"
+    |> Event.build(attributes, user_id, event_id)
+    |> EventProducer.enqueue()
+  end
+
+  # TODO
+  # update_node
+  # delete_node
+  # move_node
+
+  # list_node different case, sync call
+end

lib/radiator/outline/event.ex

@@ -1,10 +1,11 @@
 defmodule Radiator.Outline.Event do
   @moduledoc false
 
-  def build(event_id, event_type, user_id, payload) do
-    %{
+  alias Radiator.Outline.Event.InsertNodeEvent
+
+  def build("insert_node", payload, user_id, event_id) do
+    %InsertNodeEvent{
       event_id: event_id,
-      event_type: event_type,
       user_id: user_id,
       payload: payload
     }

lib/radiator/outline/event/insert_node_event.ex

@@ -0,0 +1,3 @@
+defmodule Radiator.Outline.Event.InsertNodeEvent do
+  defstruct [:event_id, :user_id, :payload]
+end

lib/radiator/outline/event_consumer.ex

@@ -1,43 +1,46 @@
 defmodule Radiator.Outline.EventConsumer do
   use GenStage
+
+  alias Radiator.Outline
   alias Radiator.Outline.EventProducer
+  alias Radiator.Outline.Event.InsertNodeEvent
 
   def start_link(opts \\ []) do
     GenStage.start_link(__MODULE__, opts, name: __MODULE__)
   end
 
-  def init(_opts) do
-    options = []
-    {:consumer, :event_producer, subscribe_to: [{EventProducer, options}]}
+  def init(opts \\ [max_demand: 1]) do
+    {:consumer, :event_producer, subscribe_to: [{EventProducer, opts}]}
   end
 
-  def handle_events(events, _from, state) do
-    IO.inspect(events, label: "EventConsumer handle_events")
-
-    Enum.each(events, fn event ->
-      process_event(event, state)
-      IO.inspect(event, label: "EventConsumer handle_events event")
-    end)
+  def handle_events([event], _from, state) do
+    process_event(event)
 
     {:noreply, [], state}
   end
 
-  defp process_event(%InsertNodeEvent{} = event) do
-    # validate
+  defp process_event(%InsertNodeEvent{payload: payload} = _event) do
+    payload
+    |> Outline.create_node()
+    |> handle_insert_result()
+
+    #      validate
     #         true->
     #           database action: insert node()
     #           create && persist event (event contains all attributes, user, event_id, timestamps)
     #           broadcast event (topic: episode_id)
+    #           broadcast node (topic: episode_id)
     #         false->
     #           log error and return error (audit log)
   end
 
-  defp handle_result(:ok, event) do
-    persist_event(event)
-    broadcast_success(event)
+  defp handle_insert_result({:ok, node}) do
+    {:ok, node}
   end
 
-  defp handle_result(:error, event) do
-    broadcast_error(event)
+  defp handle_insert_result({:error, _error}) do
+    # log_error_please :-)
+
+    :error
   end
 end

lib/radiator/outline/event_producer.ex

@@ -8,21 +8,31 @@ defmodule Radiator.Outline.EventProducer do
   end
 
   def init(_opts) do
-    {:producer, []}
+    {:producer, {:queue.new(), 0}}
   end
 
   def enqueue(event) do
     GenStage.cast(__MODULE__, {:enqueue, event})
     :ok
   end
 
-  def handle_cast({:enqueue, event}, state) do
-    IO.inspect(state, label: "EventProducer. handle_cast")
-    {:noreply, [event], state}
+  def handle_cast({:enqueue, event}, {queue, 0}) do
+    queue = :queue.in(event, queue)
+    {:noreply, [], {queue, 0}}
   end
 
-  def handle_demand(demand, state) do
-    IO.inspect(demand, label: "EventConsumer")
-    {:noreply, [], state}
+  def handle_cast({:enqueue, event}, {queue, demand}) do
+    queue = :queue.in(event, queue)
+    {{:value, event}, queue} = :queue.out(queue)
+    {:noreply, [event], {queue, demand - 1}}
+  end
+
+  def handle_demand(_incoming, {queue, demand}) do
+    with {item, queue} <- :queue.out(queue),
+         {:value, event} <- item do
+      {:noreply, [event], {queue, demand}}
+    else
+      _ -> {:noreply, [], {queue, demand + 1}}
+    end
   end
 end

test/radiator/outline/dispatch_test.exs

@@ -0,0 +1,34 @@
+defmodule Radiator.Outline.DispatchTest do
+  use Radiator.DataCase
+
+  import Radiator.AccountsFixtures
+  import Radiator.PodcastFixtures
+
+  # alias Radiator.Outline
+  alias Radiator.Outline.Dispatch
+  alias Radiator.Outline.Node
+
+  describe "outline dispatch" do
+    setup do
+      %{episode: episode_fixture()}
+    end
+
+    test "insert_node does WHAT?", %{episode: episode} do
+      user = user_fixture()
+
+      node = %Node{episode_id: episode.id, content: "something very special 1!1"}
+      attributes = Map.from_struct(node)
+
+      event_id = Ecto.UUID.generate()
+
+      Dispatch.insert_node(attributes, user.id, event_id)
+
+      # _inserted_node =
+      #  Outline.list_nodes()
+      # |> Enum.find(&(&1.content == "something very special 1!1"))
+
+      # assert inserted_node.episode_id == node.episode_id
+      # assert inserted_node.content == node.content
+    end
+  end
+end