Partition packets table by day and simplify cleanup
Convert the packets table to a time-partitioned table with daily partitions (PARTITION BY RANGE on received_at). This replaces the expensive CTE-based batch DELETE cleanup with instant DROP TABLE partition drops — no dead tuples, no WAL pressure, no VACUUM needed. - Add PartitionManager GenServer to create/drop daily partitions - Simplify PacketCleanupWorker to delegate to PartitionManager - Replace gridsquare/geocalc deps with local Maidenhead module - Remove gettext_pseudolocalize dependency - Fix Decimal comparison in packet consumer test
This commit is contained in:
parent
d5a1f3fe1e
commit
b08eba2c6b
14 changed files with 814 additions and 296 deletions
|
|
@ -123,8 +123,6 @@ config :esbuild,
|
|||
cd: Path.expand("../assets", __DIR__)
|
||||
]
|
||||
|
||||
config :gettext, :plural_forms, GettextPseudolocalize.Plural
|
||||
|
||||
# Configures Elixir's Logger
|
||||
config :logger, :console,
|
||||
format: "$time $metadata[$level] $message\n",
|
||||
|
|
|
|||
|
|
@ -35,6 +35,8 @@ defmodule Aprsme.Application do
|
|||
Aprsme.SpatialPubSub,
|
||||
# Start global streaming packets PubSub
|
||||
Aprsme.StreamingPacketsPubSub,
|
||||
# Manage daily partitions for the packets table
|
||||
Aprsme.PartitionManager,
|
||||
# Start the Endpoint (http/https)
|
||||
AprsmeWeb.Endpoint,
|
||||
# Start a worker by calling: Aprsme.Worker.start_link(arg)
|
||||
|
|
@ -49,10 +51,12 @@ defmodule Aprsme.Application do
|
|||
Aprsme.PacketPipelineSupervisor
|
||||
]
|
||||
|
||||
# Skip packet pipeline in test to avoid Sandbox ownership errors
|
||||
# Skip partition manager and packet pipeline in test to avoid Sandbox ownership errors
|
||||
children =
|
||||
if Application.get_env(:aprsme, :env) == :test do
|
||||
List.delete(children, Aprsme.PacketPipelineSupervisor)
|
||||
children
|
||||
|> List.delete(Aprsme.PartitionManager)
|
||||
|> List.delete(Aprsme.PacketPipelineSupervisor)
|
||||
else
|
||||
children
|
||||
end
|
||||
|
|
|
|||
33
lib/aprsme/maidenhead.ex
Normal file
33
lib/aprsme/maidenhead.ex
Normal file
|
|
@ -0,0 +1,33 @@
|
|||
defmodule Aprsme.Maidenhead do
|
||||
@moduledoc """
|
||||
Encodes longitude/latitude to Maidenhead grid square locators (6 characters).
|
||||
"""
|
||||
|
||||
@spec encode(number() | nil, number() | nil) :: String.t() | nil
|
||||
def encode(lon, lat) when is_number(lon) and is_number(lat) do
|
||||
lon = lon + 180.0
|
||||
lat = lat + 90.0
|
||||
|
||||
field_lon = trunc(lon / 20)
|
||||
field_lat = trunc(lat / 10)
|
||||
|
||||
lon = lon - field_lon * 20
|
||||
lat = lat - field_lat * 10
|
||||
|
||||
square_lon = trunc(lon / 2)
|
||||
square_lat = trunc(lat)
|
||||
|
||||
lon = lon - square_lon * 2
|
||||
lat = lat - square_lat
|
||||
|
||||
sub_lon = trunc(lon / (2 / 24))
|
||||
sub_lat = trunc(lat / (1 / 24))
|
||||
|
||||
sub_lon = min(sub_lon, 23)
|
||||
sub_lat = min(sub_lat, 23)
|
||||
|
||||
<<?A + field_lon, ?A + field_lat, ?0 + square_lon, ?0 + square_lat, ?a + sub_lon, ?a + sub_lat>>
|
||||
end
|
||||
|
||||
def encode(_lon, _lat), do: nil
|
||||
end
|
||||
192
lib/aprsme/partition_manager.ex
Normal file
192
lib/aprsme/partition_manager.ex
Normal file
|
|
@ -0,0 +1,192 @@
|
|||
defmodule Aprsme.PartitionManager do
|
||||
@moduledoc """
|
||||
Manages daily partitions for the packets table.
|
||||
|
||||
Creates future partitions and drops old ones past the retention period.
|
||||
Partition naming: packets_YYYYMMDD (e.g., packets_20260220).
|
||||
Each partition covers [YYYY-MM-DD 00:00:00, YYYY-MM-DD+1 00:00:00) UTC.
|
||||
"""
|
||||
|
||||
use GenServer
|
||||
|
||||
alias Aprsme.Repo
|
||||
|
||||
require Logger
|
||||
|
||||
@check_interval to_timeout(hour: 1)
|
||||
@future_days 2
|
||||
|
||||
def start_link(opts \\ []) do
|
||||
GenServer.start_link(__MODULE__, opts, name: __MODULE__)
|
||||
end
|
||||
|
||||
@doc """
|
||||
Returns the partition table name for a given date.
|
||||
"""
|
||||
@spec partition_name(Date.t()) :: String.t()
|
||||
def partition_name(%Date{} = date) do
|
||||
"packets_#{Calendar.strftime(date, "%Y%m%d")}"
|
||||
end
|
||||
|
||||
@doc """
|
||||
Returns the {start, end} UTC datetime range for a partition covering the given date.
|
||||
"""
|
||||
@spec partition_range(Date.t()) :: {DateTime.t(), DateTime.t()}
|
||||
def partition_range(%Date{} = date) do
|
||||
{:ok, start_dt} = DateTime.new(date, ~T[00:00:00], "Etc/UTC")
|
||||
next_date = Date.add(date, 1)
|
||||
{:ok, end_dt} = DateTime.new(next_date, ~T[00:00:00], "Etc/UTC")
|
||||
{start_dt, end_dt}
|
||||
end
|
||||
|
||||
@doc """
|
||||
Ensures partitions exist for today through today + #{@future_days} days.
|
||||
Returns {:ok, list_of_created_partition_names}.
|
||||
"""
|
||||
@spec ensure_partitions_exist() :: {:ok, [String.t()]}
|
||||
def ensure_partitions_exist do
|
||||
today = Date.utc_today()
|
||||
|
||||
created =
|
||||
for offset <- 0..@future_days, reduce: [] do
|
||||
acc ->
|
||||
date = Date.add(today, offset)
|
||||
name = partition_name(date)
|
||||
|
||||
if partition_exists?(name) do
|
||||
acc
|
||||
else
|
||||
create_partition(date)
|
||||
[name | acc]
|
||||
end
|
||||
end
|
||||
|
||||
{:ok, Enum.reverse(created)}
|
||||
end
|
||||
|
||||
@doc """
|
||||
Drops partitions older than the given number of retention days.
|
||||
Returns {:ok, list_of_dropped_partition_names}.
|
||||
"""
|
||||
@spec drop_old_partitions(pos_integer()) :: {:ok, [String.t()]}
|
||||
def drop_old_partitions(retention_days \\ 7) do
|
||||
cutoff_date = Date.add(Date.utc_today(), -retention_days)
|
||||
|
||||
dropped =
|
||||
list_partitions()
|
||||
|> Enum.filter(fn name ->
|
||||
case partition_date(name) do
|
||||
{:ok, date} -> Date.before?(date, cutoff_date)
|
||||
_ -> false
|
||||
end
|
||||
end)
|
||||
|> Enum.map(fn name ->
|
||||
drop_partition(name)
|
||||
name
|
||||
end)
|
||||
|
||||
{:ok, dropped}
|
||||
end
|
||||
|
||||
@doc """
|
||||
Lists all existing partition table names for the packets table.
|
||||
"""
|
||||
@spec list_partitions() :: [String.t()]
|
||||
def list_partitions do
|
||||
case Repo.query(
|
||||
"SELECT inhrelid::regclass::text FROM pg_inherits WHERE inhparent = 'packets'::regclass ORDER BY inhrelid::regclass::text"
|
||||
) do
|
||||
{:ok, %{rows: rows}} ->
|
||||
Enum.map(rows, fn [name] ->
|
||||
String.replace(name, ~r/^public\./, "")
|
||||
end)
|
||||
|
||||
{:error, _} ->
|
||||
[]
|
||||
end
|
||||
end
|
||||
|
||||
# GenServer callbacks
|
||||
|
||||
@impl true
|
||||
def init(_opts) do
|
||||
if Application.get_env(:aprsme, :env) != :test do
|
||||
send(self(), :manage_partitions)
|
||||
end
|
||||
|
||||
{:ok, %{}}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_info(:manage_partitions, state) do
|
||||
manage_partitions()
|
||||
Process.send_after(self(), :manage_partitions, @check_interval)
|
||||
{:noreply, state}
|
||||
end
|
||||
|
||||
defp manage_partitions do
|
||||
case ensure_partitions_exist() do
|
||||
{:ok, []} ->
|
||||
:ok
|
||||
|
||||
{:ok, created} ->
|
||||
Logger.info("Created #{length(created)} partition(s): #{Enum.join(created, ", ")}")
|
||||
end
|
||||
|
||||
retention_days = Application.get_env(:aprsme, :packet_retention_days, 7)
|
||||
|
||||
case drop_old_partitions(retention_days) do
|
||||
{:ok, []} ->
|
||||
:ok
|
||||
|
||||
{:ok, dropped} ->
|
||||
Logger.info("Dropped #{length(dropped)} old partition(s): #{Enum.join(dropped, ", ")}")
|
||||
end
|
||||
rescue
|
||||
error ->
|
||||
Logger.error("Partition management failed: #{inspect(error)}")
|
||||
end
|
||||
|
||||
defp partition_exists?(name) do
|
||||
case Repo.query(
|
||||
"SELECT 1 FROM pg_tables WHERE tablename = $1 AND schemaname = 'public'",
|
||||
[name]
|
||||
) do
|
||||
{:ok, %{num_rows: n}} when n > 0 -> true
|
||||
_ -> false
|
||||
end
|
||||
end
|
||||
|
||||
defp create_partition(%Date{} = date) do
|
||||
name = partition_name(date)
|
||||
{from_dt, to_dt} = partition_range(date)
|
||||
|
||||
from_str = DateTime.to_iso8601(from_dt)
|
||||
to_str = DateTime.to_iso8601(to_dt)
|
||||
|
||||
Repo.query!(
|
||||
"CREATE TABLE IF NOT EXISTS #{name} PARTITION OF packets FOR VALUES FROM ('#{from_str}') TO ('#{to_str}')"
|
||||
)
|
||||
|
||||
Logger.debug("Created partition #{name} [#{from_str}, #{to_str})")
|
||||
name
|
||||
end
|
||||
|
||||
defp drop_partition(name) do
|
||||
Repo.query!("DROP TABLE IF EXISTS #{name}")
|
||||
Logger.debug("Dropped partition #{name}")
|
||||
name
|
||||
end
|
||||
|
||||
defp partition_date("packets_" <> date_str) do
|
||||
case Date.from_iso8601(
|
||||
String.slice(date_str, 0, 4) <>
|
||||
"-" <> String.slice(date_str, 4, 2) <> "-" <> String.slice(date_str, 6, 2)
|
||||
) do
|
||||
{:ok, date} -> {:ok, date}
|
||||
_ -> nil
|
||||
end
|
||||
end
|
||||
|
||||
defp partition_date(_), do: nil
|
||||
end
|
||||
|
|
@ -2,41 +2,27 @@ 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
|
||||
Delegates to `PartitionManager.drop_old_partitions/1` which drops entire
|
||||
daily partitions past the retention period — instant, no dead tuples, no VACUUM.
|
||||
|
||||
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
|
||||
Also cleans up old `bad_packets` records via standard DELETE.
|
||||
"""
|
||||
|
||||
import Ecto.Query
|
||||
|
||||
alias Aprsme.BadPacket
|
||||
alias Aprsme.Packet
|
||||
alias Aprsme.PartitionManager
|
||||
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
|
||||
def perform(%{"cleanup_days" => days}) when is_integer(days) and days > 0 do
|
||||
Logger.info("Starting scheduled APRS packet cleanup for packets older than #{days} days")
|
||||
|
||||
{deleted_count, _} = cleanup_packets_older_than_batched(days)
|
||||
{:ok, dropped} = PartitionManager.drop_old_partitions(days)
|
||||
|
||||
Logger.info("APRS packet cleanup complete: removed #{deleted_count} packets older than #{days} days")
|
||||
Logger.info("APRS packet cleanup complete: dropped #{length(dropped)} partition(s) older than #{days} days")
|
||||
|
||||
:ok
|
||||
rescue
|
||||
|
|
@ -49,14 +35,13 @@ defmodule Aprsme.Workers.PacketCleanupWorker do
|
|||
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)
|
||||
|
||||
{:ok, dropped} = PartitionManager.drop_old_partitions(retention_days)
|
||||
bad_packet_count = cleanup_old_bad_packets()
|
||||
|
||||
Logger.info(
|
||||
"APRS packet cleanup complete: removed #{deleted_count} packets and #{bad_packet_count} bad packets older than #{retention_days} days"
|
||||
"APRS packet cleanup complete: dropped #{length(dropped)} partition(s) and removed #{bad_packet_count} bad packets older than #{retention_days} days"
|
||||
)
|
||||
|
||||
:ok
|
||||
|
|
@ -66,104 +51,6 @@ defmodule Aprsme.Workers.PacketCleanupWorker do
|
|||
{: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)
|
||||
|
|
@ -177,56 +64,4 @@ defmodule Aprsme.Workers.PacketCleanupWorker do
|
|||
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
|
||||
|
|
|
|||
|
|
@ -89,12 +89,7 @@
|
|||
<div>
|
||||
<dt class="text-xs font-medium text-gray-500 dark:text-gray-400">{gettext("Grid Square")}</dt>
|
||||
<dd class="mt-1 text-sm font-semibold">
|
||||
<%= case Gridsquare.encode(@packet.lon, @packet.lat) do %>
|
||||
<% %Gridsquare.EncodeResult{grid_reference: grid_ref} -> %>
|
||||
{grid_ref}
|
||||
<% _ -> %>
|
||||
-
|
||||
<% end %>
|
||||
{Aprsme.Maidenhead.encode(@packet.lon, @packet.lat) || "-"}
|
||||
</dd>
|
||||
</div>
|
||||
<% end %>
|
||||
|
|
|
|||
5
mix.exs
5
mix.exs
|
|
@ -65,7 +65,6 @@ defmodule Aprsme.MixProject do
|
|||
{:ecto_sql, "~> 3.11"},
|
||||
{:geo, "~> 4.1"},
|
||||
{:geo_postgis, "~> 3.4"},
|
||||
{:geocalc, "~> 0.8"},
|
||||
{:gen_stage, "~> 1.2"},
|
||||
{:gettext, "~> 0.26"},
|
||||
{:jason, "~> 1.4"},
|
||||
|
|
@ -88,7 +87,6 @@ defmodule Aprsme.MixProject do
|
|||
github: "tailwindlabs/heroicons", tag: "v2.1.1", sparse: "optimized", app: false, compile: false, depth: 1},
|
||||
{:bandit, "~> 1.5"},
|
||||
{:req, "~> 0.5"},
|
||||
{:gridsquare, "~> 0.3.1"},
|
||||
{:credo, "~> 1.7", only: [:dev, :test], runtime: false},
|
||||
{:dialyxir, "~> 1.0", only: :dev, runtime: false},
|
||||
{:exvcr, "~> 0.15", only: :test},
|
||||
|
|
@ -100,8 +98,7 @@ defmodule Aprsme.MixProject do
|
|||
{:igniter, "~> 0.7.0", only: [:dev, :test]},
|
||||
{:mox, "~> 1.2", only: :test},
|
||||
{:styler, "~> 1.10", only: :dev, runtime: false},
|
||||
{:hammer, "~> 7.0"},
|
||||
{:gettext_pseudolocalize, "~> 0.1"}
|
||||
{:hammer, "~> 7.0"}
|
||||
]
|
||||
end
|
||||
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@
|
|||
msgid ""
|
||||
msgstr ""
|
||||
"Language: xx\n"
|
||||
"Plural-Forms: nplurals=2\n"
|
||||
"Plural-Forms: nplurals=2; plural=(n != 1);\n"
|
||||
|
||||
#: lib/aprsme_web/components/core_components.ex:558
|
||||
#, elixir-autogen, elixir-format
|
||||
|
|
|
|||
|
|
@ -9,7 +9,7 @@
|
|||
msgid ""
|
||||
msgstr ""
|
||||
"Language: xx\n"
|
||||
"Plural-Forms: nplurals=2\n"
|
||||
"Plural-Forms: nplurals=2; plural=(n != 1);\n"
|
||||
|
||||
msgid "can't be blank"
|
||||
msgstr "⟦ċàñ'ť ƀê ƀĺàñǩ~~~~~~⟧"
|
||||
|
|
|
|||
366
priv/repo/migrations/20260220200000_partition_packets_table.exs
Normal file
366
priv/repo/migrations/20260220200000_partition_packets_table.exs
Normal file
|
|
@ -0,0 +1,366 @@
|
|||
defmodule Aprsme.Repo.Migrations.PartitionPacketsTable do
|
||||
use Ecto.Migration
|
||||
|
||||
def up do
|
||||
# Disable statement timeout for DDL on large table with many indexes
|
||||
execute("SET LOCAL statement_timeout = '0'")
|
||||
|
||||
# Drop existing non-partitioned table (data is ephemeral, refills from APRS-IS stream)
|
||||
execute("DROP TABLE IF EXISTS packets CASCADE")
|
||||
|
||||
# Create partitioned table with composite PK required for partitioning
|
||||
execute("""
|
||||
CREATE TABLE packets (
|
||||
id uuid NOT NULL,
|
||||
base_callsign varchar,
|
||||
data_type varchar,
|
||||
destination varchar,
|
||||
information_field text,
|
||||
path text,
|
||||
sender varchar,
|
||||
ssid varchar,
|
||||
data_extended jsonb,
|
||||
inserted_at timestamp NOT NULL,
|
||||
updated_at timestamp NOT NULL,
|
||||
received_at timestamp NOT NULL,
|
||||
region varchar,
|
||||
lat numeric,
|
||||
lon numeric,
|
||||
has_position boolean DEFAULT false,
|
||||
item_name text,
|
||||
object_name text,
|
||||
is_item boolean DEFAULT false,
|
||||
is_object boolean DEFAULT false,
|
||||
dao jsonb,
|
||||
raw_packet text,
|
||||
symbol_code varchar,
|
||||
symbol_table_id varchar,
|
||||
comment text,
|
||||
"timestamp" varchar,
|
||||
aprs_messaging boolean DEFAULT false,
|
||||
temperature double precision,
|
||||
humidity double precision,
|
||||
wind_speed double precision,
|
||||
wind_direction integer,
|
||||
wind_gust double precision,
|
||||
pressure double precision,
|
||||
rain_1h double precision,
|
||||
rain_24h double precision,
|
||||
rain_since_midnight double precision,
|
||||
manufacturer text,
|
||||
equipment_type text,
|
||||
course integer,
|
||||
speed double precision,
|
||||
altitude double precision,
|
||||
addressee text,
|
||||
message_text text,
|
||||
message_number varchar,
|
||||
location geometry(Point,4326),
|
||||
device_identifier text,
|
||||
luminosity integer,
|
||||
snow double precision,
|
||||
phg_power integer,
|
||||
phg_height integer,
|
||||
phg_gain integer,
|
||||
phg_directivity integer,
|
||||
has_weather boolean DEFAULT false,
|
||||
rain_midnight double precision,
|
||||
position_ambiguity integer,
|
||||
posresolution double precision,
|
||||
format text,
|
||||
telemetry_seq integer,
|
||||
telemetry_vals integer[],
|
||||
telemetry_bits text,
|
||||
radiorange text,
|
||||
srccallsign varchar,
|
||||
dstcallsign varchar,
|
||||
body text,
|
||||
origpacket text,
|
||||
header text,
|
||||
alive integer DEFAULT 1,
|
||||
posambiguity integer,
|
||||
symboltable varchar,
|
||||
symbolcode varchar,
|
||||
messaging integer,
|
||||
PRIMARY KEY (id, received_at)
|
||||
) PARTITION BY RANGE (received_at)
|
||||
""")
|
||||
|
||||
# Default partition catches rows outside named partitions
|
||||
execute("CREATE TABLE packets_default PARTITION OF packets DEFAULT")
|
||||
|
||||
# Create daily partitions for recent past + near future
|
||||
today = Date.utc_today()
|
||||
|
||||
for offset <- -8..2 do
|
||||
date = Date.add(today, offset)
|
||||
name = "packets_#{Calendar.strftime(date, "%Y%m%d")}"
|
||||
next_date = Date.add(date, 1)
|
||||
|
||||
execute(
|
||||
"CREATE TABLE IF NOT EXISTS #{name} PARTITION OF packets FOR VALUES FROM ('#{date} 00:00:00') TO ('#{next_date} 00:00:00')"
|
||||
)
|
||||
end
|
||||
|
||||
create_indexes()
|
||||
create_triggers("packets")
|
||||
end
|
||||
|
||||
def down do
|
||||
execute("DROP TABLE IF EXISTS packets CASCADE")
|
||||
|
||||
# Recreate as regular non-partitioned table
|
||||
execute("""
|
||||
CREATE TABLE packets (
|
||||
id uuid NOT NULL PRIMARY KEY,
|
||||
base_callsign varchar,
|
||||
data_type varchar,
|
||||
destination varchar,
|
||||
information_field text,
|
||||
path text,
|
||||
sender varchar,
|
||||
ssid varchar,
|
||||
data_extended jsonb,
|
||||
inserted_at timestamp NOT NULL,
|
||||
updated_at timestamp NOT NULL,
|
||||
received_at timestamp NOT NULL,
|
||||
region varchar,
|
||||
lat numeric,
|
||||
lon numeric,
|
||||
has_position boolean DEFAULT false,
|
||||
item_name text,
|
||||
object_name text,
|
||||
is_item boolean DEFAULT false,
|
||||
is_object boolean DEFAULT false,
|
||||
dao jsonb,
|
||||
raw_packet text,
|
||||
symbol_code varchar,
|
||||
symbol_table_id varchar,
|
||||
comment text,
|
||||
"timestamp" varchar,
|
||||
aprs_messaging boolean DEFAULT false,
|
||||
temperature double precision,
|
||||
humidity double precision,
|
||||
wind_speed double precision,
|
||||
wind_direction integer,
|
||||
wind_gust double precision,
|
||||
pressure double precision,
|
||||
rain_1h double precision,
|
||||
rain_24h double precision,
|
||||
rain_since_midnight double precision,
|
||||
manufacturer text,
|
||||
equipment_type text,
|
||||
course integer,
|
||||
speed double precision,
|
||||
altitude double precision,
|
||||
addressee text,
|
||||
message_text text,
|
||||
message_number varchar,
|
||||
location geometry(Point,4326),
|
||||
device_identifier text,
|
||||
luminosity integer,
|
||||
snow double precision,
|
||||
phg_power integer,
|
||||
phg_height integer,
|
||||
phg_gain integer,
|
||||
phg_directivity integer,
|
||||
has_weather boolean DEFAULT false,
|
||||
rain_midnight double precision,
|
||||
position_ambiguity integer,
|
||||
posresolution double precision,
|
||||
format text,
|
||||
telemetry_seq integer,
|
||||
telemetry_vals integer[],
|
||||
telemetry_bits text,
|
||||
radiorange text,
|
||||
srccallsign varchar,
|
||||
dstcallsign varchar,
|
||||
body text,
|
||||
origpacket text,
|
||||
header text,
|
||||
alive integer DEFAULT 1,
|
||||
posambiguity integer,
|
||||
symboltable varchar,
|
||||
symbolcode varchar,
|
||||
messaging integer
|
||||
)
|
||||
""")
|
||||
|
||||
create_triggers("packets")
|
||||
end
|
||||
|
||||
defp create_indexes do
|
||||
# Time-series
|
||||
execute("CREATE INDEX idx_packets_received_desc ON packets (received_at DESC)")
|
||||
execute("CREATE INDEX idx_packets_received_brin ON packets USING brin (received_at)")
|
||||
|
||||
# Sender/Callsign
|
||||
execute("CREATE INDEX idx_packets_sender_received ON packets (sender, received_at DESC)")
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_upper_sender ON packets (upper(sender::text), received_at DESC)"
|
||||
)
|
||||
|
||||
execute("CREATE INDEX idx_packets_sender_pattern ON packets (sender text_pattern_ops)")
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_base_callsign_pattern ON packets (base_callsign text_pattern_ops)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_base_callsign_time ON packets (base_callsign, received_at DESC)"
|
||||
)
|
||||
|
||||
execute("CREATE INDEX idx_packets_sender_id_desc ON packets (sender, id DESC)")
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_upper_base_callsign ON packets (upper(base_callsign::text))"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_sender_trgm ON packets USING gin (sender gin_trgm_ops) WHERE sender IS NOT NULL"
|
||||
)
|
||||
|
||||
# Position/Geographic
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_positioned ON packets (received_at DESC, lat, lon, sender) WHERE has_position = true"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_position_time ON packets (has_position, received_at DESC) WHERE has_position = true"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_location ON packets USING gist (location) WHERE has_position = true"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_location_geography ON packets USING gist ((location::geography))"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_spatial_temporal ON packets (received_at DESC, base_callsign) WHERE has_position = true"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_sender_position ON packets (sender, received_at DESC) WHERE lat IS NOT NULL AND lon IS NOT NULL"
|
||||
)
|
||||
|
||||
execute("CREATE INDEX idx_packets_lat ON packets (lat)")
|
||||
execute("CREATE INDEX idx_packets_lon ON packets (lon)")
|
||||
|
||||
# Region
|
||||
execute("CREATE INDEX idx_packets_region ON packets (region)")
|
||||
execute("CREATE INDEX idx_packets_region_received ON packets (region, received_at)")
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_region_position_time ON packets (region, has_position, received_at DESC) WHERE has_position = true AND region IS NOT NULL"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_region_data_type ON packets (region, data_type, received_at)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_region_position ON packets (region, has_position, received_at)"
|
||||
)
|
||||
|
||||
# Weather
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_weather ON packets (sender, received_at DESC) WHERE temperature IS NOT NULL OR humidity IS NOT NULL OR pressure IS NOT NULL OR wind_speed IS NOT NULL OR wind_direction IS NOT NULL OR rain_1h IS NOT NULL"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_has_weather ON packets (has_weather, sender, received_at DESC) WHERE has_weather = true"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_temperature ON packets (temperature) WHERE temperature IS NOT NULL"
|
||||
)
|
||||
|
||||
execute("CREATE INDEX idx_packets_humidity ON packets (humidity) WHERE humidity IS NOT NULL")
|
||||
execute("CREATE INDEX idx_packets_pressure ON packets (pressure) WHERE pressure IS NOT NULL")
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_wind_speed ON packets (wind_speed) WHERE wind_speed IS NOT NULL"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_received_temp ON packets (received_at, temperature) WHERE temperature IS NOT NULL"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_sender_temp ON packets (sender, temperature) WHERE temperature IS NOT NULL"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_weather_history ON packets (sender, data_type, received_at) WHERE data_type = 'weather'"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_weather_selective ON packets (received_at DESC) WHERE data_type IN ('weather', 'Weather', 'WX', 'wx')"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_weather_symbol ON packets (received_at DESC, lat, lon) WHERE (symbol_table_id = '/' AND symbol_code = '_') OR (symbol_table_id = E'\\\\' AND symbol_code = '_')"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_sender_weather_lookup ON packets (upper(sender::text), received_at DESC) INCLUDE (data_type, symbol_table_id, symbol_code) WHERE data_type = 'weather' OR (symbol_table_id = '/' AND symbol_code = '_') OR (symbol_table_id = E'\\\\' AND symbol_code = '_')"
|
||||
)
|
||||
|
||||
# Data type & Device
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_datatype_sender_time ON packets (data_type, sender, received_at DESC)"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_device_identifier ON packets (device_identifier) WHERE device_identifier IS NOT NULL"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_device_received ON packets (device_identifier, received_at) WHERE device_identifier IS NOT NULL"
|
||||
)
|
||||
|
||||
# Path
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_path_pattern ON packets (path text_pattern_ops) WHERE path IS NOT NULL AND path <> ''"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_path_trgm ON packets USING gin (path gin_trgm_ops) WHERE path IS NOT NULL AND path <> '' AND path !~ '^TCPIP' AND path !~ ',TCPIP'"
|
||||
)
|
||||
|
||||
# Digipeater/stations heard
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_digipeater ON packets (sender, received_at DESC) WHERE location IS NOT NULL"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_stations_heard ON packets (received_at DESC, sender, lat, lon) WHERE path IS NOT NULL AND path <> '' AND path !~ '^TCPIP' AND path !~ ',TCPIP' AND lat IS NOT NULL AND lon IS NOT NULL"
|
||||
)
|
||||
|
||||
# Symbol
|
||||
execute(
|
||||
"CREATE INDEX idx_packets_symbol_time ON packets (symbol_table_id, symbol_code, received_at)"
|
||||
)
|
||||
end
|
||||
|
||||
defp create_triggers(table) do
|
||||
execute(
|
||||
"CREATE TRIGGER update_has_weather_trigger BEFORE INSERT OR UPDATE ON #{table} FOR EACH ROW EXECUTE FUNCTION update_has_weather()"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE TRIGGER packets_notify_insert AFTER INSERT ON #{table} FOR EACH ROW EXECUTE FUNCTION notify_packets_insert()"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE TRIGGER packet_insert_sequence AFTER INSERT ON #{table} FOR EACH ROW EXECUTE FUNCTION increment_packet_sequence()"
|
||||
)
|
||||
|
||||
execute(
|
||||
"CREATE TRIGGER packet_delete_sequence AFTER DELETE ON #{table} FOR EACH ROW EXECUTE FUNCTION decrement_packet_sequence()"
|
||||
)
|
||||
end
|
||||
end
|
||||
31
test/aprsme/maidenhead_test.exs
Normal file
31
test/aprsme/maidenhead_test.exs
Normal file
|
|
@ -0,0 +1,31 @@
|
|||
defmodule Aprsme.MaidenheadTest do
|
||||
use ExUnit.Case, async: true
|
||||
|
||||
alias Aprsme.Maidenhead
|
||||
|
||||
describe "encode/2" do
|
||||
test "encodes known locations to 6-character grid squares" do
|
||||
# Washington DC area
|
||||
assert Maidenhead.encode(-77.036, 38.897) == "FM18lv"
|
||||
# London
|
||||
assert Maidenhead.encode(-0.1257, 51.5085) == "IO91wm"
|
||||
# Tokyo
|
||||
assert Maidenhead.encode(139.6917, 35.6895) == "PM95uq"
|
||||
# Sydney
|
||||
assert Maidenhead.encode(151.2093, -33.8688) == "QF56od"
|
||||
end
|
||||
|
||||
test "encodes edge coordinates" do
|
||||
# South pole
|
||||
assert Maidenhead.encode(0.0, -90.0) == "JA00aa"
|
||||
# North pole area
|
||||
assert Maidenhead.encode(0.0, 89.99) == "JR09ax"
|
||||
end
|
||||
|
||||
test "returns nil for nil inputs" do
|
||||
assert Maidenhead.encode(nil, 0.0) == nil
|
||||
assert Maidenhead.encode(0.0, nil) == nil
|
||||
assert Maidenhead.encode(nil, nil) == nil
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
@ -191,7 +191,7 @@ defmodule Aprsme.PacketConsumerTest do
|
|||
valid2_packet = Enum.find(packets, &(&1.sender == "VALID2"))
|
||||
assert valid2_packet.location == nil
|
||||
# Invalid latitude stored
|
||||
assert valid2_packet.lat == Decimal.new("91.000000")
|
||||
assert Decimal.equal?(valid2_packet.lat, Decimal.new("91.0"))
|
||||
|
||||
# Check that VALID1 and VALID3 have valid locations
|
||||
valid_location_packets = Enum.filter(packets, &(&1.sender in ["VALID1", "VALID3"]))
|
||||
|
|
|
|||
122
test/aprsme/partition_manager_test.exs
Normal file
122
test/aprsme/partition_manager_test.exs
Normal file
|
|
@ -0,0 +1,122 @@
|
|||
defmodule Aprsme.PartitionManagerTest do
|
||||
use Aprsme.DataCase, async: false
|
||||
|
||||
alias Aprsme.PartitionManager
|
||||
|
||||
describe "partition_name/1" do
|
||||
test "generates correct name for a date" do
|
||||
assert PartitionManager.partition_name(~D[2026-02-20]) == "packets_20260220"
|
||||
assert PartitionManager.partition_name(~D[2026-01-01]) == "packets_20260101"
|
||||
assert PartitionManager.partition_name(~D[2025-12-31]) == "packets_20251231"
|
||||
end
|
||||
end
|
||||
|
||||
describe "partition_range/1" do
|
||||
test "returns start and end timestamps for a date" do
|
||||
{start_dt, end_dt} = PartitionManager.partition_range(~D[2026-02-20])
|
||||
assert start_dt == ~U[2026-02-20 00:00:00Z]
|
||||
assert end_dt == ~U[2026-02-21 00:00:00Z]
|
||||
end
|
||||
|
||||
test "handles month boundaries" do
|
||||
{_start_dt, end_dt} = PartitionManager.partition_range(~D[2026-01-31])
|
||||
assert end_dt == ~U[2026-02-01 00:00:00Z]
|
||||
end
|
||||
|
||||
test "handles year boundaries" do
|
||||
{_start_dt, end_dt} = PartitionManager.partition_range(~D[2025-12-31])
|
||||
assert end_dt == ~U[2026-01-01 00:00:00Z]
|
||||
end
|
||||
end
|
||||
|
||||
describe "ensure_partitions_exist/0" do
|
||||
test "creates partitions for today and next 2 days" do
|
||||
today = Date.utc_today()
|
||||
|
||||
{:ok, created} = PartitionManager.ensure_partitions_exist()
|
||||
|
||||
assert is_list(created)
|
||||
|
||||
# Verify partitions exist in the database
|
||||
for offset <- 0..2 do
|
||||
date = Date.add(today, offset)
|
||||
name = PartitionManager.partition_name(date)
|
||||
|
||||
assert {:ok, %{num_rows: 1}} =
|
||||
Repo.query(
|
||||
"SELECT 1 FROM pg_tables WHERE tablename = $1 AND schemaname = 'public'",
|
||||
[name]
|
||||
)
|
||||
end
|
||||
end
|
||||
|
||||
test "is idempotent - calling twice doesn't error" do
|
||||
{:ok, _created1} = PartitionManager.ensure_partitions_exist()
|
||||
{:ok, _created2} = PartitionManager.ensure_partitions_exist()
|
||||
end
|
||||
end
|
||||
|
||||
describe "drop_old_partitions/1" do
|
||||
test "drops partitions older than retention days" do
|
||||
# Create a partition for 30 days ago
|
||||
old_date = Date.add(Date.utc_today(), -30)
|
||||
name = PartitionManager.partition_name(old_date)
|
||||
{from_dt, to_dt} = PartitionManager.partition_range(old_date)
|
||||
|
||||
Repo.query!(
|
||||
"CREATE TABLE IF NOT EXISTS #{name} PARTITION OF packets FOR VALUES FROM ('#{DateTime.to_iso8601(from_dt)}') TO ('#{DateTime.to_iso8601(to_dt)}')"
|
||||
)
|
||||
|
||||
# Verify it exists
|
||||
assert {:ok, %{num_rows: 1}} =
|
||||
Repo.query(
|
||||
"SELECT 1 FROM pg_tables WHERE tablename = $1 AND schemaname = 'public'",
|
||||
[name]
|
||||
)
|
||||
|
||||
# Drop partitions older than 7 days
|
||||
{:ok, dropped} = PartitionManager.drop_old_partitions(7)
|
||||
|
||||
assert name in dropped
|
||||
|
||||
# Verify it's gone
|
||||
assert {:ok, %{num_rows: 0}} =
|
||||
Repo.query(
|
||||
"SELECT 1 FROM pg_tables WHERE tablename = $1 AND schemaname = 'public'",
|
||||
[name]
|
||||
)
|
||||
end
|
||||
|
||||
test "does not drop recent partitions" do
|
||||
# Ensure today's partition exists
|
||||
{:ok, _} = PartitionManager.ensure_partitions_exist()
|
||||
|
||||
today_name = PartitionManager.partition_name(Date.utc_today())
|
||||
|
||||
{:ok, dropped} = PartitionManager.drop_old_partitions(7)
|
||||
|
||||
refute today_name in dropped
|
||||
|
||||
# Verify today's partition still exists
|
||||
assert {:ok, %{num_rows: 1}} =
|
||||
Repo.query(
|
||||
"SELECT 1 FROM pg_tables WHERE tablename = $1 AND schemaname = 'public'",
|
||||
[today_name]
|
||||
)
|
||||
end
|
||||
end
|
||||
|
||||
describe "list_partitions/0" do
|
||||
test "returns list of existing partition names" do
|
||||
{:ok, _} = PartitionManager.ensure_partitions_exist()
|
||||
|
||||
partitions = PartitionManager.list_partitions()
|
||||
|
||||
assert is_list(partitions)
|
||||
assert length(partitions) >= 3
|
||||
|
||||
today_name = PartitionManager.partition_name(Date.utc_today())
|
||||
assert today_name in partitions
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
@ -1,134 +1,79 @@
|
|||
defmodule Aprsme.Workers.PacketCleanupWorkerTest do
|
||||
use Aprsme.DataCase, async: true
|
||||
use Aprsme.DataCase, async: false
|
||||
|
||||
alias Aprsme.Packet
|
||||
alias Aprsme.BadPacket
|
||||
alias Aprsme.PartitionManager
|
||||
alias Aprsme.Workers.PacketCleanupWorker
|
||||
|
||||
defp insert_packet(attrs) do
|
||||
attrs =
|
||||
Map.merge(
|
||||
%{
|
||||
sender: "TEST-#{System.unique_integer([:positive])}",
|
||||
received_at: DateTime.truncate(DateTime.utc_now(), :second),
|
||||
has_position: false
|
||||
},
|
||||
attrs
|
||||
describe "perform/1 with cleanup_days" do
|
||||
test "drops old partitions via PartitionManager" do
|
||||
# Ensure current partitions exist
|
||||
{:ok, _} = PartitionManager.ensure_partitions_exist()
|
||||
|
||||
# Create an old partition (30 days ago)
|
||||
old_date = Date.add(Date.utc_today(), -30)
|
||||
name = PartitionManager.partition_name(old_date)
|
||||
{from_dt, to_dt} = PartitionManager.partition_range(old_date)
|
||||
|
||||
Repo.query!(
|
||||
"CREATE TABLE IF NOT EXISTS #{name} PARTITION OF packets FOR VALUES FROM ('#{DateTime.to_iso8601(from_dt)}') TO ('#{DateTime.to_iso8601(to_dt)}')"
|
||||
)
|
||||
|
||||
{:ok, packet} = %Packet{} |> Map.merge(attrs) |> Repo.insert()
|
||||
packet
|
||||
end
|
||||
# Verify old partition exists
|
||||
assert name in PartitionManager.list_partitions()
|
||||
|
||||
defp insert_packets(count, attrs) do
|
||||
for _ <- 1..count, do: insert_packet(attrs)
|
||||
end
|
||||
assert :ok = PacketCleanupWorker.perform(%{"cleanup_days" => 7})
|
||||
|
||||
describe "perform/1 with cleanup_days" do
|
||||
test "cleans up packets older than the specified number of days" do
|
||||
days = 30
|
||||
job_args = %{"cleanup_days" => days}
|
||||
# Old partition should be dropped
|
||||
refute name in PartitionManager.list_partitions()
|
||||
end
|
||||
|
||||
# Create old packets that should be deleted
|
||||
old_time = DateTime.utc_now() |> DateTime.add(-(days + 1) * 86_400, :second) |> DateTime.truncate(:second)
|
||||
insert_packets(5, %{received_at: old_time})
|
||||
test "does not drop recent partitions" do
|
||||
{:ok, _} = PartitionManager.ensure_partitions_exist()
|
||||
|
||||
# Create recent packets that should NOT be deleted
|
||||
recent_time = DateTime.utc_now() |> DateTime.add(-5 * 86_400, :second) |> DateTime.truncate(:second)
|
||||
insert_packets(3, %{received_at: recent_time})
|
||||
today_name = PartitionManager.partition_name(Date.utc_today())
|
||||
|
||||
assert :ok = PacketCleanupWorker.perform(job_args)
|
||||
assert :ok = PacketCleanupWorker.perform(%{"cleanup_days" => 7})
|
||||
|
||||
# Verify only recent packets remain
|
||||
assert Repo.aggregate(Packet, :count, :id) == 3
|
||||
assert today_name in PartitionManager.list_partitions()
|
||||
end
|
||||
end
|
||||
|
||||
describe "perform/1 without cleanup_days" do
|
||||
test "cleans up packets using the default retention period" do
|
||||
retention_days = Application.get_env(:aprsme, :packet_retention_days, 365)
|
||||
job_args = %{}
|
||||
test "drops old partitions and cleans up bad packets" do
|
||||
{:ok, _} = PartitionManager.ensure_partitions_exist()
|
||||
|
||||
# Create an old bad packet
|
||||
retention_days = Application.get_env(:aprsme, :packet_retention_days, 7)
|
||||
|
||||
# Create old packets that should be deleted (older than retention period)
|
||||
old_time =
|
||||
DateTime.utc_now()
|
||||
|> DateTime.add(-(retention_days + 10) * 86_400, :second)
|
||||
|> DateTime.truncate(:second)
|
||||
|> DateTime.truncate(:microsecond)
|
||||
|
||||
insert_packets(10, %{received_at: old_time})
|
||||
Repo.insert!(%BadPacket{raw_packet: "bad", error_message: "test", attempted_at: old_time})
|
||||
|
||||
# Create recent packets that should NOT be deleted (1 hour ago)
|
||||
recent_time = DateTime.utc_now() |> DateTime.add(-3600, :second) |> DateTime.truncate(:second)
|
||||
insert_packets(3, %{received_at: recent_time})
|
||||
# Create a recent bad packet
|
||||
recent_time =
|
||||
DateTime.utc_now()
|
||||
|> DateTime.add(-3600, :second)
|
||||
|> DateTime.truncate(:microsecond)
|
||||
|
||||
assert :ok = PacketCleanupWorker.perform(job_args)
|
||||
Repo.insert!(%BadPacket{raw_packet: "recent_bad", error_message: "test", attempted_at: recent_time})
|
||||
|
||||
# Verify only recent packets remain
|
||||
assert Repo.aggregate(Packet, :count, :id) == 3
|
||||
assert :ok = PacketCleanupWorker.perform(%{})
|
||||
|
||||
# Only recent bad packet should remain
|
||||
assert Repo.aggregate(BadPacket, :count) == 1
|
||||
end
|
||||
end
|
||||
|
||||
describe "cleanup_packets_older_than/1" do
|
||||
test "cleans up packets older than the given number of days" do
|
||||
days = 60
|
||||
describe "perform/1 with negative cleanup_days" do
|
||||
test "falls through to default perform and succeeds" do
|
||||
{:ok, _} = PartitionManager.ensure_partitions_exist()
|
||||
|
||||
# Create old packets that should be deleted
|
||||
old_time = DateTime.utc_now() |> DateTime.add(-(days + 10) * 86_400, :second) |> DateTime.truncate(:second)
|
||||
insert_packets(15, %{received_at: old_time})
|
||||
|
||||
# Create recent packets that should NOT be deleted
|
||||
recent_time = DateTime.utc_now() |> DateTime.add(-30 * 86_400, :second) |> DateTime.truncate(:second)
|
||||
insert_packets(5, %{received_at: recent_time})
|
||||
|
||||
deleted_count = PacketCleanupWorker.cleanup_packets_older_than(days)
|
||||
|
||||
assert deleted_count == 15
|
||||
assert Repo.aggregate(Packet, :count, :id) == 5
|
||||
end
|
||||
end
|
||||
|
||||
describe "cleanup_packets_older_than_batched/1" do
|
||||
test "returns correct tuple format with deleted count and batches" do
|
||||
days = 30
|
||||
|
||||
# Create old packets that should be deleted
|
||||
old_time = DateTime.utc_now() |> DateTime.add(-(days + 1) * 86_400, :second) |> DateTime.truncate(:second)
|
||||
insert_packets(5, %{received_at: old_time})
|
||||
|
||||
# Create recent packets that should NOT be deleted
|
||||
recent_time = DateTime.utc_now() |> DateTime.add(-5 * 86_400, :second) |> DateTime.truncate(:second)
|
||||
insert_packets(3, %{received_at: recent_time})
|
||||
|
||||
{deleted_count, batches} = PacketCleanupWorker.cleanup_packets_older_than_batched(days)
|
||||
|
||||
assert is_integer(deleted_count)
|
||||
assert is_integer(batches)
|
||||
assert deleted_count == 5
|
||||
assert batches >= 1
|
||||
assert Repo.aggregate(Packet, :count, :id) == 3
|
||||
end
|
||||
end
|
||||
|
||||
describe "get_packet_ids_for_deletion/2" do
|
||||
test "returns list of packet IDs within batch size limit" do
|
||||
cutoff_time = DateTime.utc_now() |> DateTime.add(-30, :day) |> DateTime.truncate(:second)
|
||||
batch_size = 10
|
||||
|
||||
# Create old packets that should be included
|
||||
old_time = DateTime.utc_now() |> DateTime.add(-40 * 86_400, :second) |> DateTime.truncate(:second)
|
||||
old_packets = insert_packets(15, %{received_at: old_time})
|
||||
|
||||
# Create recent packets that should NOT be included
|
||||
recent_time = DateTime.utc_now() |> DateTime.add(-5 * 86_400, :second) |> DateTime.truncate(:second)
|
||||
insert_packets(5, %{received_at: recent_time})
|
||||
|
||||
result = PacketCleanupWorker.get_packet_ids_for_deletion(cutoff_time, batch_size)
|
||||
|
||||
assert is_list(result)
|
||||
assert length(result) == batch_size
|
||||
|
||||
# Verify that returned IDs are from old packets
|
||||
old_packet_ids = Enum.map(old_packets, & &1.id)
|
||||
assert Enum.all?(result, &(&1 in old_packet_ids))
|
||||
# Negative days don't match the guard, so falls through to the catch-all
|
||||
assert :ok = PacketCleanupWorker.perform(%{"cleanup_days" => -1})
|
||||
end
|
||||
end
|
||||
end
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue