Ecto 3

lib/mongo_ecto/change.ex

@@ -34,6 +34,10 @@ defmodule Mongo.Ecto.ChangeMap do
   Change is not a value - it can't be loaded
   """
   def load(_), do: :error
+
+  def embed_as(_), do: :dump
+
+  def equal?(a, b), do: a == b
 end
 
 defmodule Mongo.Ecto.ChangeArray do
@@ -72,4 +76,8 @@ defmodule Mongo.Ecto.ChangeArray do
   Change is not a value - it can't be loaded
   """
   def load(_), do: :error
+
+  def embed_as(_), do: :dump
+
+  def equal?(a, b), do: a == b
 end

lib/mongo_ecto/connection.ex

@@ -1,17 +1,34 @@
 defmodule Mongo.Ecto.Connection do
   @moduledoc false
 
+  require Logger
+
   alias Mongo.Ecto.NormalizedQuery.ReadQuery
   alias Mongo.Ecto.NormalizedQuery.WriteQuery
   alias Mongo.Ecto.NormalizedQuery.CommandQuery
   alias Mongo.Ecto.NormalizedQuery.CountQuery
   alias Mongo.Ecto.NormalizedQuery.AggregateQuery
   alias Mongo.Query
 
+  def child_spec(opts) do
+    # 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)
+
+    #  opts = [name: pool_name] ++ Keyword.delete(opts, :pool) ++ pool_opts
+    Mongo.child_spec(opts)
+  end
+
   ## Worker
 
+  def init(_config) do
+  end
+
   def storage_down(opts) do
-    opts = Keyword.put(opts, :pool, DBConnection.Connection)
+    # opts = Keyword.put(opts, :pool, DBConnection.Connection)
 
     {:ok, _apps} = Application.ensure_all_started(:mongodb)
     {:ok, conn} = Mongo.start_link(opts)
@@ -24,6 +41,16 @@ defmodule Mongo.Ecto.Connection do
     end
   end
 
+  def storage_status(opts) do
+    {:ok, _apps} = Application.ensure_all_started(:mongodb)
+    {:ok, conn} = Mongo.start_link(opts)
+
+    case Mongo.command(conn, %{ping: true}) do
+      {:ok, %{"ok" => 1.0}} -> :up
+      _ -> :down
+    end
+  end
+
   ## Callbacks for adapter
 
   def read(repo, query, opts \\ [])
@@ -59,6 +86,7 @@ defmodule Mongo.Ecto.Connection do
     query = query.query
 
     %{deleted_count: n} = query(repo, :delete_many!, [coll, query], opts)
+
     n
   end
 
@@ -86,7 +114,7 @@ defmodule Mongo.Ecto.Connection do
     query = query.query
 
     case query(repo, :update_many, [coll, query, command], opts) do
-      {:ok, %Mongo.UpdateResult{modified_count: m}} ->
+      {:ok, %Mongo.UpdateResult{modified_count: m} = _result} ->
         m
 
       {:error, error} ->
@@ -144,45 +172,141 @@ defmodule Mongo.Ecto.Connection do
     query(repo, :command!, [command], opts)
   end
 
-  defp query(repo, operation, args, opts) do
-    {conn, default_opts} = repo.__pool__
-    args = [conn] ++ args ++ [with_log(repo, opts ++ default_opts)]
+  def query(adapter_meta, operation, args, opts) do
+    %{pid: pool, telemetry: telemetry, opts: default_opts} = adapter_meta
+
+    args = [pool] ++ args ++ [with_log(telemetry, args, opts ++ default_opts)]
     apply(Mongo, operation, args)
   end
 
-  defp with_log(repo, opts) do
-    case Keyword.pop(opts, :log, true) do
-      {true, opts} -> [log: &log(repo, &1, opts)] ++ opts
-      {false, opts} -> opts
-    end
+  defp with_log(telemetry, params, opts) do
+    [log: &log(telemetry, params, &1, opts)] ++ opts
   end
 
-  defp log(repo, entry, opts) do
+  defp log({repo, log, event_name}, _params, entry, opts) do
     %{
       connection_time: query_time,
       decode_time: decode_time,
       pool_time: queue_time,
+      idle_time: idle_time,
       result: result,
       query: query,
       params: params
     } = entry
 
     source = Keyword.get(opts, :source)
 
