graham/aprs.me
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

format

Browse source
This commit is contained in:
Graham McIntire 2025-07-27 11:16:16 -05:00
parent 9270bc1aa5
commit 1596e95eba
No known key found for this signature in database
12 changed files with 182 additions and 166 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

31
config/config.exs
View file

@ -31,24 +31,11 @@ config :aprsme, AprsmeWeb.Gettext,
locales: ~w(en es de fr),
default_locale: "en"
# Configure Exq for background jobs
# Redis connection settings are configured in runtime.exs
config :exq,
name: Exq,
concurrency: :infinite,
queues: [
{"default", 10},
{"maintenance", 2}
],
scheduler_enable: true,
scheduler_poll_timeout: 200,
poll_timeout: 50,
redis_timeout: 5000
# Configure periodic cleanup job
config :aprsme, :cleanup_scheduler,
enabled: true,
interval: 6 * 60 * 60 * 1000 # 6 hours in milliseconds
# 6 hours in milliseconds
interval: 6 * 60 * 60 * 1000
config :aprsme,
ecto_repos: [Aprsme.Repo]
@ -128,6 +115,20 @@ config :esbuild,
cd: Path.expand("../assets", __DIR__)
]
# Configure Exq for background jobs
# Redis connection settings are configured in runtime.exs
config :exq,
name: Exq,
concurrency: :infinite,
queues: [
{"default", 10},
{"maintenance", 2}
],
scheduler_enable: true,
scheduler_poll_timeout: 200,
poll_timeout: 50,
redis_timeout: 5000
config :gettext, :plural_forms, GettextPseudolocalize.Plural
# Configures Elixir's Logger

93
config/runtime.exs
View file

@ -49,6 +49,55 @@ if config_env() == :prod do
# Configure clustering based on environment
cluster_enabled = System.get_env("CLUSTER_ENABLED", "false") == "true"
# Configure Exq with Redis connection from environment
redis_url = System.get_env("REDIS_URL", "redis://localhost:6379")
redis_uri = URI.parse(redis_url)
redis_host = redis_uri.host || "localhost"
redis_port = redis_uri.port || 6379
redis_password =
case redis_uri.userinfo do
nil ->
""
userinfo ->
case String.split(userinfo, ":") do
[_, password] -> password
_ -> ""
end
end
# Extract database from path if present
redis_database =
try do
case redis_uri.path do
nil ->
0
"/" ->
0
path ->
path |> String.trim_leading("/") |> String.to_integer()
end
rescue
_ -> 0
end
exq_config = [
host: redis_host,
port: redis_port,
database: redis_database
]
# Only add password if it's not empty
exq_config =
if redis_password != "" and redis_password != nil do
Keyword.put(exq_config, :password, redis_password)
else
exq_config
end
# ## Configuring the mailer
#
# Configure Resend for email delivery
@ -119,48 +168,6 @@ if config_env() == :prod do
config :aprsme,
default_from_email: "w5isp@aprs.me"
# Configure Exq with Redis connection from environment
redis_url = System.get_env("REDIS_URL", "redis://localhost:6379")
redis_uri = URI.parse(redis_url)
redis_host = redis_uri.host || "localhost"
redis_port = redis_uri.port || 6379
redis_password = case redis_uri.userinfo do
nil -> ""
userinfo ->
case String.split(userinfo, ":") do
[_, password] -> password
_ -> ""
end
end
# Extract database from path if present
redis_database = try do
case redis_uri.path do
nil -> 0
"/" -> 0
path ->
path |> String.trim_leading("/") |> String.to_integer()
end
rescue
_ -> 0
end
exq_config = [
host: redis_host,
port: redis_port,
database: redis_database
]
# Only add password if it's not empty
exq_config = if redis_password != "" and redis_password != nil do
Keyword.put(exq_config, :password, redis_password)
else
exq_config
end
config :exq, exq_config
config :aprsme,
ecto_repos: [Aprsme.Repo],
aprs_is_server: System.get_env("APRS_SERVER", "dallas.aprs2.net"),
@ -171,6 +178,8 @@ if config_env() == :prod do
aprs_is_password: System.get_env("APRS_PASSCODE"),
env: :prod
config :exq, exq_config
# Configure Hammer for production environment
config :hammer,
backend: {Hammer.Backend.ETS, [expiry_ms: 60_000 * 60 * 4, cleanup_interval_ms: 60_000 * 10]}

6
config/test.exs
View file

@ -27,9 +27,6 @@ config :aprsme, AprsmeWeb.Endpoint,
secret_key_base: "IV9+ENaw9i8xjReRk4sULRvRgsmFVTGQwQGGrf4G+Q/SFMeHBCNWRlPXQ2YvT36R",
server: false
# Disable Exq during tests to prevent background job execution
config :exq, mode: :inline
# Disable cleanup scheduler in test environment
config :aprsme, :cleanup_scheduler, enabled: false
@ -54,6 +51,9 @@ config :aprsme, auto_migrate: false
# Only in tests, remove the complexity from the password hashing algorithm
config :bcrypt_elixir, :log_rounds, 1
# Disable Exq during tests to prevent background job execution
config :exq, mode: :inline
# Configure ExVCR
config :exvcr,
vcr_cassette_library_dir: "test/fixtures/vcr_cassettes",

4
lib/aprsme/application.ex
View file

@ -92,13 +92,14 @@ defmodule Aprsme.Application do
defp migrate do
auto_migrate = Application.get_env(:aprsme, :auto_migrate, true)
cluster_enabled = Application.get_env(:aprsme, :cluster_enabled, false)
# In cluster mode, prefer init containers or manual migration
# to avoid race conditions between nodes
if auto_migrate and not cluster_enabled do
do_migrate(true)
else
require Logger
if cluster_enabled do
Logger.info("Skipping auto-migration in cluster mode")
else
@ -176,7 +177,6 @@ defmodule Aprsme.Application do
Logger.info("Database migrations completed")
end
defp pubsub_config do
cluster_enabled = Application.get_env(:aprsme, :cluster_enabled, false)
redis_url = System.get_env("REDIS_URL")

27
lib/aprsme/cleanup_scheduler.ex
View file

@ -1,24 +1,26 @@
defmodule Aprsme.CleanupScheduler do
@moduledoc """
GenServer that schedules periodic packet cleanup jobs using Exq.
This scheduler replaces the Oban cron functionality for packet cleanup.
It runs every 6 hours by default and enqueues a cleanup job.
"""
use GenServer
require Logger
def start_link(opts) do
GenServer.start_link(__MODULE__, opts, name: __MODULE__)
end
@impl true
def init(_opts) do
config = Application.get_env(:aprsme, :cleanup_scheduler, [])
enabled = Keyword.get(config, :enabled, true)
interval = Keyword.get(config, :interval, 6 * 60 * 60 * 1000) # 6 hours
# 6 hours
interval = Keyword.get(config, :interval, 6 * 60 * 60 * 1000)
if enabled do
Logger.info("Starting packet cleanup scheduler with #{interval}ms interval")
schedule_next_cleanup(interval)
@ -28,30 +30,31 @@ defmodule Aprsme.CleanupScheduler do
{:ok, %{interval: nil}}
end
end
@impl true
def handle_info(:schedule_cleanup, %{interval: interval} = state) when is_integer(interval) do
Logger.info("Scheduling packet cleanup job")
# Enqueue cleanup job with Exq
case Exq.enqueue(Exq, "maintenance", Aprsme.Workers.PacketCleanupWorker, []) do
{:ok, _job_id} ->
Logger.info("Packet cleanup job scheduled successfully")
{:error, reason} ->
Logger.error("Failed to schedule packet cleanup job: #{inspect(reason)}")
end
schedule_next_cleanup(interval)
{:noreply, state}
end
@impl true
def handle_info(:schedule_cleanup, state) do
# Scheduler is disabled
{:noreply, state}
end
defp schedule_next_cleanup(interval) do
Process.send_after(self(), :schedule_cleanup, interval)
end
end
end

96
lib/aprsme/encoding.ex
View file

@ -24,13 +24,16 @@ defmodule Aprsme.Encoding do
"""
@spec to_float_safe(binary()) :: {:ok, float()} | nil
def to_float_safe(value) when is_binary(value) do
sanitized = value
|> String.trim()
|> String.slice(0, 30) # Reasonable max length for a number
sanitized =
value
|> String.trim()
# Reasonable max length for a number
|> String.slice(0, 30)
case Float.parse(sanitized) do
{f, _} when f > -9.0e15 and f < 9.0e15 ->
{:ok, f}
_ ->
nil
end
@ -56,48 +59,48 @@ defmodule Aprsme.Encoding do
@spec has_weather_data(any(), any(), any(), any()) :: boolean()
def has_weather_data(temperature, humidity, wind_speed, pressure) do
not is_nil(temperature) or
not is_nil(humidity) or
not is_nil(wind_speed) or
not is_nil(pressure)
not is_nil(humidity) or
not is_nil(wind_speed) or
not is_nil(pressure)
end
@doc """
Get encoding information about a binary
"""
@spec encoding_info(binary()) :: %{
valid_utf8: boolean(),
byte_count: non_neg_integer(),
char_count: non_neg_integer() | nil,
invalid_at: non_neg_integer() | nil
}
valid_utf8: boolean(),
byte_count: non_neg_integer(),
char_count: non_neg_integer() | nil,
invalid_at: non_neg_integer() | nil
}
def encoding_info(input) when is_binary(input) do
byte_count = byte_size(input)
case String.valid?(input) do
true ->
%{
valid_utf8: true,
byte_count: byte_count,
char_count: String.length(input),
invalid_at: nil
}
false ->
invalid_pos = find_invalid_byte_position(input)
%{
valid_utf8: false,
byte_count: byte_count,
char_count: nil,
invalid_at: invalid_pos
}
if String.valid?(input) do
%{
valid_utf8: true,
byte_count: byte_count,
char_count: String.length(input),
invalid_at: nil
}
else
invalid_pos = find_invalid_byte_position(input)
%{
valid_utf8: false,
byte_count: byte_count,
char_count: nil,
invalid_at: invalid_pos
}
end
end
@spec encoding_info(any()) :: %{
valid_utf8: boolean(),
byte_count: non_neg_integer(),
char_count: nil,
invalid_at: nil
}
valid_utf8: boolean(),
byte_count: non_neg_integer(),
char_count: nil,
invalid_at: nil
}
def encoding_info(_) do
%{
valid_utf8: false,
@ -121,14 +124,12 @@ defmodule Aprsme.Encoding do
@spec latin1_to_utf8(binary()) :: binary()
defp latin1_to_utf8(input) do
try do
input
|> :binary.bin_to_list()
|> Enum.map(&latin1_char_to_utf8/1)
|> IO.iodata_to_binary()
rescue
_ -> ""
end
input
|> :binary.bin_to_list()
|> Enum.map(&latin1_char_to_utf8/1)
|> IO.iodata_to_binary()
rescue
_ -> ""
end
@spec latin1_char_to_utf8(0..255) :: binary()
@ -166,6 +167,7 @@ defmodule Aprsme.Encoding do
c when c >= 128 and c <= 159 -> false
_ -> true
end
_ ->
# Multi-codepoint grapheme, keep it
true
@ -186,11 +188,9 @@ defmodule Aprsme.Encoding do
@spec valid_utf8_continuation?(binary(), non_neg_integer()) :: boolean()
defp valid_utf8_continuation?(binary, pos) do
try do
<<_::binary-size(pos), _char::utf8, _::binary>> = binary
true
rescue
_ -> false
end
<<_::binary-size(pos), _char::utf8, _::binary>> = binary
true
rescue
_ -> false
end
end
end

6
lib/aprsme/encoding_utils.ex
View file

@ -49,14 +49,12 @@ defmodule Aprsme.EncodingUtils do
"""
@spec to_float(any()) :: float() | nil
def to_float(value) when is_float(value) do
if finite_float?(value), do: value, else: nil
if finite_float?(value), do: value
end
def to_float(value) when is_integer(value) do
if value >= -9.0e15 and value <= 9.0e15 do
value * 1.0
else
nil
end
end
@ -72,7 +70,7 @@ defmodule Aprsme.EncodingUtils do
def to_float(%Decimal{} = value) do
float = Decimal.to_float(value)
if finite_float?(float), do: float, else: nil
if finite_float?(float), do: float
end
def to_float(_), do: nil

8
lib/aprsme/migration_lock.ex
View file

@ -37,7 +37,7 @@ defmodule Aprsme.MigrationLock do
defp acquire_lock(repo) do
# Try to acquire an exclusive advisory lock (non-blocking)
query = "SELECT pg_try_advisory_lock($1)"
case repo.query(query, [@migration_lock_id]) do
{:ok, %{rows: [[true]]}} ->
:ok
@ -53,7 +53,7 @@ defmodule Aprsme.MigrationLock do
defp release_lock(repo) do
query = "SELECT pg_advisory_unlock($1)"
case repo.query(query, [@migration_lock_id]) do
{:ok, %{rows: [[true]]}} ->
:ok
@ -77,7 +77,7 @@ defmodule Aprsme.MigrationLock do
defp wait_for_migrations(repo, retries) do
# Check if lock is still held
query = "SELECT pg_try_advisory_lock($1)"
case repo.query(query, [@migration_lock_id]) do
{:ok, %{rows: [[true]]}} ->
# We got the lock, which means migrations are done
@ -95,4 +95,4 @@ defmodule Aprsme.MigrationLock do
:error
end
end
end
end

46
lib/aprsme/release.ex
View file

@ -21,19 +21,21 @@ defmodule Aprsme.Release do
# Run migrations with distributed lock
cluster_enabled = Application.get_env(:aprsme, :cluster_enabled, false)
# Configure longer timeouts for migration operations
migration_timeout = String.to_integer(System.get_env("MIGRATION_TIMEOUT", "7200000")) # 2 hours default
# 2 hours default
migration_timeout = String.to_integer(System.get_env("MIGRATION_TIMEOUT", "7200000"))
if cluster_enabled do
Logger.info("Running migrations with distributed lock...")
# Ensure repo is started for advisory lock with extended timeout
repo_config = Aprsme.Repo.config()
|> Keyword.put(:timeout, migration_timeout)
|> Keyword.put(:pool_timeout, 60_000)
repo_config =
Aprsme.Repo.config()
|> Keyword.put(:timeout, migration_timeout)
|> Keyword.put(:pool_timeout, 60_000)
{:ok, _} = Aprsme.Repo.start_link(repo_config)
Aprsme.MigrationLock.with_lock(Aprsme.Repo, fn ->
run_migrations_with_timeout(migration_timeout)
end)
@ -91,21 +93,25 @@ defmodule Aprsme.Release do
defp run_migrations_with_timeout(timeout) do
require Logger
Logger.info("Running migrations with timeout: #{timeout}ms")
# Run with extended timeout configuration
_repo_config = Aprsme.Repo.config()
|> Keyword.put(:timeout, timeout)
|> Keyword.put(:pool_timeout, 60_000)
{:ok, _, _} = Ecto.Migrator.with_repo(Aprsme.Repo, fn repo ->
# Set session-level timeout for this connection
Ecto.Adapters.SQL.query!(repo, "SET statement_timeout = '#{div(timeout, 1000)}s'")
Ecto.Migrator.run(repo, :up, all: true)
end, [timeout: timeout])
_repo_config =
Aprsme.Repo.config()
|> Keyword.put(:timeout, timeout)
|> Keyword.put(:pool_timeout, 60_000)
{:ok, _, _} =
Ecto.Migrator.with_repo(
Aprsme.Repo,
fn repo ->
# Set session-level timeout for this connection
Ecto.Adapters.SQL.query!(repo, "SET statement_timeout = '#{div(timeout, 1000)}s'")
Ecto.Migrator.run(repo, :up, all: true)
end, timeout: timeout)
end
defp read_deployment_timestamp do
case File.read("/app/deployed_at.txt") do
{:ok, timestamp} ->

