Merge upstream packet hash search into the dev1 topology backend

.github/workflows/ci.yml

@@ -58,7 +58,7 @@ jobs:
       - name: Verify stats and endpoint queries against PostgreSQL 16
         env:
           BEACON_TEST_POSTGRES_DSN: postgres://postgres:backup-ci-only@127.0.0.1:5432/postgres?sslmode=disable
-        run: go test ./db -run '^Test(Signal|Paths|PacketSummaries|PacketEndpointsResolveLive|ObserverMetrics|RouteEvidence|RouteEvidenceIndex|AnalyticsRetention|AnalyticsRetentionConcurrent|DeleteOldPacketsBatches|MeshMapperCatalogue|ChannelMessageScopes|NodeLocationReset|TraceQuality|TopologyLinks|RunMigrations[A-Za-z]*|Reconfirm[A-Za-z]*)Postgres$' -count=1 -v
+        run: go test ./db -run '^Test(SearchPacketsByHashPrefix|MeshMapperRepeaters|TopAdvertisersSort|Signal|Paths|PacketSummaries|PacketEndpointsResolveLive|ObserverMetrics|RouteEvidence|RouteEvidenceIndex|AnalyticsRetention|AnalyticsRetentionConcurrent|DeleteOldPacketsBatches|MeshMapperCatalogue|ChannelMessageScopes|NodeLocationReset|TraceQuality|TopologyLinks|RunMigrations[A-Za-z]*|Reconfirm[A-Za-z]*)Postgres$' -count=1 -v
 
       - name: Verify backup command against PostgreSQL 16
         run: |

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

db/packets.go

@@ -4,6 +4,7 @@
 package db
 
 import (
+	"bytes"
 	"context"
 	"encoding/binary"
 	"encoding/hex"
@@ -254,6 +255,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,

db/queries/queries.sql

@@ -635,6 +635,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.

db/sqlc/mock/querier.go

@@ -2355,6 +2355,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()

db/sqlc/querier.go

@@ -377,6 +377,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

db/sqlc/queries.sql.go

@@ -5506,6 +5506,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
 `

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": {

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": {

docs/swagger.yaml

@@ -3173,6 +3173,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:

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

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

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)

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)

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
 }

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)