aprs.me/lib/aprsme/workers/packet_cleanup_worker.ex

defmodule Aprsme.Workers.PacketCleanupWorker do
  @moduledoc """
  Worker for cleaning up old APRS packet data.

  This worker is responsible for:
  1. Removing packets older than the retention period (7 days by default)
  2. Providing granular cleanup with custom age parameters
  3. Logging statistics about cleanup operations
  4. Supporting batch processing for large datasets

  Uses a single-query DELETE with a CTE for efficient batch cleanup,
  avoiding the overhead of a separate SELECT + DELETE per batch.

  ## Functions
  - `perform/1` - Standard cleanup using configured retention period
  - `cleanup_packets_older_than/1` - Cleanup packets older than specified days
  - `cleanup_packets_in_batches/2` - Batch cleanup for large datasets
  """

  import Ecto.Query

  alias Aprsme.BadPacket
  alias Aprsme.Packet
  alias Aprsme.Repo

  require Logger

  # Batch size for cleanup operations to prevent long-running transactions
  @batch_size 10_000
  # Maximum time to spend on cleanup in milliseconds (5 minutes)
  @max_cleanup_time 5 * 60 * 1000

  @spec perform(map()) :: :ok | {:error, String.t()}
  def perform(%{"cleanup_days" => days}) when is_integer(days) do
    Logger.info("Starting scheduled APRS packet cleanup for packets older than #{days} days")

    {deleted_count, _} = cleanup_packets_older_than_batched(days)

    Logger.info("APRS packet cleanup complete: removed #{deleted_count} packets older than #{days} days")

    :ok
  rescue
    error ->
      Logger.error("Packet cleanup failed: #{inspect(error)}\n#{inspect(__STACKTRACE__)}")
      {:error, "Cleanup failed: #{inspect(error)}"}
  end

  @spec perform(map()) :: :ok | {:error, String.t()}
  def perform(_args) do
    Logger.info("Starting scheduled APRS packet cleanup")

    {deleted_count, _} = cleanup_old_packets_batched()

    bad_packet_count = cleanup_old_bad_packets()

    retention_days = Application.get_env(:aprsme, :packet_retention_days, 7)

    Logger.info(
      "APRS packet cleanup complete: removed #{deleted_count} packets and #{bad_packet_count} bad packets older than #{retention_days} days"
    )

    :ok
  rescue
    error ->
      Logger.error("Packet cleanup failed: #{inspect(error)}\n#{inspect(__STACKTRACE__)}")
      {:error, "Cleanup failed: #{inspect(error)}"}
  end

  @doc """
  Perform cleanup of packets older than a specific number of days using batch processing.

  ## Parameters
  - `days` - Number of days to retain packets (packets older than this will be deleted)

  ## Returns
  - {deleted_count, batches_processed}
  """
  @spec cleanup_packets_older_than_batched(pos_integer()) :: {non_neg_integer(), non_neg_integer()}
  def cleanup_packets_older_than_batched(days) when is_integer(days) and days > 0 do
    Logger.info("Starting batched APRS packet cleanup for packets older than #{days} days")

    cutoff_time = DateTime.add(DateTime.utc_now(), -days * 86_400, :second)
    start_time = System.monotonic_time(:millisecond)

    {deleted_count, batches_processed} = cleanup_packets_in_batches(cutoff_time, start_time)

    duration = System.monotonic_time(:millisecond) - start_time

    :telemetry.execute(
      [:aprsme, :packet_cleanup, :complete],
      %{deleted: deleted_count, batches: batches_processed, duration_ms: duration},
      %{retention_days: days}
    )

    Logger.info(
      "APRS packet cleanup complete: removed #{deleted_count} packets in #{batches_processed} batches over #{duration}ms"
    )

    {deleted_count, batches_processed}
  end

  @doc """
  Clean packets older than a specific number of days.

  ## Parameters
  - `days` - Number of days to keep (packets older than this will be deleted)

  ## Returns
  - Number of packets deleted
  """
  @spec cleanup_packets_older_than(pos_integer()) :: non_neg_integer()
  def cleanup_packets_older_than(days) when is_integer(days) and days > 0 do
    {deleted_count, _} = cleanup_packets_older_than_batched(days)
    deleted_count
  end

  @doc """
  Clean packets in batches to prevent long-running transactions.

  Uses a single DELETE query with a subquery LIMIT per batch,
  avoiding the overhead of separate SELECT + DELETE round-trips.

  ## Parameters
  - `cutoff_time` - DateTime before which packets should be deleted
  - `start_time` - Monotonic time when cleanup started (for duration tracking)

  ## Returns
  - {total_deleted_count, batches_processed}
  """
  @spec cleanup_packets_in_batches(DateTime.t(), integer()) :: {non_neg_integer(), non_neg_integer()}
  def cleanup_packets_in_batches(cutoff_time, start_time) do
    cleanup_packets_in_batches(cutoff_time, start_time, 0, 0)
  end

  @doc """
  Get packet IDs for deletion in batches.

  ## Parameters
  - `cutoff_time` - DateTime before which packets should be deleted
  - `batch_size` - Maximum number of IDs to return

  ## Returns
  - List of packet IDs to delete
  """
  @spec get_packet_ids_for_deletion(DateTime.t(), pos_integer()) :: [integer()]
  def get_packet_ids_for_deletion(cutoff_time, batch_size) do
    Repo.all(
      from(p in Packet,
        where: p.received_at < ^cutoff_time,
        select: p.id,
        limit: ^batch_size
      )
    )
  rescue
    error ->
      Logger.error("Failed to get packet IDs for deletion: #{inspect(error)}")
      []
  end

  # Private functions

  defp cleanup_old_packets_batched do
    retention_days = Application.get_env(:aprsme, :packet_retention_days, 7)
    cleanup_packets_older_than_batched(retention_days)
  end

  defp cleanup_old_bad_packets do
    retention_days = Application.get_env(:aprsme, :packet_retention_days, 7)
    cutoff_time = DateTime.add(DateTime.utc_now(), -retention_days * 86_400, :second)

    {deleted_count, _} =
      Repo.delete_all(from(b in BadPacket, where: b.attempted_at < ^cutoff_time))

    deleted_count
  rescue
    error ->
      Logger.error("Failed to clean up bad packets: #{inspect(error)}")
      0
  end

  defp cleanup_packets_in_batches(cutoff_time, start_time, total_deleted, batches_processed) do
    current_time = System.monotonic_time(:millisecond)

    if current_time - start_time > @max_cleanup_time do
      Logger.info("Cleanup time limit reached (#{@max_cleanup_time}ms), stopping batch processing")
      {total_deleted, batches_processed}
    else
      try do
        deleted_count = delete_batch(cutoff_time, @batch_size)

        if deleted_count == 0 do
          {total_deleted, batches_processed}
        else
          new_total = total_deleted + deleted_count
          new_batches = batches_processed + 1

          Logger.debug("Cleanup batch #{new_batches}: deleted #{deleted_count} packets (total: #{new_total})")

          cleanup_packets_in_batches(cutoff_time, start_time, new_total, new_batches)
        end
      rescue
        error ->
          Logger.error("Failed to delete batch of packets: #{inspect(error)}")
          {total_deleted, batches_processed}
      end
    end
  end

  # Single-query batch delete using a CTE subquery.
  # This is significantly faster than SELECT IDs + DELETE by IDs
  # because it's a single database round-trip and allows PostgreSQL
  # to optimize the delete plan (sequential scan on received_at index).
  @spec delete_batch(DateTime.t(), pos_integer()) :: non_neg_integer()
  defp delete_batch(cutoff_time, batch_size) do
    # Use raw SQL for the CTE-based batch delete since Ecto doesn't
    # natively support DELETE ... WHERE id IN (SELECT ... LIMIT ...)
    {:ok, %{num_rows: deleted_count}} =
      Repo.query(
        """
        DELETE FROM packets
        WHERE id IN (
          SELECT id FROM packets
          WHERE received_at < $1
          LIMIT $2
        )
        """,
        [cutoff_time, batch_size]
      )

    deleted_count
  end
end