Graham McIntire
ca05b497b9
Drop 10 dead parser-compat columns (srccallsign, dstcallsign, origpacket, body, header, alive, posambiguity, symboltable, symbolcode, messaging) and move 14 display-only columns into a single `data` JSONB column (PHG, telemetry, radiorange, information_field, format, posresolution, position_ambiguity, luminosity, rain_midnight). Reduces row width from ~70 to ~48 columns. Table is ephemeral so migration uses DROP/recreate. Also fixes snow_24h -> snow bug in weather check query and adds null byte stripping for JSONB string sanitization.
263 lines
8.2 KiB
Elixir
263 lines
8.2 KiB
Elixir
defmodule Aprsme.Packets.QueryBuilder do
|
|
@moduledoc """
|
|
Query builder functions for composing Packet queries.
|
|
Provides reusable, composable functions for common query patterns.
|
|
"""
|
|
|
|
import Ecto.Query
|
|
|
|
alias Aprsme.Packet
|
|
|
|
@doc """
|
|
Filters query by time range. Supports start_time, end_time, and hours_back options.
|
|
|
|
## Options
|
|
* `:start_time` - DateTime to filter packets after
|
|
* `:end_time` - DateTime to filter packets before
|
|
* `:hours_back` - Alternative to start_time, filters packets from X hours ago
|
|
"""
|
|
@spec with_time_range(Ecto.Query.t(), map() | keyword()) :: Ecto.Query.t()
|
|
def with_time_range(query, opts) when is_list(opts) do
|
|
with_time_range(query, Map.new(opts))
|
|
end
|
|
|
|
def with_time_range(query, opts) when is_map(opts) do
|
|
query
|
|
|> maybe_filter_start_time(opts[:start_time] || opts["start_time"])
|
|
|> maybe_filter_end_time(opts[:end_time] || opts["end_time"])
|
|
|> maybe_filter_hours_back(opts[:hours_back] || opts["hours_back"])
|
|
end
|
|
|
|
@doc """
|
|
Filters query to only include packets with position data.
|
|
"""
|
|
@spec with_position(Ecto.Query.t()) :: Ecto.Query.t()
|
|
def with_position(query) do
|
|
from p in query, where: p.has_position == true
|
|
end
|
|
|
|
@doc """
|
|
Filters query by exact callsign match (case-insensitive).
|
|
Uses the functional index on upper(sender) for performance.
|
|
"""
|
|
@spec for_callsign(Ecto.Query.t(), String.t()) :: Ecto.Query.t()
|
|
def for_callsign(query, callsign) when is_binary(callsign) do
|
|
normalized = String.upcase(String.trim(callsign))
|
|
from p in query, where: fragment("upper(?)", p.sender) == ^normalized
|
|
end
|
|
|
|
@doc """
|
|
Filters query by base callsign (without SSID).
|
|
"""
|
|
@spec for_base_callsign(Ecto.Query.t(), String.t()) :: Ecto.Query.t()
|
|
def for_base_callsign(query, base_callsign) when is_binary(base_callsign) do
|
|
from p in query, where: p.base_callsign == ^base_callsign
|
|
end
|
|
|
|
@doc """
|
|
Orders query by received_at descending (most recent first).
|
|
"""
|
|
@spec recent_first(Ecto.Query.t()) :: Ecto.Query.t()
|
|
def recent_first(query) do
|
|
from p in query, order_by: [desc: p.received_at]
|
|
end
|
|
|
|
@doc """
|
|
Orders query by received_at ascending (oldest first).
|
|
"""
|
|
@spec chronological(Ecto.Query.t()) :: Ecto.Query.t()
|
|
def chronological(query) do
|
|
from p in query, order_by: [asc: p.received_at]
|
|
end
|
|
|
|
@doc """
|
|
Filters query to only weather packets.
|
|
Uses the indexed `has_weather` boolean column.
|
|
"""
|
|
@spec weather_only(Ecto.Query.t()) :: Ecto.Query.t()
|
|
def weather_only(query) do
|
|
from p in query, where: p.has_weather == true
|
|
end
|
|
|
|
@doc """
|
|
Applies pagination with limit and optional offset.
|
|
"""
|
|
@spec paginate(Ecto.Query.t(), integer(), integer()) :: Ecto.Query.t()
|
|
def paginate(query, limit, offset \\ 0) when is_integer(limit) and is_integer(offset) do
|
|
from p in query, limit: ^limit, offset: ^offset
|
|
end
|
|
|
|
@doc """
|
|
Adds PostGIS coordinate extraction to select clause.
|
|
"""
|
|
@spec with_coordinates(Ecto.Query.t()) :: Ecto.Query.t()
|
|
def with_coordinates(query) do
|
|
from p in query,
|
|
select: %{p | lat: fragment("ST_Y(?)", p.location), lon: fragment("ST_X(?)", p.location)}
|
|
end
|
|
|
|
@doc """
|
|
Selects only the fields needed for map display as a plain map.
|
|
Returns ~22 columns instead of all 73, reducing I/O significantly.
|
|
"""
|
|
@spec select_map_fields(Ecto.Query.t()) :: Ecto.Query.t()
|
|
def select_map_fields(query) do
|
|
from p in query,
|
|
select: %{
|
|
id: p.id,
|
|
sender: p.sender,
|
|
object_name: p.object_name,
|
|
item_name: p.item_name,
|
|
lat: fragment("?::float8", p.lat),
|
|
lon: fragment("?::float8", p.lon),
|
|
received_at: p.received_at,
|
|
symbol_table_id: p.symbol_table_id,
|
|
symbol_code: p.symbol_code,
|
|
comment: p.comment,
|
|
path: p.path,
|
|
temperature: p.temperature,
|
|
humidity: p.humidity,
|
|
wind_direction: p.wind_direction,
|
|
wind_speed: p.wind_speed,
|
|
wind_gust: p.wind_gust,
|
|
pressure: p.pressure,
|
|
rain_1h: p.rain_1h,
|
|
rain_24h: p.rain_24h,
|
|
rain_since_midnight: p.rain_since_midnight,
|
|
snow: p.snow,
|
|
luminosity: fragment("(?->>'luminosity')::integer", p.data)
|
|
}
|
|
end
|
|
|
|
@doc """
|
|
Filters by region if provided in options.
|
|
"""
|
|
@spec maybe_filter_region(Ecto.Query.t(), map() | keyword()) :: Ecto.Query.t()
|
|
def maybe_filter_region(query, opts) when is_list(opts) do
|
|
maybe_filter_region(query, Map.new(opts))
|
|
end
|
|
|
|
def maybe_filter_region(query, %{region: region}) when not is_nil(region) do
|
|
from p in query, where: p.region == ^region
|
|
end
|
|
|
|
def maybe_filter_region(query, _), do: query
|
|
|
|
@doc """
|
|
Filters by map bounds using bounding box coordinates.
|
|
|
|
## Example
|
|
|
|
within_bounds(query, [-74.0, 40.0, -73.0, 41.0])
|
|
# [west, south, east, north]
|
|
"""
|
|
@spec within_bounds(Ecto.Query.t(), list(number())) :: Ecto.Query.t()
|
|
def within_bounds(query, [west, south, east, north])
|
|
when is_number(west) and is_number(south) and is_number(east) and is_number(north) do
|
|
# Use ST_MakeEnvelope with && operator to leverage GiST spatial index on location column
|
|
if west > east do
|
|
# Handle antimeridian crossing with two envelopes: [west, 180] and [-180, east]
|
|
from p in query,
|
|
where: p.has_position == true,
|
|
where:
|
|
fragment("? && ST_MakeEnvelope(?, ?, 180, ?, 4326)", p.location, ^west, ^south, ^north) or
|
|
fragment("? && ST_MakeEnvelope(-180, ?, ?, ?, 4326)", p.location, ^south, ^east, ^north)
|
|
else
|
|
from p in query,
|
|
where: p.has_position == true,
|
|
where: fragment("? && ST_MakeEnvelope(?, ?, ?, ?, 4326)", p.location, ^west, ^south, ^east, ^north)
|
|
end
|
|
end
|
|
|
|
def within_bounds(query, _), do: query
|
|
|
|
@doc """
|
|
Common query composition for recent position packets.
|
|
|
|
## Options
|
|
* `:limit` - Number of packets to return (default: 100)
|
|
* `:start_time` - Filter packets after this time
|
|
* `:end_time` - Filter packets before this time
|
|
* `:region` - Filter by region
|
|
"""
|
|
@spec recent_position_packets(map() | keyword()) :: Ecto.Query.t()
|
|
def recent_position_packets(opts \\ %{}) do
|
|
limit = opts[:limit] || opts["limit"] || 100
|
|
|
|
from(p in Packet)
|
|
|> with_position()
|
|
|> with_time_range(opts)
|
|
|> maybe_filter_region(opts)
|
|
|> recent_first()
|
|
|> paginate(limit)
|
|
|> with_coordinates()
|
|
end
|
|
|
|
@doc """
|
|
Common query composition for callsign packet history.
|
|
|
|
## Options
|
|
* `:limit` - Number of packets to return (default: 100)
|
|
* `:start_time` - Filter packets after this time
|
|
* `:end_time` - Filter packets before this time
|
|
"""
|
|
@spec callsign_history(String.t(), map() | keyword()) :: Ecto.Query.t()
|
|
def callsign_history(callsign, opts \\ %{}) do
|
|
limit = opts[:limit] || opts["limit"] || 100
|
|
|
|
from(p in Packet)
|
|
|> for_callsign(callsign)
|
|
|> with_time_range(opts)
|
|
|> recent_first()
|
|
|> paginate(limit)
|
|
end
|
|
|
|
@doc """
|
|
Common query composition for weather packets.
|
|
|
|
## Options
|
|
* `:callsign` - Filter by specific callsign
|
|
* `:limit` - Number of packets to return (default: 100)
|
|
* `:start_time` - Filter packets after this time
|
|
* `:end_time` - Filter packets before this time
|
|
"""
|
|
@spec weather_packets(map() | keyword()) :: Ecto.Query.t()
|
|
def weather_packets(opts \\ %{}) do
|
|
limit = opts[:limit] || opts["limit"] || 100
|
|
callsign = opts[:callsign] || opts["callsign"]
|
|
|
|
query =
|
|
from(p in Packet)
|
|
|> weather_only()
|
|
|> with_time_range(opts)
|
|
|> recent_first()
|
|
|> paginate(limit)
|
|
|
|
if callsign do
|
|
for_callsign(query, callsign)
|
|
else
|
|
query
|
|
end
|
|
end
|
|
|
|
# Private helper functions
|
|
|
|
defp maybe_filter_start_time(query, nil), do: query
|
|
|
|
defp maybe_filter_start_time(query, start_time) do
|
|
from p in query, where: p.received_at >= ^start_time
|
|
end
|
|
|
|
defp maybe_filter_end_time(query, nil), do: query
|
|
|
|
defp maybe_filter_end_time(query, end_time) do
|
|
from p in query, where: p.received_at <= ^end_time
|
|
end
|
|
|
|
defp maybe_filter_hours_back(query, nil), do: query
|
|
|
|
defp maybe_filter_hours_back(query, hours) when is_number(hours) do
|
|
start_time = DateTime.add(DateTime.utc_now(), -hours * 3600, :second)
|
|
from p in query, where: p.received_at >= ^start_time
|
|
end
|
|
end
|