From 72972155557b3c677543043b456f4af814e6722a Mon Sep 17 00:00:00 2001 From: MrAlders0n <55921894+MrAlders0n@users.noreply.github.com> Date: Sun, 11 Oct 2026 14:18:26 -0400 Subject: [PATCH 1/2] feat(meshmapper): import repeater site details for node detail (#221) --- README.md | 5 +- cmd/beacon/main.go | 16 +- config.yaml.example | 16 + db/meshmapper.go | 62 +++ db/meshmapper_repeaters_integration_test.go | 147 +++++++ db/migrations/003_meshmapper_repeaters.sql | 22 + db/nodes.go | 8 + db/queries/queries.sql | 48 ++- db/sqlc/mock/querier.go | 72 ++++ db/sqlc/models.go | 18 + db/sqlc/querier.go | 10 + db/sqlc/queries.sql.go | 129 +++++- docs/docs.go | 122 ++++++ docs/swagger.json | 122 ++++++ docs/swagger.yaml | 85 ++++ internal/api/handlers/nodes_test.go | 25 ++ internal/api/nodes.go | 45 ++ internal/cache/cache_test.go | 12 + internal/cache/reader.go | 5 + internal/config/meshmapper.go | 32 +- internal/config/meshmapper_test.go | 18 + internal/meshmapper/directory.go | 5 + internal/meshmapper/repeaters.go | 455 ++++++++++++++++++++ internal/meshmapper/repeaters_test.go | 440 +++++++++++++++++++ 24 files changed, 1906 insertions(+), 13 deletions(-) create mode 100644 db/meshmapper_repeaters_integration_test.go create mode 100644 db/migrations/003_meshmapper_repeaters.sql create mode 100644 internal/meshmapper/repeaters.go create mode 100644 internal/meshmapper/repeaters_test.go diff --git a/README.md b/README.md index efd9530..f63db3b 100644 --- a/README.md +++ b/README.md @@ -80,8 +80,7 @@ remove rate limits or change the `get_zones.php?country=...` request shape. IP exemptions are not authentication. Beacon sends `X-API-Key` to `get_zones.php`, `get_geojson.php`, `get_scopes.php`, -and `get_channels.php`. The shared client also supports that header for -`get_repeaters.php`; this repository does not currently fetch repeaters. +`get_channels.php`, and `get_repeaters.php`. Requests are restricted to those HTTPS MeshMapper endpoints and never follow redirects. Coverage consumers keep their separate keys, scopes, quotas, and supported `?key=` authentication; this client does not call `coverage.php`. @@ -90,7 +89,7 @@ Missing keys are reported as unconfigured without stopping Beacon or deleting cached imports. Failed refreshes record `last_error`, retain the last successful data and freshness timestamp, and report 401 as an authentication failure or 403 as a permission failure. Inspect the `meshmapper.zones`, `meshmapper.scopes`, -and `meshmapper.channels` logs for freshness and the next attempt. No anonymous +`meshmapper.channels`, and `meshmapper.repeaters` logs for freshness and the next attempt. No anonymous fallback occurs. Existing refresh limits still apply, including longer `Retry-After` delays on 429/503. diff --git a/cmd/beacon/main.go b/cmd/beacon/main.go index f9aee8a..cad97ad 100644 --- a/cmd/beacon/main.go +++ b/cmd/beacon/main.go @@ -200,7 +200,7 @@ func main() { scopes.SetManualMembers(cfg.ManualScopeMembers()) slog.Info(fmt.Sprintf("loaded %d transport scopes", len(scopeEntries)), "component", "startup") directory := meshmapper.NewDirectory(store, cfg.MeshMapper.APIKey) - if cfg.MeshMapper.Scopes.Enabled || cfg.MeshMapper.Zones.Enabled || cfg.MeshMapper.Channels.Enabled { + if mm := cfg.MeshMapper; mm.Scopes.Enabled || mm.Zones.Enabled || mm.Channels.Enabled || mm.Repeaters.Enabled { if cfg.MeshMapper.APIKey == "" { slog.Warn("MeshMapper integration unconfigured: set MESHMAPPER_API_KEY or meshmapper.api_key; retaining cached data", "component", "meshmapper") } @@ -306,6 +306,17 @@ func main() { slog.Error("failed to restore MeshMapper channels", "component", "startup", "error", err) os.Exit(1) } + var onRepeatersChange func(context.Context) + if cr, ok := reader.(*cache.CachedReader); ok { + onRepeatersChange = cr.InvalidateNodeDetails + } + restoreCtx, cancelRestore = context.WithTimeout(ctx, 10*time.Second) + repeaterImporter, err := meshmapper.NewRepeaters(restoreCtx, cfg.MeshMapper.Repeaters, store, directory, onRepeatersChange) + cancelRestore() + if err != nil { + slog.Error("failed to restore MeshMapper repeaters", "component", "startup", "error", err) + os.Exit(1) + } channelImporter.OnNewKeys(func(hashes [][]byte) { // Off the refresh tick: a backfill can outlast it, and history isn't urgent. go func() { @@ -416,6 +427,9 @@ func main() { if cfg.MeshMapper.Channels.Enabled { tasks = append(tasks, background.Task{Name: "meshmapper.channels", Interval: meshmapper.PollInterval, Run: channelImporter.Refresh}) } + if cfg.MeshMapper.Repeaters.Enabled { + tasks = append(tasks, background.Task{Name: "meshmapper.repeaters", Interval: meshmapper.PollInterval, Run: repeaterImporter.Refresh}) + } profiles := configureProfiling(ctx, pool) defer profiles.Stop() for i := range tasks { diff --git a/config.yaml.example b/config.yaml.example index 1f1add3..cc19991 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -112,6 +112,22 @@ ratelimit: # channels: # enabled: false # refresh_interval: 24h # 24h-168h (one get_channels.php per region per 23.5h) +# +# Repeater site details (get_repeaters.php). Every known IATA that MeshMapper +# lists is fetched from its site, the same way as channels; a MeshMapper group is +# fetched once, since any member's site lists the whole group. Each repeater's +# admin-entered site details (power, hardware, antenna, heights, power source, +# notes), CARpeater flag, admin names, confirmed scopes and default scope are +# shown on node detail as `meshmapper`, matched by full public key. Positions, +# status and radio data are not imported. A failed refresh keeps the last good +# list and retries after 1h (or Retry-After); a 429 pauses every region, since +# the key's budget (1,000/day, 30/min) is shared. An empty list clears the region. +# 6h costs 4 requests per group or ungrouped IATA per day. Each import refreshes +# cached node detail. Turning it off deletes the import on restart but keeps the +# backoff. Logged under component=meshmapper.repeaters. +# repeaters: +# enabled: false +# refresh_interval: 6h # 1h-24h iatas: YVR: diff --git a/db/meshmapper.go b/db/meshmapper.go index fa2120d..b2ad31c 100644 --- a/db/meshmapper.go +++ b/db/meshmapper.go @@ -5,9 +5,12 @@ package db import ( "context" + "encoding/hex" + "encoding/json" "errors" sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/MeshCore-Beacon/beacon-server/internal/api" "github.com/MeshCore-Beacon/beacon-server/internal/meshmapper" "github.com/MeshCore-Beacon/beacon-server/internal/scopestore" "github.com/jackc/pgx/v5" @@ -254,3 +257,62 @@ func (s *Store) ListIATADetails(ctx context.Context) ([]meshmapper.IATADetails, } return out, nil } + +func (s *Store) ListRepeaterSources(ctx context.Context) ([]meshmapper.Catalogue, error) { + rows, err := s.q.ListRepeaterSources(ctx) + if err != nil { + return nil, err + } + out := make([]meshmapper.Catalogue, 0, len(rows)) + for _, row := range rows { + c := meshmapper.Catalogue{IATA: row.Source, URL: row.Url, Cache: meshmapper.Cache{CheckedAt: row.CheckedAt.Time, + AttemptedAt: row.AttemptedAt.Time, NextAttempt: row.NextAttempt.Time, LastError: row.LastError}} + if row.Etag != nil { + c.ETag = *row.Etag + } + out = append(out, c) + } + return out, nil +} + +// SaveRepeaterSource records a fetch; non-nil repeaters replace the source's list first, +// so a failure never records a check without its list. +func (s *Store) SaveRepeaterSource(ctx context.Context, source, url string, cache meshmapper.Cache, repeaters []meshmapper.Repeater) error { + if repeaters != nil { + type row struct { + Key string `json:"key"` + Data api.MeshMapperRepeater `json:"data"` + } + rows := make([]row, 0, len(repeaters)) + for _, r := range repeaters { + rows = append(rows, row{Key: hex.EncodeToString(r.Key), Data: r.Data}) + } + payload, err := json.Marshal(rows) + if err != nil { + return err + } + if err := s.q.ReplaceRepeaters(ctx, sqlc.ReplaceRepeatersParams{Source: source, Rows: payload}); err != nil { + return err + } + } + params := sqlc.SaveRepeaterSourceParams{ + Source: source, Url: url, LastError: cache.LastError, + CheckedAt: pgtype.Timestamptz{Time: cache.CheckedAt, Valid: !cache.CheckedAt.IsZero()}, + AttemptedAt: pgtype.Timestamptz{Time: cache.AttemptedAt, Valid: true}, + NextAttempt: pgtype.Timestamptz{Time: cache.NextAttempt, Valid: true}, + } + if !cache.CheckedAt.IsZero() { + params.Etag = &cache.ETag + } + return s.q.SaveRepeaterSource(ctx, params) +} + +func (s *Store) DeleteRepeaterSource(ctx context.Context, source string) error { + return s.q.DeleteRepeaterSource(ctx, source) +} + +// ClearRepeaters reports whether any imported repeater was removed. +func (s *Store) ClearRepeaters(ctx context.Context) (bool, error) { + n, err := s.q.ClearRepeaters(ctx) + return n > 0, err +} diff --git a/db/meshmapper_repeaters_integration_test.go b/db/meshmapper_repeaters_integration_test.go new file mode 100644 index 0000000..a905bcc --- /dev/null +++ b/db/meshmapper_repeaters_integration_test.go @@ -0,0 +1,147 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package db + +import ( + "bytes" + "context" + "os" + "strings" + "testing" + "time" + + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/MeshCore-Beacon/beacon-server/internal/meshmapper" + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +func TestMeshMapperRepeatersPostgres(t *testing.T) { + dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN") + if dsn == "" { + t.Skip("set BEACON_TEST_POSTGRES_DSN") + } + ctx, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + pool, err := pgxpool.New(ctx, dsn) + if err != nil { + t.Fatal(err) + } + if err := RunMigrations(ctx, pool); err != nil { + pool.Close() + t.Fatal(err) + } + pool.Close() + conn, err := pgx.Connect(ctx, dsn) + if err != nil { + t.Fatal(err) + } + defer conn.Close(context.Background()) + tx, err := conn.Begin(ctx) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback(context.Background()) + for _, table := range []string{"nodes", "transport_scopes", "observers", "node_iatas", "node_neighbors", + "meshmapper_repeater_sources", "meshmapper_repeaters"} { + if _, err := tx.Exec(ctx, "CREATE TEMP TABLE "+table+" (LIKE public."+table+" INCLUDING ALL) ON COMMIT DROP"); err != nil { + t.Fatal(err) + } + } + store := &Store{q: sqlc.New(tx)} + keyA, keyB := bytes.Repeat([]byte{0xa1}, 32), bytes.Repeat([]byte{0xb2}, 32) + nodeID := uuid.New() + if _, err := tx.Exec(ctx, "INSERT INTO nodes (id, public_key, node_type) VALUES ($1, $2, 2)", nodeID, keyA); err != nil { + t.Fatal(err) + } + hw := "RAK4631" + now := time.Now().UTC().Truncate(time.Second) + save := func(iata string, at time.Time, list []meshmapper.Repeater) { + t.Helper() + c := meshmapper.Cache{CheckedAt: at, AttemptedAt: at, NextAttempt: at.Add(time.Hour)} + if err := store.SaveRepeaterSource(ctx, iata, "https://"+iata+".meshmapper.net/get_repeaters.php", c, list); err != nil { + t.Fatal(err) + } + } + rows := func() (n int) { + t.Helper() + if err := tx.QueryRow(ctx, "SELECT count(*) FROM meshmapper_repeaters").Scan(&n); err != nil { + t.Fatal(err) + } + return n + } + + node, err := store.GetNode(ctx, nodeID) + if err != nil || node.MeshMapper != nil { + t.Fatal("unlisted node got MeshMapper data", node.MeshMapper, err) + } + save("YOW", now, []meshmapper.Repeater{ + {Key: keyA, Data: api.MeshMapperRepeater{IATA: "YOW", Site: api.MeshMapperSite{Hardware: &hw}, Admins: []string{"op"}, Scopes: []api.MeshMapperScope{}}}, + {Key: keyB, Data: api.MeshMapperRepeater{Admins: []string{}, Scopes: []api.MeshMapperScope{}}}, + }) + node, err = store.GetNode(ctx, nodeID) + if err != nil || node.MeshMapper == nil || node.MeshMapper.Site.Hardware == nil || *node.MeshMapper.Site.Hardware != hw || + node.MeshMapper.FetchedAt != now.UnixMilli() || node.MeshMapper.IATA != "YOW" { + t.Fatalf("node detail: %+v %v", node.MeshMapper, err) + } + + // A failed check keeps the list; a newer check of the same key from another source wins. + if err := store.SaveRepeaterSource(ctx, "YOW", "https://YOW.meshmapper.net/get_repeaters.php", + meshmapper.Cache{AttemptedAt: now, NextAttempt: now.Add(time.Hour), LastError: "HTTP 503"}, nil); err != nil { + t.Fatal(err) + } + save("ONQC", now.Add(time.Minute), []meshmapper.Repeater{{Key: keyA, Data: api.MeshMapperRepeater{IATA: "YUL", Admins: []string{}, Scopes: []api.MeshMapperScope{}}}}) + if node, err = store.GetNode(ctx, nodeID); err != nil || node.MeshMapper.IATA != "YUL" || rows() != 3 { + t.Fatalf("newest source not preferred: %+v %v rows=%d", node.MeshMapper, err, rows()) + } + var stored string + if err := tx.QueryRow(ctx, "SELECT data::text FROM meshmapper_repeaters WHERE source = 'ONQC'").Scan(&stored); err != nil || strings.Contains(stored, "fetchedAt") { + t.Fatal("fetchedAt stored", stored, err) + } + + save("YOW", now.Add(2*time.Minute), []meshmapper.Repeater{{Key: keyB, Data: api.MeshMapperRepeater{Admins: []string{}, Scopes: []api.MeshMapperScope{}}}}) + var left int + if err := tx.QueryRow(ctx, "SELECT count(*) FROM meshmapper_repeaters WHERE source = 'YOW' AND public_key = $1", keyA).Scan(&left); err != nil || left != 0 { + t.Fatal("dropped repeater kept", left, err) + } + save("YOW", now.Add(3*time.Minute), []meshmapper.Repeater{}) + if err := tx.QueryRow(ctx, "SELECT count(*) FROM meshmapper_repeaters WHERE source = 'YOW'").Scan(&left); err != nil || left != 0 { + t.Fatal("empty list kept repeaters", left, err) + } + + // A new site URL drops the old site's ETag even before the next successful check. + c := meshmapper.Cache{ETag: `"v1"`, CheckedAt: now, AttemptedAt: now, NextAttempt: now} + if err := store.SaveRepeaterSource(ctx, "YYZ", "https://yyz.meshmapper.net/get_repeaters.php", c, []meshmapper.Repeater{}); err != nil { + t.Fatal(err) + } + if err := store.SaveRepeaterSource(ctx, "YYZ", "https://other.meshmapper.net/get_repeaters.php", + meshmapper.Cache{AttemptedAt: now, NextAttempt: now, LastError: "no response"}, nil); err != nil { + t.Fatal(err) + } + sources, err := store.ListRepeaterSources(ctx) + if err != nil || len(sources) != 3 || sources[1].IATA != "YOW" || sources[1].LastError != "" || sources[2].ETag != "" { + t.Fatal("sources", sources, err) + } + + if err := store.DeleteRepeaterSource(ctx, "ONQC"); err != nil { + t.Fatal(err) + } + if node, err = store.GetNode(ctx, nodeID); err != nil || node.MeshMapper != nil { + t.Fatal("retired source still served", node.MeshMapper, err) + } + + save("YOW", now.Add(4*time.Minute), []meshmapper.Repeater{{Key: keyB, Data: api.MeshMapperRepeater{Admins: []string{}, Scopes: []api.MeshMapperScope{}}}}) + if cleared, err := store.ClearRepeaters(ctx); err != nil || !cleared || rows() != 0 { + t.Fatal("clear", cleared, err) + } + sources, _ = store.ListRepeaterSources(ctx) + if len(sources) != 2 || !sources[0].CheckedAt.IsZero() || sources[0].ETag != "" || sources[0].NextAttempt.IsZero() { + t.Fatal("clear should keep timing and drop list state", sources) + } + if cleared, err := store.ClearRepeaters(ctx); err != nil || cleared { + t.Fatal("second clear reported a change", cleared, err) + } +} diff --git a/db/migrations/003_meshmapper_repeaters.sql b/db/migrations/003_meshmapper_repeaters.sql new file mode 100644 index 0000000..d08fb8c --- /dev/null +++ b/db/migrations/003_meshmapper_repeaters.sql @@ -0,0 +1,22 @@ +-- Copyright 2026 Beacon Contributors +-- SPDX-License-Identifier: AGPL-3.0-or-later + +-- One source per MeshMapper group, or per IATA outside a group: any member's site lists the whole group. +CREATE TABLE IF NOT EXISTS meshmapper_repeater_sources ( + source text PRIMARY KEY, + url text NOT NULL, + etag text, + checked_at timestamp with time zone, + attempted_at timestamp with time zone NOT NULL, + next_attempt timestamp with time zone NOT NULL, + last_error text DEFAULT ''::text NOT NULL +); + +CREATE TABLE IF NOT EXISTS meshmapper_repeaters ( + source text NOT NULL, + public_key bytea NOT NULL, + data jsonb NOT NULL, + PRIMARY KEY (source, public_key) +); + +CREATE INDEX IF NOT EXISTS idx_meshmapper_repeaters_public_key ON meshmapper_repeaters (public_key); diff --git a/db/nodes.go b/db/nodes.go index a602cb3..ee248cd 100644 --- a/db/nodes.go +++ b/db/nodes.go @@ -195,6 +195,14 @@ func (s *Store) GetNode(ctx context.Context, nodeID uuid.UUID) (*api.Node, error node.IATAs = []api.NodeIATA{} } } + if len(row.Meshmapper) > 0 { + var mm api.MeshMapperRepeater + if err := json.Unmarshal(row.Meshmapper, &mm); err != nil { + slog.Error("store: failed to unmarshal MeshMapper repeater", "component", "db", "error", err) + } else { + node.MeshMapper = &mm + } + } if row.RadioFreqMhz != nil && row.RadioSf != nil && row.RadioBwKhz != nil { s := fmt.Sprintf("%g,%g,%d", *row.RadioFreqMhz, *row.RadioBwKhz, *row.RadioSf) node.Radio = &s diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 2274e38..1fb4096 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -816,7 +816,12 @@ SELECT n.*, ts.name AS default_scope_name, (SELECT o.id FROM observers o WHERE o.public_key = n.public_key LIMIT 1) AS observer_id, (SELECT json_agg(json_build_object('iata', ni.iata, 'lastHeard', (extract(epoch from ni.last_heard) * 1000)::bigint) ORDER BY ni.last_heard DESC) FROM node_iatas ni WHERE ni.node_id = n.id) AS iatas, - (SELECT COUNT(DISTINCT nn.neighbor_id) FROM node_neighbors nn WHERE nn.node_id = n.id)::bigint AS known_neighbor_count + (SELECT COUNT(DISTINCT nn.neighbor_id) FROM node_neighbors nn WHERE nn.node_id = n.id)::bigint AS known_neighbor_count, + -- The most recently checked source wins when group sites list the same repeater. + (SELECT m.data || jsonb_build_object('fetchedAt', (extract(epoch from s.checked_at) * 1000)::bigint) + FROM meshmapper_repeaters m JOIN meshmapper_repeater_sources s ON s.source = m.source + WHERE m.public_key = n.public_key AND s.checked_at IS NOT NULL + ORDER BY s.checked_at DESC LIMIT 1)::jsonb AS meshmapper FROM nodes n LEFT JOIN transport_scopes ts ON ts.id = n.default_scope_id WHERE n.id = $1; @@ -1838,6 +1843,47 @@ ON CONFLICT (iata, key_fingerprint) DO NOTHING; -- name: DeleteAllChannelMembers :exec DELETE FROM meshmapper_channel_members; +-- name: ListRepeaterSources :many +SELECT * FROM meshmapper_repeater_sources ORDER BY source; + +-- name: SaveRepeaterSource :exec +-- NULL etag/checked_at retain the last good list's state after an error or 304; +-- a new URL drops the old site's ETag. +INSERT INTO meshmapper_repeater_sources (source, url, etag, checked_at, attempted_at, next_attempt, last_error) +VALUES (@source, @url, sqlc.narg(etag)::text, sqlc.narg(checked_at)::timestamptz, @attempted_at, @next_attempt, @last_error) +ON CONFLICT (source) DO UPDATE SET + url = EXCLUDED.url, + etag = CASE WHEN meshmapper_repeater_sources.url = EXCLUDED.url + THEN COALESCE(EXCLUDED.etag, meshmapper_repeater_sources.etag) ELSE EXCLUDED.etag END, + checked_at = COALESCE(EXCLUDED.checked_at, meshmapper_repeater_sources.checked_at), + attempted_at = EXCLUDED.attempted_at, + next_attempt = EXCLUDED.next_attempt, + last_error = EXCLUDED.last_error; + +-- name: ReplaceRepeaters :exec +-- rows is a JSON array of {"key": hex, "data": {...}}; unchanged rows are not rewritten. +-- fetchedAt comes from the source's checked_at when read. +WITH incoming AS ( + SELECT decode(r->>'key', 'hex') AS public_key, (r->'data') - 'fetchedAt' AS data + FROM jsonb_array_elements(@rows::jsonb) r +), removed AS ( + DELETE FROM meshmapper_repeaters m + WHERE m.source = @source AND m.public_key NOT IN (SELECT public_key FROM incoming) +) +INSERT INTO meshmapper_repeaters (source, public_key, data) +SELECT @source, public_key, data FROM incoming +ON CONFLICT (source, public_key) DO UPDATE SET data = EXCLUDED.data +WHERE meshmapper_repeaters.data IS DISTINCT FROM EXCLUDED.data; + +-- name: DeleteRepeaterSource :exec +WITH removed AS (DELETE FROM meshmapper_repeaters r WHERE r.source = @source) +DELETE FROM meshmapper_repeater_sources s WHERE s.source = @source; + +-- name: ClearRepeaters :execrows +-- Keeps fetch timing so re-enabling still respects backoff; the ETag goes with the rows. +WITH reset AS (UPDATE meshmapper_repeater_sources SET etag = NULL, checked_at = NULL) +DELETE FROM meshmapper_repeaters; + -- name: ListZoneBoundaries :many SELECT * FROM meshmapper_zone_boundaries ORDER BY iata; diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index 6d5593c..6ab1f20 100644 --- a/db/sqlc/mock/querier.go +++ b/db/sqlc/mock/querier.go @@ -115,6 +115,21 @@ func (mr *MockQuerierMockRecorder) AmbiguousPrefixes(ctx any) *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AmbiguousPrefixes", reflect.TypeOf((*MockQuerier)(nil).AmbiguousPrefixes), ctx) } +// ClearRepeaters mocks base method. +func (m *MockQuerier) ClearRepeaters(ctx context.Context) (int64, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ClearRepeaters", ctx) + ret0, _ := ret[0].(int64) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// ClearRepeaters indicates an expected call of ClearRepeaters. +func (mr *MockQuerierMockRecorder) ClearRepeaters(ctx any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ClearRepeaters", reflect.TypeOf((*MockQuerier)(nil).ClearRepeaters), ctx) +} + // CreateAccount mocks base method. func (m *MockQuerier) CreateAccount(ctx context.Context, name string) (db.Account, error) { m.ctrl.T.Helper() @@ -330,6 +345,20 @@ func (mr *MockQuerierMockRecorder) DeleteRegionIATAsNotIn(ctx, arg any) *gomock. return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteRegionIATAsNotIn", reflect.TypeOf((*MockQuerier)(nil).DeleteRegionIATAsNotIn), ctx, arg) } +// DeleteRepeaterSource mocks base method. +func (m *MockQuerier) DeleteRepeaterSource(ctx context.Context, source string) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "DeleteRepeaterSource", ctx, source) + ret0, _ := ret[0].(error) + return ret0 +} + +// DeleteRepeaterSource indicates an expected call of DeleteRepeaterSource. +func (mr *MockQuerierMockRecorder) DeleteRepeaterSource(ctx, source any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteRepeaterSource", reflect.TypeOf((*MockQuerier)(nil).DeleteRepeaterSource), ctx, source) +} + // DeleteRollupHour mocks base method. func (m *MockQuerier) DeleteRollupHour(ctx context.Context, hour pgtype.Timestamptz) error { m.ctrl.T.Helper() @@ -1571,6 +1600,21 @@ func (mr *MockQuerierMockRecorder) ListRegions(ctx any) *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListRegions", reflect.TypeOf((*MockQuerier)(nil).ListRegions), ctx) } +// ListRepeaterSources mocks base method. +func (m *MockQuerier) ListRepeaterSources(ctx context.Context) ([]db.MeshmapperRepeaterSource, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ListRepeaterSources", ctx) + ret0, _ := ret[0].([]db.MeshmapperRepeaterSource) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// ListRepeaterSources indicates an expected call of ListRepeaterSources. +func (mr *MockQuerierMockRecorder) ListRepeaterSources(ctx any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListRepeaterSources", reflect.TypeOf((*MockQuerier)(nil).ListRepeaterSources), ctx) +} + // ListRouteEvidence mocks base method. func (m *MockQuerier) ListRouteEvidence(ctx context.Context, arg db.ListRouteEvidenceParams) ([]db.ListRouteEvidenceRow, error) { m.ctrl.T.Helper() @@ -1821,6 +1865,20 @@ func (mr *MockQuerierMockRecorder) RegisterRollupHours(ctx, since any) *gomock.C return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RegisterRollupHours", reflect.TypeOf((*MockQuerier)(nil).RegisterRollupHours), ctx, since) } +// ReplaceRepeaters mocks base method. +func (m *MockQuerier) ReplaceRepeaters(ctx context.Context, arg db.ReplaceRepeatersParams) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ReplaceRepeaters", ctx, arg) + ret0, _ := ret[0].(error) + return ret0 +} + +// ReplaceRepeaters indicates an expected call of ReplaceRepeaters. +func (mr *MockQuerierMockRecorder) ReplaceRepeaters(ctx, arg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ReplaceRepeaters", reflect.TypeOf((*MockQuerier)(nil).ReplaceRepeaters), ctx, arg) +} + // ResolveEndpointHashPairs mocks base method. func (m *MockQuerier) ResolveEndpointHashPairs(ctx context.Context, arg db.ResolveEndpointHashPairsParams) ([]db.ResolveEndpointHashPairsRow, error) { m.ctrl.T.Helper() @@ -2196,6 +2254,20 @@ func (mr *MockQuerierMockRecorder) SaveChannelCatalogue(ctx, arg any) *gomock.Ca return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SaveChannelCatalogue", reflect.TypeOf((*MockQuerier)(nil).SaveChannelCatalogue), ctx, arg) } +// SaveRepeaterSource mocks base method. +func (m *MockQuerier) SaveRepeaterSource(ctx context.Context, arg db.SaveRepeaterSourceParams) error { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "SaveRepeaterSource", ctx, arg) + ret0, _ := ret[0].(error) + return ret0 +} + +// SaveRepeaterSource indicates an expected call of SaveRepeaterSource. +func (mr *MockQuerierMockRecorder) SaveRepeaterSource(ctx, arg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SaveRepeaterSource", reflect.TypeOf((*MockQuerier)(nil).SaveRepeaterSource), ctx, arg) +} + // SaveScopeCatalogue mocks base method. func (m *MockQuerier) SaveScopeCatalogue(ctx context.Context, arg db.SaveScopeCatalogueParams) error { m.ctrl.T.Helper() diff --git a/db/sqlc/models.go b/db/sqlc/models.go index 78e1430..c1c9625 100644 --- a/db/sqlc/models.go +++ b/db/sqlc/models.go @@ -5,6 +5,8 @@ package db import ( + "encoding/json" + "github.com/google/uuid" "github.com/jackc/pgx/v5/pgtype" ) @@ -232,6 +234,22 @@ type MeshmapperChannelMember struct { KeyFingerprint []byte `json:"key_fingerprint"` } +type MeshmapperRepeater struct { + Source string `json:"source"` + PublicKey []byte `json:"public_key"` + Data json.RawMessage `json:"data"` +} + +type MeshmapperRepeaterSource struct { + Source string `json:"source"` + Url string `json:"url"` + Etag *string `json:"etag"` + CheckedAt pgtype.Timestamptz `json:"checked_at"` + AttemptedAt pgtype.Timestamptz `json:"attempted_at"` + NextAttempt pgtype.Timestamptz `json:"next_attempt"` + LastError string `json:"last_error"` +} + type MeshmapperScopeCatalogue struct { Iata string `json:"iata"` Url string `json:"url"` diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index 5ae22f2..1135eaa 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -20,6 +20,8 @@ type Querier interface { AddRegionIATAs(ctx context.Context, arg AddRegionIATAsParams) error // Hop prefixes that match >1 node in an IATA, per width. Computed once per reconfirm run. AmbiguousPrefixes(ctx context.Context) ([]AmbiguousPrefixesRow, error) + // Keeps fetch timing so re-enabling still respects backoff; the ETag goes with the rows. + ClearRepeaters(ctx context.Context) (int64, error) CreateAccount(ctx context.Context, name string) (Account, error) // Lock the current row before deciding the outcome, including when another // deactivation commits while this statement is waiting for its row lock. @@ -54,6 +56,7 @@ type Querier interface { DeleteOldTraceIATAs(ctx context.Context, lastHeard pgtype.Timestamptz) error DeleteOldTraceTags(ctx context.Context, lastHeardAt pgtype.Timestamptz) error DeleteRegionIATAsNotIn(ctx context.Context, arg DeleteRegionIATAsNotInParams) error + DeleteRepeaterSource(ctx context.Context, source string) error DeleteRollupHour(ctx context.Context, hour pgtype.Timestamptz) error // Dirty hours that can no longer be re-rolled. DeleteStaleDirtyHours(ctx context.Context) error @@ -257,6 +260,7 @@ type Querier interface { // REGIONS // ============================================================ ListRegions(ctx context.Context) ([]ListRegionsRow, error) + ListRepeaterSources(ctx context.Context) ([]MeshmapperRepeaterSource, error) // Leading index equalities and the time/ID boundary bound both custom and generic plans. // Full-byte equality is required even when the compact digest matches. TRACE path bytes // carry readings; unclassified legacy observations cannot be safely called ordinary paths. @@ -302,6 +306,9 @@ type Querier interface { // (and not before @since), zero-observation hours included. // No observations yet means nothing to register (GREATEST would ignore the NULL MIN). RegisterRollupHours(ctx context.Context, since pgtype.Timestamptz) error + // rows is a JSON array of {"key": hex, "data": {...}}; unchanged rows are not rewritten. + // fetchedAt comes from the source's checked_at when read. + ReplaceRepeaters(ctx context.Context, arg ReplaceRepeatersParams) error // Batch form of ResolveEndpointHashes for a page of packets. Matches the cross // product of IATAs and hashes; callers pick out the pairs they asked for. ResolveEndpointHashPairs(ctx context.Context, arg ResolveEndpointHashPairsParams) ([]ResolveEndpointHashPairsRow, error) @@ -350,6 +357,9 @@ type Querier interface { RollupUnlock(ctx context.Context) error // NULL payload/etag/checked_at retain the last good list after an error or 304. SaveChannelCatalogue(ctx context.Context, arg SaveChannelCatalogueParams) error + // NULL etag/checked_at retain the last good list's state after an error or 304; + // a new URL drops the old site's ETag. + SaveRepeaterSource(ctx context.Context, arg SaveRepeaterSourceParams) error // One statement commits the validated snapshot and its lookup identities together. // Empty arrays insert nothing. NULL payload/checked_at retain last-known-good data // after an error or 304. Imported names never replace existing manual metadata. diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 74da580..7ea8cab 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -109,6 +109,20 @@ func (q *Queries) AmbiguousPrefixes(ctx context.Context) ([]AmbiguousPrefixesRow return items, nil } +const clearRepeaters = `-- name: ClearRepeaters :execrows +WITH reset AS (UPDATE meshmapper_repeater_sources SET etag = NULL, checked_at = NULL) +DELETE FROM meshmapper_repeaters +` + +// Keeps fetch timing so re-enabling still respects backoff; the ETag goes with the rows. +func (q *Queries) ClearRepeaters(ctx context.Context) (int64, error) { + result, err := q.db.Exec(ctx, clearRepeaters) + if err != nil { + return 0, err + } + return result.RowsAffected(), nil +} + const createAccount = `-- name: CreateAccount :one INSERT INTO accounts (name) VALUES ($1) ON CONFLICT (name) WHERE deactivated_at IS NULL DO NOTHING @@ -364,6 +378,16 @@ func (q *Queries) DeleteRegionIATAsNotIn(ctx context.Context, arg DeleteRegionIA return err } +const deleteRepeaterSource = `-- name: DeleteRepeaterSource :exec +WITH removed AS (DELETE FROM meshmapper_repeaters r WHERE r.source = $1) +DELETE FROM meshmapper_repeater_sources s WHERE s.source = $1 +` + +func (q *Queries) DeleteRepeaterSource(ctx context.Context, source string) error { + _, err := q.db.Exec(ctx, deleteRepeaterSource, source) + return err +} + const getAccount = `-- name: GetAccount :one SELECT id, name, created_at, deactivated_at FROM accounts WHERE id = $1 ` @@ -585,7 +609,12 @@ SELECT n.id, n.public_key, n.node_type, n.name, n.latitude, n.longitude, n.locat (SELECT o.id FROM observers o WHERE o.public_key = n.public_key LIMIT 1) AS observer_id, (SELECT json_agg(json_build_object('iata', ni.iata, 'lastHeard', (extract(epoch from ni.last_heard) * 1000)::bigint) ORDER BY ni.last_heard DESC) FROM node_iatas ni WHERE ni.node_id = n.id) AS iatas, - (SELECT COUNT(DISTINCT nn.neighbor_id) FROM node_neighbors nn WHERE nn.node_id = n.id)::bigint AS known_neighbor_count + (SELECT COUNT(DISTINCT nn.neighbor_id) FROM node_neighbors nn WHERE nn.node_id = n.id)::bigint AS known_neighbor_count, + -- The most recently checked source wins when group sites list the same repeater. + (SELECT m.data || jsonb_build_object('fetchedAt', (extract(epoch from s.checked_at) * 1000)::bigint) + FROM meshmapper_repeaters m JOIN meshmapper_repeater_sources s ON s.source = m.source + WHERE m.public_key = n.public_key AND s.checked_at IS NOT NULL + ORDER BY s.checked_at DESC LIMIT 1)::jsonb AS meshmapper FROM nodes n LEFT JOIN transport_scopes ts ON ts.id = n.default_scope_id WHERE n.id = $1 @@ -616,6 +645,7 @@ type GetNodeByIDRow struct { ObserverID uuid.UUID `json:"observer_id"` Iatas []byte `json:"iatas"` KnownNeighborCount int64 `json:"known_neighbor_count"` + Meshmapper json.RawMessage `json:"meshmapper"` } func (q *Queries) GetNodeByID(ctx context.Context, id uuid.UUID) (GetNodeByIDRow, error) { @@ -646,6 +676,7 @@ func (q *Queries) GetNodeByID(ctx context.Context, id uuid.UUID) (GetNodeByIDRow &i.ObserverID, &i.Iatas, &i.KnownNeighborCount, + &i.Meshmapper, ) return i, err } @@ -4086,6 +4117,38 @@ func (q *Queries) ListRegions(ctx context.Context) ([]ListRegionsRow, error) { return items, nil } +const listRepeaterSources = `-- name: ListRepeaterSources :many +SELECT source, url, etag, checked_at, attempted_at, next_attempt, last_error FROM meshmapper_repeater_sources ORDER BY source +` + +func (q *Queries) ListRepeaterSources(ctx context.Context) ([]MeshmapperRepeaterSource, error) { + rows, err := q.db.Query(ctx, listRepeaterSources) + if err != nil { + return nil, err + } + defer rows.Close() + items := []MeshmapperRepeaterSource{} + for rows.Next() { + var i MeshmapperRepeaterSource + if err := rows.Scan( + &i.Source, + &i.Url, + &i.Etag, + &i.CheckedAt, + &i.AttemptedAt, + &i.NextAttempt, + &i.LastError, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const listScopeCatalogues = `-- name: ListScopeCatalogues :many SELECT iata, url, payload, etag, checked_at, attempted_at, next_attempt, last_error FROM meshmapper_scope_catalogues ORDER BY iata, attempted_at ` @@ -4528,6 +4591,32 @@ func (q *Queries) RefreshRadioPresets(ctx context.Context) error { return err } +const replaceRepeaters = `-- name: ReplaceRepeaters :exec +WITH incoming AS ( + SELECT decode(r->>'key', 'hex') AS public_key, (r->'data') - 'fetchedAt' AS data + FROM jsonb_array_elements($2::jsonb) r +), removed AS ( + DELETE FROM meshmapper_repeaters m + WHERE m.source = $1 AND m.public_key NOT IN (SELECT public_key FROM incoming) +) +INSERT INTO meshmapper_repeaters (source, public_key, data) +SELECT $1, public_key, data FROM incoming +ON CONFLICT (source, public_key) DO UPDATE SET data = EXCLUDED.data +WHERE meshmapper_repeaters.data IS DISTINCT FROM EXCLUDED.data +` + +type ReplaceRepeatersParams struct { + Source string `json:"source"` + Rows json.RawMessage `json:"rows"` +} + +// rows is a JSON array of {"key": hex, "data": {...}}; unchanged rows are not rewritten. +// fetchedAt comes from the source's checked_at when read. +func (q *Queries) ReplaceRepeaters(ctx context.Context, arg ReplaceRepeatersParams) error { + _, err := q.db.Exec(ctx, replaceRepeaters, arg.Source, arg.Rows) + return err +} + const resolveEndpointHashPairs = `-- name: ResolveEndpointHashPairs :many SELECT ns.iata, ns.prefix_1 AS hash, n.id AS node_id, n.name, n.latitude, n.longitude, n.public_key FROM node_short_ids ns @@ -5053,6 +5142,44 @@ func (q *Queries) SaveChannelCatalogue(ctx context.Context, arg SaveChannelCatal return err } +const saveRepeaterSource = `-- name: SaveRepeaterSource :exec +INSERT INTO meshmapper_repeater_sources (source, url, etag, checked_at, attempted_at, next_attempt, last_error) +VALUES ($1, $2, $3::text, $4::timestamptz, $5, $6, $7) +ON CONFLICT (source) DO UPDATE SET + url = EXCLUDED.url, + etag = CASE WHEN meshmapper_repeater_sources.url = EXCLUDED.url + THEN COALESCE(EXCLUDED.etag, meshmapper_repeater_sources.etag) ELSE EXCLUDED.etag END, + checked_at = COALESCE(EXCLUDED.checked_at, meshmapper_repeater_sources.checked_at), + attempted_at = EXCLUDED.attempted_at, + next_attempt = EXCLUDED.next_attempt, + last_error = EXCLUDED.last_error +` + +type SaveRepeaterSourceParams struct { + Source string `json:"source"` + Url string `json:"url"` + Etag *string `json:"etag"` + CheckedAt pgtype.Timestamptz `json:"checked_at"` + AttemptedAt pgtype.Timestamptz `json:"attempted_at"` + NextAttempt pgtype.Timestamptz `json:"next_attempt"` + LastError string `json:"last_error"` +} + +// NULL etag/checked_at retain the last good list's state after an error or 304; +// a new URL drops the old site's ETag. +func (q *Queries) SaveRepeaterSource(ctx context.Context, arg SaveRepeaterSourceParams) error { + _, err := q.db.Exec(ctx, saveRepeaterSource, + arg.Source, + arg.Url, + arg.Etag, + arg.CheckedAt, + arg.AttemptedAt, + arg.NextAttempt, + arg.LastError, + ) + return err +} + const saveScopeCatalogue = `-- name: SaveScopeCatalogue :exec WITH inserted AS ( INSERT INTO transport_scopes (name, transport_key, key_fingerprint, imported_only) diff --git a/docs/docs.go b/docs/docs.go index bf3fc9f..9991944 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -4085,6 +4085,120 @@ const docTemplate = `{ } } }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.MeshMapperDefaultScope": { + "type": "object", + "properties": { + "name": { + "type": "string" + }, + "source": { + "description": "advert or reported", + "type": "string" + }, + "state": { + "description": "named, unknown (set but not nameable yet) or none", + "type": "string" + }, + "updatedAt": { + "type": "integer" + } + } + }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.MeshMapperRepeater": { + "type": "object", + "properties": { + "admins": { + "type": "array", + "items": { + "type": "string" + } + }, + "carpeater": { + "description": "mobile repeater: its position moves", + "type": "boolean" + }, + "defaultScope": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.MeshMapperDefaultScope" + }, + "fetchedAt": { + "description": "when Beacon last fetched this list", + "type": "integer" + }, + "iata": { + "description": "region of its latest advert, per MeshMapper", + "type": "string" + }, + "scopes": { + "type": "array", + "items": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.MeshMapperScope" + } + }, + "scopesCheckedAt": { + "description": "set (with empty scopes) when it answered with none", + "type": "integer" + }, + "site": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.MeshMapperSite" + } + } + }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.MeshMapperScope": { + "type": "object", + "properties": { + "observedAt": { + "description": "seen forwarding scoped traffic", + "type": "integer" + }, + "observedPackets": { + "type": "integer" + }, + "proof": { + "description": "multibyte or floor", + "type": "string" + }, + "reportedAt": { + "description": "an observer asked it", + "type": "integer" + }, + "scope": { + "type": "string" + }, + "uploadedAt": { + "description": "a wardriver's app asked it", + "type": "integer" + } + } + }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.MeshMapperSite": { + "type": "object", + "properties": { + "antenna": { + "type": "string" + }, + "hardware": { + "type": "string" + }, + "heightAslM": { + "description": "antenna above mean sea level", + "type": "number" + }, + "heightM": { + "description": "antenna above ground", + "type": "number" + }, + "notes": { + "type": "string" + }, + "power": { + "type": "string" + }, + "powerSource": { + "description": "mains, solar, battery or poe", + "type": "string" + } + } + }, "github_com_MeshCore-Beacon_beacon-server_internal_api.Node": { "type": "object", "properties": { @@ -4143,6 +4257,14 @@ const docTemplate = `{ "description": "\"advert\" or \"manual\"", "type": "string" }, + "meshmapper": { + "description": "MeshMapper is what MeshMapper's repeater list says about this key; omitted when\nthe import is off or MeshMapper doesn't list it.", + "allOf": [ + { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.MeshMapperRepeater" + } + ] + }, "metadata": { "description": "raw JSONB metadata" }, diff --git a/docs/swagger.json b/docs/swagger.json index 2fb7dba..f5aee6d 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -4083,6 +4083,120 @@ } } }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.MeshMapperDefaultScope": { + "type": "object", + "properties": { + "name": { + "type": "string" + }, + "source": { + "description": "advert or reported", + "type": "string" + }, + "state": { + "description": "named, unknown (set but not nameable yet) or none", + "type": "string" + }, + "updatedAt": { + "type": "integer" + } + } + }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.MeshMapperRepeater": { + "type": "object", + "properties": { + "admins": { + "type": "array", + "items": { + "type": "string" + } + }, + "carpeater": { + "description": "mobile repeater: its position moves", + "type": "boolean" + }, + "defaultScope": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.MeshMapperDefaultScope" + }, + "fetchedAt": { + "description": "when Beacon last fetched this list", + "type": "integer" + }, + "iata": { + "description": "region of its latest advert, per MeshMapper", + "type": "string" + }, + "scopes": { + "type": "array", + "items": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.MeshMapperScope" + } + }, + "scopesCheckedAt": { + "description": "set (with empty scopes) when it answered with none", + "type": "integer" + }, + "site": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.MeshMapperSite" + } + } + }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.MeshMapperScope": { + "type": "object", + "properties": { + "observedAt": { + "description": "seen forwarding scoped traffic", + "type": "integer" + }, + "observedPackets": { + "type": "integer" + }, + "proof": { + "description": "multibyte or floor", + "type": "string" + }, + "reportedAt": { + "description": "an observer asked it", + "type": "integer" + }, + "scope": { + "type": "string" + }, + "uploadedAt": { + "description": "a wardriver's app asked it", + "type": "integer" + } + } + }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.MeshMapperSite": { + "type": "object", + "properties": { + "antenna": { + "type": "string" + }, + "hardware": { + "type": "string" + }, + "heightAslM": { + "description": "antenna above mean sea level", + "type": "number" + }, + "heightM": { + "description": "antenna above ground", + "type": "number" + }, + "notes": { + "type": "string" + }, + "power": { + "type": "string" + }, + "powerSource": { + "description": "mains, solar, battery or poe", + "type": "string" + } + } + }, "github_com_MeshCore-Beacon_beacon-server_internal_api.Node": { "type": "object", "properties": { @@ -4141,6 +4255,14 @@ "description": "\"advert\" or \"manual\"", "type": "string" }, + "meshmapper": { + "description": "MeshMapper is what MeshMapper's repeater list says about this key; omitted when\nthe import is off or MeshMapper doesn't list it.", + "allOf": [ + { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.MeshMapperRepeater" + } + ] + }, "metadata": { "description": "raw JSONB metadata" }, diff --git a/docs/swagger.yaml b/docs/swagger.yaml index 82a0dbc..77de161 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -320,6 +320,85 @@ definitions: description: stable identity within this IATA; use for route evidence links type: string type: object + github_com_MeshCore-Beacon_beacon-server_internal_api.MeshMapperDefaultScope: + properties: + name: + type: string + source: + description: advert or reported + type: string + state: + description: named, unknown (set but not nameable yet) or none + type: string + updatedAt: + type: integer + type: object + github_com_MeshCore-Beacon_beacon-server_internal_api.MeshMapperRepeater: + properties: + admins: + items: + type: string + type: array + carpeater: + description: 'mobile repeater: its position moves' + type: boolean + defaultScope: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.MeshMapperDefaultScope' + fetchedAt: + description: when Beacon last fetched this list + type: integer + iata: + description: region of its latest advert, per MeshMapper + type: string + scopes: + items: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.MeshMapperScope' + type: array + scopesCheckedAt: + description: set (with empty scopes) when it answered with none + type: integer + site: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.MeshMapperSite' + type: object + github_com_MeshCore-Beacon_beacon-server_internal_api.MeshMapperScope: + properties: + observedAt: + description: seen forwarding scoped traffic + type: integer + observedPackets: + type: integer + proof: + description: multibyte or floor + type: string + reportedAt: + description: an observer asked it + type: integer + scope: + type: string + uploadedAt: + description: a wardriver's app asked it + type: integer + type: object + github_com_MeshCore-Beacon_beacon-server_internal_api.MeshMapperSite: + properties: + antenna: + type: string + hardware: + type: string + heightAslM: + description: antenna above mean sea level + type: number + heightM: + description: antenna above ground + type: number + notes: + type: string + power: + type: string + powerSource: + description: mains, solar, battery or poe + type: string + type: object github_com_MeshCore-Beacon_beacon-server_internal_api.Node: properties: clockCheckedAt: @@ -368,6 +447,12 @@ definitions: locationSource: description: '"advert" or "manual"' type: string + meshmapper: + allOf: + - $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.MeshMapperRepeater' + description: |- + MeshMapper is what MeshMapper's repeater list says about this key; omitted when + the import is off or MeshMapper doesn't list it. metadata: description: raw JSONB metadata minFirmwareVersion: diff --git a/internal/api/handlers/nodes_test.go b/internal/api/handlers/nodes_test.go index 8ab84ba..c7ceebb 100644 --- a/internal/api/handlers/nodes_test.go +++ b/internal/api/handlers/nodes_test.go @@ -256,6 +256,31 @@ func TestGetNode_OK(t *testing.T) { if w.Code != http.StatusOK { t.Errorf("expected 200, got %d", w.Code) } + if strings.Contains(w.Body.String(), `"meshmapper"`) { + t.Error("meshmapper should be omitted when MeshMapper doesn't list the node") + } +} + +func TestGetNode_MeshMapper(t *testing.T) { + nodeID := uuid.MustParse("00000000-0000-0000-0000-000000000001") + hw, checked := "RAK4631", int64(1791100000000) + r := chi.NewRouter() + r.Get("/nodes/{nodeId}", getNode(stubReader{ + getNode: func(_ context.Context, id uuid.UUID) (*api.Node, error) { + return &api.Node{NodeSummary: api.NodeSummary{ID: id}, MeshMapper: &api.MeshMapperRepeater{ + IATA: "YOW", FetchedAt: 1791234600000, Site: api.MeshMapperSite{Hardware: &hw}, Carpeater: true, + Admins: []string{}, Scopes: []api.MeshMapperScope{}, ScopesCheckedAt: &checked, + DefaultScope: &api.MeshMapperDefaultScope{State: "none"}, + }}, nil + }, + })) + w := httptest.NewRecorder() + r.ServeHTTP(w, httptest.NewRequest(http.MethodGet, "/nodes/"+nodeID.String(), nil)) + want := `"meshmapper":{"iata":"YOW","fetchedAt":1791234600000,"site":{"hardware":"RAK4631"},"carpeater":true,` + + `"admins":[],"scopes":[],"scopesCheckedAt":1791100000000,"defaultScope":{"state":"none"}}` + if w.Code != http.StatusOK || !strings.Contains(w.Body.String(), want) { + t.Fatalf("status %d, body %s", w.Code, w.Body.String()) + } } func TestListNodeNeighbors_OK(t *testing.T) { diff --git a/internal/api/nodes.go b/internal/api/nodes.go index 9dce557..19e5f5b 100644 --- a/internal/api/nodes.go +++ b/internal/api/nodes.go @@ -80,6 +80,51 @@ type Node struct { ClockDriftSeconds *int `json:"clockDriftSeconds,omitempty"` ClockOutOfSync *bool `json:"clockOutOfSync,omitempty"` ClockCheckedAt *int64 `json:"clockCheckedAt,omitempty"` + // MeshMapper is what MeshMapper's repeater list says about this key; omitted when + // the import is off or MeshMapper doesn't list it. + MeshMapper *MeshMapperRepeater `json:"meshmapper,omitempty"` +} + +// MeshMapperRepeater is imported from MeshMapper's get_repeaters.php. Times are epoch ms. +// Text fields are admin-entered free text: escape before rendering. +type MeshMapperRepeater struct { + IATA string `json:"iata,omitempty"` // region of its latest advert, per MeshMapper + FetchedAt int64 `json:"fetchedAt"` // when Beacon last fetched this list + Site MeshMapperSite `json:"site"` + Carpeater bool `json:"carpeater"` // mobile repeater: its position moves + Admins []string `json:"admins"` + Scopes []MeshMapperScope `json:"scopes"` + ScopesCheckedAt *int64 `json:"scopesCheckedAt,omitempty"` // set (with empty scopes) when it answered with none + DefaultScope *MeshMapperDefaultScope `json:"defaultScope,omitempty"` +} + +// MeshMapperSite holds admin-entered site details; each field is omitted when unset. +type MeshMapperSite struct { + Power *string `json:"power,omitempty"` + Hardware *string `json:"hardware,omitempty"` + Antenna *string `json:"antenna,omitempty"` + HeightM *float64 `json:"heightM,omitempty"` // antenna above ground + HeightASLM *float64 `json:"heightAslM,omitempty"` // antenna above mean sea level + PowerSource *string `json:"powerSource,omitempty"` // mains, solar, battery or poe + Notes *string `json:"notes,omitempty"` +} + +// MeshMapperScope is one scope the repeater was confirmed carrying, and how. +type MeshMapperScope struct { + Scope string `json:"scope"` + ObservedAt *int64 `json:"observedAt,omitempty"` // seen forwarding scoped traffic + ObservedPackets *int64 `json:"observedPackets,omitempty"` + Proof *string `json:"proof,omitempty"` // multibyte or floor + ReportedAt *int64 `json:"reportedAt,omitempty"` // an observer asked it + UploadedAt *int64 `json:"uploadedAt,omitempty"` // a wardriver's app asked it +} + +// MeshMapperDefaultScope is the scope the repeater puts on its own adverts. +type MeshMapperDefaultScope struct { + Name *string `json:"name,omitempty"` + State string `json:"state"` // named, unknown (set but not nameable yet) or none + UpdatedAt *int64 `json:"updatedAt,omitempty"` + Source *string `json:"source,omitempty"` // advert or reported } // NodeTypeName returns a human-readable name for a node type integer. diff --git a/internal/cache/cache_test.go b/internal/cache/cache_test.go index ecedcab..5786c12 100644 --- a/internal/cache/cache_test.go +++ b/internal/cache/cache_test.go @@ -417,6 +417,18 @@ func TestCachedReader_InvalidateNode(t *testing.T) { } } +func TestInvalidateNodeDetails(t *testing.T) { + c, mr := newTestClient(t) + cr := &CachedReader{inner: &stubReader{}, c: c, ttl: CacheTTLs{}} + mr.Set(keyNodePrefix+"a", "x") + mr.Set(keyNodePrefix+"b", "x") + mr.Set(keyNodesByIDsPrefix+"a", "keep") + cr.InvalidateNodeDetails(context.Background()) + if got := mr.Keys(); len(got) != 1 || got[0] != keyNodesByIDsPrefix+"a" { + t.Fatalf("keys left: %v", got) + } +} + func TestCachedReader_InvalidateObserver(t *testing.T) { c, mr := newTestClient(t) observerID := uuid.MustParse("00000000-0000-0000-0000-000000000002") diff --git a/internal/cache/reader.go b/internal/cache/reader.go index 47b0e2d..d573ab9 100644 --- a/internal/cache/reader.go +++ b/internal/cache/reader.go @@ -113,6 +113,11 @@ func (cr *CachedReader) InvalidateNode(ctx context.Context, nodeID uuid.UUID) { cr.c.del(ctx, keyNodePrefix+id, keyNodeNeighborsPrefix+id) } +// InvalidateNodeDetails drops every cached node detail, after a MeshMapper repeater import. +func (cr *CachedReader) InvalidateNodeDetails(ctx context.Context) { + cr.c.delPrefix(ctx, keyNodePrefix) +} + // 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) { diff --git a/internal/config/meshmapper.go b/internal/config/meshmapper.go index 134b9f6..3c53c81 100644 --- a/internal/config/meshmapper.go +++ b/internal/config/meshmapper.go @@ -12,16 +12,31 @@ import ( // MeshMapper rate-limits get_zones/get_geojson to one request per target per // 23.5h and get_scopes to one per region per 55m; minimums round up. const ( - MinZonesRefresh = 24 * time.Hour - MinScopesRefresh = time.Hour - MinChannelsRefresh = 24 * time.Hour // get_channels.php: one per region per 23.5h + MinZonesRefresh = 24 * time.Hour + MinScopesRefresh = time.Hour + MinChannelsRefresh = 24 * time.Hour // get_channels.php: one per region per 23.5h + MinRepeatersRefresh = time.Hour // get_repeaters.php only has the shared daily key budget ) type MeshMapperConfig struct { - APIKey string `yaml:"api_key" json:"-"` - Scopes MeshMapperScopesConfig `yaml:"scopes"` - Zones MeshMapperZonesConfig `yaml:"zones"` - Channels MeshMapperChannelsConfig `yaml:"channels"` + APIKey string `yaml:"api_key" json:"-"` + Scopes MeshMapperScopesConfig `yaml:"scopes"` + Zones MeshMapperZonesConfig `yaml:"zones"` + Channels MeshMapperChannelsConfig `yaml:"channels"` + Repeaters MeshMapperRepeatersConfig `yaml:"repeaters"` +} + +// MeshMapperRepeatersConfig imports each known IATA's repeater site details for node detail. +type MeshMapperRepeatersConfig struct { + Enabled bool `yaml:"enabled"` + RefreshInterval duration `yaml:"refresh_interval"` +} + +func (c MeshMapperRepeatersConfig) Interval() time.Duration { + if c.RefreshInterval.Duration == 0 { + return 6 * time.Hour + } + return c.RefreshInterval.Duration } // MeshMapperChannelsConfig imports each known IATA's public hashtag channels. @@ -78,6 +93,9 @@ func (c *Config) validateMeshMapper() error { if ch := c.MeshMapper.Channels; ch.Enabled && (ch.Interval() < MinChannelsRefresh || ch.Interval() > 7*24*time.Hour) { return fmt.Errorf("meshmapper.channels.refresh_interval must be between 24h and 168h") } + if r := c.MeshMapper.Repeaters; r.Enabled && (r.Interval() < MinRepeatersRefresh || r.Interval() > 24*time.Hour) { + return fmt.Errorf("meshmapper.repeaters.refresh_interval must be between 1h and 24h") + } s := c.MeshMapper.Scopes if !s.Enabled { return nil diff --git a/internal/config/meshmapper_test.go b/internal/config/meshmapper_test.go index 15dcce7..c8beebe 100644 --- a/internal/config/meshmapper_test.go +++ b/internal/config/meshmapper_test.go @@ -120,3 +120,21 @@ func TestMeshMapperZonesConfig(t *testing.T) { }) } } + +func TestMeshMapperRepeatersConfig(t *testing.T) { + valid := "meshmapper:\n repeaters:\n enabled: true\n" + for text, bad := range map[string]bool{ + valid: false, + valid + " refresh_interval: 1h\n": false, + valid + " refresh_interval: 24h\n": false, + valid + " refresh_interval: 59m\n": true, + valid + " refresh_interval: 25h\n": true, + } { + if _, err := loadText(t, text); (err != nil) != bad { + t.Errorf("%q: error=%v", text, err) + } + } + if (MeshMapperRepeatersConfig{}).Interval() != 6*time.Hour { + t.Fatal("wrong default") + } +} diff --git a/internal/meshmapper/directory.go b/internal/meshmapper/directory.go index c50ed21..40f7b9c 100644 --- a/internal/meshmapper/directory.go +++ b/internal/meshmapper/directory.go @@ -22,6 +22,7 @@ import ( type zoneEntry struct { url string hasBoundary bool + group string // first group listing it, "" when none name string // short display name, e.g. "Nanaimo" lat, lon *float64 // nil when missing or out of range } @@ -223,6 +224,7 @@ func decodeZones(body []byte, country string) (map[string]zoneEntry, map[string] Lon *float64 `json:"lon"` URL string `json:"url"` HasBoundary bool `json:"has_boundary"` + Group *string `json:"group"` } `json:"zones"` Groups []struct { Code string `json:"code"` @@ -242,6 +244,9 @@ func decodeZones(body []byte, country string) (map[string]zoneEntry, map[string] if entry.name = strings.TrimSpace(entry.name); len(entry.name) > 128 || strings.ContainsFunc(entry.name, unicode.IsControl) { entry.name = "" } + if zone.Group != nil && groupCode.MatchString(strings.ToUpper(*zone.Group)) { + entry.group = strings.ToUpper(*zone.Group) + } if zone.Lat != nil && zone.Lon != nil && *zone.Lat >= -90 && *zone.Lat <= 90 && *zone.Lon >= -180 && *zone.Lon <= 180 { entry.lat, entry.lon = zone.Lat, zone.Lon } diff --git a/internal/meshmapper/repeaters.go b/internal/meshmapper/repeaters.go new file mode 100644 index 0000000..280dd77 --- /dev/null +++ b/internal/meshmapper/repeaters.go @@ -0,0 +1,455 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package meshmapper + +import ( + "cmp" + "context" + "encoding/hex" + "encoding/json" + "fmt" + "log/slog" + "math" + "net/http" + "regexp" + "slices" + "strings" + "time" + "unicode" + "unicode/utf8" + + "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/MeshCore-Beacon/beacon-server/internal/config" + "github.com/MeshCore-Beacon/beacon-server/internal/iatadb" +) + +const ( + MaxRepeaters = 20000 // per response; a group site lists every member's + MaxRepeaterBody = 32 << 20 // decompressed + + repeaterRetry = config.MinRepeatersRefresh + + maxAdmins = 32 + maxScopes = 64 + maxNameRunes = 64 + latestUnixTime = 32503680000 // year 3000; rejects ms or garbage in seconds fields +) + +var repeaterIATA = regexp.MustCompile(`^[A-Z0-9]{3,8}$`) + +// Repeater is one imported repeater, keyed by its full public key. +type Repeater struct { + Key []byte + Data api.MeshMapperRepeater +} + +// RepeaterStore keys sources by group code, or by IATA for a region outside a group. +type RepeaterStore interface { + ListKnownIATAs(ctx context.Context) ([]string, error) + ListRepeaterSources(ctx context.Context) ([]Catalogue, error) + // SaveRepeaterSource records a fetch; non-nil repeaters also replace the source's list. + SaveRepeaterSource(ctx context.Context, source, url string, cache Cache, repeaters []Repeater) error + DeleteRepeaterSource(ctx context.Context, source string) error + // ClearRepeaters drops every imported repeater but keeps fetch timing. + ClearRepeaters(ctx context.Context) (bool, error) +} + +type repeaterRegion struct{ iata, country string } + +type repeaterSource struct { + key, url string + cache Cache // Payload unused: the list lives in its own table +} + +// Repeaters is owned by one background task. +type Repeaters struct { + store RepeaterStore + dir *Directory + repeatersURL func(site string) (string, bool) + client *http.Client + interval time.Duration + onChange func(context.Context) + seen map[string]bool + regions []repeaterRegion + sources map[string]*repeaterSource + pausedUntil time.Time // a 429 spends the key's shared budget, so every source waits + warned map[string]time.Time // invalid site URL, logged once per interval +} + +// NewRepeaters restores fetch state without HTTP requests. Disabled, it drops every +// imported repeater. onChange runs whenever node detail may have changed. +func NewRepeaters(ctx context.Context, cfg config.MeshMapperRepeatersConfig, store RepeaterStore, dir *Directory, onChange func(context.Context)) (*Repeaters, error) { + r := &Repeaters{store: store, dir: dir, repeatersURL: repeatersEndpoint, interval: cfg.Interval(), onChange: onChange, + seen: map[string]bool{}, sources: map[string]*repeaterSource{}, warned: map[string]time.Time{}, client: dir.client} + if !cfg.Enabled { + cleared, err := store.ClearRepeaters(ctx) + if err != nil { + return nil, fmt.Errorf("clear MeshMapper repeaters: %w", err) + } + if cleared { + r.changed(ctx) + } + return r, nil + } + saved, err := store.ListRepeaterSources(ctx) + if err != nil { + return nil, fmt.Errorf("restore repeater sources: %w", err) + } + for _, cat := range saved { + s := &repeaterSource{key: cat.IATA, url: cat.URL, cache: cat.Cache} + r.sources[s.key] = s + r.log(*s, "restored", -1, 0) + } + iatas, err := store.ListKnownIATAs(ctx) + if err != nil { + return nil, fmt.Errorf("list IATAs for MeshMapper repeaters: %w", err) + } + r.track(iatas) + return r, nil +} + +func (r *Repeaters) track(iatas []string) { + for _, iata := range iatas { + if r.seen[iata] { + continue + } + r.seen[iata] = true + if country := iatadb.CountryFor(iata); country != "" { + r.regions = append(r.regions, repeaterRegion{iata, country}) + } + } + // Sorted so a group is always fetched from the same member's site. + slices.SortFunc(r.regions, func(a, b repeaterRegion) int { return strings.Compare(a.iata, b.iata) }) +} + +// Refresh makes at most one request. The scheduler serializes calls every 15s. +func (r *Repeaters) Refresh(ctx context.Context) (err error) { + parent := ctx + defer func() { + if parent.Err() != nil { + err = nil + } + }() + ctx, cancel := context.WithTimeout(ctx, refreshTimeout) + defer cancel() + now := time.Now().UTC() + if now.Before(r.pausedUntil) { + return nil + } + iatas, err := r.store.ListKnownIATAs(ctx) + if err != nil { + return fmt.Errorf("list IATAs for MeshMapper repeaters: %w", err) + } + r.track(iatas) + wanted, order, complete := map[string]string{}, []string{}, true + for _, region := range r.regions { + zones, fetched, err := r.dir.List(ctx, region.country, now) + if err != nil || fetched { + return err + } + if zones == nil { + complete = false + continue + } + entry, listed := zones[region.iata] + if !listed { + continue + } + endpoint, valid := r.repeatersURL(entry.url) + if !valid { + r.warnInvalidSite(region.iata, now) + continue + } + // Any member's site lists the whole group, so one request covers every member. + key := cmp.Or(entry.group, region.iata) + if _, ok := wanted[key]; !ok { + wanted[key], order = endpoint, append(order, key) + } + } + // Only with every country's list in hand can a missing source mean it's gone. + if complete { + if err := r.retire(ctx, wanted); err != nil { + return err + } + } + for _, key := range order { + s := r.sources[key] + if s == nil { + s = &repeaterSource{key: key} + r.sources[key] = s + } + if s.url != wanted[key] { + s.url, s.cache.ETag = wanted[key], "" + } + if now.Before(s.cache.NextAttempt) { + continue + } + return r.refresh(ctx, s, now) + } + return nil +} + +// retire drops sources no listed region maps to, e.g. a region that joined a group. +func (r *Repeaters) retire(ctx context.Context, wanted map[string]string) error { + for key, s := range r.sources { + if _, ok := wanted[key]; ok { + continue + } + if err := r.store.DeleteRepeaterSource(ctx, key); err != nil { + return fmt.Errorf("retire repeater source %s: %w", key, err) + } + delete(r.sources, key) + r.log(*s, "retired", -1, 0) + r.changed(ctx) + } + return nil +} + +func (r *Repeaters) refresh(ctx context.Context, s *repeaterSource, now time.Time) error { + update := Cache{AttemptedAt: now, NextAttempt: now.Add(r.interval)} + etag := "" + if !s.cache.CheckedAt.IsZero() { + etag = s.cache.ETag + } + // Recorded first: an abandoned request still spent the key's budget. + attempt := Cache{AttemptedAt: now, NextAttempt: now.Add(min(r.interval, repeaterRetry)), LastError: "no response"} + s.cache.NextAttempt = attempt.NextAttempt + if err := r.store.SaveRepeaterSource(ctx, s.key, s.url, attempt, nil); err != nil { + return fmt.Errorf("persist repeater source %s: %w", s.key, err) + } + s.cache.AttemptedAt, s.cache.LastError = now, attempt.LastError + status, body, header, err := get(ctx, r.client, "Beacon-MeshMapper-Repeaters/1", s.url, etag, MaxRepeaterBody) + if ctx.Err() != nil { + return ctx.Err() + } + var repeaters []Repeater + var skipped int + var retryAt time.Time + switch { + case err != nil: + update.LastError = requestProblem(err) + case status == http.StatusOK: + if repeaters, skipped, err = decodeRepeaters(body); err != nil { + update.LastError = "invalid response" + } + case status == http.StatusNotModified: + if s.cache.CheckedAt.IsZero() { + update.LastError = "304 without cached list" + } + default: + update.LastError = statusProblem(status, header, now, &retryAt) + } + if update.LastError == "" { + etag := header.Get("ETag") + if len(etag) > 256 || strings.ContainsAny(etag, "\r\n") { + update.LastError, repeaters = "invalid ETag", nil + } else { + if status == http.StatusNotModified && etag == "" { + etag = s.cache.ETag + } + update.ETag, update.CheckedAt = etag, now + } + } + if update.LastError != "" { + update.NextAttempt = now.Add(min(r.interval, repeaterRetry)) + if retryAt.After(update.NextAttempt) { + update.NextAttempt = retryAt + } + if status == http.StatusTooManyRequests { + r.pausedUntil = update.NextAttempt + } + } + if err := r.store.SaveRepeaterSource(ctx, s.key, s.url, update, repeaters); err != nil { + s.cache.NextAttempt = update.NextAttempt // never retry every tick during a DB outage + s.cache.LastError = "save failed" + r.log(*s, "checked", -1, 0) + return fmt.Errorf("persist repeater source %s: %w", s.key, err) + } + if !update.CheckedAt.IsZero() { + s.cache.CheckedAt, s.cache.ETag = update.CheckedAt, update.ETag + } + s.cache.AttemptedAt, s.cache.NextAttempt, s.cache.LastError = now, update.NextAttempt, update.LastError + count := -1 + if repeaters != nil { + count = len(repeaters) + r.changed(ctx) + } + r.log(*s, "checked", count, skipped) + return nil +} + +func (r *Repeaters) changed(ctx context.Context) { + if r.onChange != nil { + r.onChange(ctx) + } +} + +func (r *Repeaters) warnInvalidSite(iata string, now time.Time) { + if now.Before(r.warned[iata]) { + return + } + r.warned[iata] = now.Add(r.interval) + slog.Warn("MeshMapper repeaters skipped", "component", "meshmapper.repeaters", "iata", iata, "last_error", "invalid site URL") +} + +// log reports counts only when a list was just imported. +func (r *Repeaters) log(s repeaterSource, action string, count, skipped int) { + level := slog.LevelInfo + if s.cache.LastError != "" { + level = slog.LevelWarn + } + attrs := []any{"component", "meshmapper.repeaters", "source", s.key, "url", s.url, + "checked_at", s.cache.CheckedAt, "next_attempt", s.cache.NextAttempt, "last_error", s.cache.LastError} + if count >= 0 { + attrs = append(attrs, "repeaters", count, "skipped", skipped) + } + slog.Log(context.Background(), level, "MeshMapper repeaters "+action, attrs...) +} + +func repeatersEndpoint(site string) (string, bool) { return siteEndpoint(site, "get_repeaters.php") } + +type rawRepeater struct { + HexID string `json:"hex_id"` + IATA string `json:"iata"` + Power *string `json:"power"` + Hardware *string `json:"hardware"` + Antenna *string `json:"antenna"` + HeightM *float64 `json:"height_m"` + HeightASLM *float64 `json:"height_asl_m"` + PowerSource *string `json:"power_source"` + SiteNotes *string `json:"site_notes"` + IsCarpeater *int `json:"is_carpeater"` + Admins []string `json:"admins"` + Scopes []struct { + Scope string `json:"scope"` + ObservedAt *int64 `json:"observed_at"` + ObservedPkts *int64 `json:"observed_pkts"` + Proof *string `json:"proof"` + ReportedAt *int64 `json:"reported_at"` + UploadedAt *int64 `json:"uploaded_at"` + } `json:"scopes"` + ScopesCheckedAt *int64 `json:"scopes_checked_at"` + DefaultScope *string `json:"default_scope"` + DefaultScopeState *string `json:"default_scope_state"` + DefaultScopeAt *int64 `json:"default_scope_at"` + DefaultScopeSrc *string `json:"default_scope_src"` +} + +// decodeRepeaters rejects a malformed list but skips individual bad rows (counted), +// including legacy rows without a full key; nothing here feeds decryption or routing. +func decodeRepeaters(body []byte) (out []Repeater, skipped int, err error) { + if len(body) > MaxRepeaterBody || !utf8.Valid(body) { + return nil, 0, fmt.Errorf("invalid repeater list size/encoding") + } + var rows []json.RawMessage + if err := json.Unmarshal(body, &rows); err != nil { + return nil, 0, err + } + if rows == nil || len(rows) > MaxRepeaters { + return nil, 0, fmt.Errorf("invalid repeater list") + } + out = make([]Repeater, 0, len(rows)) + seen := map[string]bool{} + for _, row := range rows { + var raw rawRepeater + if json.Unmarshal(row, &raw) != nil || len(raw.HexID) != 64 { + skipped++ + continue + } + key, err := hex.DecodeString(raw.HexID) + if err != nil || seen[string(key)] { + skipped++ + continue + } + seen[string(key)] = true + out = append(out, Repeater{Key: key, Data: raw.normalize()}) + } + return out, skipped, nil +} + +func (raw rawRepeater) normalize() api.MeshMapperRepeater { + r := api.MeshMapperRepeater{ + Site: api.MeshMapperSite{ + Power: text(raw.Power, 20, false), + Hardware: text(raw.Hardware, 80, false), + Antenna: text(raw.Antenna, 80, false), + HeightM: within(raw.HeightM, -100, 2000), + HeightASLM: within(raw.HeightASLM, -500, 9000), + PowerSource: oneOf(raw.PowerSource, "mains", "solar", "battery", "poe"), + Notes: text(raw.SiteNotes, 200, true), + }, + Carpeater: raw.IsCarpeater != nil && *raw.IsCarpeater == 1, + Admins: []string{}, + Scopes: []api.MeshMapperScope{}, + ScopesCheckedAt: millis(raw.ScopesCheckedAt), + } + if repeaterIATA.MatchString(raw.IATA) { + r.IATA = raw.IATA + } + for _, name := range raw.Admins { + if n := text(&name, maxNameRunes, false); n != nil && len(r.Admins) < maxAdmins && !slices.Contains(r.Admins, *n) { + r.Admins = append(r.Admins, *n) + } + } + for _, s := range raw.Scopes { + name := text(&s.Scope, maxNameRunes, false) + if name == nil || len(r.Scopes) == maxScopes || + slices.ContainsFunc(r.Scopes, func(c api.MeshMapperScope) bool { return c.Scope == *name }) { + continue + } + scope := api.MeshMapperScope{Scope: *name, ObservedAt: millis(s.ObservedAt), Proof: oneOf(s.Proof, "multibyte", "floor"), + ReportedAt: millis(s.ReportedAt), UploadedAt: millis(s.UploadedAt)} + if s.ObservedPkts != nil && *s.ObservedPkts >= 0 { + scope.ObservedPackets = s.ObservedPkts + } + r.Scopes = append(r.Scopes, scope) + } + if state := oneOf(raw.DefaultScopeState, "named", "unknown", "none"); state != nil { + r.DefaultScope = &api.MeshMapperDefaultScope{State: *state, UpdatedAt: millis(raw.DefaultScopeAt), + Source: oneOf(raw.DefaultScopeSrc, "advert", "reported")} + if *state == "named" { + r.DefaultScope.Name = text(raw.DefaultScope, maxNameRunes, false) + } + } + return r +} + +// text trims admin-entered text and drops it when empty, over-long or carrying control +// characters (Postgres jsonb rejects NUL); multiline keeps newlines. +func text(s *string, maxRunes int, multiline bool) *string { + if s == nil { + return nil + } + v := strings.TrimSpace(*s) + if multiline { + v = strings.ReplaceAll(v, "\r\n", "\n") + } + if v == "" || utf8.RuneCountInString(v) > maxRunes || + strings.ContainsFunc(v, func(c rune) bool { return unicode.IsControl(c) && !(multiline && c == '\n') }) { + return nil + } + return &v +} + +func oneOf(s *string, allowed ...string) *string { + if s == nil || !slices.Contains(allowed, *s) { + return nil + } + return s +} + +func within(v *float64, lo, hi float64) *float64 { + if v == nil || math.IsNaN(*v) || *v < lo || *v > hi { + return nil + } + return v +} + +func millis(seconds *int64) *int64 { + if seconds == nil || *seconds <= 0 || *seconds > latestUnixTime { + return nil + } + ms := *seconds * 1000 + return &ms +} diff --git a/internal/meshmapper/repeaters_test.go b/internal/meshmapper/repeaters_test.go new file mode 100644 index 0000000..5834678 --- /dev/null +++ b/internal/meshmapper/repeaters_test.go @@ -0,0 +1,440 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package meshmapper + +import ( + "compress/gzip" + "context" + "encoding/hex" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "slices" + "strings" + "testing" + "time" + + "github.com/MeshCore-Beacon/beacon-server/internal/config" +) + +const repeaterKey = "A1B2C3D4E5F60718293A4B5C6D7E8F90A1B2C3D4E5F60718293A4B5C6D7E8F90" // the doc's key is one digit short + +// From the Repeaters API doc example, with a full-length key. +const repeaterDocExample = `[{ + "id": "A1", "hex_id": "` + repeaterKey + `", "name": "Gatineau Hills", + "lat": 45.512345, "lon": -75.912345, "last_heard": 1791234567, "created_at": 1773103950, + "enabled": 1, "iata": "YOW", "advert_bytes": 2, "hop_bytes": 2, "multibyte_capable": 1, "time_offset": 3, + "power": "1W", "hardware": "RAK4631", "antenna": "5.8 dBi fibreglass", "height_m": 12, "height_asl_m": 341.6, + "power_source": "solar", "site_notes": "On the fire tower, east face", + "backbone": 1, "backbone_share": 0.084, + "default_scope": "yow", "default_scope_state": "named", "default_scope_at": 1791230000, "default_scope_src": "advert", + "admins": ["MrAlders0n"], + "proven_neighbours": [{ "key": "B7C8D9E0F1A2B3C4", "resolved": 0, "snr": 6.25, "heard_at": 1791230011 }], + "radios": [{ "preset": "910.525,62.5,7", "last_seen": 1791234500, "witnesses": 4 }], + "preset_current": "910.525,62.5,7", + "scopes": [{ "scope": "yow", "observed_at": 1791200000, "observed_pkts": 37, "proof": "multibyte", "reported_at": 1791100000, "uploaded_at": null }], + "scopes_checked_at": 1791100000, "stale_repeater_hours": 24, "single_observer_mode": 0, + "is_observer": 0, "is_carpeater": 0 +}]` + +func TestDecodeRepeatersDocExample(t *testing.T) { + got, skipped, err := decodeRepeaters([]byte(repeaterDocExample)) + if err != nil || len(got) != 1 || skipped != 0 { + t.Fatal(got, skipped, err) + } + if hex.EncodeToString(got[0].Key) != strings.ToLower(repeaterKey) { + t.Fatal("key", got[0].Key) + } + b, _ := json.Marshal(got[0].Data) + want := `{"iata":"YOW","fetchedAt":0,"site":{"power":"1W","hardware":"RAK4631","antenna":"5.8 dBi fibreglass","heightM":12,` + + `"heightAslM":341.6,"powerSource":"solar","notes":"On the fire tower, east face"},"carpeater":false,"admins":["MrAlders0n"],` + + `"scopes":[{"scope":"yow","observedAt":1791200000000,"observedPackets":37,"proof":"multibyte","reportedAt":1791100000000}],` + + `"scopesCheckedAt":1791100000000,"defaultScope":{"name":"yow","state":"named","updatedAt":1791230000000,"source":"advert"}}` + if string(b) != want { + t.Fatalf("got %s\nwant %s", b, want) + } +} + +func TestDecodeRepeatersSkipsBadRows(t *testing.T) { + key := func(c string) string { return strings.Repeat(c, 64) } + body := `[ + {"hex_id": "A1B2C3"}, + {"hex_id": "` + key("Z") + `"}, + {"hex_id": "` + key("a") + `", "is_carpeater": 1, "power_source": "nuclear", "hardware": "` + strings.Repeat("x", 81) + `", + "admins": ["", "ok", "ok", "bad\u0000name"], "scopes": [{"scope": ""}, {"scope": "bc", "proof": "guess", "observed_at": 1791200000000}, {"scope": "bc"}], + "default_scope_state": "maybe", "iata": "", "site_notes": "nul\u0000here", "antenna": " whip ", "height_m": 1e300, "height_asl_m": -9999}, + {"hex_id": "` + key("A") + `", "power": "dup"}, + {"hex_id": "` + key("b") + `", "height_m": "12"}, + {"hex_id": "` + key("c") + `", "default_scope_state": "unknown", "default_scope": "secret", "site_notes": "line one\r\nline two", "power": "1\tW"} + ]` + got, skipped, err := decodeRepeaters([]byte(body)) + if err != nil || len(got) != 2 || skipped != 4 { + t.Fatalf("want 2 rows and 4 skipped, got %d and %d: %v", len(got), skipped, err) + } + a := got[0].Data + if !a.Carpeater || a.Site.PowerSource != nil || a.Site.Hardware != nil || a.IATA != "" || a.DefaultScope != nil || + !slices.Equal(a.Admins, []string{"ok"}) || len(a.Scopes) != 1 || a.Scopes[0].Proof != nil || a.Scopes[0].ObservedAt != nil { + t.Fatalf("bad values kept: %+v", a) + } + if a.Site.Notes != nil || a.Site.HeightM != nil || a.Site.HeightASLM != nil { + t.Fatalf("control characters or absurd heights kept: %+v", a.Site) + } + if a.Site.Antenna == nil || *a.Site.Antenna != "whip" { + t.Fatal("text not trimmed", a.Site.Antenna) + } + if a.Site.Power != nil { + t.Fatal("duplicate key replaced the first row") + } + c := got[1].Data + if c.DefaultScope == nil || c.DefaultScope.State != "unknown" || c.DefaultScope.Name != nil { + t.Fatalf("unnamed default scope: %+v", c.DefaultScope) + } + if c.Site.Notes == nil || *c.Site.Notes != "line one\nline two" || c.Site.Power != nil { + t.Fatalf("notes newline or single-line tab: %+v", c.Site) + } + for _, bad := range []string{`null`, `{}`, `[`, `"x"`} { + if _, _, err := decodeRepeaters([]byte(bad)); err == nil { + t.Fatalf("%s accepted", bad) + } + } + if got, _, err := decodeRepeaters([]byte(`[]`)); err != nil || got == nil || len(got) != 0 { + t.Fatal("empty region list rejected", got, err) + } +} + +func TestDecodeRepeatersLimits(t *testing.T) { + if _, _, err := decodeRepeaters([]byte("[" + strings.Repeat("{},", MaxRepeaters) + "{}]")); err == nil { + t.Fatal("over MaxRepeaters accepted") + } + if _, _, err := decodeRepeaters(make([]byte, MaxRepeaterBody+1)); err == nil { + t.Fatal("over MaxRepeaterBody accepted") + } +} + +type repeaterMemoryStore struct { + iatas []string + rows map[string]sourceKeyCache + repeaters map[string][]Repeater + saves int +} + +type sourceKeyCache struct { + url string + cache Cache +} + +func newRepeaterMemoryStore(iatas ...string) *repeaterMemoryStore { + return &repeaterMemoryStore{iatas: iatas, rows: map[string]sourceKeyCache{}, repeaters: map[string][]Repeater{}} +} + +func (s *repeaterMemoryStore) ListKnownIATAs(context.Context) ([]string, error) { + return slices.Clone(s.iatas), nil +} + +func (s *repeaterMemoryStore) ListRepeaterSources(context.Context) ([]Catalogue, error) { + var out []Catalogue + for key, r := range s.rows { + out = append(out, Catalogue{IATA: key, URL: r.url, Cache: r.cache}) + } + return out, nil +} + +func (s *repeaterMemoryStore) SaveRepeaterSource(_ context.Context, key, url string, next Cache, repeaters []Repeater) error { + s.saves++ + if repeaters != nil { + s.repeaters[key] = repeaters + } + if next.CheckedAt.IsZero() { + old := s.rows[key] + next.CheckedAt, next.ETag = old.cache.CheckedAt, old.cache.ETag + if old.url != url { + next.ETag = "" + } + } + s.rows[key] = sourceKeyCache{url, next} + return nil +} + +func (s *repeaterMemoryStore) DeleteRepeaterSource(_ context.Context, key string) error { + delete(s.rows, key) + delete(s.repeaters, key) + return nil +} + +func (s *repeaterMemoryStore) ClearRepeaters(context.Context) (bool, error) { + cleared := len(s.repeaters) > 0 + s.repeaters = map[string][]Repeater{} + for key, r := range s.rows { + r.cache.CheckedAt, r.cache.ETag = time.Time{}, "" + s.rows[key] = r + } + return cleared, nil +} + +type repeaterZone struct{ code, group string } + +type repeaterSite struct { + *httptest.Server + zones []repeaterZone + body string + status int + retryAfter string + etag string + listCalls int + calls int + gzipped bool + ifNoneMatch []string + requestedPaths []string +} + +func newRepeaterSite(t *testing.T, zones ...repeaterZone) *repeaterSite { + if len(zones) == 0 { + zones = []repeaterZone{{code: "YOW"}} + } + f := &repeaterSite{status: 200, body: repeaterDocExample, zones: zones} + f.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch { + case r.URL.Path == "/get_zones.php": + f.listCalls++ + var rows []string + for _, z := range f.zones { + group := "null" + if z.group != "" { + group = `"` + z.group + `"` + } + rows = append(rows, fmt.Sprintf(`{"code":%q,"url":"%s/%s/","has_boundary":true,"group":%s}`, z.code, f.URL, z.code, group)) + } + _, _ = w.Write([]byte(`{"country":"CA","zones":[` + strings.Join(rows, ",") + `]}`)) + case strings.HasSuffix(r.URL.Path, "/get_repeaters.php"): + f.calls++ + f.requestedPaths = append(f.requestedPaths, r.URL.Path) + f.ifNoneMatch = append(f.ifNoneMatch, r.Header.Get("If-None-Match")) + w.Header().Set("Retry-After", f.retryAfter) + if f.etag != "" { + w.Header().Set("ETag", f.etag) + } + if f.status != 200 { + w.WriteHeader(f.status) + return + } + if !strings.Contains(r.Header.Get("Accept-Encoding"), "gzip") { + _, _ = w.Write([]byte(f.body)) + return + } + f.gzipped = true + w.Header().Set("Content-Encoding", "gzip") + zw := gzip.NewWriter(w) + _, _ = zw.Write([]byte(f.body)) + _ = zw.Close() + default: + t.Errorf("unexpected path %s", r.URL.Path) + } + })) + t.Cleanup(f.Close) + return f +} + +type repeaterHarness struct { + *Repeaters + changes int +} + +func newRepeaterImporter(t *testing.T, f *repeaterSite, store *repeaterMemoryStore, lists *zoneListMemory, enabled bool) *repeaterHarness { + t.Helper() + dir := newTestDirectory(lists) + dir.listURL = f.URL + "/get_zones.php" + if err := dir.Restore(context.Background()); err != nil { + t.Fatal(err) + } + h := &repeaterHarness{} + r, err := NewRepeaters(context.Background(), config.MeshMapperRepeatersConfig{Enabled: enabled}, store, dir, + func(context.Context) { h.changes++ }) + if err != nil { + t.Fatal(err) + } + r.repeatersURL = func(site string) (string, bool) { return site + "get_repeaters.php", strings.HasPrefix(site, f.URL) } + h.Repeaters = r + return h +} + +func (h *repeaterHarness) refreshN(t *testing.T, n int) { + t.Helper() + for range n { + if err := h.Refresh(context.Background()); err != nil { + t.Fatal(err) + } + } +} + +func (h *repeaterHarness) due(keys ...string) { + for _, key := range keys { + h.sources[key].cache.NextAttempt = time.Time{} + } +} + +func TestRepeatersImportGzippedList(t *testing.T) { + f := newRepeaterSite(t) + store := newRepeaterMemoryStore("YOW") + h := newRepeaterImporter(t, f, store, newZoneListMemory(), true) + h.refreshN(t, 2) + if !f.gzipped || len(store.repeaters["YOW"]) != 1 || store.repeaters["YOW"][0].Data.Site.Hardware == nil || h.changes != 1 { + t.Fatal("list not imported", f.gzipped, store.repeaters, h.changes) + } + if c := h.sources["YOW"].cache; c.LastError != "" || c.NextAttempt.Sub(c.AttemptedAt) != 6*time.Hour { + t.Fatal("refresh state", c) + } + f.body = `[]` + h.due("YOW") + h.refreshN(t, 1) + if got, ok := store.repeaters["YOW"]; !ok || len(got) != 0 { + t.Fatal("empty list did not replace the region's repeaters", got) + } +} + +func TestRepeatersGroupFetchedOnce(t *testing.T) { + f := newRepeaterSite(t, repeaterZone{"YOW", "ONQC"}, repeaterZone{"YUL", "ONQC"}, repeaterZone{"YYZ", ""}) + store := newRepeaterMemoryStore("YUL", "YOW", "YYZ") + h := newRepeaterImporter(t, f, store, newZoneListMemory(), true) + h.refreshN(t, 4) + if f.calls != 2 || !slices.Equal(f.requestedPaths, []string{"/YOW/get_repeaters.php", "/YYZ/get_repeaters.php"}) { + t.Fatal("group not fetched once from its first member", f.requestedPaths) + } + if _, ok := store.repeaters["ONQC"]; !ok || len(store.repeaters) != 2 { + t.Fatal("sources", store.repeaters) + } + + // YYZ joining the group retires its own source once every country list is in hand. + f.zones[2].group = "ONQC" + h.dir.lists["CA"].fetchedAt = time.Time{} + h.dir.lists["CA"].nextAttempt = time.Time{} + h.refreshN(t, 2) + if _, ok := store.rows["YYZ"]; ok || h.sources["YYZ"] != nil || store.repeaters["YYZ"] != nil { + t.Fatal("retired source kept", store.rows) + } +} + +func TestRepeatersFailuresKeepLastGood(t *testing.T) { + for _, tc := range []struct { + status int + retryAfter string + body string + want time.Duration + }{ + {503, "", "", time.Hour}, + {429, "60", "", time.Hour}, + {429, "7200", "", 2 * time.Hour}, + {200, "", `{"error":"nope"}`, time.Hour}, + } { + f := newRepeaterSite(t) + store := newRepeaterMemoryStore("YOW") + h := newRepeaterImporter(t, f, store, newZoneListMemory(), true) + h.refreshN(t, 2) + f.status, f.retryAfter = tc.status, tc.retryAfter + if tc.body != "" { + f.body = tc.body + } + h.due("YOW") + h.refreshN(t, 2) + c := h.sources["YOW"].cache + if f.calls != 2 || c.LastError == "" || c.NextAttempt.Sub(c.AttemptedAt) != tc.want || len(store.repeaters["YOW"]) != 1 { + t.Fatalf("HTTP %d %q: calls=%d err=%q wait=%v kept=%d", tc.status, tc.body, f.calls, c.LastError, + c.NextAttempt.Sub(c.AttemptedAt), len(store.repeaters["YOW"])) + } + } +} + +func TestRepeatersRateLimitPausesEverySource(t *testing.T) { + f := newRepeaterSite(t, repeaterZone{"YOW", ""}, repeaterZone{"YYZ", ""}) + f.status, f.retryAfter = 429, "7200" + h := newRepeaterImporter(t, f, newRepeaterMemoryStore("YOW", "YYZ"), newZoneListMemory(), true) + h.refreshN(t, 4) + if f.calls != 1 || h.sources["YYZ"] != nil && !h.sources["YYZ"].cache.AttemptedAt.IsZero() { + t.Fatal("a 429 from the shared budget didn't pause the other source", f.calls) + } + if h.pausedUntil.Sub(h.sources["YOW"].cache.AttemptedAt) != 2*time.Hour { + t.Fatal("pause", h.pausedUntil) + } + h.pausedUntil = time.Time{} + f.status = 503 + h.due("YOW") + h.refreshN(t, 1) + if !h.pausedUntil.IsZero() { + t.Fatal("503 paused every source") + } +} + +func TestRepeatersNotModified(t *testing.T) { + f := newRepeaterSite(t) + f.etag = `"r1"` + store := newRepeaterMemoryStore("YOW") + h := newRepeaterImporter(t, f, store, newZoneListMemory(), true) + h.refreshN(t, 2) + f.status, f.etag = 304, "" + h.due("YOW") + h.refreshN(t, 1) + c := h.sources["YOW"].cache + if !slices.Equal(f.ifNoneMatch, []string{"", `"r1"`}) || c.LastError != "" || c.ETag != `"r1"` || len(store.repeaters["YOW"]) != 1 { + t.Fatalf("304 handling: sent=%q err=%q etag=%q", f.ifNoneMatch, c.LastError, c.ETag) + } + + // A 304 can't stand in for a list Beacon never stored. + h.sources["YOW"].cache.CheckedAt, h.sources["YOW"].cache.ETag = time.Time{}, `"stale"` + h.due("YOW") + h.refreshN(t, 1) + if got := f.ifNoneMatch[2]; got != "" || h.sources["YOW"].cache.LastError != "304 without cached list" { + t.Fatalf("sent %q without a list, err=%q", got, h.sources["YOW"].cache.LastError) + } +} + +func TestRepeatersRestoreWithoutRequests(t *testing.T) { + f := newRepeaterSite(t) + store, lists := newRepeaterMemoryStore("YOW"), newZoneListMemory() + newRepeaterImporter(t, f, store, lists, true).refreshN(t, 2) + newRepeaterImporter(t, f, store, lists, true).refreshN(t, 2) + if f.listCalls != 1 || f.calls != 1 { + t.Fatal("restart made requests", f.listCalls, f.calls) + } +} + +func TestRepeatersDisabledClearsListsKeepsTiming(t *testing.T) { + f := newRepeaterSite(t) + store, lists := newRepeaterMemoryStore("YOW"), newZoneListMemory() + newRepeaterImporter(t, f, store, lists, true).refreshN(t, 2) + h := newRepeaterImporter(t, f, store, lists, false) + row := store.rows["YOW"] + if len(store.repeaters) != 0 || len(h.sources) != 0 || h.changes != 1 { + t.Fatal("disabled import kept repeaters or skipped invalidation", store.repeaters, h.changes) + } + if row.cache.NextAttempt.IsZero() || !row.cache.CheckedAt.IsZero() || row.cache.ETag != "" { + t.Fatal("clear dropped timing or kept the list's ETag", row.cache) + } + + // Re-enabled, the saved backoff still applies. + newRepeaterImporter(t, f, store, lists, true).refreshN(t, 2) + if f.calls != 1 { + t.Fatal("re-enable ignored backoff", f.calls) + } +} + +func TestRepeatersRequestCutShortStillCounts(t *testing.T) { + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { <-r.Context().Done() })) + defer server.Close() + store := newRepeaterMemoryStore() + r, err := NewRepeaters(context.Background(), config.MeshMapperRepeatersConfig{Enabled: true}, store, newTestDirectory(newZoneListMemory()), nil) + if err != nil { + t.Fatal(err) + } + s := &repeaterSource{key: "YOW", url: server.URL} + now := time.Now().UTC() + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + _ = r.refresh(ctx, s, now) + if got := s.cache.NextAttempt.Sub(now); got != repeaterRetry { + t.Fatal("abandoned request retried early", got) + } + if row := store.rows["YOW"]; row.cache.NextAttempt.Sub(now) != repeaterRetry || row.cache.LastError != "no response" { + t.Fatal("abandoned request not persisted", row) + } +} From 354dad69258883a0c1af94745d0a92099e20cfd9 Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Mon, 12 Oct 2026 13:31:33 -0400 Subject: [PATCH 2/2] chore: bump version to 2.0.6 --- cmd/beacon/main.go | 2 +- docs/docs.go | 2 +- docs/swagger.json | 2 +- docs/swagger.yaml | 2 +- 4 files changed, 4 insertions(+), 4 deletions(-) diff --git a/cmd/beacon/main.go b/cmd/beacon/main.go index cad97ad..cc1de70 100644 --- a/cmd/beacon/main.go +++ b/cmd/beacon/main.go @@ -45,7 +45,7 @@ import ( var version = "dev" // @title MeshCore Beacon API -// @version 2.0.5 +// @version 2.0.6 // @description MeshCore network observation backend. Ingests LoRa packets from MQTT brokers, stores in PostgreSQL, and streams live events via WebSocket. // @description REST requests share a configurable per-client rate limit (default 300/minute and a 300-request one-second burst cap). Exceeded limits return HTTP 429 with error.code=rate_limited and a Retry-After header in seconds. CORS preflights and WebSocket upgrades do not consume this API budget. // @description WebSocket upgrade attempts at /ws have a configurable per-client limit (default 10/minute, including failed handshakes). Rate exhaustion returns HTTP 429 with Retry-After before upgrade. An accepted socket exceeding the concurrent cap closes with code 1013 before hello; established connections remain open. diff --git a/docs/docs.go b/docs/docs.go index 9991944..601b5db 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -6265,7 +6265,7 @@ const docTemplate = `{ // SwaggerInfo holds exported Swagger Info so clients can modify it var SwaggerInfo = &swag.Spec{ - Version: "2.0.5", + Version: "2.0.6", Host: "localhost:8080", BasePath: "/api/v1", Schemes: []string{"http", "https"}, diff --git a/docs/swagger.json b/docs/swagger.json index f5aee6d..91de39b 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -15,7 +15,7 @@ "license": { "name": "AGPL-3-or-later" }, - "version": "2.0.5" + "version": "2.0.6" }, "host": "localhost:8080", "basePath": "/api/v1", diff --git a/docs/swagger.yaml b/docs/swagger.yaml index 77de161..48561a3 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -1825,7 +1825,7 @@ info: name: AGPL-3-or-later termsOfService: https://github.com/MeshCore-Beacon/beacon-server title: MeshCore Beacon API - version: 2.0.5 + version: 2.0.6 paths: /admin/accounts: get: