feat: add redis

CONTRIBUTING.md

@@ -145,6 +145,27 @@ For paginated responses use the generic page wrapper:
 
 ---
 
+## Caching
+
+Beacon uses a Redis-backed caching layer in `internal/cache`. If you add a new
+REST endpoint backed by a new `api.Reader` method, consider whether it should be
+cached:
+
+- Add the method to `CachedReader` in `internal/cache/reader.go`
+- Pass-through to `cr.inner` if the data is highly dynamic, paginated with many
+  filter combinations, or low traffic
+- Use `getOrSet` with an appropriate TTL category (`cr.ttl.Stats`,
+  `cr.ttl.Reference`, `cr.ttl.Nodes`, or `cr.ttl.Observers`) for read-heavy,
+  slow-changing responses
+- If the data is mutated by the ingest path, add an invalidation call in
+  `internal/ingest/side_effects.go` via the `onNodeUpsert` or `onObserverUpsert`
+  callbacks, or add a new callback following the same pattern
+
+Cache keys live as constants at the top of `reader.go`. Use the `beacon:`
+namespace prefix and include all parameters that affect the response in the key.
+
+---
+
 ## Updating the IATA database
 
 Beacon includes a static IATA → country/continent mapping compiled into the

README.md

@@ -107,17 +107,20 @@ improves automatically as the mesh is observed over time.
 
 ### Environment variables (`.env`)
 
-| Variable                 | Default       | Description                                                 |
-| ------------------------ | ------------- | ----------------------------------------------------------- |
-| `LISTEN_ADDR`            | `:8080`       | HTTP listen address                                         |
-| `POSTGRES_DSN`           | —             | PostgreSQL connection string                                |
-| `CONFIG_PATH`            | `config.yaml` | Path to YAML config file                                    |
-| `MQTT_BROKER_1_URL`      | —             | Broker 1 WebSocket URL (e.g. `wss://mqtt1.example.com:443`) |
-| `MQTT_BROKER_1_USERNAME` | —             | Broker 1 username                                           |
-| `MQTT_BROKER_1_PASSWORD` | —             | Broker 1 password                                           |
-| `MQTT_BROKER_2_URL`      | —             | Broker 2 WebSocket URL                                      |
-| `MQTT_BROKER_2_USERNAME` | —             | Broker 2 username                                           |
-| `MQTT_BROKER_2_PASSWORD` | —             | Broker 2 password                                           |
+| Variable                 | Default       | Description                                                  |
+| ------------------------ | ------------- | ------------------------------------------------------------ |
+| `LISTEN_ADDR`            | `:8080`       | HTTP listen address                                          |
+| `POSTGRES_DSN`           | —             | PostgreSQL connection string                                 |
+| `REDIS_ADDR`             | —             | Redis address (`host:port`). Leave unset to disable caching. |
+| `REDIS_PASSWORD`         | —             | Redis password (optional)                                    |
+| `REDIS_DB`               | `0`           | Redis database index                                         |
+| `CONFIG_PATH`            | `config.yaml` | Path to YAML config file                                     |
+| `MQTT_BROKER_1_URL`      | —             | Broker 1 WebSocket URL (e.g. `wss://mqtt1.example.com:443`)  |
+| `MQTT_BROKER_1_USERNAME` | —             | Broker 1 username                                            |
+| `MQTT_BROKER_1_PASSWORD` | —             | Broker 1 password                                            |
+| `MQTT_BROKER_2_URL`      | —             | Broker 2 WebSocket URL                                       |
+| `MQTT_BROKER_2_USERNAME` | —             | Broker 2 username                                            |
+| `MQTT_BROKER_2_PASSWORD` | —             | Broker 2 password                                            |
 
 ### Config file (`config.yaml`)
 
@@ -174,6 +177,20 @@ packets:
 websocket:
   max_connections_per_ip: 5 # default: 5
 
+# Redis caching layer (optional).
+# Caches read-heavy, slow-changing responses to reduce PostgreSQL load.
+# Connection details (address, password, database) are set via environment
+# variables. Leave REDIS_ADDR unset to disable caching entirely.
+# TTLs are duration strings e.g. "30m", "1h". Per-category TTLs override
+# the global ttl. Any unset category inherits ttl. Default: 1h.
+cache:
+  ttl: "1h"
+  ttls:
+    stats: "1h" # stats endpoints (backed by materialized views)
+    reference: "1h" # IATAs, regions, scopes
+    nodes: "1h" # node detail (also explicitly invalidated on upsert)
+    observers: "1h" # observer detail (also explicitly invalidated on upsert)
+
 # Geographic ingest filter (optional).
 # Drop packets from observers outside the specified area.
 # Country codes are ISO 3166-1 alpha-2. Continent codes: AF AN AS EU NA OC SA.

cmd/beacon/main.go

