From 632b83c4b90201e04739c7b6dc82e63424ca44cf Mon Sep 17 00:00:00 2001 From: MrAlders0n <55921894+MrAlders0n@users.noreply.github.com> Date: Wed, 14 Oct 2026 17:04:22 -0400 Subject: [PATCH 1/2] feat(api): search packets by hash prefix (#227) --- db/packet_hash_prefix_integration_test.go | 96 ++++++++++++++ db/packets.go | 83 ++++++++++++ db/queries/queries.sql | 64 ++++++++++ db/sqlc/mock/querier.go | 15 +++ db/sqlc/querier.go | 5 + db/sqlc/queries.sql.go | 149 ++++++++++++++++++++++ docs/docs.go | 6 + docs/swagger.json | 6 + docs/swagger.yaml | 5 + internal/api/handlers/packets.go | 41 +++++- internal/api/handlers/packets_test.go | 75 +++++++++++ internal/api/handlers/stub_reader_test.go | 8 ++ internal/api/reader.go | 5 + internal/cache/cache_test.go | 4 + internal/cache/reader.go | 5 + 15 files changed, 566 insertions(+), 1 deletion(-) create mode 100644 db/packet_hash_prefix_integration_test.go diff --git a/db/packet_hash_prefix_integration_test.go b/db/packet_hash_prefix_integration_test.go new file mode 100644 index 0000000..ab9ec58 --- /dev/null +++ b/db/packet_hash_prefix_integration_test.go @@ -0,0 +1,96 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package db + +import ( + "context" + "encoding/hex" + "os" + "slices" + "testing" + "time" + + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/jackc/pgx/v5" +) + +func TestSearchPacketsByHashPrefixPostgres(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()) + 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) + } + } + // Newest first by last_heard_at: fb00 > fbff > fc00 > fa00 > ffff. + _, err = tx.Exec(ctx, ` +INSERT INTO packets (packet_hash, payload_type, payload_version, route_type, raw_payload, raw_header, first_heard_at, last_heard_at) VALUES + ('\x6ddf62fb00000000', 2, 0, 1, '\x00', '\x00', NOW(), NOW()), + ('\x6ddf62fbffffffff', 2, 0, 1, '\x00', '\x00', NOW(), NOW() - interval '1 minute'), + ('\x6ddf62fc00000000', 2, 0, 1, '\x00', '\x00', NOW(), NOW() - interval '2 minutes'), + ('\x6ddf62fa00000000', 2, 0, 1, '\x00', '\x00', NOW(), NOW() - interval '3 minutes'), + ('\xffffffffffffffff', 2, 0, 1, '\x00', '\x00', NOW(), NOW() - interval '4 minutes'); +INSERT INTO observers (id, public_key) VALUES ('00000000-0000-0000-0000-000000000001', '\x02'); +INSERT INTO packet_observations (id, packet_hash, observer_id, iata, heard_at, path_length_byte, hash_size, hop_count, path_bytes) +VALUES (1, '\x6ddf62fbffffffff', '00000000-0000-0000-0000-000000000001', 'YOW', NOW(), 0, 1, 0, '\x');`) + if err != nil { + t.Fatal(err) + } + store := &Store{q: sqlc.New(tx)} + search := func(lo, hi string, iatas []string, cursor int64, limit int32) api.Page[api.PacketSummary] { + t.Helper() + loB, _ := hex.DecodeString(lo) + var hiB []byte + if hi != "" { + hiB, _ = hex.DecodeString(hi) + } + page, err := store.SearchPacketsByHashPrefix(ctx, loB, hiB, nil, nil, iatas, nil, time.Time{}, time.Time{}, cursor, limit) + if err != nil { + t.Fatal(err) + } + return page + } + hashes := func(p api.Page[api.PacketSummary]) []string { + var out []string + for _, it := range p.Items { + out = append(out, it.PacketHash) + } + return out + } + + if got := hashes(search("6ddf62fb", "6ddf62fc", nil, 0, 50)); !slices.Equal(got, []string{"6ddf62fb00000000", "6ddf62fbffffffff"}) { + t.Errorf("8-char prefix: %v", got) + } + if got := hashes(search("6ddf60", "6ddf70", nil, 0, 50)); len(got) != 4 { + t.Errorf("odd prefix: %v", got) + } + if got := hashes(search("ffff", "", nil, 0, 50)); !slices.Equal(got, []string{"ffffffffffffffff"}) { + t.Errorf("unbounded prefix: %v", got) + } + if got := hashes(search("6ddf62fb", "6ddf62fc", []string{"YOW"}, 0, 50)); !slices.Equal(got, []string{"6ddf62fbffffffff"}) { + t.Errorf("iata filter: %v", got) + } + first := search("6ddf62fb", "6ddf62fc", nil, 0, 1) + if !first.HasMore || first.NextCursor == nil || hashes(first)[0] != "6ddf62fb00000000" { + t.Fatalf("first page: %+v", first) + } + if got := hashes(search("6ddf62fb", "6ddf62fc", nil, *first.NextCursor, 1)); !slices.Equal(got, []string{"6ddf62fbffffffff"}) { + t.Errorf("second page: %v", got) + } +} diff --git a/db/packets.go b/db/packets.go index 620acce..172084e 100644 --- a/db/packets.go +++ b/db/packets.go @@ -4,6 +4,7 @@ package db import ( + "bytes" "context" "encoding/binary" "encoding/hex" @@ -253,6 +254,88 @@ func (s *Store) listPacketsByIATAs(ctx context.Context, payloadTypes, routeTypes }, nil } +// SearchPacketsByHashPrefix lists packets whose hash is in [lo, hi); nil hi is unbounded. +func (s *Store) SearchPacketsByHashPrefix(ctx context.Context, lo, hi []byte, payloadTypes, routeTypes []int16, iatas []string, scopes []string, since, until time.Time, cursor int64, limit int32) (api.Page[api.PacketSummary], error) { + var cursorTS pgtype.Timestamptz + if cursor > 0 { + cursorTS = pgtype.Timestamptz{Time: time.UnixMilli(cursor), Valid: true} + } + var sinceTS pgtype.Timestamptz + if !since.IsZero() { + sinceTS = pgtype.Timestamptz{Time: since, Valid: true} + } + var untilTS pgtype.Timestamptz + if !until.IsZero() { + untilTS = pgtype.Timestamptz{Time: until, Valid: true} + } + if hi == nil { + // Sorts above any stored hash, keeping the range closed for the planner. + hi = bytes.Repeat([]byte{0xff}, 64) + } + rows, err := s.q.SearchPacketsByHashPrefix(ctx, sqlc.SearchPacketsByHashPrefixParams{ + HashLo: lo, + HashHi: hi, + PayloadTypes: payloadTypes, + RouteTypes: routeTypes, + SinceTs: sinceTS, + UntilTs: untilTS, + CursorTs: cursorTS, + ScopeNames: scopes, + Iatas: iatas, + PageLimit: limit + 1, + }) + if err != nil { + return api.Page[api.PacketSummary]{}, err + } + hasMore := len(rows) > int(limit) + if hasMore { + rows = rows[:limit] + } + items := make([]api.PacketSummary, 0, len(rows)) + lookups := make([]endpointLookup, 0, len(rows)) + for _, v := range rows { + var lookup endpointLookup + item := api.PacketSummary{ + PacketHash: hex.EncodeToString(v.PacketHash), + PayloadType: v.PayloadType, + PayloadTypeName: api.PayloadTypeName(v.PayloadType), + RouteType: v.RouteType, + RouteTypeName: api.RouteTypeName(v.RouteType), + Scope: v.ScopeName, + FirstHeardAt: v.FirstHeardAt.Time.UnixMilli(), + LastHeardAt: v.LastHeardAt.Time.UnixMilli(), + ObservationCount: int32(v.ObservationCount), + } + if v.Summary != "" { + item.Summary = &v.Summary + } + if v.LatestObserverID != (uuid.UUID{}) { + item.LatestObserver = &api.PacketLatestObserver{ + ID: v.LatestObserverID, + DisplayName: v.LatestObserverName, + IATA: v.LatestObserverIata, + } + lookup = endpointLookup{iata: v.LatestObserverIata, ep: parsePacketEndpoints(v.PayloadType, v.RawPayload, v.OriginPubkey)} + item.LatestObserver.PathLength, item.LatestObserver.PathBytes = buildLatestObserverPath( + &v.LatestObserverPathLengthByte, &v.LatestObserverHashSize, &v.LatestObserverHopCount, v.LatestObserverPathBytes, + ) + } + items = append(items, item) + lookups = append(lookups, lookup) + } + s.fillLatestEndpoints(ctx, items, lookups) + var nextCursor *int64 + if hasMore && len(items) > 0 { + last := items[len(items)-1].LastHeardAt + nextCursor = &last + } + return api.Page[api.PacketSummary]{ + Items: items, + NextCursor: nextCursor, + HasMore: hasMore, + }, nil +} + func (s *Store) ListPacketsAfterID(ctx context.Context, afterObservationID int64, payloadType, routeType int16, iatas []string, scope string, limit int32) ([]api.PacketSummary, error) { rows, err := s.q.ListPacketsAfterID(ctx, sqlc.ListPacketsAfterIDParams{ ID: afterObservationID, diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 1fb4096..63925c1 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -633,6 +633,70 @@ LEFT JOIN observers o ON o.id = po.observer_id LEFT JOIN transport_scopes ts ON ts.id = p.scope_id ORDER BY sh.site_heard_at DESC; +-- name: SearchPacketsByHashPrefix :many +-- Packets whose hash falls in [lo, hi). The materialized CTE pins the primary key +-- range scan; otherwise generic plans walk idx_packets_last_heard across the table. +-- IATA filtering is a membership probe, so ordering and the cursor stay on global +-- last_heard_at. +WITH hits AS MATERIALIZED ( + SELECT packet_hash FROM packets + WHERE packet_hash >= @hash_lo::bytea AND packet_hash < @hash_hi::bytea +) +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}' + WHEN p.payload_type = 3 AND p.parsed_payload ->> 'type' = 'ACK' + AND jsonb_typeof(p.parsed_payload -> 'checksum') = 'string' + AND p.parsed_payload ->> 'checksum' ~ '^[0-9a-fA-F]{8}$' + THEN 'ACK ' || lower(p.parsed_payload ->> 'checksum') + WHEN p.payload_type = 9 AND p.parsed_payload ->> 'type' IN ('TRACE', 'PING') + AND jsonb_typeof(p.parsed_payload -> 'traceTag') = 'string' + AND p.parsed_payload ->> 'traceTag' ~ '^[0-9a-fA-F]{8}$' + THEN (p.parsed_payload ->> 'type') || ' ' || lower(p.parsed_payload ->> 'traceTag') + END, '')::text AS summary, + p.route_type, + p.first_heard_at, + p.last_heard_at, + p.scope_id, + ts.name AS scope_name, + p.observation_count AS observation_count, + -- sqlc loses LATERAL nullability; these scalar defaults are ignored when observer_id is NULL. + po.observer_id AS latest_observer_id, + o.display_name AS latest_observer_name, + COALESCE(po.iata, ''::bpchar) AS latest_observer_iata, + 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, + p.raw_payload, + p.origin_pubkey +FROM hits h +JOIN packets p ON p.packet_hash = h.packet_hash +LEFT JOIN LATERAL ( + 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 + LIMIT 1 +) po ON true +LEFT JOIN observers o ON o.id = po.observer_id +LEFT JOIN transport_scopes ts ON ts.id = p.scope_id +WHERE + (COALESCE(cardinality(@payload_types::smallint[]), 0) = 0 OR p.payload_type = ANY(@payload_types::smallint[])) + AND (COALESCE(cardinality(@route_types::smallint[]), 0) = 0 OR p.route_type = ANY(@route_types::smallint[])) + AND (@since_ts::timestamptz IS NULL OR p.first_heard_at >= @since_ts) + AND (@until_ts::timestamptz IS NULL OR p.first_heard_at <= @until_ts) + AND (@cursor_ts::timestamptz IS NULL OR p.last_heard_at < @cursor_ts) + AND (COALESCE(cardinality(@scope_names::text[]), 0) = 0 OR ts.name = ANY(@scope_names::text[])) + AND (COALESCE(cardinality(@iatas::bpchar[]), 0) = 0 OR EXISTS ( + SELECT 1 FROM packet_observations pi + WHERE pi.packet_hash = p.packet_hash AND pi.iata = ANY(@iatas::bpchar[]))) +ORDER BY p.last_heard_at DESC +LIMIT @page_limit; + -- name: ListPacketsAfterID :many -- Returns packets with observations after the given observation ID, ordered oldest first. -- Used for WS reconnect backfill. Pass afterObservationId=0 to start from the beginning. diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index 6ab1f20..d590eb5 100644 --- a/db/sqlc/mock/querier.go +++ b/db/sqlc/mock/querier.go @@ -2325,6 +2325,21 @@ func (mr *MockQuerierMockRecorder) SearchKnownRoutesByNodes(ctx, arg any) *gomoc return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SearchKnownRoutesByNodes", reflect.TypeOf((*MockQuerier)(nil).SearchKnownRoutesByNodes), ctx, arg) } +// SearchPacketsByHashPrefix mocks base method. +func (m *MockQuerier) SearchPacketsByHashPrefix(ctx context.Context, arg db.SearchPacketsByHashPrefixParams) ([]db.SearchPacketsByHashPrefixRow, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "SearchPacketsByHashPrefix", ctx, arg) + ret0, _ := ret[0].([]db.SearchPacketsByHashPrefixRow) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// SearchPacketsByHashPrefix indicates an expected call of SearchPacketsByHashPrefix. +func (mr *MockQuerierMockRecorder) SearchPacketsByHashPrefix(ctx, arg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SearchPacketsByHashPrefix", reflect.TypeOf((*MockQuerier)(nil).SearchPacketsByHashPrefix), ctx, arg) +} + // SetNodeDefaultScope mocks base method. func (m *MockQuerier) SetNodeDefaultScope(ctx context.Context, arg db.SetNodeDefaultScopeParams) error { m.ctrl.T.Helper() diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index 1135eaa..a2f324f 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -372,6 +372,11 @@ type Querier interface { // A route's hops resolve in its own IATA, so candidates from other IATAs can't match. // Both && predicates use the node_ids GIN index. SearchKnownRoutesByNodes(ctx context.Context, arg SearchKnownRoutesByNodesParams) ([]SearchKnownRoutesByNodesRow, error) + // Packets whose hash falls in [lo, hi). The materialized CTE pins the primary key + // range scan; otherwise generic plans walk idx_packets_last_heard across the table. + // IATA filtering is a membership probe, so ordering and the cursor stay on global + // last_heard_at. + SearchPacketsByHashPrefix(ctx context.Context, arg SearchPacketsByHashPrefixParams) ([]SearchPacketsByHashPrefixRow, error) SetNodeDefaultScope(ctx context.Context, arg SetNodeDefaultScopeParams) error SetNodeMultibytePaths(ctx context.Context, id uuid.UUID) error SetNodeMultibyteTraces(ctx context.Context, id uuid.UUID) error diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 7ea8cab..cd99b61 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -5376,6 +5376,155 @@ func (q *Queries) SearchKnownRoutesByNodes(ctx context.Context, arg SearchKnownR return items, nil } +const searchPacketsByHashPrefix = `-- name: SearchPacketsByHashPrefix :many +WITH hits AS MATERIALIZED ( + SELECT packet_hash FROM packets + WHERE packet_hash >= $9::bytea AND packet_hash < $10::bytea +) +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}' + WHEN p.payload_type = 3 AND p.parsed_payload ->> 'type' = 'ACK' + AND jsonb_typeof(p.parsed_payload -> 'checksum') = 'string' + AND p.parsed_payload ->> 'checksum' ~ '^[0-9a-fA-F]{8}$' + THEN 'ACK ' || lower(p.parsed_payload ->> 'checksum') + WHEN p.payload_type = 9 AND p.parsed_payload ->> 'type' IN ('TRACE', 'PING') + AND jsonb_typeof(p.parsed_payload -> 'traceTag') = 'string' + AND p.parsed_payload ->> 'traceTag' ~ '^[0-9a-fA-F]{8}$' + THEN (p.parsed_payload ->> 'type') || ' ' || lower(p.parsed_payload ->> 'traceTag') + END, '')::text AS summary, + p.route_type, + p.first_heard_at, + p.last_heard_at, + p.scope_id, + ts.name AS scope_name, + p.observation_count AS observation_count, + -- sqlc loses LATERAL nullability; these scalar defaults are ignored when observer_id is NULL. + po.observer_id AS latest_observer_id, + o.display_name AS latest_observer_name, + COALESCE(po.iata, ''::bpchar) AS latest_observer_iata, + 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, + p.raw_payload, + p.origin_pubkey +FROM hits h +JOIN packets p ON p.packet_hash = h.packet_hash +LEFT JOIN LATERAL ( + 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 + LIMIT 1 +) po ON true +LEFT JOIN observers o ON o.id = po.observer_id +LEFT JOIN transport_scopes ts ON ts.id = p.scope_id +WHERE + (COALESCE(cardinality($1::smallint[]), 0) = 0 OR p.payload_type = ANY($1::smallint[])) + AND (COALESCE(cardinality($2::smallint[]), 0) = 0 OR p.route_type = ANY($2::smallint[])) + AND ($3::timestamptz IS NULL OR p.first_heard_at >= $3) + AND ($4::timestamptz IS NULL OR p.first_heard_at <= $4) + AND ($5::timestamptz IS NULL OR p.last_heard_at < $5) + AND (COALESCE(cardinality($6::text[]), 0) = 0 OR ts.name = ANY($6::text[])) + AND (COALESCE(cardinality($7::bpchar[]), 0) = 0 OR EXISTS ( + SELECT 1 FROM packet_observations pi + WHERE pi.packet_hash = p.packet_hash AND pi.iata = ANY($7::bpchar[]))) +ORDER BY p.last_heard_at DESC +LIMIT $8 +` + +type SearchPacketsByHashPrefixParams struct { + PayloadTypes []int16 `json:"payload_types"` + RouteTypes []int16 `json:"route_types"` + SinceTs pgtype.Timestamptz `json:"since_ts"` + UntilTs pgtype.Timestamptz `json:"until_ts"` + CursorTs pgtype.Timestamptz `json:"cursor_ts"` + ScopeNames []string `json:"scope_names"` + Iatas []string `json:"iatas"` + PageLimit int32 `json:"page_limit"` + HashLo []byte `json:"hash_lo"` + HashHi []byte `json:"hash_hi"` +} + +type SearchPacketsByHashPrefixRow 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"` + 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"` + RawPayload []byte `json:"raw_payload"` + OriginPubkey []byte `json:"origin_pubkey"` +} + +// Packets whose hash falls in [lo, hi). The materialized CTE pins the primary key +// range scan; otherwise generic plans walk idx_packets_last_heard across the table. +// IATA filtering is a membership probe, so ordering and the cursor stay on global +// last_heard_at. +func (q *Queries) SearchPacketsByHashPrefix(ctx context.Context, arg SearchPacketsByHashPrefixParams) ([]SearchPacketsByHashPrefixRow, error) { + rows, err := q.db.Query(ctx, searchPacketsByHashPrefix, + arg.PayloadTypes, + arg.RouteTypes, + arg.SinceTs, + arg.UntilTs, + arg.CursorTs, + arg.ScopeNames, + arg.Iatas, + arg.PageLimit, + arg.HashLo, + arg.HashHi, + ) + if err != nil { + return nil, err + } + defer rows.Close() + items := []SearchPacketsByHashPrefixRow{} + for rows.Next() { + var i SearchPacketsByHashPrefixRow + if err := rows.Scan( + &i.PacketHash, + &i.PayloadType, + &i.Summary, + &i.RouteType, + &i.FirstHeardAt, + &i.LastHeardAt, + &i.ScopeID, + &i.ScopeName, + &i.ObservationCount, + &i.LatestObserverID, + &i.LatestObserverName, + &i.LatestObserverIata, + &i.LatestObserverPathLengthByte, + &i.LatestObserverHashSize, + &i.LatestObserverHopCount, + &i.LatestObserverPathBytes, + &i.RawPayload, + &i.OriginPubkey, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const setNodeDefaultScope = `-- name: SetNodeDefaultScope :exec UPDATE nodes SET default_scope_id = $2 WHERE id = $1 ` diff --git a/docs/docs.go b/docs/docs.go index 601b5db..c8347e8 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -1894,6 +1894,12 @@ const docTemplate = `{ "description": "Max results (default 50); must be positive, values above 200 are clamped", "name": "limit", "in": "query" + }, + { + "type": "string", + "description": "Match packets whose hash starts with this hex prefix (4-16 chars). Ordering and cursor use global last_heard_at even when iatas is set", + "name": "hashPrefix", + "in": "query" } ], "responses": { diff --git a/docs/swagger.json b/docs/swagger.json index 91de39b..be7e764 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -1892,6 +1892,12 @@ "description": "Max results (default 50); must be positive, values above 200 are clamped", "name": "limit", "in": "query" + }, + { + "type": "string", + "description": "Match packets whose hash starts with this hex prefix (4-16 chars). Ordering and cursor use global last_heard_at even when iatas is set", + "name": "hashPrefix", + "in": "query" } ], "responses": { diff --git a/docs/swagger.yaml b/docs/swagger.yaml index 48561a3..01d3f8c 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -3101,6 +3101,11 @@ paths: minimum: 1 name: limit type: integer + - description: Match packets whose hash starts with this hex prefix (4-16 chars). + Ordering and cursor use global last_heard_at even when iatas is set + in: query + name: hashPrefix + type: string produces: - application/json responses: diff --git a/internal/api/handlers/packets.go b/internal/api/handlers/packets.go index 82f1026..1882382 100644 --- a/internal/api/handlers/packets.go +++ b/internal/api/handlers/packets.go @@ -4,6 +4,7 @@ package handlers import ( + "bytes" "encoding/hex" "errors" "net/http" @@ -48,6 +49,7 @@ func PacketsRouter(reader api.Reader) http.Handler { // @Param until query int false "Filter by first_heard_at <= until (epoch ms)" // @Param cursor query int false "epoch ms of last item for pagination; last_heard_at, or site-local heard_at when iatas is set" // @Param limit query int false "Max results (default 50); must be positive, values above 200 are clamped" minimum(1) maximum(200) +// @Param hashPrefix query string false "Match packets whose hash starts with this hex prefix (4-16 chars). Ordering and cursor use global last_heard_at even when iatas is set" // @Success 200 {object} object // @Failure 400 {object} handlers.APIError // @Failure 500 {object} handlers.APIError @@ -110,7 +112,17 @@ func listPackets(reader api.Reader) http.HandlerFunc { iatas = append(iatas, regionIATAs...) } scopes := parseCSVOrSingle(r, "scopes", "scope") - packets, err := reader.ListPackets(r.Context(), payloadTypes, routeTypes, iatas, scopes, since, until, cursor, limit) + var packets api.Page[api.PacketSummary] + if p := r.URL.Query().Get("hashPrefix"); p != "" { + lo, hi, err := hashPrefixRange(p) + if err != nil { + respondError(w, http.StatusBadRequest, err.Error()) + return + } + packets, err = reader.SearchPacketsByHashPrefix(r.Context(), lo, hi, payloadTypes, routeTypes, iatas, scopes, since, until, cursor, limit) + } else { + packets, err = reader.ListPackets(r.Context(), payloadTypes, routeTypes, iatas, scopes, since, until, cursor, limit) + } if err != nil { respondError(w, http.StatusInternalServerError, "internal server error") return @@ -220,3 +232,30 @@ func getPacket(reader api.Reader) http.HandlerFunc { respond(w, http.StatusOK, packet) } } + +// hashPrefixRange turns a hex prefix into the byte range [lo, hi) of hashes that start +// with it. An odd length pins the high nibble only; hi is nil when the prefix is all f. +func hashPrefixRange(prefix string) (lo, hi []byte, err error) { + if len(prefix) < 4 || len(prefix) > 16 { + return nil, nil, errors.New("hashPrefix must be 4-16 hex characters") + } + step := 1 + if len(prefix)%2 == 1 { + prefix += "0" + step = 0x10 + } + lo, err = hex.DecodeString(prefix) + if err != nil { + return nil, nil, errors.New("hashPrefix must be hex") + } + hi = bytes.Clone(lo) + for i := len(hi) - 1; i >= 0; i-- { + v := int(hi[i]) + step + hi[i] = byte(v) + if v <= 0xff { + return lo, hi, nil + } + step = 1 + } + return lo, nil, nil +} diff --git a/internal/api/handlers/packets_test.go b/internal/api/handlers/packets_test.go index 2a34328..eed4ca7 100644 --- a/internal/api/handlers/packets_test.go +++ b/internal/api/handlers/packets_test.go @@ -5,6 +5,7 @@ package handlers import ( "context" + "encoding/hex" "net/http" "net/http/httptest" "testing" @@ -270,3 +271,77 @@ func TestListPacketsBackfill_OK(t *testing.T) { t.Errorf("expected 200, got %d", w.Code) } } + +func TestHashPrefixRange(t *testing.T) { + tests := []struct { + in string + lo, hi string + }{ + {"6ddf62fb", "6ddf62fb", "6ddf62fc"}, + {"6DDF62FB", "6ddf62fb", "6ddf62fc"}, + {"6ddf6", "6ddf60", "6ddf70"}, + {"12ff", "12ff", "1300"}, + {"0012f", "0012f0", "001300"}, + {"ffff", "ffff", ""}, + {"fffff", "fffff0", ""}, + {"6ddf62fb5b9a969e", "6ddf62fb5b9a969e", "6ddf62fb5b9a969f"}, + } + for _, tt := range tests { + lo, hi, err := hashPrefixRange(tt.in) + if err != nil { + t.Fatalf("%s: %v", tt.in, err) + } + if hex.EncodeToString(lo) != tt.lo || hex.EncodeToString(hi) != tt.hi { + t.Errorf("%s: got [%x, %x), want [%s, %s)", tt.in, lo, hi, tt.lo, tt.hi) + } + if tt.hi == "" && hi != nil { + t.Errorf("%s: want nil hi", tt.in) + } + } + for _, bad := range []string{"", "abc", "6ddg", "6ddf62fb5b9a969e0"} { + if _, _, err := hashPrefixRange(bad); err == nil { + t.Errorf("%q: expected error", bad) + } + } +} + +func TestListPackets_HashPrefix(t *testing.T) { + var gotLo, gotHi []byte + var gotIATAs []string + r := chi.NewRouter() + r.Get("/packets", listPackets(stubReader{ + listPackets: func(_ context.Context, _, _ []int16, _ []string, _ []string, _, _ time.Time, _ int64, _ int32) (api.Page[api.PacketSummary], error) { + t.Error("ListPackets called for a hashPrefix search") + return api.Page[api.PacketSummary]{}, nil + }, + searchPacketsByHashPrefix: func(_ context.Context, lo, hi []byte, _, _ []int16, iatas []string, _ []string, _, _ time.Time, _ int64, _ int32) (api.Page[api.PacketSummary], error) { + gotLo, gotHi, gotIATAs = lo, hi, iatas + return api.Page[api.PacketSummary]{Items: []api.PacketSummary{{PacketHash: "6ddf62fb5b9a969e"}}}, nil + }, + })) + req := httptest.NewRequest(http.MethodGet, "/packets?hashPrefix=6DDF62FB&iatas=yow", nil) + w := httptest.NewRecorder() + r.ServeHTTP(w, req) + if w.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", w.Code) + } + if hex.EncodeToString(gotLo) != "6ddf62fb" || hex.EncodeToString(gotHi) != "6ddf62fc" { + t.Errorf("range = [%x, %x)", gotLo, gotHi) + } + if len(gotIATAs) != 1 || gotIATAs[0] != "YOW" { + t.Errorf("iatas = %v", gotIATAs) + } +} + +func TestListPackets_HashPrefixInvalid(t *testing.T) { + r := chi.NewRouter() + r.Get("/packets", listPackets(stubReader{})) + for _, q := range []string{"zz12", "abc", "00112233445566778"} { + req := httptest.NewRequest(http.MethodGet, "/packets?hashPrefix="+q, nil) + w := httptest.NewRecorder() + r.ServeHTTP(w, req) + if w.Code != http.StatusBadRequest { + t.Errorf("%s: expected 400, got %d", q, w.Code) + } + } +} diff --git a/internal/api/handlers/stub_reader_test.go b/internal/api/handlers/stub_reader_test.go index 71d7fac..d7a6481 100644 --- a/internal/api/handlers/stub_reader_test.go +++ b/internal/api/handlers/stub_reader_test.go @@ -42,6 +42,7 @@ type stubReader struct { getNode func(ctx context.Context, nodeID uuid.UUID) (*api.Node, error) getNodeNeighbors func(ctx context.Context, nodeID uuid.UUID) ([]api.NodeNeighbor, error) listNodeObservations func(ctx context.Context, nodeID uuid.UUID, cursor int64, limit int32) (api.Page[api.PacketObservationSummary], error) + searchPacketsByHashPrefix func(ctx context.Context, lo, hi []byte, payloadTypes, routeTypes []int16, iatas []string, scopes []string, since, until time.Time, cursor int64, limit int32) (api.Page[api.PacketSummary], error) listPackets func(ctx context.Context, payloadTypes, routeTypes []int16, iatas []string, scopes []string, since, until time.Time, cursor int64, limit int32) (api.Page[api.PacketSummary], error) listPacketsAfterID func(ctx context.Context, afterObservationID int64, payloadType, routeType int16, iatas []string, scope string, limit int32) ([]api.PacketSummary, error) getPacket func(ctx context.Context, packetHash []byte) (*api.Packet, error) @@ -249,6 +250,13 @@ func (s stubReader) ListPackets(ctx context.Context, payloadTypes, routeTypes [] return api.Page[api.PacketSummary]{}, nil } +func (s stubReader) SearchPacketsByHashPrefix(ctx context.Context, lo, hi []byte, payloadTypes, routeTypes []int16, iatas []string, scopes []string, since, until time.Time, cursor int64, limit int32) (api.Page[api.PacketSummary], error) { + if s.searchPacketsByHashPrefix != nil { + return s.searchPacketsByHashPrefix(ctx, lo, hi, payloadTypes, routeTypes, iatas, scopes, since, until, cursor, limit) + } + return api.Page[api.PacketSummary]{}, nil +} + func (s stubReader) ListPacketsAfterID(ctx context.Context, afterObservationID int64, payloadType, routeType int16, iatas []string, scope string, limit int32) ([]api.PacketSummary, error) { if s.listPacketsAfterID != nil { return s.listPacketsAfterID(ctx, afterObservationID, payloadType, routeType, iatas, scope, limit) diff --git a/internal/api/reader.go b/internal/api/reader.go index 672dfe2..150b52b 100644 --- a/internal/api/reader.go +++ b/internal/api/reader.go @@ -140,6 +140,11 @@ type Reader interface { // nil/empty to skip a filter. ListPackets(ctx context.Context, payloadTypes, routeTypes []int16, iatas []string, scopes []string, since, until time.Time, cursor int64, limit int32) (Page[PacketSummary], error) + // SearchPacketsByHashPrefix lists packets whose hash is in [lo, hi) (nil hi is unbounded), + // newest last_heard_at first. Other filters behave as in ListPackets; iatas only + // restricts membership, the cursor stays on global last_heard_at. + SearchPacketsByHashPrefix(ctx context.Context, lo, hi []byte, payloadTypes, routeTypes []int16, iatas []string, scopes []string, since, until time.Time, cursor int64, limit int32) (Page[PacketSummary], error) + // ListPacketsAfterID returns packets with observations after the given observation ID, // ordered oldest first. Used for WS reconnect backfill. ListPacketsAfterID(ctx context.Context, afterObservationID int64, payloadType, routeType int16, iatas []string, scope string, limit int32) ([]PacketSummary, error) diff --git a/internal/cache/cache_test.go b/internal/cache/cache_test.go index 5786c12..1e59f13 100644 --- a/internal/cache/cache_test.go +++ b/internal/cache/cache_test.go @@ -192,6 +192,10 @@ func (s *stubReader) ListObserverAdverts(_ context.Context, _ uuid.UUID, _ int64 return api.Page[api.AdvertObservation]{}, nil } +func (s *stubReader) SearchPacketsByHashPrefix(_ context.Context, _, _ []byte, _, _ []int16, _ []string, _ []string, _, _ time.Time, _ int64, _ int32) (api.Page[api.PacketSummary], error) { + return api.Page[api.PacketSummary]{}, nil +} + func (s *stubReader) ListPackets(_ context.Context, _, _ []int16, _ []string, _ []string, _, _ time.Time, _ int64, _ int32) (api.Page[api.PacketSummary], error) { return api.Page[api.PacketSummary]{}, nil } diff --git a/internal/cache/reader.go b/internal/cache/reader.go index d573ab9..5e4df4c 100644 --- a/internal/cache/reader.go +++ b/internal/cache/reader.go @@ -339,6 +339,11 @@ func (cr *CachedReader) ListObserverAdverts(ctx context.Context, observerID uuid return cr.inner.ListObserverAdverts(ctx, observerID, cursor, limit) } +// SearchPacketsByHashPrefix implements [api.Reader]. +func (cr *CachedReader) SearchPacketsByHashPrefix(ctx context.Context, lo, hi []byte, payloadTypes, routeTypes []int16, iatas []string, scopes []string, since, until time.Time, cursor int64, limit int32) (api.Page[api.PacketSummary], error) { + return cr.inner.SearchPacketsByHashPrefix(ctx, lo, hi, payloadTypes, routeTypes, iatas, scopes, since, until, cursor, limit) +} + // ListPackets implements [api.Reader]. func (cr *CachedReader) ListPackets(ctx context.Context, payloadTypes, routeTypes []int16, iatas []string, scopes []string, since, until time.Time, cursor int64, limit int32) (api.Page[api.PacketSummary], error) { return cr.inner.ListPackets(ctx, payloadTypes, routeTypes, iatas, scopes, since, until, cursor, limit) From b43da0c3a90bb343cb3b5afd4146a2e7fdb243e7 Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Wed, 14 Oct 2026 17:10:50 -0400 Subject: [PATCH 2/2] chore: bump version to 2.0.7 --- cmd/beacon/main.go | 2 +- docs/docs.go | 2 +- docs/swagger.json | 2 +- docs/swagger.yaml | 2 +- 4 files changed, 4 insertions(+), 4 deletions(-) diff --git a/cmd/beacon/main.go b/cmd/beacon/main.go index cc1de70..6c909a4 100644 --- a/cmd/beacon/main.go +++ b/cmd/beacon/main.go @@ -45,7 +45,7 @@ import ( var version = "dev" // @title MeshCore Beacon API -// @version 2.0.6 +// @version 2.0.7 // @description MeshCore network observation backend. Ingests LoRa packets from MQTT brokers, stores in PostgreSQL, and streams live events via WebSocket. // @description REST requests share a configurable per-client rate limit (default 300/minute and a 300-request one-second burst cap). Exceeded limits return HTTP 429 with error.code=rate_limited and a Retry-After header in seconds. CORS preflights and WebSocket upgrades do not consume this API budget. // @description WebSocket upgrade attempts at /ws have a configurable per-client limit (default 10/minute, including failed handshakes). Rate exhaustion returns HTTP 429 with Retry-After before upgrade. An accepted socket exceeding the concurrent cap closes with code 1013 before hello; established connections remain open. diff --git a/docs/docs.go b/docs/docs.go index c8347e8..b4a5a5f 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -6271,7 +6271,7 @@ const docTemplate = `{ // SwaggerInfo holds exported Swagger Info so clients can modify it var SwaggerInfo = &swag.Spec{ - Version: "2.0.6", + Version: "2.0.7", Host: "localhost:8080", BasePath: "/api/v1", Schemes: []string{"http", "https"}, diff --git a/docs/swagger.json b/docs/swagger.json index be7e764..d557e25 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -15,7 +15,7 @@ "license": { "name": "AGPL-3-or-later" }, - "version": "2.0.6" + "version": "2.0.7" }, "host": "localhost:8080", "basePath": "/api/v1", diff --git a/docs/swagger.yaml b/docs/swagger.yaml index 01d3f8c..93dd713 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -1825,7 +1825,7 @@ info: name: AGPL-3-or-later termsOfService: https://github.com/MeshCore-Beacon/beacon-server title: MeshCore Beacon API - version: 2.0.6 + version: 2.0.7 paths: /admin/accounts: get: