diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 88b48e8..e55dd06 100644 --- a/CONTRIBUTING.md +++ b/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 diff --git a/README.md b/README.md index df2a0a9..928e3f3 100644 --- a/README.md +++ b/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. diff --git a/cmd/beacon/main.go b/cmd/beacon/main.go index 2c7caf4..6cd51b1 100644 --- a/cmd/beacon/main.go +++ b/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, diff --git a/config.yaml.example b/config.yaml.example index cb6acc0..45618e0 100644 --- a/config.yaml.example +++ b/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). diff --git a/env.example b/env.example index 4fb7c92..96bf825 100644 --- a/env.example +++ b/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= diff --git a/go.mod b/go.mod index 5fbfc0c..565d234 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 37c91b6..c6f9972 100644 --- a/go.sum +++ b/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= diff --git a/internal/cache/cache.go b/internal/cache/cache.go new file mode 100644 index 0000000..abee276 --- /dev/null +++ b/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 +} diff --git a/internal/cache/reader.go b/internal/cache/reader.go new file mode 100644 index 0000000..587e5f8 --- /dev/null +++ b/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) +} diff --git a/internal/config/config.go b/internal/config/config.go index b3122f6..0326fd8 100644 --- a/internal/config/config.go +++ b/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. diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index 8855f1f..1d28a61 100644 --- a/internal/ingest/ingest.go +++ b/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 diff --git a/internal/ingest/side_effects.go b/internal/ingest/side_effects.go index 4115efa..1572012 100644 --- a/internal/ingest/side_effects.go +++ b/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() diff --git a/internal/ingest/status.go b/internal/ingest/status.go index 03baca4..fc32628 100644 --- a/internal/ingest/status.go +++ b/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) }