-    repo.__log__(%Ecto.LogEntry{
-      query_time: query_time,
-      decode_time: decode_time,
-      queue_time: queue_time,
+    params =
+      Enum.map(params, fn
+        %Ecto.Query.Tagged{value: value} -> value
+        value -> value
+      end)
+
+    acc = if idle_time, do: [idle_time: idle_time], else: []
+
+    measurements =
+      log_measurements(
+        [query_time: query_time, decode_time: decode_time, queue_time: queue_time],
+        0,
+        acc
+      )
+
+    metadata = %{
+      type: :ecto_sql_query,
+      repo: repo,
       result: log_result(result),
-      params: [],
+      params: params,
       query: format_query(query, params),
-      source: source
-    })
+      source: source,
+      options: Keyword.get(opts, :telemetry_options, [])
+    }
+
+    if event_name = Keyword.get(opts, :telemetry_event, event_name) do
+      :telemetry.execute(event_name, measurements, metadata)
+    end
+
+    case Keyword.get(opts, :log, log) do
+      true ->
+        Logger.log(
+          log,
+          fn -> log_iodata(measurements, metadata) end,
+          ansi_color: log_color(query)
+        )
+
+      false ->
+        :ok
+
+      level ->
+        Logger.log(
+          level,
+          fn -> log_iodata(measurements, metadata) end,
+          ansi_color: log_color(query)
+        )
+    end
+
+    :ok
   end
 
+  defp log_measurements([{_, nil} | rest], total, acc),
+    do: log_measurements(rest, total, acc)
+
+  defp log_measurements([{key, value} | rest], total, acc),
+    do: log_measurements(rest, total + value, [{key, value} | acc])
+
+  defp log_measurements([], total, acc),
+    do: Map.new([total_time: total] ++ acc)
+
+  # Currently unused
   defp log_result({:ok, _query, res}), do: {:ok, res}
   defp log_result(other), do: other
 
+  defp log_iodata(measurements, metadata) do
+    %{
+      params: params,
+      query: query,
+      result: result,
+      source: source
+    } = metadata
+
+    [
+      "QUERY",
+      ?\s,
+      log_ok_error(result),
+      log_ok_source(source),
+      log_time("db", measurements, :query_time, true),
+      log_time("decode", measurements, :decode_time, false),
+      log_time("queue", measurements, :queue_time, false),
+      log_time("idle", measurements, :idle_time, true),
+      ?\n,
+      query,
+      ?\s,
+      inspect(params, charlists: false)
+    ]
+  end
+
+  defp log_ok_error({:ok, _res}), do: "OK"
+  defp log_ok_error({:error, _err}), do: "ERROR"
+
+  defp log_ok_source(nil), do: ""
+  defp log_ok_source(source), do: " source=#{inspect(source)}"
+
+  defp log_time(label, measurements, key, force) do
+    case measurements do
+      %{^key => time} ->
+        us = System.convert_time_unit(time, :native, :microsecond)
+        ms = div(us, 100) / 10
+
+        if force or ms > 0 do
+          [?\s, label, ?=, :io_lib_format.fwrite_g(ms), ?m, ?s]
+        else
+          []
+        end
+
+      %{} ->
+        []
+    end
+  end
+
   defp check_constraint_errors(%Mongo.Error{code: 11000, message: msg}) do
     {:invalid, [unique: extract_index(msg)]}
   end
@@ -196,7 +320,7 @@ defmodule Mongo.Ecto.Connection do
 
     case Enum.reverse(parts) do
       [_, index | _] ->
-        String.strip(index)
+        String.trim(index)
 
       _ ->
         raise "failed to extract index from error message: #{inspect(msg)}"
@@ -299,4 +423,15 @@ defmodule Mongo.Ecto.Connection do
   defp format_part(name, value) do
     [" ", name, "=" | inspect(value)]
   end
+
+  defp log_color(%Query{action: :command}), do: :white
+  defp log_color(%Query{action: :find}), do: :cyan
+  defp log_color(%Query{action: :insert_one}), do: :green
+  defp log_color(%Query{action: :insert_many}), do: :green
+  defp log_color(%Query{action: :update_one}), do: :yellow
+  defp log_color(%Query{action: :update_many}), do: :yellow
+  defp log_color(%Query{action: :delete_many}), do: :red
+  defp log_color(%Query{action: :replace_one}), do: :yellow
+  defp log_color(%Query{action: :get_more}), do: :cyan
+  defp log_color(%Query{action: _}), do: nil
 end

lib/mongo_ecto/conversions.ex

