From 1d3f7ce6211bce9a3bb8fc0693eb4d10965c3fac Mon Sep 17 00:00:00 2001 From: n30nex Date: Tue, 13 Oct 2026 04:21:34 -0400 Subject: [PATCH] Add bounded topology snapshots and validate TRACE route evidence --- .github/workflows/ci.yml | 4 +- .github/workflows/docker-publish.yml | 3 +- cmd/beacon/main.go | 3 +- db/packets.go | 23 ++- db/queries/queries.sql | 36 +++- db/sqlc/mock/querier.go | 32 +++- db/sqlc/querier.go | 5 + db/sqlc/queries.sql.go | 162 +++++++++++++++-- db/topology.go | 25 +++ db/topology_integration_test.go | 54 ++++++ db/trace_pagination_integration_test.go | 3 +- db/trace_quality_integration_test.go | 77 ++++++++ db/traces.go | 179 ++++++++++++------ db/traces_integration_test.go | 13 +- db/traces_test.go | 13 +- docs/docs.go | 180 ++++++++++++++++++- docs/swagger.json | 180 ++++++++++++++++++- docs/swagger.yaml | 126 ++++++++++++- internal/api/handlers/routes.go | 1 + internal/api/handlers/scope_catalogues.go | 33 ++++ internal/api/handlers/stub_reader_test.go | 4 + internal/api/handlers/topology.go | 49 +++++ internal/api/handlers/topology_test.go | 33 ++++ internal/api/packets.go | 31 ++-- internal/api/reader.go | 2 + internal/api/router/options.go | 1 + internal/api/router/router.go | 1 + internal/api/router/scope_catalogues_test.go | 28 +++ internal/api/topology.go | 14 ++ internal/api/traces.go | 36 ++-- internal/cache/cache_test.go | 4 + internal/cache/reader.go | 9 + internal/ingest/channel_scope_test.go | 34 ++++ internal/ingest/ingest.go | 1 + internal/ingest/ingest_test.go | 3 + internal/ingest/packet.go | 58 ++++-- internal/ingest/side_effects.go | 2 + internal/ingest/trace_quality_test.go | 75 ++++++++ internal/meshmapper/catalogues.go | 75 ++++++++ internal/meshmapper/catalogues_test.go | 104 +++++++++++ internal/meshmapper/scopes.go | 24 +-- internal/tracequality/quality.go | 96 ++++++++++ internal/tracequality/quality_test.go | 54 ++++++ 43 files changed, 1746 insertions(+), 144 deletions(-) create mode 100644 db/topology.go create mode 100644 db/topology_integration_test.go create mode 100644 db/trace_quality_integration_test.go create mode 100644 internal/api/handlers/scope_catalogues.go create mode 100644 internal/api/handlers/topology.go create mode 100644 internal/api/handlers/topology_test.go create mode 100644 internal/api/router/scope_catalogues_test.go create mode 100644 internal/api/topology.go create mode 100644 internal/ingest/trace_quality_test.go create mode 100644 internal/meshmapper/catalogues.go create mode 100644 internal/meshmapper/catalogues_test.go create mode 100644 internal/tracequality/quality.go create mode 100644 internal/tracequality/quality_test.go diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 95527aa..70a3d68 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -4,7 +4,7 @@ name: CI on: push: - branches: [main, dev] + branches: [main, dev, dev1] pull_request: jobs: @@ -58,7 +58,7 @@ jobs: - name: Verify stats and endpoint queries against PostgreSQL 16 env: BEACON_TEST_POSTGRES_DSN: postgres://postgres:backup-ci-only@127.0.0.1:5432/postgres?sslmode=disable - run: go test ./db -run '^Test(Signal|Paths|PacketSummaries|PacketEndpointsResolveLive|ObserverMetrics|RouteEvidence|RouteEvidenceIndex|AnalyticsRetention|AnalyticsRetentionConcurrent|DeleteOldPacketsBatches|MeshMapperCatalogue|ChannelMessageScopes|NodeLocationReset|RunMigrations[A-Za-z]*|Reconfirm[A-Za-z]*)Postgres$' -count=1 -v + run: go test ./db -run '^Test(Signal|Paths|PacketSummaries|PacketEndpointsResolveLive|ObserverMetrics|RouteEvidence|RouteEvidenceIndex|AnalyticsRetention|AnalyticsRetentionConcurrent|DeleteOldPacketsBatches|MeshMapperCatalogue|ChannelMessageScopes|NodeLocationReset|TraceQuality|TopologyLinks|RunMigrations[A-Za-z]*|Reconfirm[A-Za-z]*)Postgres$' -count=1 -v - name: Verify backup command against PostgreSQL 16 run: | diff --git a/.github/workflows/docker-publish.yml b/.github/workflows/docker-publish.yml index 70f27ec..fae346c 100644 --- a/.github/workflows/docker-publish.yml +++ b/.github/workflows/docker-publish.yml @@ -5,7 +5,7 @@ name: Build and Publish Docker Image on: push: - branches: [main, dev] + branches: [main, dev, dev1] tags: ["v*"] env: @@ -45,6 +45,7 @@ jobs: tags: | type=raw,value=latest,enable=${{ github.ref == 'refs/heads/main' }} type=raw,value=dev,enable=${{ github.ref == 'refs/heads/dev' }} + type=raw,value=dev1,enable=${{ github.ref == 'refs/heads/dev1' }} type=semver,pattern={{version}} type=semver,pattern={{major}}.{{minor}} type=sha diff --git a/cmd/beacon/main.go b/cmd/beacon/main.go index cc1de70..b31dcd9 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.6 +// @version 2.1.0 // @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. @@ -443,6 +443,7 @@ func main() { // CachedReader. Only response projections receive the geographic annotation. reader = api.WithLocalBorders(reader, localBorders) r := router.New(h, reader, []*ingest.Worker{broker1, broker2}, router.Options{ + ScopeCatalogues: scopeImporter, MaxConnsPerIP: resolved.MaxConnsPerIP, MaxConnectsPerMinute: resolved.MaxConnectsPerMinute, WSAllowedOrigins: cfg.WebSocket.AllowedOrigins, diff --git a/db/packets.go b/db/packets.go index 620acce..a142ab6 100644 --- a/db/packets.go +++ b/db/packets.go @@ -15,6 +15,7 @@ import ( sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" "github.com/MeshCore-Beacon/beacon-server/internal/api" "github.com/MeshCore-Beacon/beacon-server/internal/ingest" + "github.com/MeshCore-Beacon/beacon-server/internal/tracequality" "github.com/google/uuid" "github.com/jackc/pgx/v5/pgtype" "github.com/meshcore-go/meshcore-go" @@ -406,7 +407,15 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet, // compute it once rather than per observation. var traceRawHashes [][]byte if row.PayloadType == int16(meshcore.PayloadTypeTrace) { - if trace, err := meshcore.TraceFromBytes(row.RawPayload); err == nil { + quality := tracequality.Assess(row.RawPayload, int(row.PayloadVersion), int(row.RouteType), -1) + for _, obs := range obsRows { + q := tracequality.Assess(row.RawPayload, int(row.PayloadVersion), int(row.RouteType), int(obs.PathLengthByte)) + for _, reason := range q.Reasons { + quality.Add(reason) + } + } + p.TraceQuality = &quality + if trace, err := meshcore.TraceFromBytes(row.RawPayload); err == nil && quality.Status == "supported" { hashSize := int(trace.PathHashSize()) for i := 0; i+hashSize <= len(trace.PathHashes); i += hashSize { traceRawHashes = append(traceRawHashes, trace.PathHashes[i:i+hashSize]) @@ -442,8 +451,11 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet, obs.PropagationTimeMs = &prop resolvedPath := []api.ResolvedHop{} if row.PayloadType == int16(meshcore.PayloadTypeTrace) { + rawPath := hex.EncodeToString(v.PathBytes) + obs.WirePathBytes = &rawPath + obs.TraceQuality = p.TraceQuality if len(traceRawHashes) > 0 { - resolved, err := s.ResolvePathHashes(ctx, v.Iata, traceRawHashes) + resolved, err := s.ResolveTracePathHashes(ctx, v.Iata, traceRawHashes) if err != nil { slog.Error(fmt.Sprintf("store: path resolution failed for observation %d", v.ID), "component", "db", "error", err) } else { @@ -491,7 +503,7 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet, } p.Observations = append(p.Observations, obs) } - if row.PayloadType == 9 && len(obsRows) > 0 { + if row.PayloadType == 9 && len(obsRows) > 0 && p.TraceQuality.Status == "supported" { iatas := make([]string, 0, len(obsRows)) seen := make(map[string]struct{}) for _, v := range obsRows { @@ -503,6 +515,11 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet, var parsed tracePayload if err := json.Unmarshal(row.ParsedPayload, &parsed); err == nil { p.ResolvedRoute = s.resolveTraceRoute(ctx, &parsed, iatas) + for _, hop := range p.ResolvedRoute { + if hop.Confidence == "ambiguous" { + p.TraceQuality.Add("ambiguous_prefix") + } + } } } return p, nil diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 1fb4096..55a31ed 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -455,6 +455,8 @@ WHERE p.packet_hash = $1; -- Return distinct observation IATAs in first-heard order for path resolution, -- without fetching full observations separately for every trace packet. SELECT encode(p.packet_hash, 'hex') AS packet_hash_hex, + p.raw_payload, p.payload_version, + EXISTS(SELECT 1 FROM packet_observations po WHERE po.packet_hash=p.packet_hash AND po.path_length_byte>=64) AS invalid_snr_path, p.route_type, p.first_heard_at, p.last_heard_at, @@ -1429,7 +1431,11 @@ SELECT WHERE ti.trace_tag = t.trace_tag AND (COALESCE(cardinality($1::bpchar[]), 0) = 0 OR ti.iata = ANY($1::bpchar[]))) AS iata_count, COALESCE(t.trace_type, '')::text AS trace_type, - t.best_payload + t.best_payload, + ARRAY(SELECT ti.iata FROM trace_iatas ti WHERE ti.trace_tag=t.trace_tag)::bpchar[] AS iatas, + EXISTS(SELECT 1 FROM packets p WHERE p.trace_tag=t.trace_tag AND p.payload_version<>0) AS unsupported_version, + EXISTS(SELECT 1 FROM packets p WHERE p.trace_tag=t.trace_tag AND p.route_type NOT IN (2,3)) AS non_direct, + EXISTS(SELECT 1 FROM packets p JOIN packet_observations po ON po.packet_hash=p.packet_hash WHERE p.trace_tag=t.trace_tag AND po.path_length_byte>=64) AS invalid_snr_path FROM trace_tags t WHERE (COALESCE(cardinality($1::bpchar[]), 0) = 0 OR EXISTS ( SELECT 1 FROM trace_iatas ti WHERE ti.trace_tag = t.trace_tag AND ti.iata = ANY($1::bpchar[]))) @@ -1920,3 +1926,31 @@ ON CONFLICT (country) DO UPDATE SET -- name: PruneZoneBoundaries :many -- Drops imports for IATAs no longer configured, so their manual border returns. DELETE FROM meshmapper_zone_boundaries WHERE NOT (iata = ANY(@keep::text[])) RETURNING iata; + +-- name: ResolveTraceHashes :many +-- TRACE can terminate at companions. Preserve the full prefix (including eight +-- bytes), and return every candidate across heard regions instead of picking one. +SELECT h.hash::bytea AS hash, si.iata, n.id AS node_id, n.name, n.latitude, n.longitude, n.public_key +FROM unnest(@hashes::bytea[]) AS h(hash) +JOIN node_short_ids si ON si.iata = ANY(@iatas::bpchar[]) AND + ((octet_length(h.hash)=1 AND si.prefix_1=h.hash) OR + (octet_length(h.hash)=2 AND si.prefix_2=h.hash) OR + (octet_length(h.hash) IN (4,8) AND si.prefix_4=substring(h.hash FROM 1 FOR 4))) +JOIN nodes n ON n.id=si.node_id +WHERE substring(n.public_key FROM 1 FOR octet_length(h.hash))=h.hash +ORDER BY si.iata, h.hash, n.id; + +-- name: GetTopologyLinks :many +-- Reduce route history to unique adjacent pairs before transferring it to a browser. +WITH pairs AS ( + SELECT h.node_id AS a, r.node_ids[(h.ord+1)::int] AS b + FROM known_routes r CROSS JOIN LATERAL unnest(r.node_ids) WITH ORDINALITY h(node_id,ord) + WHERE r.last_seen >= @since::timestamptz AND r.last_seen < @until::timestamptz + AND (COALESCE(cardinality(@iatas::bpchar[]),0)=0 OR r.iata=ANY(@iatas::bpchar[])) + AND h.ord < cardinality(r.node_ids) +) +SELECT LEAST(a,b)::uuid AS from_id, GREATEST(a,b)::uuid AS to_id +FROM pairs WHERE a IS NOT NULL AND b IS NOT NULL AND a<>b +GROUP BY LEAST(a,b), GREATEST(a,b) +ORDER BY LEAST(a,b), GREATEST(a,b) +LIMIT @link_limit::int; diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index 6ab1f20..4c5ba29 100644 --- a/db/sqlc/mock/querier.go +++ b/db/sqlc/mock/querier.go @@ -3,7 +3,7 @@ // // Generated by this command: // -// mockgen -source=db/sqlc/querier.go -destination=db/sqlc/mock/querier.go -package=mockdb +// mockgen -source db/sqlc/querier.go -destination db/sqlc/mock/querier.go -package mockdb // // Package mockdb is a generated GoMock package. @@ -1166,6 +1166,21 @@ func (mr *MockQuerierMockRecorder) GetTopNodes(ctx, arg any) *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTopNodes", reflect.TypeOf((*MockQuerier)(nil).GetTopNodes), ctx, arg) } +// GetTopologyLinks mocks base method. +func (m *MockQuerier) GetTopologyLinks(ctx context.Context, arg db.GetTopologyLinksParams) ([]db.GetTopologyLinksRow, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetTopologyLinks", ctx, arg) + ret0, _ := ret[0].([]db.GetTopologyLinksRow) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// GetTopologyLinks indicates an expected call of GetTopologyLinks. +func (mr *MockQuerierMockRecorder) GetTopologyLinks(ctx, arg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTopologyLinks", reflect.TypeOf((*MockQuerier)(nil).GetTopologyLinks), ctx, arg) +} + // GetTransportScopeByName mocks base method. func (m *MockQuerier) GetTransportScopeByName(ctx context.Context, name string) (int32, error) { m.ctrl.T.Helper() @@ -2029,6 +2044,21 @@ func (mr *MockQuerierMockRecorder) ResolveRelayHashPairsP4(ctx, arg any) *gomock return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ResolveRelayHashPairsP4", reflect.TypeOf((*MockQuerier)(nil).ResolveRelayHashPairsP4), ctx, arg) } +// ResolveTraceHashes mocks base method. +func (m *MockQuerier) ResolveTraceHashes(ctx context.Context, arg db.ResolveTraceHashesParams) ([]db.ResolveTraceHashesRow, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ResolveTraceHashes", ctx, arg) + ret0, _ := ret[0].([]db.ResolveTraceHashesRow) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// ResolveTraceHashes indicates an expected call of ResolveTraceHashes. +func (mr *MockQuerierMockRecorder) ResolveTraceHashes(ctx, arg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ResolveTraceHashes", reflect.TypeOf((*MockQuerier)(nil).ResolveTraceHashes), ctx, arg) +} + // RollAdvertHearings mocks base method. func (m *MockQuerier) RollAdvertHearings(ctx context.Context, hour pgtype.Timestamptz) error { m.ctrl.T.Helper() diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index 1135eaa..472aae8 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -162,6 +162,8 @@ type Querier interface { // Nodes by ADVERT hearings since the given hour. Live names win over the rolled snapshot; // node_id is NULL once the node row is gone. iata is a representative one. GetTopNodes(ctx context.Context, arg GetTopNodesParams) ([]GetTopNodesRow, error) + // Reduce route history to unique adjacent pairs before transferring it to a browser. + GetTopologyLinks(ctx context.Context, arg GetTopologyLinksParams) ([]GetTopologyLinksRow, error) GetTransportScopeByName(ctx context.Context, name string) (int32, error) GetTransportScopes(ctx context.Context) ([]GetTransportScopesRow, error) // ============================================================ @@ -335,6 +337,9 @@ type Querier interface { ResolveRelayHashPairsP3(ctx context.Context, arg ResolveRelayHashPairsP3Params) ([]ResolveRelayHashPairsP3Row, error) // Batch form of ResolvePathHashesP4 across IATAs; callers check uniqueness per IATA. ResolveRelayHashPairsP4(ctx context.Context, arg ResolveRelayHashPairsP4Params) ([]ResolveRelayHashPairsP4Row, error) + // TRACE can terminate at companions. Preserve the full prefix (including eight + // bytes), and return every candidate across heard regions instead of picking one. + ResolveTraceHashes(ctx context.Context, arg ResolveTraceHashesParams) ([]ResolveTraceHashesRow, error) RollAdvertHearings(ctx context.Context, hour pgtype.Timestamptz) error // Each distinct ADVERT packet counts once, under the exact set of IATAs that heard it this hour. RollAdvertSets(ctx context.Context, hour pgtype.Timestamptz) error diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 7ea8cab..efffc14 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -1422,6 +1422,8 @@ func (q *Queries) GetPacketByHash(ctx context.Context, packetHash []byte) (GetPa const getPacketsByTraceTag = `-- name: GetPacketsByTraceTag :many SELECT encode(p.packet_hash, 'hex') AS packet_hash_hex, + p.raw_payload, p.payload_version, + EXISTS(SELECT 1 FROM packet_observations po WHERE po.packet_hash=p.packet_hash AND po.path_length_byte>=64) AS invalid_snr_path, p.route_type, p.first_heard_at, p.last_heard_at, @@ -1442,14 +1444,17 @@ ORDER BY p.first_heard_at ASC ` type GetPacketsByTraceTagRow struct { - PacketHashHex string `json:"packet_hash_hex"` - RouteType int16 `json:"route_type"` - FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` - LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` - ParsedPayload []byte `json:"parsed_payload"` - ScopeID *int32 `json:"scope_id"` - ScopeName *string `json:"scope_name"` - Iatas []string `json:"iatas"` + PacketHashHex string `json:"packet_hash_hex"` + RawPayload []byte `json:"raw_payload"` + PayloadVersion int16 `json:"payload_version"` + InvalidSnrPath bool `json:"invalid_snr_path"` + RouteType int16 `json:"route_type"` + FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` + LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` + ParsedPayload []byte `json:"parsed_payload"` + ScopeID *int32 `json:"scope_id"` + ScopeName *string `json:"scope_name"` + Iatas []string `json:"iatas"` } // Return distinct observation IATAs in first-heard order for path resolution, @@ -1465,6 +1470,9 @@ func (q *Queries) GetPacketsByTraceTag(ctx context.Context, decode string) ([]Ge var i GetPacketsByTraceTagRow if err := rows.Scan( &i.PacketHashHex, + &i.RawPayload, + &i.PayloadVersion, + &i.InvalidSnrPath, &i.RouteType, &i.FirstHeardAt, &i.LastHeardAt, @@ -2254,6 +2262,59 @@ func (q *Queries) GetTopNodes(ctx context.Context, arg GetTopNodesParams) ([]Get return items, nil } +const getTopologyLinks = `-- name: GetTopologyLinks :many +WITH pairs AS ( + SELECT h.node_id AS a, r.node_ids[(h.ord+1)::int] AS b + FROM known_routes r CROSS JOIN LATERAL unnest(r.node_ids) WITH ORDINALITY h(node_id,ord) + WHERE r.last_seen >= $2::timestamptz AND r.last_seen < $3::timestamptz + AND (COALESCE(cardinality($4::bpchar[]),0)=0 OR r.iata=ANY($4::bpchar[])) + AND h.ord < cardinality(r.node_ids) +) +SELECT LEAST(a,b)::uuid AS from_id, GREATEST(a,b)::uuid AS to_id +FROM pairs WHERE a IS NOT NULL AND b IS NOT NULL AND a<>b +GROUP BY LEAST(a,b), GREATEST(a,b) +ORDER BY LEAST(a,b), GREATEST(a,b) +LIMIT $1::int +` + +type GetTopologyLinksParams struct { + LinkLimit int32 `json:"link_limit"` + Since pgtype.Timestamptz `json:"since"` + Until pgtype.Timestamptz `json:"until"` + Iatas []string `json:"iatas"` +} + +type GetTopologyLinksRow struct { + FromID uuid.UUID `json:"from_id"` + ToID uuid.UUID `json:"to_id"` +} + +// Reduce route history to unique adjacent pairs before transferring it to a browser. +func (q *Queries) GetTopologyLinks(ctx context.Context, arg GetTopologyLinksParams) ([]GetTopologyLinksRow, error) { + rows, err := q.db.Query(ctx, getTopologyLinks, + arg.LinkLimit, + arg.Since, + arg.Until, + arg.Iatas, + ) + if err != nil { + return nil, err + } + defer rows.Close() + items := []GetTopologyLinksRow{} + for rows.Next() { + var i GetTopologyLinksRow + if err := rows.Scan(&i.FromID, &i.ToID); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const getTransportScopeByName = `-- name: GetTransportScopeByName :one SELECT id FROM transport_scopes WHERE name = $1 ` @@ -4194,7 +4255,11 @@ SELECT WHERE ti.trace_tag = t.trace_tag AND (COALESCE(cardinality($1::bpchar[]), 0) = 0 OR ti.iata = ANY($1::bpchar[]))) AS iata_count, COALESCE(t.trace_type, '')::text AS trace_type, - t.best_payload + t.best_payload, + ARRAY(SELECT ti.iata FROM trace_iatas ti WHERE ti.trace_tag=t.trace_tag)::bpchar[] AS iatas, + EXISTS(SELECT 1 FROM packets p WHERE p.trace_tag=t.trace_tag AND p.payload_version<>0) AS unsupported_version, + EXISTS(SELECT 1 FROM packets p WHERE p.trace_tag=t.trace_tag AND p.route_type NOT IN (2,3)) AS non_direct, + EXISTS(SELECT 1 FROM packets p JOIN packet_observations po ON po.packet_hash=p.packet_hash WHERE p.trace_tag=t.trace_tag AND po.path_length_byte>=64) AS invalid_snr_path FROM trace_tags t WHERE (COALESCE(cardinality($1::bpchar[]), 0) = 0 OR EXISTS ( SELECT 1 FROM trace_iatas ti WHERE ti.trace_tag = t.trace_tag AND ti.iata = ANY($1::bpchar[]))) @@ -4222,13 +4287,17 @@ type ListTraceTagsParams struct { } type ListTraceTagsRow struct { - TraceTag string `json:"trace_tag"` - FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` - LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` - PacketCount int64 `json:"packet_count"` - IataCount int64 `json:"iata_count"` - TraceType string `json:"trace_type"` - BestPayload []byte `json:"best_payload"` + TraceTag string `json:"trace_tag"` + FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` + LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` + PacketCount int64 `json:"packet_count"` + IataCount int64 `json:"iata_count"` + TraceType string `json:"trace_type"` + BestPayload []byte `json:"best_payload"` + Iatas []string `json:"iatas"` + UnsupportedVersion bool `json:"unsupported_version"` + NonDirect bool `json:"non_direct"` + InvalidSnrPath bool `json:"invalid_snr_path"` } // ============================================================ @@ -4261,6 +4330,10 @@ func (q *Queries) ListTraceTags(ctx context.Context, arg ListTraceTagsParams) ([ &i.IataCount, &i.TraceType, &i.BestPayload, + &i.Iatas, + &i.UnsupportedVersion, + &i.NonDirect, + &i.InvalidSnrPath, ); err != nil { return nil, err } @@ -5103,6 +5176,63 @@ func (q *Queries) ResolveRelayHashPairsP4(ctx context.Context, arg ResolveRelayH return items, nil } +const resolveTraceHashes = `-- name: ResolveTraceHashes :many +SELECT h.hash::bytea AS hash, si.iata, n.id AS node_id, n.name, n.latitude, n.longitude, n.public_key +FROM unnest($1::bytea[]) AS h(hash) +JOIN node_short_ids si ON si.iata = ANY($2::bpchar[]) AND + ((octet_length(h.hash)=1 AND si.prefix_1=h.hash) OR + (octet_length(h.hash)=2 AND si.prefix_2=h.hash) OR + (octet_length(h.hash) IN (4,8) AND si.prefix_4=substring(h.hash FROM 1 FOR 4))) +JOIN nodes n ON n.id=si.node_id +WHERE substring(n.public_key FROM 1 FOR octet_length(h.hash))=h.hash +ORDER BY si.iata, h.hash, n.id +` + +type ResolveTraceHashesParams struct { + Hashes [][]byte `json:"hashes"` + Iatas []string `json:"iatas"` +} + +type ResolveTraceHashesRow struct { + Hash []byte `json:"hash"` + Iata string `json:"iata"` + NodeID uuid.UUID `json:"node_id"` + Name *string `json:"name"` + Latitude *float64 `json:"latitude"` + Longitude *float64 `json:"longitude"` + PublicKey []byte `json:"public_key"` +} + +// TRACE can terminate at companions. Preserve the full prefix (including eight +// bytes), and return every candidate across heard regions instead of picking one. +func (q *Queries) ResolveTraceHashes(ctx context.Context, arg ResolveTraceHashesParams) ([]ResolveTraceHashesRow, error) { + rows, err := q.db.Query(ctx, resolveTraceHashes, arg.Hashes, arg.Iatas) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ResolveTraceHashesRow{} + for rows.Next() { + var i ResolveTraceHashesRow + if err := rows.Scan( + &i.Hash, + &i.Iata, + &i.NodeID, + &i.Name, + &i.Latitude, + &i.Longitude, + &i.PublicKey, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const saveChannelCatalogue = `-- name: SaveChannelCatalogue :exec INSERT INTO meshmapper_channel_catalogues (iata, url, payload, etag, checked_at, attempted_at, next_attempt, last_error) VALUES ($1, $2, $3::jsonb, $4::text, diff --git a/db/topology.go b/db/topology.go new file mode 100644 index 0000000..24c45cc --- /dev/null +++ b/db/topology.go @@ -0,0 +1,25 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later +package db + +import ( + "context" + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgtype" + "time" +) + +func (s *Store) GetTopologyLinks(ctx context.Context, iatas []string, since, until time.Time) (*api.TopologyLinks, error) { + const limit = 100000 + rows, err := s.q.GetTopologyLinks(ctx, sqlc.GetTopologyLinksParams{Iatas: iatas, Since: pgtype.Timestamptz{Time: since, Valid: true}, Until: pgtype.Timestamptz{Time: until, Valid: true}, LinkLimit: limit + 1}) + if err != nil { + return nil, err + } + result := &api.TopologyLinks{Links: make([][2]uuid.UUID, 0, min(len(rows), limit)), Capped: len(rows) > limit, Since: since.UnixMilli(), Until: until.UnixMilli()} + for _, r := range rows[:min(len(rows), limit)] { + result.Links = append(result.Links, [2]uuid.UUID{r.FromID, r.ToID}) + } + return result, nil +} diff --git a/db/topology_integration_test.go b/db/topology_integration_test.go new file mode 100644 index 0000000..58b4c57 --- /dev/null +++ b/db/topology_integration_test.go @@ -0,0 +1,54 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later +package db + +import ( + "context" + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/google/uuid" + "testing" + "time" +) + +func TestTopologyLinksPostgres(t *testing.T) { + ctx, pool := schemaPool(t) + if err := RunMigrations(ctx, pool); err != nil { + t.Fatal(err) + } + tx, err := pool.Begin(ctx) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback(context.Background()) + if _, err = tx.Exec(ctx, "CREATE TEMP TABLE known_routes (LIKE public.known_routes INCLUDING ALL) ON COMMIT DROP"); err != nil { + t.Fatal(err) + } + ids := []uuid.UUID{uuid.New(), uuid.New(), uuid.New()} + now := time.Now().UTC().Truncate(time.Minute) + // Thousands of duplicate/reversed routes transfer only their unique adjacent pairs. + _, err = tx.Exec(ctx, `INSERT INTO known_routes (path_key,node_ids,hash_prefix,iata,hop_count,first_seen,last_seen) + SELECT decode(md5(i::text),'hex'),CASE WHEN i%2=0 THEN $1::uuid[] ELSE $2::uuid[] END,ARRAY['\xaa'::bytea,'\xbb'::bytea], + 'YKF',2,$3,$3 FROM generate_series(1,1000) i`, ids[:2], []uuid.UUID{ids[1], ids[0]}, now) + if err != nil { + t.Fatal(err) + } + _, err = tx.Exec(ctx, `INSERT INTO known_routes(path_key,node_ids,hash_prefix,iata,hop_count,first_seen,last_seen) VALUES + (decode(md5('other'),'hex'),$1,ARRAY['\xbb'::bytea,'\xcc'::bytea],'YOW',2,$3,$3), + (decode(md5('gap'),'hex'),$2,ARRAY['\xaa'::bytea,'\x00'::bytea,'\xcc'::bytea],'YKF',3,$3,$3), + (decode(md5('old'),'hex'),$1,ARRAY['\xbb'::bytea,'\xcc'::bytea],'YKF',2,$3-interval '25 hours',$3-interval '25 hours')`, ids[1:], []*uuid.UUID{&ids[0], nil, &ids[2]}, now) + if err != nil { + t.Fatal(err) + } + store := &Store{q: sqlc.New(tx)} + result, err := store.GetTopologyLinks(ctx, []string{"YKF"}, now.Add(-24*time.Hour), now.Add(time.Minute)) + if err != nil || result.Capped || len(result.Links) != 1 { + t.Fatalf("scoped: %+v %v", result, err) + } + all, err := store.GetTopologyLinks(ctx, nil, now.Add(-24*time.Hour), now.Add(time.Minute)) + if err != nil || len(all.Links) != 2 { + t.Fatalf("all regions: %+v %v", all, err) + } + if result.Since != now.Add(-24*time.Hour).UnixMilli() { + t.Fatal("window changed") + } +} diff --git a/db/trace_pagination_integration_test.go b/db/trace_pagination_integration_test.go index 90ed19d..dc123a4 100644 --- a/db/trace_pagination_integration_test.go +++ b/db/trace_pagination_integration_test.go @@ -16,6 +16,7 @@ import ( sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" "github.com/MeshCore-Beacon/beacon-server/internal/api" "github.com/MeshCore-Beacon/beacon-server/internal/ingest" + "github.com/MeshCore-Beacon/beacon-server/internal/tracequality" "github.com/jackc/pgx/v5" ) @@ -89,7 +90,7 @@ func TestTraceTagPaginationPostgres(t *testing.T) { paths := map[string][]string{"a": {"aa", "bb", "cc"}, "b": {"bb", "11", "22", "33"}, "c": {"cc"}, "d": {"dd"}} snrs := map[string][]float32{"a": {1, 2, 3}, "b": {2, 3, 4, 5}, "c": {3}, "d": {4}} summary := func(tag string, first, last int, count, iatas int64, kind string) api.TraceTagSummary { - return api.TraceTagSummary{TraceTag: strings.Repeat(tag, 8), FirstHeardAt: at(first).UnixMilli(), LastHeardAt: at(last).UnixMilli(), PacketCount: count, IATACount: iatas, TraceType: kind, PathHashes: paths[tag], SNRValues: snrs[tag]} + return api.TraceTagSummary{Quality: tracequality.Quality{Status: "suspect", Reasons: []string{"non_direct_trace"}}, TraceTag: strings.Repeat(tag, 8), FirstHeardAt: at(first).UnixMilli(), LastHeardAt: at(last).UnixMilli(), PacketCount: count, IATACount: iatas, TraceType: kind, PathHashes: paths[tag], SNRValues: snrs[tag]} } a, b, c, d := summary("a", 1, 40, 3, 2, "TRACE"), summary("b", 4, 30, 2, 1, "TRACE"), summary("c", 6, 20, 1, 1, "TRACE"), summary("d", 7, 15, 1, 1, "PING") for _, tc := range []struct { diff --git a/db/trace_quality_integration_test.go b/db/trace_quality_integration_test.go new file mode 100644 index 0000000..d623ab6 --- /dev/null +++ b/db/trace_quality_integration_test.go @@ -0,0 +1,77 @@ +package db + +import ( + "context" + "encoding/hex" + "os" + "testing" + + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/jackc/pgx/v5" +) + +func TestTraceQualityPostgres(t *testing.T) { + dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN") + if dsn == "" { + t.Skip("set BEACON_TEST_POSTGRES_DSN") + } + ctx := context.Background() + conn, err := pgx.Connect(ctx, dsn) + if err != nil { + t.Fatal(err) + } + defer conn.Close(ctx) + tx, err := conn.Begin(ctx) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback(ctx) + for _, name := range []string{"nodes", "node_short_ids"} { + if _, err := tx.Exec(ctx, "CREATE TEMP TABLE "+name+" (LIKE public."+name+" INCLUDING ALL) ON COMMIT DROP"); err != nil { + t.Fatal(err) + } + } + _, err = tx.Exec(ctx, ` + INSERT INTO nodes (id,public_key,node_type,name) VALUES + ('00000000-0000-0000-0000-000000000001',decode('aabbccdd11223344'||repeat('00',24),'hex'),1,'Companion'), + ('00000000-0000-0000-0000-000000000002',decode('aabbccdd55667788'||repeat('00',24),'hex'),2,'Repeater'); + INSERT INTO node_short_ids (node_id,iata,prefix_4) VALUES + ('00000000-0000-0000-0000-000000000001','YKF','\xaabbccdd'), + ('00000000-0000-0000-0000-000000000002','YKF','\xaabbccdd'), + ('00000000-0000-0000-0000-000000000001','YHM','\xaabbccdd');`) + if err != nil { + t.Fatal(err) + } + store := &Store{q: sqlc.New(tx)} + for _, width := range []int{1, 2, 4, 8} { + full := "aabbccdd11223344" + h := full[:width*2] + b, _ := hex.DecodeString(h) + candidates, err := store.traceCandidates(ctx, []string{"YKF", "YHM"}, [][]byte{b}) + if err != nil { + t.Fatal(err) + } + flags := byte(0) + switch width { + case 2: + flags = 1 + case 4: + flags = 2 + case 8: + flags = 3 + } + route := traceRoute(&tracePayload{Flags: flags, PathHashes: []string{h}}, []string{"YKF", "YHM"}, candidates) + count := 2 + confidence := "ambiguous" + if width == 8 { + count = 1 + confidence = "high" + } + if len(route) != 1 || len(route[0].Nodes) != count || route[0].Confidence != confidence { + t.Fatalf("width %d: %+v", width, route) + } + if route[0].Nodes[0].PublicKey[:16] != "aabbccdd11223344" { + t.Fatal("companion excluded or eight-byte hash truncated") + } + } +} diff --git a/db/traces.go b/db/traces.go index 93ed18c..cc933b1 100644 --- a/db/traces.go +++ b/db/traces.go @@ -8,6 +8,8 @@ import ( "encoding/hex" "encoding/json" "fmt" + "github.com/MeshCore-Beacon/beacon-server/internal/tracequality" + "github.com/google/uuid" "time" sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" @@ -17,9 +19,11 @@ import ( ) type tracePayload struct { - PathHashes []string `json:"pathHashes"` - Flags byte `json:"flags"` - SNRValues []float32 `json:"snrValues"` + Raw string `json:"raw"` + Quality tracequality.Quality `json:"quality"` + PathHashes []string `json:"pathHashes"` + Flags byte `json:"flags"` + SNRValues []float32 `json:"snrValues"` } func (s *Store) RecordTrace(ctx context.Context, h ingest.TraceHearing) error { @@ -71,12 +75,41 @@ func (s *Store) ListTraceTags(ctx context.Context, iatas []string, scope, traceT return nil, err } items := make([]api.TraceTagSummary, 0, len(rows)) + payloads := make([]tracePayload, 0, len(rows)) + allIATAs := map[string]bool{} + allHashes := map[string][]byte{} for _, r := range rows { var best tracePayload if len(r.BestPayload) > 0 { _ = json.Unmarshal(r.BestPayload, &best) } + quality := tracequality.Stored(best.Raw, best.Flags, best.PathHashes, best.SNRValues) + for _, reason := range best.Quality.Reasons { + if reason != "ambiguous_prefix" { + quality.Add(reason) + } + } + if r.UnsupportedVersion { + quality.Add("unsupported_version") + } + if r.InvalidSnrPath { + quality.Add("invalid_snr_path") + } + if r.NonDirect { + quality.Add("non_direct_trace") + } + payloads = append(payloads, best) + if quality.Status == "supported" { + for _, iata := range r.Iatas { + allIATAs[iata] = true + } + for _, h := range best.PathHashes { + b, _ := hex.DecodeString(h) + allHashes[h] = b + } + } items = append(items, api.TraceTagSummary{ + Quality: quality, TraceTag: r.TraceTag, FirstHeardAt: r.FirstHeardAt.Time.UnixMilli(), LastHeardAt: r.LastHeardAt.Time.UnixMilli(), @@ -87,6 +120,30 @@ func (s *Store) ListTraceTags(ctx context.Context, iatas []string, scope, traceT SNRValues: best.SNRValues, }) } + // One bounded candidate query for the entire page, including all hash widths. + regions := make([]string, 0, len(allIATAs)) + for iata := range allIATAs { + regions = append(regions, iata) + } + hashes := make([][]byte, 0, len(allHashes)) + for _, h := range allHashes { + hashes = append(hashes, h) + } + candidates, err := s.traceCandidates(ctx, regions, hashes) + if err != nil { + return nil, err + } + for i := range items { + if items[i].Quality.Status != "supported" { + continue + } + for _, hop := range traceRoute(&payloads[i], rows[i].Iatas, candidates) { + if hop.Confidence == "ambiguous" { + items[i].Quality.Add("ambiguous_prefix") + break + } + } + } return items, nil } @@ -125,76 +182,94 @@ func (s *Store) GetTraceByTag(ctx context.Context, tag string) (*api.TraceDetail } packet.RawPath = rawPath } - if len(r.Iatas) > 0 { + packet.Quality = tracequality.Assess(r.RawPayload, int(r.PayloadVersion), int(r.RouteType), len(parsed.SNRValues)) + if r.InvalidSnrPath { + packet.Quality.Add("invalid_snr_path") + } + stored := tracequality.Stored(parsed.Raw, parsed.Flags, parsed.PathHashes, parsed.SNRValues) + for _, reason := range stored.Reasons { + packet.Quality.Add(reason) + } + if len(r.Iatas) > 0 && packet.Quality.Status == "supported" { packet.ResolvedRoute = s.resolveTraceRoute(ctx, &parsed, r.Iatas) } + for _, hop := range packet.ResolvedRoute { + if hop.Confidence == "ambiguous" { + packet.Quality.Add("ambiguous_prefix") + } + } detail.Packets = append(detail.Packets, packet) } return detail, nil } +// Candidates remain separated by hearing region until a trace is assembled. +// Never replace an ambiguous region with a convenient singleton from another. +type traceCandidateMap map[string]map[string][]api.ResolvedPathEntry + +func (s *Store) traceCandidates(ctx context.Context, iatas []string, hashes [][]byte) (traceCandidateMap, error) { + out := traceCandidateMap{} + if len(iatas) == 0 || len(hashes) == 0 { + return out, nil + } + rows, err := s.q.ResolveTraceHashes(ctx, sqlc.ResolveTraceHashesParams{Iatas: iatas, Hashes: hashes}) + if err != nil { + return nil, err + } + for _, r := range rows { + if out[r.Iata] == nil { + out[r.Iata] = map[string][]api.ResolvedPathEntry{} + } + h := hex.EncodeToString(r.Hash) + out[r.Iata][h] = append(out[r.Iata][h], api.ResolvedPathEntry{NodeID: r.NodeID, Name: r.Name, Latitude: r.Latitude, Longitude: r.Longitude, PublicKey: r.PublicKey}) + } + return out, nil +} + +func (s *Store) ResolveTracePathHashes(ctx context.Context, iata string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error) { + candidates, err := s.traceCandidates(ctx, []string{iata}, hashes) + return candidates[iata], err +} + func (s *Store) resolveTraceRoute(ctx context.Context, payload *tracePayload, iatas []string) []api.ResolvedHop { - if payload == nil || len(payload.PathHashes) == 0 { + if payload == nil || tracequality.Stored(payload.Raw, payload.Flags, payload.PathHashes, payload.SNRValues).Status != "supported" { return nil } - hashSize := int(1 << (payload.Flags & 0x03)) hashes := make([][]byte, 0, len(payload.PathHashes)) for _, h := range payload.PathHashes { - b, err := hex.DecodeString(h) - if err == nil { - hashes = append(hashes, b) - } + b, _ := hex.DecodeString(h) + hashes = append(hashes, b) } - confidenceRank := map[string]int{"none": 0, "ambiguous": 1, "high": 2} - type hopResult struct { - confidence string - entries []api.ResolvedPathEntry + candidates, err := s.traceCandidates(ctx, iatas, hashes) + if err != nil { + return nil } - merged := make([]hopResult, len(hashes)) - for i := range merged { - merged[i] = hopResult{confidence: "none"} - } - for _, iata := range iatas { - resolved, err := s.ResolvePathHashes(ctx, iata, hashes) - if err != nil { - continue - } - for i, hash := range hashes { - key := hex.EncodeToString(hash[:hashSize]) - entries := resolved[key] - var confidence string - switch len(entries) { - case 0: - confidence = "none" - case 1: - confidence = "high" - default: - confidence = "ambiguous" - } - if confidenceRank[confidence] > confidenceRank[merged[i].confidence] { - merged[i] = hopResult{confidence: confidence, entries: entries} + return traceRoute(payload, iatas, candidates) +} + +func traceRoute(payload *tracePayload, iatas []string, candidates traceCandidateMap) []api.ResolvedHop { + route := make([]api.ResolvedHop, 0, len(payload.PathHashes)) + for i, h := range payload.PathHashes { + hop := api.ResolvedHop{Confidence: "none", Nodes: []api.ResolvedNode{}} + seen := map[uuid.UUID]bool{} + for _, iata := range iatas { + for _, e := range candidates[iata][h] { + if seen[e.NodeID] { + continue + } + seen[e.NodeID] = true + hop.Nodes = append(hop.Nodes, api.ResolvedNode{ID: e.NodeID, Name: e.Name, Latitude: e.Latitude, Longitude: e.Longitude, PublicKey: hex.EncodeToString(e.PublicKey)}) } } - } - route := make([]api.ResolvedHop, 0, len(hashes)) - for i, hr := range merged { - hop := api.ResolvedHop{ - Confidence: hr.confidence, - Nodes: make([]api.ResolvedNode, 0, len(hr.entries)), + if len(hop.Nodes) == 1 { + hop.Confidence = "high" + } else if len(hop.Nodes) > 1 { + hop.Confidence = "ambiguous" } if i < len(payload.SNRValues) { snr := payload.SNRValues[i] hop.SNR = &snr } - for _, e := range hr.entries { - hop.Nodes = append(hop.Nodes, api.ResolvedNode{ - ID: e.NodeID, - Name: e.Name, - Latitude: e.Latitude, - Longitude: e.Longitude, - PublicKey: hex.EncodeToString(e.PublicKey), - }) - } route = append(route, hop) } return route diff --git a/db/traces_integration_test.go b/db/traces_integration_test.go index 5130544..1d671af 100644 --- a/db/traces_integration_test.go +++ b/db/traces_integration_test.go @@ -59,7 +59,7 @@ INSERT INTO node_short_ids (node_id,iata,prefix_4) VALUES ('00000000-0000-0000-0000-000000000001','YVR','\xaa000000'), ('00000000-0000-0000-0000-000000000002','YYJ','\xaa010000'); INSERT INTO packets (packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,trace_tag,parsed_payload,first_heard_at,last_heard_at) -SELECT decode(lpad(to_hex(i),64,'0'),'hex'),9,0,1,'\x00','\x00','\x01020304', +SELECT decode(lpad(to_hex(i),64,'0'),'hex'),9,0,2,'\x010203040000000000aa','\x26','\x01020304', CASE WHEN i=100 THEN '{}'::jsonb ELSE '{"flags":0,"pathHashes":["aa"],"snrValues":[12.5]}'::jsonb END, '2026-01-01 00:00:00+00'::timestamptz+i*interval '1 second', '2026-01-01 00:00:00+00'::timestamptz+i*interval '1 second'+interval '1 second' FROM generate_series(1,100) i; @@ -89,7 +89,7 @@ ANALYZE packets; ANALYZE packet_observations; ANALYZE nodes; ANALYZE node_short_ anchor := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) for i, p := range detail.Packets { number := i + 1 - if p.PacketHash != fmt.Sprintf("%064x", number) || p.RouteType != 1 || p.FirstHeardAt != anchor.Add(time.Duration(number)*time.Second).UnixMilli() || p.LastHeardAt != p.FirstHeardAt+1000 { + if p.PacketHash != fmt.Sprintf("%064x", number) || p.RouteType != 2 || p.FirstHeardAt != anchor.Add(time.Duration(number)*time.Second).UnixMilli() || p.LastHeardAt != p.FirstHeardAt+1000 { t.Fatalf("wrong packet fields or order at %d", number) } if number == 100 { @@ -109,7 +109,10 @@ ANALYZE packets; ANALYZE packet_observations; ANALYZE nodes; ANALYZE node_short_ } want := "YVR relay" if number == 1 { - want = "YYJ relay" + if len(p.ResolvedRoute) != 1 || p.ResolvedRoute[0].Confidence != "ambiguous" || len(p.ResolvedRoute[0].Nodes) != 2 || p.Quality.Status != "ambiguous" { + t.Fatal("conflicting regions must retain both candidates") + } + continue } if len(p.ResolvedRoute) != 1 || p.ResolvedRoute[0].Confidence != "high" || len(p.ResolvedRoute[0].Nodes) != 1 || p.ResolvedRoute[0].Nodes[0].Name == nil || *p.ResolvedRoute[0].Nodes[0].Name != want { t.Fatalf("wrong region precedence/resolution at %d", number) @@ -122,8 +125,8 @@ ANALYZE packets; ANALYZE packet_observations; ANALYZE nodes; ANALYZE node_short_ if counter.observations != 0 { t.Errorf("extra observation queries: %d; want 0", counter.observations) } - if counter.queries != 100 { - t.Errorf("got %d queries; want one packet query and 99 unchanged path-resolution queries", counter.queries) + if counter.queries != 99 { + t.Errorf("got %d queries; want one packet query and 98 batched path-resolution queries", counter.queries) } counter.queries, counter.observations = 0, 0 missing, err := store.GetTraceByTag(ctx, "00000000") diff --git a/db/traces_test.go b/db/traces_test.go index 0e3d4b4..ef0a441 100644 --- a/db/traces_test.go +++ b/db/traces_test.go @@ -99,7 +99,7 @@ func TestGetTraceByTag_WithPacket(t *testing.T) { firstHeard := pgtype.Timestamptz{Time: time.UnixMilli(1700000000000), Valid: true} lastHeard := pgtype.Timestamptz{Time: time.UnixMilli(1700000001000), Valid: true} // aabbccdd is valid hex for the packet hash - parsedPayload := []byte(`{"pathHashes":["aabb"],"snrValues":[15.0],"flags":0}`) + parsedPayload := []byte(`{"pathHashes":["aabb"],"snrValues":[15.0],"flags":1}`) scopeName := "default" mock.EXPECT(). @@ -107,7 +107,8 @@ func TestGetTraceByTag_WithPacket(t *testing.T) { Return([]sqlc.GetPacketsByTraceTagRow{ { PacketHashHex: "aabbccdd", - RouteType: 1, + RouteType: 2, + RawPayload: []byte{0, 0, 0, 0, 0, 0, 0, 0, 1, 0xaa, 0xbb}, ScopeName: &scopeName, FirstHeardAt: firstHeard, LastHeardAt: lastHeard, @@ -117,11 +118,11 @@ func TestGetTraceByTag_WithPacket(t *testing.T) { }, nil) mock.EXPECT(). - ResolvePathHashesP2(gomock.Any(), sqlc.ResolvePathHashesP2Params{ - Iata: "YVR", - Column2: [][]byte{{0xaa, 0xbb}}, + ResolveTraceHashes(gomock.Any(), sqlc.ResolveTraceHashesParams{ + Iatas: []string{"YVR"}, + Hashes: [][]byte{{0xaa, 0xbb}}, }). - Return([]sqlc.ResolvePathHashesP2Row{}, nil) + Return([]sqlc.ResolveTraceHashesRow{}, nil) store := &Store{q: mock} detail, err := store.GetTraceByTag(context.Background(), "trace-001") diff --git a/docs/docs.go b/docs/docs.go index 601b5db..07a6809 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -2343,6 +2343,52 @@ const docTemplate = `{ } } }, + "/routes/topology": { + "get": { + "description": "Returns up to 100000 undirected node-ID pairs, deduplicated across known routes. Minute-aligned window includes the current minute. Existing rate limits apply; no per-node lookup or route pagination is needed. Capped is true when more links exist.", + "produces": [ + "application/json" + ], + "tags": [ + "Routes" + ], + "summary": "Unique adjacent route links for a topology snapshot", + "parameters": [ + { + "type": "string", + "description": "History window: 15m (default), 1h, or 24h", + "name": "window", + "in": "query" + }, + { + "type": "string", + "description": "Comma-separated reception IATA codes", + "name": "iatas", + "in": "query" + } + ], + "responses": { + "200": { + "description": "OK", + "schema": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.TopologyLinks" + } + }, + "400": { + "description": "Bad Request", + "schema": { + "$ref": "#/definitions/internal_api_handlers.APIError" + } + }, + "503": { + "description": "Service Unavailable", + "schema": { + "$ref": "#/definitions/internal_api_handlers.APIError" + } + } + } + } + }, "/routes/{iata}/{pathKey}/observations": { "get": { "description": "Matches the saved IATA, complete path bytes and hash width. Excludes TRACE and unclassified reports; other widths and search-result subsegments are not included. Byte matches do not prove hop identities or delivery. The route and its historical counter can outlive raw evidence. Cursors retain full timestamp precision and pin route/window scope.", @@ -2429,6 +2475,29 @@ const docTemplate = `{ } } }, + "/scope-catalogues": { + "get": { + "description": "Discovered sources only. Counts describe the source region, not individual nodes or links. No upstream request is made. An empty list means no configured importer; checkedAt=0 means no successful check. lastError or a past freshUntil marks stale metadata. Manual Beacon scopes and packet evidence are separate.", + "produces": [ + "application/json" + ], + "tags": [ + "Scopes" + ], + "summary": "Cached MeshMapper regional scope catalogues", + "responses": { + "200": { + "description": "OK", + "schema": { + "type": "array", + "items": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.PublicScopeCatalogue" + } + } + } + } + } + }, "/scopes": { "get": { "description": "Without filters, lists every stored scope name, including imported names retained after an importer is disabled or a source is removed. With IATA or region filters, lists only manual scopes configured for a matching region and imported scopes whose current MeshMapper catalogue includes a matching IATA; observed traffic does not add scopes.", @@ -4996,6 +5065,9 @@ const docTemplate = `{ "description": "matched transport scope name e.g. \"#bc\"", "type": "string" }, + "traceQuality": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality" + }, "transportCodes": { "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.PacketTransportCodes" } @@ -5127,6 +5199,13 @@ const docTemplate = `{ }, "sourceBroker": { "type": "string" + }, + "traceQuality": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality" + }, + "wirePathBytes": { + "description": "Exact on-air TRACE SNR bytes, before the display path substitution.", + "type": "string" } } }, @@ -6045,6 +6124,29 @@ const docTemplate = `{ } } }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.TopologyLinks": { + "type": "object", + "properties": { + "capped": { + "type": "boolean" + }, + "links": { + "type": "array", + "items": { + "type": "array", + "items": { + "type": "string" + } + } + }, + "since": { + "type": "integer" + }, + "until": { + "type": "integer" + } + } + }, "github_com_MeshCore-Beacon_beacon-server_internal_api.TraceDetail": { "type": "object", "properties": { @@ -6076,6 +6178,9 @@ const docTemplate = `{ "description": "hex-encoded packet hash", "type": "string" }, + "quality": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality" + }, "rawPath": { "description": "hops as received in the packet", "type": "array", @@ -6130,6 +6235,9 @@ const docTemplate = `{ "type": "string" } }, + "quality": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality" + }, "snrValues": { "description": "SNR per hop from the most complete observation", "type": "array", @@ -6186,6 +6294,76 @@ const docTemplate = `{ } } }, + "github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.CatalogueScope": { + "type": "object", + "properties": { + "default": { + "type": "integer" + }, + "monitored": { + "type": "boolean" + }, + "name": { + "type": "string" + }, + "repeaters": { + "type": "integer" + }, + "wardriving": { + "type": "boolean" + } + } + }, + "github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.PublicScopeCatalogue": { + "type": "object", + "properties": { + "checkedAt": { + "type": "integer" + }, + "freshUntil": { + "type": "integer" + }, + "generatedAt": { + "type": "integer" + }, + "iata": { + "type": "string" + }, + "lastError": { + "type": "string" + }, + "repeaters": { + "type": "integer" + }, + "scoped": { + "type": "integer" + }, + "scopes": { + "type": "array", + "items": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.CatalogueScope" + } + }, + "url": { + "type": "string" + } + } + }, + "github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality": { + "type": "object", + "properties": { + "reasons": { + "type": "array", + "items": { + "type": "string" + } + }, + "status": { + "description": "supported, suspect, or ambiguous; supported is not authenticated", + "type": "string" + } + } + }, "internal_api_handlers.APIError": { "type": "object", "properties": { @@ -6265,7 +6443,7 @@ const docTemplate = `{ // SwaggerInfo holds exported Swagger Info so clients can modify it var SwaggerInfo = &swag.Spec{ - Version: "2.0.6", + Version: "2.1.0", Host: "localhost:8080", BasePath: "/api/v1", Schemes: []string{"http", "https"}, diff --git a/docs/swagger.json b/docs/swagger.json index 91de39b..f74330e 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -15,7 +15,7 @@ "license": { "name": "AGPL-3-or-later" }, - "version": "2.0.6" + "version": "2.1.0" }, "host": "localhost:8080", "basePath": "/api/v1", @@ -2341,6 +2341,52 @@ } } }, + "/routes/topology": { + "get": { + "description": "Returns up to 100000 undirected node-ID pairs, deduplicated across known routes. Minute-aligned window includes the current minute. Existing rate limits apply; no per-node lookup or route pagination is needed. Capped is true when more links exist.", + "produces": [ + "application/json" + ], + "tags": [ + "Routes" + ], + "summary": "Unique adjacent route links for a topology snapshot", + "parameters": [ + { + "type": "string", + "description": "History window: 15m (default), 1h, or 24h", + "name": "window", + "in": "query" + }, + { + "type": "string", + "description": "Comma-separated reception IATA codes", + "name": "iatas", + "in": "query" + } + ], + "responses": { + "200": { + "description": "OK", + "schema": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.TopologyLinks" + } + }, + "400": { + "description": "Bad Request", + "schema": { + "$ref": "#/definitions/internal_api_handlers.APIError" + } + }, + "503": { + "description": "Service Unavailable", + "schema": { + "$ref": "#/definitions/internal_api_handlers.APIError" + } + } + } + } + }, "/routes/{iata}/{pathKey}/observations": { "get": { "description": "Matches the saved IATA, complete path bytes and hash width. Excludes TRACE and unclassified reports; other widths and search-result subsegments are not included. Byte matches do not prove hop identities or delivery. The route and its historical counter can outlive raw evidence. Cursors retain full timestamp precision and pin route/window scope.", @@ -2427,6 +2473,29 @@ } } }, + "/scope-catalogues": { + "get": { + "description": "Discovered sources only. Counts describe the source region, not individual nodes or links. No upstream request is made. An empty list means no configured importer; checkedAt=0 means no successful check. lastError or a past freshUntil marks stale metadata. Manual Beacon scopes and packet evidence are separate.", + "produces": [ + "application/json" + ], + "tags": [ + "Scopes" + ], + "summary": "Cached MeshMapper regional scope catalogues", + "responses": { + "200": { + "description": "OK", + "schema": { + "type": "array", + "items": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.PublicScopeCatalogue" + } + } + } + } + } + }, "/scopes": { "get": { "description": "Without filters, lists every stored scope name, including imported names retained after an importer is disabled or a source is removed. With IATA or region filters, lists only manual scopes configured for a matching region and imported scopes whose current MeshMapper catalogue includes a matching IATA; observed traffic does not add scopes.", @@ -4994,6 +5063,9 @@ "description": "matched transport scope name e.g. \"#bc\"", "type": "string" }, + "traceQuality": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality" + }, "transportCodes": { "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.PacketTransportCodes" } @@ -5125,6 +5197,13 @@ }, "sourceBroker": { "type": "string" + }, + "traceQuality": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality" + }, + "wirePathBytes": { + "description": "Exact on-air TRACE SNR bytes, before the display path substitution.", + "type": "string" } } }, @@ -6043,6 +6122,29 @@ } } }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.TopologyLinks": { + "type": "object", + "properties": { + "capped": { + "type": "boolean" + }, + "links": { + "type": "array", + "items": { + "type": "array", + "items": { + "type": "string" + } + } + }, + "since": { + "type": "integer" + }, + "until": { + "type": "integer" + } + } + }, "github_com_MeshCore-Beacon_beacon-server_internal_api.TraceDetail": { "type": "object", "properties": { @@ -6074,6 +6176,9 @@ "description": "hex-encoded packet hash", "type": "string" }, + "quality": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality" + }, "rawPath": { "description": "hops as received in the packet", "type": "array", @@ -6128,6 +6233,9 @@ "type": "string" } }, + "quality": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality" + }, "snrValues": { "description": "SNR per hop from the most complete observation", "type": "array", @@ -6184,6 +6292,76 @@ } } }, + "github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.CatalogueScope": { + "type": "object", + "properties": { + "default": { + "type": "integer" + }, + "monitored": { + "type": "boolean" + }, + "name": { + "type": "string" + }, + "repeaters": { + "type": "integer" + }, + "wardriving": { + "type": "boolean" + } + } + }, + "github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.PublicScopeCatalogue": { + "type": "object", + "properties": { + "checkedAt": { + "type": "integer" + }, + "freshUntil": { + "type": "integer" + }, + "generatedAt": { + "type": "integer" + }, + "iata": { + "type": "string" + }, + "lastError": { + "type": "string" + }, + "repeaters": { + "type": "integer" + }, + "scoped": { + "type": "integer" + }, + "scopes": { + "type": "array", + "items": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.CatalogueScope" + } + }, + "url": { + "type": "string" + } + } + }, + "github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality": { + "type": "object", + "properties": { + "reasons": { + "type": "array", + "items": { + "type": "string" + } + }, + "status": { + "description": "supported, suspect, or ambiguous; supported is not authenticated", + "type": "string" + } + } + }, "internal_api_handlers.APIError": { "type": "object", "properties": { diff --git a/docs/swagger.yaml b/docs/swagger.yaml index 48561a3..43f3ee4 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -978,6 +978,8 @@ definitions: scope: description: matched transport scope name e.g. "#bc" type: string + traceQuality: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality' transportCodes: $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.PacketTransportCodes' type: object @@ -1077,6 +1079,11 @@ definitions: type: number sourceBroker: type: string + traceQuality: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality' + wirePathBytes: + description: Exact on-air TRACE SNR bytes, before the display path substitution. + type: string type: object github_com_MeshCore-Beacon_beacon-server_internal_api.PacketObservationSummary: properties: @@ -1698,6 +1705,21 @@ definitions: senderName: type: string type: object + github_com_MeshCore-Beacon_beacon-server_internal_api.TopologyLinks: + properties: + capped: + type: boolean + links: + items: + items: + type: string + type: array + type: array + since: + type: integer + until: + type: integer + type: object github_com_MeshCore-Beacon_beacon-server_internal_api.TraceDetail: properties: packets: @@ -1720,6 +1742,8 @@ definitions: packetHash: description: hex-encoded packet hash type: string + quality: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality' rawPath: description: hops as received in the packet items: @@ -1759,6 +1783,8 @@ definitions: items: type: string type: array + quality: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality' snrValues: description: SNR per hop from the most complete observation items: @@ -1796,6 +1822,52 @@ definitions: requires_restart: type: boolean type: object + github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.CatalogueScope: + properties: + default: + type: integer + monitored: + type: boolean + name: + type: string + repeaters: + type: integer + wardriving: + type: boolean + type: object + github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.PublicScopeCatalogue: + properties: + checkedAt: + type: integer + freshUntil: + type: integer + generatedAt: + type: integer + iata: + type: string + lastError: + type: string + repeaters: + type: integer + scoped: + type: integer + scopes: + items: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.CatalogueScope' + type: array + url: + type: string + type: object + github_com_MeshCore-Beacon_beacon-server_internal_tracequality.Quality: + properties: + reasons: + items: + type: string + type: array + status: + description: supported, suspect, or ambiguous; supported is not authenticated + type: string + type: object internal_api_handlers.APIError: properties: code: @@ -1825,7 +1897,7 @@ info: name: AGPL-3-or-later termsOfService: https://github.com/MeshCore-Beacon/beacon-server title: MeshCore Beacon API - version: 2.0.6 + version: 2.1.0 paths: /admin/accounts: get: @@ -3474,6 +3546,58 @@ paths: summary: Search known routes by source and destination hash tags: - Routes + /routes/topology: + get: + description: Returns up to 100000 undirected node-ID pairs, deduplicated across + known routes. Minute-aligned window includes the current minute. Existing + rate limits apply; no per-node lookup or route pagination is needed. Capped + is true when more links exist. + parameters: + - description: 'History window: 15m (default), 1h, or 24h' + in: query + name: window + type: string + - description: Comma-separated reception IATA codes + in: query + name: iatas + type: string + produces: + - application/json + responses: + "200": + description: OK + schema: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.TopologyLinks' + "400": + description: Bad Request + schema: + $ref: '#/definitions/internal_api_handlers.APIError' + "503": + description: Service Unavailable + schema: + $ref: '#/definitions/internal_api_handlers.APIError' + summary: Unique adjacent route links for a topology snapshot + tags: + - Routes + /scope-catalogues: + get: + description: Discovered sources only. Counts describe the source region, not + individual nodes or links. No upstream request is made. An empty list means + no configured importer; checkedAt=0 means no successful check. lastError or + a past freshUntil marks stale metadata. Manual Beacon scopes and packet evidence + are separate. + produces: + - application/json + responses: + "200": + description: OK + schema: + items: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_meshmapper.PublicScopeCatalogue' + type: array + summary: Cached MeshMapper regional scope catalogues + tags: + - Scopes /scopes: get: description: Without filters, lists every stored scope name, including imported diff --git a/internal/api/handlers/routes.go b/internal/api/handlers/routes.go index ee075bc..1463438 100644 --- a/internal/api/handlers/routes.go +++ b/internal/api/handlers/routes.go @@ -24,6 +24,7 @@ import ( func RoutesRouter(reader api.Reader) http.Handler { r := chi.NewRouter() r.Get("/", listKnownRoutes(reader)) + r.Get("/topology", getTopologyLinks(reader)) r.Get("/cross", searchCrossIATARoutes(reader)) r.Get("/search", searchKnownRoutes(reader)) r.Get("/{iata}/{pathKey}/observations", getRouteEvidence(reader)) diff --git a/internal/api/handlers/scope_catalogues.go b/internal/api/handlers/scope_catalogues.go new file mode 100644 index 0000000..886a874 --- /dev/null +++ b/internal/api/handlers/scope_catalogues.go @@ -0,0 +1,33 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package handlers + +import ( + "net/http" + + "github.com/MeshCore-Beacon/beacon-server/internal/meshmapper" +) + +// ScopeCatalogueSource supplies immutable, already-cached public metadata. +type ScopeCatalogueSource interface { + Catalogues() []meshmapper.PublicScopeCatalogue +} + +// ScopeCatalogues godoc +// @Summary Cached MeshMapper regional scope catalogues +// @Description Discovered sources only. Counts describe the source region, not individual nodes or links. No upstream request is made. An empty list means no configured importer; checkedAt=0 means no successful check. lastError or a past freshUntil marks stale metadata. Manual Beacon scopes and packet evidence are separate. +// @Tags Scopes +// @Produce json +// @Success 200 {array} meshmapper.PublicScopeCatalogue +// @Router /scope-catalogues [get] +func ScopeCatalogues(source ScopeCatalogueSource) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + items := []meshmapper.PublicScopeCatalogue{} + if source != nil { + items = source.Catalogues() + } + w.Header().Set("Cache-Control", "public, max-age=60") + respond(w, http.StatusOK, items) + } +} diff --git a/internal/api/handlers/stub_reader_test.go b/internal/api/handlers/stub_reader_test.go index 71d7fac..c56afc9 100644 --- a/internal/api/handlers/stub_reader_test.go +++ b/internal/api/handlers/stub_reader_test.go @@ -403,6 +403,10 @@ func (s stubReader) GetRouteEvidence(ctx context.Context, iata, key string, q ap return nil, nil } +func (s stubReader) GetTopologyLinks(_ context.Context, _ []string, _ time.Time, _ time.Time) (*api.TopologyLinks, error) { + return &api.TopologyLinks{}, nil +} + func (s stubReader) ListObserverDirectory(ctx context.Context, q api.ObserverDirectoryQuery) (*api.ObserverDirectory, error) { if s.listObserverDirectory != nil { return s.listObserverDirectory(ctx, q) diff --git a/internal/api/handlers/topology.go b/internal/api/handlers/topology.go new file mode 100644 index 0000000..062885e --- /dev/null +++ b/internal/api/handlers/topology.go @@ -0,0 +1,49 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later +package handlers + +import ( + "context" + "github.com/MeshCore-Beacon/beacon-server/internal/api" + "net/http" + "time" +) + +// getTopologyLinks godoc +// @Summary Unique adjacent route links for a topology snapshot +// @Description Returns up to 100000 undirected node-ID pairs, deduplicated across known routes. Minute-aligned window includes the current minute. Existing rate limits apply; no per-node lookup or route pagination is needed. Capped is true when more links exist. +// @Tags Routes +// @Produce json +// @Param window query string false "History window: 15m (default), 1h, or 24h" +// @Param iatas query string false "Comma-separated reception IATA codes" +// @Success 200 {object} api.TopologyLinks +// @Failure 400 {object} handlers.APIError +// @Failure 503 {object} handlers.APIError +// @Router /routes/topology [get] +func getTopologyLinks(reader api.Reader) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + window := r.URL.Query().Get("window") + if window == "" { + window = "15m" + } + duration, ok := map[string]time.Duration{"15m": 15 * time.Minute, "1h": time.Hour, "24h": 24 * time.Hour}[window] + if !ok { + respondError(w, http.StatusBadRequest, "window must be 15m, 1h or 24h") + return + } + iatas, err := routeIATAs(r) + if err != nil { + respondError(w, http.StatusBadRequest, err.Error()) + return + } + until := time.Now().UTC().Truncate(time.Minute).Add(time.Minute) + ctx, cancel := context.WithTimeout(r.Context(), 15*time.Second) + defer cancel() + links, err := reader.GetTopologyLinks(ctx, iatas, until.Add(-duration), until) + if err != nil { + respondError(w, http.StatusServiceUnavailable, "route snapshot unavailable; try again shortly") + return + } + respond(w, http.StatusOK, links) + } +} diff --git a/internal/api/handlers/topology_test.go b/internal/api/handlers/topology_test.go new file mode 100644 index 0000000..22741b0 --- /dev/null +++ b/internal/api/handlers/topology_test.go @@ -0,0 +1,33 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later +package handlers + +import ( + "net/http/httptest" + "strings" + "testing" +) + +func TestTopologyWindowValidation(t *testing.T) { + for _, window := range []string{"", "15m", "1h", "24h", "72h", "0", "garbage"} { + w := httptest.NewRecorder() + RoutesRouter(stubReader{}).ServeHTTP(w, httptest.NewRequest("GET", "/topology?window="+window, nil)) + want := 200 + if window == "72h" || window == "0" || window == "garbage" { + want = 400 + } + if w.Code != want { + t.Fatalf("%s: %d", window, w.Code) + } + } +} + +func TestTopologyRejectsInvalidRegions(t *testing.T) { + for _, query := range []string{"iata=TOOLONG", "iatas=" + strings.Join([]string{"YKF", "invalid"}, ",")} { + w := httptest.NewRecorder() + RoutesRouter(stubReader{}).ServeHTTP(w, httptest.NewRequest("GET", "/topology?"+query, nil)) + if w.Code != 400 { + t.Fatalf("%s: %d", query, w.Code) + } + } +} diff --git a/internal/api/packets.go b/internal/api/packets.go index e5f47fe..f2fccb7 100644 --- a/internal/api/packets.go +++ b/internal/api/packets.go @@ -3,6 +3,8 @@ package api +import "github.com/MeshCore-Beacon/beacon-server/internal/tracequality" + import ( "encoding/hex" "encoding/json" @@ -53,19 +55,21 @@ type PacketPathLength struct { // PacketObservationDetail is a full observation including radio settings and resolved path. type PacketObservationDetail struct { - ID int64 `json:"id"` - ObserverID uuid.UUID `json:"observerId"` - ObserverName *string `json:"observerName,omitempty"` - IATA string `json:"iata"` - HeardAt int64 `json:"heardAt"` // epoch ms - PathLength PacketPathLength `json:"pathLength"` - PathBytes *string `json:"pathBytes,omitempty"` // hex-encoded accumulated path hashes - RSSI *int16 `json:"rssi,omitempty"` - SNR *float32 `json:"snr,omitempty"` - PropagationTimeMs *int32 `json:"propagationTimeMs"` // ms since first observation; 0 for first - Radio *PacketRadio `json:"radio,omitempty"` - SourceBroker string `json:"sourceBroker"` - ResolvedPath []ResolvedHop `json:"resolvedPath"` // per-observation resolved path hashes + WirePathBytes *string `json:"wirePathBytes,omitempty"` // Exact on-air TRACE SNR bytes, before the display path substitution. + TraceQuality *tracequality.Quality `json:"traceQuality,omitempty"` + ID int64 `json:"id"` + ObserverID uuid.UUID `json:"observerId"` + ObserverName *string `json:"observerName,omitempty"` + IATA string `json:"iata"` + HeardAt int64 `json:"heardAt"` // epoch ms + PathLength PacketPathLength `json:"pathLength"` + PathBytes *string `json:"pathBytes,omitempty"` // hex-encoded accumulated path hashes + RSSI *int16 `json:"rssi,omitempty"` + SNR *float32 `json:"snr,omitempty"` + PropagationTimeMs *int32 `json:"propagationTimeMs"` // ms since first observation; 0 for first + Radio *PacketRadio `json:"radio,omitempty"` + SourceBroker string `json:"sourceBroker"` + ResolvedPath []ResolvedHop `json:"resolvedPath"` // per-observation resolved path hashes // ResolvedSource/ResolvedDestination are the packet's endpoints, when the payload type // carries one. Prefer the snapshot captured at ingest; legacy observations without a // snapshot use the current node registry. Endpoint matching itself is unchanged: @@ -134,6 +138,7 @@ type PacketTransportCodes struct { // Packet is the full packet representation including all observations and resolved paths. type Packet struct { + TraceQuality *tracequality.Quality `json:"traceQuality,omitempty"` PacketHash string `json:"packetHash"` Header PacketHeader `json:"header"` TransportCodes *PacketTransportCodes `json:"transportCodes,omitempty"` diff --git a/internal/api/reader.go b/internal/api/reader.go index 672dfe2..932c8d1 100644 --- a/internal/api/reader.go +++ b/internal/api/reader.go @@ -226,6 +226,8 @@ type Reader interface { // GetTraceByTag returns all packets for a given trace tag with resolved routes. GetTraceByTag(ctx context.Context, tag string) (*TraceDetail, error) + GetTopologyLinks(ctx context.Context, iatas []string, since, until time.Time) (*TopologyLinks, error) + // ListKnownRoutes returns known routes filtered by IATA and optional hop count. // A nonzero cursorID breaks ties within the cursor's millisecond. ListKnownRoutes(ctx context.Context, iata string, hopCount int32, cursor time.Time, cursorID int64, limit int32) ([]KnownRoute, error) diff --git a/internal/api/router/options.go b/internal/api/router/options.go index 971a144..3d6dc93 100644 --- a/internal/api/router/options.go +++ b/internal/api/router/options.go @@ -14,6 +14,7 @@ import ( // Options names the startup settings used by the router. Adding an optional // route does not change New's signature or require unrelated callers to change. type Options struct { + ScopeCatalogues handlers.ScopeCatalogueSource MaxConnsPerIP int MaxConnectsPerMinute int WSAllowedOrigins []string diff --git a/internal/api/router/router.go b/internal/api/router/router.go index f9d1f53..f646c76 100644 --- a/internal/api/router/router.go +++ b/internal/api/router/router.go @@ -119,6 +119,7 @@ func New(h *hub.Hub, reader api.Reader, workers []*ingest.Worker, opts Options) r.Mount("/routes", handlers.RoutesRouter(reader)) r.Mount("/scopes", handlers.ScopesRouter(reader, opts.Scopes)) r.Mount("/stats", handlers.StatsRouter(reader, handlers.StatsOptions{Scopes: opts.Scopes, SeriesWindow: opts.RollupRetention})) + r.Get("/scope-catalogues", handlers.ScopeCatalogues(opts.ScopeCatalogues)) r.Mount("/traces", handlers.TracesRouter(reader)) r.Mount("/info", handlers.InfoRouter(info)) }) diff --git a/internal/api/router/scope_catalogues_test.go b/internal/api/router/scope_catalogues_test.go new file mode 100644 index 0000000..5092e4c --- /dev/null +++ b/internal/api/router/scope_catalogues_test.go @@ -0,0 +1,28 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package router + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/MeshCore-Beacon/beacon-server/internal/meshmapper" +) + +func TestScopeCataloguesDisabledAndReadOnly(t *testing.T) { + var importer *meshmapper.Importer + r := New(nil, nil, nil, Options{ScopeCatalogues: importer}) + for _, method := range []string{http.MethodGet, http.MethodPost} { + w := httptest.NewRecorder() + r.ServeHTTP(w, httptest.NewRequest(method, "/api/v1/scope-catalogues", nil)) + if method == http.MethodGet && (w.Code != 200 || strings.TrimSpace(w.Body.String()) != "[]" || w.Header().Get("Cache-Control") != "public, max-age=60") { + t.Fatal(w.Code, w.Body.String()) + } + if method == http.MethodPost && w.Code != 405 { + t.Fatal(w.Code) + } + } +} diff --git a/internal/api/topology.go b/internal/api/topology.go new file mode 100644 index 0000000..3fc85ce --- /dev/null +++ b/internal/api/topology.go @@ -0,0 +1,14 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later +package api + +import "github.com/google/uuid" + +// TopologyLinks is a bounded set of undirected, adjacent known-route segments. +// It does not imply radio reachability or join across missing path entries. +type TopologyLinks struct { + Links [][2]uuid.UUID `json:"links"` + Capped bool `json:"capped"` + Since int64 `json:"since"` + Until int64 `json:"until"` +} diff --git a/internal/api/traces.go b/internal/api/traces.go index b08fd7c..f4a70d0 100644 --- a/internal/api/traces.go +++ b/internal/api/traces.go @@ -3,29 +3,33 @@ package api +import "github.com/MeshCore-Beacon/beacon-server/internal/tracequality" + // TraceTagSummary is a single trace tag with aggregate metadata. type TraceTagSummary struct { - TraceTag string `json:"traceTag"` // hex-encoded 4-byte tag - FirstHeardAt int64 `json:"firstHeardAt"` // epoch ms - LastHeardAt int64 `json:"lastHeardAt"` // epoch ms - PacketCount int64 `json:"packetCount"` // number of packets with this trace tag - IATACount int64 `json:"iataCount"` // number of distinct IATAs where heard - TraceType string `json:"traceType"` // TRACE or PING - PathHashes []string `json:"pathHashes"` // hops from the most complete observation - SNRValues []float32 `json:"snrValues"` // SNR per hop from the most complete observation + Quality tracequality.Quality `json:"quality"` + TraceTag string `json:"traceTag"` // hex-encoded 4-byte tag + FirstHeardAt int64 `json:"firstHeardAt"` // epoch ms + LastHeardAt int64 `json:"lastHeardAt"` // epoch ms + PacketCount int64 `json:"packetCount"` // number of packets with this trace tag + IATACount int64 `json:"iataCount"` // number of distinct IATAs where heard + TraceType string `json:"traceType"` // TRACE or PING + PathHashes []string `json:"pathHashes"` // hops from the most complete observation + SNRValues []float32 `json:"snrValues"` // SNR per hop from the most complete observation } // TracePacket is a single packet within a trace series, including its // resolved route derived from the trace path hashes. type TracePacket struct { - PacketHash string `json:"packetHash"` // hex-encoded packet hash - RouteType int16 `json:"routeType"` // numeric route type - RouteTypeName string `json:"routeTypeName"` // human-readable route type - Scope *string `json:"scope,omitempty"` // transport scope name, if known - FirstHeardAt int64 `json:"firstHeardAt"` // epoch ms - LastHeardAt int64 `json:"lastHeardAt"` // epoch ms - RawPath []RawHop `json:"rawPath"` // hops as received in the packet - ResolvedRoute []ResolvedHop `json:"resolvedRoute"` // hops resolved to known nodes + Quality tracequality.Quality `json:"quality"` + PacketHash string `json:"packetHash"` // hex-encoded packet hash + RouteType int16 `json:"routeType"` // numeric route type + RouteTypeName string `json:"routeTypeName"` // human-readable route type + Scope *string `json:"scope,omitempty"` // transport scope name, if known + FirstHeardAt int64 `json:"firstHeardAt"` // epoch ms + LastHeardAt int64 `json:"lastHeardAt"` // epoch ms + RawPath []RawHop `json:"rawPath"` // hops as received in the packet + ResolvedRoute []ResolvedHop `json:"resolvedRoute"` // hops resolved to known nodes } // RawHop is a single hop in a trace path as received in the packet. diff --git a/internal/cache/cache_test.go b/internal/cache/cache_test.go index 5786c12..546910a 100644 --- a/internal/cache/cache_test.go +++ b/internal/cache/cache_test.go @@ -451,6 +451,10 @@ func (s *stubReader) GetRouteEvidence(_ context.Context, _, _ string, _ api.Rout return nil, nil } +func (s *stubReader) GetTopologyLinks(_ context.Context, _ []string, _ time.Time, _ time.Time) (*api.TopologyLinks, error) { + return &api.TopologyLinks{}, nil +} + func (s *stubReader) ListObserverDirectory(context.Context, api.ObserverDirectoryQuery) (*api.ObserverDirectory, error) { return nil, nil } diff --git a/internal/cache/reader.go b/internal/cache/reader.go index d573ab9..f67f9ac 100644 --- a/internal/cache/reader.go +++ b/internal/cache/reader.go @@ -374,6 +374,15 @@ func (cr *CachedReader) ListTraceTags(ctx context.Context, iatas []string, scope return cr.inner.ListTraceTags(ctx, iatas, scope, traceType, since, until, cursor, cursorTag, limit) } +func (cr *CachedReader) GetTopologyLinks(ctx context.Context, iatas []string, since, until time.Time) (*api.TopologyLinks, error) { + regions := append([]string{}, iatas...) + sort.Strings(regions) + key := fmt.Sprintf("beacon:topology:%s:%d:%d", strings.Join(regions, ","), since.Unix(), until.Unix()) + return getOrSet(ctx, cr.c, key, 30*time.Second, func(ctx context.Context) (*api.TopologyLinks, error) { + return cr.inner.GetTopologyLinks(ctx, iatas, since, until) + }) +} + func (cr *CachedReader) ListObserverDirectory(ctx context.Context, query api.ObserverDirectoryQuery) (*api.ObserverDirectory, error) { return cr.inner.ListObserverDirectory(ctx, query) } diff --git a/internal/ingest/channel_scope_test.go b/internal/ingest/channel_scope_test.go index 3ee6933..e229ef2 100644 --- a/internal/ingest/channel_scope_test.go +++ b/internal/ingest/channel_scope_test.go @@ -5,6 +5,7 @@ package ingest import ( "context" + "encoding/hex" "encoding/json" "testing" "time" @@ -73,3 +74,36 @@ func TestChannelMessageScopeLive(t *testing.T) { }) } } + +func TestChannelPublicMarkerUsesDecryptionKey(t *testing.T) { + for _, public := range []bool{false, true} { + w, base := newTestWorker() + w.db = &scopeMessageDB{stubDB: base, next: &InsertedChannelMessage{ID: 1}} + key := make([]byte, 16) + if public { + key, _ = hex.DecodeString("8b3387e9c5cdea6ac9e5edbaa115cd72") + } + w.keys = &mapKeys{entries: map[byte][]keystore.Entry{0x11: {{Key: key, Name: "Public"}}}} + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + go w.hub.Run() + client := w.hub.NewClient() + defer w.hub.Remove(client) + w.hub.AddScope(client, "public-marker", hub.Scope{Events: []hub.EventType{hub.EventChannelMessage, hub.EventObserverStatus}}) + waitForSummarySubscriber(t, ctx, w.hub, client) + packet := &meshcore.Packet{Header: meshcore.MakeHeader(meshcore.RouteTypeFlood, meshcore.PayloadTypeGrpTxt, 0), Payload: encryptedGroupText(t, 0x11, key, "Fixture", "test")} + w.handlePayloadTypeSideEffects(ctx, packet, "YKF", []byte{1}, RadioSettings{}, nil, nil, nil, 0) + select { + case event := <-client.Send: + var got channelMessageEvent + if err := json.Unmarshal(event.Payload, &got); err != nil { + t.Fatal(err) + } + if got.IsPublic != public { + t.Fatal("channel name/hash used instead of actual decryption key") + } + case <-ctx.Done(): + t.Fatal("no channel event") + } + } +} diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index 6c3abb7..46d7f83 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -158,6 +158,7 @@ type DB interface { // ResolvePathHashes returns a list of node UUIDs for the given path hash prefixes and IATA. ResolvePathHashes(ctx context.Context, iata string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error) + ResolveTracePathHashes(ctx context.Context, iata string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error) // ResolveEndpointHashes matches one-byte logical endpoints, including companions. ResolveEndpointHashes(ctx context.Context, iata string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error) diff --git a/internal/ingest/ingest_test.go b/internal/ingest/ingest_test.go index cffa465..0c45b48 100644 --- a/internal/ingest/ingest_test.go +++ b/internal/ingest/ingest_test.go @@ -266,6 +266,9 @@ func (s *stubDB) GetObserverScopes(_ context.Context, _ uuid.UUID) ([]string, er func (s *stubDB) ResolvePathHashes(_ context.Context, _ string, _ [][]byte) (map[string][]api.ResolvedPathEntry, error) { return nil, nil } +func (s *stubDB) ResolveTracePathHashes(_ context.Context, _ string, _ [][]byte) (map[string][]api.ResolvedPathEntry, error) { + return nil, nil +} func (s *stubDB) ResolveEndpointHashes(_ context.Context, _ string, _ [][]byte) (map[string][]api.ResolvedPathEntry, error) { return nil, nil diff --git a/internal/ingest/packet.go b/internal/ingest/packet.go index e875492..2a4b84e 100644 --- a/internal/ingest/packet.go +++ b/internal/ingest/packet.go @@ -18,6 +18,7 @@ import ( "github.com/MeshCore-Beacon/beacon-server/internal/api" "github.com/MeshCore-Beacon/beacon-server/internal/lora" "github.com/MeshCore-Beacon/beacon-server/internal/scopestore" + "github.com/MeshCore-Beacon/beacon-server/internal/tracequality" "github.com/google/uuid" "github.com/meshcore-go/meshcore-go" ) @@ -202,13 +203,14 @@ type parsedGroupEnvelope struct { // is the accumulated per-hop SNR sequence carried in the packet header's // path field (a protocol-level repurposing distinct from the payload). type parsedTrace struct { - Raw string `json:"raw"` - Type string `json:"type"` - TraceTag string `json:"traceTag"` - AuthCode uint32 `json:"authCode"` - Flags byte `json:"flags"` - PathHashes []string `json:"pathHashes"` - SNRValues []float32 `json:"snrValues"` + Quality tracequality.Quality `json:"quality"` + Raw string `json:"raw"` + Type string `json:"type"` + TraceTag string `json:"traceTag"` + AuthCode uint32 `json:"authCode"` + Flags byte `json:"flags"` + PathHashes []string `json:"pathHashes"` + SNRValues []float32 `json:"snrValues"` } // parsedAck is the parsed form of an ACK payload (ยง2.12). Checksum is the @@ -557,7 +559,10 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ hashes = append(hashes, hex.EncodeToString(h)) rawHashes = append(rawHashes, h) } - traceRawHashes = rawHashes + quality := tracequality.Assess(packet.Payload, int(packet.PayloadVer()), int(packet.RouteType()), int(packet.PathLength)) + if quality.Status == "supported" { + traceRawHashes = rawHashes + } // SNR values are in packet.Path, one signed int8 per consumed hop snrValues := make([]float32, 0, len(packet.Path)) for _, b := range packet.Path { @@ -568,6 +573,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ traceType = "PING" } pt := parsedTrace{ + Quality: quality, Raw: hex.EncodeToString(packet.Payload), Type: traceType, TraceTag: hex.EncodeToString(uint32ToBytes(trace.Tag)), @@ -585,8 +591,8 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ // originator isn't in PathHashes) and is skipped. Either side // resolving ambiguously (>1 candidate) or not at all (0 // candidates) skips that specific pair. - if len(rawHashes) >= 2 { - resolved, rErr := w.db.ResolvePathHashes(ctx, iata, rawHashes) + if quality.Status == "supported" && len(rawHashes) >= 2 { + resolved, rErr := w.db.ResolveTracePathHashes(ctx, iata, rawHashes) if rErr == nil { for i := 1; i < len(rawHashes); i++ { if i >= len(snrValues) { @@ -822,12 +828,20 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ if packet.PayloadType() == meshcore.PayloadTypeTrace { hashes = traceRawHashes } - resolved, err := w.db.ResolvePathHashes(ctx, iata, hashes) + var resolved map[string][]api.ResolvedPathEntry + if packet.PayloadType() == meshcore.PayloadTypeTrace { + resolved, err = w.db.ResolveTracePathHashes(ctx, iata, hashes) + } else { + resolved, err = w.db.ResolvePathHashes(ctx, iata, hashes) + } if err != nil { w.log.Error("path resolution failed", "error", err) } var resolvedIDs []uuid.UUID for _, entries := range resolved { + if len(entries) != 1 { + continue + } for _, e := range entries { resolvedIDs = append(resolvedIDs, e.NodeID) } @@ -846,13 +860,31 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ nodeIDs = append(nodeIDs, entries[0].NodeID) hashPrefixes = append(hashPrefixes, hash) } - if allHigh && len(nodeIDs) > 1 { + if allHigh && len(nodeIDs) > 1 && packet.PayloadType() != meshcore.PayloadTypeTrace { if err := w.db.UpsertKnownRoute(ctx, nodeIDs, hashPrefixes, iata, int32(len(nodeIDs))); err != nil { w.log.Error("failed to upsert known route", "error", err) } } } - w.runCapabilityDetection(ctx, packet.PayloadType(), packet.PathHashSize(), resolvedIDs) + hashSize := packet.PathHashSize() + if packet.PayloadType() == meshcore.PayloadTypeTrace { + hashSize = 1 + if len(traceRawHashes) > 0 { + hashSize = uint8(len(traceRawHashes[0])) + } + // Only consumed, uniquely resolved hops demonstrate trace capability. + resolvedIDs = nil + for i, hash := range traceRawHashes { + if i >= len(packet.Path) { + break + } + entries := resolved[hex.EncodeToString(hash)] + if len(entries) == 1 { + resolvedIDs = append(resolvedIDs, entries[0].NodeID) + } + } + } + w.runCapabilityDetection(ctx, packet.PayloadType(), hashSize, resolvedIDs) if inserted || repeat { if inserted { diff --git a/internal/ingest/side_effects.go b/internal/ingest/side_effects.go index 3299027..95c8441 100644 --- a/internal/ingest/side_effects.go +++ b/internal/ingest/side_effects.go @@ -48,6 +48,7 @@ type InsertedChannelMessage struct { // channelMessageEvent is the JSON payload for a channelMessage WS event. type channelMessageEvent struct { + IsPublic bool `json:"isPublic"` // True only when decryption used the documented default Public key, never inferred from a one-byte channel hash. ID int64 `json:"id"` ChannelID int `json:"channelId"` ChannelHash string `json:"channelHash"` // hex-encoded single byte @@ -223,6 +224,7 @@ func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshc if result.Message != nil { evt := channelMessageEvent{ + IsPublic: hex.EncodeToString(result.Entry.Key) == "8b3387e9c5cdea6ac9e5edbaa115cd72", ID: result.Message.ID, Scope: result.Message.Scope, ScopeStatus: result.Message.ScopeStatus, diff --git a/internal/ingest/trace_quality_test.go b/internal/ingest/trace_quality_test.go new file mode 100644 index 0000000..89c36b1 --- /dev/null +++ b/internal/ingest/trace_quality_test.go @@ -0,0 +1,75 @@ +package ingest + +import ( + "context" + "encoding/hex" + "encoding/json" + "testing" + + "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/google/uuid" + "github.com/meshcore-go/meshcore-go" +) + +type traceSafetyDB struct { + stubDB + saved UpsertPacketParams + resolutions, neighbors, routes int +} + +func (s *traceSafetyDB) UpsertPacket(_ context.Context, p UpsertPacketParams) (bool, error) { + s.saved = p + return true, nil +} +func (s *traceSafetyDB) ResolveTracePathHashes(_ context.Context, _ string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error) { + if len(hashes) > 0 { + s.resolutions++ + } + out := map[string][]api.ResolvedPathEntry{} + for i, h := range hashes { + id := uuid.UUID{} + id[15] = byte(i + 1) + out[hex.EncodeToString(h)] = []api.ResolvedPathEntry{{NodeID: id}} + } + return out, nil +} +func (s *traceSafetyDB) UpsertNodeNeighbor(_ context.Context, _, _ uuid.UUID, _ string, _ *float32, _ *string) error { + s.neighbors++ + return nil +} +func (s *traceSafetyDB) UpsertKnownRoute(_ context.Context, _ []uuid.UUID, _ [][]byte, _ string, _ int32) error { + s.routes++ + return nil +} + +func TestTraceQualityDoesNotPromoteMalformedOrUnvisitedHops(t *testing.T) { + for _, bad := range []bool{false, true} { + w, _ := newTestWorker() + db := &traceSafetyDB{} + w.db = db + payload, _ := (&meshcore.Trace{Tag: 7, Flags: 1, PathHashes: []byte{0xaa, 0xbb, 0xcc, 0xdd, 0xee, 0xff}}).ToBytes() + version := byte(0) + if bad { + version = 1 + } + pkt := &meshcore.Packet{Header: meshcore.MakeHeader(meshcore.RouteTypeDirect, meshcore.PayloadTypeTrace, version), PathLength: 2, Path: []byte{0, 252}, Payload: payload} + w.handlePacket(context.Background(), "YKF", "0102", packetEnvelope(t, pkt)) + if len(db.saved.RawPayload) == 0 || len(db.traceHearings) != 1 { + t.Fatal("raw evidence lost") + } + var parsed parsedTrace + if err := json.Unmarshal(db.saved.ParsedPayload, &parsed); err != nil { + t.Fatal(err) + } + if bad { + if parsed.Quality.Status != "suspect" || db.neighbors != 0 || db.resolutions != 0 || len(db.setCapabilityCalls) != 0 { + t.Fatalf("malformed trace promoted: %+v", db) + } + } else if parsed.Quality.Status != "supported" || db.neighbors != 1 || len(db.setCapabilityCalls) != 2 { + t.Fatalf("valid consumed links/capabilities missing: %+v", db) + } + if db.routes != 0 { + t.Fatal("planned route promoted as observed") + } + } +} diff --git a/internal/meshmapper/catalogues.go b/internal/meshmapper/catalogues.go new file mode 100644 index 0000000..7655229 --- /dev/null +++ b/internal/meshmapper/catalogues.go @@ -0,0 +1,75 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package meshmapper + +import ( + "encoding/json" + "time" + + "github.com/MeshCore-Beacon/beacon-server/internal/scopestore" +) + +// CatalogueScope is regional source metadata, not per-node forwarding evidence. +type CatalogueScope struct { + Name string `json:"name"` + Repeaters int `json:"repeaters"` + Default int `json:"default"` + Monitored bool `json:"monitored"` + Wardriving bool `json:"wardriving"` +} + +type PublicScopeCatalogue struct { + IATA string `json:"iata"` + URL string `json:"url"` + GeneratedAt int64 `json:"generatedAt"` + CheckedAt int64 `json:"checkedAt"` + FreshUntil int64 `json:"freshUntil"` + LastError string `json:"lastError,omitempty"` + Repeaters int `json:"repeaters"` + Scoped int `json:"scoped"` + Scopes []CatalogueScope `json:"scopes"` +} + +// Catalogues returns an immutable snapshot of discovered sources. No request +// triggers an upstream fetch or reads ingestion tables. Callers must not mutate it. +func (i *Importer) Catalogues() []PublicScopeCatalogue { + if i != nil { + if snapshot := i.catalogues.Load(); snapshot != nil { + return snapshot.([]PublicScopeCatalogue) + } + } + return []PublicScopeCatalogue{} +} + +func (i *Importer) publishCatalogues() { + result := make([]PublicScopeCatalogue, 0, len(i.sources)) + for _, s := range i.sources { + c := PublicScopeCatalogue{IATA: s.iata, URL: s.url, LastError: s.cache.LastError, Scopes: []CatalogueScope{}} + if !s.cache.CheckedAt.IsZero() { + c.CheckedAt = s.cache.CheckedAt.UnixMilli() + c.FreshUntil = s.cache.CheckedAt.Add(i.interval + 5*time.Minute).UnixMilli() + } + if len(s.cache.Payload) > 0 { + // Reuse the importer validation before exposing saved metadata. Invalid + // saved catalogues are omitted just as they are from the scope matcher. + if _, generated, err := decode(s.cache.Payload, s.iata); err == nil { + var payload struct { + Repeaters int + Scoped int + Scopes []CatalogueScope + } + if json.Unmarshal(s.cache.Payload, &payload) == nil { + c.GeneratedAt = generated.UnixMilli() + c.Repeaters, c.Scoped = payload.Repeaters, payload.Scoped + for _, scope := range payload.Scopes { + scope.Name = scopestore.FromName(scope.Name).Name + c.Scopes = append(c.Scopes, scope) + } + } + } + } + result = append(result, c) + } + i.catalogues.Store(result) +} diff --git a/internal/meshmapper/catalogues_test.go b/internal/meshmapper/catalogues_test.go new file mode 100644 index 0000000..14dd3d0 --- /dev/null +++ b/internal/meshmapper/catalogues_test.go @@ -0,0 +1,104 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package meshmapper + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "sync" + "testing" + "time" + + "github.com/MeshCore-Beacon/beacon-server/internal/config" + "github.com/MeshCore-Beacon/beacon-server/internal/scopestore" +) + +func TestPublicCataloguesFreshnessAndIsolation(t *testing.T) { + status := http.StatusOK + calls := 0 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls++ + w.WriteHeader(status) + if status == http.StatusOK { + _, _ = w.Write(catalogue("yow", "YOW", "monitored-zero")) + } + })) + defer server.Close() + cfg := config.MeshMapperScopesConfig{Enabled: true} + store := newMemoryStore("YOW") + imp, err := New(context.Background(), cfg, store, NewDirectory(newZoneListMemory(), ""), scopestore.New(), nil) + if err != nil { + t.Fatal(err) + } + if got := imp.Catalogues(); len(got) != 1 || got[0].CheckedAt != 0 || len(got[0].Scopes) != 0 { + t.Fatal(got) + } + imp.sources[0].url = server.URL + imp.client = server.Client() + now := time.Now() + if err := imp.refresh(context.Background(), &imp.sources[0], now); err != nil { + t.Fatal(err) + } + imp.publishCatalogues() + old := imp.Catalogues() + if old[0].Scopes[0].Name != "#yow" || old[0].Scopes[1].Name != "#YOW" || len(old[0].Scopes) != 3 || old[0].Repeaters != 5 || old[0].Scoped != 3 || old[0].FreshUntil <= old[0].CheckedAt { + t.Fatal(old) + } + status = http.StatusServiceUnavailable + if err := imp.refresh(context.Background(), &imp.sources[0], now.Add(time.Minute)); err != nil { + t.Fatal(err) + } + imp.publishCatalogues() + current := imp.Catalogues()[0] + if current.LastError != "HTTP 503" || current.CheckedAt != old[0].CheckedAt || current.GeneratedAt != old[0].GeneratedAt || len(current.Scopes) != 3 || old[0].LastError != "" { + t.Fatal(current, old) + } + // Readers only access published immutable data; the race suite exercises concurrent refreshes. + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + for range 100 { + _, _ = json.Marshal(imp.Catalogues()) + } + }() + for range 100 { + imp.publishCatalogues() + } + wg.Wait() + if calls != 2 { + t.Fatal("reading public metadata performed upstream requests", calls) + } + disabled, err := New(context.Background(), config.MeshMapperScopesConfig{}, store, NewDirectory(newZoneListMemory(), ""), scopestore.New(), nil) + if err != nil || len(disabled.Catalogues()) != 0 { + t.Fatal("disabled importer exposed old membership") + } + store.rows[sourceKey{"YOW", server.URL}] = Cache{Payload: []byte(`{"scopes":"invalid"}`)} + restored, err := New(context.Background(), cfg, store, NewDirectory(newZoneListMemory(), ""), scopestore.New(), nil) + if err != nil || len(restored.Catalogues()[0].Scopes) != 0 || restored.Catalogues()[0].LastError == "" { + t.Fatal("invalid saved metadata leaked") + } +} + +func TestPublicCataloguesIncludeDiscoveredAndUnlistedRegions(t *testing.T) { + ctx := context.Background() + store := newMemoryStore("YOW") + dir := NewDirectory(newZoneListMemory(), "") + dir.lists["CA"] = &zoneList{zones: map[string]zoneEntry{}, fetchedAt: time.Now()} + imp, err := New(ctx, config.MeshMapperScopesConfig{Enabled: true}, store, dir, scopestore.New(), nil) + if err != nil { + t.Fatal(err) + } + before := imp.Catalogues() + store.iatas = append(store.iatas, "YKF") + if err := imp.Refresh(ctx); err != nil { + t.Fatal(err) + } + after := imp.Catalogues() + if len(before) != 1 || before[0].LastError != "" || len(after) != 2 || after[1].IATA != "YKF" || after[1].LastError != "not listed" { + t.Fatal(before, after) + } +} diff --git a/internal/meshmapper/scopes.go b/internal/meshmapper/scopes.go index 35dba76..49a1dac 100644 --- a/internal/meshmapper/scopes.go +++ b/internal/meshmapper/scopes.go @@ -12,6 +12,7 @@ import ( "net/http" "slices" "strings" + "sync/atomic" "time" "unicode" "unicode/utf8" @@ -58,16 +59,17 @@ type source struct { // Importer is owned by one background task; ScopeStore synchronizes its consumers. type Importer struct { - store Store - scopes *scopestore.ScopeStore - manual []scopestore.Entry - dir *Directory - scopesURL func(site string) (string, bool) - seen map[string]bool - sources []source - interval time.Duration - client *http.Client - onChange func(context.Context) + catalogues atomic.Value // immutable []PublicScopeCatalogue, off the ingest path + store Store + scopes *scopestore.ScopeStore + manual []scopestore.Entry + dir *Directory + scopesURL func(site string) (string, bool) + seen map[string]bool + sources []source + interval time.Duration + client *http.Client + onChange func(context.Context) } // SetCacheInvalidator is wired once at startup, before the background task starts. @@ -97,6 +99,7 @@ func New(ctx context.Context, cfg config.MeshMapperScopesConfig, store Store, di } i.track(iatas, latest) i.publish() + i.publishCatalogues() return i, nil } @@ -131,6 +134,7 @@ func (i *Importer) track(iatas []string, saved map[string]Catalogue) { // Refresh checks only one due source. The scheduler serializes calls every 15s. func (i *Importer) Refresh(ctx context.Context) (err error) { + defer i.publishCatalogues() parent := ctx defer func() { if parent.Err() != nil { diff --git a/internal/tracequality/quality.go b/internal/tracequality/quality.go new file mode 100644 index 0000000..0394f49 --- /dev/null +++ b/internal/tracequality/quality.go @@ -0,0 +1,96 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +// Package tracequality checks TRACE structure, not authenticity or RF plausibility. +package tracequality + +import "encoding/hex" + +type Quality struct { + Status string `json:"status"` // supported, suspect, or ambiguous; supported is not authenticated + Reasons []string `json:"reasons"` +} + +func (q *Quality) Add(reason string) { + for _, r := range q.Reasons { + if r == reason { + return + } + } + q.Reasons = append(q.Reasons, reason) + if reason == "ambiguous_prefix" && q.Status != "suspect" { + q.Status = "ambiguous" + } else if reason != "ambiguous_prefix" { + q.Status = "suspect" + } +} + +// Assess uses TRACE's power-of-two payload hash widths, NOT the ordinary +// packet path's 1/2/3-byte widths. The path field contains consumed-hop SNRs. +// -1 means a header/path field is unavailable in a retained summary. +// Mesh.cpp onRecvPacket/sendDirect: a trace uses direct routing, flags bits +// 0..1, and fewer than 64 SNR bytes. Partial probes and return loops are valid. +func Assess(raw []byte, version, route, pathLength int) Quality { + q := Quality{Status: "supported", Reasons: []string{}} + if version > 0 { + q.Add("unsupported_version") + } + if route >= 0 && route != 2 && route != 3 { + q.Add("non_direct_trace") + } + if len(raw) < 9 { + q.Add("incomplete_payload") + return q + } + if len(raw) > 184 { + q.Add("payload_too_large") + } + if raw[8]&0xfc != 0 { + q.Add("unsupported_flags") + } + size := 1 << (raw[8] & 3) + if (len(raw)-9)%size != 0 { + q.Add("misaligned_hashes") + } + count := (len(raw) - 9) / size + if count > 63 { + q.Add("too_many_hops") + } + if pathLength > 63 || pathLength > count { + q.Add("invalid_snr_path") + } + return q +} + +// Stored also checks old parsed rows, without rewriting or deleting the evidence. +func Stored(raw string, flags byte, hashes []string, snrs []float32) Quality { + b, err := hex.DecodeString(raw) + if err != nil { + return Quality{Status: "suspect", Reasons: []string{"incomplete_payload"}} + } + if raw == "" { // Older exports may contain only decoded fields. + b = make([]byte, 9) + b[8] = flags + for _, h := range hashes { + v, e := hex.DecodeString(h) + if e != nil { + return Quality{Status: "suspect", Reasons: []string{"misaligned_hashes"}} + } + b = append(b, v...) + } + } + q := Assess(b, -1, -1, len(snrs)) + size := 1 << (flags & 3) + if len(b) >= 9 && b[8] != flags { + q.Add("misaligned_hashes") + } + for _, h := range hashes { + if len(h) != 2*size { + q.Add("misaligned_hashes") + } + } + if len(b) >= 9 && len(hashes) != (len(b)-9)/size { + q.Add("misaligned_hashes") + } + return q +} diff --git a/internal/tracequality/quality_test.go b/internal/tracequality/quality_test.go new file mode 100644 index 0000000..b216851 --- /dev/null +++ b/internal/tracequality/quality_test.go @@ -0,0 +1,54 @@ +package tracequality + +import ( + "bytes" + "slices" + "testing" +) + +func TestTraceFormat(t *testing.T) { + if q := Assess(make([]byte, 9), 0, 2, 0); q.Status != "supported" { + t.Fatalf("zero-hop trace header must remain inspectable: %+v", q) + } + for flags := byte(0); flags < 4; flags++ { + raw := make([]byte, 9) + raw[8] = flags + raw = append(raw, bytes.Repeat([]byte{0xaa}, 3*(1<