enrich packet observations

This commit is contained in:
Enot (ded) Skelly committed 2026-07-27 09:41:11 -07:00
1 parent 336b734a42
commit ae0669ccbe
8 files changed
+300 -41

No files matched your search

+37
View File
@@ -58,6 +58,32 @@ func (s *Store) SetPacketDecrypted(ctx context.Context, hash []byte) error {
return s.q.SetPacketDecrypted(ctx, hash)
}
// buildLatestObserverPath fills in PacketLatestObserver's optional path fields from the
// nullable path_length_byte/hash_size/hop_count/path_bytes columns joined in alongside the
// latest (or matching) observation. hashSize/hopCount nil means no observation joined at all
// (only possible via ListPackets/ListPacketsByIATAs' LEFT JOIN LATERAL -- ListPacketsAfterID's
// inner join always has them, callers there can pass &v.Field directly).
func buildLatestObserverPath(pathLengthByte, hashSize, hopCount *int16, pathBytes []byte) (*api.PacketPathLength, *string) {
if hashSize == nil || hopCount == nil {
return nil, nil
}
raw := ""
if pathLengthByte != nil {
raw = fmt.Sprintf("%02x", *pathLengthByte)
}
pathLength := &api.PacketPathLength{
Raw: raw,
HashSize: *hashSize,
HopCount: *hopCount,
}
var pathBytesHex *string
if pathBytes != nil {
s := hex.EncodeToString(pathBytes)
pathBytesHex = &s
}
return pathLength, pathBytesHex
}
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)
@@ -109,6 +135,9 @@ func (s *Store) ListPackets(ctx context.Context, payloadTypes, routeTypes []int1
DisplayName: v.LatestObserverName,
IATA: v.LatestObserverIata,
}
item.LatestObserver.PathLength, item.LatestObserver.PathBytes = buildLatestObserverPath(
&v.LatestObserverPathLengthByte, &v.LatestObserverHashSize, &v.LatestObserverHopCount, v.LatestObserverPathBytes,
)
}
items = append(items, item)
}
@@ -177,6 +206,9 @@ func (s *Store) listPacketsByIATAs(ctx context.Context, payloadTypes, routeTypes
DisplayName: v.LatestObserverName,
IATA: v.LatestObserverIata,
}
item.LatestObserver.PathLength, item.LatestObserver.PathBytes = buildLatestObserverPath(
&v.LatestObserverPathLengthByte, &v.LatestObserverHashSize, &v.LatestObserverHopCount, v.LatestObserverPathBytes,
)
}
items = append(items, item)
}
@@ -224,6 +256,11 @@ func (s *Store) ListPacketsAfterID(ctx context.Context, afterObservationID int64
DisplayName: v.LatestObserverName,
IATA: v.LatestObserverIata,
}
// Inner join here (unlike ListPackets/listPacketsByIATAs' LEFT JOIN LATERAL), so
// these are never nil when an observer was joined at all.
item.LatestObserver.PathLength, item.LatestObserver.PathBytes = buildLatestObserverPath(
&v.LatestObserverPathLengthByte, &v.LatestObserverHashSize, &v.LatestObserverHopCount, v.LatestObserverPathBytes,
)
}
items = append(items, item)
}
+92
View File
@@ -157,6 +157,59 @@ func TestListPackets_LatestObserverSet(t *testing.T) {
}
}
func TestListPackets_LatestObserverPathFields(t *testing.T) {
ctrl := gomock.NewController(t)
mock := mockdb.NewMockQuerier(ctrl)
heardAt := pgtype.Timestamptz{Time: time.UnixMilli(1700000000000), Valid: true}
observerID := uuid.MustParse("00000000-0000-0000-0000-000000000001")
pathLengthByte := int16(0x42)
hashSize := int16(1)
hopCount := int16(2)
pathBytes := []byte{0xa1, 0xb2}
mock.EXPECT().
ListPackets(gomock.Any(), gomock.Any()).
Return([]sqlc.ListPacketsRow{
{
PacketHash: []byte{0xde, 0xad},
FirstHeardAt: heardAt,
LastHeardAt: heardAt,
LatestObserverID: observerID,
LatestObserverPathLengthByte: pathLengthByte,
LatestObserverHashSize: hashSize,
LatestObserverHopCount: hopCount,
LatestObserverPathBytes: pathBytes,
},
}, nil)
store := &Store{q: mock}
page, err := store.ListPackets(context.Background(), nil, nil, nil, nil, time.Time{}, time.Time{}, 0, 10)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
obs := page.Items[0].LatestObserver
if obs == nil {
t.Fatal("expected LatestObserver to be set")
}
if obs.PathLength == nil {
t.Fatal("expected PathLength to be set")
}
if obs.PathLength.HashSize != 1 || obs.PathLength.HopCount != 2 {
t.Errorf("expected hashSize=1 hopCount=2, got hashSize=%d hopCount=%d", obs.PathLength.HashSize, obs.PathLength.HopCount)
}
if obs.PathLength.Raw != "42" {
t.Errorf("expected raw 42, got %s", obs.PathLength.Raw)
}
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.
if obs.ResolvedPath != nil || obs.ResolvedSource != nil || obs.ResolvedDestination != nil {
t.Error("expected no resolved path/source/destination on the list endpoint")
}
}
func TestInsertObservation_Success(t *testing.T) {
ctrl := gomock.NewController(t)
mock := mockdb.NewMockQuerier(ctrl)
@@ -352,6 +405,45 @@ func TestListPacketsAfterID_PassesIATAsAsArray(t *testing.T) {
}
}
func TestListPacketsAfterID_LatestObserverPathFields(t *testing.T) {
ctrl := gomock.NewController(t)
mock := mockdb.NewMockQuerier(ctrl)
heardAt := pgtype.Timestamptz{Time: time.UnixMilli(1700000000000), Valid: true}
observerID := uuid.MustParse("00000000-0000-0000-0000-000000000001")
mock.EXPECT().
ListPacketsAfterID(gomock.Any(), gomock.Any()).
Return([]sqlc.ListPacketsAfterIDRow{
{
PacketHash: []byte{0xde, 0xad},
FirstHeardAt: heardAt,
LastHeardAt: heardAt,
LatestObserverID: observerID,
LatestObserverPathLengthByte: 0x42,
LatestObserverHashSize: 1,
LatestObserverHopCount: 2,
LatestObserverPathBytes: []byte{0xa1, 0xb2},
},
}, nil)
store := &Store{q: mock}
items, err := store.ListPacketsAfterID(context.Background(), 0, -1, -1, nil, "", 50)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}
obs := items[0].LatestObserver
if obs == nil || obs.PathLength == nil {
t.Fatal("expected LatestObserver and PathLength to be set")
}
if obs.PathLength.HashSize != 1 || obs.PathLength.HopCount != 2 {
t.Errorf("expected hashSize=1 hopCount=2, got hashSize=%d hopCount=%d", obs.PathLength.HashSize, obs.PathLength.HopCount)
}
if obs.PathBytes == nil || *obs.PathBytes != "a1b2" {
t.Errorf("expected pathBytes a1b2, got %v", obs.PathBytes)
}
}
func TestListNodeObservations_Pagination(t *testing.T) {
ctrl := gomock.NewController(t)
mock := mockdb.NewMockQuerier(ctrl)
+16 -4
View File
@@ -393,10 +393,14 @@ SELECT
(SELECT COUNT(*) FROM packet_observations po2 WHERE po2.packet_hash = p.packet_hash) AS observation_count,
po.observer_id AS latest_observer_id,
o.display_name AS latest_observer_name,
po.iata AS latest_observer_iata
po.iata AS latest_observer_iata,
po.path_length_byte AS latest_observer_path_length_byte,
po.hash_size AS latest_observer_hash_size,
po.hop_count AS latest_observer_hop_count,
po.path_bytes AS latest_observer_path_bytes
FROM packets p
LEFT JOIN LATERAL (
SELECT observer_id, iata
SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes
FROM packet_observations
WHERE packet_hash = p.packet_hash
ORDER BY heard_at DESC
@@ -436,7 +440,11 @@ SELECT
(SELECT COUNT(*) FROM packet_observations po2 WHERE po2.packet_hash = p.packet_hash) AS observation_count,
po.observer_id AS latest_observer_id,
o.display_name AS latest_observer_name,
po.iata AS latest_observer_iata
po.iata AS latest_observer_iata,
po.path_length_byte AS latest_observer_path_length_byte,
po.hash_size AS latest_observer_hash_size,
po.hop_count AS latest_observer_hop_count,
po.path_bytes AS latest_observer_path_bytes
FROM (
SELECT hits.packet_hash, MAX(hits.heard_at)::timestamptz AS site_heard_at
FROM unnest(@iatas::bpchar[]) AS req(iata)
@@ -467,7 +475,7 @@ FROM (
) sh
JOIN packets p ON p.packet_hash = sh.packet_hash
LEFT JOIN LATERAL (
SELECT observer_id, iata
SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes
FROM packet_observations
WHERE packet_hash = p.packet_hash
ORDER BY heard_at DESC
@@ -490,6 +498,10 @@ SELECT
po.observer_id AS latest_observer_id,
o.display_name AS latest_observer_name,
po.iata AS latest_observer_iata,
po.path_length_byte AS latest_observer_path_length_byte,
po.hash_size AS latest_observer_hash_size,
po.hop_count AS latest_observer_hop_count,
po.path_bytes AS latest_observer_path_bytes,
ts.name AS scope_name
FROM packets p
JOIN packet_observations po ON po.packet_hash = p.packet_hash
+73 -37
View File
@@ -2650,10 +2650,14 @@ SELECT
(SELECT COUNT(*) FROM packet_observations po2 WHERE po2.packet_hash = p.packet_hash) AS observation_count,
po.observer_id AS latest_observer_id,
o.display_name AS latest_observer_name,
po.iata AS latest_observer_iata
po.iata AS latest_observer_iata,
po.path_length_byte AS latest_observer_path_length_byte,
po.hash_size AS latest_observer_hash_size,
po.hop_count AS latest_observer_hop_count,
po.path_bytes AS latest_observer_path_bytes
FROM packets p
LEFT JOIN LATERAL (
SELECT observer_id, iata
SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes
FROM packet_observations
WHERE packet_hash = p.packet_hash
ORDER BY heard_at DESC
@@ -2683,17 +2687,21 @@ 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"`
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"`
}
// Returns packets with the latest observation rolled in for display.
@@ -2728,6 +2736,10 @@ func (q *Queries) ListPackets(ctx context.Context, arg ListPacketsParams) ([]Lis
&i.LatestObserverID,
&i.LatestObserverName,
&i.LatestObserverIata,
&i.LatestObserverPathLengthByte,
&i.LatestObserverHashSize,
&i.LatestObserverHopCount,
&i.LatestObserverPathBytes,
); err != nil {
return nil, err
}
@@ -2750,6 +2762,10 @@ SELECT
po.observer_id AS latest_observer_id,
o.display_name AS latest_observer_name,
po.iata AS latest_observer_iata,
po.path_length_byte AS latest_observer_path_length_byte,
po.hash_size AS latest_observer_hash_size,
po.hop_count AS latest_observer_hop_count,
po.path_bytes AS latest_observer_path_bytes,
ts.name AS scope_name
FROM packets p
JOIN packet_observations po ON po.packet_hash = p.packet_hash
@@ -2774,16 +2790,20 @@ 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"`
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"`
ScopeName *string `json:"scope_name"`
}
// Returns packets with observations after the given observation ID, ordered oldest first.
@@ -2814,6 +2834,10 @@ func (q *Queries) ListPacketsAfterID(ctx context.Context, arg ListPacketsAfterID
&i.LatestObserverID,
&i.LatestObserverName,
&i.LatestObserverIata,
&i.LatestObserverPathLengthByte,
&i.LatestObserverHashSize,
&i.LatestObserverHopCount,
&i.LatestObserverPathBytes,
&i.ScopeName,
); err != nil {
return nil, err
@@ -2839,7 +2863,11 @@ SELECT
(SELECT COUNT(*) FROM packet_observations po2 WHERE po2.packet_hash = p.packet_hash) AS observation_count,
po.observer_id AS latest_observer_id,
o.display_name AS latest_observer_name,
po.iata AS latest_observer_iata
po.iata AS latest_observer_iata,
po.path_length_byte AS latest_observer_path_length_byte,
po.hash_size AS latest_observer_hash_size,
po.hop_count AS latest_observer_hop_count,
po.path_bytes AS latest_observer_path_bytes
FROM (
SELECT hits.packet_hash, MAX(hits.heard_at)::timestamptz AS site_heard_at
FROM unnest($1::bpchar[]) AS req(iata)
@@ -2870,7 +2898,7 @@ FROM (
) sh
JOIN packets p ON p.packet_hash = sh.packet_hash
LEFT JOIN LATERAL (
SELECT observer_id, iata
SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes
FROM packet_observations
WHERE packet_hash = p.packet_hash
ORDER BY heard_at DESC
@@ -2894,18 +2922,22 @@ 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"`
ObservationCount int64 `json:"observation_count"`
LatestObserverID uuid.UUID `json:"latest_observer_id"`
LatestObserverName *string `json:"latest_observer_name"`
LatestObserverIata string `json:"latest_observer_iata"`
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"`
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"`
}
// IATA-filtered packet list, driven from idx_observations_iata_heard.
@@ -2949,6 +2981,10 @@ func (q *Queries) ListPacketsByIATAs(ctx context.Context, arg ListPacketsByIATAs
&i.LatestObserverID,
&i.LatestObserverName,
&i.LatestObserverIata,
&i.LatestObserverPathLengthByte,
&i.LatestObserverHashSize,
&i.LatestObserverHopCount,
&i.LatestObserverPathBytes,
); err != nil {
return nil, err
}
+24
View File
@@ -3139,6 +3139,30 @@ const docTemplate = `{
},
"id": {
"type": "string"
},
"pathBytes": {
"description": "hex-encoded accumulated path hashes",
"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.",
"allOf": [
{
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.PacketPathLength"
}
]
},
"resolvedDestination": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ResolvedHop"
},
"resolvedPath": {
"type": "array",
"items": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ResolvedHop"
}
},
"resolvedSource": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ResolvedHop"
}
}
},
+24
View File
@@ -3137,6 +3137,30 @@
},
"id": {
"type": "string"
},
"pathBytes": {
"description": "hex-encoded accumulated path hashes",
"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.",
"allOf": [
{
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.PacketPathLength"
}
]
},
"resolvedDestination": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ResolvedHop"
},
"resolvedPath": {
"type": "array",
"items": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ResolvedHop"
}
},
"resolvedSource": {
"$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ResolvedHop"
}
}
},
+22
View File
@@ -612,6 +612,28 @@ definitions:
type: string
id:
type: string
pathBytes:
description: hex-encoded accumulated path hashes
type: string
pathLength:
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.
resolvedDestination:
$ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ResolvedHop'
resolvedPath:
items:
$ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ResolvedHop'
type: array
resolvedSource:
$ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ResolvedHop'
type: object
github_com_MeshCore-Beacon_beacon-server_internal_api.PacketObservationDetail:
properties:
+12
View File
@@ -17,6 +17,18 @@ 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 *PacketPathLength `json:"pathLength,omitempty"`
PathBytes *string `json:"pathBytes,omitempty"` // hex-encoded accumulated path hashes
ResolvedPath []ResolvedHop `json:"resolvedPath,omitempty"`
ResolvedSource *ResolvedHop `json:"resolvedSource,omitempty"`
ResolvedDestination *ResolvedHop `json:"resolvedDestination,omitempty"`
}
// PacketSummary is the minimal packet representation used in list responses.