feat: implement Ecto.Adapter.Migration for MongoDB with DDL and connection fixes

Author: tevoinea-mdb

lib/mongo_ecto.ex

@@ -384,6 +384,8 @@ defmodule Mongo.Ecto do
   @behaviour Ecto.Adapter.Storage
   @behaviour Ecto.Adapter.Schema
   @behaviour Ecto.Adapter.Queryable
+  @behaviour Ecto.Adapter.Migration
+  @behaviour Ecto.Adapter.Transaction
 
   alias Mongo.Ecto.Connection
   alias Mongo.Ecto.Conversions
@@ -919,4 +921,40 @@ defmodule Mongo.Ecto do
     Ecto.Adapter.lookup_meta(repo)
     |> Connection.query(:drop_index, [collection, indexes], opts)
   end
+
+  ## Migration
+
+  @impl Ecto.Adapter.Migration
+  defdelegate execute_ddl(adapter_meta, command, opts), to: Mongo.Ecto.Migration
+
+  @impl Ecto.Adapter.Migration
+  defdelegate supports_ddl_transaction?(), to: Mongo.Ecto.Migration
+
+  @impl Ecto.Adapter.Migration
+  defdelegate lock_for_migrations(meta, opts, fun), to: Mongo.Ecto.Migration
+
+  ## Transaction
+
+  @impl Ecto.Adapter.Transaction
+  def transaction(adapter_meta, opts, fun) do
+    %{pid: topology_pid} = adapter_meta
+
+    Mongo.transaction(topology_pid, fn ->
+      try do
+        {:ok, fun.()}
+      catch
+        :throw, {:ecto_rollback, value} -> {:error, value}
+      end
+    end, opts)
+  end
+
+  @impl Ecto.Adapter.Transaction
+  def in_transaction?(_adapter_meta) do
+    Process.get(:session) != nil
+  end
+
+  @impl Ecto.Adapter.Transaction
+  def rollback(_adapter_meta, value) do
+    throw {:ecto_rollback, value}
+  end
 end

lib/mongo_ecto/connection.ex

@@ -31,6 +31,11 @@ defmodule Mongo.Ecto.Connection do
 
   def storage_down(opts) do
     {:ok, _apps} = Application.ensure_all_started(:mongodb_driver)
+    # Rename the `:mongo_url` key so that the driver can parse it
+    opts = Enum.map(opts, fn
+      {:mongo_url, value} -> {:url, value}
+      {key, value} -> {key, value}
+    end)
     {:ok, conn} = Mongo.start_link(opts)
 
     try do
@@ -43,6 +48,11 @@ defmodule Mongo.Ecto.Connection do
 
   def storage_status(opts) do
     {:ok, _apps} = Application.ensure_all_started(:mongodb_driver)
+    # Rename the `:mongo_url` key so that the driver can parse it
+    opts = Enum.map(opts, fn
+      {:mongo_url, value} -> {:url, value}
+      {key, value} -> {key, value}
+    end)
     {:ok, conn} = Mongo.start_link(opts)
 
     case Mongo.command(conn, ping: true) do
