Packages
phoenix_kit
1.7.75
1.7.207
1.7.206
1.7.205
1.7.204
1.7.203
1.7.202
1.7.201
1.7.200
1.7.199
1.7.198
1.7.197
1.7.196
1.7.194
1.7.193
1.7.192
1.7.191
1.7.190
1.7.189
1.7.187
1.7.186
1.7.185
1.7.184
1.7.183
1.7.182
1.7.181
1.7.180
1.7.179
1.7.178
1.7.177
1.7.176
1.7.175
1.7.174
1.7.173
1.7.172
1.7.171
1.7.170
1.7.169
1.7.168
1.7.167
1.7.166
1.7.165
1.7.164
1.7.162
1.7.161
1.7.160
1.7.159
1.7.157
1.7.156
1.7.155
1.7.154
1.7.153
1.7.152
1.7.151
1.7.150
1.7.149
1.7.146
1.7.145
1.7.144
1.7.143
1.7.138
1.7.133
1.7.132
1.7.131
1.7.130
1.7.128
1.7.126
1.7.125
1.7.121
1.7.120
1.7.119
1.7.118
1.7.117
1.7.116
1.7.115
1.7.114
1.7.113
1.7.112
1.7.111
1.7.110
1.7.109
1.7.108
1.7.107
1.7.106
1.7.105
1.7.104
1.7.103
1.7.102
1.7.101
1.7.100
1.7.99
1.7.98
1.7.97
1.7.96
1.7.95
1.7.94
1.7.93
1.7.92
1.7.91
1.7.90
1.7.89
1.7.88
1.7.87
1.7.86
1.7.85
1.7.84
1.7.83
1.7.82
1.7.81
1.7.80
1.7.79
1.7.78
1.7.77
1.7.76
1.7.75
1.7.74
1.7.71
1.7.70
1.7.69
1.7.66
1.7.65
1.7.64
1.7.63
1.7.62
1.7.61
1.7.59
1.7.58
1.7.57
1.7.56
1.7.55
1.7.54
1.7.53
1.7.52
1.7.51
1.7.49
1.7.44
1.7.43
1.7.42
1.7.41
1.7.39
1.7.38
1.7.37
1.7.36
1.7.34
1.7.33
1.7.31
1.7.30
1.7.29
1.7.28
1.7.27
1.7.26
1.7.25
1.7.24
1.7.23
1.7.22
1.7.21
1.7.20
1.7.19
1.7.18
1.7.17
1.7.16
1.7.15
1.7.14
1.7.13
1.7.12
1.7.11
1.7.10
1.7.9
1.7.8
1.7.7
1.7.6
1.7.5
1.7.4
1.7.3
1.7.2
1.7.1
1.7.0
1.6.20
1.6.19
1.6.18
1.6.17
1.6.16
1.6.15
1.6.14
1.6.13
1.6.12
1.6.11
1.6.10
1.6.9
1.6.8
1.6.7
1.6.6
1.6.5
1.6.4
1.6.3
1.5.2
1.5.1
1.5.0
1.4.9
1.4.8
1.4.7
1.4.6
1.4.5
1.4.4
1.4.3
1.4.2
1.4.1
1.4.0
1.3.2
1.3.1
1.3.0
1.2.10
1.2.9
1.2.8
1.2.7
1.2.5
1.2.4
1.2.2
1.2.1
1.2.0
1.1.0
1.0.0
A foundation for building Elixir Phoenix apps — SaaS, social networks, ERP systems, marketplaces, and more
Current section
Files
Jump to
Current section
Files
lib/modules/publishing/listing_cache.ex
defmodule PhoenixKit.Modules.Publishing.ListingCache do
@moduledoc """
Caches publishing group listing metadata in :persistent_term for sub-millisecond reads.
Instead of querying the database on every request, the listing page reads from
an in-memory cache populated from the database.
## How It Works
1. When a post is created/updated/published, `regenerate/1` is called
2. This queries the database and stores post metadata in :persistent_term
3. `render_group_listing` reads from the in-memory cache
4. Cache includes: title, slug, date, status, languages, versions (no content)
## Performance
- Cache miss: ~20ms (DB query + store in :persistent_term)
- Cache hit: ~0.1μs (direct memory access, no variance)
## Cache Invalidation
Cache is regenerated when:
- Post is created
- Post is updated (metadata or content)
- Post status changes (draft/published/archived)
- Translation is added
- Version is created
## In-Memory Caching with :persistent_term
For sub-millisecond performance, parsed cache data is stored in `:persistent_term`.
- First read after restart: queries DB, stores in :persistent_term (~20ms)
- Subsequent reads: direct memory access (~0.1μs, no variance)
- On regenerate: updates :persistent_term from DB
- On invalidate: clears :persistent_term entry (next read triggers regeneration)
"""
alias PhoenixKit.Modules.Publishing.Constants
alias PhoenixKit.Modules.Publishing.DBStorage
@timestamp_modes Constants.timestamp_modes()
alias PhoenixKit.Modules.Publishing.LanguageHelpers
alias PhoenixKit.Modules.Publishing.PubSub, as: PublishingPubSub
alias PhoenixKit.Settings
alias PhoenixKit.Utils.Date, as: UtilsDate
require Logger
@persistent_term_prefix :phoenix_kit_group_listing_cache
@persistent_term_loaded_at_prefix :phoenix_kit_group_listing_cache_loaded_at
@persistent_term_cache_generated_at_prefix :phoenix_kit_group_listing_cache_generated_at
# ETS table for regeneration locks (provides atomic test-and-set via insert_new)
@lock_table :phoenix_kit_listing_cache_locks
# Settings key for memory cache toggle
@memory_cache_key "publishing_memory_cache_enabled"
@doc """
Reads the cached listing for a publishing group.
Returns `{:ok, posts}` if cache exists and is valid.
Returns `{:error, :cache_miss}` if cache doesn't exist or caching is disabled.
Respects the `publishing_memory_cache_enabled` setting.
"""
@spec read(String.t()) :: {:ok, [map()]} | {:error, :cache_miss}
def read(group_slug) do
if memory_cache_enabled?() do
term_key = persistent_term_key(group_slug)
case safe_persistent_term_get(term_key) do
{:ok, _} = hit ->
hit
:not_found ->
# Cache miss — regenerate from database
regenerate(group_slug)
case safe_persistent_term_get(term_key) do
{:ok, _} = hit -> hit
:not_found -> {:error, :cache_miss}
end
end
else
{:error, :cache_miss}
end
end
# Safely get from :persistent_term (returns :not_found instead of raising)
defp safe_persistent_term_get(key) do
{:ok, :persistent_term.get(key)}
rescue
ArgumentError -> :not_found
end
@doc """
Regenerates the listing cache for a group.
Queries the database for all posts and stores the metadata in :persistent_term.
This should be called after any post operation that changes the listing:
- create_post
- update_post
- add_language_to_post
- create_new_version
Returns `:ok` on success or `{:error, reason}` on failure.
"""
@spec regenerate(String.t()) :: :ok | {:error, any()}
def regenerate(group_slug) do
if memory_cache_enabled?() do
do_regenerate(group_slug)
else
:ok
end
rescue
error ->
Logger.error(
"[ListingCache] Failed to regenerate cache for #{group_slug}: #{inspect(error)}"
)
{:error, {:regenerate_failed, error}}
end
# Maximum number of posts to cache in :persistent_term per group.
# Groups exceeding this will still work but only cache the most recent posts.
@max_cached_posts 5000
defp do_regenerate(group_slug) do
start_time = System.monotonic_time(:millisecond)
# Posts from to_listing_map are already atom-key maps with excerpts
all_posts = DBStorage.list_posts_for_listing(group_slug)
posts =
if length(all_posts) > @max_cached_posts do
Logger.warning(
"[ListingCache] Group #{group_slug} has #{length(all_posts)} posts, caching most recent #{@max_cached_posts}"
)
Enum.take(all_posts, @max_cached_posts)
else
all_posts
end
generated_at = UtilsDate.utc_now() |> DateTime.to_iso8601()
safe_persistent_term_put(persistent_term_key(group_slug), posts)
safe_persistent_term_put(loaded_at_key(group_slug), generated_at)
safe_persistent_term_put(cache_generated_at_key(group_slug), generated_at)
elapsed = System.monotonic_time(:millisecond) - start_time
Logger.debug(
"[ListingCache] Regenerated cache from DB for #{group_slug} (#{length(posts)} posts) in #{elapsed}ms"
)
PublishingPubSub.broadcast_cache_changed(group_slug)
:ok
rescue
error ->
Logger.error(
"[ListingCache] Failed to regenerate cache for #{group_slug}: #{inspect(error)}"
)
{:error, {:regenerate_failed, error}}
end
# Lock timeout in milliseconds (30 seconds)
# If a lock is older than this, it's considered stale (process likely died)
@lock_timeout_ms 30_000
@doc """
Regenerates the cache if no other process is already regenerating it.
This prevents the "thundering herd" problem where multiple concurrent requests
all trigger cache regeneration simultaneously after a server restart.
Uses ETS with `insert_new/2` for atomic lock acquisition - only one process
can acquire the lock at a time. The lock includes a timestamp and will be
considered stale after #{@lock_timeout_ms}ms to prevent permanent lockout
if a process dies mid-regeneration.
Returns:
- `:ok` if regeneration was performed successfully
- `:already_in_progress` if another process is currently regenerating
- `{:error, reason}` if regeneration failed
## Usage
On cache miss in read paths, use this instead of `regenerate/1`:
case ListingCache.regenerate_if_not_in_progress(group_slug) do
:ok -> # Cache is ready, read from it
:already_in_progress -> # Another process is regenerating, try again later
{:error, _} -> # Regeneration failed, query DB directly
end
"""
@spec regenerate_if_not_in_progress(String.t()) :: :ok | :already_in_progress | {:error, any()}
def regenerate_if_not_in_progress(group_slug) do
ensure_lock_table_exists()
now = System.monotonic_time(:millisecond)
# Try to atomically acquire the lock using ETS insert_new
# Returns true if inserted (lock acquired), false if key already exists
case :ets.insert_new(@lock_table, {group_slug, now}) do
true ->
# We acquired the lock - perform regeneration
do_regenerate_with_lock(group_slug)
false ->
# Lock exists - check if it's stale
handle_existing_lock(group_slug, now)
end
end
# Handle case where lock already exists - check staleness
defp handle_existing_lock(group_slug, now) do
case :ets.lookup(@lock_table, group_slug) do
[{^group_slug, lock_timestamp}] ->
lock_age = now - lock_timestamp
if lock_age < @lock_timeout_ms do
# Lock is valid and recent - another process is regenerating
Logger.debug(
"[ListingCache] Regeneration already in progress for #{group_slug} (#{lock_age}ms ago), skipping"
)
:already_in_progress
else
# Lock is stale - previous process likely died
# Try to take over by deleting and re-acquiring atomically
take_over_stale_lock(group_slug, lock_timestamp, lock_age, now)
end
[] ->
# Lock was released between insert_new and lookup - try again
regenerate_if_not_in_progress(group_slug)
end
end
# Attempt to take over a stale lock using compare-and-delete
defp take_over_stale_lock(group_slug, old_timestamp, lock_age, now) do
# Use match_delete for atomic compare-and-delete
# Only deletes if the timestamp matches (no one else took over)
case :ets.select_delete(@lock_table, [{{group_slug, old_timestamp}, [], [true]}]) do
1 ->
# Successfully deleted stale lock - now try to acquire
Logger.warning(
"[ListingCache] Found stale lock for #{group_slug} (#{lock_age}ms old), taking over regeneration"
)
case :ets.insert_new(@lock_table, {group_slug, now}) do
true ->
do_regenerate_with_lock(group_slug)
false ->
# Another process beat us to it
:already_in_progress
end
0 ->
# Lock was already taken over by another process or timestamp changed
:already_in_progress
end
end
# Perform regeneration while holding the lock
defp do_regenerate_with_lock(group_slug) do
result = regenerate(group_slug)
case result do
:ok -> :ok
{:error, _} = error -> error
end
after
# Always release the lock when done (success or failure)
:ets.delete(@lock_table, group_slug)
end
# Ensure the ETS table for locks exists (lazy initialization)
defp ensure_lock_table_exists do
case :ets.whereis(@lock_table) do
:undefined ->
# Table doesn't exist - create it
# Use :public so any process can read/write
# Use :named_table so we can reference by atom
# Use :set for key-value storage
try do
:ets.new(@lock_table, [:set, :public, :named_table])
rescue
ArgumentError ->
# Table was created by another process between whereis and new
:ok
end
_tid ->
:ok
end
end
# Safely put to :persistent_term (logs warning on failure instead of crashing)
defp safe_persistent_term_put(key, value) do
:persistent_term.put(key, value)
rescue
error ->
Logger.warning("[ListingCache] Failed to write to :persistent_term: #{inspect(error)}")
:error
end
@doc """
Loads the cache from the database into :persistent_term.
Returns `:ok` if successful or `{:error, reason}` on failure.
"""
@spec load_into_memory(String.t()) :: :ok | {:error, any()}
def load_into_memory(group_slug) do
load_into_memory_from_db(group_slug)
end
defp load_into_memory_from_db(group_slug) do
posts = DBStorage.list_posts_for_listing(group_slug)
generated_at = UtilsDate.utc_now() |> DateTime.to_iso8601()
safe_persistent_term_put(persistent_term_key(group_slug), posts)
safe_persistent_term_put(loaded_at_key(group_slug), generated_at)
safe_persistent_term_put(cache_generated_at_key(group_slug), generated_at)
Logger.debug(
"[ListingCache] Loaded #{group_slug} from DB into :persistent_term (#{length(posts)} posts)"
)
PublishingPubSub.broadcast_cache_changed(group_slug)
:ok
rescue
error ->
Logger.error("[ListingCache] Failed to load #{group_slug} from DB: #{inspect(error)}")
{:error, {:load_failed, error}}
end
@doc """
Invalidates (clears) the cache for a group.
Clears the :persistent_term entries. The next read will trigger
a regeneration from the database.
"""
@spec invalidate(String.t()) :: :ok
def invalidate(group_slug) do
# Clear :persistent_term entries
term_key = persistent_term_key(group_slug)
try do
:persistent_term.erase(term_key)
rescue
ArgumentError -> :ok
end
try do
:persistent_term.erase(loaded_at_key(group_slug))
rescue
ArgumentError -> :ok
end
try do
:persistent_term.erase(cache_generated_at_key(group_slug))
rescue
ArgumentError -> :ok
end
Logger.debug("[ListingCache] Invalidated cache for #{group_slug}")
:ok
end
@doc """
Checks if a cache exists for a group in :persistent_term.
"""
@spec exists?(String.t()) :: boolean()
def exists?(group_slug) do
case safe_persistent_term_get(persistent_term_key(group_slug)) do
{:ok, _} -> true
:not_found -> false
end
end
@doc """
Finds a post by slug in the cache.
This is useful for single post views where we need metadata (language_statuses,
version_statuses, allow_version_access) without a separate DB query.
Returns `{:ok, cached_post}` if found, `{:error, :not_found}` otherwise.
"""
@spec find_post(String.t(), String.t()) :: {:ok, map()} | {:error, :not_found | :cache_miss}
def find_post(group_slug, post_slug) do
case read(group_slug) do
{:ok, posts} ->
case Enum.find(posts, fn p -> p.slug == post_slug end) do
nil -> {:error, :not_found}
post -> {:ok, post}
end
{:error, _} = error ->
error
end
end
@doc """
Finds a post by path pattern in the cache (for timestamp mode).
Matches posts where the path contains the date/time pattern.
Returns `{:ok, cached_post}` if found, `{:error, :not_found}` otherwise.
"""
@spec find_post_by_path(String.t(), String.t(), String.t()) ::
{:ok, map()} | {:error, :not_found | :cache_miss}
def find_post_by_path(group_slug, date, time) do
case read(group_slug) do
{:ok, posts} ->
# Match posts using discrete date and time fields (more robust than path string matching)
# Parse the input date string to compare with the cached Date struct
target_date = parse_date_for_lookup(date)
# Normalize time format (handles both "HH:MM" and "HH:MM:SS")
target_time = normalize_time_for_lookup(time)
case Enum.find(posts, fn p ->
dates_match?(p.date, target_date) && times_match?(p.time, target_time)
end) do
nil -> {:error, :not_found}
post -> {:ok, post}
end
{:error, _} = error ->
error
end
end
# Parse date string for lookup comparison
defp parse_date_for_lookup(date_str) when is_binary(date_str) do
case Date.from_iso8601(date_str) do
{:ok, date} -> date
_ -> date_str
end
end
defp parse_date_for_lookup(date), do: date
# Normalize time to "HH:MM" format for comparison
defp normalize_time_for_lookup(time_str) when is_binary(time_str) do
# Take just HH:MM portion
String.slice(time_str, 0, 5)
end
defp normalize_time_for_lookup(time), do: time
# Compare dates - handles both Date structs and strings
defp dates_match?(nil, _), do: false
defp dates_match?(_, nil), do: false
defp dates_match?(%Date{} = cached, %Date{} = target) do
Date.compare(cached, target) == :eq
end
defp dates_match?(%Date{} = cached, target_str) when is_binary(target_str) do
Date.to_iso8601(cached) == target_str
end
defp dates_match?(_, _), do: false
# Compare times - handles Time structs and "HH:MM" strings
defp times_match?(nil, _), do: false
defp times_match?(_, nil), do: false
defp times_match?(%Time{} = cached, target_str) when is_binary(target_str) do
# Format cached time as HH:MM and compare
cached_str = cached |> Time.to_string() |> String.slice(0, 5)
cached_str == target_str
end
defp times_match?(cached_str, target_str)
when is_binary(cached_str) and is_binary(target_str) do
String.slice(cached_str, 0, 5) == String.slice(target_str, 0, 5)
end
defp times_match?(_, _), do: false
@doc """
Finds a post by URL slug for a specific language.
This enables O(1) lookup from URL slug to internal identifier, supporting
per-language URL slugs for SEO-friendly localized URLs.
## Parameters
- `group_slug` - The publishing group
- `language` - The language code to search in
- `url_slug` - The URL slug to find
## Returns
- `{:ok, cached_post}` - Found post (includes internal `slug` for DB lookup)
- `{:error, :not_found}` - No post with this URL slug for this language
- `{:error, :cache_miss}` - Cache not available
"""
@spec find_by_url_slug(String.t(), String.t(), String.t()) ::
{:ok, map()} | {:error, :not_found | :cache_miss}
def find_by_url_slug(group_slug, language, url_slug) do
case read(group_slug) do
{:ok, posts} -> find_post_by_url_slug(posts, language, url_slug)
{:error, _} -> {:error, :cache_miss}
end
end
defp find_post_by_url_slug(posts, language, url_slug) do
case Enum.find(posts, &(Map.get(&1.language_slugs || %{}, language) == url_slug)) do
nil -> {:error, :not_found}
post -> {:ok, post}
end
end
@doc """
Finds a post by a previous URL slug for 301 redirects.
When a URL slug changes, the old slug is stored in `previous_url_slugs`.
This function finds posts that previously used the given URL slug.
## Returns
- `{:ok, cached_post}` - Found post that previously used this slug
- `{:error, :not_found}` - No post with this previous slug
- `{:error, :cache_miss}` - Cache not available
"""
@spec find_by_previous_url_slug(String.t(), String.t(), String.t()) ::
{:ok, map()} | {:error, :not_found | :cache_miss}
def find_by_previous_url_slug(group_slug, language, url_slug) do
case read(group_slug) do
{:ok, posts} -> find_post_by_previous_slug(posts, language, url_slug)
{:error, _} -> {:error, :cache_miss}
end
end
defp find_post_by_previous_slug(posts, language, url_slug) do
case Enum.find(posts, &post_has_previous_slug?(&1, language, url_slug)) do
nil -> {:error, :not_found}
post -> {:ok, post}
end
end
defp post_has_previous_slug?(post, language, url_slug) do
lang_previous_slugs = Map.get(post, :language_previous_slugs) || %{}
previous_for_lang = Map.get(lang_previous_slugs, language) || []
url_slug in previous_for_lang
end
@doc """
Finds a cached post by mode — uses date/time lookup for timestamp mode, slug for others.
"""
def find_post_by_mode(group_slug, post) do
mode = Map.get(post, :mode)
if mode in @timestamp_modes do
date = post[:date]
time = post[:time]
if date && time do
date_str = if is_struct(date, Date), do: Date.to_iso8601(date), else: to_string(date)
time_str = format_time_for_cache(time)
find_post_by_path(group_slug, date_str, time_str)
else
{:error, :not_found}
end
else
find_post(group_slug, post.slug)
end
end
defp format_time_for_cache(%Time{} = time) do
time |> Time.to_string() |> String.slice(0, 5)
end
defp format_time_for_cache(time) when is_binary(time), do: String.slice(time, 0, 5)
defp format_time_for_cache(_), do: ""
@doc """
Returns the :persistent_term key for a publishing group's cache.
"""
@spec persistent_term_key(String.t()) :: tuple()
def persistent_term_key(group_slug) do
{@persistent_term_prefix, group_slug}
end
@doc """
Returns the :persistent_term key for tracking when the memory cache was loaded.
"""
@spec loaded_at_key(String.t()) :: tuple()
def loaded_at_key(group_slug) do
{@persistent_term_loaded_at_prefix, group_slug}
end
@doc """
Returns when the memory cache was loaded (ISO 8601 string), or nil if not loaded.
"""
@spec memory_loaded_at(String.t()) :: String.t() | nil
def memory_loaded_at(group_slug) do
case safe_persistent_term_get(loaded_at_key(group_slug)) do
{:ok, loaded_at} -> loaded_at
:not_found -> nil
end
end
@doc """
Returns the :persistent_term key for tracking when the cache was last generated.
"""
@spec cache_generated_at_key(String.t()) :: tuple()
def cache_generated_at_key(group_slug) do
{@persistent_term_cache_generated_at_prefix, group_slug}
end
@doc """
Returns the timestamp of when the cache was last generated from the database.
"""
@spec cache_generated_at(String.t()) :: String.t() | nil
def cache_generated_at(group_slug) do
case safe_persistent_term_get(cache_generated_at_key(group_slug)) do
{:ok, generated_at} -> generated_at
:not_found -> nil
end
end
@doc """
Returns whether memory caching (:persistent_term) is enabled.
Uses cached settings to avoid database queries on every call.
"""
@spec memory_cache_enabled?() :: boolean()
def memory_cache_enabled? do
Settings.get_setting_cached(@memory_cache_key, "true") == "true"
end
@doc """
Returns a list of posts that need primary_language migration.
This checks all posts in a group and returns those that either:
1. Have no `primary_language` stored (need backfill)
2. Have `primary_language` different from global setting (need migration decision)
"""
@spec posts_needing_primary_language_migration(String.t()) :: [map()]
def posts_needing_primary_language_migration(group_slug) do
case read(group_slug) do
{:ok, posts} ->
global_primary = LanguageHelpers.get_primary_language()
Enum.filter(posts, fn post ->
# Use atom key since normalized posts use atoms
stored_primary = post[:primary_language]
stored_primary == nil or stored_primary != global_primary
end)
{:error, _} ->
# If cache doesn't exist, query DB directly
scan_posts_needing_migration(group_slug)
end
end
defp scan_posts_needing_migration(group_slug) do
global_primary = LanguageHelpers.get_primary_language()
DBStorage.list_posts_for_listing(group_slug)
|> Enum.filter(fn post ->
stored_primary = post[:primary_language]
stored_primary == nil or stored_primary != global_primary
end)
end
@doc """
Counts posts by primary_language status in a group.
Returns `%{current: n, needs_migration: n, needs_backfill: n}` where:
- `current` - posts with primary_language matching global setting
- `needs_migration` - posts with different primary_language (were created under old setting)
- `needs_backfill` - posts with no primary_language stored
"""
@spec count_primary_language_status(String.t()) :: map()
def count_primary_language_status(group_slug) do
case read(group_slug) do
{:ok, posts} ->
global_primary = LanguageHelpers.get_primary_language()
Enum.reduce(posts, %{current: 0, needs_migration: 0, needs_backfill: 0}, fn post, acc ->
# Use atom key since normalized posts use atoms
stored_primary = post[:primary_language]
cond do
stored_primary == nil ->
%{acc | needs_backfill: acc.needs_backfill + 1}
stored_primary == global_primary ->
%{acc | current: acc.current + 1}
true ->
%{acc | needs_migration: acc.needs_migration + 1}
end
end)
{:error, _} ->
# If cache doesn't exist, query DB directly
scan_primary_language_status(group_slug)
end
end
defp scan_primary_language_status(group_slug) do
global_primary = LanguageHelpers.get_primary_language()
DBStorage.list_posts_for_listing(group_slug)
|> Enum.reduce(%{current: 0, needs_migration: 0, needs_backfill: 0}, fn post, acc ->
stored_primary = post[:primary_language]
cond do
stored_primary == nil ->
%{acc | needs_backfill: acc.needs_backfill + 1}
stored_primary == global_primary ->
%{acc | current: acc.current + 1}
true ->
%{acc | needs_migration: acc.needs_migration + 1}
end
end)
end
end