From 42afbb1398ab87cb125482ac55564e2d9af94d8d Mon Sep 17 00:00:00 2001 From: Kpa-clawbot <259247574+Kpa-clawbot@users.noreply.github.com> Date: Fri, 27 Mar 2026 09:17:24 -0700 Subject: [PATCH] perf(go): switch channels + RF analytics to in-memory store MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace SQLite-backed handlers for /api/channels, /api/channels/:hash/messages, /api/analytics/rf, and /api/analytics/channels with in-memory PacketStore queries. Before (SQLite via packets_v VIEW on 1.2M rows): /api/channels 7.2s /api/channels/:hash/msgs 8.2s /api/analytics/rf 4.2s After (in-memory scan of ~50K transmissions): Target: all under 100ms Three new PacketStore methods: - GetChannels(region) — filters payload_type 5 + decoded type CHAN - GetChannelMessages(hash, limit, offset) — deduplicates by sender+hash - GetAnalyticsRF(region) — full RF stats with histograms, scatter, per-type SNR All handlers fall back to DB queries when store is nil (test compat). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- cmd/server/routes.go | 26 +- cmd/server/store.go | 743 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 767 insertions(+), 2 deletions(-) diff --git a/cmd/server/routes.go b/cmd/server/routes.go index 6f72cdac..72036cb8 100644 --- a/cmd/server/routes.go +++ b/cmd/server/routes.go @@ -843,7 +843,12 @@ func (s *Server) handleNodeAnalytics(w http.ResponseWriter, r *http.Request) { // --- Analytics Handlers --- func (s *Server) handleAnalyticsRF(w http.ResponseWriter, r *http.Request) { - // Basic RF analytics from SQL + if s.store != nil { + region := r.URL.Query().Get("region") + writeJSON(w, s.store.GetAnalyticsRF(region)) + return + } + // Fallback: basic RF analytics from SQL region := r.URL.Query().Get("region") regionFilter := "" var rArgs []interface{} @@ -926,7 +931,13 @@ func (s *Server) handleAnalyticsTopology(w http.ResponseWriter, r *http.Request) } func (s *Server) handleAnalyticsChannels(w http.ResponseWriter, r *http.Request) { - channels, _ := s.db.GetChannels() + var channels []map[string]interface{} + if s.store != nil { + region := r.URL.Query().Get("region") + channels = s.store.GetChannels(region) + } else { + channels, _ = s.db.GetChannels() + } writeJSON(w, map[string]interface{}{ "activeChannels": len(channels), "decryptable": len(channels), @@ -1037,6 +1048,12 @@ func (s *Server) handleResolveHops(w http.ResponseWriter, r *http.Request) { } func (s *Server) handleChannels(w http.ResponseWriter, r *http.Request) { + if s.store != nil { + region := r.URL.Query().Get("region") + channels := s.store.GetChannels(region) + writeJSON(w, map[string]interface{}{"channels": channels}) + return + } channels, err := s.db.GetChannels() if err != nil { writeError(w, 500, err.Error()) @@ -1049,6 +1066,11 @@ func (s *Server) handleChannelMessages(w http.ResponseWriter, r *http.Request) { hash := mux.Vars(r)["hash"] limit := queryInt(r, "limit", 100) offset := queryInt(r, "offset", 0) + if s.store != nil { + messages, total := s.store.GetChannelMessages(hash, limit, offset) + writeJSON(w, map[string]interface{}{"messages": messages, "total": total}) + return + } messages, total, err := s.db.GetChannelMessages(hash, limit, offset) if err != nil { writeError(w, 500, err.Error()) diff --git a/cmd/server/store.go b/cmd/server/store.go index c20148a7..8a8b3152 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -3,7 +3,9 @@ package main import ( "database/sql" "encoding/json" + "fmt" "log" + "math" "sort" "strings" "sync" @@ -965,3 +967,744 @@ func filterTxSlice(s []*StoreTx, fn func(*StoreTx) bool) []*StoreTx { } return result } + +// GetChannels returns channel list from in-memory packets (payload_type 5, decoded type CHAN). +func (s *PacketStore) GetChannels(region string) []map[string]interface{} { + s.mu.RLock() + defer s.mu.RUnlock() + + var regionObs map[string]bool + if region != "" { + regionObs = s.resolveRegionObservers(region) + } + + type chanInfo struct { + Hash string + Name string + LastMessage interface{} + LastSender interface{} + MessageCount int + LastActivity string + } + channelMap := map[string]*chanInfo{} + + for _, tx := range s.packets { + if tx.PayloadType == nil || *tx.PayloadType != 5 { + continue + } + if tx.DecodedJSON == "" { + continue + } + + // Region filter: check if any observation is from a regional observer + if regionObs != nil { + match := false + for _, obs := range tx.Observations { + if regionObs[obs.ObserverID] { + match = true + break + } + } + if !match { + continue + } + } + + var decoded map[string]interface{} + if json.Unmarshal([]byte(tx.DecodedJSON), &decoded) != nil { + continue + } + dtype, _ := decoded["type"].(string) + if dtype != "CHAN" { + continue + } + + channelName, _ := decoded["channel"].(string) + if channelName == "" { + channelName = "unknown" + } + key := channelName + + ch := channelMap[key] + if ch == nil { + ch = &chanInfo{ + Hash: key, Name: channelName, + LastActivity: tx.FirstSeen, + } + channelMap[key] = ch + } + ch.MessageCount++ + if tx.FirstSeen >= ch.LastActivity { + ch.LastActivity = tx.FirstSeen + if text, ok := decoded["text"].(string); ok && text != "" { + idx := strings.Index(text, ": ") + if idx > 0 { + ch.LastMessage = text[idx+2:] + } else { + ch.LastMessage = text + } + if sender, ok := decoded["sender"].(string); ok { + ch.LastSender = sender + } + } + } + } + + channels := make([]map[string]interface{}, 0, len(channelMap)) + for _, ch := range channelMap { + channels = append(channels, map[string]interface{}{ + "hash": ch.Hash, "name": ch.Name, + "lastMessage": ch.LastMessage, "lastSender": ch.LastSender, + "messageCount": ch.MessageCount, "lastActivity": ch.LastActivity, + }) + } + return channels +} + +// GetChannelMessages returns deduplicated messages for a channel from in-memory packets. +func (s *PacketStore) GetChannelMessages(channelHash string, limit, offset int) ([]map[string]interface{}, int) { + s.mu.RLock() + defer s.mu.RUnlock() + + if limit <= 0 { + limit = 100 + } + + type msgEntry struct { + Data map[string]interface{} + Repeats int + Observers []string + } + msgMap := map[string]*msgEntry{} + var msgOrder []string + + // Iterate oldest-first (packets stored newest-first, so reverse) + for i := len(s.packets) - 1; i >= 0; i-- { + tx := s.packets[i] + if tx.PayloadType == nil || *tx.PayloadType != 5 || tx.DecodedJSON == "" { + continue + } + + var decoded map[string]interface{} + if json.Unmarshal([]byte(tx.DecodedJSON), &decoded) != nil { + continue + } + dtype, _ := decoded["type"].(string) + if dtype != "CHAN" { + continue + } + ch, _ := decoded["channel"].(string) + if ch == "" { + ch = "unknown" + } + if ch != channelHash { + continue + } + + text, _ := decoded["text"].(string) + sender, _ := decoded["sender"].(string) + if sender == "" && text != "" { + idx := strings.Index(text, ": ") + if idx > 0 && idx < 50 { + sender = text[:idx] + } + } + + dedupeKey := sender + ":" + tx.Hash + + if existing, ok := msgMap[dedupeKey]; ok { + existing.Repeats++ + existing.Data["repeats"] = existing.Repeats + // Add observer if new + obsName := tx.ObserverName + if obsName == "" { + obsName = tx.ObserverID + } + if obsName != "" { + found := false + for _, o := range existing.Observers { + if o == obsName { + found = true + break + } + } + if !found { + existing.Observers = append(existing.Observers, obsName) + existing.Data["observers"] = existing.Observers + } + } + } else { + displaySender := sender + displayText := text + if text != "" { + idx := strings.Index(text, ": ") + if idx > 0 && idx < 50 { + displaySender = text[:idx] + displayText = text[idx+2:] + } + } + + hops := pathLen(tx.PathJSON) + + var snrVal interface{} + if tx.SNR != nil { + snrVal = *tx.SNR + } + + senderTs, _ := decoded["sender_timestamp"] + + observers := []string{} + obsName := tx.ObserverName + if obsName == "" { + obsName = tx.ObserverID + } + if obsName != "" { + observers = []string{obsName} + } + + entry := &msgEntry{ + Data: map[string]interface{}{ + "sender": displaySender, + "text": displayText, + "timestamp": strOrNil(tx.FirstSeen), + "sender_timestamp": senderTs, + "packetId": tx.ID, + "packetHash": strOrNil(tx.Hash), + "repeats": 1, + "observers": observers, + "hops": hops, + "snr": snrVal, + }, + Repeats: 1, + Observers: observers, + } + msgMap[dedupeKey] = entry + msgOrder = append(msgOrder, dedupeKey) + } + } + + total := len(msgOrder) + // Return latest messages (tail) + start := total - limit - offset + if start < 0 { + start = 0 + } + end := total - offset + if end < 0 { + end = 0 + } + if end > total { + end = total + } + + messages := make([]map[string]interface{}, 0, end-start) + for i := start; i < end; i++ { + messages = append(messages, msgMap[msgOrder[i]].Data) + } + return messages, total +} + +// GetAnalyticsRF returns full RF analytics computed from in-memory observations. +func (s *PacketStore) GetAnalyticsRF(region string) map[string]interface{} { + s.mu.RLock() + defer s.mu.RUnlock() + + ptNames := map[int]string{0: "REQ", 1: "RESPONSE", 2: "TXT_MSG", 3: "ACK", 4: "ADVERT", 5: "GRP_TXT", 7: "ANON_REQ", 8: "PATH", 9: "TRACE", 11: "CONTROL"} + + var regionObs map[string]bool + if region != "" { + regionObs = s.resolveRegionObservers(region) + } + + // Collect all observations matching the region + var snrVals, rssiVals []float64 + var packetSizes []int + seenSizeHashes := map[string]bool{} + seenTypeHashes := map[string]bool{} + typeBuckets := map[int]int{} + hourBuckets := map[string]int{} + snrByType := map[string]*struct{ vals []float64 }{} + sigTime := map[string]*struct { + snrs []float64 + count int + }{} + var scatterAll []struct{ snr, rssi float64 } + totalObs := 0 + regionalHashes := map[string]bool{} + + if regionObs != nil { + // Regional: iterate observations from matching observers + for obsID := range regionObs { + obsList := s.byObserver[obsID] + for _, obs := range obsList { + totalObs++ + tx := s.byTxID[obs.TransmissionID] + hash := "" + if tx != nil { + hash = tx.Hash + } + if hash != "" { + regionalHashes[hash] = true + } + + ts := obs.Timestamp + + // SNR/RSSI + if obs.SNR != nil { + snrVals = append(snrVals, *obs.SNR) + typeName := "UNK" + if tx != nil && tx.PayloadType != nil { + if n, ok := ptNames[*tx.PayloadType]; ok { + typeName = n + } else { + typeName = fmt.Sprintf("UNK(%d)", *tx.PayloadType) + } + } + if snrByType[typeName] == nil { + snrByType[typeName] = &struct{ vals []float64 }{} + } + snrByType[typeName].vals = append(snrByType[typeName].vals, *obs.SNR) + + if obs.RSSI != nil { + scatterAll = append(scatterAll, struct{ snr, rssi float64 }{*obs.SNR, *obs.RSSI}) + } + + // Signal over time + if len(ts) >= 13 { + hr := ts[:13] + if sigTime[hr] == nil { + sigTime[hr] = &struct { + snrs []float64 + count int + }{} + } + sigTime[hr].snrs = append(sigTime[hr].snrs, *obs.SNR) + sigTime[hr].count++ + } + } + if obs.RSSI != nil { + rssiVals = append(rssiVals, *obs.RSSI) + } + + // Packets per hour + if len(ts) >= 13 { + hr := ts[:13] + hourBuckets[hr]++ + } + + // Packet sizes (unique by hash) + if hash != "" && !seenSizeHashes[hash] && tx != nil && tx.RawHex != "" { + seenSizeHashes[hash] = true + packetSizes = append(packetSizes, len(tx.RawHex)/2) + } + + // Payload type distribution (unique by hash) + if hash != "" && !seenTypeHashes[hash] && tx != nil && tx.PayloadType != nil { + seenTypeHashes[hash] = true + typeBuckets[*tx.PayloadType]++ + } + } + } + } else { + // No region: iterate all transmissions and their observations + for _, tx := range s.packets { + if len(tx.Observations) > 0 { + for _, obs := range tx.Observations { + totalObs++ + if tx.Hash != "" { + regionalHashes[tx.Hash] = true + } + ts := obs.Timestamp + + if obs.SNR != nil { + snrVals = append(snrVals, *obs.SNR) + typeName := "UNK" + if tx.PayloadType != nil { + if n, ok := ptNames[*tx.PayloadType]; ok { + typeName = n + } else { + typeName = fmt.Sprintf("UNK(%d)", *tx.PayloadType) + } + } + if snrByType[typeName] == nil { + snrByType[typeName] = &struct{ vals []float64 }{} + } + snrByType[typeName].vals = append(snrByType[typeName].vals, *obs.SNR) + + if obs.RSSI != nil { + scatterAll = append(scatterAll, struct{ snr, rssi float64 }{*obs.SNR, *obs.RSSI}) + } + + if len(ts) >= 13 { + hr := ts[:13] + if sigTime[hr] == nil { + sigTime[hr] = &struct { + snrs []float64 + count int + }{} + } + sigTime[hr].snrs = append(sigTime[hr].snrs, *obs.SNR) + sigTime[hr].count++ + } + } + if obs.RSSI != nil { + rssiVals = append(rssiVals, *obs.RSSI) + } + + if len(ts) >= 13 { + hourBuckets[ts[:13]]++ + } + + if tx.Hash != "" && !seenSizeHashes[tx.Hash] && tx.RawHex != "" { + seenSizeHashes[tx.Hash] = true + packetSizes = append(packetSizes, len(tx.RawHex)/2) + } + if tx.Hash != "" && !seenTypeHashes[tx.Hash] && tx.PayloadType != nil { + seenTypeHashes[tx.Hash] = true + typeBuckets[*tx.PayloadType]++ + } + } + } else { + // Legacy: transmission without observations + totalObs++ + if tx.Hash != "" { + regionalHashes[tx.Hash] = true + } + if tx.SNR != nil { + snrVals = append(snrVals, *tx.SNR) + } + if tx.RSSI != nil { + rssiVals = append(rssiVals, *tx.RSSI) + } + ts := tx.FirstSeen + if len(ts) >= 13 { + hourBuckets[ts[:13]]++ + } + if tx.Hash != "" && !seenSizeHashes[tx.Hash] && tx.RawHex != "" { + seenSizeHashes[tx.Hash] = true + packetSizes = append(packetSizes, len(tx.RawHex)/2) + } + if tx.Hash != "" && !seenTypeHashes[tx.Hash] && tx.PayloadType != nil { + seenTypeHashes[tx.Hash] = true + typeBuckets[*tx.PayloadType]++ + } + } + } + } + + // Stats helpers + sortedF64 := func(arr []float64) []float64 { + c := make([]float64, len(arr)) + copy(c, arr) + sort.Float64s(c) + return c + } + medianF64 := func(arr []float64) float64 { + s := sortedF64(arr) + if len(s) == 0 { + return 0 + } + return s[len(s)/2] + } + stddevF64 := func(arr []float64, avg float64) float64 { + if len(arr) == 0 { + return 0 + } + sum := 0.0 + for _, v := range arr { + d := v - avg + sum += d * d + } + return math.Sqrt(sum / float64(len(arr))) + } + minF64 := func(arr []float64) float64 { + if len(arr) == 0 { + return 0 + } + m := arr[0] + for _, v := range arr[1:] { + if v < m { + m = v + } + } + return m + } + maxF64 := func(arr []float64) float64 { + if len(arr) == 0 { + return 0 + } + m := arr[0] + for _, v := range arr[1:] { + if v > m { + m = v + } + } + return m + } + minInt := func(arr []int) int { + if len(arr) == 0 { + return 0 + } + m := arr[0] + for _, v := range arr[1:] { + if v < m { + m = v + } + } + return m + } + maxInt := func(arr []int) int { + if len(arr) == 0 { + return 0 + } + m := arr[0] + for _, v := range arr[1:] { + if v > m { + m = v + } + } + return m + } + + snrAvg := 0.0 + if len(snrVals) > 0 { + sum := 0.0 + for _, v := range snrVals { + sum += v + } + snrAvg = sum / float64(len(snrVals)) + } + rssiAvg := 0.0 + if len(rssiVals) > 0 { + sum := 0.0 + for _, v := range rssiVals { + sum += v + } + rssiAvg = sum / float64(len(rssiVals)) + } + + // Packets per hour + type hourCount struct { + Hour string `json:"hour"` + Count int `json:"count"` + } + hourKeys := make([]string, 0, len(hourBuckets)) + for k := range hourBuckets { + hourKeys = append(hourKeys, k) + } + sort.Strings(hourKeys) + packetsPerHour := make([]hourCount, len(hourKeys)) + for i, k := range hourKeys { + packetsPerHour[i] = hourCount{Hour: k, Count: hourBuckets[k]} + } + + // Payload types + type ptEntry struct { + Type int `json:"type"` + Name string `json:"name"` + Count int `json:"count"` + } + payloadTypes := make([]ptEntry, 0, len(typeBuckets)) + for t, c := range typeBuckets { + name := ptNames[t] + if name == "" { + name = fmt.Sprintf("UNK(%d)", t) + } + payloadTypes = append(payloadTypes, ptEntry{Type: t, Name: name, Count: c}) + } + sort.Slice(payloadTypes, func(i, j int) bool { return payloadTypes[i].Count > payloadTypes[j].Count }) + + // SNR by type + type snrTypeEntry struct { + Name string `json:"name"` + Count int `json:"count"` + Avg float64 `json:"avg"` + Min float64 `json:"min"` + Max float64 `json:"max"` + } + snrByTypeArr := make([]snrTypeEntry, 0, len(snrByType)) + for name, d := range snrByType { + sum := 0.0 + for _, v := range d.vals { + sum += v + } + snrByTypeArr = append(snrByTypeArr, snrTypeEntry{ + Name: name, Count: len(d.vals), + Avg: sum / float64(len(d.vals)), + Min: minF64(d.vals), Max: maxF64(d.vals), + }) + } + sort.Slice(snrByTypeArr, func(i, j int) bool { return snrByTypeArr[i].Count > snrByTypeArr[j].Count }) + + // Signal over time + type sigTimeEntry struct { + Hour string `json:"hour"` + Count int `json:"count"` + AvgSnr float64 `json:"avgSnr"` + } + sigKeys := make([]string, 0, len(sigTime)) + for k := range sigTime { + sigKeys = append(sigKeys, k) + } + sort.Strings(sigKeys) + signalOverTime := make([]sigTimeEntry, len(sigKeys)) + for i, k := range sigKeys { + d := sigTime[k] + sum := 0.0 + for _, v := range d.snrs { + sum += v + } + signalOverTime[i] = sigTimeEntry{Hour: k, Count: d.count, AvgSnr: sum / float64(d.count)} + } + + // Scatter (downsample to 500) + type scatterPoint struct { + SNR float64 `json:"snr"` + RSSI float64 `json:"rssi"` + } + scatterStep := 1 + if len(scatterAll) > 500 { + scatterStep = len(scatterAll) / 500 + } + scatterData := make([]scatterPoint, 0, 500) + for i, p := range scatterAll { + if i%scatterStep == 0 { + scatterData = append(scatterData, scatterPoint{SNR: p.snr, RSSI: p.rssi}) + } + } + + // Histograms + buildHistogramF64 := func(values []float64, bins int) map[string]interface{} { + if len(values) == 0 { + return map[string]interface{}{"bins": []interface{}{}, "min": 0, "max": 0} + } + mn, mx := minF64(values), maxF64(values) + rng := mx - mn + if rng == 0 { + rng = 1 + } + binWidth := rng / float64(bins) + counts := make([]int, bins) + for _, v := range values { + idx := int((v - mn) / binWidth) + if idx >= bins { + idx = bins - 1 + } + counts[idx]++ + } + binArr := make([]map[string]interface{}, bins) + for i, c := range counts { + binArr[i] = map[string]interface{}{"x": mn + float64(i)*binWidth, "w": binWidth, "count": c} + } + return map[string]interface{}{"bins": binArr, "min": mn, "max": mx} + } + buildHistogramInt := func(values []int, bins int) map[string]interface{} { + if len(values) == 0 { + return map[string]interface{}{"bins": []interface{}{}, "min": 0, "max": 0} + } + mn, mx := float64(minInt(values)), float64(maxInt(values)) + rng := mx - mn + if rng == 0 { + rng = 1 + } + binWidth := rng / float64(bins) + counts := make([]int, bins) + for _, v := range values { + idx := int((float64(v) - mn) / binWidth) + if idx >= bins { + idx = bins - 1 + } + counts[idx]++ + } + binArr := make([]map[string]interface{}, bins) + for i, c := range counts { + binArr[i] = map[string]interface{}{"x": mn + float64(i)*binWidth, "w": binWidth, "count": c} + } + return map[string]interface{}{"bins": binArr, "min": mn, "max": mx} + } + + snrHistogram := buildHistogramF64(snrVals, 20) + rssiHistogram := buildHistogramF64(rssiVals, 20) + sizeHistogram := buildHistogramInt(packetSizes, 25) + + // Time span + timeSpanHours := 0.0 + if totalObs > 0 { + var minT, maxT int64 + first := true + // Scan a representative set for time bounds + for _, tx := range s.packets { + for _, obs := range tx.Observations { + t, err := time.Parse("2006-01-02 15:04:05", obs.Timestamp) + if err != nil { + t, err = time.Parse(time.RFC3339, obs.Timestamp) + } + if err != nil { + continue + } + ms := t.UnixMilli() + if first { + minT, maxT = ms, ms + first = false + } else { + if ms < minT { + minT = ms + } + if ms > maxT { + maxT = ms + } + } + } + } + if !first { + timeSpanHours = float64(maxT-minT) / 3600000.0 + } + } + + // Avg packet size + avgPktSize := 0 + if len(packetSizes) > 0 { + sum := 0 + for _, v := range packetSizes { + sum += v + } + avgPktSize = sum / len(packetSizes) + } + + snrStats := map[string]interface{}{"min": 0.0, "max": 0.0, "avg": 0.0, "median": 0.0, "stddev": 0.0} + if len(snrVals) > 0 { + snrStats = map[string]interface{}{ + "min": minF64(snrVals), "max": maxF64(snrVals), + "avg": snrAvg, "median": medianF64(snrVals), + "stddev": stddevF64(snrVals, snrAvg), + } + } + rssiStats := map[string]interface{}{"min": 0.0, "max": 0.0, "avg": 0.0, "median": 0.0, "stddev": 0.0} + if len(rssiVals) > 0 { + rssiStats = map[string]interface{}{ + "min": minF64(rssiVals), "max": maxF64(rssiVals), + "avg": rssiAvg, "median": medianF64(rssiVals), + "stddev": stddevF64(rssiVals, rssiAvg), + } + } + + return map[string]interface{}{ + "totalPackets": len(snrVals), + "totalAllPackets": totalObs, + "totalTransmissions": len(regionalHashes), + "snr": snrStats, + "rssi": rssiStats, + "snrValues": snrHistogram, + "rssiValues": rssiHistogram, + "packetSizes": sizeHistogram, + "minPacketSize": minInt(packetSizes), + "maxPacketSize": maxInt(packetSizes), + "avgPacketSize": avgPktSize, + "packetsPerHour": packetsPerHour, + "payloadTypes": payloadTypes, + "snrByType": snrByTypeArr, + "signalOverTime": signalOverTime, + "scatterData": scatterData, + "timeSpanHours": timeSpanHours, + } +}