Packages
phoenix_kit
1.7.185
1.7.208
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/storage/workers/sync_files_job.ex
defmodule PhoenixKit.Modules.Storage.Workers.SyncFilesJob do
@moduledoc """
Oban worker that syncs under-replicated files to meet the redundancy target.
Broadcasts progress via PubSub so the Health LiveView can display real-time
updates. Stores sync state in persistent_term so the UI survives page refreshes.
"""
use Oban.Worker, queue: :file_processing, max_attempts: 1
require Logger
alias PhoenixKit.Modules.Storage
alias PhoenixKit.PubSub.Manager, as: PubSubManager
alias PhoenixKit.Settings
@sync_topic "media:sync_progress"
@sync_state_key :phoenix_kit_media_sync_state
@sync_paused_key :phoenix_kit_media_sync_paused
@sync_stopped_key :phoenix_kit_media_sync_stopped
@impl Oban.Worker
def perform(%Oban.Job{}) do
redundancy_target =
Settings.get_setting("storage_redundancy_copies", "1")
|> String.to_integer()
clear_paused()
clear_stopped()
put_sync_state(%{done: 0, total: 0, synced: 0, failed: 0, status: :starting})
Storage.sync_under_replicated_with_progress(
redundancy_target,
fn progress ->
wait_while_paused()
put_sync_state(progress)
PubSubManager.broadcast(@sync_topic, {:sync_progress, progress})
end,
check_cancelled: &stopped?/0
)
if stopped?() do
Logger.info("Media sync stopped by admin")
state = get_sync_state() || %{done: 0, total: 0, synced: 0, failed: 0}
PubSubManager.broadcast(@sync_topic, {:sync_progress, Map.put(state, :status, :stopped)})
end
clear_sync_state()
clear_paused()
clear_stopped()
:ok
end
@impl Oban.Worker
def timeout(_job), do: :timer.minutes(30)
def pause, do: :persistent_term.put(@sync_paused_key, true)
def resume, do: :persistent_term.erase(@sync_paused_key)
def paused?, do: :persistent_term.get(@sync_paused_key, false) == true
def stop, do: :persistent_term.put(@sync_stopped_key, true)
def stopped?, do: :persistent_term.get(@sync_stopped_key, false) == true
defp wait_while_paused do
if paused?() and not stopped?() do
Process.sleep(500)
wait_while_paused()
end
end
defp put_sync_state(state) do
:persistent_term.put(@sync_state_key, state)
end
defp clear_sync_state do
:persistent_term.erase(@sync_state_key)
rescue
ArgumentError -> :ok
end
defp clear_paused do
:persistent_term.erase(@sync_paused_key)
rescue
ArgumentError -> :ok
end
defp clear_stopped do
:persistent_term.erase(@sync_stopped_key)
rescue
ArgumentError -> :ok
end
defp get_sync_state do
:persistent_term.get(@sync_state_key, nil)
rescue
ArgumentError -> nil
end
end