dandri/beacon-server
1
0
Fork 0
mirror of https://github.com/MeshCore-Beacon/beacon-server.git synced 2026-10-17 18:00:13 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(packets): populate advert summaries in history and live events (#143)

Browse Source
This commit is contained in:
n30nex
2026-09-18 06:18:19 -04:00
committed by GitHub
parent bcdea76c26
commit 85b1c3961d
10 changed files with 269 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
db/packet_summary_integration_test.go
+116
View File
@@ -0,0 +1,116 @@
// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later
package db
import (
"context"
"os"
"testing"
"time"
sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
"github.com/MeshCore-Beacon/beacon-server/internal/api"
"github.com/jackc/pgx/v5"
)
type summaryQueryCounter struct {
pgx.Tx
calls int
}
func (c *summaryQueryCounter) Query(ctx context.Context, query string, args ...any) (pgx.Rows, error) {
c.calls++
return c.Tx.Query(ctx, query, args...)
}
func TestPacketSummariesPostgres(t *testing.T) {
dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN")
if dsn == "" {
t.Skip("set BEACON_TEST_POSTGRES_DSN for PostgreSQL regression")
}
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())
for _, table := range []string{"packets", "packet_observations", "observers", "transport_scopes"} {
if _, err := tx.Exec(ctx, "CREATE TEMP TABLE "+table+" (LIKE public."+table+" INCLUDING ALL) ON COMMIT DROP"); err != nil {
t.Fatal(err)
}
}
_, err = tx.Exec(ctx, `
INSERT INTO packets(packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,parsed_payload,first_heard_at,last_heard_at)
SELECT decode(lpad(to_hex(id),2,'0'),'hex'),kind,0,1,'\x00','\x00',parsed::jsonb,
'2026-01-01'::timestamptz+id*interval '1 second','2026-01-01'::timestamptz+id*interval '1 second'
FROM (VALUES
(1,4,'{"type":"ADVERT","appData":{"name":"MD00-Repeater"}}'),
(2,4,'{"type":"ADVERT","appData":{"name":"Relay 📡"}}'),
(3,4,'{"type":"ADVERT","appData":{"name":""}}'),
(4,4,'{"type":"ADVERT","appData":{}}'),
(5,4,NULL),
(6,4,'{"type":"ADVERT","appData":{"name":null}}'),
(7,4,'{"type":"ADVERT","appData":{"name":42}}'),
(8,2,'{"appData":{"name":"not an advert"}}'),
(9,4,'{"type":"ADVERT","appData":[]}')
) v(id,kind,parsed);
`)
if err != nil {
t.Fatal(err)
}
counter := &summaryQueryCounter{Tx: tx}
store := &Store{q: sqlc.New(counter)}
check := func(items []api.PacketSummary) {
t.Helper()
if len(items) != 9 {
t.Fatalf("got %d packets, want 9", len(items))
}
for _, item := range items {
want := map[string]string{"01": "MD00-Repeater", "02": "Relay 📡"}[item.PacketHash]
if want == "" {
if item.Summary != nil {
t.Fatalf("unexpected summary for %s: %q", item.PacketHash, *item.Summary)
}
} else if item.Summary == nil || *item.Summary != want {
t.Fatalf("summary for %s = %v, want %q", item.PacketHash, item.Summary, want)
}
}
if counter.calls != 1 {
t.Fatalf("list made %d queries, want 1", counter.calls)
}
counter.calls = 0
}
page, err := store.ListPackets(ctx, nil, nil, nil, nil, time.Time{}, time.Time{}, 0, 20)
if err != nil {
t.Fatal(err)
}
check(page.Items) // names remain available before any observation arrives
_, err = tx.Exec(ctx, `
INSERT INTO observers(id,public_key) VALUES ('00000000-0000-0000-0000-000000000001','\x01');
INSERT INTO packet_observations(id,packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count)
SELECT id,decode(lpad(to_hex(id),2,'0'),'hex'),'00000000-0000-0000-0000-000000000001','YVR',
'2026-01-01'::timestamptz+id*interval '1 second',0,1,0 FROM generate_series(1,9) id;
`)
if err != nil {
t.Fatal(err)
}
for _, iatas := range [][]string{nil, {"YVR"}} {
page, err := store.ListPackets(ctx, nil, nil, iatas, nil, time.Time{}, time.Time{}, 0, 20)
if err != nil {
t.Fatal(err)
}
check(page.Items)
rows, err := store.ListPacketsAfterID(ctx, 0, -1, -1, iatas, "", 20)
if err != nil {
t.Fatal(err)
}
check(rows)
}
}
db/packets.go
+9
View File
@@ -137,6 +137,9 @@ func (s *Store) ListPackets(ctx context.Context, payloadTypes, routeTypes []int1
LastHeardAt: v.LastHeardAt.Time.UnixMilli(),
ObservationCount: int32(v.ObservationCount),
}
if v.Summary != "" {
item.Summary = &v.Summary
}
if v.LatestObserverID != (uuid.UUID{}) {
endpoints, _ := decodePacketEndpointSnapshot(v.LatestObserverResolvedEndpoints)
item.LatestObserver = &api.PacketLatestObserver{
@@ -219,6 +222,9 @@ func (s *Store) listPacketsByIATAs(ctx context.Context, payloadTypes, routeTypes
LastHeardAt: v.LastHeardAt.Time.UnixMilli(),
ObservationCount: int32(v.ObservationCount),
}
if v.Summary != "" {
item.Summary = &v.Summary
}
if v.LatestObserverID != (uuid.UUID{}) {
endpoints, _ := decodePacketEndpointSnapshot(v.LatestObserverResolvedEndpoints)
item.LatestObserver = &api.PacketLatestObserver{
@@ -278,6 +284,9 @@ func (s *Store) ListPacketsAfterID(ctx context.Context, afterObservationID int64
LastHeardAt: v.LastHeardAt.Time.UnixMilli(),
ObservationCount: int32(v.ObservationCount),
}
if v.Summary != "" {
item.Summary = &v.Summary
}
if v.LatestObserverID != (uuid.UUID{}) {
endpoints, _ := decodePacketEndpointSnapshot(v.LatestObserverResolvedEndpoints)
item.LatestObserver = &api.PacketLatestObserver{
db/queries/queries.sql
+6
View File
@@ -463,6 +463,8 @@ SELECT COUNT(*) FROM packet_observations WHERE packet_hash = $1;
SELECT
p.packet_hash,
p.payload_type,
COALESCE(CASE WHEN p.payload_type = 4 AND jsonb_typeof(p.parsed_payload #> '{appData,name}') = 'string'
THEN p.parsed_payload #>> '{appData,name}' END, '')::text AS summary,
p.route_type,
p.first_heard_at,
p.last_heard_at,
@@ -558,6 +560,8 @@ page AS (
SELECT
p.packet_hash,
p.payload_type,
COALESCE(CASE WHEN p.payload_type = 4 AND jsonb_typeof(p.parsed_payload #> '{appData,name}') = 'string'
THEN p.parsed_payload #>> '{appData,name}' END, '')::text AS summary,
p.route_type,
p.first_heard_at,
p.last_heard_at,
@@ -596,6 +600,8 @@ ORDER BY sh.site_heard_at DESC;
SELECT
p.packet_hash,
p.payload_type,
COALESCE(CASE WHEN p.payload_type = 4 AND jsonb_typeof(p.parsed_payload #> '{appData,name}') = 'string'
THEN p.parsed_payload #>> '{appData,name}' END, '')::text AS summary,
p.route_type,
p.first_heard_at,
p.last_heard_at,
db/sqlc/queries.sql.go
+12
View File
@@ -3157,6 +3157,8 @@ const listPackets = `-- name: ListPackets :many
SELECT
p.packet_hash,
p.payload_type,
COALESCE(CASE WHEN p.payload_type = 4 AND jsonb_typeof(p.parsed_payload #> '{appData,name}') = 'string'
THEN p.parsed_payload #>> '{appData,name}' END, '')::text AS summary,
p.route_type,
p.first_heard_at,
p.last_heard_at,
@@ -3206,6 +3208,7 @@ type ListPacketsParams struct {
type ListPacketsRow struct {
PacketHash []byte `json:"packet_hash"`
PayloadType int16 `json:"payload_type"`
Summary string `json:"summary"`
RouteType int16 `json:"route_type"`
FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"`
LastHeardAt pgtype.Timestamptz `json:"last_heard_at"`
@@ -3245,6 +3248,7 @@ func (q *Queries) ListPackets(ctx context.Context, arg ListPacketsParams) ([]Lis
if err := rows.Scan(
&i.PacketHash,
&i.PayloadType,
&i.Summary,
&i.RouteType,
&i.FirstHeardAt,
&i.LastHeardAt,
@@ -3274,6 +3278,8 @@ const listPacketsAfterID = `-- name: ListPacketsAfterID :many
SELECT
p.packet_hash,
p.payload_type,
COALESCE(CASE WHEN p.payload_type = 4 AND jsonb_typeof(p.parsed_payload #> '{appData,name}') = 'string'
THEN p.parsed_payload #>> '{appData,name}' END, '')::text AS summary,
p.route_type,
p.first_heard_at,
p.last_heard_at,
@@ -3312,6 +3318,7 @@ type ListPacketsAfterIDParams struct {
type ListPacketsAfterIDRow struct {
PacketHash []byte `json:"packet_hash"`
PayloadType int16 `json:"payload_type"`
Summary string `json:"summary"`
RouteType int16 `json:"route_type"`
FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"`
LastHeardAt pgtype.Timestamptz `json:"last_heard_at"`
@@ -3348,6 +3355,7 @@ func (q *Queries) ListPacketsAfterID(ctx context.Context, arg ListPacketsAfterID
if err := rows.Scan(
&i.PacketHash,
&i.PayloadType,
&i.Summary,
&i.RouteType,
&i.FirstHeardAt,
&i.LastHeardAt,
@@ -3419,6 +3427,8 @@ page AS (
SELECT
p.packet_hash,
p.payload_type,
COALESCE(CASE WHEN p.payload_type = 4 AND jsonb_typeof(p.parsed_payload #> '{appData,name}') = 'string'
THEN p.parsed_payload #>> '{appData,name}' END, '')::text AS summary,
p.route_type,
p.first_heard_at,
p.last_heard_at,
@@ -3467,6 +3477,7 @@ type ListPacketsByIATAsParams struct {
type ListPacketsByIATAsRow struct {
PacketHash []byte `json:"packet_hash"`
PayloadType int16 `json:"payload_type"`
Summary string `json:"summary"`
RouteType int16 `json:"route_type"`
FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"`
LastHeardAt pgtype.Timestamptz `json:"last_heard_at"`
@@ -3521,6 +3532,7 @@ func (q *Queries) ListPacketsByIATAs(ctx context.Context, arg ListPacketsByIATAs
if err := rows.Scan(
&i.PacketHash,
&i.PayloadType,
&i.Summary,
&i.RouteType,
&i.FirstHeardAt,
&i.LastHeardAt,
docs/docs.go
+1 -1
View File
@@ -3568,7 +3568,7 @@ const docTemplate = `{
"type": "string"
},
"summary": {
"description": "human-readable payload summary",
"description": "advert name from this packet; omitted when unavailable or unsupported",
"type": "string"
}
}
docs/swagger.json
+1 -1
View File
@@ -3566,7 +3566,7 @@
"type": "string"
},
"summary": {
"description": "human-readable payload summary",
"description": "advert name from this packet; omitted when unavailable or unsupported",
"type": "string"
}
}
docs/swagger.yaml
+1 -1
View File
@@ -813,7 +813,7 @@ definitions:
description: matched transport scope name e.g. "#bc"
type: string
summary:
description: human-readable payload summary
description: advert name from this packet; omitted when unavailable or unsupported
type: string
type: object
github_com_MeshCore-Beacon_beacon-server_internal_api.PacketTransportCodes:
internal/api/packets.go
+1 -1
View File
@@ -40,7 +40,7 @@ type PacketSummary struct {
LastHeardAt int64 `json:"lastHeardAt"` // epoch ms
ObservationCount int32 `json:"observationCount"`
LatestObserver *PacketLatestObserver `json:"latestObserver,omitempty"`
Summary *string `json:"summary,omitempty"` // human-readable payload summary
Summary *string `json:"summary,omitempty"` // advert name from this packet; omitted when unavailable or unsupported
}
// PacketPathLength is the decoded path_length byte from a packet observation.
internal/ingest/packet.go
+6
View File
@@ -81,6 +81,7 @@ type packetObservationEvent struct {
IsFirstObservation bool `json:"isFirstObservation"`
ObservationCount int64 `json:"observationCount"`
Scope *string `json:"scope,omitempty"`
Summary *string `json:"summary,omitempty"` // same advert name as REST list/backfill rows
} `json:"packet"`
Observation struct {
ObserverID string `json:"observerId"`
@@ -343,6 +344,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
var channelHash []byte
originPubkey := []byte(nil)
var parsedPayload json.RawMessage
var summary *string
var traceTag []byte
// For PayloadTypeTrace, packet.Path holds one SNR byte per hop (not hashes -- see
// below), so the "physical route" hashes for resolvedPath/known-route purposes come
@@ -415,6 +417,9 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
if hasName {
n := strings.ToValidUTF8(appData.Name, "\uFFFD")
name = &n
if n != "" {
summary = name
}
}
deviceRole := int(flags & 0x0F)
@@ -892,6 +897,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
evt.Packet.RouteType = packet.RouteType()
evt.Packet.RouteTypeName = api.RouteTypeName(int16(packet.RouteType()))
evt.Packet.IsFirstObservation = isNew
evt.Packet.Summary = summary
evt.Observation.ObserverID = id.String()
evt.Observation.ObserverName = observerName
evt.Observation.IATA = iata
internal/ingest/packet_summary_test.go
+116
View File
@@ -0,0 +1,116 @@
// Copyright 2026 Beacon Contributors
// SPDX-License-Identifier: AGPL-3.0-or-later
package ingest
import (
"context"
"crypto/ed25519"
"encoding/json"
"testing"
"time"
"github.com/MeshCore-Beacon/beacon-server/internal/hub"
"github.com/meshcore-go/meshcore-go"
)
// The hub has no subscription acknowledgement. Wait for a probe to make it
// through before ingesting the test packet, rather than assuming a short sleep.
func waitForSummarySubscriber(t *testing.T, ctx context.Context, h *hub.Hub, client *hub.Client) {
t.Helper()
ticker := time.NewTicker(time.Millisecond)
defer ticker.Stop()
for {
h.Broadcast(hub.Event{Type: hub.EventObserverStatus})
select {
case <-client.Send:
return
case <-ticker.C:
case <-ctx.Done():
t.Fatal("hub subscription did not become ready")
}
}
}
func TestAdvertSummaryLive(t *testing.T) {
for _, tc := range []struct {
name string
hasName, tamper bool
want string
}{
{"MD00-Repeater", true, false, "MD00-Repeater"},
{"Relay 📡", true, false, "Relay 📡"},
{"R\xffP", true, false, "R\uFFFDP"},
{"", true, false, ""},
{"", false, false, ""},
{"unverified", true, true, ""},
} {
t.Run(tc.name, func(t *testing.T) {
t.Run("wire", func(t *testing.T) {
w, base := newTestWorker()
db := &frameCaptureDB{stubDB: base}
w.db = db
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
go w.hub.Run()
client := w.hub.NewClient()
w.hub.AddScope(client, "summary", hub.Scope{Events: []hub.EventType{hub.EventPacketObservation, hub.EventObserverStatus}})
waitForSummarySubscriber(t, ctx, w.hub, client)
defer w.hub.Remove(client)
pub, priv, err := ed25519.GenerateKey(nil)
if err != nil {
t.Fatal(err)
}
identity, err := meshcore.NewIdentityFromBytes(pub)
if err != nil {
t.Fatal(err)
}
flags := byte(meshcore.AdvertTypeRepeater)
if tc.hasName {
flags |= 0x80
}
advert := &meshcore.Advert{PublicKey: identity, Timestamp: 12345, RawAppData: append([]byte{flags}, []byte(tc.name)...)}
advert.Sign(priv)
if tc.tamper {
advert.RawAppData[0] ^= 1
}
payload, err := advert.ToBytes()
if err != nil {
t.Fatal(err)
}
packet := &meshcore.Packet{Header: meshcore.MakeHeader(meshcore.RouteTypeFlood, meshcore.PayloadTypeAdvert, 0), Payload: payload}
w.handlePacket(ctx, "YVR", "0102", packetEnvelope(t, packet))
for {
select {
case event := <-client.Send:
if event.Type != hub.EventPacketObservation {
continue
}
for _, raw := range []json.RawMessage{event.Payload, event.PayloadResolved} {
var got struct {
Packet struct {
Summary *string `json:"summary"`
} `json:"packet"`
}
if err := json.Unmarshal(raw, &got); err != nil {
t.Fatal(err)
}
if tc.want == "" {
if got.Packet.Summary != nil {
t.Fatalf("unexpected summary %q", *got.Packet.Summary)
}
} else if got.Packet.Summary == nil || *got.Packet.Summary != tc.want {
t.Fatalf("live summary = %v, want %q", got.Packet.Summary, tc.want)
}
}
return
case <-ctx.Done():
t.Fatal("packet observation event missing")
}
}
})
})
}
}