fix: update encrypt jwt secret migration

lib/realtime_web/channels/realtime_channel.ex

@@ -214,8 +214,11 @@ defmodule RealtimeWeb.RealtimeChannel do
       when is_binary(refresh_token) do
     cancel_timer(ref)
 
-    with {:ok, %{"exp" => exp} = claims} when is_integer(exp) <-
-           ChannelsAuthorization.authorize_conn(refresh_token, jwt_secret),
+    secure_key = Application.get_env(:realtime, :db_enc_key)
+
+    with jwt_secret_dec <- decrypt!(jwt_secret, secure_key),
+         {:ok, %{"exp" => exp} = claims} when is_integer(exp) <-
+           ChannelsAuthorization.authorize_conn(refresh_token, jwt_secret_dec),
          exp_diff when exp_diff > 0 <- exp - Joken.current_time(),
          expire_ref <- Process.send_after(self(), :expire_token, exp_diff * 1_000) do
       unless is_nil(postgres_config) do

priv/repo/migrations/20220715115810_encrypt_jwt_secret.exs

@@ -1,30 +1,59 @@
 defmodule Realtime.Repo.Migrations.EncryptJwtSecret do
   use Ecto.Migration
-  import Ecto.Query
+
+  import Ecto.Query, only: [from: 2]
+
+  alias Ecto.Changeset
   alias Realtime.{Repo, Api.Tenant}
-  import Realtime.Helpers, only: [encrypt!: 2]
+
+  @batch_size 5_000
 
   def change do
-    secure_key = System.get_env("DB_ENC_KEY")
+    stream =
+      from(t in Tenant,
+        select: t
+      )
+      |> Repo.stream(max_rows: @batch_size)
 
     Repo.transaction(fn ->
-      from(t in Realtime.Api.Tenant, select: t)
-      |> Realtime.Repo.all()
-      |> Enum.each(fn e ->
-        %{
-          Realtime.Api.Tenant.changeset(e, %{})
-          | action: :update,
-            changes: %{jwt_secret: encrypt!(e.jwt_secret, secure_key)}
-        }
-        |> Realtime.Repo.update()
-      end)
-      |> case do
-        {:error, reason} ->
-          raise(reason)
+      stream
+      |> Stream.map(fn %Tenant{jwt_secret: jwt_secret} = tenant ->
+        tenant
+        |> Map.drop([:jwt_secret])
+        |> Tenant.changeset(%{jwt_secret: jwt_secret})
+        |> case do
+          %Changeset{changes: changes, data: data, valid?: true} ->
+            data
+            |> Map.take([
+              :external_id,
+              :id,
+              :inserted_at,
+              :jwt_secret,
+              :max_concurrent_users,
+              :max_events_per_second,
+              :name,
+              :updated_at
+            ])
+            |> Map.put(:updated_at, NaiveDateTime.utc_now() |> NaiveDateTime.truncate(:second))
+            |> Map.merge(changes)
 
-        _ ->
-          :ok
-      end
+          changeset ->
+            Repo.rollback(changeset)
+        end
+      end)
+      |> Stream.chunk_every(@batch_size)
+      |> Stream.each(
+        &Repo.insert_all(Tenant, &1,
+          conflict_target: [:id],
+          on_conflict: {:replace, [:jwt_secret, :updated_at]}
+        )
+      )
+      |> Enum.to_list()
     end)
+    |> case do
+      {:error, reason} -> reason |> inspect() |> raise()
+      {:error, _, reason, _} -> reason |> inspect() |> raise()
+      {:ok, _} -> :ok
+    end
   end
 end