diff --git a/config/config.exs b/config/config.exs index 0d1421c..e705f8e 100644 --- a/config/config.exs +++ b/config/config.exs @@ -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 diff --git a/config/runtime.exs b/config/runtime.exs index 5116182..bf96e88 100644 --- a/config/runtime.exs +++ b/config/runtime.exs @@ -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]} diff --git a/config/test.exs b/config/test.exs index f37a111..bee7f95 100644 --- a/config/test.exs +++ b/config/test.exs @@ -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", diff --git a/lib/aprsme/application.ex b/lib/aprsme/application.ex index 86f5721..a8bfd46 100644 --- a/lib/aprsme/application.ex +++ b/lib/aprsme/application.ex @@ -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") diff --git a/lib/aprsme/cleanup_scheduler.ex b/lib/aprsme/cleanup_scheduler.ex index ec9c071..aeb2d06 100644 --- a/lib/aprsme/cleanup_scheduler.ex +++ b/lib/aprsme/cleanup_scheduler.ex @@ -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 \ No newline at end of file +end diff --git a/lib/aprsme/encoding.ex b/lib/aprsme/encoding.ex index 4685b2e..7ede38f 100644 --- a/lib/aprsme/encoding.ex +++ b/lib/aprsme/encoding.ex @@ -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 \ No newline at end of file +end diff --git a/lib/aprsme/encoding_utils.ex b/lib/aprsme/encoding_utils.ex index 45feb30..2d2e41e 100644 --- a/lib/aprsme/encoding_utils.ex +++ b/lib/aprsme/encoding_utils.ex @@ -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 diff --git a/lib/aprsme/migration_lock.ex b/lib/aprsme/migration_lock.ex index 418be35..8dfc6be 100644 --- a/lib/aprsme/migration_lock.ex +++ b/lib/aprsme/migration_lock.ex @@ -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 \ No newline at end of file +end diff --git a/lib/aprsme/release.ex b/lib/aprsme/release.ex index 36000be..28fc8f0 100644 --- a/lib/aprsme/release.ex +++ b/lib/aprsme/release.ex @@ -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} -> diff --git a/mix.exs b/mix.exs index 6744852..0c36749 100644 --- a/mix.exs +++ b/mix.exs @@ -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 diff --git a/priv/repo/migrations/20250726215048_add_optimized_indexes.exs b/priv/repo/migrations/20250726215048_add_optimized_indexes.exs index b1a41e0..50888c7 100644 --- a/priv/repo/migrations/20250726215048_add_optimized_indexes.exs +++ b/priv/repo/migrations/20250726215048_add_optimized_indexes.exs @@ -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;" diff --git a/priv/repo/migrations/20250727000000_add_has_weather_column.exs b/priv/repo/migrations/20250727000000_add_has_weather_column.exs index fed7b08..11fc5b8 100644 --- a/priv/repo/migrations/20250727000000_add_has_weather_column.exs +++ b/priv/repo/migrations/20250727000000_add_has_weather_column.exs @@ -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 \ No newline at end of file +end