From 59c225593fe5e2bc830bb38b4e0a58a924f76ea1 Mon Sep 17 00:00:00 2001 From: Kpa-clawbot <259247574+Kpa-clawbot@users.noreply.github.com> Date: Fri, 27 Mar 2026 16:06:24 -0700 Subject: [PATCH] =?UTF-8?q?fix(go):=20resolve=206=20backend=20issues=20?= =?UTF-8?q?=E2=80=94=20#165=20#168=20#169=20#170=20#171=20#172?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #165 — build_time in API: already implemented (BuildTime ldflags in Dockerfile.go, main.go, StatsResponse, HealthResponse) #168 — subpaths API slow: cache (subpathCache with TTL) and invalidation already in place; verified working #169 — distance API slow: cache (distCache with TTL) and invalidation already in place; verified working #170 — audio-lab/buckets: in-memory store path already implemented, matching Node.js pktStore.packets iteration with type grouping and size-distributed sampling #171 — channels stale latest message: add companion bridge handling to Go ingestor for meshcore/message/channel/ and meshcore/message/direct/ MQTT topics. Stores decoded channel messages with type CHAN in decoded_json, enabling the channels endpoint to find them. Also handles direct messages. #172 — packets page not live-updating: add missing direction field to WS broadcast packet map for full parity with txToMap/Node.js fullPacket shape. WS broadcast shape verified correct (type, data.packet structure, timestamp, payload_type, observer_id all present). Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- cmd/ingestor/main.go | 163 ++++++++++++++++++++++++++++++++++++++++++- cmd/server/store.go | 1 + 2 files changed, 162 insertions(+), 2 deletions(-) diff --git a/cmd/ingestor/main.go b/cmd/ingestor/main.go index 92d789fa..ce48eb60 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -1,7 +1,9 @@ package main import ( + "crypto/sha256" "crypto/tls" + "encoding/hex" "encoding/json" "flag" "fmt" @@ -10,6 +12,7 @@ import ( "os/signal" "strings" "syscall" + "time" mqtt "github.com/eclipse/paho.mqtt.golang" ) @@ -228,8 +231,164 @@ func handleMessage(store *Store, tag string, source MQTTSource, m mqtt.Message) return } - // Other message formats (companion bridge etc.) are not handled yet. - // This first iteration focuses on the raw packet format (Format 1). + // Format 2: Companion bridge channel message (meshcore/message/channel/) + if strings.HasPrefix(topic, "meshcore/message/channel/") { + text, _ := msg["text"].(string) + if text == "" { + return + } + + channelIdx := "" + if len(parts) >= 4 { + channelIdx = parts[3] + } + if ci, ok := msg["channel_idx"]; ok { + channelIdx = fmt.Sprintf("%v", ci) + } + + // Extract sender from "Name: message" format + sender := "" + if idx := strings.Index(text, ": "); idx > 0 && idx < 50 { + sender = text[:idx] + } + + channelName := fmt.Sprintf("ch%s", channelIdx) + + // Build decoded JSON matching Node.js CHAN format + channelMsg := map[string]interface{}{ + "type": "CHAN", + "channel": channelName, + "text": text, + "sender": sender, + } + if st, ok := msg["sender_timestamp"]; ok { + channelMsg["sender_timestamp"] = st + } + + decodedJSON, _ := json.Marshal(channelMsg) + + now := time.Now().UTC().Format(time.RFC3339) + hashInput := fmt.Sprintf("ch:%s:%s:%s", channelIdx, text, now) + h := sha256.Sum256([]byte(hashInput)) + hash := hex.EncodeToString(h[:])[:16] + + var snr, rssi *float64 + if v, ok := msg["SNR"]; ok { + if f, ok := toFloat64(v); ok { + snr = &f + } + } else if v, ok := msg["snr"]; ok { + if f, ok := toFloat64(v); ok { + snr = &f + } + } + if v, ok := msg["RSSI"]; ok { + if f, ok := toFloat64(v); ok { + rssi = &f + } + } else if v, ok := msg["rssi"]; ok { + if f, ok := toFloat64(v); ok { + rssi = &f + } + } + + pktData := &PacketData{ + Timestamp: now, + ObserverID: "companion", + ObserverName: "L1 Pro (BLE)", + SNR: snr, + RSSI: rssi, + Hash: hash, + RouteType: 1, // FLOOD + PayloadType: 5, // GRP_TXT + PathJSON: "[]", + DecodedJSON: string(decodedJSON), + } + + if err := store.InsertTransmission(pktData); err != nil { + log.Printf("MQTT [%s] channel insert error: %v", tag, err) + } + + // Upsert sender as a companion node + if sender != "" { + senderKey := "sender-" + strings.ToLower(sender) + if err := store.UpsertNode(senderKey, sender, "companion", nil, nil, now); err != nil { + log.Printf("MQTT [%s] sender node upsert error: %v", tag, err) + } + } + + log.Printf("MQTT [%s] channel message: ch%s from %s", tag, channelIdx, firstNonEmpty(sender, "unknown")) + return + } + + // Format 2b: Companion bridge direct message (meshcore/message/direct/) + if strings.HasPrefix(topic, "meshcore/message/direct/") { + text, _ := msg["text"].(string) + if text == "" { + return + } + + sender := "" + if idx := strings.Index(text, ": "); idx > 0 && idx < 50 { + sender = text[:idx] + } + + dm := map[string]interface{}{ + "type": "DM", + "text": text, + "sender": sender, + } + if st, ok := msg["sender_timestamp"]; ok { + dm["sender_timestamp"] = st + } + + decodedJSON, _ := json.Marshal(dm) + + now := time.Now().UTC().Format(time.RFC3339) + hashInput := fmt.Sprintf("dm:%s:%s", text, now) + h := sha256.Sum256([]byte(hashInput)) + hash := hex.EncodeToString(h[:])[:16] + + var snr, rssi *float64 + if v, ok := msg["SNR"]; ok { + if f, ok := toFloat64(v); ok { + snr = &f + } + } else if v, ok := msg["snr"]; ok { + if f, ok := toFloat64(v); ok { + snr = &f + } + } + if v, ok := msg["RSSI"]; ok { + if f, ok := toFloat64(v); ok { + rssi = &f + } + } else if v, ok := msg["rssi"]; ok { + if f, ok := toFloat64(v); ok { + rssi = &f + } + } + + pktData := &PacketData{ + Timestamp: now, + ObserverID: "companion", + ObserverName: "L1 Pro (BLE)", + SNR: snr, + RSSI: rssi, + Hash: hash, + RouteType: 1, // FLOOD + PayloadType: 2, // TXT_MSG + PathJSON: "[]", + DecodedJSON: string(decodedJSON), + } + + if err := store.InsertTransmission(pktData); err != nil { + log.Printf("MQTT [%s] DM insert error: %v", tag, err) + } + + log.Printf("MQTT [%s] direct message from %s", tag, firstNonEmpty(sender, "unknown")) + return + } } func toFloat64(v interface{}) (float64, bool) { diff --git a/cmd/server/store.go b/cmd/server/store.go index 9e233bbe..f213e5b6 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -961,6 +961,7 @@ func (s *PacketStore) IngestNewFromDB(sinceID, limit int) ([]map[string]interfac "snr": floatPtrOrNil(tx.SNR), "rssi": floatPtrOrNil(tx.RSSI), "path_json": strOrNil(tx.PathJSON), + "direction": strOrNil(tx.Direction), "observation_count": tx.ObservationCount, } // Broadcast map: top-level fields for live.js + nested packet for packets.js