chore: Merge pull request #12 from chasehult/pk-fix

Fix import for PK and SP

Author: atlas-oc

lib/octocon/workers/plural_kit_import_worker.ex

@@ -8,6 +8,8 @@ defmodule Octocon.Workers.PluralKitImportWorker do
   - `pk_token` (binary): The PluralKit API token to use for the request.
   """
 
+  @insert_concurrency 10
+
   alias Octocon.{
     Accounts,
     Alters,
@@ -53,45 +55,39 @@ defmodule Octocon.Workers.PluralKitImportWorker do
 
     alter_count = length(alters)
 
-    chunked_alters =
-      alters
-      |> Stream.map(
-        &Map.drop(&1, [:__meta__, :__struct__, :fronts, :user, :global_journals, :tags])
-      )
-      |> Enum.chunk_every(1000)
-
-    Repo.transaction(fn ->
-      chunked_alters
-      |> Enum.each(fn chunk ->
-        Repo.insert_all_regional(Alter, chunk, {:user, {:system, system_id}})
-      end)
+    alters
+    |> Task.async_stream(
+      &Repo.insert_regional(&1, {:user, {:system, system_id}}),
+      max_concurrency: @insert_concurrency,
+      ordered: false
+    )
+    |> Stream.run()
 
-      user = Accounts.get_user!({:system, system_id})
+    user = Accounts.get_user!({:system, system_id})
 
-      Accounts.update_user(
-        user,
-        %{
-          lifetime_alter_count: user.lifetime_alter_count + alter_count,
-          description: default_if_empty(description, 3000, user.description)
-        }
-      )
-
-      Accounts.update_discord_settings(
-        user,
-        %{
-          system_tag:
-            default_if_empty(
-              system_tag,
-              20,
-              Map.get(
-                user.discord_settings || %Octocon.Accounts.DiscordSettings{},
-                :system_tag,
-                nil
-              )
+    Accounts.update_user(
+      user,
+      %{
+        lifetime_alter_count: user.lifetime_alter_count + alter_count,
+        description: default_if_empty(description, 3000, user.description)
+      }
+    )
+
+    Accounts.update_discord_settings(
+      user,
+      %{
+        system_tag:
+          default_if_empty(
+            system_tag,
+            20,
+            Map.get(
+              user.discord_settings || %Octocon.Accounts.DiscordSettings{},
+              :system_tag,
+              nil
             )
-        }
-      )
-    end)
+          )
+      }
+    )
 
     OctoconWeb.Endpoint.broadcast!("system:#{system_id}", "alters_created", %{
       alters: Enum.map(alters, &OctoconWeb.System.AlterJSON.data_me(&1))

lib/octocon/workers/simply_plural_import_worker.ex

@@ -8,6 +8,8 @@ defmodule Octocon.Workers.SimplyPluralImportWorker do
   - `sp_token` (binary): The Simply Plural API token to use for the request.
   """
 
+  @insert_concurrency 10
+
   alias Octocon.{
     Accounts,
     Alters,
@@ -53,29 +55,23 @@ defmodule Octocon.Workers.SimplyPluralImportWorker do
 
     alter_count = length(alters)
 
-    chunked_alters =
-      alters
-      |> Stream.map(
-        &Map.drop(&1, [:__meta__, :__struct__, :fronts, :user, :global_journals, :tags])
-      )
-      |> Enum.chunk_every(1000)
-
-    Repo.transaction(fn ->
-      chunked_alters
-      |> Enum.each(fn chunk ->
-        Repo.insert_all_regional(Alter, chunk, {:user, {:system, system_id}})
-      end)
+    alters
+    |> Task.async_stream(
+      &Repo.insert_regional(&1, {:user, {:system, system_id}}),
+      max_concurrency: @insert_concurrency,
+      ordered: false
+    )
+    |> Stream.run()
 
-      user = Accounts.get_user!({:system, system_id})
+    user = Accounts.get_user!({:system, system_id})
 
-      Accounts.update_user(
-        user,
-        %{
-          lifetime_alter_count: user.lifetime_alter_count + alter_count,
-          description: default_if_empty(description, 3000, user.description)
-        }
-      )
-    end)
+    Accounts.update_user(
+      user,
+      %{
+        lifetime_alter_count: user.lifetime_alter_count + alter_count,
+        description: default_if_empty(description, 3000, user.description)
+      }
+    )
 
     OctoconWeb.Endpoint.broadcast!("system:#{system_id}", "alters_created", %{
       alters: Enum.map(alters, &OctoconWeb.System.AlterJSON.data_me(&1))