mirror of
https://github.com/Kpa-clawbot/meshcore-analyzer.git
synced 2026-10-22 22:01:18 +00:00
feat(go-server): full API parity with Node.js server
Performance: - QueryGroupedPackets: 8s → <100ms (transmissions table, not packets_v VIEW) Field parity: - /api/stats: totalNodes uses 7-day window, added totalNodesAllTime - /api/stats: role counts filtered by 7-day (matching Node.js) - /api/nodes: role counts use all-time (matching Node.js) - /api/packets/🆔 path field returns parsed path_json hops - /api/packets: added multi-node filter (?nodes=pk1,pk2) - /api/observers: packetsLastHour, lat, lon, nodeRole computed - /api/observers/🆔 packetsLastHour computed - /api/nodes/bulk-health: per-node stats from SQL Tests updated with dynamic timestamps for 7-day filter compat. All tests pass, go vet clean. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
1 parent
38d4840b10
commit
5c68605f2c
6 files changed
+423
-54
No files matched your search
@@ -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).
|
||||
@@ -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.
|
||||
+207
-27
@@ -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
|
||||
|
||||
+18
-10
@@ -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) {
|
||||
|
||||
+141
-16
@@ -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,
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in new issue
Block a user