Execute ddl

lib/alog/connection.ex

@@ -2,15 +2,100 @@ defmodule Alog.Connection do
   @behaviour Ecto.Adapters.SQL.Connection
 
   @impl true
-  defdelegate ddl_logs(result), to: Ecto.Adapter.Postgres
+  defdelegate child_spec(opts), to: Ecto.Adapters.Postgres.Connection
+
+  @impl true
+  defdelegate ddl_logs(result), to: Ecto.Adapters.Postgres.Connection
 
   @impl true
   defdelegate prepare_execute(connection, name, statement, params, options),
-    to: Ecto.Adapter.Postgres
+    to: Ecto.Adapters.Postgres.Connection
+
+  @impl true
+  defdelegate query(connection, statement, params, options), to: Ecto.Adapters.Postgres.Connection
+
+  @impl true
+  defdelegate stream(connection, statement, params, options),
+    to: Ecto.Adapters.Postgres.Connection
+
+  @impl true
+  def execute_ddl({c, %Ecto.Migration.Table{} = table, columns} = command)
+      when c in [:create, :create_if_not_exists] do
+    # TODO: need to determine if migration_source has been set in config
+    # else name is :schema_migrations
+    with name when name != :schema_migrations <- Map.get(table, :name),
+         true <-
+           Enum.any?(
+             columns,
+             fn
+               {:add, field, type, [primary_key: true]} -> true
+               _ -> false
+             end
+           ) do
+      raise ArgumentError, "you cannot add a primary key"
+    else
+      :schema_migrations ->
+        Ecto.Adapters.Postgres.Connection.execute_ddl({c, table, columns})
+
+      _ ->
+        Ecto.Adapters.Postgres.Connection.execute_ddl({c, table, update_columns(columns)})
+    end
+  end
+
+  def execute_ddl({:alter, %Ecto.Migration.Table{}, changes} = command) do
+    with :ok <-
+           Enum.each(
+             changes,
+             fn
+               {:remove, :cid, _, _} ->
+                 raise ArgumentError, "you cannot remove cid"
+
+               {_, _, _, [primary_key: true]} ->
+                 raise ArgumentError, "you cannot add a primary key"
+
+               _ ->
+                 nil
+             end
+           ) do
+      Ecto.Adapters.Postgres.Connection.execute_ddl(command)
+    end
+  end
+
+  def execute_ddl({c, %Ecto.Migration.Index{unique: true}})
+      when c in [:create, :create_if_not_exists] do
+    raise ArgumentError, "you cannot create a unique index"
+  end
+
+  defdelegate execute_ddl(command), to: Ecto.Adapters.Postgres.Connection
+
+  # Add required columns if they are missing
+  defp update_columns(columns) do
+    [
+      {:add, :cid, :string, [primary_key: true]},
+      {:add, :entry_id, :string, [null: false]},
+      {:add, :deleted, :boolean, [default: false]},
+      {:add, :inserted_at, :naive_datetime_usec, [null: false]},
+      {:add, :updated_at, :naive_datetime_usec, [null: false]}
+    ]
+    |> Enum.reduce(columns, fn {_, c, _, _} = col, acc ->
+      case Enum.find(acc, fn {_, a, _, _} -> a == c end) do
+        nil -> acc ++ [col]
+        _ -> acc
+      end
+    end)
+  end
+
+  # Temporary delegate functions to make tests work
+
+  @impl true
+  defdelegate all(a), to: Ecto.Adapters.Postgres.Connection
+
+  @impl true
+  defdelegate insert(a, b, c, d, e, f), to: Ecto.Adapters.Postgres.Connection
 
   @impl true
-  defdelegate query(connection, statement, params, options), to: Ecto.Adapter.Postgres
+  defdelegate execute(a, b, c, d), to: Ecto.Adapters.Postgres.Connection
 
   @impl true
-  defdelegate stream(connection, statement, params, options), to: Ecto.Adapter.Postgres
+  defdelegate delete_all(a), to: Ecto.Adapters.Postgres.Connection
 end

test/migration_test.exs

