backup

lib/sanbase/metric/registry/sync.ex

@@ -7,13 +7,17 @@ defmodule Sanbase.Metric.Registry.Sync do
   alias Sanbase.Metric.Registry
   alias Sanbase.Utils.Config
 
+  require Logger
+
   def by_uuid(uuid), do: Registry.SyncSchema.by_uuid(uuid)
 
   @doc ~s"""
-
+  TODO
   """
   @spec sync(list(non_neg_integer())) :: :ok
   def sync(metric_registry_ids) when is_list(metric_registry_ids) do
+    Logger.info("Initiating sync for #{length(metric_registry_ids)} metric registry records")
+
     with :ok <- no_running_syncs(),
          :ok <- check_initiate_env(),
          {:ok, content} <- get_sync_content(metric_registry_ids),
@@ -22,31 +26,40 @@ defmodule Sanbase.Metric.Registry.Sync do
          {:ok, sync} <- Registry.SyncSchema.update_status(sync, "executing") do
       {:ok, sync}
     end
-    |> dbg()
   end
 
-  def mark_sync_as_finished(sync_uuid) do
+  @doc ~s"""
+  When the sync application on prod is completed, it sends a HTTP POST request
+  back to the initiator that the sync is finished. The initiator calls this function
+  to mark the sync as completed.
+  """
+  def mark_sync_as_completed(sync_uuid) when is_binary(sync_uuid) do
+    Logger.info("Marking Metric Registry sync as finished for UUID #{sync_uuid}")
+
     with {:ok, sync} <- Registry.SyncSchema.by_uuid(sync_uuid),
-         {:ok, metric_registry_ids} <- extract_metric_registry_ids(sync),
-         :ok <- mark_metric_registries_as_synced(metric_registry_ids),
-         :ok <- Registry.SyncSchema.update_status(sync, "completed") do
+         {:ok, list} <- extract_metric_registry_list(sync),
+         :ok <- mark_metric_registries_as_synced(list),
+         {:ok, sync} <- Registry.SyncSchema.update_status(sync, "completed") do
       {:ok, sync}
     end
   end
 
-  def apply_sync(json) do
+  def apply_sync(params) do
+    Logger.info("Applying Metric Registry sync")
+
     with :ok <- check_apply_env(),
-         {:ok, map} when is_map(map) <- Jason.decode(json),
-         list when is_list(list) <- map["content"],
+         {:ok, list} when is_list(list) <- Jason.decode(params["content"]),
          :ok <- do_apply_sync_content(list),
-         :ok <- confirm_sync_completed(map["confirmation_endpoint"]) do
+         :ok <- send_sync_completed_confirmation(params["confirmation_endpoint"]) do
       :ok
     end
   end
 
   # Private functions
 
-  defp confirm_sync_completed(url) do
+  defp send_sync_completed_confirmation(url) do
+    Logger.info("Confirming that a Metric Registry sync was completed to url #{url}")
+
     Req.post(url)
   end
 
@@ -55,6 +68,11 @@ defmodule Sanbase.Metric.Registry.Sync do
     |> Enum.reduce(Ecto.Multi.new(), fn params, multi ->
       multi_metric_registry_update(multi, params)
     end)
+    |> Sanbase.Repo.transaction()
+    |> case do
+      {:ok, %{} = _map} -> :ok
+      {:error, _name, error, _changes_so_far} -> {:error, error}
+    end
   end
 
   defp multi_metric_registry_update(multi, params) do
@@ -64,7 +82,7 @@ defmodule Sanbase.Metric.Registry.Sync do
       params
 
     with {:ok, metric_registry} <- Registry.by_name(metric, data_type, fixed_parameters) do
-      changeset = Registry.changeset(metric, params)
+      changeset = Registry.changeset(metric_registry, params)
       Ecto.Multi.update(multi, metric_registry.id, changeset)
     end
   end
