improvement: add unique constraints to changeset for custom unique in…

…dexes

improvement: separate out concurrent index creations and do them in a separate transaction

lib/custom_index.ex

@@ -9,12 +9,13 @@ defmodule AshPostgres.CustomIndex do
     :using,
     :prefix,
     :where,
-    :include
+    :include,
+    :message
   ]
 
   @schema [
     fields: [
-      type: {:list, :string},
+      type: {:list, {:or, [:atom, :string]}},
       doc: "The fields to include in the index."
     ],
     name: [
@@ -43,6 +44,10 @@ defmodule AshPostgres.CustomIndex do
       type: :string,
       doc: "specify conditions for a partial index."
     ],
+    message: [
+      type: :string,
+      doc: "A custom message to use for unique indexes that have been violated"
+    ],
     include: [
       type: {:list, :string},
       doc:
@@ -52,6 +57,21 @@ defmodule AshPostgres.CustomIndex do
 
   def schema, do: @schema
 
+  # sobelow_skip ["DOS.StringToAtom"]
+  def transform(%__MODULE__{fields: fields} = index) do
+    %{
+      index
+      | fields:
+          Enum.map(fields, fn field ->
+            if is_atom(field) do
+              field
+            else
+              String.to_atom(field)
+            end
+          end)
+    }
+  end
+
   def name(_resource, %{name: name}) when is_binary(name) do
     name
   end

lib/data_layer.ex

@@ -1200,6 +1200,20 @@ defmodule AshPostgres.DataLayer do
         Ecto.Changeset.unique_constraint(changeset, identity.keys, opts)
       end)
 
+    changeset =
+      resource
+      |> AshPostgres.DataLayer.Info.custom_indexes()
+      |> Enum.reduce(changeset, fn index, changeset ->
+        opts =
+          if index.message do
+            [name: index.name, message: index.message]
+          else
+            [name: index.name]
+          end
+
+        Ecto.Changeset.unique_constraint(changeset, index.fields, opts)
+      end)
+
     names =
       resource
       |> AshPostgres.DataLayer.Info.unique_index_names()

lib/migration_generator/migration_generator.ex

@@ -320,13 +320,46 @@ defmodule AshPostgres.MigrationGenerator do
           end
 
           operations
-          |> organize_operations
-          |> build_up_and_down()
-          |> write_migration!(snapshots, repo, opts, tenant?)
+          |> split_into_migrations()
+          |> Enum.each(fn operations ->
+            run_without_transaction? =
+              Enum.any?(operations, fn
+                %Operation.AddCustomIndex{index: %{concurrently: true}} ->
+                  true
+
+                _ ->
+                  false
+              end)
+
+            operations
+            |> organize_operations
+            |> build_up_and_down()
+            |> write_migration!(repo, opts, tenant?, run_without_transaction?)
+          end)
+
+          create_new_snapshot(snapshots, repo_name(repo), opts, tenant?)
       end
     end)
   end
 
+  defp split_into_migrations(operations) do
+    operations
+    |> Enum.split_with(fn
+      %Operation.AddCustomIndex{index: %{concurrently: true}} ->
+        true
+
+      _ ->
+        false
+    end)
+    |> case do
+      {[], ops} ->
+        [ops]
+
+      {concurrent_indexes, ops} ->
+        [ops, concurrent_indexes]
+    end
+  end
+
   defp add_order_to_operations({snapshot, operations}) do
     operations_with_order = Enum.map(operations, &add_order_to_operation(&1, snapshot.attributes))
 
@@ -666,9 +699,7 @@ defmodule AshPostgres.MigrationGenerator do
     repo |> Module.split() |> List.last() |> Macro.underscore()
   end
 
-  defp write_migration!({up, down}, snapshots, repo, opts, tenant?) do
-    repo_name = repo_name(repo)
-
+  defp write_migration!({up, down}, repo, opts, tenant?, run_without_transaction?) do
     migration_path = migration_path(opts, repo, tenant?)
 
     {migration_name, last_part} =
@@ -696,6 +727,14 @@ defmodule AshPostgres.MigrationGenerator do
         Module.concat([repo, Migrations, Macro.camelize(last_part)])
       end
 
