Merge pull request #222 from MeshCore-Beacon/dev

release: v2.0.6
This commit is contained in:
MrAlders0n
2026-10-12 13:32:55 -04:00
committed by GitHub
24 changed files with 1910 additions and 17 deletions
+2 -3
View File
@@ -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.
+16 -2
View File
@@ -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.
@@ -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 {
+16
View File
@@ -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:
+62
View File
@@ -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
}
+147
View File
@@ -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)
}
}
@@ -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);
+8
View File
@@ -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
+47 -1
View File
@@ -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;
+72
View File
@@ -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()
+18
View File
@@ -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"`
+10
View File
@@ -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.
+128 -1
View File
@@ -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)
+123 -1
View File
@@ -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"
},
@@ -6143,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"},
+123 -1
View File
@@ -15,7 +15,7 @@
"license": {
"name": "AGPL-3-or-later"
},
"version": "2.0.5"
"version": "2.0.6"
},
"host": "localhost:8080",
"basePath": "/api/v1",
@@ -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"
},
+86 -1
View File
@@ -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:
@@ -1740,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:
+25
View File
@@ -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) {
+45
View File
@@ -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.
+12
View File
@@ -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")
+5
View File
@@ -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) {
+25 -7
View File
@@ -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
+18
View File
@@ -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")
}
}
+5
View File
@@ -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
}
+455
View File
@@ -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
}
+440
View File
@@ -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": "<b>", "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)
}
}