diff --git a/db/packet_summary_integration_test.go b/db/packet_summary_integration_test.go new file mode 100644 index 0000000..7f3003d --- /dev/null +++ b/db/packet_summary_integration_test.go @@ -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) + } +} diff --git a/db/packets.go b/db/packets.go index ddf0ad2..2ea2c71 100644 --- a/db/packets.go +++ b/db/packets.go @@ -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{ diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 2e72472..c44a974 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -462,6 +462,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, @@ -557,6 +559,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, @@ -595,6 +599,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, diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 998889f..d17cc38 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -3042,6 +3042,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, @@ -3091,6 +3093,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"` @@ -3130,6 +3133,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, @@ -3159,6 +3163,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, @@ -3197,6 +3203,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"` @@ -3233,6 +3240,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, @@ -3304,6 +3312,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, @@ -3352,6 +3362,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"` @@ -3406,6 +3417,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, diff --git a/docs/docs.go b/docs/docs.go index b7cf2ac..b3a2fc0 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -3535,7 +3535,7 @@ const docTemplate = `{ "type": "string" }, "summary": { - "description": "human-readable payload summary", + "description": "advert name from this packet; omitted when unavailable or unsupported", "type": "string" } } diff --git a/docs/swagger.json b/docs/swagger.json index 5182f72..0019da2 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -3533,7 +3533,7 @@ "type": "string" }, "summary": { - "description": "human-readable payload summary", + "description": "advert name from this packet; omitted when unavailable or unsupported", "type": "string" } } diff --git a/docs/swagger.yaml b/docs/swagger.yaml index 64962a7..6fc8907 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -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: diff --git a/internal/api/packets.go b/internal/api/packets.go index c75aa6f..bb4ac37 100644 --- a/internal/api/packets.go +++ b/internal/api/packets.go @@ -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. diff --git a/internal/ingest/packet.go b/internal/ingest/packet.go index be8d007..59408de 100644 --- a/internal/ingest/packet.go +++ b/internal/ingest/packet.go @@ -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 diff --git a/internal/ingest/packet_summary_test.go b/internal/ingest/packet_summary_test.go new file mode 100644 index 0000000..0700335 --- /dev/null +++ b/internal/ingest/packet_summary_test.go @@ -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") + } + } + }) + }) + } +}