diff --git a/cmd/beacon/main.go b/cmd/beacon/main.go index 640a882..85f1a59 100644 --- a/cmd/beacon/main.go +++ b/cmd/beacon/main.go @@ -429,6 +429,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/nodes.go b/db/nodes.go index a602cb3..cff1210 100644 --- a/db/nodes.go +++ b/db/nodes.go @@ -127,6 +127,7 @@ func (s *Store) ListNodes(ctx context.Context, nodeType int16, iatas []string, s Longitude: v.Longitude, IsObserver: v.IsObserver, ObserverID: nullableUUID(v.ObserverID), + DefaultScope: v.DefaultScopeName, KnownNeighborCount: v.KnownNeighborCount, NeighborIDs: v.NeighborIds, Stale: v.LastSeen.Valid && v.LastSeen.Time.Before(time.Now().Add(-s.staleThreshold)), diff --git a/db/nodes_test.go b/db/nodes_test.go index db9a0dd..590c08c 100644 --- a/db/nodes_test.go +++ b/db/nodes_test.go @@ -706,3 +706,15 @@ func TestDeleteOldNodes(t *testing.T) { t.Fatalf("unexpected error: %v", err) } } + +func TestListNodes_DefaultScopePreservesCaseAndMissingValue(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + scope := "#YOW" + mock.EXPECT().ListNodes(gomock.Any(), gomock.Any()).Return([]sqlc.ListNodesRow{{DefaultScopeName: &scope}, {}}, nil) + store := &Store{q: mock} + page, err := store.ListNodes(context.Background(), 0, nil, nil, nil, nil, "", "", "", 0, 10, false) + if err != nil || len(page.Items) != 2 || page.Items[0].DefaultScope == nil || *page.Items[0].DefaultScope != scope || page.Items[1].DefaultScope != nil { + t.Fatalf("scope metadata lost or invented: %+v %v", page, err) + } +} 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 2a2c470..e31d9ec 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, @@ -1422,7 +1424,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[]))) @@ -1834,3 +1840,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/route_evidence.go b/db/route_evidence.go index e7017ea..f31f50c 100644 --- a/db/route_evidence.go +++ b/db/route_evidence.go @@ -39,6 +39,16 @@ func (s *Store) GetRouteEvidence(ctx context.Context, iata, key string, query ap if c.IATA != iata || c.PathKey != key || !c.Since.Equal(query.Since) || !c.Until.Equal(query.Until) || c.ID <= 0 || c.HeardAt.Before(query.Since) || !c.HeardAt.Before(query.Until) { return nil, api.ErrRouteEvidenceInput } + if c.HashSize != 0 || c.PathBytes != "" { + if (query.HashSize != 0 || query.PathBytes != "") && (query.HashSize != c.HashSize || query.PathBytes != c.PathBytes) { + return nil, api.ErrRouteEvidenceInput + } + query.HashSize, query.PathBytes = c.HashSize, c.PathBytes + } + } + pinned := query.HashSize != 0 || query.PathBytes != "" + if pinned && !api.ValidRouteEvidencePath(query.HashSize, query.PathBytes) { + return nil, api.ErrRouteEvidenceInput } keyBytes, _ := hex.DecodeString(key) row, err := s.q.GetRouteEvidenceRoute(ctx, sqlc.GetRouteEvidenceRouteParams{Iata: iata, PathKey: keyBytes}) @@ -49,10 +59,35 @@ func (s *Store) GetRouteEvidence(ctx context.Context, iata, key string, query ap if err != nil { return nil, err } + width, path, valid := savedRoutePath(row.HashPrefix, row.HopCount) + if pinned { + selected, _ := hex.DecodeString(query.PathBytes) + hops := len(selected) / int(query.HashSize) + if hops != len(row.NodeIds) || hops != int(row.HopCount) { + return nil, api.ErrRouteEvidenceInput + } + prefixes := make([][]byte, hops) + for i, id := range row.NodeIds { + prefixes[i] = selected[i*int(query.HashSize) : (i+1)*int(query.HashSize)] + // A changed representation must still belong to this node chain. + // Reuse the already-loaded keys; never widen the evidence query. + if !valid || width != query.HashSize || !bytes.Equal(path, selected) { + node := nodes[id] + if node == nil { + return nil, api.ErrRouteEvidenceInput + } + pubkey, err := hex.DecodeString(node.PublicKey) + if err != nil || !bytes.HasPrefix(pubkey, prefixes[i]) { + return nil, api.ErrRouteEvidenceInput + } + } + } + row.HashPrefix = prefixes + width, path, valid = query.HashSize, selected, true + } route := toKnownRoutes([]knownRouteRow{{ID: row.ID, NodeIds: row.NodeIds, HashPrefix: row.HashPrefix, Iata: row.Iata, HopCount: row.HopCount, FirstSeen: row.FirstSeen, LastSeen: row.LastSeen, ObservationCount: row.ObservationCount}}, nodes)[0] route.PathKey = hex.EncodeToString(row.PathKey) out := &api.RouteEvidence{Items: []api.RouteObservation{}, Route: route, WindowStart: query.Since.UnixMilli(), WindowEnd: query.Until.UnixMilli(), GeneratedAt: time.Now().UnixMilli(), MatchType: "saved_path_prefixes"} - width, path, valid := savedRoutePath(row.HashPrefix, row.HopCount) if !valid { return out, nil } @@ -86,7 +121,7 @@ func (s *Store) GetRouteEvidence(ctx context.Context, iata, key string, query ap } if out.HasMore { last := rows[len(rows)-1] - cursor := (api.RouteEvidenceCursor{IATA: iata, PathKey: key, Since: query.Since, Until: query.Until, HeardAt: last.HeardAt.Time, ID: last.ID}).String() + cursor := (api.RouteEvidenceCursor{IATA: iata, PathKey: key, Since: query.Since, Until: query.Until, HeardAt: last.HeardAt.Time, ID: last.ID, HashSize: width, PathBytes: out.PathBytes}).String() out.NextPageCursor = &cursor } return out, nil diff --git a/db/route_evidence_integration_test.go b/db/route_evidence_integration_test.go index a84b534..23abc34 100644 --- a/db/route_evidence_integration_test.go +++ b/db/route_evidence_integration_test.go @@ -36,12 +36,13 @@ func TestRouteEvidencePostgres(t *testing.T) { applyBaseline(t, ctx, tx) _, err := tx.Exec(ctx, ` INSERT INTO observers(id,public_key,display_name) VALUES ('00000000-0000-0000-0000-000000000001','\x01','One'),('00000000-0000-0000-0000-000000000002','\x02','Two'); - INSERT INTO nodes(id,public_key,node_type,name) VALUES ('00000000-0000-0000-0000-000000000001','\xaa',2,'A'),('00000000-0000-0000-0000-000000000002','\xbb',2,'B'); + INSERT INTO nodes(id,public_key,node_type,name) VALUES ('00000000-0000-0000-0000-000000000001','\xaa0102',2,'A'),('00000000-0000-0000-0000-000000000002','\xbb0304',2,'B'); INSERT INTO packets(packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,first_heard_at,last_heard_at) SELECT int4send(i),4,0,1,'\x00','\x00',NOW(),NOW() FROM generate_series(1,10) i;`) if err != nil { t.Fatal(err) } + capture := &evidenceCapture{Tx: tx} store := &Store{q: sqlc.New(capture)} ids := []uuid.UUID{uuid.MustParse("00000000-0000-0000-0000-000000000001"), uuid.MustParse("00000000-0000-0000-0000-000000000002")} @@ -80,6 +81,36 @@ func TestRouteEvidencePostgres(t *testing.T) { if err != nil { t.Fatal(err) } + if query.Cursor.HashSize != 1 || query.Cursor.PathBytes != "aabb" { + t.Fatalf("cursor did not pin the original bytes: %+v", query.Cursor) + } + // A new representation arrives while page one is open. New reads follow it, + // but pagination and copied links retain the exact original representation. + if _, err = tx.Exec(ctx, `INSERT INTO packets(packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,first_heard_at,last_heard_at) SELECT int4send(i),4,0,1,'\x00','\x00',NOW(),NOW() FROM generate_series(12,14) i`); err != nil { + t.Fatal(err) + } + _, err = tx.Exec(ctx, `INSERT INTO packet_observations(id,packet_hash,observer_id,iata,heard_at,hash_size,hop_count,path_bytes,payload_type,path_length_byte) + SELECT i,int4send(i),'00000000-0000-0000-0000-000000000001','YOW',$1,2,2,'\xaa01bb03',4,66 FROM generate_series(12,14) i`, at) + if err != nil { + t.Fatal(err) + } + if err := store.UpsertKnownRoute(ctx, ids, [][]byte{{0xaa, 1}, {0xbb, 3}}, "YOW", 2); err != nil { + t.Fatal(err) + } + freshQuery := api.RouteEvidenceQuery{Since: since, Until: until, Limit: 2} + fresh, err := store.GetRouteEvidence(ctx, "YOW", key, freshQuery) + if err != nil || fresh.HashSize != 2 || fresh.PathBytes != "aa01bb03" || len(fresh.Items) != 2 || fresh.Items[0].ID != 14 || fresh.Route.Hops[0].HashBytes != "aa01" { + t.Fatalf("new representation lost: %+v %v", fresh, err) + } + if fresh.Route.ID != first.Route.ID || fresh.Route.PathKey != first.Route.PathKey || fresh.Route.FirstSeen != first.Route.FirstSeen || fresh.Route.ObservationCount != first.Route.ObservationCount+1 { + t.Fatalf("route identity/counters changed: %+v", fresh.Route) + } + shared := freshQuery + shared.HashSize, shared.PathBytes = first.HashSize, first.PathBytes + copyPage, err := store.GetRouteEvidence(ctx, "YOW", key, shared) + if err != nil || len(copyPage.Items) != 2 || copyPage.Items[0].ID != 3 || copyPage.Route.Hops[0].HashBytes != "aa" { + t.Fatalf("shared representation changed: %+v %v", copyPage, err) + } second, err := store.GetRouteEvidence(ctx, "YOW", key, query) if err != nil { t.Fatal(err) @@ -90,6 +121,44 @@ func TestRouteEvidencePostgres(t *testing.T) { if _, err = json.Marshal(second); err != nil { t.Fatalf("non-finite JSON: %v", err) } + for _, prefixes := range [][][]byte{{{0xaa, 1, 2}, {0xbb, 3, 4}}, {{0xaa, 1, 2}, {0xbb, 3, 4}}} { + if err := store.UpsertKnownRoute(ctx, ids, prefixes, "YOW", 2); err != nil { + t.Fatal(err) + } + } + thirdWidth, err := store.GetRouteEvidence(ctx, "YOW", key, freshQuery) + if err != nil || thirdWidth.HashSize != 3 || thirdWidth.PathBytes != "aa0102bb0304" || len(thirdWidth.Items) != 0 || thirdWidth.Route.ID != first.Route.ID || thirdWidth.Route.ObservationCount != first.Route.ObservationCount+3 { + t.Fatalf("third width or repeated upsert: %+v %v", thirdWidth, err) + } + // Missing identities or mismatched bytes cannot make a pinned query search + // a different chain. Prefix collisions likewise never widen the SQL match. + wrong := shared + wrong.PathBytes = "aacc" + if _, err := store.GetRouteEvidence(ctx, "YOW", key, wrong); !errors.Is(err, api.ErrRouteEvidenceInput) { + t.Fatalf("unrelated path accepted: %v", err) + } + wrong.PathBytes = "aabbcc" + if _, err := store.GetRouteEvidence(ctx, "YOW", key, wrong); !errors.Is(err, api.ErrRouteEvidenceInput) { + t.Fatalf("wrong hop count accepted: %v", err) + } + _, err = tx.Exec(ctx, `INSERT INTO nodes(id,public_key,node_type,name) VALUES ('00000000-0000-0000-0000-000000000003','\xaa9988',2,'Collision'); + DELETE FROM nodes WHERE id='00000000-0000-0000-0000-000000000001'`) + if err != nil { + t.Fatal(err) + } + if _, err := store.GetRouteEvidence(ctx, "YOW", key, shared); !errors.Is(err, api.ErrRouteEvidenceInput) { + t.Fatalf("missing node substituted by colliding node: %v", err) + } + if _, err = tx.Exec(ctx, `INSERT INTO nodes(id,public_key,node_type,name) VALUES ('00000000-0000-0000-0000-000000000001','\xaa0102',2,'A')`); err != nil { + t.Fatal(err) + } + copyPage, err = store.GetRouteEvidence(ctx, "YOW", key, shared) + if err != nil || len(copyPage.Items) != 2 || copyPage.Items[0].ID != 3 { + t.Fatalf("collision broadened evidence: %+v %v", copyPage, err) + } + if err := store.UpsertKnownRoute(ctx, ids, [][]byte{{0xaa}, {0xbb}}, "YOW", 2); err != nil { + t.Fatal(err) + } query.Cursor.IATA = "YVR" if _, err = store.GetRouteEvidence(ctx, "YOW", key, query); !errors.Is(err, api.ErrRouteEvidenceInput) { t.Fatal("cross-route cursor accepted") diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index 114058a..30b71ae 100644 --- a/db/sqlc/mock/querier.go +++ b/db/sqlc/mock/querier.go @@ -1137,6 +1137,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() @@ -1911,6 +1926,21 @@ func (mr *MockQuerierMockRecorder) ResolvePathHashesP4(ctx, arg any) *gomock.Cal return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ResolvePathHashesP4", reflect.TypeOf((*MockQuerier)(nil).ResolvePathHashesP4), 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 8bb7be5..c2418c7 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -156,6 +156,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) // ============================================================ @@ -317,6 +319,9 @@ type Querier interface { ResolvePathHashesP2(ctx context.Context, arg ResolvePathHashesP2Params) ([]ResolvePathHashesP2Row, error) ResolvePathHashesP3(ctx context.Context, arg ResolvePathHashesP3Params) ([]ResolvePathHashesP3Row, error) ResolvePathHashesP4(ctx context.Context, arg ResolvePathHashesP4Params) ([]ResolvePathHashesP4Row, 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 46b13b5..fcc503f 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -1405,6 +1405,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, @@ -1425,14 +1427,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, @@ -1448,6 +1453,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, @@ -2229,6 +2237,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 ` @@ -4137,7 +4198,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[]))) @@ -4165,13 +4230,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"` } // ============================================================ @@ -4204,6 +4273,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 } @@ -4856,6 +4929,63 @@ func (q *Queries) ResolvePathHashesP4(ctx context.Context, arg ResolvePathHashes 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) - } - } - confidenceRank := map[string]int{"none": 0, "ambiguous": 1, "high": 2} - type hopResult struct { - confidence string - entries []api.ResolvedPathEntry + b, _ := hex.DecodeString(h) + hashes = append(hashes, b) } - merged := make([]hopResult, len(hashes)) - for i := range merged { - merged[i] = hopResult{confidence: "none"} + candidates, err := s.traceCandidates(ctx, iatas, hashes) + if err != nil { + return nil } - 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 4260553..382f971 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -2320,9 +2320,55 @@ 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.", + "description": "Matches one complete path representation in the saved IATA. Defaults to the latest processed prefixes; hashSize/pathBytes pin a prior response. Excludes TRACE and unclassified reports; widths are never combined. Pinned prefixes must match the saved node chain or its current metadata. Byte matches do not prove hop identities or delivery. The route counter can outlive raw evidence. New cursors pin path, route, window and full timestamp precision; legacy unpinned cursors remain supported.", "produces": [ "application/json" ], @@ -2363,9 +2409,23 @@ const docTemplate = `{ "name": "until", "in": "query" }, + { + "maximum": 3, + "minimum": 1, + "type": "integer", + "description": "Exact hash width (1โ€“3); provide with pathBytes to pin a response's representation", + "name": "hashSize", + "in": "query" + }, + { + "type": "string", + "description": "Complete lower-case hex path from the response; provide with hashSize", + "name": "pathBytes", + "in": "query" + }, { "type": "string", - "description": "Opaque precise cursor from nextPageCursor; window is pinned", + "description": "Opaque precise cursor from nextPageCursor; route, window and path are pinned", "name": "pageCursor", "in": "query" }, @@ -2406,6 +2466,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.", @@ -4817,6 +4900,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" } @@ -4948,6 +5034,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" } } }, @@ -5866,6 +5959,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": { @@ -5897,6 +6013,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", @@ -5951,6 +6070,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", @@ -6007,6 +6129,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": { diff --git a/docs/swagger.json b/docs/swagger.json index fbd00b5..a6dfdd6 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -2318,9 +2318,55 @@ } } }, + "/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.", + "description": "Matches one complete path representation in the saved IATA. Defaults to the latest processed prefixes; hashSize/pathBytes pin a prior response. Excludes TRACE and unclassified reports; widths are never combined. Pinned prefixes must match the saved node chain or its current metadata. Byte matches do not prove hop identities or delivery. The route counter can outlive raw evidence. New cursors pin path, route, window and full timestamp precision; legacy unpinned cursors remain supported.", "produces": [ "application/json" ], @@ -2361,9 +2407,23 @@ "name": "until", "in": "query" }, + { + "maximum": 3, + "minimum": 1, + "type": "integer", + "description": "Exact hash width (1โ€“3); provide with pathBytes to pin a response's representation", + "name": "hashSize", + "in": "query" + }, + { + "type": "string", + "description": "Complete lower-case hex path from the response; provide with hashSize", + "name": "pathBytes", + "in": "query" + }, { "type": "string", - "description": "Opaque precise cursor from nextPageCursor; window is pinned", + "description": "Opaque precise cursor from nextPageCursor; route, window and path are pinned", "name": "pageCursor", "in": "query" }, @@ -2404,6 +2464,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.", @@ -4815,6 +4898,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" } @@ -4946,6 +5032,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" } } }, @@ -5864,6 +5957,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": { @@ -5895,6 +6011,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", @@ -5949,6 +6068,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", @@ -6005,6 +6127,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 8477b1a..0a9de10 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -880,6 +880,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 @@ -979,6 +981,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: @@ -1600,6 +1607,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: @@ -1622,6 +1644,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: @@ -1661,6 +1685,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: @@ -1698,6 +1724,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: @@ -3211,11 +3283,13 @@ paths: - Routes /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. + description: Matches one complete path representation in the saved IATA. Defaults + to the latest processed prefixes; hashSize/pathBytes pin a prior response. + Excludes TRACE and unclassified reports; widths are never combined. Pinned + prefixes must match the saved node chain or its current metadata. Byte matches + do not prove hop identities or delivery. The route counter can outlive raw + evidence. New cursors pin path, route, window and full timestamp precision; + legacy unpinned cursors remain supported. parameters: - description: Three-character IATA code in: path @@ -3241,7 +3315,20 @@ paths: in: query name: until type: integer - - description: Opaque precise cursor from nextPageCursor; window is pinned + - description: Exact hash width (1โ€“3); provide with pathBytes to pin a response's + representation + in: query + maximum: 3 + minimum: 1 + name: hashSize + type: integer + - description: Complete lower-case hex path from the response; provide with + hashSize + in: query + name: pathBytes + type: string + - description: Opaque precise cursor from nextPageCursor; route, window and + path are pinned in: query name: pageCursor type: string @@ -3354,6 +3441,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/route_evidence.go b/internal/api/handlers/route_evidence.go index a194306..3819cbe 100644 --- a/internal/api/handlers/route_evidence.go +++ b/internal/api/handlers/route_evidence.go @@ -17,7 +17,7 @@ import ( func routeEvidenceQuery(r *http.Request, iata, key string, now time.Time) (api.RouteEvidenceQuery, error) { q := api.RouteEvidenceQuery{} values := r.URL.Query() - for _, name := range []string{"pageCursor", "since", "until", "range", "limit"} { + for _, name := range []string{"pageCursor", "since", "until", "range", "limit", "hashSize", "pathBytes"} { if len(values[name]) > 1 { return q, api.ErrRouteEvidenceInput } @@ -38,6 +38,7 @@ func routeEvidenceQuery(r *http.Request, iata, key string, now time.Time) (api.R q.Cursor = c q.Since = c.Since q.Until = c.Until + q.HashSize, q.PathBytes = c.HashSize, c.PathBytes } else { q.Until = now.UTC().Truncate(time.Millisecond) q.Since = q.Until.Add(-24 * time.Hour) @@ -67,6 +68,17 @@ func routeEvidenceQuery(r *http.Request, iata, key string, now time.Time) (api.R return q, api.ErrRouteEvidenceInput } } + if values.Has("hashSize") || values.Has("pathBytes") { + width, err := strconv.ParseInt(values.Get("hashSize"), 10, 16) + path := values.Get("pathBytes") + if err != nil || !api.ValidRouteEvidencePath(int16(width), path) || strconv.FormatInt(width, 10) != values.Get("hashSize") { + return q, api.ErrRouteEvidenceInput + } + if q.Cursor != nil && q.Cursor.HashSize != 0 && (q.Cursor.HashSize != int16(width) || q.Cursor.PathBytes != path) { + return q, api.ErrRouteEvidenceInput + } + q.HashSize, q.PathBytes = int16(width), path + } if !api.ValidRouteEvidenceWindow(q.Since, q.Until) || q.Until.After(now.Add(clockSkewTolerance)) { return q, api.ErrRouteEvidenceInput } @@ -76,7 +88,7 @@ func routeEvidenceQuery(r *http.Request, iata, key string, now time.Time) (api.R // getRouteEvidence godoc // // @Summary Get retained reports matching a full saved route prefix sequence -// @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. +// @Description Matches one complete path representation in the saved IATA. Defaults to the latest processed prefixes; hashSize/pathBytes pin a prior response. Excludes TRACE and unclassified reports; widths are never combined. Pinned prefixes must match the saved node chain or its current metadata. Byte matches do not prove hop identities or delivery. The route counter can outlive raw evidence. New cursors pin path, route, window and full timestamp precision; legacy unpinned cursors remain supported. // @Tags Routes // @Produce json // @Param iata path string true "Three-character IATA code" @@ -84,7 +96,9 @@ func routeEvidenceQuery(r *http.Request, iata, key string, now time.Time) (api.R // @Param range query string false "Server-anchored duration (default 24h, max 720h); exclusive with since/until/pageCursor" // @Param since query int false "Window start epoch ms; provide with until (default last 24h)" // @Param until query int false "Exclusive window end epoch ms; maximum span 30d, end may be up to 5 minutes ahead of server time (clock skew tolerance)" -// @Param pageCursor query string false "Opaque precise cursor from nextPageCursor; window is pinned" +// @Param hashSize query int false "Exact hash width (1โ€“3); provide with pathBytes to pin a response's representation" minimum(1) maximum(3) +// @Param pathBytes query string false "Complete lower-case hex path from the response; provide with hashSize" +// @Param pageCursor query string false "Opaque precise cursor from nextPageCursor; route, window and path are pinned" // @Param limit query int false "Default 50; positive, capped at 200" minimum(1) maximum(200) // @Success 200 {object} api.RouteEvidence // @Failure 400 {object} handlers.APIError @@ -100,7 +114,7 @@ func getRouteEvidence(reader api.Reader) http.HandlerFunc { } query, err := routeEvidenceQuery(r, iata, key, time.Now()) if err != nil { - respondError(w, 400, "invalid route evidence window, limit or page cursor") + respondError(w, 400, "invalid route evidence window, path, limit or page cursor") return } result, err := reader.GetRouteEvidence(r.Context(), iata, key, query) diff --git a/internal/api/handlers/route_evidence_test.go b/internal/api/handlers/route_evidence_test.go index f2268d8..88f057d 100644 --- a/internal/api/handlers/route_evidence_test.go +++ b/internal/api/handlers/route_evidence_test.go @@ -85,3 +85,28 @@ func TestRouteEvidenceHandler(t *testing.T) { } } } + +func TestRouteEvidencePinnedQuery(t *testing.T) { + now := time.Now().UTC().Truncate(time.Millisecond) + q, err := routeEvidenceQuery(httptest.NewRequest("GET", "/?hashSize=2&pathBytes=aa01bb02", nil), "YOW", evidenceKey, now) + if err != nil || q.HashSize != 2 || q.PathBytes != "aa01bb02" { + t.Fatalf("selected path: %+v %v", q, err) + } + c := api.RouteEvidenceCursor{IATA: "YOW", PathKey: evidenceKey, Since: q.Since, Until: q.Until, HeardAt: q.Since.Add(time.Hour), ID: 5, HashSize: q.HashSize, PathBytes: q.PathBytes} + encoded := url.QueryEscape(c.String()) + next, err := routeEvidenceQuery(httptest.NewRequest("GET", "/?pageCursor="+encoded, nil), "YOW", evidenceKey, now) + if err != nil || next.HashSize != 2 || next.PathBytes != q.PathBytes { + t.Fatalf("cursor path lost: %+v %v", next, err) + } + for _, query := range []string{ + "hashSize=1", "pathBytes=aabb", "hashSize=0&pathBytes=aabb", "hashSize=4&pathBytes=aa01bb02", + "hashSize=01&pathBytes=aabb", "hashSize=1&hashSize=2&pathBytes=aabb", + "hashSize=1&pathBytes=aabb&pathBytes=aabb", "hashSize=1&pathBytes=aa", "hashSize=1&pathBytes=AABB", + "pageCursor=" + encoded + "&hashSize=1&pathBytes=aabb", + "pageCursor=" + encoded + "&hashSize=2&pathBytes=aa02bb01", + } { + if _, err := routeEvidenceQuery(httptest.NewRequest("GET", "/?"+query, nil), "YOW", evidenceKey, now); err == nil { + t.Errorf("accepted %s", query) + } + } +} diff --git a/internal/api/handlers/routes.go b/internal/api/handlers/routes.go index 86a7c86..48a147b 100644 --- a/internal/api/handlers/routes.go +++ b/internal/api/handlers/routes.go @@ -20,6 +20,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 941937e..44aca8c 100644 --- a/internal/api/handlers/stub_reader_test.go +++ b/internal/api/handlers/stub_reader_test.go @@ -419,6 +419,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..b0fa104 --- /dev/null +++ b/internal/api/handlers/topology.go @@ -0,0 +1,44 @@ +// 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 + } + 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, parseIATAs(r), 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..c4274dc --- /dev/null +++ b/internal/api/handlers/topology_test.go @@ -0,0 +1,22 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later +package handlers + +import ( + "net/http/httptest" + "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) + } + } +} 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 15b8693..a049b35 100644 --- a/internal/api/reader.go +++ b/internal/api/reader.go @@ -225,6 +225,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/route_evidence.go b/internal/api/route_evidence.go index d320ade..490c91b 100644 --- a/internal/api/route_evidence.go +++ b/internal/api/route_evidence.go @@ -50,16 +50,22 @@ type RouteEvidenceQuery struct { Until time.Time Cursor *RouteEvidenceCursor Limit int32 + // Optional exact representation copied from a prior response or cursor. + HashSize int16 + PathBytes string } -// RouteEvidenceCursor binds the precise (heard_at DESC,id DESC) boundary to its route/window. +// RouteEvidenceCursor pins the route, window and exact representation as well as +// the precise (heard_at DESC,id DESC) boundary. Legacy v1 cursors have no path pin. type RouteEvidenceCursor struct { - IATA string - PathKey string - Since time.Time - Until time.Time - HeardAt time.Time - ID int64 + IATA string + PathKey string + Since time.Time + Until time.Time + HeardAt time.Time + ID int64 + HashSize int16 + PathBytes string } func ValidRouteEvidenceKey(iata, key string) bool { @@ -79,20 +85,32 @@ func ValidRouteEvidenceWindow(since, until time.Time) bool { return !since.Before(time.Unix(0, 0)) && until.After(since) && until.Sub(since) <= MaxRouteEvidenceWindow } +func ValidRouteEvidencePath(width int16, path string) bool { + if width < 1 || width > 3 || len(path) < 4*int(width) || len(path) > 126*int(width) || len(path)%(2*int(width)) != 0 { + return false + } + _, err := hex.DecodeString(path) + return err == nil && path == strings.ToLower(path) +} + func (c RouteEvidenceCursor) String() string { - return fmt.Sprintf("v1:%s:%s:%d:%d:%d:%d", c.IATA, c.PathKey, c.Since.UnixMilli(), c.Until.UnixMilli(), c.HeardAt.UnixMicro(), c.ID) + value := fmt.Sprintf("%s:%s:%d:%d:%d:%d", c.IATA, c.PathKey, c.Since.UnixMilli(), c.Until.UnixMilli(), c.HeardAt.UnixMicro(), c.ID) + if c.HashSize != 0 || c.PathBytes != "" { + return fmt.Sprintf("v2:%s:%d:%s", value, c.HashSize, c.PathBytes) + } + return "v1:" + value } func ParseRouteEvidenceCursor(raw string) (*RouteEvidenceCursor, error) { - if len(raw) > 200 { + if len(raw) > 512 { return nil, ErrRouteEvidenceInput } p := strings.Split(raw, ":") - if len(p) != 7 || p[0] != "v1" || !ValidRouteEvidenceKey(p[1], p[2]) { + if !((len(p) == 7 && p[0] == "v1") || (len(p) == 9 && p[0] == "v2")) || !ValidRouteEvidenceKey(p[1], p[2]) { return nil, ErrRouteEvidenceInput } values := [4]int64{} - for i, s := range p[3:] { + for i, s := range p[3:7] { n, err := strconv.ParseInt(s, 10, 64) if err != nil || n < 0 || strconv.FormatInt(n, 10) != s { return nil, ErrRouteEvidenceInput @@ -104,6 +122,13 @@ func ParseRouteEvidenceCursor(raw string) (*RouteEvidenceCursor, error) { return nil, ErrRouteEvidenceInput } c := &RouteEvidenceCursor{IATA: p[1], PathKey: p[2], Since: time.UnixMilli(values[0]).UTC(), Until: time.UnixMilli(values[1]).UTC(), HeardAt: time.UnixMicro(values[2]).UTC(), ID: values[3]} + if p[0] == "v2" { + width, err := strconv.ParseInt(p[7], 10, 16) + if err != nil || strconv.FormatInt(width, 10) != p[7] || !ValidRouteEvidencePath(int16(width), p[8]) { + return nil, ErrRouteEvidenceInput + } + c.HashSize, c.PathBytes = int16(width), p[8] + } if !ValidRouteEvidenceWindow(c.Since, c.Until) || c.HeardAt.Before(c.Since) || !c.HeardAt.Before(c.Until) { return nil, ErrRouteEvidenceInput } diff --git a/internal/api/route_evidence_test.go b/internal/api/route_evidence_test.go index 42c9a73..a025a44 100644 --- a/internal/api/route_evidence_test.go +++ b/internal/api/route_evidence_test.go @@ -23,3 +23,23 @@ func TestRouteEvidenceCursor(t *testing.T) { } } } + +func TestRouteEvidencePinnedCursor(t *testing.T) { + c := RouteEvidenceCursor{IATA: "YOW", PathKey: strings.Repeat("a", 32), Since: time.UnixMilli(1000).UTC(), Until: time.UnixMilli(3000).UTC(), HeardAt: time.UnixMicro(2000123).UTC(), ID: 9, HashSize: 3, PathBytes: strings.Repeat("aabbcc", 63)} + got, err := ParseRouteEvidenceCursor(c.String()) + if err != nil || *got != c || !strings.HasPrefix(c.String(), "v2:") { + t.Fatalf("pinned cursor round-trip: %+v %v", got, err) + } + for _, bad := range []string{ + strings.Replace(c.String(), ":3:", ":4:", 1), + strings.Replace(c.String(), ":3:", ":03:", 1), + c.String() + "00", + strings.TrimSuffix(c.String(), c.PathBytes) + "aabbcc", + strings.ToUpper(c.String()), + strings.TrimSuffix(c.String(), c.PathBytes) + "zzzzzzzzzzzz", + } { + if _, err := ParseRouteEvidenceCursor(bad); err == nil { + t.Errorf("accepted invalid pinned cursor %q", bad) + } + } +} diff --git a/internal/api/router/options.go b/internal/api/router/options.go index 9145091..51689e3 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 45d7a88..a1db19a 100644 --- a/internal/api/router/router.go +++ b/internal/api/router/router.go @@ -115,6 +115,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 e00b124..9e6a48a 100644 --- a/internal/cache/cache_test.go +++ b/internal/cache/cache_test.go @@ -447,6 +447,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 e032daf..a06e9d6 100644 --- a/internal/cache/reader.go +++ b/internal/cache/reader.go @@ -379,6 +379,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..e885753 --- /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) < 10 { + 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..f811ae3 --- /dev/null +++ b/internal/tracequality/quality_test.go @@ -0,0 +1,51 @@ +package tracequality + +import ( + "bytes" + "slices" + "testing" +) + +func TestTraceFormat(t *testing.T) { + for flags := byte(0); flags < 4; flags++ { + raw := make([]byte, 9) + raw[8] = flags + raw = append(raw, bytes.Repeat([]byte{0xaa}, 3*(1<