fix CI dependency issues and Phoenix.PubSub configuration

Author: InvalidPathException

.github/workflows/ci.yml

@@ -38,20 +38,22 @@ jobs:
         elixir-version: ${{ env.ELIXIR_VERSION }}
         otp-version: ${{ env.OTP_VERSION }}
 
-    - name: Restore dependencies cache
-      uses: actions/cache@v4
-      with:
-        path: |
-          deps
-          _build
-        key: ${{ runner.os }}-mix-otp28-${{ hashFiles('**/mix.lock') }}
-        restore-keys: ${{ runner.os }}-mix-otp28-
-    
+    # Temporarily disabled cache to ensure clean builds
+    # - name: Restore dependencies cache
+    #   uses: actions/cache@v4
+    #   with:
+    #     path: |
+    #       deps
+    #       _build
+    #     key: ${{ runner.os }}-mix-otp28-${{ hashFiles('**/mix.lock') }}
+    #     restore-keys: ${{ runner.os }}-mix-otp28-
+
     - name: Install dependencies
       run: |
         mix local.hex --force
         mix local.rebar --force
         mix deps.get
+        mix deps.compile
     
     - name: Check formatting
       run: mix format --check-formatted

apps/kafkaesque_core/lib/kafkaesque/core/application.ex

@@ -7,8 +7,8 @@ defmodule Kafkaesque.Core.Application do
   def start(_type, _args) do
     children = [
       {Registry, keys: :unique, name: Kafkaesque.TopicRegistry},
-      {DynamicSupervisor, name: Kafkaesque.TopicSupervisor, strategy: :one_for_one},
-      {Phoenix.PubSub, name: Kafkaesque.PubSub},
+      Kafkaesque.Topic.Supervisor,
+      {Phoenix.PubSub, name: Kafkaesque.PubSub, adapter: Phoenix.PubSub.PG2},
       Kafkaesque.Telemetry.Supervisor
     ]
 

apps/kafkaesque_core/test/pipeline/produce_broadway_test.exs

@@ -45,15 +45,21 @@ defmodule Kafkaesque.Pipeline.ProduceBroadwayTest do
           max_queue_size: 100
         )
 
-      assert {:ok, pid} =
-               ProduceBroadway.start_link(
-                 topic: @test_topic,
-                 partition: @test_partition,
-                 batch_size: 10,
-                 batch_timeout: 1
-               )
+      result =
+        ProduceBroadway.start_link(
+          topic: @test_topic,
+          partition: @test_partition,
+          batch_size: 10,
+          batch_timeout: 1
+        )
 
-      assert Process.alive?(pid)
+      case result do
+        {:ok, pid} ->
+          assert Process.alive?(pid)
+
+        {:error, {:already_started, pid}} ->
+          assert Process.alive?(pid)
+      end
     end
   end
 

apps/kafkaesque_core/test/support/test_case.ex

@@ -18,61 +18,11 @@ defmodule Kafkaesque.TestCase do
   end
 
   setup do
-    # Start required registries and services
-    ensure_registry_started()
-    # Only start PubSub if the application isn't already managing it
-    unless Application.started_applications() |> Enum.any?(fn {app, _, _} -> app == :kafkaesque_core end) do
-      ensure_pubsub_started()
-    end
-    ensure_telemetry_started()
-
+    # The application is started in test_helper.exs, so we don't need to start services here
+    # They should already be running
     :ok
   end
 
-  @doc """
-  Ensures the Registry is started.
-  """
-  def ensure_registry_started do
-    case Registry.start_link(keys: :unique, name: Kafkaesque.TopicRegistry) do
-      {:ok, _} -> :ok
-      {:error, {:already_started, _}} -> :ok
-    end
-  end
-
-  @doc """
-  Ensures PubSub is started.
-  """
-  def ensure_pubsub_started do
-    # Check if PubSub is already running
-    case Process.whereis(Kafkaesque.PubSub) do
-      nil ->
-        # Not running, start it
-        children = [
-          {Phoenix.PubSub, name: Kafkaesque.PubSub}
-        ]
-
-        case Supervisor.start_link(children, strategy: :one_for_one, name: KafkaesqueTestPubSub) do
-          {:ok, _} -> :ok
-          {:error, {:already_started, _}} -> :ok
-          {:error, {:shutdown, {:failed_to_start_child, Phoenix.PubSub, {:already_started, _}}}} -> :ok
-        end
-
-      _pid ->
-        # Already running
-        :ok
-    end
-  end
-
-  @doc """
-  Ensures Telemetry is started.
-  """
-  def ensure_telemetry_started do
-    case Kafkaesque.Telemetry.start_link([]) do
-      {:ok, _} -> :ok
-      {:error, {:already_started, _}} -> :ok
-    end
-  end
-
   @doc """
   Sets up a test directory and ensures cleanup on exit.
   """

apps/kafkaesque_core/test/test_helper.exs

@@ -1,3 +1,6 @@
+# Start the application to ensure all services are available
+Application.ensure_all_started(:kafkaesque_core)
+
 # Compile test support files
 Code.require_file("support/test_case.ex", __DIR__)
 

apps/kafkaesque_core/test/topic/supervisor_test.exs

@@ -13,16 +13,9 @@ defmodule Kafkaesque.Topic.SupervisorTest do
     Application.put_env(:kafkaesque_core, :data_dir, test_dir)
     Application.put_env(:kafkaesque_core, :offsets_dir, Path.join(test_dir, "offsets"))
 
-    # Start the topic supervisor if not already started
-    case Process.whereis(TopicSupervisor) do
-      nil ->
-        case TopicSupervisor.start_link([]) do
-          {:ok, pid} -> {:ok, pid}
-          {:error, {:already_started, pid}} -> {:ok, pid}
-        end
-      pid ->
-        {:ok, pid}
-    end
+    # The supervisor should already be started by the application
+    # Just verify it's running (registered as the module name)
+    assert Process.whereis(Kafkaesque.Topic.Supervisor) != nil
 
     on_exit(fn ->
       # Clean up any created topics