From 04fc0266d13a8dec8135da49a94f5fcf1db3bea9 Mon Sep 17 00:00:00 2001 From: n30nex Date: Wed, 9 Sep 2026 18:45:52 -0400 Subject: [PATCH] fix(packets): preserve captured endpoint names in history (#128) --- db/endpoint_matching_integration_test.go | 219 +++++++++++++++ .../026_observation_endpoint_snapshots.sql | 9 + db/packet_endpoints_integration_test.go | 257 ++++++++++++++++++ db/packets.go | 103 ++++--- db/packets_test.go | 2 +- db/queries/queries.sql | 31 ++- db/sqlc/mock/querier.go | 15 + db/sqlc/models.go | 1 + db/sqlc/querier.go | 5 + db/sqlc/queries.sql.go | 182 +++++++++---- db/store.go | 26 ++ docs/docs.go | 4 +- docs/swagger.json | 4 +- docs/swagger.yaml | 14 +- internal/api/packets.go | 29 +- internal/api/packets_test.go | 22 ++ internal/ingest/endpoint_matching_test.go | 50 ++++ internal/ingest/ingest.go | 3 + internal/ingest/ingest_test.go | 4 + internal/ingest/packet.go | 57 ++-- internal/ingest/packet_endpoints_test.go | 123 +++++++++ 21 files changed, 1018 insertions(+), 142 deletions(-) create mode 100644 db/endpoint_matching_integration_test.go create mode 100644 db/migrations/026_observation_endpoint_snapshots.sql create mode 100644 db/packet_endpoints_integration_test.go create mode 100644 internal/ingest/endpoint_matching_test.go create mode 100644 internal/ingest/packet_endpoints_test.go diff --git a/db/endpoint_matching_integration_test.go b/db/endpoint_matching_integration_test.go new file mode 100644 index 0000000..734c208 --- /dev/null +++ b/db/endpoint_matching_integration_test.go @@ -0,0 +1,219 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package db + +import ( + "context" + "encoding/json" + "fmt" + "os" + "strings" + "testing" + "time" + + 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" +) + +type endpointMatchQueries struct { + sqlc.DBTX + calls int + query string +} + +func (q *endpointMatchQueries) Query(ctx context.Context, query string, args ...any) (pgx.Rows, error) { + q.calls++ + q.query = query + return q.DBTX.Query(ctx, query, args...) +} + +func TestPacketEndpointRolesPostgres(t *testing.T) { + dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN") + if dsn == "" { + t.Skip("set BEACON_TEST_POSTGRES_DSN for the PostgreSQL regression test") + } + ctx, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + conn, err := pgx.Connect(ctx, dsn) + if err != nil { + t.Fatal(err) + } + defer conn.Close(context.Background()) + tx, err := conn.Begin(ctx) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback(context.Background()) + exec := func(sql string, args ...any) { + t.Helper() + if _, err := tx.Exec(ctx, sql, args...); err != nil { + t.Fatal(err) + } + } + for _, table := range []string{"packets", "packet_observations", "observers", "transport_scopes", "channel_messages", "nodes", "node_short_ids"} { + exec("CREATE TEMP TABLE " + table + " (LIKE public." + table + " INCLUDING ALL) ON COMMIT DROP") + } + id := func(i int) uuid.UUID { return uuid.MustParse(fmt.Sprintf("00000000-0000-0000-0000-%012d", i)) } + exec(`INSERT INTO nodes(id,public_key,name,node_type) +SELECT ('00000000-0000-0000-0000-'||lpad(i::text,12,'0'))::uuid, + decode(prefix||repeat('00',28),'hex'),name,role +FROM (VALUES (1,'aa000001','Companion',1),(2,'aa000002','Repeater',2), + (3,'ee000003','Room',3),(4,'aa000004','Sensor',4),(5,'aa000005','Unknown role',0), + (6,'bb000006','Companion only',1),(7,'aa000007','Other region',1), + (8,'cc000008','Relay',2),(9,'dd000009',NULL,1),(10,'aa000010','No membership',1)) v(i,prefix,name,role); +INSERT INTO node_short_ids(node_id,iata,prefix_4) +SELECT id,CASE WHEN name='Other region' THEN 'YYZ' ELSE 'YVR' END,substring(public_key from 1 for 4) +FROM nodes WHERE name IS DISTINCT FROM 'No membership';`) + exec("INSERT INTO observers(id,public_key) VALUES ($1,'\\x01'),($2,'\\x02')", id(101), id(102)) + queries := &endpointMatchQueries{DBTX: tx} + store := &Store{q: sqlc.New(queries)} + check := func(hop *api.ResolvedHop, confidence string, members ...int) { + t.Helper() + if hop == nil || hop.Confidence != confidence || len(hop.Nodes) != len(members) { + t.Errorf("endpoint candidate roles incorrect: confidence=%s, want %s with %d nodes", func() string { + if hop == nil { + return "nil" + } + return hop.Confidence + }(), confidence, len(members)) + return + } + want := make(map[uuid.UUID]bool) + for _, member := range members { + want[id(member)] = true + } + for _, node := range hop.Nodes { + if !want[node.ID] { + t.Errorf("unexpected endpoint node %s", node.ID) + } + delete(want, node.ID) + } + if len(want) != 0 { + t.Error("endpoint candidate missing") + } + } + for i, kind := range []int{0, 1, 2, 8} { + // These encrypted envelopes carry one-byte destination/source prefixes. + hash := []byte{byte(i + 1)} + payload := append([]byte{0xbb, 0xaa, 0, 0}, make([]byte, 16)...) + exec(`INSERT INTO packets(packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,first_heard_at,last_heard_at) +VALUES ($1,$2,0,1,$3,$4,NOW(),NOW())`, hash, kind, payload, []byte{byte(kind<<2 | 1)}) + exec(`INSERT INTO packet_observations(id,packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count,path_bytes,source_broker) +VALUES ($1,$2,$3,'YVR',NOW(),1,1,1,'\xcc','fixture')`, i+1, hash, id(101)) + packet, err := store.GetPacket(ctx, hash) + if err != nil || len(packet.Observations) != 1 { + t.Fatalf("packet detail: %v", err) + } + observation := packet.Observations[0] + check(observation.ResolvedSource, "ambiguous", 1, 2, 4, 5) + check(observation.ResolvedDestination, "high", 6) + if len(observation.ResolvedPath) != 1 { + t.Fatal("relay path missing") + } + check(&observation.ResolvedPath[0], "high", 8) + } + exec(`INSERT INTO packet_observations(id,packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count,source_broker) +VALUES (10,'\x01',$1,'YYZ',NOW()+interval '1 second',0,1,0,'fixture')`, id(102)) + packet, err := store.GetPacket(ctx, []byte{1}) + if err != nil || len(packet.Observations) != 2 { + t.Fatalf("regional observations: %v", err) + } + check(packet.Observations[1].ResolvedSource, "high", 7) + check(packet.Observations[1].ResolvedDestination, "none") + // Intermediate relay matching keeps its existing infrastructure-only boundary. + for width := 1; width <= 4; width++ { + hash := []byte{0xaa, 0, 0, 2}[:width] + resolved, err := store.ResolvePathHashes(ctx, "YVR", [][]byte{hash}) + if err != nil { + t.Fatal(err) + } + hops := api.BuildResolvedPath([][]byte{hash}, resolved) + check(&hops[0], "high", 2) + } + t.Log("endpoint roles and unchanged relay matching checked across four payload types") + for _, hashes := range [][][]byte{nil, {}, {{}}, {{0xaa, 0xbb}}, {{0xaa}, {0xbb, 0xcc}}} { + queries.calls = 0 + got, err := store.ResolveEndpointHashes(ctx, "YVR", hashes) + if err != nil || len(got) != 0 || queries.calls != 0 { + t.Fatal("invalid endpoint width reached the database") + } + } + queries.calls = 0 + resolved, err := store.ResolveEndpointHashes(ctx, "YVR", [][]byte{{0xee}, {0xdd}, {0xff}, {0xbb}, {0xbb}}) + if err != nil || queries.calls != 1 { + t.Fatalf("batched endpoint query: %v, calls=%d", err, queries.calls) + } + hops := api.BuildResolvedPath([][]byte{{0xee}, {0xdd}, {0xff}, {0xbb}}, resolved) + check(&hops[0], "high", 3) + check(&hops[1], "high", 9) + if hops[1].Nodes[0].Name != nil { + t.Error("unnamed node acquired a name") + } + check(&hops[2], "none") + check(&hops[3], "high", 6) + missing, err := store.ResolveEndpointHashes(ctx, "ZZZ", [][]byte{{0xaa}}) + if err != nil || len(missing) != 0 { + t.Fatal("endpoint escaped its observation region") + } + + // A large nonmatching population must not turn a short-hash lookup into a full scan. + exec(`INSERT INTO nodes(id,public_key,node_type) +SELECT ('00000000-0000-0000-0000-'||lpad(i::text,12,'0'))::uuid, + decode(lpad(to_hex(i),8,'0')||repeat('00',28),'hex'),1 FROM generate_series(10000,29999) i; +INSERT INTO node_short_ids(node_id,iata,prefix_4) +SELECT id,'YVR',substring(public_key from 1 for 4) FROM nodes WHERE substring(public_key from 1 for 1)='\x00'; +ANALYZE nodes; +ANALYZE node_short_ids;`) + if _, err := store.ResolveEndpointHashes(ctx, "YVR", [][]byte{{0xaa}}); err != nil { + t.Fatal(err) + } + exec("PREPARE endpoint_role_plan AS " + queries.query) + defer tx.Exec(context.Background(), "DEALLOCATE endpoint_role_plan") + for _, mode := range []string{"force_custom_plan", "force_generic_plan"} { + exec("SET LOCAL plan_cache_mode=" + mode) + var raw []byte + if err := tx.QueryRow(ctx, `EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON) EXECUTE endpoint_role_plan('YVR',ARRAY['\xaa'::bytea])`).Scan(&raw); err != nil { + t.Fatal(err) + } + var report []struct { + Plan map[string]any + Milliseconds float64 `json:"Execution Time"` + } + if err := json.Unmarshal(raw, &report); err != nil { + t.Fatal(err) + } + var examined, nodeRows float64 + indexed := false + var walk func(map[string]any) + walk = func(plan map[string]any) { + if cond, ok := plan["Index Cond"].(string); ok && strings.Contains(cond, "prefix_1") && strings.Contains(cond, "iata") { + indexed = true + } + if plan["Relation Name"] == "node_short_ids" { + rows, _ := plan["Actual Rows"].(float64) + removed, _ := plan["Rows Removed by Filter"].(float64) + loops, _ := plan["Actual Loops"].(float64) + examined += (rows + removed) * loops + } + if plan["Relation Name"] == "nodes" { + rows, _ := plan["Actual Rows"].(float64) + removed, _ := plan["Rows Removed by Filter"].(float64) + loops, _ := plan["Actual Loops"].(float64) + nodeRows += (rows + removed) * loops + } + if children, ok := plan["Plans"].([]any); ok { + for _, child := range children { + walk(child.(map[string]any)) + } + } + } + walk(report[0].Plan) + if !indexed || examined > 10 || nodeRows > 10 { + t.Errorf("%s endpoint lookup scanned %.0f short IDs and %.0f nodes, indexed=%v", mode, examined, nodeRows, indexed) + } + t.Logf("%s: %.0f short-ID rows and %.0f node rows examined in %.3f ms", mode, examined, nodeRows, report[0].Milliseconds) + } +} diff --git a/db/migrations/026_observation_endpoint_snapshots.sql b/db/migrations/026_observation_endpoint_snapshots.sql new file mode 100644 index 0000000..77a163b --- /dev/null +++ b/db/migrations/026_observation_endpoint_snapshots.sql @@ -0,0 +1,9 @@ +-- Copyright 2026 Beacon Contributors +-- SPDX-License-Identifier: AGPL-3.0-or-later + +-- Preserve the endpoint resolution already computed for the live observation. +-- SQL NULL means no endpoint nodes were captured; empty legacy captures also +-- permit detail-time lookup once the node is known (for example, first adverts). +-- No default/backfill: current node metadata cannot reconstruct past names. +ALTER TABLE packet_observations + ADD COLUMN IF NOT EXISTS resolved_endpoints JSONB; diff --git a/db/packet_endpoints_integration_test.go b/db/packet_endpoints_integration_test.go new file mode 100644 index 0000000..756d9cd --- /dev/null +++ b/db/packet_endpoints_integration_test.go @@ -0,0 +1,257 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package db + +import ( + "context" + "encoding/json" + "fmt" + "os" + "reflect" + "testing" + "time" + + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/MeshCore-Beacon/beacon-server/internal/ingest" + "github.com/google/uuid" + "github.com/jackc/pgx/v5" +) + +type endpointQueryCounter struct { + sqlc.DBTX + calls int +} + +func (q *endpointQueryCounter) Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) { + q.calls++ + return q.DBTX.Query(ctx, sql, args...) +} +func (q *endpointQueryCounter) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row { + q.calls++ + return q.DBTX.QueryRow(ctx, sql, args...) +} + +func TestPacketEndpointSnapshotsPostgres(t *testing.T) { + dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN") + if dsn == "" { + t.Skip("set BEACON_TEST_POSTGRES_DSN for the PostgreSQL regression test") + } + ctx, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + conn, err := pgx.Connect(ctx, dsn) + if err != nil { + t.Fatal(err) + } + defer conn.Close(context.Background()) + tx, err := conn.Begin(ctx) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback(context.Background()) + exec := func(query string, args ...any) { + t.Helper() + if _, err := tx.Exec(ctx, query, args...); err != nil { + t.Fatal(err) + } + } + for _, table := range []string{"packets", "packet_observations", "observers", "transport_scopes", "nodes", "channel_messages"} { + exec("CREATE TEMP TABLE " + table + " (LIKE public." + table + " INCLUDING ALL) ON COMMIT DROP") + } + // Also lets the unchanged Store run against the regression fixture before migration. + exec("ALTER TABLE pg_temp.packet_observations ADD COLUMN IF NOT EXISTS resolved_endpoints JSONB") + exec("CREATE TEMP SEQUENCE endpoint_observation_id START 1000") + exec("ALTER TABLE pg_temp.packet_observations ALTER COLUMN id SET DEFAULT nextval('pg_temp.endpoint_observation_id')") + observer1 := uuid.MustParse("00000000-0000-0000-0000-000000000001") + observer2 := uuid.MustParse("00000000-0000-0000-0000-000000000002") + nodeID := uuid.MustParse("00000000-0000-0000-0000-000000000003") + key := "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + name, earlier := "Captured 👋 companion", "Earlier companion" + high := &api.ResolvedHop{Confidence: "high", Nodes: []api.ResolvedNode{{ID: nodeID, Name: &name, PublicKey: key}}} + old := &api.ResolvedHop{Confidence: "high", Nodes: []api.ResolvedNode{{ID: nodeID, Name: &earlier, PublicKey: key}}} + ambiguous := &api.ResolvedHop{Confidence: "ambiguous", Nodes: []api.ResolvedNode{high.Nodes[0], {ID: observer2, PublicKey: "bb"}}} + none := &api.ResolvedHop{Confidence: "none", Nodes: []api.ResolvedNode{}} + snapshot := func(source, destination *api.ResolvedHop) []byte { + value, err := json.Marshal(struct { + Source *api.ResolvedHop `json:"source,omitempty"` + Destination *api.ResolvedHop `json:"destination,omitempty"` + }{source, destination}) + if err != nil { + t.Fatal(err) + } + return value + } + check := func(source, destination, wantSource, wantDestination *api.ResolvedHop) { + t.Helper() + if !reflect.DeepEqual(source, wantSource) || !reflect.DeepEqual(destination, wantDestination) { + t.Error("historical packet lost endpoint snapshot") + } + } + exec("INSERT INTO observers(id,public_key) VALUES ($1,'\\x01'),($2,'\\x02')", observer1, observer2) + exec("INSERT INTO nodes(id,public_key,name,node_type) VALUES ($1,decode($2,'hex'),$3,1)", nodeID, key, name) + anchor := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + exec(`INSERT INTO packets(packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,origin_pubkey,first_heard_at,last_heard_at) +SELECT decode(lpad(to_hex(i),64,'0'),'hex'),kind,0,1,'\x00','\x00',decode($2,'hex'), + $1::timestamptz+i*interval '1 second',$1::timestamptz+i*interval '1 second' +FROM (VALUES (1,4),(2,2),(3,3),(4,4),(5,3)) v(i,kind)`, anchor, key) + for _, obs := range []struct { + id, packet int + observer uuid.UUID + iata string + data []byte + }{ + {1, 1, observer1, "YYZ", snapshot(old, nil)}, + {2, 1, observer2, "YVR", snapshot(high, nil)}, + {3, 2, observer1, "YYZ", snapshot(ambiguous, none)}, + {4, 3, observer2, "YVR", snapshot(nil, nil)}, + {5, 4, observer1, "YYZ", nil}, + } { + exec(`INSERT INTO packet_observations(id,packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count,source_broker,resolved_endpoints) +VALUES ($1,decode($2,'hex'),$3,$4,$5,0,1,0,'fixture',$6)`, obs.id, fmt.Sprintf("%064x", obs.packet), obs.observer, obs.iata, anchor.Add(time.Duration(obs.id)*time.Second), obs.data) + } + exec("UPDATE nodes SET name='Current name'") + q := &endpointQueryCounter{DBTX: tx} + store := &Store{q: sqlc.New(q)} + for _, iatas := range [][]string{nil, {"YYZ"}, {"YVR"}, {"YYZ", "YVR", "YYZ"}} { + q.calls = 0 + page, err := store.ListPackets(ctx, nil, nil, iatas, nil, time.Time{}, time.Time{}, 0, 50) + if err != nil { + t.Fatal(err) + } + if q.calls != 1 || len(page.Items) == 0 { + t.Fatalf("list used %d queries or lost its page", q.calls) + } + for _, item := range page.Items { + switch item.PacketHash { + case fmt.Sprintf("%064x", 1): + // The existing regional roll-up also selects the global latest observer. + if item.LatestObserver == nil || item.LatestObserver.ID != observer2 || item.LatestObserver.IATA != "YVR" { + t.Fatal("latest observer selection changed") + } + check(item.LatestObserver.ResolvedSource, item.LatestObserver.ResolvedDestination, high, nil) + case fmt.Sprintf("%064x", 2): + check(item.LatestObserver.ResolvedSource, item.LatestObserver.ResolvedDestination, ambiguous, none) + case fmt.Sprintf("%064x", 3), fmt.Sprintf("%064x", 4): + check(item.LatestObserver.ResolvedSource, item.LatestObserver.ResolvedDestination, nil, nil) + case fmt.Sprintf("%064x", 5): + if item.LatestObserver != nil { + t.Fatal("invented an observation for a packet without one") + } + } + } + } + q.calls = 0 + backfill, err := store.ListPacketsAfterID(ctx, 0, -1, -1, nil, "", 50) + if err != nil || len(backfill) != 5 || q.calls != 1 { + t.Fatalf("backfill: rows=%d queries=%d err=%v", len(backfill), q.calls, err) + } + check(backfill[0].LatestObserver.ResolvedSource, backfill[0].LatestObserver.ResolvedDestination, old, nil) + check(backfill[1].LatestObserver.ResolvedSource, backfill[1].LatestObserver.ResolvedDestination, high, nil) + check(backfill[2].LatestObserver.ResolvedSource, backfill[2].LatestObserver.ResolvedDestination, ambiguous, none) + packet1 := append(make([]byte, 31), 1) + for pass := 0; pass < 2; pass++ { + q.calls = 0 + packet, err := store.GetPacket(ctx, packet1) + if err != nil || len(packet.Observations) != 2 { + t.Fatalf("packet detail: %v", err) + } + check(packet.Observations[0].ResolvedSource, packet.Observations[0].ResolvedDestination, old, nil) + check(packet.Observations[1].ResolvedSource, packet.Observations[1].ResolvedDestination, high, nil) + if q.calls != 2 { + t.Errorf("snapshot detail used %d queries, want packet plus observations only", q.calls) + } + if pass == 0 { + legacy, err := store.GetPacket(ctx, append(make([]byte, 31), 4)) + if err != nil || legacy.Observations[0].ResolvedSource == nil || *legacy.Observations[0].ResolvedSource.Nodes[0].Name != "Current name" { + t.Fatalf("legacy detail lookup changed: %v", err) + } + exec("DELETE FROM nodes") + } + } + exec(`INSERT INTO packets(packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,first_heard_at,last_heard_at) +SELECT decode(lpad(to_hex(i),64,'0'),'hex'),4,0,1,'\x00','\x00',$1::timestamptz+i*interval '1 second',$1::timestamptz+i*interval '1 second' +FROM generate_series(6,55) i`, anchor) + exec(`INSERT INTO packet_observations(id,packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count,source_broker,resolved_endpoints) +SELECT i,decode(lpad(to_hex(i),64,'0'),'hex'),$1,'YYZ',$2::timestamptz+i*interval '1 second',0,1,0,'fixture',$3 +FROM generate_series(6,55) i`, observer1, anchor, snapshot(high, nil)) + q.calls = 0 + page, err := store.ListPackets(ctx, nil, nil, nil, nil, time.Time{}, time.Time{}, 0, 50) + if err != nil || len(page.Items) != 50 || q.calls != 1 { + t.Fatalf("50-row page: queries=%d err=%v", q.calls, err) + } + for _, item := range page.Items { + check(item.LatestObserver.ResolvedSource, item.LatestObserver.ResolvedDestination, high, nil) + } + t.Logf("50 historical rows use %d database query", q.calls) + + // Capture uses the ordinary insert; duplicate delivery cannot replace its snapshot. + exec(`INSERT INTO packets(packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,first_heard_at,last_heard_at) +VALUES (decode($1,'hex'),4,0,1,'\x00','\x00',NOW(),NOW())`, fmt.Sprintf("%064x", 56)) + observation := ingest.InsertObservationParams{PacketHash: append(make([]byte, 31), 56), + ObserverID: observer1, IATA: "YYZ", HeardAt: anchor, HashSize: 1, SourceBroker: "fixture", + PayloadType: 4, ResolvedEndpoints: snapshot(high, none)} + for _, wantInserted := range []bool{true, false} { + q.calls = 0 + inserted, err := store.InsertObservation(ctx, observation) + if err != nil || inserted != wantInserted || q.calls != 1 { + t.Fatalf("snapshot insert: inserted=%v queries=%d err=%v", inserted, q.calls, err) + } + observation.ResolvedEndpoints = snapshot(old, nil) + observation.IATA = "YVR" + } + stored, err := store.GetPacket(ctx, observation.PacketHash) + if err != nil || len(stored.Observations) != 1 || stored.Observations[0].IATA != "YYZ" { + t.Fatalf("duplicate observation changed: %v", err) + } + check(stored.Observations[0].ResolvedSource, stored.Observations[0].ResolvedDestination, high, none) + // First adverts can be captured before their node exists. Empty old captures + // must allow detail-time lookup once that node has been advertised. + exec("INSERT INTO nodes(id,public_key,name,node_type) VALUES ($1,decode($2,'hex'),'Later discovered companion',1)", nodeID, key) + exec("UPDATE packets SET origin_pubkey=decode($2,'hex') WHERE packet_hash=$1", observation.PacketHash, key) + for _, raw := range []string{`{}`, `{"source":{"confidence":"none","nodes":[]}}`} { + exec("UPDATE packet_observations SET resolved_endpoints=$2::jsonb,source_broker=NULL WHERE packet_hash=$1", observation.PacketHash, raw) + q.calls = 0 + stored, err = store.GetPacket(ctx, observation.PacketHash) + if err != nil || q.calls != 4 || stored.Observations[0].SourceBroker != "" { + t.Fatalf("empty snapshot fallback or nullable legacy broker: queries=%d err=%v", q.calls, err) + } + if source := stored.Observations[0].ResolvedSource; source == nil || len(source.Nodes) != 1 || source.Nodes[0].Name == nil || *source.Nodes[0].Name != "Later discovered companion" { + t.Fatal("empty capture suppressed the later advertised node") + } + } + exec(`INSERT INTO packets(packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,origin_pubkey,first_heard_at,last_heard_at) +VALUES (decode($1,'hex'),4,0,1,'\x00','\x00',decode($2,'hex'),$3,$3)`, fmt.Sprintf("%064x", 57), key, anchor) + observation.PacketHash, observation.ResolvedEndpoints = append(make([]byte, 31), 57), nil + inserted, err := store.InsertObservation(ctx, observation) + if err != nil || !inserted { + t.Fatalf("uncaptured observation insert: %v", err) + } + var isNull bool + if err := tx.QueryRow(ctx, "SELECT resolved_endpoints IS NULL FROM packet_observations WHERE packet_hash=$1", observation.PacketHash).Scan(&isNull); err != nil || !isNull { + t.Fatalf("uncaptured resolution was not SQL NULL: %v", err) + } + empty, err := store.GetPacket(ctx, append(make([]byte, 31), 5)) + if err != nil || len(empty.Observations) != 0 { + t.Fatalf("packet awaiting its first observation: %v", err) + } +} + +func TestDecodePacketEndpointSnapshot(t *testing.T) { + for _, tc := range []struct { + raw string + captured bool + }{ + {"", false}, {"null", false}, {"[", false}, {"[]", false}, {"{}", false}, + {`{"source":{"confidence":"none","nodes":[]}}`, false}, + {`{"destination":{"confidence":"none","nodes":[]}}`, false}, + {`{"source":{},"destination":null}`, false}, + {`{"source":{"confidence":"high","nodes":[{"publicKey":"aa"}]}}`, true}, + {`{"destination":{"confidence":"ambiguous","nodes":[{"publicKey":"aa"},{"publicKey":"ab"}]}}`, true}, + } { + _, captured := decodePacketEndpointSnapshot(json.RawMessage(tc.raw)) + if captured != tc.captured { + t.Errorf("snapshot %q: captured=%v, want %v", tc.raw, captured, tc.captured) + } + } +} diff --git a/db/packets.go b/db/packets.go index f806a42..b03eda6 100644 --- a/db/packets.go +++ b/db/packets.go @@ -84,6 +84,14 @@ func buildLatestObserverPath(pathLengthByte, hashSize, hopCount *int16, pathByte return pathLength, pathBytesHex } +func decodePacketEndpointSnapshot(raw json.RawMessage) (api.PacketEndpointSnapshot, bool) { + var snapshot *api.PacketEndpointSnapshot + if len(raw) == 0 || json.Unmarshal(raw, &snapshot) != nil || snapshot == nil || !snapshot.HasResolvedNodes() { + return api.PacketEndpointSnapshot{}, false + } + return *snapshot, true +} + func (s *Store) ListPackets(ctx context.Context, payloadTypes, routeTypes []int16, iatas []string, scopes []string, since, until time.Time, cursor int64, limit int32) (api.Page[api.PacketSummary], error) { if len(iatas) > 0 { return s.listPacketsByIATAs(ctx, payloadTypes, routeTypes, iatas, scopes, since, until, cursor, limit) @@ -130,10 +138,13 @@ func (s *Store) ListPackets(ctx context.Context, payloadTypes, routeTypes []int1 ObservationCount: int32(v.ObservationCount), } if v.LatestObserverID != (uuid.UUID{}) { + endpoints, _ := decodePacketEndpointSnapshot(v.LatestObserverResolvedEndpoints) item.LatestObserver = &api.PacketLatestObserver{ - ID: v.LatestObserverID, - DisplayName: v.LatestObserverName, - IATA: v.LatestObserverIata, + ID: v.LatestObserverID, + DisplayName: v.LatestObserverName, + IATA: v.LatestObserverIata, + ResolvedSource: endpoints.Source, + ResolvedDestination: endpoints.Destination, } item.LatestObserver.PathLength, item.LatestObserver.PathBytes = buildLatestObserverPath( &v.LatestObserverPathLengthByte, &v.LatestObserverHashSize, &v.LatestObserverHopCount, v.LatestObserverPathBytes, @@ -209,10 +220,13 @@ func (s *Store) listPacketsByIATAs(ctx context.Context, payloadTypes, routeTypes ObservationCount: int32(v.ObservationCount), } if v.LatestObserverID != (uuid.UUID{}) { + endpoints, _ := decodePacketEndpointSnapshot(v.LatestObserverResolvedEndpoints) item.LatestObserver = &api.PacketLatestObserver{ - ID: v.LatestObserverID, - DisplayName: v.LatestObserverName, - IATA: v.LatestObserverIata, + ID: v.LatestObserverID, + DisplayName: v.LatestObserverName, + IATA: v.LatestObserverIata, + ResolvedSource: endpoints.Source, + ResolvedDestination: endpoints.Destination, } item.LatestObserver.PathLength, item.LatestObserver.PathBytes = buildLatestObserverPath( &v.LatestObserverPathLengthByte, &v.LatestObserverHashSize, &v.LatestObserverHopCount, v.LatestObserverPathBytes, @@ -265,10 +279,13 @@ func (s *Store) ListPacketsAfterID(ctx context.Context, afterObservationID int64 ObservationCount: int32(v.ObservationCount), } if v.LatestObserverID != (uuid.UUID{}) { + endpoints, _ := decodePacketEndpointSnapshot(v.LatestObserverResolvedEndpoints) item.LatestObserver = &api.PacketLatestObserver{ - ID: v.LatestObserverID, - DisplayName: v.LatestObserverName, - IATA: v.LatestObserverIata, + ID: v.LatestObserverID, + DisplayName: v.LatestObserverName, + IATA: v.LatestObserverIata, + ResolvedSource: endpoints.Source, + ResolvedDestination: endpoints.Destination, } // Inner join here (unlike ListPackets/listPacketsByIATAs' LEFT JOIN LATERAL), so // these are never nil when an observer was joined at all. @@ -343,7 +360,10 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet, ObservationCount: int32(len(obsRows)), Observations: make([]api.PacketObservationDetail, 0, len(obsRows)), } - minHeardAt := obsRows[0].HeardAt.Time + var minHeardAt time.Time + if len(obsRows) > 0 { + minHeardAt = obsRows[0].HeardAt.Time + } if len(obsRows) > 1 { maxHeardAt := obsRows[0].HeardAt.Time for _, v := range obsRows[1:] { @@ -420,16 +440,9 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet, destHashByte = []byte{path.Destination} } } - // ADVERT's source is an exact pubkey match, not ambiguous like the above -- and unlike - // them it doesn't depend on IATA, so resolve it once here rather than per observation. + // Legacy ADVERT sources need an exact lookup, at most once for this packet. var resolvedAdvertSource *api.ResolvedNode - if row.PayloadType == int16(meshcore.PayloadTypeAdvert) && row.OriginPubkey != nil { - if nodeID, err := s.GetNodeByPubkey(ctx, row.OriginPubkey); err == nil { - if nodes, err := s.GetNodesByIDs(ctx, []uuid.UUID{nodeID}); err == nil { - resolvedAdvertSource = nodes[nodeID] - } - } - } + advertSourceLookedUp := false for _, v := range obsRows { obs := api.PacketObservationDetail{ ID: v.ID, @@ -442,9 +455,11 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet, HashSize: v.HashSize, HopCount: v.HopCount, }, - RSSI: v.Rssi, - SNR: v.Snr, - SourceBroker: *v.SourceBroker, + RSSI: v.Rssi, + SNR: v.Snr, + } + if v.SourceBroker != nil { + obs.SourceBroker = *v.SourceBroker } prop := int32(v.HeardAt.Time.Sub(minHeardAt).Milliseconds()) obs.PropagationTimeMs = &prop @@ -472,23 +487,36 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet, } } obs.ResolvedPath = resolvedPath - if row.PayloadType == int16(meshcore.PayloadTypeAdvert) { - hop := api.ResolveExactNode(resolvedAdvertSource) - obs.ResolvedSource = &hop - } else if len(sourceHashByte) == 1 { - if r, err := s.ResolvePathHashes(ctx, v.Iata, [][]byte{sourceHashByte}); err != nil { - log.Printf("store: source resolution failed for observation %d: %v", v.ID, err) - } else { - hop := api.BuildResolvedPath([][]byte{sourceHashByte}, r)[0] + if endpoints, captured := decodePacketEndpointSnapshot(v.ResolvedEndpoints); captured { + obs.ResolvedSource = endpoints.Source + obs.ResolvedDestination = endpoints.Destination + } else { + if row.PayloadType == int16(meshcore.PayloadTypeAdvert) { + if !advertSourceLookedUp && row.OriginPubkey != nil { + if nodeID, err := s.GetNodeByPubkey(ctx, row.OriginPubkey); err == nil { + if nodes, err := s.GetNodesByIDs(ctx, []uuid.UUID{nodeID}); err == nil { + resolvedAdvertSource = nodes[nodeID] + } + } + advertSourceLookedUp = true + } + hop := api.ResolveExactNode(resolvedAdvertSource) obs.ResolvedSource = &hop + } else if len(sourceHashByte) == 1 { + if r, err := s.ResolveEndpointHashes(ctx, v.Iata, [][]byte{sourceHashByte}); err != nil { + log.Printf("store: source resolution failed for observation %d: %v", v.ID, err) + } else { + hop := api.BuildResolvedPath([][]byte{sourceHashByte}, r)[0] + obs.ResolvedSource = &hop + } } - } - if len(destHashByte) == 1 { - if r, err := s.ResolvePathHashes(ctx, v.Iata, [][]byte{destHashByte}); err != nil { - log.Printf("store: destination resolution failed for observation %d: %v", v.ID, err) - } else { - hop := api.BuildResolvedPath([][]byte{destHashByte}, r)[0] - obs.ResolvedDestination = &hop + if len(destHashByte) == 1 { + if r, err := s.ResolveEndpointHashes(ctx, v.Iata, [][]byte{destHashByte}); err != nil { + log.Printf("store: destination resolution failed for observation %d: %v", v.ID, err) + } else { + hop := api.BuildResolvedPath([][]byte{destHashByte}, r)[0] + obs.ResolvedDestination = &hop + } } } if row.PayloadType == int16(meshcore.PayloadTypeTrace) && len(traceRawHashes) > 0 { @@ -557,6 +585,7 @@ func (s *Store) InsertObservation(ctx context.Context, o ingest.InsertObservatio CodingRate: &o.CodingRate, SourceBroker: &o.SourceBroker, PayloadType: &o.PayloadType, + ResolvedEndpoints: o.ResolvedEndpoints, } row, err := s.q.InsertObservation(ctx, params) if errors.Is(err, pgx.ErrNoRows) { diff --git a/db/packets_test.go b/db/packets_test.go index f2c8248..6541ba3 100644 --- a/db/packets_test.go +++ b/db/packets_test.go @@ -204,7 +204,7 @@ func TestListPackets_LatestObserverPathFields(t *testing.T) { if obs.PathBytes == nil || *obs.PathBytes != "a1b2" { t.Errorf("expected pathBytes a1b2, got %v", obs.PathBytes) } - // Resolution stays a detail-view-only feature on this list endpoint -- deliberately unset. + // Legacy rows without a captured snapshot still omit resolved endpoints on lists. if obs.ResolvedPath != nil || obs.ResolvedSource != nil || obs.ResolvedDestination != nil { t.Error("expected no resolved path/source/destination on the list endpoint") } diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 9771a26..c2d7305 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -425,10 +425,11 @@ SELECT COALESCE(po.path_length_byte, 0::smallint) AS latest_observer_path_length_byte, COALESCE(po.hash_size, 0::smallint) AS latest_observer_hash_size, COALESCE(po.hop_count, 0::smallint) AS latest_observer_hop_count, - po.path_bytes AS latest_observer_path_bytes + po.path_bytes AS latest_observer_path_bytes, + po.resolved_endpoints AS latest_observer_resolved_endpoints FROM packets p LEFT JOIN LATERAL ( - SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes + SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes, resolved_endpoints FROM packet_observations WHERE packet_hash = p.packet_hash ORDER BY heard_at DESC @@ -522,12 +523,13 @@ SELECT COALESCE(po.path_length_byte, 0::smallint) AS latest_observer_path_length_byte, COALESCE(po.hash_size, 0::smallint) AS latest_observer_hash_size, COALESCE(po.hop_count, 0::smallint) AS latest_observer_hop_count, - po.path_bytes AS latest_observer_path_bytes + po.path_bytes AS latest_observer_path_bytes, + po.resolved_endpoints AS latest_observer_resolved_endpoints FROM page sh CROSS JOIN saturation sat JOIN packets p ON p.packet_hash = sh.packet_hash LEFT JOIN LATERAL ( - SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes + SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes, resolved_endpoints FROM packet_observations WHERE packet_hash = p.packet_hash ORDER BY heard_at DESC @@ -554,6 +556,7 @@ SELECT po.hash_size AS latest_observer_hash_size, po.hop_count AS latest_observer_hop_count, po.path_bytes AS latest_observer_path_bytes, + po.resolved_endpoints AS latest_observer_resolved_endpoints, ts.name AS scope_name FROM packets p JOIN packet_observations po ON po.packet_hash = p.packet_hash @@ -622,9 +625,10 @@ INSERT INTO packet_observations ( bandwidth_khz, coding_rate, source_broker, - payload_type + payload_type, + resolved_endpoints ) VALUES ( - $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17 + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18 ) ON CONFLICT (packet_hash, observer_id) DO NOTHING RETURNING *; @@ -1236,6 +1240,21 @@ WHERE ns.iata = $1 AND n.node_type IN (2, 3) AND ns.prefix_1 = ANY($2::bytea[]); +-- name: ResolveEndpointHashes :many +-- Logical endpoints can be any advertised role, unlike intermediate relay hops. +-- Endpoint hashes are always one byte; use the existing (iata, prefix_1) index. +-- LIMIT 1 keeps generic plans on a node PK lookup per candidate instead of +-- flattening the join into a scan of all nodes. The PK already guarantees one row. +SELECT ns.prefix_1 AS hash, n.id AS node_id, n.name, n.latitude, n.longitude, n.public_key +FROM node_short_ids ns +CROSS JOIN LATERAL ( + SELECT id, name, latitude, longitude, public_key + FROM nodes WHERE id = ns.node_id + LIMIT 1 +) n +WHERE ns.iata = $1 + AND ns.prefix_1 = ANY($2::bytea[]); + -- name: ResolvePathHashesP2 :many SELECT ns.prefix_4 AS hash, n.id AS node_id, n.name, n.latitude, n.longitude, n.public_key FROM node_short_ids ns diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index c34fe77..5742d19 100644 --- a/db/sqlc/mock/querier.go +++ b/db/sqlc/mock/querier.go @@ -1167,6 +1167,21 @@ func (mr *MockQuerierMockRecorder) RefreshTopTalkers(ctx any) *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RefreshTopTalkers", reflect.TypeOf((*MockQuerier)(nil).RefreshTopTalkers), ctx) } +// ResolveEndpointHashes mocks base method. +func (m *MockQuerier) ResolveEndpointHashes(ctx context.Context, arg db.ResolveEndpointHashesParams) ([]db.ResolveEndpointHashesRow, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ResolveEndpointHashes", ctx, arg) + ret0, _ := ret[0].([]db.ResolveEndpointHashesRow) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// ResolveEndpointHashes indicates an expected call of ResolveEndpointHashes. +func (mr *MockQuerierMockRecorder) ResolveEndpointHashes(ctx, arg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ResolveEndpointHashes", reflect.TypeOf((*MockQuerier)(nil).ResolveEndpointHashes), ctx, arg) +} + // ResolvePathHashesP1 mocks base method. func (m *MockQuerier) ResolvePathHashesP1(ctx context.Context, arg db.ResolvePathHashesP1Params) ([]db.ResolvePathHashesP1Row, error) { m.ctrl.T.Helper() diff --git a/db/sqlc/models.go b/db/sqlc/models.go index 9a40355..5bed269 100644 --- a/db/sqlc/models.go +++ b/db/sqlc/models.go @@ -292,6 +292,7 @@ type PacketObservation struct { CodingRate *int16 `json:"coding_rate"` SourceBroker *string `json:"source_broker"` PayloadType *int16 `json:"payload_type"` + ResolvedEndpoints []byte `json:"resolved_endpoints"` } type Region struct { diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index 1c5f8c9..7a2b051 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -192,6 +192,11 @@ type Querier interface { RefreshTopNodes(ctx context.Context) error RefreshTopObservers(ctx context.Context) error RefreshTopTalkers(ctx context.Context) error + // Logical endpoints can be any advertised role, unlike intermediate relay hops. + // Endpoint hashes are always one byte; use the existing (iata, prefix_1) index. + // LIMIT 1 keeps generic plans on a node PK lookup per candidate instead of + // flattening the join into a scan of all nodes. The PK already guarantees one row. + ResolveEndpointHashes(ctx context.Context, arg ResolveEndpointHashesParams) ([]ResolveEndpointHashesRow, error) // ============================================================ // HELPERS // ============================================================ diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 8581c4c..2d2a05d 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -1721,12 +1721,13 @@ INSERT INTO packet_observations ( bandwidth_khz, coding_rate, source_broker, - payload_type + payload_type, + resolved_endpoints ) VALUES ( - $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17 + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18 ) ON CONFLICT (packet_hash, observer_id) DO NOTHING -RETURNING id, packet_hash, observer_id, iata, heard_at, path_length_byte, hash_size, hop_count, path_bytes, rssi, snr, propagation_time_ms, radio_freq_mhz, spread_factor, bandwidth_khz, coding_rate, source_broker, payload_type +RETURNING id, packet_hash, observer_id, iata, heard_at, path_length_byte, hash_size, hop_count, path_bytes, rssi, snr, propagation_time_ms, radio_freq_mhz, spread_factor, bandwidth_khz, coding_rate, source_broker, payload_type, resolved_endpoints ` type InsertObservationParams struct { @@ -1747,6 +1748,7 @@ type InsertObservationParams struct { CodingRate *int16 `json:"coding_rate"` SourceBroker *string `json:"source_broker"` PayloadType *int16 `json:"payload_type"` + ResolvedEndpoints []byte `json:"resolved_endpoints"` } // ============================================================ @@ -1771,6 +1773,7 @@ func (q *Queries) InsertObservation(ctx context.Context, arg InsertObservationPa arg.CodingRate, arg.SourceBroker, arg.PayloadType, + arg.ResolvedEndpoints, ) var i PacketObservation err := row.Scan( @@ -1792,6 +1795,7 @@ func (q *Queries) InsertObservation(ctx context.Context, arg InsertObservationPa &i.CodingRate, &i.SourceBroker, &i.PayloadType, + &i.ResolvedEndpoints, ) return i, err } @@ -2481,7 +2485,7 @@ func (q *Queries) ListNodes(ctx context.Context, arg ListNodesParams) ([]ListNod } const listObservationsForPacket = `-- name: ListObservationsForPacket :many -SELECT po.id, po.packet_hash, po.observer_id, po.iata, po.heard_at, po.path_length_byte, po.hash_size, po.hop_count, po.path_bytes, po.rssi, po.snr, po.propagation_time_ms, po.radio_freq_mhz, po.spread_factor, po.bandwidth_khz, po.coding_rate, po.source_broker, po.payload_type, o.display_name AS observer_name +SELECT po.id, po.packet_hash, po.observer_id, po.iata, po.heard_at, po.path_length_byte, po.hash_size, po.hop_count, po.path_bytes, po.rssi, po.snr, po.propagation_time_ms, po.radio_freq_mhz, po.spread_factor, po.bandwidth_khz, po.coding_rate, po.source_broker, po.payload_type, po.resolved_endpoints, o.display_name AS observer_name FROM packet_observations po LEFT JOIN observers o ON o.id = po.observer_id WHERE po.packet_hash = $1 @@ -2507,6 +2511,7 @@ type ListObservationsForPacketRow struct { CodingRate *int16 `json:"coding_rate"` SourceBroker *string `json:"source_broker"` PayloadType *int16 `json:"payload_type"` + ResolvedEndpoints []byte `json:"resolved_endpoints"` ObserverName *string `json:"observer_name"` } @@ -2538,6 +2543,7 @@ func (q *Queries) ListObservationsForPacket(ctx context.Context, packetHash []by &i.CodingRate, &i.SourceBroker, &i.PayloadType, + &i.ResolvedEndpoints, &i.ObserverName, ); err != nil { return nil, err @@ -2756,10 +2762,11 @@ SELECT COALESCE(po.path_length_byte, 0::smallint) AS latest_observer_path_length_byte, COALESCE(po.hash_size, 0::smallint) AS latest_observer_hash_size, COALESCE(po.hop_count, 0::smallint) AS latest_observer_hop_count, - po.path_bytes AS latest_observer_path_bytes + po.path_bytes AS latest_observer_path_bytes, + po.resolved_endpoints AS latest_observer_resolved_endpoints FROM packets p LEFT JOIN LATERAL ( - SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes + SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes, resolved_endpoints FROM packet_observations WHERE packet_hash = p.packet_hash ORDER BY heard_at DESC @@ -2789,21 +2796,22 @@ type ListPacketsParams struct { } type ListPacketsRow struct { - PacketHash []byte `json:"packet_hash"` - PayloadType int16 `json:"payload_type"` - RouteType int16 `json:"route_type"` - FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` - LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` - ScopeID *int32 `json:"scope_id"` - ScopeName *string `json:"scope_name"` - ObservationCount int64 `json:"observation_count"` - LatestObserverID uuid.UUID `json:"latest_observer_id"` - LatestObserverName *string `json:"latest_observer_name"` - LatestObserverIata string `json:"latest_observer_iata"` - LatestObserverPathLengthByte int16 `json:"latest_observer_path_length_byte"` - LatestObserverHashSize int16 `json:"latest_observer_hash_size"` - LatestObserverHopCount int16 `json:"latest_observer_hop_count"` - LatestObserverPathBytes []byte `json:"latest_observer_path_bytes"` + PacketHash []byte `json:"packet_hash"` + PayloadType int16 `json:"payload_type"` + RouteType int16 `json:"route_type"` + FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` + LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` + ScopeID *int32 `json:"scope_id"` + ScopeName *string `json:"scope_name"` + ObservationCount int64 `json:"observation_count"` + LatestObserverID uuid.UUID `json:"latest_observer_id"` + LatestObserverName *string `json:"latest_observer_name"` + LatestObserverIata string `json:"latest_observer_iata"` + LatestObserverPathLengthByte int16 `json:"latest_observer_path_length_byte"` + LatestObserverHashSize int16 `json:"latest_observer_hash_size"` + LatestObserverHopCount int16 `json:"latest_observer_hop_count"` + LatestObserverPathBytes []byte `json:"latest_observer_path_bytes"` + LatestObserverResolvedEndpoints []byte `json:"latest_observer_resolved_endpoints"` } // Returns packets with the latest observation rolled in for display. @@ -2842,6 +2850,7 @@ func (q *Queries) ListPackets(ctx context.Context, arg ListPacketsParams) ([]Lis &i.LatestObserverHashSize, &i.LatestObserverHopCount, &i.LatestObserverPathBytes, + &i.LatestObserverResolvedEndpoints, ); err != nil { return nil, err } @@ -2868,6 +2877,7 @@ SELECT po.hash_size AS latest_observer_hash_size, po.hop_count AS latest_observer_hop_count, po.path_bytes AS latest_observer_path_bytes, + po.resolved_endpoints AS latest_observer_resolved_endpoints, ts.name AS scope_name FROM packets p JOIN packet_observations po ON po.packet_hash = p.packet_hash @@ -2892,20 +2902,21 @@ type ListPacketsAfterIDParams struct { } type ListPacketsAfterIDRow struct { - PacketHash []byte `json:"packet_hash"` - PayloadType int16 `json:"payload_type"` - RouteType int16 `json:"route_type"` - FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` - LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` - ObservationCount int64 `json:"observation_count"` - LatestObserverID uuid.UUID `json:"latest_observer_id"` - LatestObserverName *string `json:"latest_observer_name"` - LatestObserverIata string `json:"latest_observer_iata"` - LatestObserverPathLengthByte int16 `json:"latest_observer_path_length_byte"` - LatestObserverHashSize int16 `json:"latest_observer_hash_size"` - LatestObserverHopCount int16 `json:"latest_observer_hop_count"` - LatestObserverPathBytes []byte `json:"latest_observer_path_bytes"` - ScopeName *string `json:"scope_name"` + PacketHash []byte `json:"packet_hash"` + PayloadType int16 `json:"payload_type"` + RouteType int16 `json:"route_type"` + FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` + LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` + ObservationCount int64 `json:"observation_count"` + LatestObserverID uuid.UUID `json:"latest_observer_id"` + LatestObserverName *string `json:"latest_observer_name"` + LatestObserverIata string `json:"latest_observer_iata"` + LatestObserverPathLengthByte int16 `json:"latest_observer_path_length_byte"` + LatestObserverHashSize int16 `json:"latest_observer_hash_size"` + LatestObserverHopCount int16 `json:"latest_observer_hop_count"` + LatestObserverPathBytes []byte `json:"latest_observer_path_bytes"` + LatestObserverResolvedEndpoints []byte `json:"latest_observer_resolved_endpoints"` + ScopeName *string `json:"scope_name"` } // Returns packets with observations after the given observation ID, ordered oldest first. @@ -2940,6 +2951,7 @@ func (q *Queries) ListPacketsAfterID(ctx context.Context, arg ListPacketsAfterID &i.LatestObserverHashSize, &i.LatestObserverHopCount, &i.LatestObserverPathBytes, + &i.LatestObserverResolvedEndpoints, &i.ScopeName, ); err != nil { return nil, err @@ -3015,12 +3027,13 @@ SELECT COALESCE(po.path_length_byte, 0::smallint) AS latest_observer_path_length_byte, COALESCE(po.hash_size, 0::smallint) AS latest_observer_hash_size, COALESCE(po.hop_count, 0::smallint) AS latest_observer_hop_count, - po.path_bytes AS latest_observer_path_bytes + po.path_bytes AS latest_observer_path_bytes, + po.resolved_endpoints AS latest_observer_resolved_endpoints FROM page sh CROSS JOIN saturation sat JOIN packets p ON p.packet_hash = sh.packet_hash LEFT JOIN LATERAL ( - SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes + SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes, resolved_endpoints FROM packet_observations WHERE packet_hash = p.packet_hash ORDER BY heard_at DESC @@ -3044,24 +3057,25 @@ type ListPacketsByIATAsParams struct { } type ListPacketsByIATAsRow struct { - PacketHash []byte `json:"packet_hash"` - PayloadType int16 `json:"payload_type"` - RouteType int16 `json:"route_type"` - FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` - LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` - ScopeID *int32 `json:"scope_id"` - ScopeName *string `json:"scope_name"` - SiteHeardAt pgtype.Timestamptz `json:"site_heard_at"` - ScanSaturated bool `json:"scan_saturated"` - ScanFloor pgtype.Timestamptz `json:"scan_floor"` - ObservationCount int64 `json:"observation_count"` - LatestObserverID uuid.UUID `json:"latest_observer_id"` - LatestObserverName *string `json:"latest_observer_name"` - LatestObserverIata string `json:"latest_observer_iata"` - LatestObserverPathLengthByte int16 `json:"latest_observer_path_length_byte"` - LatestObserverHashSize int16 `json:"latest_observer_hash_size"` - LatestObserverHopCount int16 `json:"latest_observer_hop_count"` - LatestObserverPathBytes []byte `json:"latest_observer_path_bytes"` + PacketHash []byte `json:"packet_hash"` + PayloadType int16 `json:"payload_type"` + RouteType int16 `json:"route_type"` + FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` + LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` + ScopeID *int32 `json:"scope_id"` + ScopeName *string `json:"scope_name"` + SiteHeardAt pgtype.Timestamptz `json:"site_heard_at"` + ScanSaturated bool `json:"scan_saturated"` + ScanFloor pgtype.Timestamptz `json:"scan_floor"` + ObservationCount int64 `json:"observation_count"` + LatestObserverID uuid.UUID `json:"latest_observer_id"` + LatestObserverName *string `json:"latest_observer_name"` + LatestObserverIata string `json:"latest_observer_iata"` + LatestObserverPathLengthByte int16 `json:"latest_observer_path_length_byte"` + LatestObserverHashSize int16 `json:"latest_observer_hash_size"` + LatestObserverHopCount int16 `json:"latest_observer_hop_count"` + LatestObserverPathBytes []byte `json:"latest_observer_path_bytes"` + LatestObserverResolvedEndpoints []byte `json:"latest_observer_resolved_endpoints"` } // IATA-filtered packet list, driven from idx_observations_iata_heard. @@ -3115,6 +3129,7 @@ func (q *Queries) ListPacketsByIATAs(ctx context.Context, arg ListPacketsByIATAs &i.LatestObserverHashSize, &i.LatestObserverHopCount, &i.LatestObserverPathBytes, + &i.LatestObserverResolvedEndpoints, ); err != nil { return nil, err } @@ -3444,6 +3459,63 @@ func (q *Queries) RefreshTopTalkers(ctx context.Context) error { return err } +const resolveEndpointHashes = `-- name: ResolveEndpointHashes :many +SELECT ns.prefix_1 AS hash, n.id AS node_id, n.name, n.latitude, n.longitude, n.public_key +FROM node_short_ids ns +CROSS JOIN LATERAL ( + SELECT id, name, latitude, longitude, public_key + FROM nodes WHERE id = ns.node_id + LIMIT 1 +) n +WHERE ns.iata = $1 + AND ns.prefix_1 = ANY($2::bytea[]) +` + +type ResolveEndpointHashesParams struct { + Iata string `json:"iata"` + Column2 [][]byte `json:"column_2"` +} + +type ResolveEndpointHashesRow struct { + Hash []byte `json:"hash"` + NodeID uuid.UUID `json:"node_id"` + Name *string `json:"name"` + Latitude *float64 `json:"latitude"` + Longitude *float64 `json:"longitude"` + PublicKey []byte `json:"public_key"` +} + +// Logical endpoints can be any advertised role, unlike intermediate relay hops. +// Endpoint hashes are always one byte; use the existing (iata, prefix_1) index. +// LIMIT 1 keeps generic plans on a node PK lookup per candidate instead of +// flattening the join into a scan of all nodes. The PK already guarantees one row. +func (q *Queries) ResolveEndpointHashes(ctx context.Context, arg ResolveEndpointHashesParams) ([]ResolveEndpointHashesRow, error) { + rows, err := q.db.Query(ctx, resolveEndpointHashes, arg.Iata, arg.Column2) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ResolveEndpointHashesRow{} + for rows.Next() { + var i ResolveEndpointHashesRow + if err := rows.Scan( + &i.Hash, + &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 resolvePathHashesP1 = `-- name: ResolvePathHashesP1 :many diff --git a/db/store.go b/db/store.go index d2699a7..a22b960 100644 --- a/db/store.go +++ b/db/store.go @@ -83,6 +83,32 @@ func (s *Store) ResolvePathHashes(ctx context.Context, iata string, hashes [][]b return result, nil } +// ResolveEndpointHashes matches one-byte packet endpoints across advertised roles. +// Keep this separate from the infrastructure-only intermediate path resolver. +func (s *Store) ResolveEndpointHashes(ctx context.Context, iata string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error) { + if len(hashes) == 0 { + return nil, nil + } + for _, hash := range hashes { + if len(hash) != 1 { + return nil, nil + } + } + rows, err := s.q.ResolveEndpointHashes(ctx, sqlc.ResolveEndpointHashesParams{Iata: iata, Column2: hashes}) + if err != nil { + return nil, err + } + result := make(map[string][]api.ResolvedPathEntry) + for _, row := range rows { + key := hex.EncodeToString(row.Hash) + result[key] = append(result[key], api.ResolvedPathEntry{ + NodeID: row.NodeID, Name: row.Name, Latitude: row.Latitude, + Longitude: row.Longitude, PublicKey: row.PublicKey, + }) + } + return result, nil +} + // nullableUUID returns nil for a zero UUID, or a pointer to the UUID otherwise. func nullableUUID(id uuid.UUID) *uuid.UUID { if id == (uuid.UUID{}) { diff --git a/docs/docs.go b/docs/docs.go index 94cc919..b9f685a 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -3191,7 +3191,7 @@ const docTemplate = `{ "type": "string" }, "pathLength": { - "description": "PathLength/PathBytes are cheap -- already-stored columns on packet_observations -- and\npopulated everywhere PacketLatestObserver appears: the REST list/backfill endpoints and\nthe WS feed alike. ResolvedPath/ResolvedSource/ResolvedDestination require a per-hash DB\nresolution lookup; they're populated on the WS feed (already computed once at ingest, so\neffectively free there) but deliberately left nil on the REST endpoints, which are\npaginated/high-volume and used only for scrollback and reconnect-gap backfill -- full\nresolution stays a GET /packets/{packetHash}-only feature.", + "description": "PathLength/PathBytes and captured endpoint resolutions are stored on the observation,\nso list/backfill reads need no per-hash resolution queries. Legacy observations have\nno endpoint snapshot. ResolvedPath remains a detail/opted-in WS feature.", "allOf": [ { "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.PacketPathLength" @@ -3256,7 +3256,7 @@ const docTemplate = `{ } }, "resolvedSource": { - "description": "ResolvedSource/ResolvedDestination are the packet's endpoints, when the payload type\ncarries a resolvable one: an exact match for ADVERT's full pubkey, an ambiguous\nhash-prefix match (like intermediate hops) for TEXT_MESSAGE/PATH/ANON_REQ's 1-byte\nsource/destination hashes. Nil when the payload type doesn't carry one at all (e.g.\nGRP_TXT/GRP_DATA/TRACE aren't node-to-node addressed) -- see BuildResolvedPath and\nResolveExactNode for how each is built.", + "description": "ResolvedSource/ResolvedDestination are the packet's endpoints, when the payload type\ncarries one. Prefer the snapshot captured at ingest; legacy observations without a\nsnapshot use the current node registry. Endpoint matching itself is unchanged:\nan exact match for ADVERT's full pubkey, an ambiguous\nhash-prefix match (like intermediate hops) for TEXT_MESSAGE/PATH/ANON_REQ's 1-byte\nsource/destination hashes. Nil when the payload type doesn't carry one at all (e.g.\nGRP_TXT/GRP_DATA/TRACE aren't node-to-node addressed) -- see BuildResolvedPath and\nResolveExactNode for how each is built.", "allOf": [ { "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ResolvedHop" diff --git a/docs/swagger.json b/docs/swagger.json index c943d3b..7db537c 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -3189,7 +3189,7 @@ "type": "string" }, "pathLength": { - "description": "PathLength/PathBytes are cheap -- already-stored columns on packet_observations -- and\npopulated everywhere PacketLatestObserver appears: the REST list/backfill endpoints and\nthe WS feed alike. ResolvedPath/ResolvedSource/ResolvedDestination require a per-hash DB\nresolution lookup; they're populated on the WS feed (already computed once at ingest, so\neffectively free there) but deliberately left nil on the REST endpoints, which are\npaginated/high-volume and used only for scrollback and reconnect-gap backfill -- full\nresolution stays a GET /packets/{packetHash}-only feature.", + "description": "PathLength/PathBytes and captured endpoint resolutions are stored on the observation,\nso list/backfill reads need no per-hash resolution queries. Legacy observations have\nno endpoint snapshot. ResolvedPath remains a detail/opted-in WS feature.", "allOf": [ { "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.PacketPathLength" @@ -3254,7 +3254,7 @@ } }, "resolvedSource": { - "description": "ResolvedSource/ResolvedDestination are the packet's endpoints, when the payload type\ncarries a resolvable one: an exact match for ADVERT's full pubkey, an ambiguous\nhash-prefix match (like intermediate hops) for TEXT_MESSAGE/PATH/ANON_REQ's 1-byte\nsource/destination hashes. Nil when the payload type doesn't carry one at all (e.g.\nGRP_TXT/GRP_DATA/TRACE aren't node-to-node addressed) -- see BuildResolvedPath and\nResolveExactNode for how each is built.", + "description": "ResolvedSource/ResolvedDestination are the packet's endpoints, when the payload type\ncarries one. Prefer the snapshot captured at ingest; legacy observations without a\nsnapshot use the current node registry. Endpoint matching itself is unchanged:\nan exact match for ADVERT's full pubkey, an ambiguous\nhash-prefix match (like intermediate hops) for TEXT_MESSAGE/PATH/ANON_REQ's 1-byte\nsource/destination hashes. Nil when the payload type doesn't carry one at all (e.g.\nGRP_TXT/GRP_DATA/TRACE aren't node-to-node addressed) -- see BuildResolvedPath and\nResolveExactNode for how each is built.", "allOf": [ { "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ResolvedHop" diff --git a/docs/swagger.yaml b/docs/swagger.yaml index c6667f3..5bb289b 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -619,13 +619,9 @@ definitions: allOf: - $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.PacketPathLength' description: |- - PathLength/PathBytes are cheap -- already-stored columns on packet_observations -- and - populated everywhere PacketLatestObserver appears: the REST list/backfill endpoints and - the WS feed alike. ResolvedPath/ResolvedSource/ResolvedDestination require a per-hash DB - resolution lookup; they're populated on the WS feed (already computed once at ingest, so - effectively free there) but deliberately left nil on the REST endpoints, which are - paginated/high-volume and used only for scrollback and reconnect-gap backfill -- full - resolution stays a GET /packets/{packetHash}-only feature. + PathLength/PathBytes and captured endpoint resolutions are stored on the observation, + so list/backfill reads need no per-hash resolution queries. Legacy observations have + no endpoint snapshot. ResolvedPath remains a detail/opted-in WS feature. resolvedDestination: $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ResolvedHop' resolvedPath: @@ -670,7 +666,9 @@ definitions: - $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ResolvedHop' description: |- ResolvedSource/ResolvedDestination are the packet's endpoints, when the payload type - carries a resolvable one: an exact match for ADVERT's full pubkey, an ambiguous + carries one. Prefer the snapshot captured at ingest; legacy observations without a + snapshot use the current node registry. Endpoint matching itself is unchanged: + an exact match for ADVERT's full pubkey, an ambiguous hash-prefix match (like intermediate hops) for TEXT_MESSAGE/PATH/ANON_REQ's 1-byte source/destination hashes. Nil when the payload type doesn't carry one at all (e.g. GRP_TXT/GRP_DATA/TRACE aren't node-to-node addressed) -- see BuildResolvedPath and diff --git a/internal/api/packets.go b/internal/api/packets.go index 305fcfb..c75aa6f 100644 --- a/internal/api/packets.go +++ b/internal/api/packets.go @@ -17,13 +17,9 @@ type PacketLatestObserver struct { ID uuid.UUID `json:"id"` DisplayName *string `json:"displayName,omitempty"` IATA string `json:"iata"` - // PathLength/PathBytes are cheap -- already-stored columns on packet_observations -- and - // populated everywhere PacketLatestObserver appears: the REST list/backfill endpoints and - // the WS feed alike. ResolvedPath/ResolvedSource/ResolvedDestination require a per-hash DB - // resolution lookup; they're populated on the WS feed (already computed once at ingest, so - // effectively free there) but deliberately left nil on the REST endpoints, which are - // paginated/high-volume and used only for scrollback and reconnect-gap backfill -- full - // resolution stays a GET /packets/{packetHash}-only feature. + // PathLength/PathBytes and captured endpoint resolutions are stored on the observation, + // so list/backfill reads need no per-hash resolution queries. Legacy observations have + // no endpoint snapshot. ResolvedPath remains a detail/opted-in WS feature. PathLength *PacketPathLength `json:"pathLength,omitempty"` PathBytes *string `json:"pathBytes,omitempty"` // hex-encoded accumulated path hashes ResolvedPath []ResolvedHop `json:"resolvedPath,omitempty"` @@ -71,7 +67,9 @@ type PacketObservationDetail struct { 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 a resolvable one: an exact match for ADVERT's full pubkey, an ambiguous + // carries one. Prefer the snapshot captured at ingest; legacy observations without a + // snapshot use the current node registry. Endpoint matching itself is unchanged: + // an exact match for ADVERT's full pubkey, an ambiguous // hash-prefix match (like intermediate hops) for TEXT_MESSAGE/PATH/ANON_REQ's 1-byte // source/destination hashes. Nil when the payload type doesn't carry one at all (e.g. // GRP_TXT/GRP_DATA/TRACE aren't node-to-node addressed) -- see BuildResolvedPath and @@ -96,6 +94,21 @@ type ResolvedHop struct { Nodes []ResolvedNode `json:"nodes"` // empty for "none", one for "high", multiple for "ambiguous" } +// PacketEndpointSnapshot is the internal storage shape for an observation's +// endpoint resolution at ingest. Names and confidence are preserved even when +// the current node registry changes. REST/WS expose the existing per-endpoint fields. +type PacketEndpointSnapshot struct { + Source *ResolvedHop `json:"source,omitempty"` + Destination *ResolvedHop `json:"destination,omitempty"` +} + +// HasResolvedNodes distinguishes a historical capture from an unresolved lookup +// that should be retried when a node becomes known (including a first advert). +func (s PacketEndpointSnapshot) HasResolvedNodes() bool { + return (s.Source != nil && len(s.Source.Nodes) > 0) || + (s.Destination != nil && len(s.Destination.Nodes) > 0) +} + // ResolvedNode is a node reference within a resolved path hop. type ResolvedNode struct { ID uuid.UUID `json:"id"` diff --git a/internal/api/packets_test.go b/internal/api/packets_test.go index f430f60..2befa56 100644 --- a/internal/api/packets_test.go +++ b/internal/api/packets_test.go @@ -10,6 +10,28 @@ import ( "github.com/meshcore-go/meshcore-go" ) +func TestPacketEndpointSnapshotHasResolvedNodes(t *testing.T) { + none := &api.ResolvedHop{Confidence: "none", Nodes: []api.ResolvedNode{}} + known := &api.ResolvedHop{Confidence: "high", Nodes: []api.ResolvedNode{{PublicKey: "aa"}}} + for _, tc := range []struct { + name string + snapshot api.PacketEndpointSnapshot + want bool + }{ + {"absent", api.PacketEndpointSnapshot{}, false}, + {"unresolved source", api.PacketEndpointSnapshot{Source: none}, false}, + {"both unresolved", api.PacketEndpointSnapshot{Source: none, Destination: none}, false}, + {"source only", api.PacketEndpointSnapshot{Source: known, Destination: none}, true}, + {"destination only", api.PacketEndpointSnapshot{Destination: known}, true}, + } { + t.Run(tc.name, func(t *testing.T) { + if got := tc.snapshot.HasResolvedNodes(); got != tc.want { + t.Fatalf("resolved nodes = %v, want %v", got, tc.want) + } + }) + } +} + func TestPayloadTypeName(t *testing.T) { tests := []struct { input int16 diff --git a/internal/ingest/endpoint_matching_test.go b/internal/ingest/endpoint_matching_test.go new file mode 100644 index 0000000..1b60062 --- /dev/null +++ b/internal/ingest/endpoint_matching_test.go @@ -0,0 +1,50 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package ingest + +import ( + "context" + "encoding/hex" + "reflect" + "testing" + + "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/meshcore-go/meshcore-go" +) + +type endpointRoutingDB struct { + *stubDB + endpoints []string + paths []string +} + +func (s *endpointRoutingDB) ResolveEndpointHashes(_ context.Context, iata string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error) { + for _, hash := range hashes { + s.endpoints = append(s.endpoints, iata+":"+hex.EncodeToString(hash)) + } + return nil, nil +} +func (s *endpointRoutingDB) ResolvePathHashes(_ context.Context, iata string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error) { + for _, hash := range hashes { + s.paths = append(s.paths, iata+":"+hex.EncodeToString(hash)) + } + return nil, nil +} + +func TestHandlePacketSeparatesEndpointAndRelayMatching(t *testing.T) { + for _, kind := range []uint8{meshcore.PayloadTypeReq, meshcore.PayloadTypeResponse, meshcore.PayloadTypeTxtMsg, meshcore.PayloadTypePath} { + w, base := newTestWorker() + db := &endpointRoutingDB{stubDB: base} + w.db = db + packet := &meshcore.Packet{Header: meshcore.MakeHeader(meshcore.RouteTypeFlood, kind, 0), + PathLength: 1, Path: []byte{0xcc}, Payload: append([]byte{0xbb, 0xaa, 0, 0}, make([]byte, 16)...)} + w.handlePacket(context.Background(), "YVR", "0102", packetEnvelope(t, packet)) + if !reflect.DeepEqual(db.endpoints, []string{"YVR:aa", "YVR:bb"}) { + t.Errorf("payload %d endpoint dispatch: %v", kind, db.endpoints) + } + if !reflect.DeepEqual(db.paths, []string{"YVR:cc"}) { + t.Errorf("payload %d relay dispatch: %v", kind, db.paths) + } + } +} diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index 16743f6..c2e485f 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -153,6 +153,9 @@ 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) + // ResolveEndpointHashes matches one-byte logical endpoints, including companions. + ResolveEndpointHashes(ctx context.Context, iata string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error) + // UpsertChannel upserts a channel row by (hash, keyFingerprint) and returns its integer ID. // Pass nil keyFingerprint to record a hash-only row when the key is unknown. UpsertChannel(ctx context.Context, channelHash []byte, keyFingerprint []byte, name string, hashtag string) (int, error) diff --git a/internal/ingest/ingest_test.go b/internal/ingest/ingest_test.go index 6e11dfe..94bd6fd 100644 --- a/internal/ingest/ingest_test.go +++ b/internal/ingest/ingest_test.go @@ -252,6 +252,10 @@ func (s *stubDB) ResolvePathHashes(_ context.Context, _ string, _ [][]byte) (map return nil, nil } +func (s *stubDB) ResolveEndpointHashes(_ context.Context, _ string, _ [][]byte) (map[string][]api.ResolvedPathEntry, error) { + return nil, nil +} + func (s *stubDB) UpsertChannel(_ context.Context, _ []byte, _ []byte, _, _ string) (int, error) { s.upsertChannelCalls++ return 0, nil diff --git a/internal/ingest/packet.go b/internal/ingest/packet.go index 44b5d2f..4a16a04 100644 --- a/internal/ingest/packet.go +++ b/internal/ingest/packet.go @@ -55,6 +55,7 @@ type InsertObservationParams struct { CodingRate int16 SourceBroker string PayloadType int16 + ResolvedEndpoints json.RawMessage } // RadioSettings holds the radio configuration for an observer, populated from @@ -754,6 +755,38 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ if err != nil { log.Printf("ingest[%s]: db: get observer radio failed for %s: %v", w.cfg.BrokerName, pubkeyHex, err) } + // Save the same endpoint resolution used by the live event in the observation INSERT. + var resolvedSource, resolvedDestination *api.ResolvedHop + if packet.PayloadType() == meshcore.PayloadTypeAdvert && originPubkey != nil { + // Exact match: ADVERT carries the sender's real identity pubkey, not a + // short ambiguous hash prefix like the other resolvable payload types. + if nodeID, err := w.db.GetNodeByPubkey(ctx, originPubkey); err == nil { + if nodes, err := w.db.GetNodesByIDs(ctx, []uuid.UUID{nodeID}); err == nil { + hop := api.ResolveExactNode(nodes[nodeID]) + resolvedSource = &hop + } + } + } else if len(sourceHashByte) == 1 { + if r, err := w.db.ResolveEndpointHashes(ctx, iata, [][]byte{sourceHashByte}); err == nil { + hop := api.BuildResolvedPath([][]byte{sourceHashByte}, r)[0] + resolvedSource = &hop + } + } + if len(destHashByte) == 1 { + if r, err := w.db.ResolveEndpointHashes(ctx, iata, [][]byte{destHashByte}); err == nil { + hop := api.BuildResolvedPath([][]byte{destHashByte}, r)[0] + resolvedDestination = &hop + } + } + var resolvedEndpoints json.RawMessage + snapshot := api.PacketEndpointSnapshot{Source: resolvedSource, Destination: resolvedDestination} + if snapshot.HasResolvedNodes() { + resolvedEndpoints, err = json.Marshal(snapshot) + if err != nil { + log.Printf("ingest[%s]: endpoint snapshot encoding failed: %v", w.cfg.BrokerName, err) + resolvedEndpoints = nil // optional enrichment must not discard the observation + } + } oParams := InsertObservationParams{ PacketHash: packetHash[:], ObserverID: id, @@ -772,6 +805,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ CodingRate: radio.CR, SourceBroker: w.cfg.BrokerName, PayloadType: int16(packet.PayloadType()), + ResolvedEndpoints: resolvedEndpoints, } inserted, err := w.db.InsertObservation(ctx, oParams) if err != nil { @@ -839,29 +873,6 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ } w.runCapabilityDetection(ctx, packet.PayloadType(), packet.PathHashSize(), resolvedIDs) - var resolvedSource, resolvedDestination *api.ResolvedHop - if packet.PayloadType() == meshcore.PayloadTypeAdvert && originPubkey != nil { - // Exact match: ADVERT carries the sender's real identity pubkey, not a - // short ambiguous hash prefix like the other resolvable payload types. - if nodeID, err := w.db.GetNodeByPubkey(ctx, originPubkey); err == nil { - if nodes, err := w.db.GetNodesByIDs(ctx, []uuid.UUID{nodeID}); err == nil { - hop := api.ResolveExactNode(nodes[nodeID]) - resolvedSource = &hop - } - } - } else if len(sourceHashByte) == 1 { - if r, err := w.db.ResolvePathHashes(ctx, iata, [][]byte{sourceHashByte}); err == nil { - hop := api.BuildResolvedPath([][]byte{sourceHashByte}, r)[0] - resolvedSource = &hop - } - } - if len(destHashByte) == 1 { - if r, err := w.db.ResolvePathHashes(ctx, iata, [][]byte{destHashByte}); err == nil { - hop := api.BuildResolvedPath([][]byte{destHashByte}, r)[0] - resolvedDestination = &hop - } - } - if inserted { w.handlePayloadTypeSideEffects(ctx, packet, iata, packetHash[:], radio, scopeID, matchedScope, pubkeyBytes, float32(parseNumber(envelope.SNR))) evt := packetObservationEvent{} diff --git a/internal/ingest/packet_endpoints_test.go b/internal/ingest/packet_endpoints_test.go new file mode 100644 index 0000000..61a4706 --- /dev/null +++ b/internal/ingest/packet_endpoints_test.go @@ -0,0 +1,123 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package ingest + +import ( + "context" + "encoding/hex" + "encoding/json" + "errors" + "math" + "reflect" + "testing" + + "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/google/uuid" + "github.com/meshcore-go/meshcore-go" +) + +type endpointCaptureDB struct { + *stubDB + node api.ResolvedNode + observed []InsertObservationParams + lookups int + pathLookups int + missingNode bool +} + +func (s *endpointCaptureDB) GetNodeByPubkey(context.Context, []byte) (uuid.UUID, error) { + if s.missingNode { + return uuid.Nil, errors.New("node not yet advertised") + } + return s.node.ID, nil +} + +func (s *endpointCaptureDB) UpsertNode(context.Context, UpsertNodeParams, RadioSettings) (uuid.UUID, error) { + s.missingNode = false + return s.node.ID, nil +} +func (s *endpointCaptureDB) GetNodesByIDs(context.Context, []uuid.UUID) (map[uuid.UUID]*api.ResolvedNode, error) { + return map[uuid.UUID]*api.ResolvedNode{s.node.ID: &s.node}, nil +} +func (s *endpointCaptureDB) ResolvePathHashes(_ context.Context, _ string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error) { + if len(hashes) > 0 { + s.pathLookups++ + } + return nil, nil // a companion must not be found by the relay-only resolver +} + +func (s *endpointCaptureDB) ResolveEndpointHashes(_ context.Context, iata string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error) { + if len(hashes) == 0 { + return nil, nil + } + s.lookups++ + if iata != "YYZ" || s.missingNode { + return nil, nil + } + return map[string][]api.ResolvedPathEntry{ + "aa": {{NodeID: s.node.ID, Name: s.node.Name, PublicKey: []byte{0xaa}}, {NodeID: uuid.Nil, PublicKey: []byte{0xab}}}, + }, nil +} + +func (s *endpointCaptureDB) InsertObservation(_ context.Context, observation InsertObservationParams) (bool, error) { + s.observed = append(s.observed, observation) + return true, nil +} + +func TestHandlePacketCapturesEndpoints(t *testing.T) { + name := "Companion 👋" + for _, kind := range []string{"advert", "first advert", "direct message", "unresolved direct", "unaddressed", "encoding failure"} { + t.Run(kind, func(t *testing.T) { + w, base := newTestWorker() + db := &endpointCaptureDB{stubDB: base, node: api.ResolvedNode{ID: uuid.New(), Name: &name, PublicKey: "aa"}} + w.db = db + packet := buildAdvertPacket(t, false) + want := api.PacketEndpointSnapshot{} + switch kind { + case "advert": + hop := api.ResolveExactNode(&db.node) + want.Source = &hop + case "first advert": + db.missingNode = true + case "direct message", "unresolved direct": + // Destination, source, MAC and a minimal ciphertext envelope. + packet = &meshcore.Packet{Header: meshcore.MakeHeader(meshcore.RouteTypeFlood, meshcore.PayloadTypeTxtMsg, 0), Payload: append([]byte{0xbb, 0xaa, 0, 0}, make([]byte, 16)...)} + db.missingNode = kind == "unresolved direct" + resolved, _ := db.ResolveEndpointHashes(context.Background(), "YYZ", [][]byte{{0xaa}}) + source := api.BuildResolvedPath([][]byte{{0xaa}}, resolved)[0] + destination := api.BuildResolvedPath([][]byte{{0xbb}}, nil)[0] + want.Source, want.Destination = &source, &destination + db.lookups = 0 + case "unaddressed": + packet = buildTracePacket(t) + case "encoding failure": + nan := math.NaN() + db.node.Latitude = &nan + } + w.handlePacket(context.Background(), "YYZ", hex.EncodeToString([]byte{1, 2}), packetEnvelope(t, packet)) + if len(db.observed) != 1 { + t.Fatalf("observation was lost or written twice: %d", len(db.observed)) + } + if kind == "first advert" || kind == "unresolved direct" || kind == "unaddressed" || kind == "encoding failure" { + if db.observed[0].ResolvedEndpoints != nil { + t.Fatal("empty or failed endpoint resolution must remain SQL NULL") + } + if kind == "first advert" && db.missingNode { + t.Fatal("advert side effects did not make the node available for a later lookup") + } + return + } + var got api.PacketEndpointSnapshot + if err := json.Unmarshal(db.observed[0].ResolvedEndpoints, &got); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(got, want) { + t.Fatalf("captured endpoints differ: got %+v, want %+v", got, want) + } + if kind == "direct message" && (db.lookups != 2 || db.pathLookups != 0) { + t.Fatalf("wrong resolver or repeated work: endpoint=%d relay=%d", db.lookups, db.pathLookups) + } + }) + } +}