@@ -12,12 +12,15 @@ import (
 	"net/http"
 	"os"
 	"os/signal"
+	"strconv"
 	"syscall"
 	"time"
 
 	"github.com/MeshCore-Beacon/beacon-server/db"
 	_ "github.com/MeshCore-Beacon/beacon-server/docs"
+	"github.com/MeshCore-Beacon/beacon-server/internal/api"
 	"github.com/MeshCore-Beacon/beacon-server/internal/api/router"
+	"github.com/MeshCore-Beacon/beacon-server/internal/cache"
 	"github.com/MeshCore-Beacon/beacon-server/internal/config"
 	"github.com/MeshCore-Beacon/beacon-server/internal/hub"
 	"github.com/MeshCore-Beacon/beacon-server/internal/iatadb"
@@ -113,6 +116,33 @@ func main() {
 
 	store := db.New(pool)
 
+	// ── Redis cache layer ────────────────────────────────────────────────────
+	var reader api.Reader = store
+	if redisAddr := os.Getenv("REDIS_ADDR"); redisAddr != "" {
+		redisClient := cache.NewClient(
+			redisAddr,
+			os.Getenv("REDIS_PASSWORD"),
+			func() int {
+				if db := os.Getenv("REDIS_DB"); db != "" {
+					n, err := strconv.Atoi(db)
+					if err == nil {
+						return n
+					}
+				}
+				return 0
+			}(),
+		)
+		if err := redisClient.Ping(ctx); err != nil {
+			log.Printf("warning: redis unavailable at %s, caching disabled: %v", redisAddr, err)
+		} else {
+			ttls := cache.ResolveTTLs(cfg.Cache)
+			reader = cache.NewCachedReader(store, redisClient, ttls)
+			defer redisClient.Close()
+			log.Printf("cache: Redis connected at %s (stats=%s reference=%s nodes=%s observers=%s)",
+				redisAddr, ttls.Stats, ttls.Reference, ttls.Nodes, ttls.Observers)
+		}
+	}
+
 	// refresh meterialized views on boot or restart to stay fresh
 	refreshMaterializedViews(ctx, store)
 
@@ -208,6 +238,11 @@ func main() {
 		scopes,
 	)
 
+	if cr, ok := reader.(*cache.CachedReader); ok {
+		broker1.SetCacheInvalidators(cr.InvalidateNode, cr.InvalidateObserver)
+		broker2.SetCacheInvalidators(cr.InvalidateNode, cr.InvalidateObserver)
+	}
+
 	go broker1.Start(ctx)
 	go broker2.Start(ctx)
 
@@ -232,7 +267,7 @@ func main() {
 	}()
 
 	// ── HTTP server ──────────────────────────────────────────────────────────
-	r := router.New(h, store, []*ingest.Worker{broker1, broker2}, maxConnsPerIP)
+	r := router.New(h, reader, []*ingest.Worker{broker1, broker2}, maxConnsPerIP)
 
 	srv := &http.Server{
 		Addr:    addr,

config.yaml.example

@@ -75,6 +75,25 @@ packets:
 websocket:
   max_connections_per_ip: 5 # default: 5
 
+# Redis caching layer.
+# If REDIS_ADDR is not set, caching is disabled and all reads go directly to PostgreSQL.
+# TTLs are specified as duration strings e.g. "30m", "1h", "2h".
+# Per-category TTLs override the global ttl. Any category left unset inherits ttl.
+cache:
+  # Global default TTL for all cached responses. Defaults to 1h if not set.
+  ttl: "1h"
+  ttls:
+    # Aggregated stats endpoints (overview, observations, breakdowns, top nodes/observers,
+    # radio presets, scope stats). Backed by materialized views refreshed hourly.
+    stats: "1h"
+    # Mostly-static reference data (IATAs, regions, scopes).
+    # Only changes when new observers arrive or config is reseeded.
+    reference: "1h"
+    # Individual node detail. Also explicitly invalidated on upsert.
+    nodes: "1h"
+    # Individual observer detail. Also explicitly invalidated on upsert.
+    observers: "1h"
+
 # Geographic ingest filter (optional).
 # Drop packets from observers outside the specified area at ingest time.
 # Country codes are ISO 3166-1 alpha-2 (e.g. CA, US, GB).

env.example

@@ -2,7 +2,10 @@ LISTEN_ADDR=:8080
 
 POSTGRES_DSN=postgres://beacon:beacon@localhost:5432/beacon?sslmode=disable
 
+# Redis (optional — leave REDIS_ADDR unset to disable caching)
 REDIS_ADDR=localhost:6379
+REDIS_PASSWORD=
+REDIS_DB=0
 
 MQTT_BROKER_1_URL=wss://mqtt1.meshcore.ca:443
 MQTT_BROKER_1_USERNAME=

go.mod

@@ -18,6 +18,7 @@ require (
 require (
 	filippo.io/edwards25519 v1.2.0 // indirect
 	github.com/KyleBanks/depth v1.2.1 // indirect
+	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/go-openapi/jsonpointer v0.19.5 // indirect
 	github.com/go-openapi/jsonreference v0.20.0 // indirect
 	github.com/go-openapi/spec v0.20.6 // indirect
@@ -28,8 +29,10 @@ require (
 	github.com/jackc/puddle/v2 v2.2.2 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/mailru/easyjson v0.7.6 // indirect
+	github.com/redis/go-redis/v9 v9.20.0 // indirect
 	github.com/rogpeppe/go-internal v1.14.1 // indirect
 	github.com/swaggo/files v0.0.0-20220610200504-28940afbdbfe // indirect
+	go.uber.org/atomic v1.11.0 // indirect
 	golang.org/x/crypto v0.51.0 // indirect
 	golang.org/x/mod v0.35.0 // indirect
 	golang.org/x/net v0.53.0 // indirect

go.sum

@@ -2,6 +2,8 @@ filippo.io/edwards25519 v1.2.0 h1:crnVqOiS4jqYleHd9vaKZ+HKtHfllngJIiOpNpoJsjo=
 filippo.io/edwards25519 v1.2.0/go.mod h1:xzAOLCNug/yB62zG1bQ8uziwrIqIuxhctzJT18Q77mc=
 github.com/KyleBanks/depth v1.2.1 h1:5h8fQADFrWtarTdtDudMmGsC7GPbOAu6RVB3ffsVFHc=
 github.com/KyleBanks/depth v1.2.1/go.mod h1:jzSb9d0L43HxTQfT+oSA1EEp2q+ne2uh6XgeJcm8brE=
+github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
+github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/coder/websocket v1.8.14 h1:9L0p0iKiNOibykf283eHkKUHHrpG7f65OE3BhhO7v9g=
 github.com/coder/websocket v1.8.14/go.mod h1:NX3SzP+inril6yawo5CQXx8+fk145lPDC6pumgx0mVg=
 github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E=
@@ -56,6 +58,8 @@ github.com/meshcore-go/meshcore-go v1.0.6/go.mod h1:u+Lvlg4Wy4blqCAJB+yGL3j2y7JR
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/redis/go-redis/v9 v9.20.0 h1:WnQYxLkgO2xiXTCJY0ldIiI8dNqCDlQAG+AtaH7a2a0=
+github.com/redis/go-redis/v9 v9.20.0/go.mod h1:v/M13XI1PVCDcm01VtPFOADfZtHf8YW3baQf57KlIkA=
 github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
 github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
@@ -70,6 +74,8 @@ github.com/swaggo/http-swagger v1.3.4 h1:q7t/XLx0n15H1Q9/tk3Y9L4n210XzJF5WtnDX64
 github.com/swaggo/http-swagger v1.3.4/go.mod h1:9dAh0unqMBAlbp1uE2Uc2mQTxNMU/ha4UbucIg1MFkQ=
 github.com/swaggo/swag v1.16.6 h1:qBNcx53ZaX+M5dxVyTrgQ0PJ/ACK+NzhwcbieTt+9yI=
 github.com/swaggo/swag v1.16.6/go.mod h1:ngP2etMK5a0P3QBizic5MEwpRmluJZPHjXcMoj4Xesg=
+go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
+go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=
 golang.org/x/crypto v0.51.0 h1:IBPXwPfKxY7cWQZ38ZCIRPI50YLeevDLlLnyC5wRGTI=
 golang.org/x/crypto v0.51.0/go.mod h1:8AdwkbraGNABw2kOX6YFPs3WM22XqI4EXEd8g+x7Oc8=
 golang.org/x/mod v0.35.0 h1:Ww1D637e6Pg+Zb2KrWfHQUnH2dQRLBQyAtpr/haaJeM=

internal/cache/cache.go

@@ -0,0 +1,119 @@
+// Copyright 2026 Beacon Contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+// Package cache provides a Redis-backed caching layer for the Beacon API.
+// It wraps api.Reader with a CachedReader that transparently caches responses
+// for read-heavy, slow-changing endpoints. If Redis is unavailable, all
+// operations degrade gracefully to the underlying reader.
+package cache
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"time"
+
+	"github.com/MeshCore-Beacon/beacon-server/internal/config"
+	redis "github.com/redis/go-redis/v9"
+)
+
+// Client wraps a Redis client with helper methods used by CachedReader.
+type Client struct {
+	rdb *redis.Client
+}
+
+// NewClient creates a new Redis client from the given address, password, and
+// database index. addr should be in "host:port" form e.g. "localhost:6379".
+// password may be empty if the Redis instance requires no authentication.
+// db is the Redis database index, typically 0.
+func NewClient(addr, password string, db int) *Client {
+	opts := redis.Options{
+		Addr:     addr,
+		Password: password,
+		DB:       db,
+	}
+	return &Client{
+		rdb: redis.NewClient(&opts),
+	}
+}
+
+// ResolveTTLs builds a CacheTTLs from the given config, applying the fallback
+// chain: category TTL → global TTL → default (1h).
+func ResolveTTLs(cfg config.CacheConfig) CacheTTLs {
+	ttls := CacheTTLs{
+		Stats:     resolve(cfg.TTLs.Stats.Duration, cfg.TTL.Duration),
+		Reference: resolve(cfg.TTLs.Reference.Duration, cfg.TTL.Duration),
+		Nodes:     resolve(cfg.TTLs.Nodes.Duration, cfg.TTL.Duration),
+		Observers: resolve(cfg.TTLs.Observers.Duration, cfg.TTL.Duration),
+	}
+	return ttls
+}
+
+// resolve returns the first non-zero duration from category, global, or 1h.
+func resolve(category, global time.Duration) time.Duration {
+	if category != 0 {
+		return category
+	}
+	if global != 0 {
+		return global
+	}
+	return time.Hour
+}
+
+// Ping checks connectivity to Redis. Call this on startup to verify the
+// connection before serving requests.
+func (c *Client) Ping(ctx context.Context) error {
+	return c.rdb.Ping(ctx).Err()
+}
+
+// Close closes the underlying Redis connection pool.
+func (c *Client) Close() error {
+	return c.rdb.Close()
+}
+
+func (c *Client) del(ctx context.Context, keys ...string) {
+	c.rdb.Del(ctx, keys...)
+}
+
+// getOrSet retrieves a cached value by key, deserialising it into T.
+// On a cache miss it calls fetch, stores the result under key with the given
+// TTL, and returns it. If Redis is unavailable or returns an unexpected error,
+// fetch is called directly and the result is not cached. If a cached entry
+// fails to unmarshal (corrupt or schema-changed), the entry is overwritten
+// with a fresh fetch. Errors from Set are ignored so a Redis hiccup never
+// fails a request.
+func getOrSet[T any](ctx context.Context, c *Client, key string, ttl time.Duration, fetch func() (T, error)) (T, error) {
+	raw, err := c.rdb.Get(ctx, key).Bytes()
+	if err != nil && !errors.Is(err, redis.Nil) {
+		// real Redis error, degrade gracefully
+		return fetch()
+	}
+	var zero, out T
+	if errors.Is(err, redis.Nil) {
+		// cache miss — fetch, store, return
+		val, err := fetch()
+		if err != nil {
+			return zero, err
+		}
+		data, jsonErr := json.Marshal(val)
+		if jsonErr != nil {
+			return val, nil
+		}
+		_ = c.rdb.Set(ctx, key, data, ttl)
+		return val, nil
+	}
+	if err = json.Unmarshal(raw, &out); err != nil {
+		// corrupt cache entry — overwrite it
+		val, err := fetch()
+		if err != nil {
+			return zero, err
+		}
+		data, jsonErr := json.Marshal(val)
+		if jsonErr != nil {
+			return val, nil
+		}
+		_ = c.rdb.Set(ctx, key, data, ttl)
+		return val, nil
+	}
+	return out, nil
+}

internal/cache/reader.go

@@ -0,0 +1,342 @@
+// Copyright 2026 Beacon Contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package cache
+
+import (
+	"context"
+	"fmt"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/MeshCore-Beacon/beacon-server/internal/api"
+	"github.com/google/uuid"
+)
+
+const (
+	keyIATAs                = "beacon:iatas"
+	keyIATAPrefix           = "beacon:iata:"
+	keyRegions              = "beacon:regions"
+	keyRegionPrefix         = "beacon:region:"
+	keyRegionSlugPrefix     = "beacon:region:slug:"
+	keyScopeNames           = "beacon:scope:names"
+	keyScopeStats           = "beacon:scope:stats"
+	keyScopesByIATAsPrefix  = "beacon:scopes:iatas:"
+	keyScopeByNamePrefix    = "beacon:scope:name:"
+	keyStatsOverviewPrefix  = "beacon:stats:overview:"
+	keyStatsObsPrefix       = "beacon:stats:observations:"
+	keyStatsBreakdownPrefix = "beacon:stats:breakdown:"
+	keyStatsTopNodesPrefix  = "beacon:stats:top-nodes:"
+	keyStatsTopObsPrefix    = "beacon:stats:top-observers:"
+	keyRadioPresetsPrefix   = "beacon:radio-presets:"
+	keyNodePrefix           = "beacon:node:"
+	keyNodeNeighborsPrefix  = "beacon:node:neighbors:"
+	keyNodesByIDsPrefix     = "beacon:nodes:ids:"
+	keyObserverPrefix       = "beacon:observer:"
+	keyObserverScopesPrefix = "beacon:observer:scopes:"
+)
+
+// CachedReader wraps an api.Reader with a Redis caching layer.
+// It implements api.Reader and is a drop-in replacement for db.Store
+// at the wiring point in main.go.
+type CachedReader struct {
+	inner api.Reader
+	c     *Client
+	ttl   CacheTTLs
+}
+
+// CacheTTLs holds the resolved per-category TTLs for the cache layer.
+// All fields should be non-zero — use ResolveTTLs to build this from
+// config with fallback to the global TTL and then the default.
+type CacheTTLs struct {
+	Stats     time.Duration
+	Reference time.Duration
+	Nodes     time.Duration
+	Observers time.Duration
+}
+
+// NewCachedReader returns an api.Reader that transparently caches responses
+// using the provided Redis client and TTL configuration. inner is the
+// underlying db.Store that is called on cache misses.
+func NewCachedReader(inner api.Reader, c *Client, ttl CacheTTLs) api.Reader {
+	return &CachedReader{
+		inner: inner,
+		c:     c,
+		ttl:   ttl,
+	}
+}
+
+// InvalidateNode removes the cached entries for a node by UUID.
+// Should be called from the ingest path after a node upsert.
+func (cr *CachedReader) InvalidateNode(ctx context.Context, nodeID uuid.UUID) {
+	id := nodeID.String()
+	cr.c.del(ctx, keyNodePrefix+id, keyNodeNeighborsPrefix+id)
+}
+
+// InvalidateObserver removes the cached entries for an observer by UUID.
+// Should be called from the ingest path after an observer upsert.
+func (cr *CachedReader) InvalidateObserver(ctx context.Context, observerID uuid.UUID) {
+	id := observerID.String()
+	cr.c.del(ctx, keyObserverPrefix+id, keyObserverScopesPrefix+id)
+}
+
+// ListIATAs implements [api.Reader].
+func (cr *CachedReader) ListIATAs(ctx context.Context) ([]api.IATA, error) {
+	return getOrSet(ctx, cr.c, keyIATAs, cr.ttl.Reference, func() ([]api.IATA, error) {
+		return cr.inner.ListIATAs(ctx)
+	})
+}
+
+// GetIATA implements [api.Reader].
+func (cr *CachedReader) GetIATA(ctx context.Context, iata string) (*api.IATA, error) {
+	return getOrSet(ctx, cr.c, keyIATAPrefix+iata, cr.ttl.Reference, func() (*api.IATA, error) {
+		return cr.inner.GetIATA(ctx, iata)
+	})
+}
+
+// ListRegions implements [api.Reader].
+func (cr *CachedReader) ListRegions(ctx context.Context) ([]api.RegionSummary, error) {
+	return getOrSet(ctx, cr.c, keyRegions, cr.ttl.Reference, func() ([]api.RegionSummary, error) {
+		return cr.inner.ListRegions(ctx)
+	})
+}
+
+// GetRegion implements [api.Reader].
+func (cr *CachedReader) GetRegion(ctx context.Context, regionID int32) (*api.Region, error) {
+	return getOrSet(ctx, cr.c, fmt.Sprintf("%s%d", keyRegionPrefix, regionID), cr.ttl.Reference, func() (*api.Region, error) {
+		return cr.inner.GetRegion(ctx, regionID)
+	})
+}
+
+// GetRegionBySlug implements [api.Reader].
+func (cr *CachedReader) GetRegionBySlug(ctx context.Context, slug string) (*api.Region, error) {
+	return getOrSet(ctx, cr.c, keyRegionSlugPrefix+slug, cr.ttl.Reference, func() (*api.Region, error) {
+		return cr.inner.GetRegionBySlug(ctx, slug)
+	})
+}
+
+// GetScopeNames implements [api.Reader].
+func (cr *CachedReader) GetScopeNames(ctx context.Context) ([]string, error) {
+	return getOrSet(ctx, cr.c, keyScopeNames, cr.ttl.Reference, func() ([]string, error) {
+		return cr.inner.GetScopeNames(ctx)
+	})
+}
+
+// GetScopeStats implements [api.Reader].
+func (cr *CachedReader) GetScopeStats(ctx context.Context) ([]api.ScopeStats, error) {
+	return getOrSet(ctx, cr.c, keyScopeStats, cr.ttl.Reference, func() ([]api.ScopeStats, error) {
+		return cr.inner.GetScopeStats(ctx)
+	})
+}
+
+// GetScopesByIATAs implements [api.Reader].
+func (cr *CachedReader) GetScopesByIATAs(ctx context.Context, iatas []string) ([]api.ScopeSummary, error) {
+	sorted := make([]string, len(iatas))
+	copy(sorted, iatas)
+	sort.Strings(sorted)
+	key := keyScopesByIATAsPrefix + strings.Join(sorted, ",")
+	return getOrSet(ctx, cr.c, key, cr.ttl.Reference, func() ([]api.ScopeSummary, error) {
+		return cr.inner.GetScopesByIATAs(ctx, iatas)
+	})
+}
+
+// GetScopeByName implements [api.Reader].
+func (cr *CachedReader) GetScopeByName(ctx context.Context, name string) (*api.ScopeDetail, error) {
+	return getOrSet(ctx, cr.c, keyScopeByNamePrefix+name, cr.ttl.Reference, func() (*api.ScopeDetail, error) {
+		return cr.inner.GetScopeByName(ctx, name)
+	})
+}
+
+// GetStatsOverview implements [api.Reader].
+func (cr *CachedReader) GetStatsOverview(ctx context.Context, iata string) (*api.StatsOverview, error) {
+	return getOrSet(ctx, cr.c, keyStatsOverviewPrefix+iata, cr.ttl.Stats, func() (*api.StatsOverview, error) {
+		return cr.inner.GetStatsOverview(ctx, iata)
+	})
+}
+
+// GetStatsObservations implements [api.Reader].
+func (cr *CachedReader) GetStatsObservations(ctx context.Context, iata string, since time.Time) ([]api.ObservationPoint, error) {
+	key := fmt.Sprintf("%s%s:%d", keyStatsObsPrefix, iata, since.UnixMilli())
+	return getOrSet(ctx, cr.c, key, cr.ttl.Stats, func() ([]api.ObservationPoint, error) {
+		return cr.inner.GetStatsObservations(ctx, iata, since)
+	})
+}
+
+// GetStatsPayloadBreakdown implements [api.Reader].
+func (cr *CachedReader) GetStatsPayloadBreakdown(ctx context.Context, iata string, since time.Time) ([]api.PayloadBreakdownItem, error) {
+	key := fmt.Sprintf("%s%s:%d", keyStatsBreakdownPrefix, iata, since.UnixMilli())
+	return getOrSet(ctx, cr.c, key, cr.ttl.Stats, func() ([]api.PayloadBreakdownItem, error) {
+		return cr.inner.GetStatsPayloadBreakdown(ctx, iata, since)
+	})
+}
+
+// GetStatsTopNodes implements [api.Reader].
+func (cr *CachedReader) GetStatsTopNodes(ctx context.Context, iata string, limit int32) ([]api.TopNode, error) {
+	key := fmt.Sprintf("%s%s:%d", keyStatsTopNodesPrefix, iata, limit)
+	return getOrSet(ctx, cr.c, key, cr.ttl.Stats, func() ([]api.TopNode, error) {
+		return cr.inner.GetStatsTopNodes(ctx, iata, limit)
+	})
+}
+
+// GetStatsTopObservers implements [api.Reader].
+func (cr *CachedReader) GetStatsTopObservers(ctx context.Context, iata string, since time.Time, limit int32) ([]api.TopObserver, error) {
+	key := fmt.Sprintf("%s%s:%d:%d", keyStatsTopObsPrefix, iata, since.UnixMilli(), limit)
+	return getOrSet(ctx, cr.c, key, cr.ttl.Stats, func() ([]api.TopObserver, error) {
+		return cr.inner.GetStatsTopObservers(ctx, iata, since, limit)
+	})
+}
+
+// GetRadioPresets implements [api.Reader].
+func (cr *CachedReader) GetRadioPresets(ctx context.Context, preset string, iata string) ([]api.RadioPreset, error) {
+	key := keyRadioPresetsPrefix + preset + ":" + iata
+	return getOrSet(ctx, cr.c, key, cr.ttl.Stats, func() ([]api.RadioPreset, error) {
+		return cr.inner.GetRadioPresets(ctx, preset, iata)
+	})
+}
+
+// GetNode implements [api.Reader].
+func (cr *CachedReader) GetNode(ctx context.Context, nodeID uuid.UUID) (*api.Node, error) {
+	return getOrSet(ctx, cr.c, keyNodePrefix+nodeID.String(), cr.ttl.Nodes, func() (*api.Node, error) {
+		return cr.inner.GetNode(ctx, nodeID)
+	})
+}
+
+// GetNodeNeighbors implements [api.Reader].
+func (cr *CachedReader) GetNodeNeighbors(ctx context.Context, nodeID uuid.UUID) ([]api.NodeNeighbor, error) {
+	return getOrSet(ctx, cr.c, keyNodeNeighborsPrefix+nodeID.String(), cr.ttl.Nodes, func() ([]api.NodeNeighbor, error) {
+		return cr.inner.GetNodeNeighbors(ctx, nodeID)
+	})
+}
+
+// GetNodesByIDs implements [api.Reader].
+func (cr *CachedReader) GetNodesByIDs(ctx context.Context, ids []uuid.UUID) (map[uuid.UUID]*api.ResolvedNode, error) {
+	strs := make([]string, len(ids))
+	for i, id := range ids {
+		strs[i] = id.String()
+	}
+	sort.Strings(strs)
+	key := keyNodesByIDsPrefix + strings.Join(strs, ",")
+	return getOrSet(ctx, cr.c, key, cr.ttl.Nodes, func() (map[uuid.UUID]*api.ResolvedNode, error) {
+		return cr.inner.GetNodesByIDs(ctx, ids)
+	})
+}
+
+// GetObserver implements [api.Reader].
+func (cr *CachedReader) GetObserver(ctx context.Context, observerID uuid.UUID) (*api.Observer, error) {
+	return getOrSet(ctx, cr.c, keyObserverPrefix+observerID.String(), cr.ttl.Observers, func() (*api.Observer, error) {
+		return cr.inner.GetObserver(ctx, observerID)
+	})
+}
+
+// GetObserverScopes implements [api.Reader].
+func (cr *CachedReader) GetObserverScopes(ctx context.Context, observerID uuid.UUID) ([]string, error) {
+	return getOrSet(ctx, cr.c, keyObserverScopesPrefix+observerID.String(), cr.ttl.Observers, func() ([]string, error) {
+		return cr.inner.GetObserverScopes(ctx, observerID)
+	})
+}
+
+// GetObserverTelemetry implements [api.Reader].
+func (cr *CachedReader) GetObserverTelemetry(ctx context.Context, observerID uuid.UUID, since, until time.Time, afterID int64) (*api.ObserverTelemetry, error) {
+	return cr.inner.GetObserverTelemetry(ctx, observerID, since, until, afterID)
+}
+
+// GetObserverTelemetryBucketed implements [api.Reader].
+func (cr *CachedReader) GetObserverTelemetryBucketed(ctx context.Context, observerID uuid.UUID, since, until time.Time, bucketHours int32) ([]api.ObserverTelemetryPoint, error) {
+	return cr.inner.GetObserverTelemetryBucketed(ctx, observerID, since, until, bucketHours)
+}
+
+// GetPacket implements [api.Reader].
+func (cr *CachedReader) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet, error) {
+	return cr.inner.GetPacket(ctx, packetHash)
+}
+
+// GetChannel implements [api.Reader].
+func (cr *CachedReader) GetChannel(ctx context.Context, channelID int32) (*api.Channel, error) {
+	return cr.inner.GetChannel(ctx, channelID)
+}
+
+// GetTraceByTag implements [api.Reader].
+func (cr *CachedReader) GetTraceByTag(ctx context.Context, tag string) (*api.TraceDetail, error) {
+	return cr.inner.GetTraceByTag(ctx, tag)
+}
+
+// GetKnownRoutesByNode implements [api.Reader].
+func (cr *CachedReader) GetKnownRoutesByNode(ctx context.Context, iata string, nodeID uuid.UUID) ([]api.KnownRoute, error) {
+	return cr.inner.GetKnownRoutesByNode(ctx, iata, nodeID)
+}
+
+// GetCrossIATANeighbors implements [api.Reader].
+func (cr *CachedReader) GetCrossIATANeighbors(ctx context.Context, nodeID uuid.UUID, iata string) ([]api.NodeNeighbor, error) {
+	return cr.inner.GetCrossIATANeighbors(ctx, nodeID, iata)
+}
+
+// ListChannels implements [api.Reader].
+func (cr *CachedReader) ListChannels(ctx context.Context, limit int32, hash []byte, iata string, cursor int64) (api.Page[api.ChannelSummary], error) {
+	return cr.inner.ListChannels(ctx, limit, hash, iata, cursor)
+}
+
+// ListChannelMessages implements [api.Reader].
+func (cr *CachedReader) ListChannelMessages(ctx context.Context, channelID *int32, since time.Time, limit int32, iatas []string, scope string, cursor int64) (api.Page[api.ChannelMessage], error) {
+	return cr.inner.ListChannelMessages(ctx, channelID, since, limit, iatas, scope, cursor)
+}
+
+// ListChannelMessagesByHash implements [api.Reader].
+func (cr *CachedReader) ListChannelMessagesByHash(ctx context.Context, hash []byte, since time.Time, limit int32, iatas []string, scope string, cursor int64) (api.Page[api.ChannelMessage], error) {
+	return cr.inner.ListChannelMessagesByHash(ctx, hash, since, limit, iatas, scope, cursor)
+}
+
+// ListMessagesAfterID implements [api.Reader].
+func (cr *CachedReader) ListMessagesAfterID(ctx context.Context, afterID int64, iatas []string, scope string, limit int32) ([]api.ChannelMessage, error) {
+	return cr.inner.ListMessagesAfterID(ctx, afterID, iatas, scope, limit)
+}
+
+// ListNodes implements [api.Reader].
+func (cr *CachedReader) ListNodes(ctx context.Context, nodeType int16, iatas []string, supportsMultibytePaths, supportsMultibyteTraces *bool, pubkey []byte, name, scope string, cursor int64, limit int32) (api.Page[api.NodeSummary], error) {
+	return cr.inner.ListNodes(ctx, nodeType, iatas, supportsMultibytePaths, supportsMultibyteTraces, pubkey, name, scope, cursor, limit)
+}
+
+// ListNodeObservations implements [api.Reader].
+func (cr *CachedReader) ListNodeObservations(ctx context.Context, nodeID uuid.UUID, cursor int64, limit int32) (api.Page[api.PacketObservationSummary], error) {
+	return cr.inner.ListNodeObservations(ctx, nodeID, cursor, limit)
+}
+
+// ListObservers implements [api.Reader].
+func (cr *CachedReader) ListObservers(ctx context.Context, iatas []string, observerType, broker, status, name, scope string, cursor int64, limit int32) (api.Page[api.ObserverSummary], error) {
+	return cr.inner.ListObservers(ctx, iatas, observerType, broker, status, name, scope, cursor, limit)
+}
+
+// ListObserverAdverts implements [api.Reader].
+func (cr *CachedReader) ListObserverAdverts(ctx context.Context, observerID uuid.UUID, cursor int64, limit int32) (api.Page[api.AdvertObservation], error) {
+	return cr.inner.ListObserverAdverts(ctx, observerID, cursor, limit)
+}
+
+// ListPackets implements [api.Reader].
+func (cr *CachedReader) ListPackets(ctx context.Context, payloadType, routeType int16, iatas []string, scope string, since, until time.Time, cursor int64, limit int32) (api.Page[api.PacketSummary], error) {
+	return cr.inner.ListPackets(ctx, payloadType, routeType, iatas, scope, since, until, cursor, limit)
+}
+
+// ListPacketsAfterID implements [api.Reader].
+func (cr *CachedReader) ListPacketsAfterID(ctx context.Context, afterObservationID int64, payloadType, routeType int16, iatas []string, scope string, limit int32) ([]api.PacketSummary, error) {
+	return cr.inner.ListPacketsAfterID(ctx, afterObservationID, payloadType, routeType, iatas, scope, limit)
+}
+
+// ListKnownRoutes implements [api.Reader].
+func (cr *CachedReader) ListKnownRoutes(ctx context.Context, iata string, hopCount int32, cursor time.Time, limit int32) ([]api.KnownRoute, error) {
+	return cr.inner.ListKnownRoutes(ctx, iata, hopCount, cursor, limit)
+}
+
+// SearchKnownRoutes implements [api.Reader].
+func (cr *CachedReader) SearchKnownRoutes(ctx context.Context, iata, fromHash, toHash string) ([]api.KnownRoute, error) {
+	return cr.inner.SearchKnownRoutes(ctx, iata, fromHash, toHash)
+}
+
+// SearchCrossIATARoutes implements [api.Reader].
+func (cr *CachedReader) SearchCrossIATARoutes(ctx context.Context, fromHash, fromIATA, toHash, toIATA string) ([]api.CrossIATARoute, error) {
+	return cr.inner.SearchCrossIATARoutes(ctx, fromHash, fromIATA, toHash, toIATA)
+}
+
+// ListTraceTags implements [api.Reader].
+func (cr *CachedReader) ListTraceTags(ctx context.Context, iatas []string, scope string, since, until time.Time, cursor time.Time, limit int32) ([]api.TraceTagSummary, error) {
+	return cr.inner.ListTraceTags(ctx, iatas, scope, since, until, cursor, limit)
+}

internal/config/config.go

@@ -22,6 +22,42 @@ type Config struct {
 	Packets     PacketsConfig         `yaml:"packets"`
 	Ingest      IngestFilterConfig    `yaml:"ingest"`
 	Scopes      []ScopeConfig         `yaml:"scopes"`
+	Cache       CacheConfig           `yaml:"cache"`
+}
+
+// CacheConfig controls Redis caching behaviour.
+// If Addr is not set (via REDIS_ADDR env var), caching is disabled entirely
+// and Beacon falls back to querying PostgreSQL directly for all reads.
+type CacheConfig struct {
+	// TTL is the default cache entry lifetime applied to all categories
+	// that do not have an explicit override in TTLs. Defaults to 1h if not set.
+	TTL duration `yaml:"ttl"`
+
+	// TTLs holds optional per-category TTL overrides. Any category left
+	// at zero will fall back to TTL.
+	TTLs CacheTTLsConfig `yaml:"ttls"`
+}
+
+// CacheTTLsConfig holds per-category TTL overrides for the cache layer.
+// Each field is optional — omit it in config to inherit the global TTL.
+type CacheTTLsConfig struct {
+	// Stats controls the TTL for aggregated network statistics endpoints
+	// (overview, observations, payload breakdown, top nodes/observers, radio presets, scope stats).
+	// These are backed by materialized views refreshed hourly, so values under 1m are rarely useful.
+	Stats duration `yaml:"stats"`
+
+	// Reference controls the TTL for mostly-static reference data
+	// (IATAs, regions, scopes). These change only when new observers
+	// arrive or config is reseeded.
+	Reference duration `yaml:"reference"`
+
+	// Nodes controls the TTL for individual node detail responses.
+	// Acts as a safety-net expiry alongside explicit invalidation on upsert.
+	Nodes duration `yaml:"nodes"`
+
+	// Observers controls the TTL for individual observer detail responses.
+	// Acts as a safety-net expiry alongside explicit invalidation on upsert.
+	Observers duration `yaml:"observers"`
 }
 
 // ScopeConfig defines a regional transport scope.

internal/ingest/ingest.go

@@ -174,12 +174,14 @@ type ScopeStore interface {
 
 // Worker holds the dependencies for one broker's ingest loop.
 type Worker struct {
-	cfg    Config
-	db     DB
-	hub    *hub.Hub
-	keys   ChannelKeyStore
-	scopes ScopeStore
-	client mqtt.Client
+	cfg              Config
+	db               DB
+	hub              *hub.Hub
+	keys             ChannelKeyStore
+	scopes           ScopeStore
+	client           mqtt.Client
+	onNodeUpsert     func(ctx context.Context, nodeID uuid.UUID)
+	onObserverUpsert func(ctx context.Context, observerID uuid.UUID)
 }
 
 // New creates an ingest Worker. Call Start() to connect and begin processing.
@@ -236,6 +238,11 @@ func (w *Worker) IsConnected() bool {
 	return w.client.IsConnected()
 }
 
+func (w *Worker) SetCacheInvalidators(onNode, onObserver func(ctx context.Context, id uuid.UUID)) {
+	w.onNodeUpsert = onNode
+	w.onObserverUpsert = onObserver
+}
+
 // subscribe registers the wildcard topic handler after (re)connect.
 func (w *Worker) subscribe(client mqtt.Client) {
 	// meshcore/{IATA}/{pubkey}/packets

internal/ingest/side_effects.go

@@ -95,6 +95,10 @@ func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshc
 			log.Printf("ingest[%s]: db: upsert node failed: %v", w.cfg.BrokerName, err)
 			return
 		}
+		// invalidate cache for this node
+		if w.onNodeUpsert != nil {
+			w.onNodeUpsert(ctx, nodeID)
+		}
 		// if the advert was forwarded, the first hop is a neighbor
 		if packet.PathHashCount() > 0 && (advert.Type() == meshcore.AdvertTypeRepeater || advert.Type() == meshcore.AdvertTypeRoom) {
 			firstHop := packet.PathHashes()

internal/ingest/status.go

@@ -84,6 +84,10 @@ func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte)
 		log.Printf("ingest[%s]: db: upsert observer failed in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err)
 		return
 	}
+	// invalidate cache for observer details
+	if w.onObserverUpsert != nil {
+		w.onObserverUpsert(ctx, id)
+	}
 	if err := w.db.UpsertObserverBroker(ctx, id, w.cfg.BrokerName); err != nil {
 		log.Printf("ingest[%s]: db: upsert observer broker failed in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err)
 	}