rindle

defmodule Rindle do
  alias Rindle.Domain.MediaAsset
  alias Rindle.Domain.MediaAttachment
  alias Rindle.Domain.MediaUploadSession
  alias Rindle.Domain.MediaVariant
  alias Rindle.Error
  alias Rindle.Internal.OwnerErasure
  alias Rindle.Internal.VariantFailureLogger
  alias Rindle.Ops.LifecycleRepair
  alias Rindle.Ops.RuntimeStatus
  alias Rindle.Security.UploadValidation
  alias Rindle.Upload.Broker
  alias Rindle.Upload.TusPlug
  alias Rindle.Workers.ProcessVariant
  alias Rindle.Workers.PromoteAsset
  alias Rindle.Workers.PurgeStorage

  @moduledoc """
  Phoenix/Ecto-native media lifecycle library.

  Rindle manages the full post-upload lifecycle: upload sessions, staged object
  verification, asset modeling, attachment associations, variant/derivative
  generation, background processing, secure delivery, observability, and
  day-2 operations.

  ## Owner/account erasure contract

  The recommended `v1.10` owner/account erasure facade is
  `Rindle.preview_owner_erasure/2` for dry-run planning and
  `Rindle.erase_owner/2` for the execute lane. Preview and execute share the
  same `owner_erasure_report()` vocabulary so adopters can audit what will be
  detached now, what purge work is enqueued later, and which shared assets are
  intentionally retained.

  For multi-owner orchestration, `Rindle.preview_batch_owner_erasure/2` and
  `Rindle.erase_batch_owner_erasure/2` are the supported batch preview and
  execute entrypoints. Batch reports use `owner_erasure_batch_report()` with
  per-owner nested `owner_erasure_report()` entries.

  The stable report buckets are `attachments_to_detach`, `assets_to_purge`,
  and `retained_shared_assets`.

  Shared assets are retained whenever a surviving attachment remains.
  Execute semantics stay honest: detach work happens transactionally first,
  then purge-enqueued async cleanup handles newly orphaned assets later. This
  facade contract does not promise inline storage deletion,
  scheduler/cron erasure jobs, or force-delete behavior for assets with
  surviving attachments.

  Rindle now ships a mountable admin console, mounted from your router via
  `Rindle.Admin.Router.rindle_admin/2`. The console is host-authenticated and
  separate from this facade; see `guides/admin_console.md` for the adopter
  setup. Admin read composition lives in `Rindle.Admin.Queries`, which is
  internal and is not promoted onto this `Rindle` facade.

  `detach/3` remains the slot-scoped attachment API, and
  `mix rindle.cleanup_orphans` remains the maintenance-only upload-residue
  cleanup lane rather than the owner/account erasure surface.
  """

  import Ecto.Query

  @typedoc "Tagged storage result shape: {:ok, result} | {:error, reason}"
  @type storage_result :: {:ok, term()} | {:error, term()}

  @typedoc """
  Count-plus-list reporting bucket used by `owner_erasure_report()`.
  """
  @type owner_erasure_bucket :: %{
          count: non_neg_integer(),
          entries: [map()]
        }

  @typedoc """
  Stable public owner/account erasure report vocabulary shared by
  `Rindle.preview_owner_erasure/2` and `Rindle.erase_owner/2`.

  `attachments_to_detach` reports the owner attachment rows targeted by the
  operation, `assets_to_purge` reports newly orphaned assets whose purge work
  is enqueued, and `retained_shared_assets` reports assets retained because a
  surviving attachment still exists.
  """
  @type owner_erasure_report :: %{
          mode: :preview | :execute,
          attachments_to_detach: owner_erasure_bucket(),
          assets_to_purge: owner_erasure_bucket(),
          retained_shared_assets: owner_erasure_bucket(),
          purge_enqueued: non_neg_integer(),
          purge_already_queued: non_neg_integer()
        }

  @typedoc """
  Owner identity reference extracted from owner structs.
  Matches the tuple `Rindle.Internal.OwnerErasure` derives internally.
  """
  @type owner_ref :: {owner_type :: String.t(), owner_id :: Ecto.UUID.t()}

  @typedoc """
  Per-owner entry in a batch erasure report.
  """
  @type owner_erasure_batch_entry :: %{
          owner: owner_ref(),
          report: owner_erasure_report()
        }

  @typedoc """
  Aggregate batch erasure report with per-owner nested reports.
  """
  @type owner_erasure_batch_report :: %{
          mode: :preview | :execute,
          attachments_to_detach: owner_erasure_bucket(),
          assets_to_purge: owner_erasure_bucket(),
          retained_shared_assets: owner_erasure_bucket(),
          owners: [owner_erasure_batch_entry()]
        }

  @typedoc "Detail map for batch size limit violations."
  @type batch_too_large_detail :: %{
          requested: non_neg_integer(),
          max: pos_integer()
        }

  @typedoc "Detail map when batch processing stops after a per-owner failure."
  @type batch_owner_failed_detail :: %{
          owner: owner_ref(),
          reason: term(),
          partial_report: owner_erasure_batch_report()
        }

  @type batch_owner_erasure_result ::
          {:ok, owner_erasure_batch_report()}
          | {:error, :empty_batch}
          | {:error, {:batch_too_large, batch_too_large_detail()}}
          | {:error, term()}

  @doc """
  Returns the current version of Rindle.

  ## Examples

      iex> is_binary(Rindle.version())
      true

  """
  @spec version :: String.t()
  def version do
    Application.spec(:rindle, :vsn) |> to_string()
  end

  @doc """
  Initiates a direct upload session through the broker.

  Delegates to `Broker.initiate_session/2`. Returns
  `{:ok, %MediaUploadSession{}}` on success.

  ## Examples

      # Requires `config :rindle, :repo, MyApp.Repo` and a configured profile module.
      iex> {:ok, session} = Rindle.initiate_upload(MyApp.MediaProfile, filename: "photo.png")
      iex> session.state
      "initialized"

  """
  @spec initiate_upload(module(), keyword()) :: {:ok, MediaUploadSession.t()} | {:error, term()}
  def initiate_upload(profile, opts \\ []) do
    Broker.initiate_session(profile, opts)
  end

  @doc """
  Initiates a multipart direct upload session through the broker.
  """
  @spec initiate_multipart_upload(module(), keyword()) :: Broker.initiate_multipart_result()
  def initiate_multipart_upload(profile, opts \\ []) do
    Broker.initiate_multipart_session(profile, opts)
  end

  @doc """
  Signs a single multipart upload part through the broker.
  """
  @spec sign_multipart_part(binary(), pos_integer(), keyword()) :: Broker.sign_part_result()
  def sign_multipart_part(session_id, part_number, opts \\ []) do
    Broker.sign_multipart_part(session_id, part_number, opts)
  end

  @doc """
  Completes a multipart upload through the broker and reuses upload verification.
  """
  @spec complete_multipart_upload(binary(), [map()], keyword()) :: Broker.verify_result()
  def complete_multipart_upload(session_id, parts, opts \\ []) do
    Broker.complete_multipart_upload(session_id, parts, opts)
  end

  @doc """
  Initiates a resumable upload session through the broker.
  """
  @spec initiate_resumable_session(module(), keyword()) :: Broker.initiate_resumable_result()
  def initiate_resumable_session(profile, opts \\ []) do
    Broker.initiate_resumable_session(profile, opts)
  end

  @doc """
  Initiates a tus upload resource through the broker and returns the signed
  upload URL needed by browser tus clients.

  Requires an explicit mounted tus `:path`, the adopter `:secret_key_base`, and
  the file `:length` in bytes. The returned `upload_url` is a bearer credential
  and should only be handed to the client that will upload the file.
  """
  @spec initiate_tus_upload(module(), keyword()) :: TusPlug.create_upload_result()
  def initiate_tus_upload(profile, opts \\ []) do
    TusPlug.create_upload(profile, opts)
  end

  @doc """
  Plans owner/account erasure without changing DB or storage state.

  Returns the same semantic report vocabulary as `erase_owner/2`, but with
  `mode: :preview` and no purge work enqueued.
  """
  @spec preview_owner_erasure(struct(), keyword()) ::
          {:ok, owner_erasure_report()} | {:error, term()}
  def preview_owner_erasure(owner, opts \\ []) do
    OwnerErasure.preview(owner, opts)
  end

  @doc """
  Erases an owner's Rindle-managed attachment rows and enqueues orphan-only purge work.

  Execute semantics stay honest: attachment detach happens transactionally,
  while storage deletion remains asynchronous through `PurgeStorage`.
  """
  @spec erase_owner(struct(), keyword()) :: {:ok, owner_erasure_report()} | {:error, term()}
  def erase_owner(owner, opts \\ []) do
    OwnerErasure.execute(owner, opts)
  end

  @doc """
  Plans batch owner/account erasure without changing DB or storage state.

  Processes each deduped owner sequentially via `OwnerErasure.preview/2` in its
  own read-only planner pass. Empty batches and over-limit owner counts are
  rejected before any planner work runs.

  On per-owner failure after earlier owners succeeded, returns
  `{:error, {:batch_owner_failed, detail}}` with `partial_report` containing only
  completed owners.
  """
  @spec preview_batch_owner_erasure([struct()], keyword()) :: batch_owner_erasure_result()
  def preview_batch_owner_erasure(owners, opts \\ []) do
    with :ok <- validate_batch_owners(owners, opts) do
      run_batch_owner_erasure(owners, :preview, &OwnerErasure.preview/2, opts)
    end
  end

  @doc """
  Erases multiple owners' Rindle-managed attachment rows and enqueues orphan-only purge work.

  Processes each deduped owner sequentially via `OwnerErasure.execute/2`, one
  transaction per owner. Empty batches and over-limit owner counts are rejected
  before any planner work runs.

  On per-owner failure after earlier owners succeeded, returns
  `{:error, {:batch_owner_failed, detail}}` with `partial_report` containing only
  completed owners; earlier owners remain committed.
  """
  @spec erase_batch_owner_erasure([struct()], keyword()) :: batch_owner_erasure_result()
  def erase_batch_owner_erasure(owners, opts \\ []) do
    with :ok <- validate_batch_owners(owners, opts) do
      run_batch_owner_erasure(owners, :execute, &OwnerErasure.execute/2, opts)
    end
  end

  @doc """
  Polls the broker-owned resumable upload session without changing completion trust.
  """
  @spec resumable_session_status(binary(), keyword()) :: Broker.resumable_status_result()
  def resumable_session_status(session_id, opts \\ []) do
    Broker.resumable_session_status(session_id, opts)
  end

  @doc """
  Cancels a broker-owned resumable upload session.
  """
  @spec cancel_resumable_session(binary(), keyword()) :: Broker.cancel_resumable_result()
  def cancel_resumable_session(session_id, opts \\ []) do
    Broker.cancel_resumable_session(session_id, opts)
  end

  @doc """
  Verifies a direct upload completion through the broker.

  Delegates to `Broker.verify_completion/2`. Promotes the
  session to `completed` and the asset to `validating`.

  ## Examples

      # Requires a configured Rindle repo + the upload object to exist in storage.
      iex> {:ok, %{session: session, asset: asset}} = Rindle.verify_completion(session_id)
      iex> session.state
      "completed"
      iex> asset.state
      "validating"

  """
  @spec verify_completion(binary(), keyword()) :: Broker.verify_result()
  def verify_completion(session_id, opts \\ []) do
    Broker.verify_completion(session_id, opts)
  end

  @doc deprecated: "Use verify_completion/2"
  @doc """
  Verifies a direct upload completion through the broker.

  Legacy compatibility shim for `0.1.x`. Delegates to
  `verify_completion/2` while the older name remains supported.

  ## Examples

      # Requires a configured Rindle repo + the upload object to exist in storage.
      iex> {:ok, %{session: session, asset: asset}} = Rindle.verify_upload(session_id)
      iex> session.state
      "completed"
      iex> asset.state
      "validating"

  """
  @spec verify_upload(binary(), keyword()) :: Broker.verify_result()
  def verify_upload(session_id, opts \\ []) do
    verify_completion(session_id, opts)
  end

  @doc """
  Resolves the storage adapter module for a given profile.

  ## Examples

      # Requires a profile module that defines `storage_adapter/0`.
      iex> Rindle.storage_adapter_for(MyApp.MediaProfile)
      Rindle.Storage.Local

  """
  @spec storage_adapter_for(module()) :: module()
  def storage_adapter_for(profile) do
    profile.storage_adapter()
  end

  @doc """
  Stores an object through the profile-specific storage adapter.

  ## Examples

      # Requires a configured storage adapter and a readable source file.
      iex> {:ok, _meta} = Rindle.store(MyApp.MediaProfile, "uploads/abc.png", "/tmp/abc.png")
      iex> :ok
      :ok

  """
  @spec store(module(), String.t(), Path.t(), keyword()) :: storage_result()
  def store(profile, key, source_path, opts \\ []) do
    invoke_storage(profile, :store, [key, source_path, opts])
  end

  @doc """
  Attaches a MediaAsset to an owner at a specific slot.

  If an attachment already exists in that slot, it is replaced and the old
  asset is purged asynchronously via `PurgeStorage`.

  ## Examples

      # Requires a configured Rindle repo + an existing MediaAsset and owner record.
      iex> {:ok, attachment} = Rindle.attach(asset_id, %MyApp.User{id: user_id}, "avatar")
      iex> attachment.slot
      "avatar"

  """
  @spec attach(MediaAsset.t() | binary(), struct(), String.t(), keyword()) ::
          {:ok, MediaAttachment.t()} | {:error, term()}
  def attach(asset_or_id, owner, slot, _opts \\ []) do
    repo = Rindle.Config.repo()
    asset_id = get_asset_id(asset_or_id)
    {owner_type, owner_id} = get_owner_info(owner)

    Ecto.Multi.new()
    |> Ecto.Multi.run(:existing, fn repo, _ ->
      existing =
        repo.one(
          from a in MediaAttachment,
            where: a.owner_type == ^owner_type and a.owner_id == ^owner_id and a.slot == ^slot
        )

      {:ok, existing}
    end)
    |> Ecto.Multi.run(:detach_old, fn repo, %{existing: existing} ->
      if existing do
        repo.delete(existing)
      else
        {:ok, nil}
      end
    end)
    |> Ecto.Multi.insert(:attachment, fn _ ->
      %MediaAttachment{}
      |> MediaAttachment.changeset(%{
        asset_id: asset_id,
        owner_type: owner_type,
        owner_id: owner_id,
        slot: slot
      })
    end)
    |> Ecto.Multi.run(:old_asset, fn tx_repo, %{existing: existing} ->
      if existing do
        {:ok, tx_repo.get!(MediaAsset, existing.asset_id)}
      else
        {:ok, nil}
      end
    end)
    |> Ecto.Multi.merge(fn %{old_asset: old_asset} ->
      if old_asset do
        Ecto.Multi.new()
        |> Oban.insert(
          :purge_old,
          PurgeStorage.new(%{
            "asset_id" => old_asset.id,
            "profile" => old_asset.profile
          })
        )
      else
        Ecto.Multi.new()
      end
    end)
    |> repo.transaction()
    |> case do
      {:ok, %{attachment: attachment}} -> {:ok, attachment}
      {:error, _name, reason, _changes} -> {:error, reason}
    end
  end

  @doc """
  Detaches any MediaAsset from an owner at a specific slot and triggers a purge.

  Idempotent: returns `:ok` even when no attachment exists at the slot.

  ## Examples

      # Requires a configured Rindle repo + an existing attachment at owner+slot.
      iex> :ok = Rindle.detach(%MyApp.User{id: user_id}, "avatar")
      iex> :ok
      :ok

  """
  @spec detach(struct(), String.t(), keyword()) :: :ok | {:error, term()}
  def detach(owner, slot, _opts \\ []) do
    repo = Rindle.Config.repo()
    {owner_type, owner_id} = get_owner_info(owner)

    Ecto.Multi.new()
    |> Ecto.Multi.run(:existing, fn repo, _ ->
      existing =
        repo.one(
          from a in MediaAttachment,
            where: a.owner_type == ^owner_type and a.owner_id == ^owner_id and a.slot == ^slot
        )

      if existing, do: {:ok, existing}, else: {:error, :not_found}
    end)
    |> Ecto.Multi.run(:old_asset, fn tx_repo, %{existing: existing} ->
      {:ok, tx_repo.get!(MediaAsset, existing.asset_id)}
    end)
    |> Ecto.Multi.delete(:attachment, fn %{existing: existing} -> existing end)
    |> Oban.insert(
      :purge,
      fn %{old_asset: old_asset} ->
        PurgeStorage.new(%{
          "asset_id" => old_asset.id,
          "profile" => old_asset.profile
        })
      end
    )
    |> repo.transaction()
    |> case do
      {:ok, _} -> :ok
      # Idempotent
      {:error, :existing, :not_found, _} -> :ok
      {:error, _name, reason, _changes} -> {:error, reason}
    end
  end

  # Bang variants — Phase 19 (API-11).

  @doc "Same as `attach/4` but raises `Rindle.Error` on failure or re-raises the original exception for storage adapter exceptions. Database constraint failures (e.g., foreign-key violations) surface as `Rindle.Error` with the underlying changeset as the reason."
  @spec attach!(MediaAsset.t() | binary(), struct(), String.t()) :: MediaAttachment.t()
  @spec attach!(MediaAsset.t() | binary(), struct(), String.t(), keyword()) :: MediaAttachment.t()
  def attach!(asset_or_id, owner, slot, opts \\ []) do
    case attach(asset_or_id, owner, slot, opts) do
      {:ok, attachment} ->
        attachment

      {:error, {:storage_adapter_exception, exception}} when is_exception(exception) ->
        raise exception

      {:error, reason} ->
        raise Error, action: :attach, reason: reason
    end
  end

  @doc "Same as `detach/3` but raises `Rindle.Error` on failure."
  @spec detach!(struct(), String.t()) :: :ok
  @spec detach!(struct(), String.t(), keyword()) :: :ok
  def detach!(owner, slot, opts \\ []) do
    case detach(owner, slot, opts) do
      :ok ->
        :ok

      {:error, %Ecto.Changeset{} = cs} ->
        raise Ecto.InvalidChangesetError, action: :delete, changeset: cs

      {:error, {:storage_adapter_exception, exception}} when is_exception(exception) ->
        raise exception

      {:error, reason} ->
        raise Error, action: :detach, reason: reason
    end
  end

  # Convenience read helpers — Phase 19 (API-09 / API-10).

  @doc """
  Fetches the most recent `MediaAttachment` for an `(owner, slot)` pair.

  Returns the attachment row with `:asset` preloaded by default, or `nil`
  when no attachment exists at the slot. Pass `preload: <list>` to override
  the default preload (the override **replaces** `[:asset]` rather than
  merging — pass `preload: [asset: :variants]` to extend, `preload: []` to
  disable preloading entirely).

  When multiple attachment rows exist for the same `(owner, slot)` (possible
  because the join schema enforces uniqueness only at the application level
  via `attach/4`'s last-write-wins replacement), the most-recent row by
  `:inserted_at` is returned.

  This helper does **not** issue a write or a side-effect query; it is safe
  to call in render paths.

  ## Examples

      # Requires a configured Rindle repo + an existing owner record.
      iex> attachment = Rindle.attachment_for(%MyApp.User{id: user_id}, "avatar")
      iex> attachment && attachment.slot
      "avatar"

  """
  @spec attachment_for(struct(), String.t()) :: MediaAttachment.t() | nil
  @spec attachment_for(struct(), String.t(), keyword()) :: MediaAttachment.t() | nil
  def attachment_for(owner, slot, opts \\ []) do
    repo = Rindle.Config.repo()
    {owner_type, owner_id} = get_owner_info(owner)
    preloads = Keyword.get(opts, :preload, [:asset])

    query =
      from a in MediaAttachment,
        where: a.owner_type == ^owner_type and a.owner_id == ^owner_id and a.slot == ^slot,
        order_by: [desc: a.inserted_at],
        limit: 1

    case repo.one(query) do
      nil -> nil
      attachment -> repo.preload(attachment, preloads)
    end
  end

  @doc """
  Lists `MediaVariant` rows in the `"ready"` state for a given asset.

  Accepts either a `%MediaAsset{}` struct or a binary asset id. Returns a
  list of variants ordered by `:name` ascending; returns `[]` when no
  variants are ready. The unique constraint on `(asset_id, name)` makes the
  ordering deterministic.

  Only variants in the `"ready"` state are returned — variants in
  `"planned"`, `"queued"`, `"processing"`, `"stale"`, `"missing"`,
  `"failed"`, or `"purged"` are excluded. Adopters wanting fallback
  behavior should call `variant_url/4`, which already orchestrates the
  stale-policy fallback.

  ## Examples

      # Requires a configured Rindle repo + at least one ready variant row.
      iex> variants = Rindle.ready_variants_for(asset)
      iex> Enum.all?(variants, &(&1.state == "ready"))
      true

  """
  @spec ready_variants_for(MediaAsset.t() | binary()) :: [MediaVariant.t()]
  def ready_variants_for(asset_or_id) do
    repo = Rindle.Config.repo()
    asset_id = get_asset_id(asset_or_id)

    repo.all(
      from v in MediaVariant,
        where: v.asset_id == ^asset_id and v.state == "ready",
        order_by: [asc: v.name]
    )
  end

  @doc """
  Cancels active variant processing for an asset.

  Returns `:ok` when the asset has queued or executing variant work that can be
  cancelled. Returns `{:error, :not_processing}` when the asset has no queued
  or executing variant work.

  The public surface remains asset-scoped; callers do not need to know variant
  ids, job ids, or Oban internals to stop in-flight processing.

  ## Examples

      iex> Rindle.cancel_processing(asset_id)
      :ok

      iex> Rindle.cancel_processing("missing-or-idle-asset")
      {:error, :not_processing}

  """
  @spec cancel_processing(MediaAsset.t() | binary()) :: :ok | {:error, :not_processing}
  def cancel_processing(asset_or_id) do
    asset_or_id
    |> get_asset_id()
    |> ProcessVariant.cancel_processing()
  end

  @doc """
  Reruns probe detection for an asset and persists only probe-derived fields.

  Accepts either a `%MediaAsset{}` struct or a binary asset id. Reprobe is
  asset-scoped and refreshes only `content_type`, `kind`, `width`, `height`,
  `duration_ms`, `has_video_track`, and `has_audio_track`; fields that no
  longer apply are cleared explicitly, while unrelated lifecycle state and
  ownership data stay untouched.

  Returns `{:ok, report}` on a completed probe refresh and `{:error, reason}`
  when the run could not be completed.

  ## Examples

      iex> {:ok, report} = Rindle.reprobe(asset_id)
      iex> report.kind
      "image"

  """
  @spec reprobe(MediaAsset.t() | binary()) ::
          {:ok, LifecycleRepair.reprobe_report()} | {:error, term()}
  def reprobe(asset_or_id) do
    LifecycleRepair.reprobe_asset(asset_or_id)
  end

  @doc """
  Requeues failed or cancelled variants for a single asset.

  Accepts either a `%MediaAsset{}` struct or a binary asset id. By default,
  only this asset's variants currently in `failed` or `cancelled` state are
  targeted. Pass `variant_names: [...]` to narrow the repair to explicit
  variant names; unknown names fail loudly, and already-ready siblings stay
  untouched.

  Returns `{:ok, report}` after the enqueue attempt finishes, including
  deterministic counters for selected, enqueued, skipped, and errored
  variants. Equivalent in-flight jobs are counted as skipped through Oban
  uniqueness rather than double-enqueued.

  ## Examples

      iex> {:ok, report} = Rindle.requeue_variants(asset_id)
      iex> report.enqueued
      1

      iex> {:ok, report} = Rindle.requeue_variants(asset_id, variant_names: ["thumb"])
      iex> report.selected
      1

  """
  @spec requeue_variants(MediaAsset.t() | binary(), keyword() | map()) ::
          {:ok, LifecycleRepair.requeue_report()} | {:error, term()}
  def requeue_variants(asset_or_id, opts \\ []) do
    LifecycleRepair.requeue_failed_variants(asset_or_id, opts)
  end

  @doc """
  Returns a bounded runtime diagnostics report for operators.

  The report is read-only and groups lifecycle drift, stuck work, and upload
  residue into a stable map shape with counts, oldest age, and bounded
  examples. Supported filters are intentionally narrow: `:profile`,
  `:older_than`, `:limit`, and `:format`.

  ## Examples

      iex> {:ok, report} = Rindle.runtime_status(limit: 3)
      iex> is_map(report.variants)
      true

  """
  @spec runtime_status(keyword() | map()) :: {:ok, map()} | {:error, term()}
  def runtime_status(opts \\ []) do
    RuntimeStatus.runtime_status(opts)
  end

  defp get_asset_id(%MediaAsset{id: id}), do: id
  defp get_asset_id(id) when is_binary(id), do: id

  defp get_owner_info(%{__struct__: module, id: id}) do
    {to_string(module), id}
  end

  @doc """
  Downloads an object through the profile-specific storage adapter.

  ## Examples

      # Requires a configured storage adapter and an existing object.
      iex> {:ok, _meta} = Rindle.download(MyApp.MediaProfile, "uploads/abc.png", "/tmp/abc.png")
      iex> :ok
      :ok

  """
  @spec download(module(), String.t(), Path.t(), keyword()) :: storage_result()
  def download(profile, key, destination_path, opts \\ []) do
    invoke_storage(profile, :download, [key, destination_path, opts])
  end

  @doc """
  Deletes an object through the profile-specific storage adapter.

  ## Examples

      # Requires a configured storage adapter.
      iex> {:ok, _} = Rindle.delete(MyApp.MediaProfile, "uploads/abc.png")
      iex> :ok
      :ok

  """
  @spec delete(module(), String.t(), keyword()) :: storage_result()
  def delete(profile, key, opts \\ []) do
    invoke_storage(profile, :delete, [key, opts])
  end

  @doc """
  Generates a delivery URL through the profile-specific storage adapter.

  Delegates to `Rindle.Delivery.url/3` so policy (public vs. signed) is honored.

  ## Examples

      # Requires a configured storage adapter and a key that exists in storage.
      iex> {:ok, url} = Rindle.url(MyApp.MediaProfile, "uploads/abc.png")
      iex> is_binary(url)
      true

  """
  @spec url(module(), String.t(), keyword()) :: storage_result()
  def url(profile, key, opts \\ []) do
    Rindle.Delivery.url(profile, key, opts)
  end

  @doc "Same as `url/3` but raises `Rindle.Error` on failure or re-raises the original exception for storage adapter exceptions."
  @spec url!(module(), String.t()) :: String.t()
  @spec url!(module(), String.t(), keyword()) :: String.t()
  def url!(profile, key, opts \\ []) do
    case url(profile, key, opts) do
      {:ok, url_string} ->
        url_string

      {:error, %Ecto.Changeset{} = cs} ->
        raise Ecto.InvalidChangesetError, action: :insert, changeset: cs

      {:error, {:storage_adapter_exception, exception}} when is_exception(exception) ->
        raise exception

      {:error, reason} ->
        raise Error, action: :url, reason: reason
    end
  end

  @doc """
  Generates a delivery URL for a variant, falling back when needed.

  Delegates to `Rindle.Delivery.variant_url/4`. Stale or non-ready variants
  fall back to the original asset URL per the configured stale-serving policy.

  ## Examples

      # Requires a configured storage adapter and ready/stale variant rows.
      iex> {:ok, url} = Rindle.variant_url(MyApp.MediaProfile, asset, variant)
      iex> is_binary(url)
      true

  """
  @spec variant_url(module(), map(), map(), keyword()) :: storage_result()
  def variant_url(profile, asset, variant, opts \\ []) do
    Rindle.Delivery.variant_url(profile, asset, variant, opts)
  end

  @doc "Same as `variant_url/4` but raises `Rindle.Error` on failure or re-raises the original exception for storage adapter exceptions."
  @spec variant_url!(module(), map(), map()) :: String.t()
  @spec variant_url!(module(), map(), map(), keyword()) :: String.t()
  def variant_url!(profile, asset, variant, opts \\ []) do
    case variant_url(profile, asset, variant, opts) do
      {:ok, url_string} ->
        url_string

      {:error, %Ecto.Changeset{} = cs} ->
        raise Ecto.InvalidChangesetError, action: :insert, changeset: cs

      {:error, {:storage_adapter_exception, exception}} when is_exception(exception) ->
        raise exception

      {:error, reason} ->
        raise Error, action: :variant_url, reason: reason
    end
  end

  @doc """
  Uploads a file directly through the server (proxied upload).

  Accepts a profile module and an upload (map or `%Plug.Upload{}`).
  The file is validated against the profile's `upload_policy/0`, stored
  via the profile's storage adapter, and a `MediaAsset` row is inserted
  in the `analyzing` state.

  ## Examples

      # Requires a configured Rindle repo + a configured storage adapter + a Plug.Upload.
      iex> {:ok, asset} = Rindle.upload(MyApp.MediaProfile, %Plug.Upload{path: "/tmp/x.png", filename: "x.png"})
      iex> asset.state
      "analyzing"

  """
  @spec upload(module(), map() | Plug.Upload.t(), keyword()) ::
          {:ok, MediaAsset.t()} | {:error, term()}
  def upload(profile_module, upload, opts \\ []) do
    repo = Rindle.Config.repo()
    upload = normalize_upload(upload)
    profile_name = to_string(profile_module)
    asset_id = Ecto.UUID.generate()

    # Get policy from profile
    policy = profile_module.upload_policy()

    with {:ok, validation} <-
           UploadValidation.validate_for_promotion(
             upload,
             policy,
             profile_name,
             asset_id
           ),
         {:ok, _storage_meta} <- store(profile_module, validation.storage_key, upload.path, opts) do
      Ecto.Multi.new()
      |> Ecto.Multi.insert(
        :asset,
        %MediaAsset{id: asset_id}
        |> MediaAsset.changeset(%{
          state: "analyzing",
          profile: profile_name,
          storage_key: validation.storage_key,
          filename: validation.sanitized_filename,
          content_type: validation.detected_mime,
          byte_size: upload.byte_size
        })
      )
      |> Oban.insert(:promote_job, PromoteAsset.new(%{asset_id: asset_id}))
      |> repo.transaction()
      |> case do
        {:ok, %{asset: asset}} -> {:ok, asset}
        {:error, _name, reason, _changes} -> {:error, reason}
      end
    else
      {:error, {:quarantine, reason}} ->
        # Handle quarantine (Pitfall 2)
        # For now, we return the error
        {:error, {:quarantine, reason}}

      {:error, reason} ->
        {:error, reason}
    end
  end

  @doc "Same as `upload/3` but raises `Rindle.Error` on failure, `Ecto.InvalidChangesetError` for changeset failures, or re-raises the original exception for storage adapter exceptions."
  @spec upload!(module(), map() | Plug.Upload.t()) :: MediaAsset.t()
  @spec upload!(module(), map() | Plug.Upload.t(), keyword()) :: MediaAsset.t()
  def upload!(profile_module, upload, opts \\ []) do
    case upload(profile_module, upload, opts) do
      {:ok, asset} ->
        asset

      {:error, %Ecto.Changeset{} = cs} ->
        raise Ecto.InvalidChangesetError, action: :insert, changeset: cs

      {:error, {:storage_adapter_exception, exception}} when is_exception(exception) ->
        raise exception

      {:error, reason} ->
        raise Error, action: :upload, reason: reason
    end
  end

  @doc """
  Checks for object existence through the profile-specific storage adapter.

  ## Examples

      # Requires a configured storage adapter.
      iex> {:ok, _meta} = Rindle.head(MyApp.MediaProfile, "uploads/abc.png")
      iex> :ok
      :ok

  """
  @spec head(module(), String.t(), keyword()) :: storage_result()
  def head(profile, key, opts \\ []) do
    invoke_storage(profile, :head, [key, opts])
  end

  @doc """
  Generates a presigned PUT payload through the profile-specific storage adapter.

  ## Examples

      # Requires an S3-compatible storage adapter with :presigned_put capability.
      iex> {:ok, %{url: url}} = Rindle.presigned_put(MyApp.MediaProfile, "uploads/abc.png", 3600)
      iex> is_binary(url)
      true

  """
  @spec presigned_put(module(), String.t(), pos_integer(), keyword()) :: storage_result()
  def presigned_put(profile, key, expires_in, opts \\ []) do
    invoke_storage(profile, :presigned_put, [key, expires_in, opts])
  end

  @doc """
  Executes variant storage and logs failures with required context metadata.

  Wraps `store/4` with structured failure logging that captures the
  `asset_id` and `variant_name` for observability dashboards.

  ## Examples

      # Requires a configured storage adapter.
      iex> {:ok, _meta} = Rindle.store_variant(MyApp.MediaProfile, "variants/abc-thumb.png", "/tmp/abc-thumb.png", asset_id: asset_id, variant_name: "thumb")
      iex> :ok
      :ok

  """
  @spec store_variant(module(), String.t(), Path.t(), keyword()) :: storage_result()
  def store_variant(profile, key, source_path, opts \\ []) do
    asset_id = Keyword.get(opts, :asset_id)
    variant_name = Keyword.get(opts, :variant_name)
    adapter_opts = Keyword.drop(opts, [:asset_id, :variant_name])

    case store(profile, key, source_path, adapter_opts) do
      {:ok, result} ->
        {:ok, result}

      {:error, reason} = error ->
        VariantFailureLogger.log(asset_id, variant_name, reason)
        error
    end
  end

  @deprecated "Use Rindle.Internal.VariantFailureLogger.log/3 instead — facade shim kept for 0.1.x compatibility only"
  @doc false
  @spec log_variant_processing_failure(term(), term(), term()) :: :ok
  def log_variant_processing_failure(asset_id, variant_name, reason) do
    VariantFailureLogger.log(asset_id, variant_name, reason)
  end

  defp invoke_storage(profile, function_name, args) do
    adapter = storage_adapter_for(profile)

    try do
      normalize_storage_result(apply(adapter, function_name, args))
    rescue
      exception ->
        {:error, {:storage_adapter_exception, exception}}
    end
  end

  defp normalize_storage_result({:ok, _result} = result), do: result
  defp normalize_storage_result({:error, _reason} = result), do: result
  defp normalize_storage_result(other), do: {:error, {:invalid_storage_response, other}}

  defp normalize_upload(%{__struct__: Plug.Upload} = upload) do
    %{
      path: upload.path,
      filename: upload.filename,
      content_type: upload.content_type,
      byte_size: File.stat!(upload.path).size
    }
  end

  defp normalize_upload(upload) when is_map(upload) do
    # Ensure byte_size is present if path is given
    if Map.has_key?(upload, :path) and not Map.has_key?(upload, :byte_size) do
      Map.put(upload, :byte_size, File.stat!(upload.path).size)
    else
      upload
    end
  end

  defp dedupe_batch_owners(owners), do: Enum.uniq_by(owners, &owner_ref/1)

  @doc false
  @spec per_owner_erasure_opts(keyword()) :: keyword()
  def per_owner_erasure_opts(opts) when is_list(opts), do: Keyword.drop(opts, [:max_owners])

  defp run_batch_owner_erasure(owners, mode, runner, opts) do
    per_owner_opts = per_owner_erasure_opts(opts)

    owners
    |> dedupe_batch_owners()
    |> Enum.reduce_while({:ok, []}, fn owner, {:ok, acc} ->
      case runner.(owner, per_owner_opts) do
        {:ok, report} ->
          {:cont, {:ok, [%{owner: owner_ref(owner), report: report} | acc]}}

        {:error, reason} ->
          partial_entries = Enum.reverse(acc)

          {:halt,
           {:error,
            {:batch_owner_failed,
             %{
               owner: owner_ref(owner),
               reason: reason,
               partial_report: build_batch_report(mode, partial_entries)
             }}}}
      end
    end)
    |> case do
      {:ok, entries} -> {:ok, build_batch_report(mode, Enum.reverse(entries))}
      error -> error
    end
  end

  defp build_batch_report(mode, entries) do
    reports = Enum.map(entries, & &1.report)

    %{
      mode: mode,
      attachments_to_detach: aggregate_bucket(reports, :attachments_to_detach),
      assets_to_purge: aggregate_bucket(reports, :assets_to_purge),
      retained_shared_assets: aggregate_bucket(reports, :retained_shared_assets),
      owners: entries
    }
  end

  defp aggregate_bucket(reports, key) do
    buckets = Enum.map(reports, &Map.fetch!(&1, key))

    %{
      count: Enum.sum(Enum.map(buckets, & &1.count)),
      entries: Enum.flat_map(buckets, & &1.entries)
    }
  end

  defp validate_batch_owners([], _opts), do: {:error, :empty_batch}

  defp validate_batch_owners(owners, opts) when is_list(owners) do
    max = resolve_max_batch_owners(opts)
    requested = owners |> Enum.map(&owner_ref/1) |> Enum.uniq() |> length()

    if requested > max do
      {:error, {:batch_too_large, %{requested: requested, max: max}}}
    else
      :ok
    end
  end

  defp resolve_max_batch_owners(opts) do
    Keyword.get(
      opts,
      :max_owners,
      Application.get_env(:rindle, :max_batch_erasure_owners, 100)
    )
  end

  defp owner_ref(%{__struct__: module, id: id}), do: {to_string(module), id}
end