@@ -26,7 +26,7 @@ defmodule Mongo.Ecto.Conversions do
     do: map(list, &inject_params(&1, params, pk))
 
   def inject_params(
-        %Ecto.Query.Tagged{tag: tag, type: type, value: {:^, _, [idx]} = value},
+        %Ecto.Query.Tagged{tag: _tag, type: _type, value: {:^, _, [idx]} = _value},
         params,
         pk
       ) do
@@ -93,7 +93,10 @@ defmodule Mongo.Ecto.Conversions do
   end
 
   defp key(pk, pk), do: :_id
-  defp key(key, _), do: key
+
+  defp key(key, _) do
+    key
+  end
 
   defp map(map, _fun) when is_map(map) and map_size(map) == 0 do
     {:ok, %{}}

lib/mongo_ecto/normalized_query.ex

@@ -40,7 +40,6 @@ defmodule Mongo.Ecto.NormalizedQuery do
   end
 
   alias Mongo.Ecto.Conversions
-  alias Ecto.Query.Tagged
   alias Ecto.Query
 
   defmacrop is_op(op) do
@@ -49,7 +48,7 @@ defmodule Mongo.Ecto.NormalizedQuery do
     end
   end
 
-  def all(%Query{} = original, params) do
+  def all(original, params) do
     check_query!(original, [:limit, :offset])
 
     from = from(original)
@@ -125,7 +124,7 @@ defmodule Mongo.Ecto.NormalizedQuery do
     %WriteQuery{coll: coll, query: query, command: command, database: original.prefix}
   end
 
-  def update(%{source: {prefix, coll}, schema: schema}, fields, filter) do
+  def update(%{source: coll, prefix: prefix, schema: schema}, fields, filter) do
     command = command(:update, fields, primary_key(schema))
     query = query(filter, primary_key(schema))
 
@@ -143,13 +142,13 @@ defmodule Mongo.Ecto.NormalizedQuery do
     %WriteQuery{coll: coll, query: query, database: original.prefix}
   end
 
-  def delete(%{source: {prefix, coll}, schema: schema}, filter) do
+  def delete(%{source: coll, schema: schema, prefix: prefix}, filter) do
     query = query(filter, primary_key(schema))
 
     %WriteQuery{coll: coll, query: query, database: prefix}
   end
 
-  def insert(%{source: {prefix, coll}, schema: schema}, document) do
+  def insert(%{source: coll, schema: schema, prefix: prefix}, document) do
     command = command(:insert, document, primary_key(schema))
 
     %WriteQuery{coll: coll, command: command, database: prefix}
@@ -159,11 +158,11 @@ defmodule Mongo.Ecto.NormalizedQuery do
     %CommandQuery{command: command, database: Keyword.get(opts, :database, nil)}
   end
 
-  defp from(%Query{from: {coll, model}}) do
+  defp from(%Query{from: %{source: {coll, model}}}) do
     {coll, model, primary_key(model)}
   end
 
-  defp from(%Query{from: %Ecto.SubQuery{}}) do
+  defp from(%Query{from: %{source: %Ecto.SubQuery{}}}) do
     raise ArgumentError, "MongoDB does not support subqueries"
   end
 
@@ -172,8 +171,13 @@ defmodule Mongo.Ecto.NormalizedQuery do
 
   defp projection(%Query{select: nil}, _params, _from), do: {:find, %{}, []}
 
-  defp projection(%Query{select: %Query.SelectExpr{fields: fields}} = query, params, from),
-    do: projection(fields, params, from, query, %{}, [])
+  defp projection(
+         %Query{select: %Query.SelectExpr{fields: fields} = _select} = query,
+         params,
+         from
+       ) do
+    projection(fields, params, from, query, %{}, [])
+  end
 
   defp projection([], _params, _from, _query, pacc, facc), do: {:find, pacc, Enum.reverse(facc)}
 
@@ -340,8 +344,9 @@ defmodule Mongo.Ecto.NormalizedQuery do
     ["$set": values |> value(pk, "update command") |> map_unless_empty]
   end
 
-  defp both_nil(nil, nil), do: true
-  defp both_nil(_, _), do: false
+  # Currently unused
+  # defp both_nil(nil, nil), do: true
+  # defp both_nil(_, _), do: false
 
   defp offset_limit(nil, _params, _pk, _query, _where), do: nil
 

lib/mongo_ecto/regex.ex

@@ -60,4 +60,8 @@ defmodule Mongo.Ecto.Regex do
   """
   def load(%BSON.Regex{} = js), do: {:ok, Map.put(js, :__struct__, __MODULE__)}
   def load(_), do: :error
+
+  def embed_as(_), do: :dump
+
+  def equal?(a, b), do: a == b
 end