diff --git a/config/config.exs b/config/config.exs index 020c038..f3c8c06 100644 --- a/config/config.exs +++ b/config/config.exs @@ -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", diff --git a/lib/aprsme/application.ex b/lib/aprsme/application.ex index f73d032..c64cceb 100644 --- a/lib/aprsme/application.ex +++ b/lib/aprsme/application.ex @@ -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 diff --git a/lib/aprsme/maidenhead.ex b/lib/aprsme/maidenhead.ex new file mode 100644 index 0000000..1eeeaa9 --- /dev/null +++ b/lib/aprsme/maidenhead.ex @@ -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) + + <> + end + + def encode(_lon, _lat), do: nil +end diff --git a/lib/aprsme/partition_manager.ex b/lib/aprsme/partition_manager.ex new file mode 100644 index 0000000..64b2762 --- /dev/null +++ b/lib/aprsme/partition_manager.ex @@ -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 diff --git a/lib/aprsme/workers/packet_cleanup_worker.ex b/lib/aprsme/workers/packet_cleanup_worker.ex index 4d3a3ad..5e4b790 100644 --- a/lib/aprsme/workers/packet_cleanup_worker.ex +++ b/lib/aprsme/workers/packet_cleanup_worker.ex @@ -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 diff --git a/lib/aprsme_web/live/info_live/show.html.heex b/lib/aprsme_web/live/info_live/show.html.heex index ce2962c..d0966e3 100644 --- a/lib/aprsme_web/live/info_live/show.html.heex +++ b/lib/aprsme_web/live/info_live/show.html.heex @@ -89,12 +89,7 @@
{gettext("Grid Square")}
- <%= case Gridsquare.encode(@packet.lon, @packet.lat) do %> - <% %Gridsquare.EncodeResult{grid_reference: grid_ref} -> %> - {grid_ref} - <% _ -> %> - - - <% end %> + {Aprsme.Maidenhead.encode(@packet.lon, @packet.lat) || "-"}
<% end %> diff --git a/mix.exs b/mix.exs index cb18436..1883404 100644 --- a/mix.exs +++ b/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 diff --git a/priv/gettext/xx/LC_MESSAGES/default.po b/priv/gettext/xx/LC_MESSAGES/default.po index 346c0af..9e43bff 100644 --- a/priv/gettext/xx/LC_MESSAGES/default.po +++ b/priv/gettext/xx/LC_MESSAGES/default.po @@ -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 diff --git a/priv/gettext/xx/LC_MESSAGES/errors.po b/priv/gettext/xx/LC_MESSAGES/errors.po index 5c132c8..615acaf 100644 --- a/priv/gettext/xx/LC_MESSAGES/errors.po +++ b/priv/gettext/xx/LC_MESSAGES/errors.po @@ -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 "⟦ċàñ'ť ƀê ƀĺàñǩ~~~~~~⟧" diff --git a/priv/repo/migrations/20260220200000_partition_packets_table.exs b/priv/repo/migrations/20260220200000_partition_packets_table.exs new file mode 100644 index 0000000..9fb900f --- /dev/null +++ b/priv/repo/migrations/20260220200000_partition_packets_table.exs @@ -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 diff --git a/test/aprsme/maidenhead_test.exs b/test/aprsme/maidenhead_test.exs new file mode 100644 index 0000000..60d0381 --- /dev/null +++ b/test/aprsme/maidenhead_test.exs @@ -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 diff --git a/test/aprsme/packet_consumer_test.exs b/test/aprsme/packet_consumer_test.exs index ed94d7e..b6aced7 100644 --- a/test/aprsme/packet_consumer_test.exs +++ b/test/aprsme/packet_consumer_test.exs @@ -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"])) diff --git a/test/aprsme/partition_manager_test.exs b/test/aprsme/partition_manager_test.exs new file mode 100644 index 0000000..d57bdf3 --- /dev/null +++ b/test/aprsme/partition_manager_test.exs @@ -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 diff --git a/test/aprsme/workers/packet_cleanup_worker_test.exs b/test/aprsme/workers/packet_cleanup_worker_test.exs index 01d7caa..fd20b6f 100644 --- a/test/aprsme/workers/packet_cleanup_worker_test.exs +++ b/test/aprsme/workers/packet_cleanup_worker_test.exs @@ -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