3
mix.exs
View file

@ -52,7 +52,6 @@ defmodule Aprsme.MixProject do
defp elixirc_paths(:test), do: ["lib", "test/support"]
defp elixirc_paths(_), do: ["lib"]
# Specifies your project dependencies.
#
# Type `mix help deps` for examples and options.
@ -107,7 +106,7 @@ defmodule Aprsme.MixProject do
{:hammer, "~> 7.0"},
{:cachex, "~> 4.1"},
{:gettext_pseudolocalize, "~> 0.1"},
{:sentry, "~> 11.0"},
{:sentry, "~> 11.0"}
# Gleam dependencies
]
end

2
priv/repo/migrations/20250726215048_add_optimized_indexes.exs
View file

@ -103,7 +103,7 @@ defmodule Aprsme.Repo.Migrations.AddOptimizedIndexes do
# Analyze the table to update statistics after adding indexes
execute "ANALYZE packets;"
# Reset statement timeout
execute "RESET statement_timeout;"
execute "RESET lock_timeout;"

26
priv/repo/migrations/20250727000000_add_has_weather_column.exs
View file

@ -6,14 +6,14 @@ defmodule Aprsme.Repo.Migrations.AddHasWeatherColumn do
def up do
# This migration adds a generated column which requires a full table rewrite
# It should be run during a maintenance window or when load is low
# Set aggressive timeout for very large table
execute "SET statement_timeout = '2h';"
execute "SET lock_timeout = '30s';"
# Log progress
execute "SELECT NOW() AS start_time, 'Starting has_weather column addition' AS status;"
# Add the column - this will rewrite the entire table
# Consider running this during off-peak hours
execute """
@ -28,24 +28,24 @@ defmodule Aprsme.Repo.Migrations.AddHasWeatherColumn do
OR rain_1h IS NOT NULL
) STORED
"""
# Create index on the new column
execute """
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_packets_has_weather
ON packets (has_weather, sender, received_at DESC)
WHERE has_weather = true
"""
# Drop the functional index if it exists (replaced by column-based index)
execute "DROP INDEX IF EXISTS idx_packets_has_weather_functional;"
# Update statistics
execute "ANALYZE packets (has_weather);"
# Reset timeouts
execute "RESET statement_timeout;"
execute "RESET lock_timeout;"
# Log completion
execute "SELECT NOW() AS end_time, 'Completed has_weather column addition' AS status;"
end
@ -53,13 +53,13 @@ defmodule Aprsme.Repo.Migrations.AddHasWeatherColumn do
def down do
execute "SET statement_timeout = '2h';"
execute "SET lock_timeout = '30s';"
# Drop the index first
execute "DROP INDEX IF EXISTS idx_packets_has_weather;"
# Drop the column (this will also rewrite the table)
execute "ALTER TABLE packets DROP COLUMN IF EXISTS has_weather;"
# Recreate the functional index
execute """
CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_packets_has_weather_functional
@ -71,8 +71,8 @@ defmodule Aprsme.Repo.Migrations.AddHasWeatherColumn do
OR wind_direction IS NOT NULL
OR rain_1h IS NOT NULL)
"""
execute "RESET statement_timeout;"
execute "RESET lock_timeout;"
end
end
end