Merge pull request #58 from tryriot/check-publication-exists

Check publication exists

lib/walex/replication/query_builder.ex

@@ -1,4 +1,8 @@
 defmodule WalEx.Replication.QueryBuilder do
+  def publication_exists(state) do
+    "SELECT 1 FROM pg_publication WHERE pubname = '#{state.publication}' LIMIT 1;"
+  end
+
   def create_temporary_slot(state) do
     "CREATE_REPLICATION_SLOT #{state.slot_name} TEMPORARY LOGICAL pgoutput NOEXPORT_SNAPSHOT;"
   end

lib/walex/replication/server.ex

@@ -63,10 +63,21 @@ defmodule WalEx.Replication.Server do
 
   @impl true
   def handle_connect(state) do
+    query = QueryBuilder.publication_exists(state)
+    {:query, query, %{state | step: :publication_exists}}
+  end
+
+  @impl true
+  def handle_result([%Postgrex.Result{num_rows: 1}], state = %{step: :publication_exists}) do
     query = QueryBuilder.create_temporary_slot(state)
     {:query, query, %{state | step: :create_slot}}
   end
 
+  @impl true
+  def handle_result(results, %{step: :publication_exists}) do
+    raise "Publication does not exist. #{inspect(results)}"
+  end
+
   @impl true
   def handle_result([%Postgrex.Result{} | _results], state = %{step: :create_slot}) do
     query = QueryBuilder.start_replication_slot(state)

test/walex/database_test.exs

@@ -36,6 +36,16 @@ defmodule WalEx.DatabaseTest do
       assert [%{"wal_level" => "logical"}] == query(pid, "SHOW wal_level;")
     end
 
+    test "error early if publication doesn't exists" do
+      Process.flag(:trap_exit, true)
+      config = Keyword.put(@base_configs, :publication, "non_existent_publication")
+
+      {_pid, ref} =
+        spawn_monitor(fn -> WalExSupervisor.start_link(config) end)
+
+      assert_receive {:DOWN, ^ref, _, _, {:shutdown, _}}
+    end
+
     test "should start replication slot", %{database_pid: database_pid} do
       assert {:ok, replication_pid} = WalExSupervisor.start_link(@base_configs)
       assert is_pid(replication_pid)