@@ -106,13 +124,20 @@ defmodule Sanbase.Metric.Registry.Sync do
     end
   end
 
-  defp mark_metric_registries_as_synced(metric_registry_ids) do
-    for id <- metric_registry_ids do
-      with {:ok, metric} <- Registry.by_id(id),
+  defp mark_metric_registries_as_synced(list) do
+    Enum.reduce_while(list, :ok, fn map, _acc ->
+      with {:ok, metric} <-
+             Registry.by_name(
+               Map.fetch!(map, "metric"),
+               Map.fetch!(map, "data_type"),
+               Map.fetch!(map, "fixed_parameters")
+             ),
            {:ok, _metric} <- Registry.update(metric, %{"sync_status" => "synced"}) do
-        :ok
+        {:cont, :ok}
+      else
+        error -> {:halt, error}
       end
-    end
+    end)
   end
 
   defp store_sync_in_db(content) do
@@ -122,7 +147,7 @@ defmodule Sanbase.Metric.Registry.Sync do
   defp get_sync_target_url() do
     secret = Config.module_get(Sanbase.Metric.Registry.Sync, :sync_secret)
     deployment_env = Config.module_get(Sanbase, :deployment_env)
-    port = Config.module_get(SanbaseWeb.Endpoint, :port)
+    port = Config.module_get(SanbaseWeb.Endpoint, [:http, :port])
 
     case deployment_env do
       "dev" -> "http://localhost:#{port}/sync_metric_registry?secret=#{secret}"
@@ -172,10 +197,9 @@ defmodule Sanbase.Metric.Registry.Sync do
     end
   end
 
-  defp extract_metric_registry_ids(sync) do
+  defp extract_metric_registry_list(sync) do
     sync.content
-    |> Jason.decode!()
-    |> Enum.map(& &1["id"])
+    |> Jason.decode()
   end
 
   defp get_confirmation_endpoint(sync) do

lib/sanbase_web/controllers/metric_registry_controller.ex

@@ -4,16 +4,19 @@ defmodule SanbaseWeb.MetricRegistryController do
   def sync(conn, %{"secret" => secret} = params) do
     case secret == get_sync_secret() do
       true ->
-        # TODO: Remove
-
-        {:ok, data, _conn_details} = Plug.Conn.read_body(conn)
-
-        IO.inspect({params})
-        IO.inspect({data})
+        case Sanbase.Metric.Registry.Sync.apply_sync(
+               Map.take(params, ["content", "confirmation_endpoint"])
+             ) do
+          {:ok, _sync} ->
+            conn
+            |> resp(200, "OK")
+            |> send_resp()
 
-        conn
-        |> resp(200, "OK")
-        |> send_resp()
+          {:error, error} ->
+            conn
+            |> resp(500, "Error Syncing: #{inspect(error)}")
+            |> send_resp()
+        end
 
       false ->
         conn
@@ -22,10 +25,10 @@ defmodule SanbaseWeb.MetricRegistryController do
     end
   end
 
-  def mark_sync_as_finished(conn, %{"sync_uuid" => sync_uuid, "secret" => secret}) do
+  def mark_sync_as_completed(conn, %{"sync_uuid" => sync_uuid, "secret" => secret}) do
     case secret == get_sync_secret() do
       true ->
-        case Sanbase.Metric.Registry.Sync.mark_sync_as_finished(sync_uuid) do
+        case Sanbase.Metric.Registry.Sync.mark_sync_as_completed(sync_uuid) do
           {:ok, _} ->
             conn
             |> resp(200, "OK")

lib/sanbase_web/router.ex

@@ -187,7 +187,7 @@ defmodule SanbaseWeb.Router do
     post(
       "/mark_metric_registry_sync_as_finished/:sync_uuid",
       MetricRegistryController,
-      :mark_sync_as_finished
+      :mark_sync_as_completed
     )
 
     get("/api_metric_name_mapping", MetricNameController, :api_metric_name_mapping)