@@ -425,6 +435,9 @@ defmodule Mongo.Ecto.Connection do
           should be avoided if possible.
           """
         end
+
+      _ ->
+        check_constraint_errors(error)
     end
   end
 

lib/mongo_ecto/migration.ex

@@ -0,0 +1,88 @@
+defmodule Mongo.Ecto.Migration do
+  @moduledoc false
+
+  @behaviour Ecto.Adapter.Migration
+
+  @impl true
+  def supports_ddl_transaction?, do: false
+
+  # No-ops: MongoDB is schema-less; column-level DDL has no equivalent.
+  @impl true
+  def lock_for_migrations(_meta, _opts, fun), do: fun.()
+
+  @impl true
+  def execute_ddl(_meta, {:alter, %Ecto.Migration.Table{}, _changes}, _opts), do: {:ok, []}
+  def execute_ddl(_meta, {:rename, %Ecto.Migration.Table{}, %Ecto.Migration.Table{}}, _opts), do: {:ok, []}
+  def execute_ddl(_meta, {:rename, %Ecto.Migration.Table{}, _old_col, _new_col}, _opts), do: {:ok, []}
+  def execute_ddl(_meta, {:rename, %Ecto.Migration.Index{}, _new_name}, _opts), do: {:ok, []}
+  def execute_ddl(%{pid: pool}, {:create_if_not_exists, %Ecto.Migration.Table{name: name}, commands}, _opts) do
+    collection = to_string(name)
+
+    case Mongo.create(pool, collection) do
+      :ok -> :ok
+      {:error, %Mongo.Error{code: 48}} -> :ok
+      {:error, reason} -> raise reason
+    end
+
+    # Skip :binary_id and :uuid pks — they map to MongoDB's _id which is auto-indexed.
+    # Only create a unique index for non-_id primary keys (e.g. :integer version field).
+    pk_fields = for {:add, field, type, opts} <- commands,
+                    is_list(opts) and Keyword.get(opts, :primary_key, false),
+                    type not in [:binary_id, :uuid],
+                    do: to_string(field)
+
+    if pk_fields != [] do
+      key = pk_fields |> Enum.map(fn f -> {f, 1} end) |> Map.new()
+      index_def = %{key: key, unique: true, name: "#{collection}_pk"}
+      case Mongo.create_indexes(pool, collection, [index_def]) do
+        :ok -> :ok
+        {:error, reason} -> raise reason
+      end
+    end
+
+    {:ok, []}
+  end
+  def execute_ddl(_meta, {:drop_if_exists, %Ecto.Migration.Table{}, _mode}, _opts), do: {:ok, []}
+  def execute_ddl(_meta, {:create_if_not_exists, %Ecto.Migration.Index{}}, _opts), do: {:ok, []}
+  def execute_ddl(_meta, {:drop_if_exists, %Ecto.Migration.Index{}, _mode}, _opts), do: {:ok, []}
+  def execute_ddl(%{pid: pool}, {:create, %Ecto.Migration.Table{name: name}, _columns}, _opts) do
+    case Mongo.create(pool, to_string(name)) do
+      :ok -> {:ok, []}
+      {:error, %Mongo.Error{code: 48}} -> {:ok, []}
+      {:error, reason} -> raise reason
+    end
+  end
+
+  def execute_ddl(%{pid: pool}, {:drop, %Ecto.Migration.Table{name: name}, _mode}, _opts) do
+    Mongo.drop_collection(pool, to_string(name))
+    {:ok, []}
+  end
+  def execute_ddl(%{pid: pool}, {:create, %Ecto.Migration.Index{} = index}, _opts) do
+    collection = to_string(index.table)
+    key = index.columns |> Enum.map(fn col -> {to_string(col), 1} end) |> Map.new()
+    index_def = %{key: key, name: index_name(index), unique: index.unique || false, sparse: false}
+
+    case Mongo.create_indexes(pool, collection, [index_def]) do
+      :ok -> {:ok, []}
+      {:error, reason} -> raise reason
+    end
+  end
+
+  def execute_ddl(%{pid: pool}, {:drop, %Ecto.Migration.Index{} = index, _mode}, _opts) do
+    case Mongo.drop_index(pool, to_string(index.table), index_name(index)) do
+      :ok -> {:ok, []}
+      {:error, reason} -> raise reason
+    end
+  end
+
+  def execute_ddl(_meta, {:create, %Ecto.Migration.Constraint{}}, _opts), do: {:ok, []}
+  def execute_ddl(_meta, {:drop, %Ecto.Migration.Constraint{}, _mode}, _opts), do: {:ok, []}
+  def execute_ddl(_meta, string, _opts) when is_binary(string), do: {:ok, []}
+  def execute_ddl(_meta, keyword, _opts) when is_list(keyword), do: {:ok, []}
+
+  defp index_name(%Ecto.Migration.Index{name: name}) when not is_nil(name),
+    do: to_string(name)
+
+  defp index_name(%Ecto.Migration.Index{table: table, columns: columns}),
+    do: "#{to_string(table)}_#{columns |> Enum.map(&to_string/1) |> Enum.join("_")}_index"
+end

lib/mongo_ecto/normalized_query.ex

@@ -917,7 +917,15 @@ defmodule Mongo.Ecto.NormalizedQuery do
         nil
 
       [pk] ->
-        pk
+        # Only map pk to _id when the field type is :binary_id.
+        # Integer primary keys like SchemaMigration's :version are stored as
+        # regular fields so that string-source queries (which have no schema and
+        # therefore no pk information) can still find them.
+        case schema.__schema__(:type, pk) do
+          :binary_id -> pk
+          :id -> pk
+          _ -> nil
+        end
 
       keys ->
         raise ArgumentError,

lib/mongo_ecto/regex.ex

@@ -35,7 +35,7 @@ defmodule Mongo.Ecto.Regex do
 
   @behaviour Ecto.Type
 
-  defstruct BSON.Regex |> Map.from_struct() |> Enum.to_list()
+  defstruct %BSON.Regex{} |> Map.from_struct() |> Enum.to_list()
   @type t :: %__MODULE__{pattern: String.t(), options: String.t()}
 
   @doc """

