Implements async member sync with global progress

Author: gueldenstone

lib/valkyrie/application.ex

@@ -14,6 +14,7 @@ defmodule Valkyrie.Application do
        repos: Application.fetch_env!(:valkyrie, :ecto_repos), skip: skip_migrations?()},
       {DNSCluster, query: Application.get_env(:valkyrie, :dns_cluster_query) || :ignore},
       {Phoenix.PubSub, name: Valkyrie.PubSub},
+      Valkyrie.Members.SyncState,
       # Start a worker by calling: Valkyrie.Worker.start_link(arg)
       # {Valkyrie.Worker, arg},
       # Start to serve requests, typically the last entry

lib/valkyrie/authentik.ex

@@ -11,6 +11,8 @@ defmodule Valkyrie.Authentik do
     ]
   end
 
+  @page_size 100
+
   defp bearer_token do
     Application.fetch_env!(:valkyrie, :authentik_token)
   end
@@ -19,25 +21,44 @@ defmodule Valkyrie.Authentik do
     Application.fetch_env!(:valkyrie, :authentik_url)
   end
 
-  defp do_get_all_users(page \\ 1, acc \\ []) do
+  defp do_get_all_users(page, acc, progress_callback) do
     Logger.debug("Fetching users from page #{page}")
 
     url = "#{base_url()}/api/v3/core/users/"
 
     case Req.get(url,
            auth: {:bearer, bearer_token()},
-           params: [page: page, page_size: 20, is_active: true],
+           params: [page: page, page_size: @page_size, is_active: true],
            receive_timeout: 20_000
          ) do
       {:ok, %Req.Response{status: 200, body: %{"results" => results, "pagination" => pagination}}} ->
-        Logger.debug("Found #{length(results)} users on page #{page}")
-
         new_acc = Enum.concat(acc, results)
 
+        # Calculate total pages if available in pagination
+        total_pages =
+          case pagination do
+            %{"count" => count} when is_integer(count) ->
+              # Calculate total pages from count and page_size
+              ceil(count / @page_size)
+
+            _ ->
+              nil
+          end
+
+        # Call progress callback if provided
+        if not is_nil(progress_callback) do
+          progress_callback.(%{
+            page: page,
+            total_pages: total_pages,
+            users_fetched: length(new_acc),
+            status: :in_progress
+          })
+        end
+
         case pagination["next"] do
           next_page when next_page > 0 and is_integer(next_page) ->
             # Fetch next page recursively
-            do_get_all_users(next_page, new_acc)
+            do_get_all_users(next_page, new_acc, progress_callback)
 
           _ ->
             # No more pages
@@ -52,8 +73,8 @@ defmodule Valkyrie.Authentik do
     end
   end
 
-  def get_all_users do
-    case do_get_all_users() do
+  def get_all_users(progress_callback \\ nil) do
+    case do_get_all_users(1, [], progress_callback) do
       {:ok, users} ->
         {:ok,
          users

lib/valkyrie/members.ex

@@ -2,11 +2,14 @@ defmodule Valkyrie.Members do
   use Ash.Domain, otp_app: :valkyrie, extensions: [AshAdmin.Domain, AshPaperTrail.Domain]
 
   require Logger
+  require Ash.Query
 
   alias Valkyrie.Authentik
   alias Valkyrie.Members.Member
+  alias Valkyrie.Members.SyncState
 
   @member_group_uuid "005b0c1c-c1bf-4a57-99b2-66cf04f94cda"
+  @sync_progress_topic "sync_members:progress"
 
   admin do
     show? true
@@ -40,32 +43,128 @@ defmodule Valkyrie.Members do
   end
 
   def update_members_from_xhain_account_system do
-    case Authentik.get_all_users() do
-      {:ok, users} ->
-        {:ok,
-         users
-         |> Enum.filter(&has_required_attributes?/1)
-         |> Enum.map(&xhain_account_to_member_info/1)
-         |> Enum.each(fn member ->
-           Ash.create!(Member, member, action: :create)
-         end)}
-
-      {:error, reason} ->
-        {:error, reason}
+    perform_sync()
+  end
+
+  def update_members_from_xhain_account_system_async() do
+    case SyncState.start_sync() do
+      :ok ->
+        task = Task.async(fn -> perform_sync() end)
+        {:ok, task}
+
+      {:error, :already_syncing} ->
+        {:error, :already_syncing}
     end
   end
 
+  defp perform_sync() do
+    try do
+      progress_callback = create_progress_callback()
+
+      case Authentik.get_all_users(progress_callback) do
+        {:ok, users} ->
+          valid_users = process_valid_users(users)
+          broadcast_progress(:processing, length(valid_users))
+          remove_obsolete_members(valid_users)
+          create_members(valid_users)
+          broadcast_progress(:completed, length(valid_users))
+
+        {:error, reason} ->
+          handle_sync_error(reason)
+      end
+    rescue
+      e ->
+        handle_sync_error(Exception.message(e))
+        raise e
+    after
+      SyncState.finish_sync()
+    end
+  end
+
+  defp create_progress_callback() do
+    fn progress ->
+      Phoenix.PubSub.broadcast(
+        Valkyrie.PubSub,
+        @sync_progress_topic,
+        {:sync_progress, progress}
+      )
+    end
+  end
+
+  defp process_valid_users(users) do
+    users
+    |> Enum.filter(&has_required_attributes?/1)
+    |> Enum.map(&xhain_account_to_member_info/1)
+  end
+
+  defp remove_obsolete_members(valid_users) do
+    Member
+    |> Ash.Query.filter(is_manual_entry: false)
+    |> Ash.read!()
+    |> Enum.filter(fn member -> not member_exists_in_list?(member, valid_users) end)
+    |> Enum.each(fn member ->
+      Logger.info("Removing member #{inspect(member)}")
+      Ash.destroy!(member, action: :destroy)
+    end)
+  end
+
+  defp create_members(valid_users) do
+    Enum.each(valid_users, fn member ->
+      Ash.create!(Member, member, action: :create)
+    end)
+  end
+
+  defp broadcast_progress(status, users_count) do
+    progress_data = %{
+      page: nil,
+      total_pages: nil,
+      users_fetched: users_count,
+      status: status
+    }
+
+    Phoenix.PubSub.broadcast(
+      Valkyrie.PubSub,
+      @sync_progress_topic,
+      {:sync_progress, progress_data}
+    )
+  end
+
+  defp handle_sync_error(reason) do
+    progress_data = %{
+      page: nil,
+      total_pages: nil,
+      users_fetched: 0,
+      status: :error,
+      error: reason
+    }
+
+    Phoenix.PubSub.broadcast(
+      Valkyrie.PubSub,
+      @sync_progress_topic,
+      {:sync_progress, progress_data}
+    )
+  end
+
+  ## Helper functions ##
+
   defp xhain_account_to_member_info(%Authentik.XHainAccount{} = xhain_account) do
     %{
       username: xhain_account.username,
       xhain_account_id: xhain_account.xhain_account_id,
       ssh_public_key: xhain_account.ssh_public_key,
       tree_name: xhain_account.tree_name,
-      is_active: is_member_in_group(xhain_account.groups, @member_group_uuid),
-      has_key: get_key_status(xhain_account)
+      is_active: is_member_in_group(xhain_account.groups, @member_group_uuid)
     }
   end
 
+  defp member_exists_in_list?(member, valid_users) do
+    Enum.any?(valid_users, fn valid_user -> is_same_user?(member, valid_user) end)
+  end
+
+  defp is_same_user?(user1, user2) do
+    user1.username == user2.username
+  end
+
   defp has_required_attributes?(xhain_account) do
     has_username = not is_nil(xhain_account.username) and xhain_account.username != ""
     has_xhain_account_id = not is_nil(xhain_account.xhain_account_id)
@@ -85,14 +184,4 @@ defmodule Valkyrie.Members do
   defp is_member_in_group(groups, group_uuid) do
     Enum.any?(groups, fn group -> group == group_uuid end)
   end
-
-  defp get_key_status(member) do
-    case get_member_by_username(member.username) do
-      {:ok, member} ->
-        member.has_key
-
-      _ ->
-        false
-    end
-  end
 end

lib/valkyrie/members/sync_state.ex

@@ -0,0 +1,48 @@
+defmodule Valkyrie.Members.SyncState do
+  @moduledoc """
+  Agent to track global sync state and prevent concurrent sync operations.
+  """
+  use Agent
+
+  @name __MODULE__
+
+  def start_link(_opts) do
+    Agent.start_link(fn -> nil end, name: @name)
+  end
+
+  @doc """
+  Returns true if a sync is currently running, false otherwise.
+  """
+  def is_syncing? do
+    Agent.get(@name, fn state -> not is_nil(state) end)
+  end
+
+  @doc """
+  Attempts to start a sync.
+  Returns `:ok` if the lock was acquired, or `{:error, :already_syncing}` if a sync is already running.
+  """
+  def start_sync do
+    Agent.get_and_update(@name, fn
+      nil ->
+        state = %{started_at: DateTime.utc_now()}
+        {:ok, state}
+
+      existing_state ->
+        {{:error, :already_syncing}, existing_state}
+    end)
+  end
+
+  @doc """
+  Finishes the current sync, clearing the state.
+  """
+  def finish_sync do
+    Agent.update(@name, fn _state -> nil end)
+  end
+
+  @doc """
+  Gets the current sync state (for debugging/monitoring).
+  """
+  def get_state do
+    Agent.get(@name, fn state -> state end)
+  end
+end