diff --git a/.squad/agents/hicks/history.md b/.squad/agents/hicks/history.md index 2ff4d61f..42d0e7bd 100644 --- a/.squad/agents/hicks/history.md +++ b/.squad/agents/hicks/history.md @@ -12,3 +12,8 @@ User: User - Split the monolithic "Frontend coverage (instrumented Playwright)" CI step into 5 discrete steps: Instrument frontend JS, Start test server (with health-check poll replacing sleep 5), Run Playwright E2E tests, Extract coverage + generate report, Stop test server. Cleanup/report steps use `if: always()` so server shutdown happens even on test failure. Server PID shared across steps via .server.pid file. "Frontend E2E only" fast-path left untouched. - Fixed memory explosion in packet-store.js: observations no longer duplicate transmission fields (hash, raw_hex, decoded_json, payload_type, route_type). Instead, observations store only `transmission_id` as a reference. Added `_enrichObs()` to hydrate observations at API boundaries (getById, getSiblings, enrichObservations). Replaced `.all()` with `.iterate()` for streaming load. Updated `_transmissionsForObserver()` to use transmission_id instead of hash. For a 185MB DB with 50K transmissions × 23 observations avg, this eliminates ~1.17M copies of hex dumps and JSON — projected ~2GB RAM savings. - Built standalone Go MQTT ingestor (`cmd/ingestor/`). Ported decoder.js → Go (header parsing, path extraction, all payload types, advert decoding with flags/lat/lon/name). Ported db.js v3 schema (transmissions + observations + nodes + observers). Ported computeContentHash (SHA-256 based, path-independent). Uses modernc.org/sqlite (pure Go, no CGO) and paho.mqtt.golang. 25 tests passing (decoder golden fixtures from production data + DB schema compatibility). Supports same config.json format as Node.js server. Handles Format 1 (raw packet) messages; companion bridge format deferred. System Go was 1.17 — installed Go 1.22.5 to support modern dependencies. +- Built standalone Go web server (`cmd/server/`) — READ side of the Go rewrite. 35+ REST API endpoints ported from server.js. All queries go directly to SQLite (no in-memory packet store). WebSocket broadcast via SQLite polling. Static file server with SPA fallback. Uses gorilla/mux for routing, gorilla/websocket for WS, modernc.org/sqlite for DB. 42 tests passing (20 DB query tests, 20+ route integration tests, 2 WebSocket tests). `go vet` clean. Binary compiles to single executable. Analytics endpoints that required Node.js in-memory store (topology, distance, hash-sizes, subpaths) return structural stubs — core data (RF stats, channels, node health, etc.) fully functional via SQL. System Go 1.17 → installed Go 1.22 for build. Each cmd/* module has its own go.mod (no root-level go.mod). +- Go server API parity fix: Rewrote QueryPackets from observation-centric (packets_v view) to transmission-centric (transmissions table + correlated subqueries). This fixes both performance (9s to sub-100ms for unfiltered queries on 1.2M rows) and response shape. Packets now return first_seen, timestamp (= first_seen), observation_count, and NOT created_at/payload_version/score. Node responses now include last_heard (= last_seen fallback), hash_size (null), hash_size_inconsistent (false). Added schema version detection (v2 vs v3 observations table). Fixed QueryGroupedPackets first_seen. Added GetRecentTransmissionsForNode. All tests pass, build clean with Go 1.22. +- Fixed #133 (node count keeps climbing): `db.getStats().totalNodes` used `SELECT COUNT(*) FROM nodes` which counts every node ever seen — 6800+ on a ~200-400 node mesh. Changed `totalNodes` to count only nodes with `last_seen` within 7 days. Added `totalNodesAllTime` for the full historical count. Also filtered role counts in `/api/stats` to the same 7-day window. Added `countActiveNodes` and `countActiveNodesByRole` prepared statements in db.js. 6 new tests (95 total in test-db.js). The existing `idx_nodes_last_seen` index covers the new queries. +- Go server FULL API parity: Rewrote QueryGroupedPackets from packets_v VIEW scan (8s on 1.2M rows) to transmission-centric query (<100ms). Fixed GetStats to use 7-day window for totalNodes + added totalNodesAllTime. Split GetRoleCounts into 7-day (for /api/stats) and all-time (for /api/nodes). Added packetsLastHour + node lat/lon/role to /api/observers via batch queries (GetObserverPacketCounts, GetNodeLocations). Added multi-node filter support (/api/packets?nodes=pk1,pk2). Fixed /api/packets/:id to return parsed path_json in path field. Populated bulk-health per-node stats from SQL. Updated test seed data to use dynamic timestamps for 7-day filter compatibility. All 42+ tests pass, go vet clean. +- Fixed #133 ROOT CAUSE (phantom nodes): `autoLearnHopNodes` in server.js was calling `db.upsertNode()` for every unresolved hop prefix, creating thousands of fake "repeater" nodes with short public_keys (just the 2-4 byte hop prefix). Removed the `upsertNode` call entirely — unresolved hops are now simply cached to skip repeat DB lookups, and display as raw hex prefixes via hop-resolver. Added `db.removePhantomNodes()` that deletes nodes with `LENGTH(public_key) <= 16` (real pubkeys are 64 hex chars). Called at server startup to purge existing phantoms. 14 new test assertions (109 total in test-db.js). diff --git a/.squad/decisions/inbox/hicks-go-full-parity.md b/.squad/decisions/inbox/hicks-go-full-parity.md new file mode 100644 index 00000000..43ef0cec --- /dev/null +++ b/.squad/decisions/inbox/hicks-go-full-parity.md @@ -0,0 +1,51 @@ +# Decision: Go Server Full API Parity + +**Author:** Hicks (Backend Dev) +**Date:** 2026-03-27 + +## Context + +The Go server (`cmd/server/`) is intended as a drop-in replacement for the Node.js server. Multiple endpoints had response shape mismatches, missing fields, and performance issues. + +## Changes Made + +### Performance Fix: `/api/packets?groupByHash=true` (8s → <100ms) +- **Root cause**: `QueryGroupedPackets` was scanning `packets_v` VIEW (1.2M observations) with `GROUP BY hash` +- **Fix**: Rewrite to query `transmissions` table (52K rows) directly with correlated subqueries for count/observer_count/latest. Uses same `buildTransmissionWhere` filter builder and `idx_observations_transmission_id` index as the already-fast `QueryPackets`. + +### `/api/stats` — Field parity +- `totalNodes` now uses 7-day active window (was counting ALL nodes) +- Added `totalNodesAllTime` field +- `GetRoleCounts` for stats uses 7-day filter (matches Node.js `server.js` line 880-886) + +### `/api/nodes` — Counts parity +- Node listing counts use `GetAllRoleCounts` (no time filter) — matches Node.js behavior +- Stats endpoint uses `GetRoleCounts` (7-day filter) — matches Node.js behavior +- Two separate methods to avoid conflating different use cases + +### `/api/packets/:id` — Path data +- Now parses `path_json` and returns actual hop array in `path` field (was returning empty array) + +### `/api/observers` — Live computed fields +- `packetsLastHour`: batch query via `GetObserverPacketCounts` (observations table with observer_idx join) +- `lat`, `lon`, `nodeRole`: batch lookup via `GetNodeLocations` (nodes table where public_key matches observer ID) + +### `/api/observers/:id` — packetsLastHour +- Now computes actual value from observations table + +### `/api/nodes/bulk-health` — Per-node stats +- `totalTransmissions`, `totalObservations`, `packetsToday`, `avgSnr`, `lastHeard` now computed from SQL +- Was returning all zeros + +### `/api/packets` — Multi-node filter +- Added `nodes` query parameter support (comma-separated pubkeys) +- New `QueryMultiNodePackets` DB method + +## Files Modified +- `cmd/server/db.go` — QueryGroupedPackets rewrite, GetStats, GetRoleCounts, GetAllRoleCounts, GetObserverPacketCounts, GetNodeLocations, QueryMultiNodePackets, nullStrVal, nilIfEmpty helpers +- `cmd/server/routes.go` — handlePackets (multi-node), handleObservers, handleObserverDetail, handleBulkHealth, handlePacketDetail, handleStats +- `cmd/server/db_test.go` — Dynamic timestamps in seedTestData (7-day window compat) +- `cmd/server/routes_test.go` — Updated error test to drop observations table (grouped query no longer needs packets_v) + +## Remaining Stubs (Acceptable) +Analytics endpoints (topology, distance, hash-sizes, subpaths) return structural stubs with correct field names. These require in-memory packet store or complex path resolution not yet ported to Go. The frontend degrades gracefully with empty arrays. diff --git a/cmd/server/db.go b/cmd/server/db.go index 51d9cf7c..b366c79d 100644 --- a/cmd/server/db.go +++ b/cmd/server/db.go @@ -184,11 +184,12 @@ type Stats struct { TotalTransmissions int `json:"totalTransmissions"` TotalObservations int `json:"totalObservations"` TotalNodes int `json:"totalNodes"` + TotalNodesAllTime int `json:"totalNodesAllTime"` TotalObservers int `json:"totalObservers"` PacketsLastHour int `json:"packetsLastHour"` } -// GetStats returns aggregate counts. +// GetStats returns aggregate counts (matches Node.js db.getStats shape). func (db *DB) GetStats() (*Stats, error) { s := &Stats{} err := db.conn.QueryRow("SELECT COUNT(*) FROM transmissions").Scan(&s.TotalTransmissions) @@ -198,7 +199,10 @@ func (db *DB) GetStats() (*Stats, error) { s.TotalPackets = s.TotalTransmissions db.conn.QueryRow("SELECT COUNT(*) FROM observations").Scan(&s.TotalObservations) - db.conn.QueryRow("SELECT COUNT(*) FROM nodes").Scan(&s.TotalNodes) + // Node.js uses 7-day active nodes for totalNodes + sevenDaysAgo := time.Now().Add(-7 * 24 * time.Hour).Format(time.RFC3339) + db.conn.QueryRow("SELECT COUNT(*) FROM nodes WHERE last_seen > ?", sevenDaysAgo).Scan(&s.TotalNodes) + db.conn.QueryRow("SELECT COUNT(*) FROM nodes").Scan(&s.TotalNodesAllTime) db.conn.QueryRow("SELECT COUNT(*) FROM observers").Scan(&s.TotalObservers) oneHourAgo := time.Now().Add(-1 * time.Hour).Unix() @@ -207,8 +211,20 @@ func (db *DB) GetStats() (*Stats, error) { return s, nil } -// GetRoleCounts returns count per role. +// GetRoleCounts returns count per role (7-day active, matching Node.js /api/stats). func (db *DB) GetRoleCounts() map[string]int { + sevenDaysAgo := time.Now().Add(-7 * 24 * time.Hour).Format(time.RFC3339) + counts := map[string]int{} + for _, role := range []string{"repeater", "room", "companion", "sensor"} { + var c int + db.conn.QueryRow("SELECT COUNT(*) FROM nodes WHERE role = ? AND last_seen > ?", role, sevenDaysAgo).Scan(&c) + counts[role+"s"] = c + } + return counts +} + +// GetAllRoleCounts returns count per role (all nodes, no time filter — matching Node.js /api/nodes). +func (db *DB) GetAllRoleCounts() map[string]int { counts := map[string]int{} for _, role := range []string{"repeater", "room", "companion", "sensor"} { var c int @@ -288,27 +304,62 @@ func (db *DB) QueryPackets(q PacketQuery) (*PacketResult, error) { return &PacketResult{Packets: packets, Total: total}, nil } -// QueryGroupedPackets groups by hash (transmissions). +// QueryGroupedPackets groups by hash (transmissions) — queries transmissions table directly for performance. func (db *DB) QueryGroupedPackets(q PacketQuery) (*PacketResult, error) { if q.Limit <= 0 { q.Limit = 50 } - where, args := db.buildPacketWhere(q) + + where, args := db.buildTransmissionWhere(q) w := "" if len(where) > 0 { w = "WHERE " + strings.Join(where, " AND ") } - qry := fmt.Sprintf(`SELECT hash, COUNT(*) as count, COUNT(DISTINCT observer_id) as observer_count, - MAX(timestamp) as latest, - (SELECT first_seen FROM transmissions WHERE hash = packets_v.hash LIMIT 1) as first_seen, - MIN(observer_id) as observer_id, MIN(observer_name) as observer_name, - MIN(path_json) as path_json, MIN(payload_type) as payload_type, MIN(route_type) as route_type, - MIN(raw_hex) as raw_hex, MIN(decoded_json) as decoded_json, MIN(snr) as snr, MIN(rssi) as rssi - FROM packets_v %s GROUP BY hash ORDER BY latest DESC LIMIT ? OFFSET ?`, w) - args = append(args, q.Limit, q.Offset) + // Count total transmissions (fast — queries transmissions directly, not packets_v) + var total int + if len(where) == 0 { + db.conn.QueryRow("SELECT COUNT(*) FROM transmissions").Scan(&total) + } else { + db.conn.QueryRow(fmt.Sprintf("SELECT COUNT(*) FROM transmissions t %s", w), args...).Scan(&total) + } - rows, err := db.conn.Query(qry, args...) + // Build grouped query using transmissions table with correlated subqueries + 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(datetime(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, + o.snr, o.rssi, o.path_json + FROM transmissions t + LEFT JOIN observations o ON o.id = ( + SELECT id FROM observations WHERE transmission_id = t.id + ORDER BY length(COALESCE(path_json,'')) DESC LIMIT 1 + ) + LEFT JOIN observers obs ON obs.rowid = o.observer_idx + %s ORDER BY latest DESC LIMIT ? OFFSET ?`, w) + } else { + 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_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, + o.snr, o.rssi, o.path_json + FROM transmissions t + LEFT JOIN observations o ON o.id = ( + SELECT id FROM observations WHERE transmission_id = t.id + ORDER BY length(COALESCE(path_json,'')) DESC LIMIT 1 + ) + %s ORDER BY latest DESC LIMIT ? OFFSET ?`, w) + } + + qArgs := make([]interface{}, len(args)) + copy(qArgs, args) + qArgs = append(qArgs, q.Limit, q.Offset) + + rows, err := db.conn.Query(querySQL, qArgs...) if err != nil { return nil, err } @@ -316,20 +367,24 @@ func (db *DB) QueryGroupedPackets(q PacketQuery) (*PacketResult, error) { packets := make([]map[string]interface{}, 0) for rows.Next() { - var hash, latest, firstSeen, observerID, observerName, pathJSON, rawHex, decodedJSON sql.NullString - var count, observerCount int + var hash, firstSeen, rawHex, decodedJSON, latest, observerID, observerName, pathJSON sql.NullString var payloadType, routeType sql.NullInt64 + var count, observerCount int var snr, rssi sql.NullFloat64 - if err := rows.Scan(&hash, &count, &observerCount, &latest, &firstSeen, &observerID, &observerName, &pathJSON, &payloadType, &routeType, &rawHex, &decodedJSON, &snr, &rssi); err != nil { + + if err := rows.Scan(&hash, &firstSeen, &rawHex, &decodedJSON, &payloadType, &routeType, + &count, &observerCount, &latest, + &observerID, &observerName, &snr, &rssi, &pathJSON); err != nil { continue } - p := map[string]interface{}{ + + packets = append(packets, map[string]interface{}{ "hash": nullStr(hash), + "first_seen": nullStr(firstSeen), "count": count, "observer_count": observerCount, "observation_count": count, "latest": nullStr(latest), - "first_seen": nullStr(firstSeen), "observer_id": nullStr(observerID), "observer_name": nullStr(observerName), "path_json": nullStr(pathJSON), @@ -339,15 +394,9 @@ func (db *DB) QueryGroupedPackets(q PacketQuery) (*PacketResult, error) { "decoded_json": nullStr(decodedJSON), "snr": nullFloat(snr), "rssi": nullFloat(rssi), - } - packets = append(packets, p) + }) } - var total int - countSQL := fmt.Sprintf("SELECT COUNT(DISTINCT hash) FROM packets_v %s", w) - baseArgs := args[:len(args)-2] // remove LIMIT/OFFSET - db.conn.QueryRow(countSQL, baseArgs...).Scan(&total) - return &PacketResult{Packets: packets, Total: total}, nil } @@ -580,7 +629,7 @@ func (db *DB) GetNodes(limit, offset int, role, search, before, lastHeard, sortB } } - counts := db.GetRoleCounts() + counts := db.GetAllRoleCounts() return nodes, total, counts, nil } @@ -1181,6 +1230,123 @@ func (db *DB) GetMaxTransmissionID() int { return maxID } +// GetObserverPacketCounts returns packetsLastHour for all observers (batch query). +func (db *DB) GetObserverPacketCounts(sinceEpoch int64) map[string]int { + counts := make(map[string]int) + var rows *sql.Rows + var err error + if db.isV3 { + rows, err = db.conn.Query(`SELECT obs.id, COUNT(*) as cnt + FROM observations o + JOIN observers obs ON obs.rowid = o.observer_idx + WHERE o.timestamp > ? + GROUP BY obs.id`, sinceEpoch) + } else { + rows, err = db.conn.Query(`SELECT o.observer_id, COUNT(*) as cnt + FROM observations o + WHERE o.observer_id IS NOT NULL AND o.timestamp > ? + GROUP BY o.observer_id`, sinceEpoch) + } + if err != nil { + return counts + } + defer rows.Close() + for rows.Next() { + var id string + var cnt int + rows.Scan(&id, &cnt) + counts[id] = cnt + } + return counts +} + +// GetNodeLocations returns a map of lowercase public_key → {lat, lon, role} for node geo lookups. +func (db *DB) GetNodeLocations() map[string]map[string]interface{} { + result := make(map[string]map[string]interface{}) + rows, err := db.conn.Query("SELECT public_key, lat, lon, role FROM nodes") + if err != nil { + return result + } + defer rows.Close() + for rows.Next() { + var pk string + var role sql.NullString + var lat, lon sql.NullFloat64 + rows.Scan(&pk, &lat, &lon, &role) + result[strings.ToLower(pk)] = map[string]interface{}{ + "lat": nullFloat(lat), + "lon": nullFloat(lon), + "role": nullStr(role), + } + } + return result +} + +// QueryMultiNodePackets returns transmissions referencing any of the given pubkeys. +func (db *DB) QueryMultiNodePackets(pubkeys []string, limit, offset int, order, since, until string) (*PacketResult, error) { + if len(pubkeys) == 0 { + return &PacketResult{Packets: []map[string]interface{}{}, Total: 0}, nil + } + if limit <= 0 { + limit = 50 + } + if order == "" { + order = "DESC" + } + + // Build OR conditions for decoded_json LIKE %pubkey% + var conditions []string + var args []interface{} + for _, pk := range pubkeys { + // Resolve pubkey to also check by name + resolved := db.resolveNodePubkey(pk) + conditions = append(conditions, "t.decoded_json LIKE ?") + args = append(args, "%"+resolved+"%") + } + jsonWhere := "(" + strings.Join(conditions, " OR ") + ")" + + var timeFilters []string + if since != "" { + timeFilters = append(timeFilters, "t.first_seen >= ?") + args = append(args, since) + } + if until != "" { + timeFilters = append(timeFilters, "t.first_seen <= ?") + args = append(args, until) + } + + w := "WHERE " + jsonWhere + if len(timeFilters) > 0 { + w += " AND " + strings.Join(timeFilters, " AND ") + } + + var total int + db.conn.QueryRow(fmt.Sprintf("SELECT COUNT(*) FROM transmissions t %s", w), args...).Scan(&total) + + selectCols, observerJoin := db.transmissionBaseSQL() + querySQL := fmt.Sprintf("SELECT %s FROM transmissions t %s %s ORDER BY t.first_seen %s LIMIT ? OFFSET ?", + selectCols, observerJoin, w, order) + + qArgs := make([]interface{}, len(args)) + copy(qArgs, args) + qArgs = append(qArgs, limit, offset) + + rows, err := db.conn.Query(querySQL, qArgs...) + if err != nil { + return nil, err + } + defer rows.Close() + + packets := make([]map[string]interface{}, 0) + for rows.Next() { + p := db.scanTransmissionRow(rows) + if p != nil { + packets = append(packets, p) + } + } + return &PacketResult{Packets: packets, Total: total}, nil +} + // --- Helpers --- func scanPacketRow(rows *sql.Rows) map[string]interface{} { @@ -1243,6 +1409,20 @@ func nullStr(ns sql.NullString) interface{} { return nil } +func nullStrVal(ns sql.NullString) string { + if ns.Valid { + return ns.String + } + return "" +} + +func nilIfEmpty(s string) interface{} { + if s == "" { + return nil + } + return s +} + func nullFloat(nf sql.NullFloat64) interface{} { if nf.Valid { return nf.Float64 diff --git a/cmd/server/db_test.go b/cmd/server/db_test.go index 0dd87c6c..45dd7e65 100644 --- a/cmd/server/db_test.go +++ b/cmd/server/db_test.go @@ -91,33 +91,41 @@ func setupTestDB(t *testing.T) *DB { func seedTestData(t *testing.T, db *DB) { t.Helper() + // Use recent timestamps so 7-day window filters don't exclude test data + now := time.Now().UTC() + recent := now.Add(-1 * time.Hour).Format(time.RFC3339) + yesterday := now.Add(-24 * time.Hour).Format(time.RFC3339) + twoDaysAgo := now.Add(-48 * time.Hour).Format(time.RFC3339) + recentEpoch := now.Add(-1 * time.Hour).Unix() + yesterdayEpoch := now.Add(-24 * time.Hour).Unix() + // Seed observers db.conn.Exec(`INSERT INTO observers (id, name, iata, last_seen, first_seen, packet_count) - VALUES ('obs1', 'Observer One', 'SJC', '2026-01-15T10:00:00Z', '2026-01-01T00:00:00Z', 100)`) + VALUES ('obs1', 'Observer One', 'SJC', ?, '2026-01-01T00:00:00Z', 100)`, recent) db.conn.Exec(`INSERT INTO observers (id, name, iata, last_seen, first_seen, packet_count) - VALUES ('obs2', 'Observer Two', 'SFO', '2026-01-15T09:00:00Z', '2026-01-01T00:00:00Z', 50)`) + VALUES ('obs2', 'Observer Two', 'SFO', ?, '2026-01-01T00:00:00Z', 50)`, yesterday) // Seed nodes db.conn.Exec(`INSERT INTO nodes (public_key, name, role, lat, lon, last_seen, first_seen, advert_count) - VALUES ('aabbccdd11223344', 'TestRepeater', 'repeater', 37.5, -122.0, '2026-01-15T10:00:00Z', '2026-01-01T00:00:00Z', 50)`) + VALUES ('aabbccdd11223344', 'TestRepeater', 'repeater', 37.5, -122.0, ?, '2026-01-01T00:00:00Z', 50)`, recent) db.conn.Exec(`INSERT INTO nodes (public_key, name, role, lat, lon, last_seen, first_seen, advert_count) - VALUES ('eeff00112233aabb', 'TestCompanion', 'companion', 37.6, -122.1, '2026-01-15T09:00:00Z', '2026-01-01T00:00:00Z', 10)`) + VALUES ('eeff00112233aabb', 'TestCompanion', 'companion', 37.6, -122.1, ?, '2026-01-01T00:00:00Z', 10)`, yesterday) db.conn.Exec(`INSERT INTO nodes (public_key, name, role, lat, lon, last_seen, first_seen, advert_count) - VALUES ('1122334455667788', 'TestRoom', 'room', 37.4, -121.9, '2026-01-14T10:00:00Z', '2026-01-01T00:00:00Z', 5)`) + VALUES ('1122334455667788', 'TestRoom', 'room', 37.4, -121.9, ?, '2026-01-01T00:00:00Z', 5)`, twoDaysAgo) // Seed transmissions db.conn.Exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json) - VALUES ('AABB', 'abc123def4567890', '2026-01-15T10:00:00Z', 1, 4, '{"pubKey":"aabbccdd11223344","name":"TestRepeater","type":"ADVERT"}')`) + VALUES ('AABB', 'abc123def4567890', ?, 1, 4, '{"pubKey":"aabbccdd11223344","name":"TestRepeater","type":"ADVERT"}')`, recent) db.conn.Exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json) - VALUES ('CCDD', '1234567890abcdef', '2026-01-15T09:30:00Z', 1, 5, '{"type":"CHAN","channel":"#test","text":"Hello: World","sender":"TestUser"}')`) + VALUES ('CCDD', '1234567890abcdef', ?, 1, 5, '{"type":"CHAN","channel":"#test","text":"Hello: World","sender":"TestUser"}')`, yesterday) // Seed observations (use unix timestamps) db.conn.Exec(`INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp) - VALUES (1, 1, 12.5, -90, '["aa","bb"]', 1736935200)`) + VALUES (1, 1, 12.5, -90, '["aa","bb"]', ?)`, recentEpoch) db.conn.Exec(`INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp) - VALUES (1, 2, 8.0, -95, '["aa"]', 1736935100)`) + VALUES (1, 2, 8.0, -95, '["aa"]', ?)`, recentEpoch-100) db.conn.Exec(`INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp) - VALUES (2, 1, 15.0, -85, '[]', 1736933400)`) + VALUES (2, 1, 15.0, -85, '[]', ?)`, yesterdayEpoch) } func TestGetStats(t *testing.T) { diff --git a/cmd/server/routes.go b/cmd/server/routes.go index a206f672..3c1d7826 100644 --- a/cmd/server/routes.go +++ b/cmd/server/routes.go @@ -304,6 +304,7 @@ func (s *Server) handleStats(w http.ResponseWriter, r *http.Request) { "totalTransmissions": stats.TotalTransmissions, "totalObservations": stats.TotalObservations, "totalNodes": stats.TotalNodes, + "totalNodesAllTime": stats.TotalNodesAllTime, "totalObservers": stats.TotalObservers, "packetsLastHour": stats.PacketsLastHour, "counts": counts, @@ -339,6 +340,36 @@ func (s *Server) handlePerf(w http.ResponseWriter, r *http.Request) { // --- Packet Handlers --- func (s *Server) handlePackets(w http.ResponseWriter, r *http.Request) { + // Multi-node filter: comma-separated pubkeys (Node.js parity) + if nodesParam := r.URL.Query().Get("nodes"); nodesParam != "" { + pubkeys := strings.Split(nodesParam, ",") + var cleaned []string + for _, pk := range pubkeys { + pk = strings.TrimSpace(pk) + if pk != "" { + cleaned = append(cleaned, pk) + } + } + order := "DESC" + if r.URL.Query().Get("order") == "asc" { + order = "ASC" + } + result, err := s.db.QueryMultiNodePackets(cleaned, + queryInt(r, "limit", 50), queryInt(r, "offset", 0), + order, r.URL.Query().Get("since"), r.URL.Query().Get("until")) + if err != nil { + writeError(w, 500, err.Error()) + return + } + writeJSON(w, map[string]interface{}{ + "packets": result.Packets, + "total": result.Total, + "limit": queryInt(r, "limit", 50), + "offset": queryInt(r, "offset", 0), + }) + return + } + q := PacketQuery{ Limit: queryInt(r, "limit", 50), Offset: queryInt(r, "offset", 0), @@ -434,9 +465,20 @@ func (s *Server) handlePacketDetail(w http.ResponseWriter, r *http.Request) { observationCount = 1 } + // Parse path from path_json + var pathHops []interface{} + if pj, ok := packet["path_json"]; ok && pj != nil { + if pjStr, ok := pj.(string); ok && pjStr != "" { + json.Unmarshal([]byte(pjStr), &pathHops) + } + } + if pathHops == nil { + pathHops = []interface{}{} + } + writeJSON(w, map[string]interface{}{ "packet": packet, - "path": []interface{}{}, + "path": pathHops, "breakdown": map[string]interface{}{}, "observation_count": observationCount, "observations": observations, @@ -517,26 +559,81 @@ func (s *Server) handleBulkHealth(w http.ResponseWriter, r *http.Request) { } defer rows.Close() - results := make([]map[string]interface{}, 0) + type nodeInfo struct { + pk, name, role, lastSeen string + lat, lon interface{} + } + var nodes []nodeInfo for rows.Next() { var pk string var name, role, lastSeen sql.NullString var lat, lon sql.NullFloat64 rows.Scan(&pk, &name, &role, &lat, &lon, &lastSeen) + nodes = append(nodes, nodeInfo{ + pk: pk, name: nullStrVal(name), role: nullStrVal(role), + lastSeen: nullStrVal(lastSeen), + lat: nullFloat(lat), lon: nullFloat(lon), + }) + } + + // Batch query: per-node transmission stats + todayStart := time.Now().UTC().Truncate(24 * time.Hour).Format(time.RFC3339) + results := make([]map[string]interface{}, 0, len(nodes)) + for _, n := range nodes { + pk := "%" + n.pk + "%" + np := "%" + n.name + "%" + + var txCount, obsCount, packetsToday int + var avgSnr sql.NullFloat64 + var lastHeard sql.NullString + + whereClause := "t.decoded_json LIKE ?" + queryArgs := []interface{}{pk} + if n.name != "" { + whereClause = "(t.decoded_json LIKE ? OR t.decoded_json LIKE ?)" + queryArgs = []interface{}{pk, np} + } + + s.db.conn.QueryRow(fmt.Sprintf("SELECT COUNT(*) FROM transmissions t WHERE %s", whereClause), queryArgs...).Scan(&txCount) + + if txCount > 0 { + // Observation count + s.db.conn.QueryRow(fmt.Sprintf(`SELECT COALESCE(SUM( + (SELECT COUNT(*) FROM observations oi WHERE oi.transmission_id = t.id) + ), 0) FROM transmissions t WHERE %s`, whereClause), queryArgs...).Scan(&obsCount) + + // Packets today + todayArgs := append(queryArgs, todayStart) + s.db.conn.QueryRow(fmt.Sprintf("SELECT COUNT(*) FROM transmissions t WHERE %s AND t.first_seen > ?", whereClause), todayArgs...).Scan(&packetsToday) + + // Avg SNR from best observation per transmission + s.db.conn.QueryRow(fmt.Sprintf(`SELECT AVG(o.snr) FROM transmissions t + LEFT JOIN observations o ON o.id = ( + SELECT id FROM observations WHERE transmission_id = t.id AND snr IS NOT NULL LIMIT 1 + ) WHERE %s AND o.snr IS NOT NULL`, whereClause), queryArgs...).Scan(&avgSnr) + + // Last heard + s.db.conn.QueryRow(fmt.Sprintf("SELECT MAX(t.first_seen) FROM transmissions t WHERE %s", whereClause), queryArgs...).Scan(&lastHeard) + } + + lh := n.lastSeen + if lastHeard.Valid && lastHeard.String > lh { + lh = lastHeard.String + } results = append(results, map[string]interface{}{ - "public_key": pk, - "name": nullStr(name), - "role": nullStr(role), - "lat": nullFloat(lat), - "lon": nullFloat(lon), + "public_key": n.pk, + "name": nilIfEmpty(n.name), + "role": nilIfEmpty(n.role), + "lat": n.lat, + "lon": n.lon, "stats": map[string]interface{}{ - "totalTransmissions": 0, - "totalObservations": 0, - "totalPackets": 0, - "packetsToday": 0, - "avgSnr": nil, - "lastHeard": nullStr(lastSeen), + "totalTransmissions": txCount, + "totalObservations": obsCount, + "totalPackets": txCount, + "packetsToday": packetsToday, + "avgSnr": nullFloat(avgSnr), + "lastHeard": nilIfEmpty(lh), }, "observers": []interface{}{}, }) @@ -913,8 +1010,27 @@ func (s *Server) handleObservers(w http.ResponseWriter, r *http.Request) { writeError(w, 500, err.Error()) return } + + // Batch lookup: packetsLastHour per observer + oneHourAgo := time.Now().Add(-1 * time.Hour).Unix() + pktCounts := s.db.GetObserverPacketCounts(oneHourAgo) + + // Batch lookup: node locations (observer ID may match a node public_key) + nodeLocations := s.db.GetNodeLocations() + result := make([]map[string]interface{}, 0, len(observers)) for _, o := range observers { + plh := 0 + if c, ok := pktCounts[o.ID]; ok { + plh = c + } + var lat, lon, nodeRole interface{} + if nodeLoc, ok := nodeLocations[strings.ToLower(o.ID)]; ok { + lat = nodeLoc["lat"] + lon = nodeLoc["lon"] + nodeRole = nodeLoc["role"] + } + m := map[string]interface{}{ "id": o.ID, "name": o.Name, "iata": o.IATA, "last_seen": o.LastSeen, "first_seen": o.FirstSeen, @@ -923,8 +1039,8 @@ func (s *Server) handleObservers(w http.ResponseWriter, r *http.Request) { "client_version": o.ClientVersion, "radio": o.Radio, "battery_mv": o.BatteryMv, "uptime_secs": o.UptimeSecs, "noise_floor": o.NoiseFloor, - "packetsLastHour": 0, - "lat": nil, "lon": nil, "nodeRole": nil, + "packetsLastHour": plh, + "lat": lat, "lon": lon, "nodeRole": nodeRole, } result = append(result, m) } @@ -941,6 +1057,15 @@ func (s *Server) handleObserverDetail(w http.ResponseWriter, r *http.Request) { writeError(w, 404, "Observer not found") return } + + // Compute packetsLastHour from observations + oneHourAgo := time.Now().Add(-1 * time.Hour).Unix() + pktCounts := s.db.GetObserverPacketCounts(oneHourAgo) + plh := 0 + if c, ok := pktCounts[id]; ok { + plh = c + } + writeJSON(w, map[string]interface{}{ "id": obs.ID, "name": obs.Name, "iata": obs.IATA, "last_seen": obs.LastSeen, "first_seen": obs.FirstSeen, @@ -949,7 +1074,7 @@ func (s *Server) handleObserverDetail(w http.ResponseWriter, r *http.Request) { "client_version": obs.ClientVersion, "radio": obs.Radio, "battery_mv": obs.BatteryMv, "uptime_secs": obs.UptimeSecs, "noise_floor": obs.NoiseFloor, - "packetsLastHour": 0, + "packetsLastHour": plh, }) } diff --git a/cmd/server/routes_test.go b/cmd/server/routes_test.go index ecc93228..fa9aed35 100644 --- a/cmd/server/routes_test.go +++ b/cmd/server/routes_test.go @@ -1530,7 +1530,7 @@ func TestHandlerErrorPacketsGrouped(t *testing.T) { router := mux.NewRouter() srv.RegisterRoutes(router) - db.conn.Exec("DROP VIEW IF EXISTS packets_v") + db.conn.Exec("DROP TABLE IF EXISTS observations") req := httptest.NewRequest("GET", "/api/packets?limit=10&groupByHash=true", nil) w := httptest.NewRecorder()