diff --git a/docs/improvement-todos.md b/docs/improvement-todos.md new file mode 100644 index 0000000..6ae6c71 --- /dev/null +++ b/docs/improvement-todos.md @@ -0,0 +1,151 @@ +# APRS.me Improvement TODOs + +This document tracks potential improvements identified during the multi-replica Kubernetes deployment setup. + +## High Priority + +### 1. Implement Distributed Caching with Redis +- **Status**: Pending +- **Impact**: High - Reduce database load, improve response times +- **Details**: + - Cache frequently accessed packet queries + - Cache callsign lookups + - Cache weather data aggregations + - Cache map viewport data + - Use Cachex with Redis adapter or direct Redis commands + - Implement cache invalidation strategies + +### 2. Optimize Database Queries with Better Indexes +- **Status**: Pending +- **Impact**: High - Improve query performance +- **Details**: + - Add composite indexes for common query patterns + - Optimize spatial queries with better PostGIS indexes + - Consider materialized views for complex aggregations + - Analyze slow query logs to identify bottlenecks + +### 3. Add Metrics and Monitoring with Prometheus +- **Status**: Pending +- **Impact**: High - Production visibility +- **Details**: + - Add Prometheus metrics exporter (prometheus_ex) + - Track packet processing rates and latencies + - Monitor connection pool usage (PgBouncer & app) + - Track cache hit rates + - Add custom business metrics + - Monitor APRS-IS connection stability + +## Medium Priority + +### 4. Add Connection Draining for Graceful Shutdowns +- **Status**: Pending +- **Impact**: Medium - Better user experience during deployments +- **Details**: + - Implement proper shutdown handlers for WebSocket connections + - Allow in-flight requests to complete before pod termination + - Add preStop hooks to Kubernetes deployment + - Handle SIGTERM gracefully + +### 5. Implement Distributed Rate Limiting Across Replicas +- **Status**: Pending +- **Impact**: Medium - Consistent rate limiting +- **Details**: + - Currently using ETS-based rate limiting (not distributed) + - Use Redis for distributed rate limit counters + - Implement sliding window rate limiting + - Share rate limit state across all pods + - Consider using Hammer with Redis backend + +### 6. Add Comprehensive Health Checks +- **Status**: Pending +- **Impact**: Medium - Better Kubernetes integration +- **Details**: + - Enhance beyond basic /health endpoint + - Add database connectivity checks + - Add Redis connectivity checks + - Add APRS-IS connection status checks + - Add resource usage checks (memory, connections) + - Separate readiness vs liveness probes + +### 7. Implement Horizontal Pod Autoscaling +- **Status**: Pending +- **Impact**: Medium - Auto-scaling based on load +- **Details**: + - Configure HPA based on CPU/memory usage + - Consider custom metrics (packet processing rate) + - Ensure proper resource requests/limits + - Test scaling behavior under load + +## Low Priority + +### 8. Enhance Circuit Breakers +- **Status**: Pending +- **Impact**: Low - Resilience improvement +- **Details**: + - Already have Aprsme.CircuitBreaker module + - Add circuit breakers for database connections + - Implement fallback mechanisms + - Add circuit breaker metrics + - Consider using fuse library + +## Additional Improvements Identified + +### 9. Session Affinity for WebSockets +- Consider implementing sticky sessions for WebSocket connections +- Or implement WebSocket connection state migration +- May improve user experience during pod scaling + +### 10. Background Job Optimization +- Oban jobs could use Redis for better distributed processing +- Implement job priorities and queues +- Add job monitoring and metrics +- Consider using Oban Pro features + +### 11. Optimize JavaScript Bundle Size Further +- Analyze bundle with webpack-bundle-analyzer equivalent +- Consider lazy loading more components +- Implement code splitting for routes +- Remove any remaining unused dependencies + +### 12. Database Connection Pool Tuning +- Monitor PgBouncer pool usage patterns +- Adjust pool sizes based on actual usage +- Consider separate pools for read/write operations +- Implement connection pool warmup + +### 13. Implement Distributed Tracing +- Add OpenTelemetry support +- Trace requests across the system +- Identify performance bottlenecks +- Integrate with Jaeger or similar + +### 14. Security Enhancements +- Implement CSRF protection for non-API routes +- Add rate limiting per IP/user +- Implement API key management for external access +- Add security headers (HSTS, CSP, etc.) + +### 15. Performance Optimizations +- Implement ETL for historical data +- Add data archival strategies +- Optimize Phoenix Channels for large subscriber counts +- Consider read replicas for heavy read workloads + +## Implementation Priority + +Based on current system state with Redis and PgBouncer already deployed: + +1. **Distributed Caching** - Immediate high impact, infrastructure ready +2. **Metrics/Monitoring** - Essential for production visibility +3. **Database Indexes** - Query performance improvements +4. **Enhanced Health Checks** - Better Kubernetes integration +5. **Connection Draining** - Improved deployment experience + +## Notes + +- Redis infrastructure is already in place (used for PubSub) +- PgBouncer is configured and working for connection pooling +- Kubernetes cluster is configured with StatefulSet for stable networking +- Current setup handles ~8-21 packets/second with 2 replicas + +Last updated: 2025-07-26 \ No newline at end of file diff --git a/lib/aprsme/application.ex b/lib/aprsme/application.ex index 90f5305..b681a03 100644 --- a/lib/aprsme/application.ex +++ b/lib/aprsme/application.ex @@ -23,12 +23,8 @@ defmodule Aprsme.Application do Aprsme.Repo, # Start the PubSub system pubsub_config(), - # Start the rate limiter - Aprsme.RateLimiter, - # Start cache systems - %{id: :query_cache, start: {Cachex, :start_link, [:query_cache, [limit: 10_000]]}}, - %{id: :device_cache, start: {Cachex, :start_link, [:device_cache, [limit: 5_000]]}}, - %{id: :symbol_cache, start: {Cachex, :start_link, [:symbol_cache, [limit: 1_000]]}}, + # Start Redis-based rate limiter and caches (only if Redis is available) + redis_children(), # Start circuit breaker Aprsme.CircuitBreaker, # Start device cache manager @@ -184,4 +180,33 @@ defmodule Aprsme.Application do {Phoenix.PubSub, name: Aprsme.PubSub} end end + + defp redis_children do + if System.get_env("REDIS_URL") do + require Logger + + Logger.info("Starting Redis-based caching and rate limiting") + + [ + # Redis-based rate limiter + Aprsme.RedisRateLimiter, + # Redis-based caches + {Aprsme.RedisCache, name: :query_cache}, + {Aprsme.RedisCache, name: :device_cache}, + {Aprsme.RedisCache, name: :symbol_cache} + ] + else + require Logger + + Logger.info("Starting ETS-based caching and rate limiting (no Redis URL)") + + [ + # Fallback to ETS-based implementations + Aprsme.RateLimiter, + %{id: :query_cache, start: {Cachex, :start_link, [:query_cache, [limit: 10_000]]}}, + %{id: :device_cache, start: {Cachex, :start_link, [:device_cache, [limit: 5_000]]}}, + %{id: :symbol_cache, start: {Cachex, :start_link, [:symbol_cache, [limit: 1_000]]}} + ] + end + end end diff --git a/lib/aprsme/cache.ex b/lib/aprsme/cache.ex new file mode 100644 index 0000000..a6ab023 --- /dev/null +++ b/lib/aprsme/cache.ex @@ -0,0 +1,124 @@ +defmodule Aprsme.Cache do + @moduledoc """ + Cache abstraction layer that works with both Cachex and RedisCache. + Provides a unified API regardless of the underlying implementation. + """ + + @doc """ + Get a value from cache + """ + def get(cache_name, key) do + if using_redis?() do + Aprsme.RedisCache.get(cache_name, key) + else + Cachex.get(cache_name, key) + end + end + + @doc """ + Put a value in cache with optional TTL + """ + def put(cache_name, key, value, opts \\ []) do + if using_redis?() do + # Convert TTL from milliseconds to seconds for Redis + opts = convert_ttl_to_seconds(opts) + Aprsme.RedisCache.put(cache_name, key, value, opts) + else + Cachex.put(cache_name, key, value, opts) + end + end + + @doc """ + Delete a key from cache + """ + def del(cache_name, key) do + if using_redis?() do + Aprsme.RedisCache.del(cache_name, key) + else + Cachex.del(cache_name, key) + end + end + + @doc """ + Clear all keys from cache + """ + def clear(cache_name) do + if using_redis?() do + Aprsme.RedisCache.clear(cache_name) + else + Cachex.clear(cache_name) + end + end + + @doc """ + Get cache statistics + """ + def stats(cache_name) do + if using_redis?() do + Aprsme.RedisCache.stats(cache_name) + else + Cachex.stats(cache_name) + end + end + + @doc """ + Check if key exists + """ + def exists?(cache_name, key) do + if using_redis?() do + Aprsme.RedisCache.exists?(cache_name, key) + else + Cachex.exists?(cache_name, key) + end + end + + @doc """ + Get TTL for a key + """ + def ttl(cache_name, key) do + if using_redis?() do + Aprsme.RedisCache.ttl(cache_name, key) + else + Cachex.ttl(cache_name, key) + end + end + + # Helper functions + + defp using_redis? do + System.get_env("REDIS_URL") != nil + end + + defp convert_ttl_to_seconds(opts) do + case Keyword.get(opts, :ttl) do + nil -> + opts + + ttl_ms when is_integer(ttl_ms) -> + # Convert milliseconds to seconds + Keyword.put(opts, :ttl, div(ttl_ms, 1000)) + + _ -> + opts + end + end + + @doc """ + Convert timeout keyword list to milliseconds + """ + def to_timeout(opts) do + # Cachex's to_timeout is private, so we implement our own + Enum.reduce(opts, 0, fn + {:second, n}, acc -> acc + n * 1000 + {:seconds, n}, acc -> acc + n * 1000 + {:minute, n}, acc -> acc + n * 60 * 1000 + {:minutes, n}, acc -> acc + n * 60 * 1000 + {:hour, n}, acc -> acc + n * 60 * 60 * 1000 + {:hours, n}, acc -> acc + n * 60 * 60 * 1000 + {:day, n}, acc -> acc + n * 24 * 60 * 60 * 1000 + {:days, n}, acc -> acc + n * 24 * 60 * 60 * 1000 + {:millisecond, n}, acc -> acc + n + {:milliseconds, n}, acc -> acc + n + end) + end +end diff --git a/lib/aprsme/cached_queries.ex b/lib/aprsme/cached_queries.ex index b4c88a3..9d53b8c 100644 --- a/lib/aprsme/cached_queries.ex +++ b/lib/aprsme/cached_queries.ex @@ -3,15 +3,16 @@ defmodule Aprsme.CachedQueries do Caching layer for database queries to improve performance """ + alias Aprsme.Cache alias Aprsme.Packet alias Aprsme.Packets alias Aprsme.Repo alias Ecto.Adapters.SQL # 1 minute for frequently changing data - @cache_ttl_short to_timeout(minute: 1) + @cache_ttl_short Cache.to_timeout(minute: 1) # 1 minute for moderately changing data - @cache_ttl_medium to_timeout(minute: 1) + @cache_ttl_medium Cache.to_timeout(minute: 1) @doc """ Get recent packets with caching @@ -19,13 +20,13 @@ defmodule Aprsme.CachedQueries do def get_recent_packets_cached(opts) do cache_key = generate_cache_key("recent_packets", opts) - case Cachex.get(:query_cache, cache_key) do + case Cache.get(:query_cache, cache_key) do {:ok, result} when not is_nil(result) -> result _ -> result = Packets.get_recent_packets_optimized(opts) - Cachex.put(:query_cache, cache_key, result, ttl: @cache_ttl_short) + Cache.put(:query_cache, cache_key, result, ttl: @cache_ttl_short) result end end @@ -36,13 +37,13 @@ defmodule Aprsme.CachedQueries do def get_weather_packets_cached(callsign, start_time, end_time, opts) do cache_key = generate_cache_key("weather", {callsign, start_time, end_time, opts}) - case Cachex.get(:query_cache, cache_key) do + case Cache.get(:query_cache, cache_key) do {:ok, result} when not is_nil(result) -> result _ -> result = Packets.get_weather_packets(callsign, start_time, end_time, opts) - Cachex.put(:query_cache, cache_key, result, ttl: @cache_ttl_medium) + Cache.put(:query_cache, cache_key, result, ttl: @cache_ttl_medium) result end end @@ -54,7 +55,7 @@ defmodule Aprsme.CachedQueries do def get_total_packet_count_cached do cache_key = "total_packet_count" - case Cachex.get(:query_cache, cache_key) do + case Cache.get(:query_cache, cache_key) do {:ok, result} when not is_nil(result) -> result @@ -62,7 +63,7 @@ defmodule Aprsme.CachedQueries do # This is now extremely fast due to the counter table result = Packets.get_total_packet_count() # Cache for only 5 seconds since the query is now instant - Cachex.put(:query_cache, cache_key, result, ttl: to_timeout(second: 5)) + Cache.put(:query_cache, cache_key, result, ttl: Cache.to_timeout(second: 5)) result end end @@ -73,14 +74,14 @@ defmodule Aprsme.CachedQueries do def get_latest_packet_for_callsign_cached(callsign) do cache_key = generate_cache_key("latest_packet", callsign) - case Cachex.get(:query_cache, cache_key) do + case Cache.get(:query_cache, cache_key) do {:ok, result} when not is_nil(result) -> result _ -> result = Packets.get_latest_packet_for_callsign(callsign) # Shorter TTL for latest packets as they change frequently - Cachex.put(:query_cache, cache_key, result, ttl: @cache_ttl_short) + Cache.put(:query_cache, cache_key, result, ttl: @cache_ttl_short) result end end @@ -92,14 +93,14 @@ defmodule Aprsme.CachedQueries do def get_latest_weather_packet_cached(callsign) do cache_key = generate_cache_key("latest_weather_packet", callsign) - case Cachex.get(:query_cache, cache_key) do + case Cache.get(:query_cache, cache_key) do {:ok, result} when not is_nil(result) -> result _ -> result = Packets.get_latest_weather_packet(callsign) # Cache for 5 minutes since weather updates are less frequent - Cachex.put(:query_cache, cache_key, result, ttl: @cache_ttl_short) + Cache.put(:query_cache, cache_key, result, ttl: @cache_ttl_short) result end end @@ -111,7 +112,7 @@ defmodule Aprsme.CachedQueries do def has_weather_packets_cached?(callsign) do cache_key = generate_cache_key("has_weather_packets", callsign) - case Cachex.get(:query_cache, cache_key) do + case Cache.get(:query_cache, cache_key) do {:ok, result} when not is_nil(result) -> result @@ -130,7 +131,7 @@ defmodule Aprsme.CachedQueries do result = Repo.exists?(query) # Cache for 15 minutes - Cachex.put(:query_cache, cache_key, result, ttl: @cache_ttl_medium) + Cache.put(:query_cache, cache_key, result, ttl: @cache_ttl_medium) result end end @@ -148,7 +149,7 @@ defmodule Aprsme.CachedQueries do ] Enum.each(patterns, fn pattern -> - Cachex.del(:query_cache, pattern) + Cache.del(:query_cache, pattern) end) end @@ -156,14 +157,14 @@ defmodule Aprsme.CachedQueries do Invalidate all cached queries """ def invalidate_all_cache do - Cachex.clear(:query_cache) + Cache.clear(:query_cache) end @doc """ Get cache statistics """ def get_cache_stats do - {:ok, stats} = Cachex.stats(:query_cache) + {:ok, stats} = Cache.stats(:query_cache) stats end @@ -173,7 +174,7 @@ defmodule Aprsme.CachedQueries do def get_path_station_positions_cached(callsigns) when is_list(callsigns) do cache_key = generate_cache_key("path_stations", Enum.sort(callsigns)) - case Cachex.get(:query_cache, cache_key) do + case Cache.get(:query_cache, cache_key) do {:ok, result} when not is_nil(result) -> result @@ -214,7 +215,7 @@ defmodule Aprsme.CachedQueries do end # Cache for 5 minutes - Cachex.put(:query_cache, cache_key, result, ttl: @cache_ttl_medium) + Cache.put(:query_cache, cache_key, result, ttl: @cache_ttl_medium) result end end diff --git a/lib/aprsme/device_cache.ex b/lib/aprsme/device_cache.ex index 862c8d4..8635440 100644 --- a/lib/aprsme/device_cache.ex +++ b/lib/aprsme/device_cache.ex @@ -5,11 +5,12 @@ defmodule Aprsme.DeviceCache do use GenServer + alias Aprsme.Cache alias Aprsme.Devices alias Aprsme.Repo @cache_name :device_cache - @refresh_interval to_timeout(day: 1) + @refresh_interval Cache.to_timeout(day: 1) # Client API @@ -27,7 +28,7 @@ defmodule Aprsme.DeviceCache do def lookup_device(nil), do: nil def lookup_device(identifier) when is_binary(identifier) do - case Cachex.get(@cache_name, :all_devices) do + case Cache.get(@cache_name, :all_devices) do {:ok, nil} -> # Cache miss - trigger async refresh and return nil for now GenServer.cast(__MODULE__, :refresh_cache_async) @@ -97,7 +98,7 @@ defmodule Aprsme.DeviceCache do devices = Repo.all(Devices) # Store all devices in cache - case Cachex.put(@cache_name, :all_devices, devices) do + case Cache.put(@cache_name, :all_devices, devices) do {:ok, true} -> :ok error -> error end @@ -106,7 +107,7 @@ defmodule Aprsme.DeviceCache do # Handle case where database or table doesn't exist yet Logger.warning("Failed to load devices: #{inspect(error)}. Will retry later.") # Store empty list for now - Cachex.put(@cache_name, :all_devices, []) + Cache.put(@cache_name, :all_devices, []) :error end end diff --git a/lib/aprsme/rate_limiter_wrapper.ex b/lib/aprsme/rate_limiter_wrapper.ex new file mode 100644 index 0000000..11e1230 --- /dev/null +++ b/lib/aprsme/rate_limiter_wrapper.ex @@ -0,0 +1,46 @@ +defmodule Aprsme.RateLimiterWrapper do + @moduledoc """ + Wrapper that provides the same API for both Hammer and RedisRateLimiter + """ + + @doc """ + Check rate limit - compatible with Hammer.hit/3 API + """ + def hit(bucket, scale_ms, limit) do + if using_redis?() do + Aprsme.RedisRateLimiter.check_rate(bucket, limit, scale_ms) + else + Aprsme.RateLimiter.hit(bucket, scale_ms, limit) + end + end + + @doc """ + Reset rate limit for a bucket + """ + def reset(bucket) do + if using_redis?() do + Aprsme.RedisRateLimiter.reset(bucket) + else + # Hammer doesn't provide a reset function + # Best we can do is return ok + :ok + end + end + + @doc """ + Get current count for a bucket + """ + def count(bucket, scale_ms) do + if using_redis?() do + Aprsme.RedisRateLimiter.count(bucket, scale_ms) + else + # Hammer doesn't provide a count function + # Return 0 as default + {:ok, 0} + end + end + + defp using_redis? do + System.get_env("REDIS_URL") != nil + end +end diff --git a/lib/aprsme/redis_cache.ex b/lib/aprsme/redis_cache.ex new file mode 100644 index 0000000..e9f48b5 --- /dev/null +++ b/lib/aprsme/redis_cache.ex @@ -0,0 +1,253 @@ +defmodule Aprsme.RedisCache do + @moduledoc """ + Redis-backed distributed cache implementation. + Provides a similar API to Cachex but uses Redis for distributed caching. + """ + + require Logger + + # 5 minutes in seconds + @default_ttl 300 + @redis_pool_size 5 + + def child_spec(opts) do + name = Keyword.fetch!(opts, :name) + + children = [ + {Redix, + name: redis_name(name), + host: redis_host(), + port: redis_port(), + password: redis_password(), + pool_size: @redis_pool_size} + ] + + %{ + id: {__MODULE__, name}, + type: :supervisor, + start: {Supervisor, :start_link, [children, [strategy: :one_for_one, name: :"#{name}_redis_supervisor"]]} + } + end + + @doc """ + Get a value from the cache + """ + def get(cache_name, key) do + redis_key = make_redis_key(cache_name, key) + + case Redix.command(redis_name(cache_name), ["GET", redis_key]) do + {:ok, nil} -> + {:ok, nil} + + {:ok, value} -> + {:ok, deserialize(value)} + + {:error, reason} -> + Logger.error("Redis GET error for #{redis_key}: #{inspect(reason)}") + {:error, reason} + end + end + + @doc """ + Put a value in the cache with optional TTL + """ + def put(cache_name, key, value, opts \\ []) do + redis_key = make_redis_key(cache_name, key) + ttl = Keyword.get(opts, :ttl, @default_ttl) + serialized = serialize(value) + + case Redix.command(redis_name(cache_name), ["SETEX", redis_key, ttl, serialized]) do + {:ok, "OK"} -> + {:ok, true} + + {:error, reason} -> + Logger.error("Redis SETEX error for #{redis_key}: #{inspect(reason)}") + {:error, reason} + end + end + + @doc """ + Delete a value from the cache + """ + def del(cache_name, key) do + redis_key = make_redis_key(cache_name, key) + + case Redix.command(redis_name(cache_name), ["DEL", redis_key]) do + {:ok, _} -> + {:ok, true} + + {:error, reason} -> + Logger.error("Redis DEL error for #{redis_key}: #{inspect(reason)}") + {:error, reason} + end + end + + @doc """ + Delete multiple keys matching a pattern + """ + def del_pattern(cache_name, pattern) do + redis_pattern = make_redis_key(cache_name, pattern) + + # Use SCAN to find keys matching pattern + case scan_keys(cache_name, redis_pattern) do + {:ok, keys} when keys != [] -> + case Redix.command(redis_name(cache_name), ["DEL" | keys]) do + {:ok, count} -> + {:ok, count} + + {:error, reason} -> + Logger.error("Redis DEL error for pattern #{redis_pattern}: #{inspect(reason)}") + {:error, reason} + end + + {:ok, []} -> + {:ok, 0} + + {:error, reason} -> + {:error, reason} + end + end + + @doc """ + Clear all keys for a cache + """ + def clear(cache_name) do + pattern = make_redis_key(cache_name, "*") + + case scan_keys(cache_name, pattern) do + {:ok, keys} when keys != [] -> + case Redix.pipeline(redis_name(cache_name), Enum.map(keys, &["DEL", &1])) do + {:ok, _results} -> + {:ok, true} + + {:error, reason} -> + Logger.error("Redis CLEAR error for #{cache_name}: #{inspect(reason)}") + {:error, reason} + end + + {:ok, []} -> + {:ok, true} + + {:error, reason} -> + {:error, reason} + end + end + + @doc """ + Get cache statistics + """ + def stats(cache_name) do + pattern = make_redis_key(cache_name, "*") + + case scan_keys(cache_name, pattern) do + {:ok, keys} -> + {:ok, + %{ + size: length(keys), + keys: keys + }} + + {:error, reason} -> + {:error, reason} + end + end + + @doc """ + Check if a key exists + """ + def exists?(cache_name, key) do + redis_key = make_redis_key(cache_name, key) + + case Redix.command(redis_name(cache_name), ["EXISTS", redis_key]) do + {:ok, 1} -> true + {:ok, 0} -> false + {:error, _} -> false + end + end + + @doc """ + Get remaining TTL for a key + """ + def ttl(cache_name, key) do + redis_key = make_redis_key(cache_name, key) + + case Redix.command(redis_name(cache_name), ["TTL", redis_key]) do + # Key doesn't exist + {:ok, -2} -> {:ok, nil} + # Key exists but has no TTL + {:ok, -1} -> {:ok, :infinity} + {:ok, seconds} -> {:ok, seconds} + {:error, reason} -> {:error, reason} + end + end + + # Private functions + + defp redis_name(cache_name), do: :"#{cache_name}_redis" + + defp make_redis_key(cache_name, key) when is_binary(key) do + "aprsme:#{cache_name}:#{key}" + end + + defp make_redis_key(cache_name, key) do + "aprsme:#{cache_name}:#{:erlang.phash2(key)}" + end + + defp serialize(value) do + :erlang.term_to_binary(value) + end + + defp deserialize(binary) when is_binary(binary) do + :erlang.binary_to_term(binary) + rescue + _ -> nil + end + + defp scan_keys(cache_name, pattern) do + scan_keys(cache_name, pattern, "0", []) + end + + defp scan_keys(cache_name, pattern, cursor, acc) do + case Redix.command(redis_name(cache_name), ["SCAN", cursor, "MATCH", pattern, "COUNT", "100"]) do + {:ok, [new_cursor, keys]} -> + new_acc = acc ++ keys + + if new_cursor == "0" do + {:ok, new_acc} + else + scan_keys(cache_name, pattern, new_cursor, new_acc) + end + + {:error, reason} -> + {:error, reason} + end + end + + defp redis_host do + redis_url = System.get_env("REDIS_URL", "redis://localhost:6379") + uri = URI.parse(redis_url) + uri.host || "localhost" + end + + defp redis_port do + redis_url = System.get_env("REDIS_URL", "redis://localhost:6379") + uri = URI.parse(redis_url) + uri.port || 6379 + end + + defp redis_password do + redis_url = System.get_env("REDIS_URL", "redis://localhost:6379") + uri = URI.parse(redis_url) + + case uri.userinfo do + nil -> + nil + + userinfo -> + case String.split(userinfo, ":") do + [_, password] -> password + _ -> nil + end + end + end +end diff --git a/lib/aprsme/redis_rate_limiter.ex b/lib/aprsme/redis_rate_limiter.ex new file mode 100644 index 0000000..d907c6b --- /dev/null +++ b/lib/aprsme/redis_rate_limiter.ex @@ -0,0 +1,159 @@ +defmodule Aprsme.RedisRateLimiter do + @moduledoc """ + Redis-backed distributed rate limiter using sliding window algorithm. + """ + + require Logger + + @redis_pool_size 5 + + def child_spec(_opts) do + children = [ + {Redix, + name: __MODULE__, host: redis_host(), port: redis_port(), password: redis_password(), pool_size: @redis_pool_size} + ] + + %{ + id: __MODULE__, + type: :supervisor, + start: {Supervisor, :start_link, [children, [strategy: :one_for_one, name: :"#{__MODULE__}_supervisor"]]} + } + end + + @doc """ + Check rate limit using sliding window algorithm. + + ## Parameters + - bucket: The rate limit bucket (e.g., "user:123" or "ip:192.168.1.1") + - limit: Maximum number of requests allowed + - window_ms: Time window in milliseconds + + ## Returns + - {:allow, count} if under limit + - {:deny, limit} if over limit + """ + def check_rate(bucket, limit, window_ms) do + now = System.system_time(:millisecond) + window_start = now - window_ms + key = make_key(bucket) + + # Lua script for atomic sliding window rate limiting + lua_script = """ + local key = KEYS[1] + local now = tonumber(ARGV[1]) + local window_start = tonumber(ARGV[2]) + local limit = tonumber(ARGV[3]) + + -- Remove old entries + redis.call('ZREMRANGEBYSCORE', key, 0, window_start) + + -- Count current entries + local current = redis.call('ZCARD', key) + + if current < limit then + -- Add new entry + redis.call('ZADD', key, now, now) + -- Set expiry + redis.call('EXPIRE', key, math.ceil(ARGV[4] / 1000)) + return {1, current + 1} + else + return {0, limit} + end + """ + + case Redix.command(__MODULE__, ["EVAL", lua_script, 1, key, now, window_start, limit, window_ms]) do + {:ok, [1, count]} -> + {:allow, count} + + {:ok, [0, limit]} -> + {:deny, limit} + + {:error, reason} -> + Logger.error("Redis rate limit error for #{bucket}: #{inspect(reason)}") + # Fail open - allow request if Redis is down + {:allow, 1} + end + end + + @doc """ + Reset rate limit for a bucket + """ + def reset(bucket) do + key = make_key(bucket) + + case Redix.command(__MODULE__, ["DEL", key]) do + {:ok, _} -> + :ok + + {:error, reason} -> + Logger.error("Redis rate limit reset error for #{bucket}: #{inspect(reason)}") + {:error, reason} + end + end + + @doc """ + Get current count for a bucket + """ + def count(bucket, window_ms) do + now = System.system_time(:millisecond) + window_start = now - window_ms + key = make_key(bucket) + + # Clean up old entries and count + case Redix.pipeline(__MODULE__, [ + ["ZREMRANGEBYSCORE", key, 0, window_start], + ["ZCARD", key] + ]) do + {:ok, [_, count]} -> + {:ok, count} + + {:error, reason} -> + Logger.error("Redis rate limit count error for #{bucket}: #{inspect(reason)}") + {:error, reason} + end + end + + @doc """ + Get remaining requests for a bucket + """ + def remaining(bucket, limit, window_ms) do + case count(bucket, window_ms) do + {:ok, current} -> {:ok, max(0, limit - current)} + error -> error + end + end + + # Private functions + + defp make_key(bucket) do + "aprsme:rate_limit:#{bucket}" + end + + defp redis_host do + redis_url = System.get_env("REDIS_URL", "redis://localhost:6379") + uri = URI.parse(redis_url) + uri.host || "localhost" + end + + defp redis_port do + redis_url = System.get_env("REDIS_URL", "redis://localhost:6379") + uri = URI.parse(redis_url) + uri.port || 6379 + end + + defp redis_password do + redis_url = System.get_env("REDIS_URL", "redis://localhost:6379") + uri = URI.parse(redis_url) + + case uri.userinfo do + nil -> + nil + + userinfo -> + case String.split(userinfo, ":") do + [_, password] -> password + _ -> nil + end + end + end +end diff --git a/lib/aprsme_web/aprs_symbol.ex b/lib/aprsme_web/aprs_symbol.ex index b9a16e0..b7b9202 100644 --- a/lib/aprsme_web/aprs_symbol.ex +++ b/lib/aprsme_web/aprs_symbol.ex @@ -238,14 +238,14 @@ defmodule AprsmeWeb.AprsSymbol do if is_nil(callsign) do cache_key = "symbol_html:#{symbol_table}:#{symbol_code}:#{size}" - case Cachex.get(:symbol_cache, cache_key) do + case Aprsme.Cache.get(:symbol_cache, cache_key) do {:ok, html} when not is_nil(html) -> html _ -> html = generate_marker_html(symbol_table, symbol_code, nil, size) # Cache for 1 hour since symbols don't change - Cachex.put(:symbol_cache, cache_key, html, ttl: to_timeout(hour: 1)) + Aprsme.Cache.put(:symbol_cache, cache_key, html, ttl: Aprsme.Cache.to_timeout(hour: 1)) html end else diff --git a/lib/aprsme_web/plugs/rate_limiter.ex b/lib/aprsme_web/plugs/rate_limiter.ex index 9c3e7c0..1f175a2 100644 --- a/lib/aprsme_web/plugs/rate_limiter.ex +++ b/lib/aprsme_web/plugs/rate_limiter.ex @@ -27,7 +27,7 @@ defmodule AprsmeWeb.Plugs.RateLimiter do limit = opts[:limit] error_message = opts[:error_message] - case Aprsme.RateLimiter.hit("rate_limit:#{key}", scale, limit) do + case Aprsme.RateLimiterWrapper.hit("rate_limit:#{key}", scale, limit) do {:allow, _count} -> conn