fix(packets): preserve captured endpoint names in history (#128)

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)
+	}
+}

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;

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)
+		}
+	}
+}

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) {

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")
 	}

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

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()

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 {

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
 	// ============================================================

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
 
 

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{}) {

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"

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"

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

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"`

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

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)
+		}
+	}
+}

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)

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

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{}

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)
+			}
+		})
+	}
+}