Packages
phoenix_kit
1.7.102
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/storage.ex
defmodule PhoenixKit.Modules.Storage do
@moduledoc """
Storage context for managing files, buckets, and dimensions.
Provides a distributed file storage system with support for multiple storage providers
(local filesystem, AWS S3, Backblaze B2, Cloudflare R2) with automatic redundancy
and failover capabilities.
## Features
- Multi-location storage with configurable redundancy (1-5 copies)
- Support for local, S3, B2, and R2 storage providers
- Automatic variant generation for images and videos
- Priority-based storage selection
- Built-in usage tracking and statistics
- PostgreSQL-backed file registry
## Module Status
This module is **always enabled** and cannot be disabled. It provides core
functionality for file management across PhoenixKit.
"""
use PhoenixKit.Module
import Ecto.Query, warn: false
require Logger
alias PhoenixKit.Utils.Date, as: UtilsDate
alias PhoenixKit.Modules.Storage.Bucket
alias PhoenixKit.Modules.Storage.Dimension
alias PhoenixKit.Modules.Storage.FileInstance
alias PhoenixKit.Modules.Storage.FileLocation
alias PhoenixKit.Modules.Storage.Folder
alias PhoenixKit.Modules.Storage.FolderLink
alias PhoenixKit.Modules.Storage.Manager
alias PhoenixKit.Modules.Storage.ProcessFileJob
alias PhoenixKit.Modules.Storage.ProviderRegistry
# NOTE: Temporary helper for Publishing component system.
# The dedicated storage/media APIs under development should replace this fallback once available.
alias PhoenixKit.Modules.Storage.URLSigner
alias PhoenixKit.Modules.Storage.VariantGenerator
alias PhoenixKit.Settings
@default_path "priv/uploads"
# ===== MODULE STATUS =====
@doc """
Checks if the Storage module is enabled.
This module is always enabled and cannot be disabled.
## Examples
iex> PhoenixKit.Modules.Storage.module_enabled?()
true
"""
def module_enabled?, do: true
# ===== BUCKETS =====
@doc """
Returns a list of all storage buckets, ordered by priority.
"""
def list_buckets do
Bucket
|> order_by(asc: :priority)
|> repo().all()
end
@doc """
Gets a single bucket by ID.
Returns `nil` if bucket does not exist.
"""
def get_bucket(id), do: repo().get(Bucket, id)
@doc """
Gets a bucket by name.
"""
def get_bucket_by_name(name) do
repo().get_by(Bucket, name: name)
end
@doc """
Gets enabled buckets, ordered by priority.
"""
def list_enabled_buckets do
Bucket
|> where([b], b.enabled == true)
|> order_by(asc: :priority)
|> repo().all()
end
@doc """
Creates a new bucket.
## Examples
iex> create_bucket(%{name: "Local Storage", provider: "local"})
{:ok, %Bucket{}}
iex> create_bucket(%{name: nil})
{:error, %Ecto.Changeset{}}
"""
def create_bucket(attrs \\ %{}) do
%Bucket{}
|> Bucket.changeset(attrs)
|> repo().insert()
end
@doc """
Updates a bucket.
## Examples
iex> update_bucket(bucket, %{name: "New Name"})
{:ok, %Bucket{}}
iex> update_bucket(bucket, %{name: nil})
{:error, %Ecto.Changeset{}}
"""
def update_bucket(%Bucket{} = bucket, attrs) do
bucket
|> Bucket.changeset(attrs)
|> repo().update()
end
@doc """
Deletes a bucket.
## Examples
iex> delete_bucket(bucket)
{:ok, %Bucket{}}
iex> delete_bucket(bucket)
{:error, %Ecto.Changeset{}}
"""
def delete_bucket(%Bucket{} = bucket) do
repo().delete(bucket)
end
@doc """
Returns an `%Ecto.Changeset{}` for tracking bucket changes.
"""
def change_bucket(%Bucket{} = bucket, attrs \\ %{}) do
Bucket.changeset(bucket, attrs)
end
@doc """
Tests connectivity for a bucket configuration.
Builds a temporary Bucket struct from the given params and delegates
to the appropriate provider's `test_connection/1` callback.
Returns `:ok` or `{:error, reason}`.
"""
def test_connection(bucket_params) when is_map(bucket_params) do
provider = bucket_params["provider"]
bucket = %Bucket{
provider: provider,
region: bucket_params["region"],
endpoint: bucket_params["endpoint"],
bucket_name: bucket_params["bucket_name"],
access_key_id: bucket_params["access_key_id"],
secret_access_key: bucket_params["secret_access_key"]
}
case ProviderRegistry.get_provider(provider) do
{:ok, provider_module} -> provider_module.test_connection(bucket)
{:error, reason} -> {:error, reason}
end
rescue
error -> {:error, "Connection test failed: #{Exception.message(error)}"}
end
@doc """
Calculates storage usage for a bucket in MB.
Returns total size of all files stored in this bucket by summing up all
file instances that have locations in this bucket.
"""
def calculate_bucket_usage(bucket_uuid) do
from(fl in FileLocation,
join: fi in FileInstance,
on: fl.file_instance_uuid == fi.uuid,
where: fl.bucket_uuid == ^bucket_uuid and fl.status == "active",
select: fragment("SUM(? / (1024 * 1024))", fi.size)
)
|> repo().one()
|> case do
nil -> 0
total -> Decimal.to_float(total)
end
end
@doc """
Returns a health report comparing file location counts against the redundancy target.
Groups by file (not instance) — a file is "under-replicated" if any of its
instances have fewer active locations than the redundancy target.
Returns a map with:
- `total` — total files
- `healthy` — files where all instances meet the redundancy target
- `under_replicated` — list of files with at least one under-replicated instance
- `health_percentage` — percentage of healthy files
"""
def get_health_report(redundancy_target) do
# Per-instance location counts, then aggregate per file
instance_counts_query =
from(fi in FileInstance,
left_join: fl in FileLocation,
on: fl.file_instance_uuid == fi.uuid and fl.status == "active",
where: fi.processing_status == "completed",
group_by: [fi.uuid, fi.file_uuid],
select: %{
file_uuid: fi.file_uuid,
location_count: count(fl.uuid)
}
)
instance_counts = repo().all(instance_counts_query)
# Group by file, take the minimum location count per file
file_min_counts =
instance_counts
|> Enum.group_by(& &1.file_uuid)
|> Enum.map(fn {file_uuid, instances} ->
min_count = Enum.min_by(instances, & &1.location_count).location_count
{file_uuid, min_count}
end)
# Load file details for under-replicated ones
under_replicated_uuids =
file_min_counts
|> Enum.filter(fn {_uuid, min_count} -> min_count < redundancy_target end)
|> Enum.map(fn {uuid, _} -> uuid end)
total = length(file_min_counts)
under_replicated_count = length(under_replicated_uuids)
healthy = total - under_replicated_count
under_replicated_files =
if under_replicated_uuids != [] do
min_counts_map = Map.new(file_min_counts)
from(f in PhoenixKit.Modules.Storage.File,
where: f.uuid in ^under_replicated_uuids,
order_by: [asc: f.original_file_name],
select: %{
file_uuid: f.uuid,
original_file_name: f.original_file_name,
file_type: f.file_type
}
)
|> repo().all()
|> Enum.map(fn file ->
Map.put(file, :min_location_count, Map.get(min_counts_map, file.file_uuid, 0))
end)
else
[]
end
health_percentage =
if total > 0, do: Float.round(healthy / total * 100, 1), else: 100.0
%{
total: total,
healthy: healthy,
under_replicated: under_replicated_files,
health_percentage: health_percentage,
redundancy_target: redundancy_target
}
rescue
error ->
Logger.error("Health report failed: #{inspect(error)}")
%{
total: 0,
healthy: 0,
under_replicated: [],
health_percentage: 100.0,
redundancy_target: redundancy_target
}
end
@doc """
Syncs under-replicated files to meet the redundancy target.
For each under-replicated file, retrieves it from an existing bucket
and replicates it to the missing buckets. Returns a summary of results.
"""
def sync_under_replicated(redundancy_target) do
enabled_buckets = list_enabled_buckets()
enabled_bucket_uuids = Enum.map(enabled_buckets, & &1.uuid)
buckets_by_uuid = Map.new(enabled_buckets, &{&1.uuid, &1})
# Get all instances with their location counts and existing bucket UUIDs
instance_data =
from(fi in FileInstance,
left_join: fl in FileLocation,
on: fl.file_instance_uuid == fi.uuid and fl.status == "active",
where: fi.processing_status == "completed",
group_by: [fi.uuid, fi.file_name],
having: count(fl.uuid) < ^redundancy_target,
select: %{
instance_uuid: fi.uuid,
file_name: fi.file_name,
location_count: count(fl.uuid)
}
)
|> repo().all()
results =
Enum.map(instance_data, fn item ->
# Get existing bucket UUIDs for this instance
existing_bucket_uuids = get_file_instance_bucket_uuids(item.instance_uuid)
# Find missing buckets (enabled but no location for this instance)
missing_bucket_uuids =
enabled_bucket_uuids
|> Enum.filter(&(&1 not in existing_bucket_uuids))
|> Enum.take(redundancy_target - item.location_count)
missing_buckets =
Enum.map(missing_bucket_uuids, &Map.get(buckets_by_uuid, &1))
|> Enum.reject(&is_nil/1)
if missing_buckets == [] do
{:skip, item.instance_uuid}
else
case Manager.replicate_to_buckets(item.file_name, missing_buckets) do
{:ok, storage_info} ->
create_file_locations_for_instance(
item.instance_uuid,
storage_info.bucket_ids,
item.file_name
)
{:ok, item.instance_uuid, length(storage_info.bucket_ids)}
{:error, reason} ->
Logger.warning("Sync failed for instance #{item.instance_uuid}: #{reason}")
{:error, item.instance_uuid, reason}
end
end
end)
synced = Enum.count(results, &match?({:ok, _, _}, &1))
failed = Enum.count(results, &match?({:error, _, _}, &1))
skipped = Enum.count(results, &match?({:skip, _}, &1))
%{synced: synced, failed: failed, skipped: skipped, total: length(results)}
end
@doc """
Syncs under-replicated files with progress reporting via callback.
The callback receives a map with `:done`, `:total`, `:synced`, `:failed`,
and `:status` (`:in_progress` or `:complete`) after each file is processed.
"""
def sync_under_replicated_with_progress(redundancy_target, callback, opts \\ []) do
enabled_buckets = list_enabled_buckets()
enabled_bucket_uuids = Enum.map(enabled_buckets, & &1.uuid)
buckets_by_uuid = Map.new(enabled_buckets, &{&1.uuid, &1})
# Get under-replicated instances grouped by file
instance_data =
from(fi in FileInstance,
join: f in PhoenixKit.Modules.Storage.File,
on: f.uuid == fi.file_uuid,
left_join: fl in FileLocation,
on: fl.file_instance_uuid == fi.uuid and fl.status == "active",
where: fi.processing_status == "completed",
group_by: [fi.uuid, fi.file_name, fi.file_uuid, f.original_file_name],
having: count(fl.uuid) < ^redundancy_target,
select: %{
instance_uuid: fi.uuid,
file_uuid: fi.file_uuid,
file_name: fi.file_name,
original_file_name: f.original_file_name,
location_count: count(fl.uuid)
}
)
|> repo().all()
# Group by file so progress tracks files, not instances
files_with_instances =
instance_data
|> Enum.group_by(& &1.file_uuid)
|> Enum.to_list()
total = length(files_with_instances)
check_cancelled = Keyword.get(opts, :check_cancelled, fn -> false end)
sync_ctx = %{
enabled_bucket_uuids: enabled_bucket_uuids,
buckets_by_uuid: buckets_by_uuid,
redundancy_target: redundancy_target
}
{synced, failed} =
Enum.reduce_while(Enum.with_index(files_with_instances, 1), {0, 0}, fn {{_file_uuid,
instances}, index},
{synced_acc,
failed_acc} ->
if check_cancelled.() do
{:halt, {synced_acc, failed_acc}}
else
{new_synced, new_failed, log_entry} =
sync_file_instances(instances, synced_acc, failed_acc, sync_ctx, check_cancelled)
callback.(%{
done: index,
total: total,
synced: new_synced,
failed: new_failed,
log: log_entry,
status: :in_progress
})
{:cont, {new_synced, new_failed}}
end
end)
callback.(%{
done: total,
total: total,
synced: synced,
failed: failed,
log: nil,
status: :complete
})
%{synced: synced, failed: failed, total: total}
end
defp sync_file_instances(instances, synced_acc, failed_acc, ctx, check_cancelled) do
file_name = List.first(instances)[:original_file_name] || List.first(instances)[:file_name]
if check_cancelled.() do
{synced_acc, failed_acc, %{file: file_name, status: :error, message: "Cancelled"}}
else
instance_results =
Enum.map(instances, fn item ->
if check_cancelled.() do
{:error, "Cancelled"}
else
sync_instance(
item,
ctx.enabled_bucket_uuids,
ctx.buckets_by_uuid,
ctx.redundancy_target
)
end
end)
if Enum.all?(instance_results, &match?({:ok, _}, &1)) do
{synced_acc + 1, failed_acc,
%{file: file_name, status: :ok, message: "Synced successfully"}}
else
errors =
instance_results
|> Enum.filter(&match?({:error, _}, &1))
|> Enum.reject(&(&1 == {:error, "Cancelled"}))
|> Enum.map_join("; ", fn {:error, reason} -> reason end)
{synced_acc, failed_acc + 1, %{file: file_name, status: :error, message: errors}}
end
end
end
defp sync_instance(item, enabled_bucket_uuids, buckets_by_uuid, redundancy_target) do
existing = get_file_instance_bucket_uuids(item.instance_uuid)
missing_uuids =
enabled_bucket_uuids
|> Enum.filter(&(&1 not in existing))
|> Enum.take(redundancy_target - item.location_count)
missing_buckets =
Enum.map(missing_uuids, &Map.get(buckets_by_uuid, &1))
|> Enum.reject(&is_nil/1)
if missing_buckets == [] do
{:ok, :already_synced}
else
case Manager.replicate_to_buckets(item.file_name, missing_buckets) do
{:ok, storage_info} ->
create_file_locations_for_instance(
item.instance_uuid,
storage_info.bucket_ids,
item.file_name
)
{:ok, :synced}
{:error, reason} ->
Logger.warning("Sync failed for instance #{item.instance_uuid}: #{reason}")
{:error, to_string(reason)}
end
end
end
@doc """
Calculates free space for a bucket.
For local storage, checks actual disk space.
For cloud storage, returns the configured max_size_mb minus usage.
"""
def calculate_bucket_free_space(%Bucket{} = bucket) do
used_mb = calculate_bucket_usage(bucket.uuid)
case bucket.provider do
"local" ->
calculate_local_free_space(bucket)
_ ->
# For cloud storage, use the configured max size
max(bucket.max_size_mb - used_mb, 0)
end
end
def calculate_bucket_free_space(bucket_uuid) when is_binary(bucket_uuid) do
bucket = get_bucket(bucket_uuid)
if bucket, do: calculate_bucket_free_space(bucket), else: 0
end
# ===== DIMENSIONS =====
@doc """
Returns a list of all dimensions, ordered by size (width x height).
"""
def list_dimensions do
Dimension
|> order_by(asc: :width, asc: :height)
|> repo().all()
end
@doc """
Returns enabled dimensions for a specific file type.
"""
def list_dimensions_for_type(file_type) when file_type in ["image", "video"] do
Dimension
|> where([d], d.enabled == true and (d.applies_to == ^file_type or d.applies_to == "both"))
|> order_by(asc: :width, asc: :height)
|> repo().all()
end
def list_dimensions_for_type(_), do: []
@doc """
Gets a single dimension by ID.
"""
def get_dimension(id), do: repo().get(Dimension, id)
@doc """
Gets a dimension by name.
"""
def get_dimension_by_name(name) do
repo().get_by(Dimension, name: name)
end
@doc """
Resets all dimensions to default seeded values.
Deletes all current dimensions and recreates the 8 default ones.
"""
def reset_dimensions_to_defaults do
repo().transaction(fn ->
# Delete all existing dimensions
repo().delete_all(Dimension)
# Insert default dimensions
now = UtilsDate.utc_now()
default_dimensions = [
# Image dimensions
%{
name: "thumbnail",
width: 150,
height: 150,
quality: 85,
format: "jpg",
applies_to: "image",
enabled: true,
order: 1,
alternative_formats: [],
inserted_at: now,
updated_at: now
},
%{
name: "small",
width: 300,
height: 300,
quality: 85,
format: "jpg",
applies_to: "image",
enabled: true,
alternative_formats: [],
order: 2,
inserted_at: now,
updated_at: now
},
%{
name: "medium",
width: 800,
height: 600,
quality: 85,
format: "jpg",
applies_to: "image",
enabled: true,
alternative_formats: [],
order: 3,
inserted_at: now,
updated_at: now
},
%{
name: "large",
width: 1920,
height: 1080,
quality: 85,
format: "jpg",
applies_to: "image",
enabled: true,
alternative_formats: [],
order: 4,
inserted_at: now,
updated_at: now
},
# Video dimensions
%{
name: "360p",
width: 640,
height: 360,
quality: 28,
format: "mp4",
applies_to: "video",
enabled: true,
alternative_formats: [],
order: 5,
inserted_at: now,
updated_at: now
},
%{
name: "720p",
width: 1280,
height: 720,
quality: 28,
format: "mp4",
applies_to: "video",
enabled: true,
alternative_formats: [],
order: 6,
inserted_at: now,
updated_at: now
},
%{
name: "1080p",
width: 1920,
height: 1080,
quality: 28,
format: "mp4",
applies_to: "video",
enabled: true,
alternative_formats: [],
order: 7,
inserted_at: now,
updated_at: now
},
%{
name: "video_thumbnail",
width: 640,
height: 360,
quality: 85,
format: "jpg",
applies_to: "video",
enabled: true,
alternative_formats: [],
order: 8,
inserted_at: now,
updated_at: now
}
]
# Insert all default dimensions
Enum.each(default_dimensions, fn dim ->
%Dimension{}
|> Dimension.changeset(dim)
|> repo().insert!()
end)
end)
end
@doc """
Repairs the storage module by resetting configuration to defaults.
This is a safe, non-destructive operation that:
1. Creates a default local bucket if no buckets exist
2. Resets dimensions to 8 defaults (4 image + 4 video)
3. Resets storage settings to recommended defaults
All existing files are preserved.
## Returns
- `{:ok, repairs}` - List of repairs performed
- `{:error, reason}` - If repair failed
## Examples
iex> repair_storage_module()
{:ok, [{:bucket_created, "Local Storage"}, {:dimensions_reset, 8}, {:settings_reset, 3}]}
"""
def repair_storage_module do
repo().transaction(fn ->
repairs = []
# 1. Ensure at least one bucket exists
repairs =
case ensure_default_bucket_exists() do
{:created, bucket} -> [{:bucket_created, bucket.name} | repairs]
:exists -> repairs
end
# 2. Reset dimensions to defaults
case reset_dimensions_to_defaults() do
{:ok, _} -> :ok
{:error, reason} -> repo().rollback(reason)
end
repairs = [{:dimensions_reset, 8} | repairs]
# 3. Reset settings to defaults
reset_settings_to_defaults()
repairs = [{:settings_reset, 3} | repairs]
Enum.reverse(repairs)
end)
end
@doc """
Ensures at least one default bucket exists.
If no buckets exist, creates a default local storage bucket.
## Returns
- `{:created, bucket}` - If a new bucket was created
- `:exists` - If buckets already exist
## Examples
iex> ensure_default_bucket_exists()
{:created, %Bucket{name: "Local Storage"}}
iex> ensure_default_bucket_exists()
:exists
"""
def ensure_default_bucket_exists do
if Enum.empty?(list_buckets()) do
{:ok, bucket} =
create_bucket(%{
name: "Local Storage",
provider: "local",
endpoint: "priv/media",
enabled: true,
priority: 0
})
{:created, bucket}
else
:exists
end
end
@doc """
Resets storage settings to their default values.
Resets:
- `storage_redundancy_copies` to "1"
- `storage_auto_generate_variants` to "true"
- `storage_default_bucket_uuid` to nil
## Returns
- `:ok`
"""
def reset_settings_to_defaults do
Settings.update_setting("storage_redundancy_copies", "1")
Settings.update_setting("storage_auto_generate_variants", "true")
Settings.update_setting("storage_default_bucket_uuid", nil)
:ok
end
@doc """
Creates a new dimension.
## Examples
iex> create_dimension(%{name: "thumbnail", width: 150, height: 150})
{:ok, %Dimension{}}
iex> create_dimension(%{name: nil})
{:error, %Ecto.Changeset{}}
"""
def create_dimension(attrs \\ %{}) do
%Dimension{}
|> Dimension.changeset(attrs)
|> repo().insert()
end
@doc """
Updates a dimension.
## Examples
iex> update_dimension(dimension, %{name: "New Name"})
{:ok, %Dimension{}}
iex> update_dimension(dimension, %{name: nil})
{:error, %Ecto.Changeset{}}
"""
def update_dimension(%Dimension{} = dimension, attrs) do
dimension
|> Dimension.changeset(attrs)
|> repo().update()
end
@doc """
Deletes a dimension.
## Examples
iex> delete_dimension(dimension)
{:ok, %Dimension{}}
iex> delete_dimension(dimension)
{:error, %Ecto.Changeset{}}
"""
def delete_dimension(%Dimension{} = dimension) do
repo().delete(dimension)
end
@doc """
Returns an `%Ecto.Changeset{}` for tracking dimension changes.
"""
def change_dimension(%Dimension{} = dimension, attrs \\ %{}) do
Dimension.changeset(dimension, attrs)
end
# ===== FOLDERS =====
@doc "Returns all folders as a flat list ordered by name, for building a tree."
def list_all_folders do
from(f in Folder, order_by: [asc: f.name])
|> repo().all()
end
@doc "Builds a folder tree structure from a flat list of folders."
def build_folder_tree(folders) do
by_parent = Enum.group_by(folders, & &1.parent_uuid)
build_tree_nodes(by_parent, nil)
end
defp build_tree_nodes(by_parent, parent_uuid) do
(Map.get(by_parent, parent_uuid) || [])
|> Enum.map(fn folder ->
%{folder: folder, children: build_tree_nodes(by_parent, folder.uuid)}
end)
end
@doc """
Returns folder tree rooted at scope_folder_id (exclusive of scope itself).
For nil scope, returns the real-root tree.
"""
def list_folder_tree(scope_folder_id \\ nil) do
all_folders = list_all_folders()
by_parent = Enum.group_by(all_folders, & &1.parent_uuid)
if scope_folder_id do
build_tree_nodes(by_parent, scope_folder_id)
else
build_tree_nodes(by_parent, nil)
end
end
@doc """
Lists folders within a parent folder (nil = root).
When parent_uuid is nil and scope_folder_id is set, returns children of
scope_folder_id instead of real root.
"""
def list_folders(parent_uuid \\ nil, scope_folder_id \\ nil)
def list_folders(nil, nil) do
from(f in Folder, where: is_nil(f.parent_uuid), order_by: [asc: f.name])
|> repo().all()
end
def list_folders(nil, scope_folder_id) do
from(f in Folder, where: f.parent_uuid == ^scope_folder_id, order_by: [asc: f.name])
|> repo().all()
end
def list_folders(parent_uuid, _scope_folder_id) do
from(f in Folder, where: f.parent_uuid == ^parent_uuid, order_by: [asc: f.name])
|> repo().all()
end
@doc "Gets a single folder by UUID."
def get_folder(nil), do: nil
def get_folder(uuid), do: repo().get(Folder, uuid)
@doc """
Creates a new folder.
When scope_folder_id is set:
- If attrs.parent_uuid is outside scope, returns `{:error, :out_of_scope}`.
- If attrs.parent_uuid is nil, rewrites to scope_folder_id (new folder at scope root).
"""
def create_folder(attrs, scope_folder_id \\ nil)
def create_folder(attrs, nil) do
%Folder{}
|> Folder.changeset(attrs)
|> repo().insert()
end
def create_folder(attrs, scope_folder_id) do
parent_uuid = attrs[:parent_uuid] || attrs["parent_uuid"]
cond do
is_nil(parent_uuid) ->
attrs = Map.put(attrs, :parent_uuid, scope_folder_id)
%Folder{} |> Folder.changeset(attrs) |> repo().insert()
within_scope?(parent_uuid, scope_folder_id) ->
%Folder{} |> Folder.changeset(attrs) |> repo().insert()
true ->
{:error, :out_of_scope}
end
end
@doc """
Updates a folder (rename, color change, move).
Returns `{:error, :cycle}` if the move would create a circular reference.
Returns `{:error, :out_of_scope}` if the folder or new parent is outside scope.
"""
def update_folder(folder, attrs, scope_folder_id \\ nil)
def update_folder(%Folder{} = folder, attrs, nil) do
new_parent = attrs[:parent_uuid] || attrs["parent_uuid"]
if new_parent && new_parent != folder.parent_uuid && ancestor_of?(folder.uuid, new_parent) do
{:error, :cycle}
else
folder
|> Folder.changeset(attrs)
|> repo().update()
end
end
def update_folder(%Folder{} = folder, attrs, scope_folder_id) do
if within_scope?(folder.uuid, scope_folder_id) do
new_parent = attrs[:parent_uuid] || attrs["parent_uuid"]
cond do
new_parent && not within_scope?(new_parent, scope_folder_id) ->
{:error, :out_of_scope}
new_parent && new_parent != folder.parent_uuid && ancestor_of?(folder.uuid, new_parent) ->
{:error, :cycle}
true ->
folder
|> Folder.changeset(attrs)
|> repo().update()
end
else
{:error, :out_of_scope}
end
end
@doc """
Deletes a folder.
Moves child folders and home files to the deleted folder's parent.
Folder links are cascade-deleted by the database FK.
Returns `{:error, :out_of_scope}` if the folder is outside scope.
"""
def delete_folder(folder, scope_folder_id \\ nil)
def delete_folder(%Folder{} = folder, scope_folder_id) when not is_nil(scope_folder_id) do
if within_scope?(folder.uuid, scope_folder_id) do
do_delete_folder(folder)
else
{:error, :out_of_scope}
end
end
def delete_folder(%Folder{} = folder, nil) do
do_delete_folder(folder)
end
defp do_delete_folder(%Folder{} = folder) do
repo().transaction(fn ->
# Move child folders to parent
from(f in Folder, where: f.parent_uuid == ^folder.uuid)
|> repo().update_all(set: [parent_uuid: folder.parent_uuid])
# Move home files to parent
from(f in PhoenixKit.Modules.Storage.File, where: f.folder_uuid == ^folder.uuid)
|> repo().update_all(set: [folder_uuid: folder.parent_uuid])
# Delete folder (links cascade via FK)
case repo().delete(folder) do
{:ok, deleted} -> deleted
{:error, changeset} -> repo().rollback(changeset)
end
end)
end
@doc """
Returns the ancestor chain from root to the given folder (for breadcrumbs).
When scope_folder_id is set, the chain stops before scope (scope itself not included —
it is the virtual root).
"""
def folder_breadcrumbs(folder_uuid, scope_folder_id \\ nil)
def folder_breadcrumbs(folder_uuid, nil) do
do_folder_breadcrumbs(folder_uuid, 50)
end
def folder_breadcrumbs(folder_uuid, scope_folder_id) do
folder_uuid
|> do_folder_breadcrumbs(50)
|> Enum.drop_while(fn f -> f.uuid != scope_folder_id end)
|> case do
[] -> []
[_ | rest] -> rest
end
end
defp do_folder_breadcrumbs(nil, _limit), do: []
defp do_folder_breadcrumbs(_uuid, 0), do: []
defp do_folder_breadcrumbs(folder_uuid, limit) do
case get_folder(folder_uuid) do
nil -> []
folder -> do_folder_breadcrumbs(folder.parent_uuid, limit - 1) ++ [folder]
end
end
@doc "Returns true if `folder_uuid` is an ancestor of `target_uuid`."
def ancestor_of?(_folder_uuid, nil), do: false
def ancestor_of?(folder_uuid, target_uuid) do
ancestor_of?(folder_uuid, target_uuid, 50)
end
defp ancestor_of?(_folder_uuid, nil, _limit), do: false
defp ancestor_of?(_folder_uuid, _target_uuid, 0), do: false
defp ancestor_of?(folder_uuid, target_uuid, limit) do
case get_folder(target_uuid) do
nil -> false
%{uuid: ^folder_uuid} -> true
target -> ancestor_of?(folder_uuid, target.parent_uuid, limit - 1)
end
end
@doc """
Returns true if folder_uuid is within the given scope.
- When scope_folder_id is nil, always returns true (no scope restriction).
- When folder_uuid equals scope_folder_id, returns true (scope is the virtual root).
- When scope_folder_id is an ancestor of folder_uuid, returns true (folder is a descendant).
- Returns false otherwise, including when folder_uuid is nil and scope is set
(real root is outside any non-nil scope).
"""
def within_scope?(_folder_uuid, nil), do: true
def within_scope?(folder_uuid, scope_folder_id) when folder_uuid == scope_folder_id, do: true
def within_scope?(folder_uuid, scope_folder_id), do: ancestor_of?(scope_folder_id, folder_uuid)
@doc "Counts files in a folder (home files + linked files)."
def count_folder_contents(nil) do
home =
from(f in PhoenixKit.Modules.Storage.File, where: is_nil(f.folder_uuid), select: count())
|> repo().one()
home || 0
end
def count_folder_contents(folder_uuid) do
home =
from(f in PhoenixKit.Modules.Storage.File,
where: f.folder_uuid == ^folder_uuid,
select: count()
)
|> repo().one()
links =
from(fl in FolderLink, where: fl.folder_uuid == ^folder_uuid, select: count())
|> repo().one()
(home || 0) + (links || 0)
end
@doc """
Lists files within the given scope with optional folder filter, search, and pagination.
## Options
- `:folder_uuid` — specific folder within scope; returns `{:error, :out_of_scope}` if outside.
- `:search` — ilike search on original_file_name; restricted to scope descendants when scope set.
- `:include_orphaned` — boolean (default false); only meaningful when scope is nil.
When true, returns only files with folder_uuid IS NULL.
`include_orphaned: true` is ignored when `scope_folder_id` is non-nil (orphans are always outside any scope).
- `:page` — page number (default 1).
- `:per_page` — page size (default 20).
## Returns
`{files, total_count}` or `{:error, :out_of_scope}`.
When `scope_folder_id == nil` and no `folder_uuid` is specified, ALL files are returned (not
just orphans). To fetch orphans only at real root, pass `include_orphaned: true` AND use a
dedicated orphan-only branch. Task 4 callers must preserve the current `/admin/media` behavior
(list orphans only when `filter_orphaned` is on) via a separate code path.
"""
def list_files_in_scope(scope_folder_id, opts \\ []) do
folder_uuid = opts[:folder_uuid]
search = opts[:search]
include_orphaned = Keyword.get(opts, :include_orphaned, false)
page = Keyword.get(opts, :page, 1)
per_page = Keyword.get(opts, :per_page, 20)
if not is_nil(folder_uuid) and not is_nil(scope_folder_id) and
not within_scope?(folder_uuid, scope_folder_id) do
{:error, :out_of_scope}
else
query =
build_scope_file_query(scope_folder_id, folder_uuid, search, include_orphaned)
|> where([f], f.status != "trashed")
total = repo().aggregate(query, :count, :uuid)
files =
query
|> order_by([f], desc: f.inserted_at)
|> offset(^((page - 1) * per_page))
|> limit(^per_page)
|> repo().all()
{files, total}
end
end
defp build_scope_file_query(nil, nil, nil, false) do
from(f in PhoenixKit.Modules.Storage.File)
end
defp build_scope_file_query(nil, nil, nil, true) do
from(f in PhoenixKit.Modules.Storage.File, where: is_nil(f.folder_uuid))
end
defp build_scope_file_query(nil, folder_uuid, search, _orphaned) when not is_nil(folder_uuid) do
from(f in PhoenixKit.Modules.Storage.File, where: f.folder_uuid == ^folder_uuid)
|> apply_file_search(search)
end
defp build_scope_file_query(nil, nil, search, _orphaned) when not is_nil(search) do
from(f in PhoenixKit.Modules.Storage.File)
|> apply_file_search(search)
end
defp build_scope_file_query(scope_folder_id, folder_uuid, search, _orphaned) do
cond do
folder_uuid ->
# Specific folder already validated within scope — no CTE needed
from(f in PhoenixKit.Modules.Storage.File, where: f.folder_uuid == ^folder_uuid)
search && search != "" ->
# Search: walk the full scope subtree via recursive CTE
scope_subtree_query(scope_folder_id)
true ->
# Scope root without search: show only direct children (files directly in scope)
from(f in PhoenixKit.Modules.Storage.File, where: f.folder_uuid == ^scope_folder_id)
end
|> apply_file_search(search)
end
defp scope_subtree_query(scope_folder_id) do
cte_base =
from(f in Folder,
where: f.uuid == ^scope_folder_id,
select: %{uuid: f.uuid}
)
cte_recursive =
from(f in Folder,
join: d in "scope_descendants",
on: f.parent_uuid == d.uuid,
select: %{uuid: f.uuid}
)
cte = union_all(cte_base, ^cte_recursive)
from(f in PhoenixKit.Modules.Storage.File,
join: d in "scope_descendants",
on: f.folder_uuid == d.uuid,
select: f
)
|> recursive_ctes(true)
|> with_cte("scope_descendants", as: ^cte)
end
defp apply_file_search(query, nil), do: query
defp apply_file_search(query, ""), do: query
defp apply_file_search(query, search) do
term = "%#{search}%"
where(
query,
[f],
ilike(f.original_file_name, ^term) or
fragment("CAST(? AS TEXT) ILIKE ?", f.uuid, ^term)
)
end
@doc "Moves a file's home folder."
def move_file_to_folder(file_uuid, target_folder_uuid, scope_folder_id \\ nil)
def move_file_to_folder(file_uuid, target_folder_uuid, nil) do
file = repo().get(PhoenixKit.Modules.Storage.File, file_uuid)
if file do
file
|> Ecto.Changeset.change(%{folder_uuid: target_folder_uuid})
|> repo().update()
else
{:error, :not_found}
end
end
def move_file_to_folder(file_uuid, target_folder_uuid, scope_folder_id) do
file = repo().get(PhoenixKit.Modules.Storage.File, file_uuid)
cond do
is_nil(file) ->
{:error, :not_found}
not within_scope?(file.folder_uuid, scope_folder_id) ->
{:error, :out_of_scope}
not within_scope?(target_folder_uuid, scope_folder_id) ->
{:error, :out_of_scope}
true ->
file
|> Ecto.Changeset.change(%{folder_uuid: target_folder_uuid})
|> repo().update()
end
end
@doc "Creates a link (shortcut) of a file in a folder."
def create_folder_link(folder_uuid, file_uuid, scope_folder_id \\ nil)
def create_folder_link(folder_uuid, file_uuid, nil) do
%FolderLink{}
|> FolderLink.changeset(%{folder_uuid: folder_uuid, file_uuid: file_uuid})
|> repo().insert()
end
def create_folder_link(folder_uuid, file_uuid, scope_folder_id) do
file = repo().get(PhoenixKit.Modules.Storage.File, file_uuid)
cond do
not within_scope?(folder_uuid, scope_folder_id) ->
{:error, :out_of_scope}
is_nil(file) ->
{:error, :not_found}
not within_scope?(file.folder_uuid, scope_folder_id) ->
{:error, :out_of_scope}
true ->
%FolderLink{}
|> FolderLink.changeset(%{folder_uuid: folder_uuid, file_uuid: file_uuid})
|> repo().insert()
end
end
@doc "Removes a folder link."
def delete_folder_link(link_uuid) do
case repo().get(FolderLink, link_uuid) do
nil -> {:error, :not_found}
link -> repo().delete(link)
end
end
# ===== FILES =====
@doc """
Returns a list of files, optionally filtered by bucket.
## Options
- `:bucket_uuid` - Filter by bucket UUID
- `:limit` - Maximum number of results
- `:offset` - Number of results to skip
- `:order_by` - Ordering (default: `[desc: :inserted_at]`)
"""
def list_files(opts \\ []) do
PhoenixKit.Modules.Storage.File
|> where([f], f.status != "trashed")
|> maybe_filter_by_bucket(opts[:bucket_uuid])
|> maybe_order_by(opts[:order_by])
|> maybe_limit(opts[:limit])
|> maybe_offset(opts[:offset])
|> repo().all()
end
@doc """
Gets a single file by ID.
"""
def get_file(id) when is_binary(id),
do: repo().get(PhoenixKit.Modules.Storage.File, id)
@doc """
Calculates user-specific file checksum (salted with user_uuid).
This creates a unique checksum per user+file combination for duplicate detection,
while preserving the original file checksum for popularity queries.
## Parameters
- user_uuid: The user UUID
- file_checksum: The SHA256 checksum of the file content
## Returns
String representing the SHA256 checksum of "user_uuid + file_checksum"
"""
def calculate_user_file_checksum(user_uuid, file_checksum) do
"#{user_uuid}#{file_checksum}"
|> then(fn data -> :crypto.hash(:sha256, data) end)
|> Base.encode16(case: :lower)
end
@doc """
Gets a file by its user-specific checksum.
This checks for duplicates for a specific user.
"""
def get_file_by_user_checksum(user_file_checksum) do
repo().get_by(PhoenixKit.Modules.Storage.File, user_file_checksum: user_file_checksum)
end
@doc """
Gets a file by its original content checksum (file_checksum).
This can find files uploaded by any user with the same content.
Useful for popularity queries.
"""
def get_file_by_checksum(file_checksum) do
repo().get_by(PhoenixKit.Modules.Storage.File, file_checksum: file_checksum)
end
@doc """
Creates a new file record.
This only creates the database record. Use `store_file/4` to actually
store the file data in storage buckets.
"""
def create_file(attrs \\ %{}) do
%PhoenixKit.Modules.Storage.File{}
|> PhoenixKit.Modules.Storage.File.changeset(attrs)
|> repo().insert()
end
@doc """
Updates a file.
"""
def update_file(%PhoenixKit.Modules.Storage.File{} = file, attrs) do
file
|> PhoenixKit.Modules.Storage.File.changeset(attrs)
|> repo().update()
end
@doc """
Deletes a file.
This only removes the database record. Use `delete_file_data/1` to
remove the actual file data from storage buckets.
"""
def delete_file(%PhoenixKit.Modules.Storage.File{} = file) do
repo().delete(file)
end
@doc """
Returns an `%Ecto.Changeset{}` for tracking file changes.
"""
def change_file(%PhoenixKit.Modules.Storage.File{} = file, attrs \\ %{}) do
PhoenixKit.Modules.Storage.File.changeset(file, attrs)
end
# ===== ORPHAN DETECTION =====
@doc """
Returns a list of orphaned files (files not referenced by any known entity).
## Options
- `:limit` - Maximum number of results
- `:offset` - Number of results to skip
"""
def find_orphaned_files(opts \\ []) do
orphaned_files_query()
|> order_by([f], desc: f.inserted_at)
|> maybe_limit(opts[:limit])
|> maybe_offset(opts[:offset])
|> repo().all()
end
@doc """
Returns the count of orphaned files.
When scope_folder_id is set, returns 0 because orphaned files (folder_uuid IS NULL)
are always outside any non-nil scope.
"""
def count_orphaned_files(scope_folder_id \\ nil)
def count_orphaned_files(scope_folder_id) when not is_nil(scope_folder_id), do: 0
def count_orphaned_files(nil) do
orphaned_files_query()
|> repo().aggregate(:count, :uuid)
end
@doc """
Returns true if the given file UUID is not referenced by any known entity.
"""
def file_orphaned?(file_uuid) when is_binary(file_uuid) do
orphaned_files_query()
|> where([f], f.uuid == ^file_uuid)
|> repo().exists?()
end
@doc """
Queues a list of file UUIDs for orphan cleanup via Oban.
Each file is scheduled for deletion after a 60-second delay to protect
against race conditions (another entity may reference the file).
Only files that are still orphaned at job execution time will be deleted.
"""
def queue_file_cleanup(file_uuids) when is_list(file_uuids) do
alias PhoenixKit.Modules.Storage.Workers.DeleteOrphanedFileJob
file_uuids
|> Enum.reject(&is_nil/1)
|> Enum.uniq()
|> Enum.each(fn uuid ->
%{"file_uuid" => uuid}
|> DeleteOrphanedFileJob.new(schedule_in: 60)
|> Oban.insert()
end)
end
defp orphaned_files_query do
existing = existing_optional_tables()
protected_uuids = protected_file_uuids()
# Core check — phoenix_kit_users always exists; exclude trashed files
base =
from(f in PhoenixKit.Modules.Storage.File,
where:
f.status != "trashed" and
fragment(
"NOT EXISTS (SELECT 1 FROM phoenix_kit_users u WHERE u.custom_fields->>'avatar_file_uuid' = ?::text)",
f.uuid
)
)
# Exclude UUIDs that the parent app has explicitly marked as protected
base =
if protected_uuids == [] do
base
else
where(base, [f], f.uuid not in ^protected_uuids)
end
# Optional module tables — only included if the table exists
optional_checks = [
{"phoenix_kit_post_media",
dynamic(
[f],
fragment(
"NOT EXISTS (SELECT 1 FROM phoenix_kit_post_media pm WHERE pm.file_uuid = ?)",
f.uuid
)
)},
{"phoenix_kit_ticket_attachments",
dynamic(
[f],
fragment(
"NOT EXISTS (SELECT 1 FROM phoenix_kit_ticket_attachments ta WHERE ta.file_uuid = ?)",
f.uuid
)
)},
{"phoenix_kit_post_groups",
dynamic(
[f],
fragment(
"NOT EXISTS (SELECT 1 FROM phoenix_kit_post_groups pg WHERE pg.cover_image_uuid = ?)",
f.uuid
)
)},
{"phoenix_kit_shop_products",
dynamic(
[f],
fragment(
"NOT EXISTS (SELECT 1 FROM phoenix_kit_shop_products sp WHERE sp.featured_image_uuid = ?) AND NOT EXISTS (SELECT 1 FROM phoenix_kit_shop_products sp WHERE ? = ANY(sp.image_uuids)) AND NOT EXISTS (SELECT 1 FROM phoenix_kit_shop_products sp WHERE sp.file_uuid = ?)",
f.uuid,
f.uuid,
f.uuid
)
)},
{"phoenix_kit_shop_categories",
dynamic(
[f],
fragment(
"NOT EXISTS (SELECT 1 FROM phoenix_kit_shop_categories sc WHERE sc.image_uuid = ?)",
f.uuid
)
)},
{"phoenix_kit_publishing_contents",
dynamic(
[f],
fragment(
"NOT EXISTS (SELECT 1 FROM phoenix_kit_publishing_contents pc WHERE pc.data->>'featured_image_uuid' = ?::text)",
f.uuid
)
)},
{"phoenix_kit_publishing_versions",
dynamic(
[f],
fragment(
"NOT EXISTS (SELECT 1 FROM phoenix_kit_publishing_versions pv WHERE pv.data->>'featured_image_uuid' = ?::text)",
f.uuid
)
)},
{"phoenix_kit_posts",
dynamic(
[f],
fragment(
"NOT EXISTS (SELECT 1 FROM phoenix_kit_posts p WHERE p.metadata->>'featured_image_uuid' = ?::text)",
f.uuid
)
)},
{"phoenix_kit_entity_data",
dynamic(
[f],
fragment(
"NOT EXISTS (SELECT 1 FROM phoenix_kit_entity_data ed WHERE ed.data::text LIKE '%' || ?::text || '%')",
f.uuid
)
)}
]
Enum.reduce(optional_checks, base, fn {table, condition}, query ->
if table in existing do
where(query, ^condition)
else
query
end
end)
end
defp existing_optional_tables do
repo = repo()
%{rows: rows} =
repo.query!(
"SELECT table_name FROM information_schema.tables WHERE table_schema = 'public' AND table_name LIKE 'phoenix_kit_%'"
)
List.flatten(rows)
end
# Returns UUIDs that should never be considered orphans, as configured by the parent app.
#
# Parent apps can register protected file UUIDs in their config:
#
# config :phoenix_kit, :protected_file_uuids, ["uuid1", "uuid2", ...]
#
# Or via a zero-arity function/MFA for dynamic resolution:
#
# config :phoenix_kit, :protected_file_uuids, {MyApp.Media, :all_uuids, []}
# config :phoenix_kit, :protected_file_uuids, fn -> MyApp.Media.all_uuids() end
#
defp protected_file_uuids do
case Application.get_env(:phoenix_kit, :protected_file_uuids, []) do
uuids when is_list(uuids) -> uuids
fun when is_function(fun, 0) -> fun.()
{mod, fun, args} -> apply(mod, fun, args)
end
end
# ===== FILE INSTANCES =====
@doc """
Returns a list of file instances for a given file.
"""
def list_file_instances(file_uuid) do
FileInstance
|> where([fi], fi.file_uuid == ^file_uuid)
|> order_by(asc: :variant_name)
|> repo().all()
end
@doc """
Gets a single file instance by ID.
"""
def get_file_instance(id), do: repo().get(FileInstance, id)
@doc """
Gets a file instance by file UUID and variant name.
"""
def get_file_instance_by_name(file_uuid, variant_name) do
repo().get_by(FileInstance, file_uuid: file_uuid, variant_name: variant_name)
end
@doc """
Gets the bucket UUIDs where a file instance is stored.
Returns a list of bucket UUIDs from the file_locations for the given file instance.
"""
def get_file_instance_bucket_uuids(file_instance_uuid) do
FileLocation
|> where([fl], fl.file_instance_uuid == ^file_instance_uuid and fl.status == "active")
|> select([fl], fl.bucket_uuid)
|> repo().all()
end
@doc """
Creates a new file instance.
"""
def create_file_instance(attrs \\ %{}) do
%FileInstance{}
|> FileInstance.changeset(attrs)
|> repo().insert()
end
@doc """
Updates a file instance.
"""
def update_file_instance(%FileInstance{} = instance, attrs) do
instance
|> FileInstance.changeset(attrs)
|> repo().update()
end
@doc """
Deletes a file instance.
"""
def delete_file_instance(%FileInstance{} = instance) do
repo().delete(instance)
end
@doc """
Returns an `%Ecto.Changeset{}` for tracking file instance changes.
"""
def change_file_instance(%FileInstance{} = instance, attrs \\ %{}) do
FileInstance.changeset(instance, attrs)
end
@doc """
Updates a file instance's processing status.
"""
def update_instance_status(instance, status)
when status in ["pending", "processing", "completed", "failed"] do
update_file_instance(instance, %{processing_status: status})
end
@doc """
Updates a file instance with file information after processing.
"""
def update_instance_with_file_info(instance, file_path, dimensions \\ nil) do
{:ok, stat} = Elixir.File.stat(file_path)
size = stat.size
checksum = calculate_file_hash(file_path)
attrs = %{
checksum: checksum,
size: size
}
attrs =
case dimensions do
{width, height} ->
Map.merge(attrs, %{width: width, height: height})
_ ->
attrs
end
update_file_instance(instance, attrs)
end
# ===== CONFIGURATION =====
@impl PhoenixKit.Module
@doc """
Gets the current storage configuration.
"""
def get_config do
buckets = list_buckets()
active_count = Enum.count(buckets, & &1.enabled)
%{
module_enabled: true,
default_path: get_default_path(),
redundancy_copies: get_redundancy_copies(),
auto_generate_variants: get_auto_generate_variants(),
default_bucket_uuid: get_default_bucket_uuid(),
buckets_count: length(buckets),
active_buckets_count: active_count
}
end
# ============================================================================
# Module Behaviour Callbacks
# ============================================================================
@impl PhoenixKit.Module
def module_key, do: "storage"
@impl PhoenixKit.Module
def module_name, do: "Storage"
@impl PhoenixKit.Module
def enabled?, do: module_enabled?()
@impl PhoenixKit.Module
def enable_system, do: {:ok, :always_enabled}
@impl PhoenixKit.Module
def disable_system, do: {:ok, :always_enabled}
@impl PhoenixKit.Module
def permission_metadata do
%{
key: "storage",
label: "Storage",
icon: "hero-circle-stack",
description: "Distributed file storage with multi-location redundancy"
}
end
@doc """
Gets the default storage path for local file uploads (relative path).
Returns the configured relative path or the default "priv/uploads" if not set.
"""
def get_default_path do
Settings.get_setting_cached("storage_default_path", @default_path)
end
@doc """
Gets the absolute path for local storage.
"""
def get_absolute_path do
default_path = get_default_path()
Path.expand(default_path, Elixir.File.cwd!())
end
@doc """
Validates and normalizes a storage path.
Returns `{:ok, relative_path}` if valid, or error tuple if invalid.
"""
def validate_and_normalize_path(path) when is_binary(path) do
expanded_path = Path.expand(path, Elixir.File.cwd!())
cond do
not Elixir.File.exists?(expanded_path) ->
{:error, :does_not_exist, expanded_path}
not Elixir.File.dir?(expanded_path) ->
{:error, "Path is not a directory: #{expanded_path}"}
not writable?(expanded_path) ->
{:error, "Directory is not writable: #{expanded_path}"}
true ->
relative_path = Path.relative_to(expanded_path, Elixir.File.cwd!())
{:ok, relative_path}
end
end
def validate_and_normalize_path(_path), do: {:error, :invalid_path}
@doc """
Updates the default storage path.
"""
def update_default_path(relative_path) when is_binary(relative_path) do
Settings.update_setting("storage_default_path", relative_path)
end
@doc """
Creates a directory if it doesn't exist.
"""
def create_directory(path) when is_binary(path) do
case Elixir.File.mkdir_p(path) do
:ok -> {:ok, path}
{:error, reason} -> {:error, reason}
end
end
def create_directory(_path), do: {:error, :invalid_path}
# ===== FILE STORAGE OPERATIONS =====
@doc """
Stores a file in the storage system.
This will:
1. Store the file in multiple buckets based on redundancy settings
2. Generate variants if enabled
3. Create database records for the file and its variants
## Options
- `:filename` - Original filename (required)
- `:content_type` - MIME type (required)
- `:size_bytes` - File size in bytes (required)
- `:user_uuid` - User UUID who owns the file
- `:metadata` - Additional metadata map
"""
def store_file(source_path, opts \\ []) do
filename = Keyword.fetch!(opts, :filename)
content_type = Keyword.fetch!(opts, :content_type)
size_bytes = Keyword.fetch!(opts, :size_bytes)
user_uuid = Keyword.get(opts, :user_uuid)
metadata = Keyword.get(opts, :metadata, %{})
# Validate required fields
if Elixir.File.exists?(source_path) do
# Calculate file checksum
file_checksum = calculate_file_hash(source_path)
# Calculate user-specific checksum for duplicate detection
user_file_checksum = calculate_user_file_checksum(user_uuid, file_checksum)
# Check if this user already uploaded this file
case get_file_by_user_checksum(user_file_checksum) do
%PhoenixKit.Modules.Storage.File{} = existing_file ->
# File already exists for this user, return existing file
{:ok, existing_file}
nil ->
# New file for this user, proceed with storage
store_new_file(
source_path,
file_checksum,
user_file_checksum,
filename,
content_type,
size_bytes,
user_uuid,
metadata
)
end
else
{:error, "Source file does not exist"}
end
end
@doc """
Retrieves a file from storage by file UUID.
Will try buckets in priority order until the file is found.
"""
def retrieve_file(file_uuid) do
case get_file(file_uuid) do
%PhoenixKit.Modules.Storage.File{} = file ->
# Look up the original variant path from file_instances table
case get_file_instance_by_name(file_uuid, "original") do
%FileInstance{file_name: file_path} ->
destination_path = generate_temp_path()
case Manager.retrieve_file(file_path,
destination_path: destination_path
) do
{:ok, _path} -> {:ok, destination_path, file}
error -> error
end
nil ->
{:error, "Original file instance not found"}
end
nil ->
{:error, "File not found"}
end
end
@doc """
Retrieves a file by its hash.
"""
def retrieve_file_by_hash(hash) do
case get_file_by_checksum(hash) do
%PhoenixKit.Modules.Storage.File{} = file ->
retrieve_file(file.uuid)
nil ->
{:error, "File not found"}
end
end
@doc """
Deletes file data from all storage buckets for all variants.
"""
def delete_file_data(%PhoenixKit.Modules.Storage.File{} = file) do
instances = list_file_instances(file.uuid)
if instances == [] do
{:error, "No file instances found"}
else
results =
Enum.map(instances, fn instance ->
case Manager.delete_file(instance.file_name) do
:ok ->
:ok
error ->
Logger.warning(
"Failed to delete variant #{instance.variant_name}: #{inspect(error)}"
)
error
end
end)
if Enum.any?(results, &(&1 == :ok)),
do: :ok,
else: {:error, "Failed to delete from all buckets"}
end
end
# ===== TRASH =====
@doc "Moves a file to trash (soft-delete). Sets status to 'trashed' and records timestamp."
def trash_file(%PhoenixKit.Modules.Storage.File{} = file) do
file
|> Ecto.Changeset.change(%{
status: "trashed",
trashed_at: DateTime.utc_now() |> DateTime.truncate(:second)
})
|> repo().update()
end
def trash_file(file_uuid) when is_binary(file_uuid) do
case get_file(file_uuid) do
nil -> {:error, :not_found}
file -> trash_file(file)
end
end
@doc "Restores a trashed file back to active status."
def restore_file(%PhoenixKit.Modules.Storage.File{} = file) do
file
|> Ecto.Changeset.change(%{status: "active", trashed_at: nil})
|> repo().update()
end
def restore_file(file_uuid) when is_binary(file_uuid) do
case get_file(file_uuid) do
nil -> {:error, :not_found}
file -> restore_file(file)
end
end
@doc "Returns trashed files ordered by trashed_at descending, with pagination and optional scope."
def list_trashed_files(scope \\ nil, opts \\ []) do
query =
build_trashed_query(scope)
|> order_by([f], desc: f.trashed_at)
query = if opts[:limit], do: limit(query, ^opts[:limit]), else: query
query = if opts[:offset], do: offset(query, ^opts[:offset]), else: query
repo().all(query)
end
@doc "Returns the count of trashed files, optionally scoped."
def count_trashed_files(scope \\ nil) do
build_trashed_query(scope)
|> repo().aggregate(:count, :uuid)
end
defp build_trashed_query(nil) do
from(f in PhoenixKit.Modules.Storage.File, where: f.status == "trashed")
end
defp build_trashed_query(scope_folder_id) do
cte_base =
from(f in Folder,
where: f.uuid == ^scope_folder_id,
select: %{uuid: f.uuid}
)
cte_recursive =
from(f in Folder,
join: d in "scope_descendants",
on: f.parent_uuid == d.uuid,
select: %{uuid: f.uuid}
)
cte = union_all(cte_base, ^cte_recursive)
from(f in PhoenixKit.Modules.Storage.File,
join: d in "scope_descendants",
on: f.folder_uuid == d.uuid,
where: f.status == "trashed"
)
|> with_cte("scope_descendants", as: ^cte)
|> recursive_ctes(true)
end
@doc "Permanently deletes all trashed files, optionally scoped."
def empty_trash(scope \\ nil) do
trashed = list_trashed_files(scope)
Enum.each(trashed, &delete_file_completely/1)
{:ok, length(trashed)}
end
@doc "Permanently deletes trashed files older than the given number of days."
def prune_trash(days) do
cutoff = DateTime.utc_now() |> DateTime.add(-days * 86_400, :second)
expired =
PhoenixKit.Modules.Storage.File
|> where([f], f.status == "trashed" and f.trashed_at < ^cutoff)
|> repo().all()
Enum.each(expired, &delete_file_completely/1)
{:ok, length(expired)}
end
@doc "Returns the configured trash retention period in days (default 30)."
def trash_retention_days do
Settings.get_setting("trash_retention_days", "30")
|> to_string()
|> Integer.parse()
|> case do
{n, _} -> n
:error -> 30
end
rescue
_ -> 30
end
@doc """
Deletes a file completely - physical data from all storage buckets and database record.
## Examples
iex> delete_file_completely(file)
{:ok, %File{}}
"""
def delete_file_completely(%PhoenixKit.Modules.Storage.File{} = file) do
if other_files_share_path?(file) do
# Other files share the same storage — only delete the DB record, keep physical files
Logger.info(
"Storage: skipping physical deletion for #{file.uuid} (shared path: #{file.file_path})"
)
else
# Last reference — safe to delete physical files
case delete_file_data(file) do
:ok ->
Logger.info("Storage: physical files deleted for #{file.uuid}")
{:error, reason} ->
Logger.warning("Storage: partial physical deletion for #{file.uuid}: #{reason}")
end
end
# Delete DB record (CASCADE handles instances + locations)
delete_file(file)
end
def delete_file_completely(file_uuid) when is_binary(file_uuid) do
case get_file(file_uuid) do
nil -> {:error, :not_found}
file -> delete_file_completely(file)
end
end
@doc """
Gets a public URL for a file.
"""
def get_public_url(%PhoenixKit.Modules.Storage.File{} = file) do
# Look up the actual file path from file_instances where "original" variant is stored
case get_file_instance_by_name(file.uuid, "original") do
%PhoenixKit.Modules.Storage.FileInstance{file_name: file_path} ->
Manager.public_url(file_path) || signed_file_url(file.uuid, "original")
nil ->
nil
end
end
@doc """
Gets a public URL for a specific file variant.
## Variants
For images: "original", "thumbnail", "small", "medium", "large"
For videos: "original", "360p", "720p", "1080p", "video_thumbnail"
## Examples
iex> get_public_url_by_variant(file, "thumbnail")
"https://cdn.example.com/12/a1/a1b2c3d4e5f6/a1b2c3d4e5f6_thumbnail.jpg"
iex> get_public_url_by_variant(file, "medium")
"https://cdn.example.com/12/a1/a1b2c3d4e5f6/a1b2c3d4e5f6_medium.jpg"
"""
def get_public_url_by_variant(%PhoenixKit.Modules.Storage.File{} = file, variant_name) do
case get_file_instance_by_name(file.uuid, variant_name) do
%PhoenixKit.Modules.Storage.FileInstance{file_name: file_path} ->
Manager.public_url(file_path) || signed_file_url(file.uuid, variant_name)
nil ->
# Fallback to original if variant doesn't exist
get_public_url(file)
end
end
@doc """
Gets a public URL for a file by file ID.
Convenience function that fetches the file and returns its URL.
## Examples
iex> get_public_url_by_uuid("018e3c4a-9f6b-7890-abcd-ef1234567890")
"https://cdn.example.com/12/a1/a1b2c3d4e5f6/a1b2c3d4e5f6_original.jpg"
iex> get_public_url_by_uuid("invalid-uuid")
nil
"""
def get_public_url_by_uuid(file_uuid) when is_binary(file_uuid) do
case get_file(file_uuid) do
%PhoenixKit.Modules.Storage.File{} = file ->
get_public_url(file)
nil ->
nil
end
end
def get_public_url_by_uuid(_), do: nil
@doc """
Gets a public URL for a specific file variant by file ID.
## Examples
iex> get_public_url_by_uuid("018e3c4a-9f6b-7890-abcd-ef1234567890", "thumbnail")
"https://cdn.example.com/12/a1/a1b2c3d4e5f6/a1b2c3d4e5f6_thumbnail.jpg"
"""
def get_public_url_by_uuid(file_uuid, variant_name) when is_binary(file_uuid) do
case get_file(file_uuid) do
%PhoenixKit.Modules.Storage.File{} = file ->
get_public_url_by_variant(file, variant_name)
nil ->
nil
end
end
@doc """
Returns variant data for building an `<.image_set>` `<picture>` element.
Returns a list of maps with `:variant_name`, `:mime_type`, `:width`, and `:url`
for all completed image instances of the given file.
"""
def list_image_set_variants(file_uuid) when is_binary(file_uuid) do
repo = repo()
FileInstance
|> where([fi], fi.file_uuid == ^file_uuid)
|> where([fi], fi.processing_status == "completed")
|> where([fi], like(fi.mime_type, ^"image/%"))
|> repo.all()
|> Enum.map(fn fi ->
%{
variant_name: fi.variant_name,
mime_type: fi.mime_type,
width: fi.width,
height: fi.height,
url: URLSigner.signed_url(file_uuid, fi.variant_name, locale: :none)
}
end)
end
@doc """
Bulk version of `list_image_set_variants/1` for multiple files.
Returns a map of `%{file_uuid => [variant_maps]}`. Uses a single DB query.
"""
def list_image_set_variants_for_files(file_uuids) when is_list(file_uuids) do
if file_uuids == [] do
%{}
else
repo = repo()
FileInstance
|> where([fi], fi.file_uuid in ^file_uuids)
|> where([fi], fi.processing_status == "completed")
|> where([fi], like(fi.mime_type, ^"image/%"))
|> repo.all()
|> Enum.group_by(& &1.file_uuid, fn fi ->
%{
variant_name: fi.variant_name,
mime_type: fi.mime_type,
width: fi.width,
height: fi.height,
url: URLSigner.signed_url(fi.file_uuid, fi.variant_name, locale: :none)
}
end)
end
end
defp signed_file_url(file_uuid, variant_name) do
URLSigner.signed_url(file_uuid, variant_name, locale: :none)
rescue
_ -> nil
end
@doc """
Checks if a file exists in storage.
"""
def file_exists?(%PhoenixKit.Modules.Storage.File{} = file) do
# Look up the actual file path from file_instances where "original" variant is stored
case get_file_instance_by_name(file.uuid, "original") do
%PhoenixKit.Modules.Storage.FileInstance{file_name: file_path} ->
Manager.file_exists?(file_path)
nil ->
false
end
end
@doc """
Stores a file in buckets with hierarchical path structure.
## Path Structure
Files are stored using the pattern:
`{user_uuid[0..1]}/{hash[0..1]}/{full_hash}/{full_hash}_{variant}.{format}`
## Examples
User ID: "12345678"
File hash: "a1b2c3d4e5f6..."
Original: "12/a1/a1b2c3d4e5f6/a1b2c3d4e5f6_original.jpg"
Thumbnail: "12/a1/a1b2c3d4e5f6/a1b2c3d4e5f6_thumbnail.jpg"
"""
def store_file_in_buckets(
source_path,
file_type,
user_uuid,
file_checksum,
ext,
original_filename \\ nil
) do
# Check if any enabled buckets exist
case list_enabled_buckets() do
[] ->
{:error, :no_buckets_configured}
_buckets ->
# Proceed with storage
store_file_with_buckets_available(
source_path,
file_type,
user_uuid,
file_checksum,
ext,
original_filename
)
end
end
defp store_file_with_buckets_available(
source_path,
file_type,
user_uuid,
file_checksum,
ext,
original_filename
) do
# Calculate user-specific hash for duplicate detection
user_file_checksum = calculate_user_file_checksum(user_uuid, file_checksum)
# Check if this user already uploaded this file
case get_file_by_user_checksum(user_file_checksum) do
%PhoenixKit.Modules.Storage.File{} = existing_file ->
Logger.info("=== DUPLICATE FILE DETECTED ===")
Logger.info("File ID: #{existing_file.uuid}, Checksum: #{file_checksum}")
Logger.info("File path: #{existing_file.file_path}")
# File already exists, but check if instances and actual files are healthy
case get_file_instance_by_name(existing_file.uuid, "original") do
%FileInstance{file_name: stored_file_path} ->
Logger.info("Original instance record found: #{stored_file_path}")
# Instance record exists, verify actual file exists in storage
case verify_file_in_storage(stored_file_path) do
:exists ->
Logger.info("Duplicate file is healthy in storage. Queueing variant generation.")
# File is healthy in storage, ensure other variants are generated
_ = queue_variant_generation(existing_file, user_uuid, original_filename)
{:ok, existing_file, :duplicate}
:missing ->
# File record exists but actual file is missing from storage
# Need to re-store the file and recreate instances
Logger.warning(
"Duplicate file detected but missing from storage: #{existing_file.uuid}"
)
restore_missing_file(
existing_file,
source_path,
file_checksum,
user_uuid,
original_filename
)
end
nil ->
# File record exists but instance record is missing
# Need to recreate instances from the stored file
Logger.warning(
"Duplicate file detected but missing instance record: #{existing_file.uuid}"
)
Logger.info("Attempting to recreate instances...")
recreate_file_instances(
existing_file,
source_path,
file_checksum,
user_uuid,
original_filename
)
end
nil ->
# No per-user match — check for cross-user duplicate (same file uploaded by another user)
case get_active_file_by_checksum(file_checksum) do
%PhoenixKit.Modules.Storage.File{} = donor_file ->
Logger.info("=== CROSS-USER DUPLICATE DETECTED ===")
Logger.info("Donor file: #{donor_file.uuid} (user: #{donor_file.user_uuid})")
clone_file_for_user(
donor_file,
user_uuid,
file_checksum,
ext,
original_filename
)
nil ->
Logger.info("New file detected (no existing hash match). Proceeding with storage.")
store_new_file_in_buckets(
source_path,
file_type,
user_uuid,
file_checksum,
user_file_checksum,
ext,
original_filename
)
end
end
end
defp store_new_file_in_buckets(
source_path,
file_type,
user_uuid,
file_checksum,
user_file_checksum,
ext,
original_filename
) do
# Calculate MD5 hash for path structure
md5_hash =
source_path
|> Elixir.File.read!()
|> then(fn data -> :crypto.hash(:md5, data) end)
|> Base.encode16(case: :lower)
# Generate UUIDv7 for file UUID
file_uuid = UUIDv7.generate()
# Build hierarchical path - organized by user_prefix/hash_prefix/md5_hash
user_prefix = String.slice(to_string(user_uuid), 0, 2)
hash_prefix = String.slice(md5_hash, 0, 2)
file_path = "#{user_prefix}/#{hash_prefix}/#{md5_hash}"
# Use provided original filename or fall back to source basename
orig_filename = original_filename || Path.basename(source_path)
# Create file record
file_attrs = %{
uuid: file_uuid,
file_name: md5_hash <> "." <> ext,
original_file_name: orig_filename,
file_path: file_path,
mime_type: determine_mime_type(ext),
file_type: file_type,
ext: ext,
file_checksum: file_checksum,
user_file_checksum: user_file_checksum,
size: get_file_size(source_path),
status: "processing",
user_uuid: user_uuid
}
case create_file(file_attrs) do
{:ok, file} ->
# Store in buckets with redundancy - use MD5 hash for organized structure
original_path = "#{file_path}/#{md5_hash}_original.#{ext}"
case Manager.store_file(source_path, path_prefix: original_path) do
{:ok, storage_info} ->
# Create file instance for original
original_instance_attrs = %{
variant_name: "original",
file_name: original_path,
mime_type: file.mime_type,
ext: ext,
checksum: file_checksum,
size: get_file_size(source_path),
processing_status: "completed",
file_uuid: file.uuid
}
case create_file_instance(original_instance_attrs) do
{:ok, instance} ->
# Create file location records for each bucket where the file was stored
_ = create_file_locations(instance.uuid, storage_info.bucket_ids, original_path)
# Queue background job for variant processing
_ =
%{file_uuid: file.uuid, user_uuid: user_uuid, filename: orig_filename}
|> ProcessFileJob.new()
|> Oban.insert()
{:ok, file}
{:error, changeset} ->
# Clean up if instance creation fails
Manager.delete_file(original_path)
{:error, changeset}
end
{:error, reason} ->
# Clean up file record if storage fails
repo().delete(file)
{:error, reason}
end
{:error, changeset} ->
{:error, changeset}
end
end
# ===== CROSS-USER DEDUPLICATION =====
# Find any active file with the given checksum (regardless of user) for cross-user dedup
defp get_active_file_by_checksum(file_checksum) do
PhoenixKit.Modules.Storage.File
|> where([f], f.file_checksum == ^file_checksum and f.status == "active")
|> limit(1)
|> repo().one()
end
# Create a new File record for a different user, reusing the same storage path
defp clone_file_for_user(donor_file, user_uuid, file_checksum, ext, original_filename) do
user_file_checksum = calculate_user_file_checksum(user_uuid, file_checksum)
file_attrs = %{
file_name: donor_file.file_name,
original_file_name: original_filename || donor_file.original_file_name,
file_path: donor_file.file_path,
mime_type: donor_file.mime_type,
file_type: donor_file.file_type,
ext: ext,
file_checksum: file_checksum,
user_file_checksum: user_file_checksum,
size: donor_file.size,
width: donor_file.width,
height: donor_file.height,
status: "active",
user_uuid: user_uuid
}
case create_file(file_attrs) do
{:ok, new_file} ->
clone_file_instances(donor_file.uuid, new_file.uuid)
Logger.info("Cross-user clone created: #{new_file.uuid} from donor #{donor_file.uuid}")
{:ok, new_file, :duplicate}
{:error, changeset} ->
Logger.error("Failed to clone file for user: #{inspect(changeset.errors)}")
{:error, changeset}
end
end
# Copy all FileInstance records from one file to another (same storage paths)
defp clone_file_instances(donor_file_uuid, new_file_uuid) do
donor_instances = list_file_instances(donor_file_uuid)
Enum.each(donor_instances, fn instance ->
attrs = %{
variant_name: instance.variant_name,
file_name: instance.file_name,
mime_type: instance.mime_type,
ext: instance.ext,
checksum: instance.checksum,
size: instance.size,
width: instance.width,
height: instance.height,
processing_status: instance.processing_status,
file_uuid: new_file_uuid
}
%FileInstance{}
|> FileInstance.changeset(attrs)
|> repo().insert()
end)
# Also clone file locations for each new instance
clone_file_locations(donor_file_uuid, new_file_uuid)
end
# Copy FileLocation records from donor instances to new instances
defp clone_file_locations(donor_file_uuid, new_file_uuid) do
donor_instances = list_file_instances(donor_file_uuid)
new_instances = list_file_instances(new_file_uuid)
# Match instances by variant_name and copy their locations
Enum.each(new_instances, fn new_inst ->
donor_inst = Enum.find(donor_instances, &(&1.variant_name == new_inst.variant_name))
if donor_inst do
FileLocation
|> where([fl], fl.file_instance_uuid == ^donor_inst.uuid and fl.status == "active")
|> repo().all()
|> Enum.each(fn loc ->
%FileLocation{}
|> FileLocation.changeset(%{
path: loc.path,
status: "active",
priority: loc.priority,
file_instance_uuid: new_inst.uuid,
bucket_uuid: loc.bucket_uuid
})
|> repo().insert()
end)
end
end)
end
# Check if other File records share the same storage path
defp other_files_share_path?(file) do
PhoenixKit.Modules.Storage.File
|> where([f], f.file_path == ^file.file_path and f.uuid != ^file.uuid)
|> repo().exists?()
end
# ===== HELPER FUNCTIONS =====
defp queue_variant_generation(file, user_uuid, original_filename) do
# Queue variant generation to ensure all variants exist for this file
Task.start(fn ->
%{file_uuid: file.uuid, user_uuid: user_uuid, filename: original_filename}
|> ProcessFileJob.new()
|> Oban.insert()
end)
end
defp verify_file_in_storage(stored_file_path) do
# Check if file actually exists in storage buckets
Logger.info("Verifying file in storage: #{stored_file_path}")
exists = Manager.file_exists?(stored_file_path)
Logger.info("File exists? #{exists}")
if exists, do: :exists, else: :missing
end
defp restore_missing_file(existing_file, source_path, file_hash, user_uuid, original_filename) do
# File record exists but actual file is missing from storage
# Delete broken instances and recreate them (which will also store the file)
Logger.warning("=== RECOVERING MISSING FILE ===")
Logger.warning("File ID: #{existing_file.uuid}")
Logger.warning("File path: #{existing_file.file_path}")
Logger.warning("Source path: #{source_path}")
# First, delete all broken instances for this file
deleted_count = delete_file_instances_for_file(existing_file.uuid)
Logger.info("Deleted #{deleted_count} broken instances for file: #{existing_file.uuid}")
# Recreate instance and store the file (combined in one operation)
Logger.info("Recreating instances for file #{existing_file.uuid}")
recreate_file_instances(existing_file, source_path, file_hash, user_uuid, original_filename)
end
defp delete_file_instances_for_file(file_uuid) do
# Delete all file instances for a file (to clean up broken ones)
{deleted_count, _} =
from(fi in FileInstance, where: fi.file_uuid == ^file_uuid)
|> repo().delete_all()
Logger.info("Deleted #{deleted_count} file instances for file_uuid: #{file_uuid}")
deleted_count
end
defp recreate_file_instances(file, source_path, file_checksum, user_uuid, original_filename) do
# File record exists but instances are missing or broken
# First store the file in buckets, then recreate the instance record
Logger.info(
"Starting recreate_file_instances for file: #{file.uuid}, file_path: #{file.file_path}"
)
{:ok, stat} = Elixir.File.stat(source_path)
file_size = stat.size
# Reconstruct the full storage path for the original instance
# file.file_path is "user_prefix/hash_prefix/md5_hash"
# We need to extract md5_hash and build the original path
[_user_prefix, _hash_prefix, md5_hash | _rest] = String.split(file.file_path, "/")
original_path = "#{file.file_path}/#{md5_hash}_original.#{file.ext}"
Logger.info("Reconstructed original path for instance: #{original_path}")
Logger.info(
"About to store file from source_path: #{source_path} to storage path: #{original_path}"
)
# First, store the file in buckets using Manager
case Manager.store_file(source_path, path_prefix: original_path) do
{:ok, storage_info} ->
Logger.info(
"File stored in buckets: #{original_path}, bucket_ids: #{inspect(storage_info.bucket_ids)}"
)
# Now create the file instance record pointing to the stored file
original_instance_attrs = %{
variant_name: "original",
file_name: original_path,
mime_type: file.mime_type,
ext: file.ext,
checksum: file_checksum,
size: file_size,
processing_status: "completed",
file_uuid: file.uuid
}
case create_file_instance(original_instance_attrs) do
{:ok, _instance} ->
Logger.info(
"Recreated original instance for file: #{file.uuid}, path: #{original_path}"
)
# Delete any remaining broken variant instances BEFORE queuing ProcessFileJob
# This ensures ProcessFileJob creates fresh instances with correct paths
deleted_variants = delete_variant_instances(file.uuid)
Logger.info(
"Deleted #{deleted_variants} broken variant instances before regeneration"
)
# Queue variant generation for the recovered file
_ = queue_variant_generation(file, user_uuid, original_filename)
{:ok, file, :duplicate}
{:error, reason} ->
# Instance creation failed, might be duplicate constraint
# Try deleting old broken instances and recreating
Logger.warning(
"Instance creation failed for file #{file.uuid}: #{inspect(reason)}, attempting cleanup and retry"
)
_ = delete_file_instances_for_file(file.uuid)
case create_file_instance(original_instance_attrs) do
{:ok, _instance} ->
Logger.info(
"Recreated original instance for file (after cleanup): #{file.uuid}, path: #{original_path}"
)
# Delete any remaining broken variant instances
deleted_variants = delete_variant_instances(file.uuid)
Logger.info(
"Deleted #{deleted_variants} broken variant instances before regeneration"
)
_ = queue_variant_generation(file, user_uuid, original_filename)
{:ok, file, :duplicate}
{:error, final_reason} ->
Logger.error(
"Failed to recreate instance for file #{file.uuid}: #{inspect(final_reason)}"
)
{:error, final_reason}
end
end
{:error, store_error} ->
Logger.error(
"Failed to store file in buckets for recreate_file_instances: #{inspect(store_error)}"
)
{:error, store_error}
end
end
defp delete_variant_instances(file_uuid) do
# Delete only the variant instances (not the original), to clean up broken ones
{deleted_count, _} =
from(fi in FileInstance,
where: fi.file_uuid == ^file_uuid and fi.variant_name != "original"
)
|> repo().delete_all()
deleted_count
end
defp get_file_size(source_path) do
case Elixir.File.stat(source_path) do
{:ok, stat} -> stat.size
_ -> 0
end
end
defp determine_mime_type(ext) do
case String.downcase(ext) do
"jpg" -> "image/jpeg"
"jpeg" -> "image/jpeg"
"png" -> "image/png"
"gif" -> "image/gif"
"webp" -> "image/webp"
"mp4" -> "video/mp4"
"webm" -> "video/webm"
"mov" -> "video/quicktime"
"avi" -> "video/x-msvideo"
"pdf" -> "application/pdf"
_ -> "application/octet-stream"
end
end
defp get_redundancy_copies do
Settings.get_setting_cached("storage_redundancy_copies", "1")
|> String.to_integer()
|> max(1)
|> min(5)
end
def get_auto_generate_variants do
Settings.get_setting_cached("storage_auto_generate_variants", "true") == "true"
end
defp get_default_bucket_uuid do
Settings.get_setting_cached("storage_default_bucket_uuid", nil)
end
defp calculate_local_free_space(bucket) do
# For local storage, return configured max_size_mb or default 1000 MB
# Note: Real disk space monitoring should be implemented via System.cmd("df")
# or external monitoring tools, as :disksup is not reliably available
bucket.max_size_mb || 1000
end
# Check if directory is writable
defp writable?(path) do
test_file = Path.join(path, ".phoenix_kit_write_test")
case Elixir.File.write(test_file, "test") do
:ok ->
Elixir.File.rm(test_file)
true
{:error, _} ->
false
end
end
# ===== REPO HELPERS =====
defp repo do
PhoenixKit.Config.get_repo()
end
# Query builders for file listing
defp maybe_filter_by_bucket(query, nil), do: query
defp maybe_filter_by_bucket(query, bucket_uuid) do
where(query, [f], f.bucket_uuid == ^bucket_uuid)
end
defp maybe_order_by(query, nil), do: order_by(query, [f], desc: f.inserted_at)
defp maybe_order_by(query, order_by), do: order_by(query, [f], ^order_by)
defp maybe_limit(query, nil), do: query
defp maybe_limit(query, limit), do: limit(query, ^limit)
defp maybe_offset(query, nil), do: query
defp maybe_offset(query, offset), do: offset(query, ^offset)
# ===== FILE STORAGE HELPERS =====
defp store_new_file(
source_path,
file_checksum,
user_file_checksum,
filename,
content_type,
size_bytes,
user_uuid,
metadata
) do
# Store file using manager
case Manager.store_file(source_path) do
{:ok, storage_info} ->
file_attrs =
build_file_attrs(
storage_info,
filename,
content_type,
file_checksum,
user_file_checksum,
size_bytes,
metadata,
user_uuid
)
case create_file(file_attrs) do
{:ok, file} ->
# Create original instance and variants (non-critical operations)
create_original_instance_and_variants(file, file_checksum, size_bytes)
{:ok, file}
{:error, changeset} ->
# Clean up stored files if database creation fails
Manager.delete_file(storage_info.destination_path)
{:error, changeset}
end
{:error, reason} ->
{:error, reason}
end
end
defp build_file_attrs(
storage_info,
filename,
content_type,
file_checksum,
user_file_checksum,
size_bytes,
metadata,
user_uuid
) do
# Extract file_path (directory) from destination_path
# destination_path is like "01/ab/0123456789abcdef_original.jpg"
# file_path should be "01/ab/0123456789abcdef" (without filename)
file_path = Path.dirname(storage_info.destination_path)
%{
original_file_name: filename,
file_name: storage_info.destination_path,
file_path: file_path,
mime_type: content_type,
file_type: determine_file_type(content_type),
ext: Path.extname(filename),
file_checksum: file_checksum,
user_file_checksum: user_file_checksum,
size: size_bytes,
# Convert to MB
size_mb: size_bytes / (1024 * 1024),
status: "active",
metadata: metadata,
user_uuid: user_uuid
}
end
defp create_original_instance_and_variants(file, file_checksum, size_bytes) do
original_instance_attrs = %{
variant_name: "original",
file_name: file.file_name,
mime_type: file.mime_type,
ext: file.ext,
checksum: file_checksum,
size: size_bytes,
# Will be populated if we can detect dimensions
width: nil,
# Will be populated if we can detect dimensions
height: nil,
processing_status: "completed",
file_uuid: file.uuid
}
case create_file_instance(original_instance_attrs) do
{:ok, _original_instance} ->
# Generate variants if enabled (failure is non-critical)
case VariantGenerator.generate_variants(file) do
{:ok, _variants} -> :ok
{:error, _reason} -> :ok
end
{:error, _changeset} ->
# Original instance creation failed, but file was stored (non-critical)
:ok
end
end
defp calculate_file_hash(file_path) do
file_path
|> Elixir.File.read!()
|> then(fn data -> :crypto.hash(:sha256, data) end)
|> Base.encode16(case: :lower)
end
defp determine_file_type(mime_type) do
cond do
String.starts_with?(mime_type, "image/") ->
"image"
String.starts_with?(mime_type, "video/") ->
"video"
String.starts_with?(mime_type, "audio/") ->
"audio"
String.starts_with?(mime_type, "text/") ->
"document"
mime_type in ["application/pdf"] ->
"document"
mime_type in [
"application/msword",
"application/vnd.openxmlformats-officedocument.wordprocessingml.document"
] ->
"document"
String.contains?(mime_type, "zip") or String.contains?(mime_type, "archive") ->
"archive"
true ->
"other"
end
end
@doc """
Creates file locations for a file instance across specified buckets.
Returns `{:ok, locations}` on success or `{:error, :file_locations_failed, errors}` if any insertions fail.
## Parameters
* `file_instance_uuid` - The UUID of the file instance
* `bucket_uuids` - List of bucket UUIDs to create locations for
* `file_path` - The storage path for the file
## Examples
iex> create_file_locations_for_instance(instance_uuid, [bucket_uuid], "path/to/file")
{:ok, [%FileLocation{}]}
iex> create_file_locations_for_instance(instance_uuid, [invalid_bucket], "path")
{:error, :file_locations_failed, [{bucket_uuid, changeset}]}
"""
def create_file_locations_for_instance(file_instance_uuid, bucket_uuids, file_path) do
create_file_locations(file_instance_uuid, bucket_uuids, file_path)
end
defp generate_temp_path do
temp_dir = System.tmp_dir!()
random_name = :crypto.strong_rand_bytes(8) |> Base.encode16(case: :lower)
Path.join(temp_dir, "phoenix_kit_#{random_name}")
end
defp create_file_locations(file_instance_uuid, bucket_uuids, file_path) do
results =
Enum.map(bucket_uuids, fn bucket_uuid ->
location_attrs = %{
path: file_path,
status: "active",
priority: 0,
file_instance_uuid: file_instance_uuid,
bucket_uuid: bucket_uuid
}
case repo().insert(%FileLocation{} |> FileLocation.changeset(location_attrs)) do
{:ok, location} -> {:ok, location}
{:error, changeset} -> {:error, bucket_uuid, changeset}
end
end)
errors = Enum.filter(results, &match?({:error, _, _}, &1))
if errors == [] do
locations = Enum.map(results, fn {:ok, loc} -> loc end)
{:ok, locations}
else
error_details =
Enum.map(errors, fn {:error, bucket_uuid, changeset} -> {bucket_uuid, changeset} end)
{:error, :file_locations_failed, error_details}
end
end
end