mix.exs

@@ -13,23 +13,27 @@ defmodule Mongo.Ecto.Mixfile do
       dialyzer: dialyzer(),
       docs: docs(),
       package: package(),
-      preferred_cli_env: [docs: :dev],
       test_coverage: [tool: ExCoveralls]
     ]
   end
 
   # Configuration for the OTP application.
   #
   # Type `mix help compile.app` for more information.
+  def cli do
+    [preferred_envs: [docs: :dev]]
+  end
+
   def application do
     [extra_applications: [:logger]]
   end
 
   defp deps do
     [
-      {:credo, "~> 1.5.6", only: [:dev, :test], runtime: false},
+      {:credo, "~> 1.7", only: [:dev, :test], runtime: false},
       {:dialyxir, "~> 1.1.0", only: :dev, runtime: false},
       {:ecto, "~> 3.12"},
+      {:ecto_sql, "~> 3.12"},
       {:ex_doc, ">= 0.0.0", only: :dev, runtime: false},
       {:excoveralls, "~> 0.16", only: :test},
       {:mongodb_driver, "~> 1.4"},

mix.lock

@@ -1,19 +1,20 @@
 %{
-  "bunt": {:hex, :bunt, "0.2.0", "951c6e801e8b1d2cbe58ebbd3e616a869061ddadcc4863d0a2182541acae9a38", [:mix], [], "hexpm", "7af5c7e09fe1d40f76c8e4f9dd2be7cebd83909f31fee7cd0e9eadc567da8353"},
+  "bunt": {:hex, :bunt, "1.0.0", "081c2c665f086849e6d57900292b3a161727ab40431219529f13c4ddcf3e7a44", [:mix], [], "hexpm", "dc5f86aa08a5f6fa6b8096f0735c4e76d54ae5c9fa2c143e5a1fc7c1cd9bb6b5"},
   "certifi": {:hex, :certifi, "2.9.0", "6f2a475689dd47f19fb74334859d460a2dc4e3252a3324bd2111b8f0429e7e21", [:rebar3], [], "hexpm", "266da46bdb06d6c6d35fde799bcb28d36d985d424ad7c08b5bb48f5b5cdd4641"},
-  "credo": {:hex, :credo, "1.5.6", "e04cc0fdc236fefbb578e0c04bd01a471081616e741d386909e527ac146016c6", [:mix], [{:bunt, "~> 0.2.0", [hex: :bunt, repo: "hexpm", optional: false]}, {:file_system, "~> 0.2.8", [hex: :file_system, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "4b52a3e558bd64e30de62a648518a5ea2b6e3e5d2b164ef5296244753fc7eb17"},
+  "credo": {:hex, :credo, "1.7.18", "5c5596bf7aedf9c8c227f13272ac499fe8eae6237bd326f2f07dfc173786f042", [:mix], [{:bunt, "~> 0.2.1 or ~> 1.0", [hex: :bunt, repo: "hexpm", optional: false]}, {:file_system, "~> 0.2 or ~> 1.0", [hex: :file_system, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "a189d164685fd945809e862fe76a7420c4398fa288d76257662aecb909d6b3e5"},
   "db_connection": {:hex, :db_connection, "2.7.0", "b99faa9291bb09892c7da373bb82cba59aefa9b36300f6145c5f201c7adf48ec", [:mix], [{:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "dcf08f31b2701f857dfc787fbad78223d61a32204f217f15e881dd93e4bdd3ff"},
   "decimal": {:hex, :decimal, "2.1.1", "5611dca5d4b2c3dd497dec8f68751f1f1a54755e8ed2a966c2633cf885973ad6", [:mix], [], "hexpm", "53cfe5f497ed0e7771ae1a475575603d77425099ba5faef9394932b35020ffcc"},
   "dialyxir": {:hex, :dialyxir, "1.1.0", "c5aab0d6e71e5522e77beff7ba9e08f8e02bad90dfbeffae60eaf0cb47e29488", [:mix], [{:erlex, ">= 0.2.6", [hex: :erlex, repo: "hexpm", optional: false]}], "hexpm", "07ea8e49c45f15264ebe6d5b93799d4dd56a44036cf42d0ad9c960bc266c0b9a"},
   "earmark_parser": {:hex, :earmark_parser, "1.4.18", "e1b2be73eb08a49fb032a0208bf647380682374a725dfb5b9e510def8397f6f2", [:mix], [], "hexpm", "114a0e85ec3cf9e04b811009e73c206394ffecfcc313e0b346de0d557774ee97"},
   "ecto": {:hex, :ecto, "3.12.5", "4a312960ce612e17337e7cefcf9be45b95a3be6b36b6f94dfb3d8c361d631866", [:mix], [{:decimal, "~> 2.0", [hex: :decimal, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "6eb18e80bef8bb57e17f5a7f068a1719fbda384d40fc37acb8eb8aeca493b6ea"},
+  "ecto_sql": {:hex, :ecto_sql, "3.12.1", "c0d0d60e85d9ff4631f12bafa454bc392ce8b9ec83531a412c12a0d415a3a4d0", [:mix], [{:db_connection, "~> 2.4.1 or ~> 2.5", [hex: :db_connection, repo: "hexpm", optional: false]}, {:ecto, "~> 3.12", [hex: :ecto, repo: "hexpm", optional: false]}, {:myxql, "~> 0.7", [hex: :myxql, repo: "hexpm", optional: true]}, {:postgrex, "~> 0.19 or ~> 1.0", [hex: :postgrex, repo: "hexpm", optional: true]}, {:tds, "~> 2.1.1 or ~> 2.2", [hex: :tds, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.0 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "aff5b958a899762c5f09028c847569f7dfb9cc9d63bdb8133bff8a5546de6bf5"},
   "erlex": {:hex, :erlex, "0.2.6", "c7987d15e899c7a2f34f5420d2a2ea0d659682c06ac607572df55a43753aa12e", [:mix], [], "hexpm", "2ed2e25711feb44d52b17d2780eabf998452f6efda104877a3881c2f8c0c0c75"},
   "ex_doc": {:hex, :ex_doc, "0.26.0", "1922164bac0b18b02f84d6f69cab1b93bc3e870e2ad18d5dacb50a9e06b542a3", [:mix], [{:earmark_parser, "~> 1.4.0", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_elixir, "~> 0.14", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1", [hex: :makeup_erlang, repo: "hexpm", optional: false]}], "hexpm", "2775d66e494a9a48355db7867478ffd997864c61c65a47d31c4949459281c78d"},
   "excoveralls": {:hex, :excoveralls, "0.16.0", "41f4cfbf7caaa3bc2cf411db6f89c1f53afedf0f1fe8debac918be1afa19c668", [:mix], [{:hackney, "~> 1.16", [hex: :hackney, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "401205356482ab99fb44d9812cd14dd83b65de8e7ae454697f8b34ba02ecd916"},
-  "file_system": {:hex, :file_system, "0.2.10", "fb082005a9cd1711c05b5248710f8826b02d7d1784e7c3451f9c1231d4fc162d", [:mix], [], "hexpm", "41195edbfb562a593726eda3b3e8b103a309b733ad25f3d642ba49696bf715dc"},
+  "file_system": {:hex, :file_system, "1.1.1", "31864f4685b0148f25bd3fbef2b1228457c0c89024ad67f7a81a3ffbc0bbad3a", [:mix], [], "hexpm", "7a15ff97dfe526aeefb090a7a9d3d03aa907e100e262a0f8f7746b78f8f87a5d"},
   "hackney": {:hex, :hackney, "1.18.1", "f48bf88f521f2a229fc7bae88cf4f85adc9cd9bcf23b5dc8eb6a1788c662c4f6", [:rebar3], [{:certifi, "~>2.9.0", [hex: :certifi, repo: "hexpm", optional: false]}, {:idna, "~>6.1.0", [hex: :idna, repo: "hexpm", optional: false]}, {:metrics, "~>1.0.0", [hex: :metrics, repo: "hexpm", optional: false]}, {:mimerl, "~>1.1", [hex: :mimerl, repo: "hexpm", optional: false]}, {:parse_trans, "3.3.1", [hex: :parse_trans, repo: "hexpm", optional: false]}, {:ssl_verify_fun, "~>1.1.0", [hex: :ssl_verify_fun, repo: "hexpm", optional: false]}, {:unicode_util_compat, "~>0.7.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "a4ecdaff44297e9b5894ae499e9a070ea1888c84afdd1fd9b7b2bc384950128e"},
   "idna": {:hex, :idna, "6.1.1", "8a63070e9f7d0c62eb9d9fcb360a7de382448200fbbd1b106cc96d3d8099df8d", [:rebar3], [{:unicode_util_compat, "~> 0.7.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "92376eb7894412ed19ac475e4a86f7b413c1b9fbb5bd16dccd57934157944cea"},
-  "jason": {:hex, :jason, "1.4.4", "b9226785a9aa77b6857ca22832cffa5d5011a667207eb2a0ad56adb5db443b8a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "c5eb0cab91f094599f94d55bc63409236a8ec69a21a67814529e8d5f6cc90b3b"},
+  "jason": {:hex, :jason, "1.4.5", "2e3a008590b0b8d7388c20293e9dcc9cf3e5d642fd2a114e4cbbb52e595d940a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "b0c823996102bcd0239b3c2444eb00409b72f6a140c1950bc8b457d836b30684"},
   "makeup": {:hex, :makeup, "1.0.5", "d5a830bc42c9800ce07dd97fa94669dfb93d3bf5fcf6ea7a0c67b2e0e4a7f26c", [:mix], [{:nimble_parsec, "~> 0.5 or ~> 1.0", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "cfa158c02d3f5c0c665d0af11512fed3fba0144cf1aadee0f2ce17747fba2ca9"},
   "makeup_elixir": {:hex, :makeup_elixir, "0.15.2", "dc72dfe17eb240552857465cc00cce390960d9a0c055c4ccd38b70629227e97c", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.1", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "fd23ae48d09b32eff49d4ced2b43c9f086d402ee4fd4fcb2d7fad97fa8823e75"},
   "makeup_erlang": {:hex, :makeup_erlang, "0.1.1", "3fcb7f09eb9d98dc4d208f49cc955a34218fc41ff6b84df7c75b3e6e533cc65f", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm", "174d0809e98a4ef0b3309256cbf97101c6ec01c4ab0b23e926a9e17df2077cbb"},