@@ -0,0 +1,179 @@
+defmodule AlogTest.MigrationTest do
+  use ExUnit.Case, async: true
+
+  alias Alog.Repo
+
+  # Avoid migration out of order warnings
+  @moduletag :capture_log
+  @base_migration 3_000_000
+
+  setup do
+    {:ok, migration_number: System.unique_integer([:positive]) + @base_migration}
+  end
+
+  defmodule AddColumnIfNotExistsMigration do
+    use Ecto.Migration
+
+    def up do
+      create(table(:add_col_if_not_exists_migration, primary_key: false))
+
+      alter table(:add_col_if_not_exists_migration) do
+        add_if_not_exists(:value, :integer)
+        add_if_not_exists(:to_be_added, :integer)
+      end
+
+      execute(
+        "INSERT INTO add_col_if_not_exists_migration (value, to_be_added, cid, entry_id, inserted_at, updated_at) VALUES (1, 2, 'a', 'a', '2019-02-10 10:04:30', '2019-02-10 10:04:30')"
+      )
+    end
+
+    def down do
+      drop(table(:add_col_if_not_exists_migration))
+    end
+  end
+
+  defmodule DropColumnIfExistsMigration do
+    use Ecto.Migration
+
+    def up do
+      create table(:drop_col_if_exists_migration, primary_key: false) do
+        add(:value, :integer)
+        add(:to_be_removed, :integer)
+      end
+
+      execute(
+        "INSERT INTO drop_col_if_exists_migration (value, to_be_removed, cid, entry_id, inserted_at, updated_at) VALUES (1, 2, 'a', 'a', '2019-02-10 10:04:30', '2019-02-10 10:04:30')"
+      )
+
+      alter table(:drop_col_if_exists_migration) do
+        remove_if_exists(:to_be_removed, :integer)
+      end
+    end
+
+    def down do
+      drop(table(:drop_col_if_exists_migration))
+    end
+  end
+
+  defmodule DuplicateTableMigration do
+    use Ecto.Migration
+
+    def change do
+      create_if_not_exists(table(:duplicate_table, primary_key: false))
+      create_if_not_exists(table(:duplicate_table, primary_key: false))
+    end
+  end
+
+  defmodule NoErrorOnConditionalColumnMigration do
+    use Ecto.Migration
+
+    def up do
+      create(table(:no_error_on_conditional_column_migration, primary_key: false))
+
+      alter table(:no_error_on_conditional_column_migration) do
+        add_if_not_exists(:value, :integer)
+        add_if_not_exists(:value, :integer)
+
+        remove_if_exists(:value, :integer)
+        remove_if_exists(:value, :integer)
+      end
+    end
+
+    def down do
+      drop(table(:no_error_on_conditional_column_migration))
+    end
+  end
+
+  defmodule DefaultMigration do
+    use Ecto.Migration
+
+    def up do
+      create table(:default_migration, primary_key: false) do
+        add(:name, :string)
+      end
+
+      execute(
+        "INSERT INTO default_migration (name, cid, entry_id, inserted_at, updated_at) VALUES ('a', 'b', 'a', '2019-02-10 10:04:30', '2019-02-10 10:04:30')"
+      )
+    end
+
+    def down do
+      drop(table(:default_migration))
+    end
+  end
+
+  defmodule ExistingDefaultMigration do
+    use Ecto.Migration
+
+    def change do
+      create table(:existing_default_migration, primary_key: false) do
+        timestamps()
+      end
+    end
+  end
+
+  import Ecto.Query, only: [from: 2]
+  import Ecto.Migrator, only: [up: 4, down: 4]
+
+  test "logs Postgres notice messages" do
+    log =
+      ExUnit.CaptureLog.capture_log(fn ->
+        num = @base_migration + System.unique_integer([:positive])
+        up(Repo, num, DuplicateTableMigration, log: false)
+      end)
+
+    assert log =~ ~s(relation "duplicate_table" already exists, skipping)
+  end
+
+  @tag :no_error_on_conditional_column_migration
+  test "add if not exists and drop if exists does not raise on failure", %{migration_number: num} do
+    assert :ok == up(Repo, num, NoErrorOnConditionalColumnMigration, log: false)
+    assert :ok == down(Repo, num, NoErrorOnConditionalColumnMigration, log: false)
+  end
+
+  @tag :add_column_if_not_exists
+  test "add column if not exists", %{migration_number: num} do
+    assert :ok == up(Repo, num, AddColumnIfNotExistsMigration, log: false)
+
+    assert [2] == Repo.all(from(p in "add_col_if_not_exists_migration", select: p.to_be_added))
+
+    :ok = down(Repo, num, AddColumnIfNotExistsMigration, log: false)
+  end
+
+  @tag :remove_column_if_exists
+  test "remove column when exists", %{migration_number: num} do
+    assert :ok == up(Repo, num, DropColumnIfExistsMigration, log: false)
+
+    assert catch_error(
+             Repo.all(from(p in "drop_col_if_exists_migration", select: p.to_be_removed))
+           )
+
+    :ok = down(Repo, num, DropColumnIfExistsMigration, log: false)
+  end
+
+  test "creates default columns", %{migration_number: num} do
+    assert :ok == up(Repo, num, DefaultMigration, log: false)
+
+    assert [%{name: _, cid: _, entry_id: _, inserted_at: _, updated_at: _, deleted: false}] =
+             Repo.all(
+               from(a in "default_migration",
+                 select: %{
+                   name: a.name,
+                   cid: a.cid,
+                   entry_id: a.entry_id,
+                   inserted_at: a.inserted_at,
+                   updated_at: a.updated_at,
+                   deleted: a.deleted
+                 }
+               )
+             )
+
+    :ok = down(Repo, num, DefaultMigration, log: false)
+  end
+
+  test "existing default columns don't throw errors", %{migration_number: num} do
+    assert :ok == up(Repo, num, ExistingDefaultMigration, log: false)
+
+    :ok = down(Repo, num, ExistingDefaultMigration, log: false)
+  end
+end