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

Add Redis-based distributed caching and rate limiting

Browse source 
This commit introduces a dual-mode caching and rate limiting system that can use
either Redis (for distributed deployments) or ETS/Cachex (for single-node setups).

Key changes:
- Add Cache abstraction layer that automatically switches between Redis and Cachex
- Implement RedisCache module with full distributed cache functionality
- Implement RedisRateLimiter with sliding window algorithm for accurate rate limiting
- Add RateLimiterWrapper to provide unified API for both implementations
- Update application startup to conditionally use Redis when REDIS_URL is set
- Migrate all cache operations to use the new Cache abstraction
- Support graceful fallback to ETS-based solutions when Redis is unavailable

The system automatically detects Redis availability via REDIS_URL environment
variable and switches between implementations without code changes. This enables
proper distributed caching and rate limiting in Kubernetes deployments while
maintaining backward compatibility for development environments.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Graham McIntire 2025-07-26 15:29:38 -05:00
parent 4fcdc33ec0
commit 226a53bf61
No known key found for this signature in database
10 changed files with 792 additions and 32 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

151
docs/improvement-todos.md Normal file
View file

@ -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

37
lib/aprsme/application.ex
View file

@ -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

124
lib/aprsme/cache.ex Normal file
View file

@ -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

39
lib/aprsme/cached_queries.ex
View file

@ -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

9
lib/aprsme/device_cache.ex
View file

@ -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

46
lib/aprsme/rate_limiter_wrapper.ex Normal file
View file

@ -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

253
lib/aprsme/redis_cache.ex Normal file
View file

@ -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

159
lib/aprsme/redis_rate_limiter.ex Normal file
View file

@ -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

4
lib/aprsme_web/aprs_symbol.ex
View file

@ -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

2
lib/aprsme_web/plugs/rate_limiter.ex
View file

@ -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