defmodule OhMyExOhIrcd.Images do
@moduledoc """
Context for ImgServ: per-channel config, channel+user-bound API tokens,
hosted image files + metadata, per-user pin budgets, retention GC, and a
request log. Mirrors `OhMyExOhIrcd.Scrobbles` for the shared concepts.
Image bytes are stored on the local filesystem under `upload_dir/0`;
metadata lives in Postgres. Channel lookups go through
`OhMyExOhIrcd.Channels.get/1`, so only registered channels can be
configured, and `on_delete: :delete_all` cascades config/tokens/images
when a registration is dropped (the on-disk files are reaped by GC, or
call `delete_image/2`).
"""
import Ecto.Query, warn: false
require Logger
alias OhMyExOhIrcd.Repo
alias OhMyExOhIrcd.Accounts.User
alias OhMyExOhIrcd.Channels
alias OhMyExOhIrcd.Channels.ChannelRecord
alias OhMyExOhIrcd.Images.{ChannelConfig, Image, RequestLog, Token}
@log_retention 200
@ext_for %{
"image/png" => "png",
"image/jpeg" => "jpg",
"image/gif" => "gif",
"image/webp" => "webp"
}
## Server-configured settings
defp env, do: Application.get_env(:oh_my_ex_oh_ircd, :imgserv, [])
def upload_dir do
case env() |> Keyword.get(:upload_dir) do
dir when is_binary(dir) -> dir
_ -> Application.app_dir(:oh_my_ex_oh_ircd, "priv/uploads")
end
end
def max_bytes, do: env() |> Keyword.get(:max_bytes, 10_485_760)
def allowed_types, do: env() |> Keyword.get(:allowed_types, Map.keys(@ext_for))
def default_retention_days, do: env() |> Keyword.get(:retention_days, 30)
def gc_interval_ms, do: env() |> Keyword.get(:gc_interval_ms, 3_600_000)
def gc_enabled?, do: env() |> Keyword.get(:gc_enabled, true)
def pin_budget_bytes, do: (env() |> Keyword.get(:pin_budget_mb, 100)) * 1_048_576
## Config
@doc "Fetch the config for a channel by name. Returns nil if absent."
def get_config(channel_name) when is_binary(channel_name) do
case Channels.get(channel_name) do
nil -> nil
%ChannelRecord{id: id} -> Repo.get_by(ChannelConfig, channel_id: id)
end
end
@doc """
Fetch the config for a registered channel, creating it (enabled) if it
doesn't exist yet. Returns `{:ok, config}` or `{:error, :not_registered}`.
"""
def ensure_config(channel_name, attrs \\ %{}) when is_binary(channel_name) do
case Channels.get(channel_name) do
nil ->
{:error, :not_registered}
%ChannelRecord{id: id} ->
case Repo.get_by(ChannelConfig, channel_id: id) do
nil ->
%ChannelConfig{}
|> ChannelConfig.changeset(Map.merge(%{channel_id: id}, attrs))
|> Repo.insert()
%ChannelConfig{} = cfg ->
{:ok, cfg}
end
end
end
@doc """
Update config fields for a registered channel, creating the row first if
needed. Returns `{:ok, config}`, `{:error, changeset}`, or
`{:error, :not_registered}`.
"""
def update_config(channel_name, attrs) when is_binary(channel_name) do
with {:ok, cfg} <- ensure_config(channel_name) do
cfg
|> ChannelConfig.changeset(attrs)
|> Repo.update()
end
end
def set_enabled(channel_name, value), do: update_config(channel_name, %{enabled: !!value})
def set_joined(channel_name, value), do: update_config(channel_name, %{joined: !!value})
def set_format(channel_name, format), do: update_config(channel_name, %{format: format})
def set_restrict(channel_name, value),
do: update_config(channel_name, %{restrict_to_allowlist: !!value})
def set_rate_limit(channel_name, seconds),
do: update_config(channel_name, %{rate_limit_seconds: seconds})
def set_retention_days(channel_name, days),
do: update_config(channel_name, %{retention_days: days})
@doc "Add a `who` display name to the allowlist (stored downcased, deduped)."
def add_allow(channel_name, who) when is_binary(who) do
who = normalize_who(who)
with {:ok, cfg} <- ensure_config(channel_name) do
list = Enum.uniq([who | cfg.allowlist])
update_config(channel_name, %{allowlist: list})
end
end
@doc "Remove a `who` display name from the allowlist."
def remove_allow(channel_name, who) when is_binary(who) do
who = normalize_who(who)
with {:ok, cfg} <- ensure_config(channel_name) do
list = Enum.reject(cfg.allowlist, &(&1 == who))
update_config(channel_name, %{allowlist: list})
end
end
@doc "Channel names where ImgServ should currently be present."
def list_joined_channels do
Repo.all(
from c in ChannelConfig,
where: c.joined == true,
join: ch in ChannelRecord,
on: ch.id == c.channel_id,
select: ch.name
)
end
## Tokens
@doc """
Mint a token for a registered channel, bound to `user` and announcing as
`who`. Returns `{:ok, display_token, token}`; `display_token` is shown
once.
"""
def mint_token(channel_name, %User{} = user, who, label \\ nil)
when is_binary(channel_name) and is_binary(who) do
who = String.trim(who)
case Channels.get(channel_name) do
_ when who == "" ->
{:error, :blank_who}
nil ->
{:error, :not_registered}
%ChannelRecord{id: id} ->
{display, struct} = Token.build(id, user.id, who, label)
case struct |> Token.changeset(%{}) |> Repo.insert() do
{:ok, token} -> {:ok, display, token}
{:error, cs} -> {:error, cs}
end
end
end
@doc "List (non-revoked) tokens for a channel. Never exposes plaintext."
def list_tokens(channel_name) when is_binary(channel_name) do
case Channels.get(channel_name) do
nil ->
[]
%ChannelRecord{id: id} ->
Repo.all(
from t in Token,
where: t.channel_id == ^id and is_nil(t.revoked_at),
order_by: [desc: t.inserted_at]
)
end
end
@doc "Revoke a token by id, scoped to the channel. Idempotent."
def revoke_token(channel_name, token_id) when is_binary(channel_name) do
case Channels.get(channel_name) do
nil ->
:ok
%ChannelRecord{id: id} ->
from(t in Token,
where: t.id == ^token_id and t.channel_id == ^id and is_nil(t.revoked_at)
)
|> Repo.update_all(set: [revoked_at: DateTime.utc_now(:second)])
:ok
end
end
@doc """
Verify a presented bearer token. Returns `{:ok, %{channel:, who:,
user_id:}}` for a live token — `who` is the bound name, not anything the
client supplied — else `:error`. Bumps `last_used_at` on success.
"""
def verify_token(presented) when is_binary(presented) do
with {:ok, hash} <- Token.hash_presented(presented),
%Token{} = token <-
Repo.one(from t in Token, where: t.token_hash == ^hash and is_nil(t.revoked_at)),
%ChannelRecord{} = channel <- Repo.get(ChannelRecord, token.channel_id) do
Repo.update_all(
from(t in Token, where: t.id == ^token.id),
set: [last_used_at: DateTime.utc_now(:second)]
)
{:ok, %{channel: channel, who: token.who, user_id: token.user_id}}
else
_ -> :error
end
end
def verify_token(_), do: :error
## Image storage
@doc "Absolute on-disk path for an image (or bare filename)."
def path_for(%Image{filename: f}), do: path_for(f)
def path_for(filename) when is_binary(filename), do: Path.join(upload_dir(), filename)
@doc "File extension for a known content type, or nil."
def ext_for(content_type), do: Map.get(@ext_for, content_type)
@doc """
Store an uploaded image for a channel and record its metadata.
`source` is `{:path, tmp_path}` (e.g. a `Plug.Upload`/LiveView temp file)
or `{:binary, bytes}`. `attrs` carries `who` (required), `content_type`
(required, must be allowed), `byte_size` (required), and optional
`user_id` / `title`. Returns `{:ok, image}` or `{:error,
:bad_type | :too_large | :not_registered | changeset | reason}`.
"""
def store_upload(channel_name, attrs, source) when is_binary(channel_name) do
attrs = upload_attrs(attrs)
content_type = attrs.content_type
with :ok <- validate_upload(content_type, attrs.byte_size) do
case Channels.get(channel_name) do
nil -> {:error, :not_registered}
%ChannelRecord{id: id} -> do_store(id, attrs, content_type, source)
end
end
end
@upload_attrs [:who, :user_id, :title, :content_type, :byte_size]
# attrs may come atom- or string-keyed; read once into atom keys
defp upload_attrs(attrs), do: Map.new(@upload_attrs, &{&1, upload_attr(attrs, &1)})
defp upload_attr(attrs, key), do: Map.get(attrs, key) || Map.get(attrs, Atom.to_string(key))
defp validate_upload(content_type, byte_size) do
cond do
content_type not in allowed_types() -> {:error, :bad_type}
not is_integer(byte_size) or byte_size <= 0 -> {:error, :empty}
byte_size > max_bytes() -> {:error, :too_large}
true -> :ok
end
end
defp do_store(channel_id, attrs, content_type, source) do
filename = gen_filename(content_type)
dest = path_for(filename)
:ok = File.mkdir_p!(upload_dir())
with :ok <- write_source(dest, source),
{:ok, image} <-
%Image{}
|> Image.changeset(%{
channel_id: channel_id,
user_id: attrs[:user_id],
who: attrs[:who],
filename: filename,
content_type: content_type,
byte_size: attrs[:byte_size],
title: attrs[:title]
})
|> Repo.insert() do
{:ok, image}
else
{:error, reason} ->
File.rm(dest)
{:error, reason}
end
end
defp write_source(dest, {:binary, bin}), do: File.write(dest, bin)
defp write_source(dest, {:path, src}), do: File.cp(src, dest)
defp gen_filename(content_type) do
slug = :crypto.strong_rand_bytes(9) |> Base.url_encode64(padding: false)
case ext_for(content_type) do
nil -> slug
ext -> "#{slug}.#{ext}"
end
end
@doc "Fetch an image by its public filename, or nil."
def get_image_by_filename(filename) when is_binary(filename),
do: Repo.get_by(Image, filename: filename)
@doc "Most recent `n` images for a channel, newest first."
def recent(channel_name, n \\ 30) when is_binary(channel_name) do
case Channels.get(channel_name) do
nil ->
[]
%ChannelRecord{id: id} ->
Repo.all(
from i in Image,
where: i.channel_id == ^id,
order_by: [desc: i.inserted_at, desc: i.id],
limit: ^n
)
end
end
@doc "Delete an image (row + file), scoped to the channel. Idempotent."
def delete_image(channel_name, image_id) when is_binary(channel_name) do
case Channels.get(channel_name) do
nil ->
:ok
%ChannelRecord{id: id} ->
case Repo.get_by(Image, id: image_id, channel_id: id) do
nil ->
:ok
%Image{} = image ->
delete_file(image)
Repo.delete(image)
:ok
end
end
end
## Pinning (per-user byte budget)
@doc "Total bytes a user currently has pinned, across all channels."
def pinned_bytes_for_user(user_id) when is_integer(user_id) do
Repo.one(
from i in Image,
where: i.pinned == true and i.pinned_by_user_id == ^user_id,
select: coalesce(sum(i.byte_size), 0)
)
end
@doc """
Pin an image (exempt from GC), attributed to `user` and counted against
their pin budget. Returns `{:ok, image}`, `{:error, :over_budget}`, or
`{:error, :not_found}`. Already-pinned images are returned as-is.
"""
def pin_image(channel_name, image_id, %User{} = user) when is_binary(channel_name) do
with %ChannelRecord{id: cid} <- Channels.get(channel_name),
%Image{} = image <- Repo.get_by(Image, id: image_id, channel_id: cid) do
cond do
image.pinned ->
{:ok, image}
pinned_bytes_for_user(user.id) + image.byte_size > pin_budget_bytes() ->
{:error, :over_budget}
true ->
image |> Image.pin_changeset(user) |> Repo.update()
end
else
_ -> {:error, :not_found}
end
end
@doc "Unpin an image, scoped to the channel. Idempotent."
def unpin_image(channel_name, image_id) when is_binary(channel_name) do
with %ChannelRecord{id: cid} <- Channels.get(channel_name),
%Image{} = image <- Repo.get_by(Image, id: image_id, channel_id: cid) do
image |> Image.unpin_changeset() |> Repo.update()
else
_ -> {:error, :not_found}
end
end
## Retention GC
@doc """
Delete non-pinned images older than their channel's retention window
(per-channel `retention_days` override, else the server default),
removing both the DB row and the on-disk file. Returns the number of
images deleted. Safe to run concurrently / repeatedly.
"""
def gc do
global = default_retention_days()
retention =
Repo.all(from c in ChannelConfig, select: {c.channel_id, c.retention_days}) |> Map.new()
now = DateTime.utc_now()
from(i in Image, where: not i.pinned, select: i.channel_id, distinct: true)
|> Repo.all()
|> Enum.reduce(0, fn cid, acc ->
days = Map.get(retention, cid) || global
cutoff = DateTime.add(now, -days * 86_400, :second)
stale =
Repo.all(
from i in Image,
where: i.channel_id == ^cid and not i.pinned and i.inserted_at < ^cutoff
)
Enum.each(stale, &delete_file/1)
ids = Enum.map(stale, & &1.id)
{n, _} = Repo.delete_all(from i in Image, where: i.id in ^ids)
acc + n
end)
end
defp delete_file(%Image{} = image) do
case File.rm(path_for(image)) do
:ok ->
:ok
{:error, :enoent} ->
:ok
{:error, reason} ->
Logger.warning("ImgServ GC could not delete #{image.filename}: #{inspect(reason)}")
end
end
## Request logs
@doc """
Record the outcome of an authenticated upload request. `attrs` carries
`who`, `status`, and optional `detail`. Prunes to the most recent
`#{@log_retention}` rows per channel. Best-effort: never raises.
"""
def log_request(channel_id, attrs) when is_integer(channel_id) do
%RequestLog{}
|> RequestLog.changeset(Map.put(attrs, :channel_id, channel_id))
|> Repo.insert()
|> case do
{:ok, _} = ok ->
prune_logs(channel_id)
ok
other ->
other
end
rescue
_ -> :error
end
@doc "Most recent `n` request-log rows for a channel, newest first."
def recent_logs(channel_name, n \\ 50) when is_binary(channel_name) do
case Channels.get(channel_name) do
nil ->
[]
%ChannelRecord{id: id} ->
Repo.all(
from l in RequestLog,
where: l.channel_id == ^id,
order_by: [desc: l.inserted_at, desc: l.id],
limit: ^n
)
end
end
defp prune_logs(channel_id) do
keep_ids =
from(l in RequestLog,
where: l.channel_id == ^channel_id,
order_by: [desc: l.inserted_at, desc: l.id],
limit: @log_retention,
select: l.id
)
Repo.delete_all(
from l in RequestLog,
where: l.channel_id == ^channel_id and l.id not in subquery(keep_ids)
)
end
defp normalize_who(who), do: who |> String.trim() |> String.downcase()
end
neiam
/oh-my-ex-oh-ircd
public · Issues · Pulls · Labels · Forks · Compare · Actions success · Badges · Packages
⭐
Log in to mark this repository.
15.0 KiB · text
History
bc25690