feat(packets): show hop ambiguity, resolved per observer (port of upstream 2099)

.github/workflows/deploy.yml

@@ -441,6 +441,7 @@ jobs:
           BASE_URL=http://localhost:13581 node test-slideover-1056-e2e.js 2>&1 | tee -a e2e-output.txt
           BASE_URL=http://localhost:13581 node test-issue-1692-packets-init-parallel-e2e.js 2>&1 | tee -a e2e-output.txt
           BASE_URL=http://localhost:13581 node test-issue-147-packets-url-obs-e2e.js 2>&1 | tee -a e2e-output.txt
+          BASE_URL=http://localhost:13581 node test-issue-165-grouped-hop-warn-e2e.js 2>&1 | tee -a e2e-output.txt
           BASE_URL=http://localhost:13581 node test-packet-detail-sender-hash-size-obs-e2e.js 2>&1 | tee -a e2e-output.txt
           BASE_URL=http://localhost:13581 node test-issue-96-hide-control-e2e.js 2>&1 | tee -a e2e-output.txt
           BASE_URL=http://localhost:13581 node test-issue-180-packets-url-modal-e2e.js 2>&1 | tee -a e2e-output.txt

cmd/server/db.go

@@ -792,14 +792,20 @@ func (db *DB) QueryGroupedPackets(q PacketQuery) (*PacketResult, error) {
 	if db.hasScopeName() {
 		groupedScopeCol = ", t.scope_name"
 	}
+	// #165 — the displayed observation's resolved_path. Always one column
+	// (NULL without the column), so the Scan below does not depend on it.
+	groupedRPCol := "NULL"
+	if db.hasResolvedPath() {
+		groupedRPCol = "o.resolved_path"
+	}
 	var querySQL string
 	if db.isV3() {
 		querySQL = fmt.Sprintf(`SELECT t.hash, t.first_seen, t.raw_hex, t.decoded_json, t.payload_type, t.route_type,
 			COALESCE((SELECT COUNT(*) FROM observations oi WHERE oi.transmission_id = t.id), 0) AS count,
 			COALESCE((SELECT COUNT(DISTINCT oi.observer_idx) FROM observations oi WHERE oi.transmission_id = t.id), 0) AS observer_count,
 			COALESCE((SELECT MAX(strftime('%%Y-%%m-%%dT%%H:%%M:%%fZ', oi.timestamp, 'unixepoch')) FROM observations oi WHERE oi.transmission_id = t.id), t.first_seen) AS latest,
 			obs.id AS observer_id, obs.name AS observer_name, COALESCE(obs.iata, '') AS observer_iata,
-			o.snr, o.rssi, o.path_json,
+			o.snr, o.rssi, o.path_json, `+groupedRPCol+`,
 			COALESCE((SELECT GROUP_CONCAT(DISTINCT obi.iata) FROM observations oi JOIN observers obi ON obi.rowid = oi.observer_idx WHERE oi.transmission_id = t.id AND obi.iata IS NOT NULL AND obi.iata != ''), '') AS distinct_iatas`+groupedScopeCol+`
 		FROM transmissions t
 		LEFT JOIN observations o ON o.id = (
@@ -814,7 +820,7 @@ func (db *DB) QueryGroupedPackets(q PacketQuery) (*PacketResult, error) {
 			COALESCE((SELECT COUNT(DISTINCT oi.observer_id) FROM observations oi WHERE oi.transmission_id = t.id), 0) AS observer_count,
 			COALESCE((SELECT MAX(oi.timestamp) FROM observations oi WHERE oi.transmission_id = t.id), t.first_seen) AS latest,
 			o.observer_id, o.observer_name, COALESCE(obs2.iata, '') AS observer_iata,
-			o.snr, o.rssi, o.path_json,
+			o.snr, o.rssi, o.path_json, `+groupedRPCol+`,
 			COALESCE((SELECT GROUP_CONCAT(DISTINCT obi.iata) FROM observations oi JOIN observers obi ON obi.id = oi.observer_id WHERE oi.transmission_id = t.id AND obi.iata IS NOT NULL AND obi.iata != ''), '') AS distinct_iatas`+groupedScopeCol+`
 		FROM transmissions t
 		LEFT JOIN observations o ON o.id = (
@@ -837,23 +843,23 @@ func (db *DB) QueryGroupedPackets(q PacketQuery) (*PacketResult, error) {
 
 	packets := make([]map[string]interface{}, 0)
 	for rows.Next() {
-		var hash, firstSeen, rawHex, decodedJSON, latest, observerID, observerName, observerIATA, pathJSON, distinctIatasCSV sql.NullString
+		var hash, firstSeen, rawHex, decodedJSON, latest, observerID, observerName, observerIATA, pathJSON, resolvedPath, distinctIatasCSV sql.NullString
 		var payloadType, routeType sql.NullInt64
 		var count, observerCount int
 		var snr, rssi sql.NullFloat64
 		var scopeName sql.NullString
 
 		scanArgs := []interface{}{&hash, &firstSeen, &rawHex, &decodedJSON, &payloadType, &routeType,
 			&count, &observerCount, &latest,
-			&observerID, &observerName, &observerIATA, &snr, &rssi, &pathJSON, &distinctIatasCSV}
+			&observerID, &observerName, &observerIATA, &snr, &rssi, &pathJSON, &resolvedPath, &distinctIatasCSV}
 		if db.hasScopeName() {
 			scanArgs = append(scanArgs, &scopeName)
 		}
 		if err := rows.Scan(scanArgs...); err != nil {
 			continue
 		}
 
-		packets = append(packets, map[string]interface{}{
+		row := map[string]interface{}{
 			"hash":              nullStr(hash),
 			"first_seen":        nullStr(firstSeen),
 			"count":             count,
@@ -872,7 +878,15 @@ func (db *DB) QueryGroupedPackets(q PacketQuery) (*PacketResult, error) {
 			"snr":               nullFloat(snr),
 			"rssi":              nullFloat(rssi),
 			"scope_name":        nullStr(scopeName),
-		})
+		}
+		// #165 — the displayed observation's resolved_path, as the in-memory
+		// store sends it (groupedPageWithRP).
+		if resolvedPath.Valid {
+			if rp := resolvedPathRaw(resolvedPath.String); rp != nil {
+				row["resolved_path"] = rp
+			}
+		}
+		packets = append(packets, row)
 	}
 
 	return &PacketResult{Packets: packets, Total: total}, nil

cmd/server/grouped_resolved_path_165_bench_test.go

@@ -0,0 +1,74 @@
+package main
+
+import (
+	"bytes"
+	"compress/gzip"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"testing"
+	"time"
+)
+
+// BenchmarkGroupedPackets165 measures the default packets-page request
+// (/api/packets?groupByHash=true, desktop limit 50000) on 30K transmissions
+// with two observations each and a 3-hop resolved_path, the shape #165
+// asked to be checked before the grouped rows carry resolved_path. It
+// reports the JSON size of the page as json-bytes, and gzipped (the server
+// compresses responses, compress.go) as gzip-bytes.
+//
+//	go test -run '^$' -bench GroupedPackets165 -benchtime 20x .
+func BenchmarkGroupedPackets165(b *testing.B) {
+	const nTx = 30000
+	db := setupTestDB(b)
+	defer db.Close()
+	now := time.Now().UTC()
+	ts := now.Format(time.RFC3339)
+	epoch := now.Add(-time.Minute).Unix()
+	tx, err := db.conn.Begin()
+	if err != nil {
+		b.Fatal(err)
+	}
+	for i := 1; i <= 2; i++ {
+		tx.Exec(`INSERT INTO observers (id, name, iata, last_seen, first_seen, packet_count) VALUES (?, ?, 'SJC', ?, ?, 1)`,
+			fmt.Sprintf("OBS%02d", i), fmt.Sprintf("O%d", i), ts, ts)
+	}
+	// Pubkeys look random, as real ones do, so gzip-bytes is not flattered.
+	pk := func(i, h int) string {
+		sum := sha256.Sum256([]byte(fmt.Sprint(i, ":", h)))
+		return fmt.Sprintf("%02x", (i+h)%256) + hex.EncodeToString(sum[:])[2:]
+	}
+	for i := 1; i <= nTx; i++ {
+		tx.Exec(`INSERT INTO transmissions (id, raw_hex, hash, first_seen, route_type, payload_type, decoded_json) VALUES (?, ?, ?, ?, 1, 4, '{}')`,
+			i, "11223344556677889900aabbccddeeff", fmt.Sprintf("%016x", i), ts)
+		path := fmt.Sprintf(`["%02x","%02x","%02x"]`, (i+0)%256, (i+1)%256, (i+2)%256)
+		rp := fmt.Sprintf(`["%s","%s",null]`, pk(i, 0), pk(i, 1))
+		tx.Exec(`INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp, resolved_path) VALUES (?, 1, 5, -90, ?, ?, ?)`, i, path, epoch-int64(i), rp)
+		tx.Exec(`INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp, resolved_path) VALUES (?, 2, 5, -90, ?, ?, ?)`,
+			i, fmt.Sprintf(`["%02x"]`, (i+2)%256), epoch-int64(i), fmt.Sprintf(`["%s"]`, pk(i, 2)))
+	}
+	if err := tx.Commit(); err != nil {
+		b.Fatal(err)
+	}
+	store := NewPacketStore(db, nil)
+	if err := store.Load(); err != nil {
+		b.Fatal(err)
+	}
+	q := PacketQuery{Limit: 50000}
+	r := store.QueryGroupedPackets(q)
+	if r.Total != nTx {
+		b.Fatalf("total = %d, want %d", r.Total, nTx)
+	}
+	body, _ := json.Marshal(r)
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		store.QueryGroupedPackets(q)
+	}
+	var gz bytes.Buffer
+	zw := gzip.NewWriter(&gz)
+	zw.Write(body)
+	zw.Close()
+	b.ReportMetric(float64(len(body)), "json-bytes")
+	b.ReportMetric(float64(gz.Len()), "gzip-bytes")
+}

cmd/server/grouped_resolved_path_165_test.go

@@ -0,0 +1,102 @@
+package main
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+	"time"
+)
+
+// #165 — the packets page's default (grouped) view resolves hop names from
+// the grouped row. Without the server's resolved_path on that row, the client
+// heuristic guessed every hop and the list flagged hops the server had a
+// definite answer for (23 of 60 rows on the CI fixture). The grouped row must
+// carry the resolved_path of the observation it displays: the same observer
+// and path_json as the row's observer_id/path_json, since the client caches
+// it under hop:observer.
+
+const (
+	rp165A = "aa11111111111111111111111111111111111111111111111111111111111111"
+	rp165B = "bb22222222222222222222222222222222222222222222222222222222222222"
+	rp165X = "aa99999999999999999999999999999999999999999999999999999999999999"
+)
+
+// seedGroupedRP165 inserts two transmissions:
+//   - "165a…": observer A hears path [aa,bb] with resolved_path [A, null]
+//     (the header: longest path); observer B hears [aa] resolved to X.
+//   - "165b…": one observation with no resolved_path at all.
+func seedGroupedRP165(t *testing.T, db *DB) {
+	t.Helper()
+	now := time.Now().UTC()
+	ts := now.Format(time.RFC3339)
+	epoch := now.Add(-time.Minute).Unix()
+	mustExec := func(q string, args ...interface{}) {
+		t.Helper()
+		if _, err := db.conn.Exec(q, args...); err != nil {
+			t.Fatalf("seed: %v\n%s", err, q)
+		}
+	}
+	mustExec(`INSERT INTO observers (id, name, iata, last_seen, first_seen, packet_count) VALUES ('OBSA', 'A', 'SJC', ?, ?, 1)`, ts, ts)
+	mustExec(`INSERT INTO observers (id, name, iata, last_seen, first_seen, packet_count) VALUES ('OBSB', 'B', 'SFO', ?, ?, 1)`, ts, ts)
+	mustExec(`INSERT INTO transmissions (id, raw_hex, hash, first_seen, route_type, payload_type, decoded_json) VALUES (1, 'AABB', '165a000000000001', ?, 1, 4, '{}')`, ts)
+	mustExec(`INSERT INTO transmissions (id, raw_hex, hash, first_seen, route_type, payload_type, decoded_json) VALUES (2, 'CCDD', '165b000000000002', ?, 1, 4, '{}')`, ts)
+	mustExec(`INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp, resolved_path) VALUES (1, 2, 5, -90, '["aa"]', ?, ?)`, epoch, `["`+rp165X+`"]`)
+	mustExec(`INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp, resolved_path) VALUES (1, 1, 6, -88, '["aa","bb"]', ?, ?)`, epoch, `["`+rp165A+`",null]`)
+	mustExec(`INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp) VALUES (2, 1, 7, -80, '["cc"]', ?)`, epoch)
+}
+
+func groupedRowsByHash165(t *testing.T, r *PacketResult) map[string]map[string]interface{} {
+	t.Helper()
+	out := map[string]map[string]interface{}{}
+	for _, p := range r.Packets {
+		h, _ := p["hash"].(string)
+		out[h] = p
+	}
+	if len(out) != 2 {
+		t.Fatalf("expected 2 grouped rows, got %d: %#v", len(out), r.Packets)
+	}
+	return out
+}
+
+func assertGroupedRP165(t *testing.T, rows map[string]map[string]interface{}) {
+	t.Helper()
+	a := rows["165a000000000001"]
+	if got := strings.ToUpper(a["observer_id"].(string)); got != "OBSA" {
+		t.Fatalf("header observer = %q, want OBSA (longest path)", got)
+	}
+	rp, ok := a["resolved_path"]
+	if !ok {
+		t.Fatalf("grouped row carries no resolved_path; the client then guesses hops the server resolved (#165): %#v", a)
+	}
+	b, _ := json.Marshal(rp)
+	if want := `["` + rp165A + `",null]`; string(b) != want {
+		t.Fatalf("grouped resolved_path = %s, want the header observation's %s", b, want)
+	}
+	if _, ok := rows["165b000000000002"]["resolved_path"]; ok {
+		t.Fatalf("a row whose observation has no resolved_path must omit the key, got %#v", rows["165b000000000002"]["resolved_path"])
+	}
+}
+
+func TestGroupedPacketsCarryHeaderResolvedPath165(t *testing.T) {
+	db := setupTestDB(t)
+	defer db.Close()
+	seedGroupedRP165(t, db)
+	store := NewPacketStore(db, nil)
+	if err := store.Load(); err != nil {
+		t.Fatal(err)
+	}
+	assertGroupedRP165(t, groupedRowsByHash165(t, store.QueryGroupedPackets(PacketQuery{Limit: 50})))
+	// Second call is served from the grouped sort cache; same contract.
+	assertGroupedRP165(t, groupedRowsByHash165(t, store.QueryGroupedPackets(PacketQuery{Limit: 50})))
+}
+
+func TestGroupedPacketsSQLCarryHeaderResolvedPath165(t *testing.T) {
+	db := setupTestDB(t)
+	defer db.Close()
+	seedGroupedRP165(t, db)
+	r, err := db.QueryGroupedPackets(PacketQuery{Limit: 50})
+	if err != nil {
+		t.Fatal(err)
+	}
+	assertGroupedRP165(t, groupedRowsByHash165(t, r))
+}

cmd/server/resolved_index.go

@@ -11,8 +11,10 @@ package main
 
 import (
 	"database/sql"
+	"encoding/json"
 	"hash/fnv"
 	"log"
+	"strconv"
 	"strings"
 	"time"
 )
@@ -198,6 +200,64 @@ func (s *PacketStore) fetchResolvedPathsForTx(txID int) map[int][]*string {
 	return result
 }
 
+// fetchResolvedPathsForObsIDs reads resolved_path for many observations in
+// one query (#165, grouped packet pages of up to 50K rows): the ids go in as
+// one JSON array parameter, so there is no bound-variable limit and one
+// statement to prepare. Values come back as the stored JSON, unparsed
+// (resolvedPathRaw), because the caller only re-encodes them. Id 0 is
+// skipped. It bypasses the LRU on purpose: a page is far larger than
+// lruMaxSize and would evict every entry other endpoints rely on.
+func (s *PacketStore) fetchResolvedPathsForObsIDs(ids []int) map[int]json.RawMessage {
+	out := make(map[int]json.RawMessage)
+	if s.db == nil || s.db.conn == nil || !s.db.hasResolvedPath() || len(ids) == 0 {
+		return out
+	}
+	var b strings.Builder
+	b.Grow(len(ids) * 8)
+	b.WriteByte('[')
+	n := 0
+	for _, id := range ids {
+		if id == 0 {
+			continue
+		}
+		if n > 0 {
+			b.WriteByte(',')
+		}
+		b.WriteString(strconv.Itoa(id))
+		n++
+	}
+	b.WriteByte(']')
+	if n == 0 {
+		return out
+	}
+	rows, err := s.db.conn.Query(`SELECT o.id, o.resolved_path FROM observations o
+		WHERE o.id IN (SELECT value FROM json_each(?)) AND o.resolved_path IS NOT NULL`, b.String())
+	if err != nil {
+		return out
+	}
+	defer rows.Close()
+	for rows.Next() {
+		var id int
+		var rpJSON sql.NullString
+		if rows.Scan(&id, &rpJSON) == nil && rpJSON.Valid {
+			if rp := resolvedPathRaw(rpJSON.String); rp != nil {
+				out[id] = rp
+			}
+		}
+	}
+	return out
+}
+
+// resolvedPathRaw returns a stored resolved_path as JSON to embed in a
+// response, or nil when it is empty or not a JSON array (an invalid
+// RawMessage would fail the whole response's encoding).
+func resolvedPathRaw(s string) json.RawMessage {
+	if len(s) < 2 || s[0] != '[' || !json.Valid([]byte(s)) {
+		return nil
+	}
+	return json.RawMessage(s)
+}
+
 // fetchResolvedPathForObs fetches resolved_path for a single observation,
 // using the LRU cache.
 func (s *PacketStore) fetchResolvedPathForObs(obsID int) []*string {