neiam /gitgud

defmodule GitGud.Packages.Pruner do
  @moduledoc """
  Carries out what `GitGud.Packages.Retention` decided.

  Split from the planning deliberately: a plan is a query and can be
  shown to somebody, while this deletes bytes. Nothing here runs unless
  `packages.retention_enforced` is on, or the caller passes
  `force: true` — which is what the "prune now" button in the admin
  view does, once an admin has looked at the plan.

  ## Two kinds of bytes

  A file-backed version (PyPI, Hex, Cargo, apt) owns rows in
  `package_files` pointing at `GitGud.Storage`. An OCI version owns a
  manifest in zot. Both are deleted, and the OCI half is the one that
  can fail for reasons outside this instance — if zot refuses or is
  unreachable, the mirror row **stays**, because a row with no manifest
  is a package page advertising something nobody can pull.

  ## Shared bytes

  Storage keys are content-addressed per `(repository, format,
  sha256)`, so two versions that published identical bytes share a key
  — a wheel rebuilt reproducibly, or the same `.deb` in two suites.
  Deleting one version must not delete bytes another still points at,
  so every object is refcounted against `package_files` before it goes.
  """

  import Ecto.Query, warn: false

  require Logger

  alias GitGud.Packages.Package
  alias GitGud.Packages.PackageFile
  alias GitGud.Packages.PackageVersion
  alias GitGud.Packages.Retention
  alias GitGud.Packages.Zot
  alias GitGud.Repo
  alias GitGud.Repositories.Repository
  alias GitGud.Storage

  @typedoc "What a prune actually did."
  @type result :: %{
          status: :pruned | :reported,
          pruned: non_neg_integer(),
          bytes: non_neg_integer(),
          failed: [%{version: String.t(), reason: term()}]
        }

  @doc """
  Apply retention to one package.

  Reports rather than deletes unless enforcement is on. The return
  shape is the same either way, so a caller can render "freed 2.1 GiB"
  and "would free 2.1 GiB" from one code path.
  """
  @spec prune(Package.t(), keyword()) :: {:ok, result()}
  def prune(%Package{} = package, opts \\ []) do
    plan = Retention.plan(package)

    if Keyword.get(opts, :force, false) or Retention.enforced?() do
      {:ok, execute(package, plan)}
    else
      {:ok,
       %{
         status: :reported,
         pruned: length(plan.prune),
         bytes: plan.bytes,
         failed: []
       }}
    end
  end

  @doc """
  Apply retention across every package, or a scoped subset
  (`repository:` / `organization_id:`).

  One package's failure doesn't stop the walk: a zot that is down
  should delay OCI pruning, not stop a PyPI registry from being tidied.
  """
  @spec prune_all(keyword()) :: {:ok, result()}
  def prune_all(opts \\ []) do
    packages =
      Package
      |> scope(opts)
      |> Repo.all()

    result =
      Enum.reduce(packages, %{status: :reported, pruned: 0, bytes: 0, failed: []}, fn package,
                                                                                      acc ->
        {:ok, one} = prune(package, opts)

        %{
          status: if(one.status == :pruned, do: :pruned, else: acc.status),
          pruned: acc.pruned + one.pruned,
          bytes: acc.bytes + one.bytes,
          failed: acc.failed ++ one.failed
        }
      end)

    {:ok, result}
  end

  defp scope(query, opts) do
    Enum.reduce(opts, query, fn
      {:repository, %Repository{id: id}}, q -> where(q, [p], p.repository_id == ^id)
      {:repository_id, id}, q -> where(q, [p], p.repository_id == ^id)
      {:organization_id, id}, q -> scope_org(q, id)
      _, q -> q
    end)
  end

  defp scope_org(query, org_id) do
    repos = from(r in Repository, where: r.organization_id == ^org_id, select: r.id)
    where(query, [p], p.repository_id in subquery(repos))
  end

  # ── doing it ─────────────────────────────────────────────────────────

  defp execute(%Package{} = package, plan) do
    repo = Repo.get!(Repository, package.repository_id)

    {pruned, bytes, failed} =
      Enum.reduce(plan.prune, {0, 0, []}, fn version, {count, bytes, failed} ->
        size = Retention.version_bytes(version)

        case delete_version(repo, package, version) do
          :ok ->
            {count + 1, bytes + size, failed}

          {:error, reason} ->
            Logger.warning(
              "retention: keeping #{package.name} #{version.version}: #{inspect(reason)}"
            )

            {count, bytes, [%{version: version.version, reason: reason} | failed]}
        end
      end)

    if pruned > 0, do: refresh_latest_version(package)

    %{status: :pruned, pruned: pruned, bytes: bytes, failed: Enum.reverse(failed)}
  end

  # OCI first and separately: the manifest has to be gone from zot
  # before the row goes, or the page stops advertising something that
  # is still taking up disk.
  defp delete_version(repo, %Package{format: "oci"}, %PackageVersion{digest: digest} = version)
       when is_binary(digest) do
    case Zot.delete_manifest(repo, digest) do
      :ok -> delete_rows(version)
      {:error, :no_registry} -> delete_rows(version)
      {:error, _} = err -> err
    end
  end

  defp delete_version(_repo, _package, %PackageVersion{} = version) do
    files = Repo.all(from f in PackageFile, where: f.package_version_id == ^version.id)

    Enum.each(files, &delete_object/1)
    delete_rows(version)
  end

  defp delete_rows(%PackageVersion{} = version) do
    Repo.transaction(fn ->
      Repo.delete_all(from f in PackageFile, where: f.package_version_id == ^version.id)
      Repo.delete!(version)
    end)

    :ok
  end

  # Content-addressed keys are shared, so "this row is going" is not the
  # same as "these bytes are going".
  defp delete_object(%PackageFile{storage_bucket: bucket, storage_key: key} = file)
       when is_binary(bucket) and is_binary(key) do
    if last_reference?(file) do
      case Storage.delete(bucket, key) do
        :ok ->
          :ok

        other ->
          # A stranded object costs disk; a missing object breaks a
          # download. Losing the race in this direction is the right way
          # round, so log and carry on.
          Logger.warning("retention: could not delete #{bucket}/#{key}: #{inspect(other)}")
          :ok
      end
    else
      :ok
    end
  end

  defp delete_object(_file), do: :ok

  defp last_reference?(%PackageFile{id: id, storage_bucket: bucket, storage_key: key}) do
    count =
      from(f in PackageFile,
        where: f.storage_bucket == ^bucket and f.storage_key == ^key and f.id != ^id,
        select: count(f.id)
      )
      |> Repo.one()

    count == 0
  end

  defp refresh_latest_version(%Package{} = package) do
    latest =
      from(v in PackageVersion,
        where: v.package_id == ^package.id,
        order_by: [desc_nulls_last: v.pushed_at, desc: v.id],
        limit: 50
      )
      |> Repo.all()
      |> PackageVersion.latest()
      |> case do
        nil -> nil
        version -> version.version
      end

    package
    |> Package.changeset(%{latest_version: latest})
    |> Repo.update()
  end
end