+    module_attributes =
+      if run_without_transaction? do
+        """
+        @disable_ddl_transaction true
+        @disable_migration_lock true
+        """
+      end
+
     contents = """
     defmodule #{inspect(module_name)} do
       @moduledoc \"\"\"
@@ -706,6 +745,8 @@ defmodule AshPostgres.MigrationGenerator do
 
       use Ecto.Migration
 
+      #{module_attributes}
+
       def up do
         #{up}
       end
@@ -719,8 +760,6 @@ defmodule AshPostgres.MigrationGenerator do
     try do
       contents = format(contents, opts)
 
-      create_new_snapshot(snapshots, repo_name, opts, tenant?)
-
       if opts.dry_run do
         Mix.shell().info(contents)
       else
@@ -1061,6 +1100,25 @@ defmodule AshPostgres.MigrationGenerator do
     true
   end
 
+  defp after?(
+         %Operation.AddCustomIndex{
+           table: table,
+           schema: schema,
+           index: %{
+             concurrently: true
+           }
+         },
+         %Operation.AddCustomIndex{
+           table: table,
+           schema: schema,
+           index: %{
+             concurrently: false
+           }
+         }
+       ) do
+    true
+  end
+
   defp after?(
          %Operation.AddCheckConstraint{table: table, schema: schema, constraint: %{name: name}},
          %Operation.RemoveCheckConstraint{
@@ -1347,7 +1405,7 @@ defmodule AshPostgres.MigrationGenerator do
     custom_indexes_to_add =
       Enum.filter(snapshot.custom_indexes, fn index ->
         !Enum.find(old_snapshot.custom_indexes, fn old_custom_index ->
-          old_custom_index == index
+          indexes_match?(old_custom_index, index)
         end)
       end)
       |> Enum.map(fn custom_index ->
@@ -1364,7 +1422,7 @@ defmodule AshPostgres.MigrationGenerator do
       Enum.filter(old_snapshot.custom_indexes, fn old_custom_index ->
         rewrite_all_identities? ||
           !Enum.find(snapshot.custom_indexes, fn index ->
-            old_custom_index == index
+            indexes_match?(old_custom_index, index)
           end)
       end)
       |> Enum.map(fn custom_index ->
@@ -1490,6 +1548,20 @@ defmodule AshPostgres.MigrationGenerator do
     |> Enum.map(&Map.put(&1, :old_multitenancy, old_snapshot.multitenancy))
   end
 
+  defp indexes_match?(left, right) do
+    left =
+      Map.update!(left, :fields, fn fields ->
+        Enum.map(fields, &to_string/1)
+      end)
+
+    right =
+      Map.update!(right, :fields, fn fields ->
+        Enum.map(fields, &to_string/1)
+      end)
+
+    left == right
+  end
+
   defp attribute_operations(snapshot, old_snapshot, opts) do
     attributes_to_add =
       Enum.reject(snapshot.attributes, fn attribute ->

lib/migration_generator/operation.ex

@@ -775,10 +775,10 @@ defmodule AshPostgres.MigrationGenerator.Operation do
       keys =
         case multitenancy.strategy do
           :attribute ->
-            [to_string(multitenancy.attribute) | index.fields]
+            [to_string(multitenancy.attribute) | Enum.map(index.fields, &to_string/1)]
 
           _ ->
-            index.fields
+            Enum.map(index.fields, &to_string/1)
         end
 
       index =
@@ -811,10 +811,10 @@ defmodule AshPostgres.MigrationGenerator.Operation do
       keys =
         case multitenancy.strategy do
           :attribute ->
-            [to_string(multitenancy.attribute) | index.fields]
+            [to_string(multitenancy.attribute) | Enum.map(index.fields, &to_string/1)]
 
           _ ->
-            index.fields
+            Enum.map(index.fields, &to_string/1)
         end
 
       "drop_if_exists index(:#{table}, [#{Enum.map_join(keys, ", ", &inspect/1)}], #{join(["name: \"#{index_name}\"", option(:prefix, schema)])})"
@@ -832,10 +832,10 @@ defmodule AshPostgres.MigrationGenerator.Operation do
       keys =
         case multitenancy.strategy do
           :attribute ->
-            [to_string(multitenancy.attribute) | index.fields]
+            [to_string(multitenancy.attribute) | Enum.map(index.fields, &to_string/1)]
 
           _ ->
-            index.fields
+            Enum.map(index.fields, &to_string/1)
         end
 
       "drop_if_exists index(:#{table}, [#{Enum.map_join(keys, ", ", &inspect/1)}], #{join(["name: \"#{index_name}\"", option(:prefix, schema)])})"
@@ -851,10 +851,10 @@ defmodule AshPostgres.MigrationGenerator.Operation do
       keys =
         case multitenancy.strategy do
           :attribute ->
-            [to_string(multitenancy.attribute) | index.fields]
+            [to_string(multitenancy.attribute) | Enum.map(index.fields, &to_string/1)]
 
           _ ->
-            index.fields
+            Enum.map(index.fields, &to_string/1)
         end
 
       index =