Upgrade to Ecto 3

.iex.exs

@@ -102,7 +102,6 @@ alias SanbaseWeb.Graphql.Resolvers.{
   UserResolver,
   ApikeyResolver,
   EthAccountResolver,
-  ElasticsearchResolver,
   EtherbiResolver,
   FileResolver,
   GithubResolver,

config/config.exs

@@ -14,7 +14,7 @@ config :sanbase, ecto_repos: [Sanbase.Repo]
 
 config :phoenix, :json_library, Jason
 
-config :ecto, json_library: Jason
+config :postgrex, :json_library, Jason
 
 config :sanbase, Sanbase, environment: "#{Mix.env()}"
 
@@ -51,7 +51,8 @@ config :sanbase, Sanbase.Repo,
   pool_size: {:system, "SANBASE_POOL_SIZE", "20"},
   max_overflow: 5,
   # because of pgbouncer
-  prepare: :unnamed
+  prepare: :unnamed,
+  migration_timestamps: [type: :naive_datetime_usec]
 
 config :sanbase, Sanbase.Auth.Hmac, secret_key: {:system, "APIKEY_HMAC_SECRET_KEY", nil}
 
@@ -179,7 +180,6 @@ import_config "ex_admin_config.exs"
 import_config "influxdb_config.exs"
 import_config "scrapers_config.exs"
 import_config "notifications_config.exs"
-import_config "elasticsearch_config.exs"
 import_config "prometheus_config.exs"
 import_config "stripe_config.exs"
 import_config "scheduler_config.exs"

config/dev.exs

@@ -71,7 +71,8 @@ config :sanbase, Sanbase.ClickhouseRepo,
   password: "",
   pool_timeout: 60_000,
   timeout: 60_000,
-  pool_size: {:system, "CLICKHOUSE_POOL_SIZE", "3"}
+  pool_size: {:system, "CLICKHOUSE_POOL_SIZE", "3"},
+  show_sensitive_data_on_connection_error: true
 
 config :ex_admin,
   basic_auth: [

config/test.exs

@@ -33,6 +33,11 @@ config :sanbase, Sanbase.Repo,
   database: "sanbase_test",
   pool_size: 5
 
+config :sanbase, Sanbase.ClickhouseRepo,
+  pool: Ecto.Adapters.SQL.Sandbox,
+  database: "sanbase_test",
+  pool_size: 5
+
 config :sanbase, Sanbase.Auth.Hmac, secret_key: "Non_empty_key_used_in_tests_only"
 
 config :sanbase, Sanbase.ExternalServices.Coinmarketcap, sync_enabled: false
@@ -60,10 +65,6 @@ config :arc,
   storage: Arc.Storage.Local,
   storage_dir: "/tmp/sanbase/filestore-test/"
 
-config :sanbase, Sanbase.Elasticsearch.Cluster, api: Sanbase.ElasticsearchMock
-
-config :sanbase, Sanbase.Elasticsearch, indices: "index1,index2,index3,index4"
-
 config :sanbase, SanbaseWeb.Plug.VerifyStripeWebhook, webhook_secret: "stripe_webhook_secret"
 
 config :sanbase, Sanbase.Signal, email_channel_enabled: {:system, "EMAIL_CHANNEL_ENABLED", "true"}

lib/sanbase/application.ex

@@ -1,8 +1,6 @@
 defmodule Sanbase.Application do
   use Application
 
-  import Sanbase.ApplicationUtils
-
   require Logger
   require Sanbase.Utils.Config, as: Config
 
@@ -167,7 +165,7 @@ defmodule Sanbase.Application do
       Sanbase.Repo,
 
       # Start the Clickhouse Repo
-      start_in({Sanbase.ClickhouseRepo, []}, [:dev, :prod]),
+      Sanbase.ClickhouseRepo,
 
       # Time series Prices DB connection
       Sanbase.Prices.Store.child_spec(),

lib/sanbase/auth/user.ex

@@ -1,6 +1,7 @@
 defmodule Sanbase.Auth.User do
   use Ecto.Schema
-  use Timex.Ecto.Timestamps
+
+  # use Timex.Ecto.Timestamps
 
   import Ecto.Changeset
   import Ecto.Query
@@ -100,15 +101,17 @@ defmodule Sanbase.Auth.User do
     timestamps()
   end
 
-  def generate_salt do
+  def generate_salt() do
     :crypto.strong_rand_bytes(@salt_length) |> Base.url_encode64() |> binary_part(0, @salt_length)
   end
 
-  def generate_email_token do
+  def generate_email_token() do
     :crypto.strong_rand_bytes(@email_token_length) |> Base.url_encode64()
   end
 
   def changeset(%User{} = user, attrs \\ %{}) do
+    attrs = Sanbase.DateTimeUtils.truncate_datetimes(attrs)
+
     user
     |> cast(attrs, [
       :email,
@@ -188,15 +191,20 @@ defmodule Sanbase.Auth.User do
   def san_balance_cache_stale?(%User{san_balance_updated_at: nil}), do: true
 
   def san_balance_cache_stale?(%User{san_balance_updated_at: san_balance_updated_at}) do
-    Timex.diff(Timex.now(), san_balance_updated_at, :seconds) > @san_balance_cache_seconds
+    naive_now = NaiveDateTime.utc_now() |> NaiveDateTime.truncate(:second)
+    Timex.diff(naive_now, san_balance_updated_at, :seconds) > @san_balance_cache_seconds
   end
 
   def update_san_balance_changeset(user) do
     user = Repo.preload(user, :eth_accounts)
     san_balance = san_balance_for_eth_accounts(user)
+    naive_now = NaiveDateTime.utc_now() |> NaiveDateTime.truncate(:second)
 
     user
-    |> change(san_balance: san_balance, san_balance_updated_at: Timex.now())
+    |> change(
+      san_balance_updated_at: naive_now,
+      san_balance: san_balance
+    )
   end
 
   @spec san_balance(%User{}) :: {:ok, float()} | {:ok, nil} | {:error, String.t()}
@@ -273,7 +281,7 @@ defmodule Sanbase.Auth.User do
     user
     |> change(
       email_token: generate_email_token(),
-      email_token_generated_at: Timex.now(),
+      email_token_generated_at: NaiveDateTime.utc_now() |> NaiveDateTime.truncate(:second),
       email_token_validated_at: nil,
       consent_id: consent
     )
@@ -285,44 +293,57 @@ defmodule Sanbase.Auth.User do
     |> changeset(%{
       email_candidate: email_candidate,
       email_candidate_token: generate_email_token(),
-      email_candidate_token_generated_at: Timex.now(),
+      email_candidate_token_generated_at:
+        NaiveDateTime.utc_now() |> NaiveDateTime.truncate(:second),
       email_candidate_token_validated_at: nil
     })
     |> Repo.update()
   end
 
   def mark_email_token_as_validated(user) do
+    validated_at =
+      (user.email_token_validated_at || Timex.now())
+      |> Timex.to_naive_datetime()
+      |> NaiveDateTime.truncate(:second)
+
     user
     |> change(
-      email_token_validated_at: user.email_token_validated_at || Timex.now(),
+      email_token_validated_at: validated_at,
       is_registered: true
     )
     |> Repo.update()
   end
 
   def update_email_from_email_candidate(user) do
+    validated_at =
+      (user.email_candidate_token_validated_at || Timex.now())
+      |> Timex.to_naive_datetime()
+      |> NaiveDateTime.truncate(:second)
+
     user
     |> changeset(%{
       email: user.email_candidate,
       email_candidate: nil,
-      email_candidate_token_validated_at: user.email_candidate_token_validated_at || Timex.now()
+      email_candidate_token_validated_at: validated_at
     })
     |> Repo.update()
   end
 
   def email_token_valid?(user, token) do
+    naive_now = NaiveDateTime.utc_now() |> NaiveDateTime.truncate(:second)
+
     cond do
       user.email_token != token ->
         false
 
-      Timex.diff(Timex.now(), user.email_token_generated_at, :minutes) >
+      Timex.diff(naive_now, user.email_token_generated_at, :minutes) >
           @login_email_valid_minutes ->
         false
 
       user.email_token_validated_at == nil ->
         true
 
-      Timex.diff(Timex.now(), user.email_token_validated_at, :minutes) >
+      Timex.diff(naive_now, user.email_token_validated_at, :minutes) >
           @login_email_valid_after_validation_minutes ->
         false
 
@@ -332,18 +353,20 @@ defmodule Sanbase.Auth.User do
   end
 
   def email_candidate_token_valid?(user, email_candidate_token) do
+    naive_now = NaiveDateTime.utc_now() |> NaiveDateTime.truncate(:second)
+
     cond do
       user.email_candidate_token != email_candidate_token ->
         false
 
-      Timex.diff(Timex.now(), user.email_candidate_token_generated_at, :minutes) >
+      Timex.diff(naive_now, user.email_candidate_token_generated_at, :minutes) >
           @login_email_valid_minutes ->
         false
 
       user.email_candidate_token_validated_at == nil ->
         true
 
-      Timex.diff(Timex.now(), user.email_candidate_token_validated_at, :minutes) >
+      Timex.diff(naive_now, user.email_candidate_token_validated_at, :minutes) >
           @login_email_valid_after_validation_minutes ->
         false
 

lib/sanbase/clickhouse/erc20_transfers.ex

@@ -22,7 +22,7 @@ defmodule Sanbase.Clickhouse.Erc20Transfers do
   @table "erc20_transfers"
 
   @primary_key false
-  @timestamps_opts updated_at: false
+  @timestamps_opts [updated_at: false]
   schema @table do
     field(:datetime, :utc_datetime, source: :dt)
     field(:contract, :string, primary_key: true)

lib/sanbase/clickhouse/eth_transfers.ex

@@ -39,7 +39,7 @@ defmodule Sanbase.Clickhouse.EthTransfers do
   @eth_decimals 1_000_000_000_000_000_000
 
   @primary_key false
-  @timestamps_opts updated_at: false
+  @timestamps_opts [updated_at: false]
   schema @table do
     field(:datetime, :utc_datetime, source: :dt)
     field(:from_address, :string, primary_key: true, source: :from)

lib/sanbase/clickhouse/github/github.ex

@@ -37,7 +37,7 @@ defmodule Sanbase.Clickhouse.Github do
   @table "github"
 
   @primary_key false
-  @timestamps_opts updated_at: false
+  @timestamps_opts [updated_at: false]
   schema @table do
     field(:datetime, :utc_datetime, source: :dt, primary_key: true)
     field(:repo, :string, primary_key: true)

lib/sanbase/clickhouse_repo.ex

@@ -1,5 +1,7 @@
 defmodule Sanbase.ClickhouseRepo do
-  use Ecto.Repo, otp_app: :sanbase
+  # Clickhouse tests are done only through mocking the results.
+  @adapter if Mix.env() == :test, do: Ecto.Adapters.Postgres, else: ClickhouseEcto
+  use Ecto.Repo, otp_app: :sanbase, adapter: @adapter
 
   require Sanbase.Utils.Config, as: Config
 
@@ -23,6 +25,9 @@ defmodule Sanbase.ClickhouseRepo do
       try do
         require Sanbase.ClickhouseRepo, as: ClickhouseRepo
 
+        ordered_params = ClickhouseRepo.order_params(query, args)
+        sanitized_query = ClickhouseRepo.sanitize_query(query)
+
         ClickhouseRepo.query(query, args)
         |> case do
           {:ok, result} ->
@@ -46,55 +51,67 @@ defmodule Sanbase.ClickhouseRepo do
     end
   end
 
-  defmacro query_transform(query, args, transform_fn) do
-    quote bind_quoted: [query: query, args: args, transform_fn: transform_fn] do
-      try do
-        require Sanbase.ClickhouseRepo, as: ClickhouseRepo
+  def query_transform(query, args, transform_fn) do
+    try do
+      ordered_params = order_params(query, args)
+      sanitized_query = sanitize_query(query)
 
-        ClickhouseRepo.query(query, args)
-        |> case do
-          {:ok, result} ->
-            result =
-              Enum.map(
-                result.rows,
-                transform_fn
-              )
+      __MODULE__.query(sanitized_query, ordered_params)
+      |> case do
+        {:ok, result} -> {:ok, Enum.map(result.rows, transform_fn)}
+        {:error, error} -> {:error, error}
+      end
+    rescue
+      e -> {:error, "Cannot execute ClickHouse query. Reason: #{Exception.message(e)}"}
+    end
+  end
 
-            {:ok, result}
+  def query_reduce(query, args, init, reducer) do
+    try do
+      ordered_params = order_params(query, args)
+      sanitized_query = sanitize_query(query)
 
-          {:error, error} ->
-            {:error, error}
-        end
-      rescue
-        e -> {:error, "Cannot execute ClickHouse query. Reason: #{Exception.message(e)}"}
+      __MODULE__.query(sanitized_query, ordered_params)
+      |> case do
+        {:ok, result} -> {:ok, Enum.reduce(result.rows, init, reducer)}
+        {:error, error} -> {:error, error}
       end
+    rescue
+      e ->
+        {:error, "Cannot execute ClickHouse query. Reason: #{Exception.message(e)}"}
     end
   end
 
-  defmacro query_reduce(query, args, init, reducer) do
-    quote bind_quoted: [query: query, args: args, init: init, reducer: reducer] do
-      try do
-        require Sanbase.ClickhouseRepo, as: ClickhouseRepo
+  def sanitize_query(query) do
+    query
+    |> IO.iodata_to_binary()
+    |> String.replace(~r/(\?([0-9]+))(?=(?:[^\\"']|[\\"'][^\\"']*[\\"'])*$)/, "?")
+  end
 
-        ClickhouseRepo.query(query, args)
-        |> case do
-          {:ok, result} ->
-            result =
-              Enum.reduce(
-                result.rows,
-                init,
-                reducer
-              )
+  def order_params(query, params) do
+    sanitised =
+      Regex.replace(~r/(([^\\]|^))["'].*?[^\\]['"]/, IO.iodata_to_binary(query), "\\g{1}")
 
-            {:ok, result}
+    ordering =
+      Regex.scan(~r/\?([0-9]+)/, sanitised)
+      |> Enum.map(fn [_, x] -> String.to_integer(x) end)
 
-          {:error, error} ->
-            {:error, error}
-        end
-      rescue
-        e ->
-          {:error, "Cannot execute ClickHouse query. Reason: #{Exception.message(e)}"}
-      end
+    ordering_count = Enum.max_by(ordering, fn x -> x end, fn -> 0 end)
+
+    if ordering_count != length(params) do
+      raise "\nError: number of params received (#{length(params)}) does not match expected (#{
+              ordering_count
+            })"
+    end
+
+    ordered_params =
+      ordering
+      |> Enum.reduce([], fn ix, acc -> [Enum.at(params, ix - 1) | acc] end)
+      |> Enum.reverse()
+
+    case ordered_params do
+      [] -> params
+      _ -> ordered_params
     end
   end
 end