implement storage layer and Broadway pipeline

Author: InvalidPathException

.credo.exs

@@ -30,7 +30,7 @@
           {Credo.Check.Consistency.TabsOrSpaces, []},
           {Credo.Check.Design.AliasUsage,
            [priority: :low, if_nested_deeper_than: 2, if_called_more_often_than: 0]},
-          {Credo.Check.Design.TagTODO, [exit_status: 2]},
+          {Credo.Check.Design.TagTODO, false},
           {Credo.Check.Design.TagFIXME, []},
           {Credo.Check.Readability.AliasOrder, []},
           {Credo.Check.Readability.FunctionNames, []},

apps/kafkaesque_client/mix.exs

@@ -43,4 +43,4 @@ defmodule KafkaesqueClient.MixProject do
       test: ["test"]
     ]
   end
-end
+end

apps/kafkaesque_core/lib/kafkaesque/constants.ex

@@ -0,0 +1,35 @@
+defmodule Kafkaesque.Constants do
+  @moduledoc """
+  Shared constants used throughout the Kafkaesque system.
+  """
+
+  @magic_byte <<0x4B>>
+  @frame_header_size 4
+  @index_interval 10
+  @index_bytes_threshold 4096
+  @fsync_interval_ms 1000
+  @default_batch_size 500
+  @default_batch_timeout 5
+  @default_processor_concurrency 10
+  @default_max_queue_size 10_000
+  @default_retention_hours 168
+  @max_key_size 256 * 1024
+  @max_value_size 1024 * 1024
+  @max_headers_size 32 * 1024
+  @max_batch_size 1000
+
+  def magic_byte, do: @magic_byte
+  def frame_header_size, do: @frame_header_size
+  def index_interval, do: @index_interval
+  def index_bytes_threshold, do: @index_bytes_threshold
+  def fsync_interval_ms, do: @fsync_interval_ms
+  def default_batch_size, do: @default_batch_size
+  def default_batch_timeout, do: @default_batch_timeout
+  def default_processor_concurrency, do: @default_processor_concurrency
+  def default_max_queue_size, do: @default_max_queue_size
+  def default_retention_hours, do: @default_retention_hours
+  def max_key_size, do: @max_key_size
+  def max_value_size, do: @max_value_size
+  def max_headers_size, do: @max_headers_size
+  def max_batch_size, do: @max_batch_size
+end

apps/kafkaesque_core/lib/kafkaesque/pipeline/message_buffer.ex

@@ -0,0 +1,134 @@
+defmodule Kafkaesque.Pipeline.MessageBuffer do
+  @moduledoc """
+  Message buffer that queues messages for Broadway consumption.
+  Acts as a simple queue service that Broadway's producer can pull from.
+  """
+
+  use GenServer
+  require Logger
+
+  defstruct [
+    :topic,
+    :partition,
+    :queue,
+    :max_queue_size,
+    :waiting_demand
+  ]
+
+  def start_link(opts) do
+    topic = Keyword.fetch!(opts, :topic)
+    partition = Keyword.fetch!(opts, :partition)
+
+    GenServer.start_link(__MODULE__, opts, name: via_tuple(topic, partition))
+  end
+
+  @doc """
+  Enqueue messages to be processed by Broadway.
+  Returns {:ok, :enqueued} or {:error, :queue_full}
+  """
+  def enqueue(topic, partition, messages) do
+    GenServer.call(via_tuple(topic, partition), {:enqueue, messages})
+  end
+
+  @doc """
+  Pull messages from the buffer.
+  Called by Broadway's producer when it needs messages.
+  """
+  def pull(topic, partition, demand) do
+    GenServer.call(via_tuple(topic, partition), {:pull, demand})
+  end
+
+  @doc """
+  Get current queue size.
+  """
+  def get_stats(topic, partition) do
+    GenServer.call(via_tuple(topic, partition), :get_stats)
+  end
+
+  defp via_tuple(topic, partition) do
+    {:via, Registry, {Kafkaesque.TopicRegistry, {:message_buffer, topic, partition}}}
+  end
+
+  @impl true
+  def init(opts) do
+    state = %__MODULE__{
+      topic: opts[:topic],
+      partition: opts[:partition],
+      queue: :queue.new(),
+      max_queue_size: opts[:max_queue_size] || 10_000,
+      waiting_demand: nil
+    }
+
+    Logger.info("MessageBuffer started for #{state.topic}/#{state.partition}")
+
+    {:ok, state}
+  end
+
+  @impl true
+  def handle_call({:enqueue, messages}, _from, state) do
+    current_size = :queue.len(state.queue)
+    new_size = current_size + length(messages)
+
+    if new_size > state.max_queue_size do
+      {:reply, {:error, :queue_full}, state}
+    else
+      new_queue =
+        Enum.reduce(messages, state.queue, fn msg, q ->
+          :queue.in(msg, q)
+        end)
+
+      new_state = %{state | queue: new_queue}
+
+      new_state = maybe_send_to_waiting(new_state)
+
+      {:reply, {:ok, :enqueued}, new_state}
+    end
+  end
+
+  @impl true
+  def handle_call({:pull, demand}, from, state) do
+    {messages, new_queue} = pull_from_queue(state.queue, demand, [])
+
+    if length(messages) > 0 do
+      {:reply, {:ok, messages}, %{state | queue: new_queue}}
+    else
+      {:reply, {:ok, []}, %{state | waiting_demand: {from, demand}}}
+    end
+  end
+
+  @impl true
+  def handle_call(:get_stats, _from, state) do
+    stats = %{
+      queue_size: :queue.len(state.queue),
+      max_queue_size: state.max_queue_size,
+      has_waiting_demand: state.waiting_demand != nil
+    }
+
+    {:reply, stats, state}
+  end
+
+  defp pull_from_queue(queue, 0, acc), do: {Enum.reverse(acc), queue}
+
+  defp pull_from_queue(queue, demand, acc) do
+    case :queue.out(queue) do
+      {{:value, message}, new_queue} ->
+        pull_from_queue(new_queue, demand - 1, [message | acc])
+
+      {:empty, _queue} ->
+        {Enum.reverse(acc), queue}
+    end
+  end
+
+  defp maybe_send_to_waiting(%{waiting_demand: nil} = state), do: state
+
+  defp maybe_send_to_waiting(%{waiting_demand: {from, demand}} = state) do
+    {messages, new_queue} = pull_from_queue(state.queue, demand, [])
+
+    if length(messages) > 0 do
+      GenServer.reply(from, {:ok, messages})
+      %{state | queue: new_queue, waiting_demand: nil}
+    else
+      state
+    end
+  end
+end