mirror of
https://github.com/MeshCore-Beacon/beacon-server.git
synced 2026-10-20 11:01:56 +00:00
MrAlders0n
b30668bac4
A full send buffer dropped the oldest event and then the new one too while reporting one drop. Subscribe accepted routeTypes and observerIds but never applied them.
320 lines
10 KiB
Go
320 lines
10 KiB
Go
// Copyright 2026 Beacon Contributors
|
|
// SPDX-License-Identifier: AGPL-3.0-or-later
|
|
|
|
package ingest
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"fmt"
|
|
"slices"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/MeshCore-Beacon/beacon-server/internal/hub"
|
|
"github.com/MeshCore-Beacon/beacon-server/internal/lora"
|
|
)
|
|
|
|
// UpdateObserverStatusParams carries the fields parsed from a /status message.
|
|
type UpdateObserverStatusParams struct {
|
|
PublicKey []byte
|
|
StatusMetadata json.RawMessage
|
|
LastStatusAt time.Time
|
|
BatteryLevel *float32
|
|
UptimeSeconds *int64
|
|
SoftwareVersion string
|
|
ObserverType string // only set if we can detect it; never downgrade to unknown
|
|
DisplayName string // only set if current value is NULL
|
|
HardwareModel string
|
|
FirmwareVersion string
|
|
FirmwareBuild string
|
|
RadioFreqMHz float32
|
|
RadioSF int16
|
|
RadioBWKHz float32
|
|
RadioCR int16
|
|
}
|
|
|
|
// statusEvent is the JSON payload for an observerStatus WS event.
|
|
// Shape matches the design doc § Server → Client events.
|
|
type statusEvent struct {
|
|
ObserverID string `json:"observerId"`
|
|
DisplayName string `json:"displayName"`
|
|
ObserverType *string `json:"observerType,omitempty"`
|
|
IATA string `json:"iata,omitempty"`
|
|
Online bool `json:"online"`
|
|
Radio *string `json:"radio,omitempty"`
|
|
Scopes []string `json:"scopes"`
|
|
BatteryMV int `json:"batteryMv,omitempty"`
|
|
UptimeSeconds int64 `json:"uptimeSeconds"`
|
|
LastStatusAt int64 `json:"lastStatusAt"`
|
|
}
|
|
|
|
type statusStats struct {
|
|
UptimeSeconds int64 `json:"uptime_secs"`
|
|
BatteryMV int `json:"battery_mv"`
|
|
NoiseFloor float32 `json:"noise_floor"`
|
|
QueueLen int `json:"queue_len"`
|
|
DebugFlags int `json:"debug_flags"`
|
|
TxAirSecs float64 `json:"tx_air_secs"`
|
|
RxAirSecs float64 `json:"rx_air_secs"`
|
|
RecvErrors int `json:"recv_errors"`
|
|
}
|
|
|
|
// usable reports whether the stats carry real radio readings. A running observer
|
|
// never reports zero uptime, and a live radio never reads a 0 dB noise floor, so
|
|
// that with zero airtime marks a status sent before the radio stats were filled in.
|
|
func (s statusStats) usable() bool {
|
|
if s.UptimeSeconds == 0 {
|
|
return false
|
|
}
|
|
return s.NoiseFloor != 0 || s.TxAirSecs != 0 || s.RxAirSecs != 0
|
|
}
|
|
|
|
// LoRa radios span 137 MHz sub-GHz to 2.5 GHz SX128x.
|
|
const (
|
|
minRadioFreqMHz = 100
|
|
maxRadioFreqMHz = 3000
|
|
)
|
|
|
|
// parseRadioFloat rejects NaN, Inf and anything outside [lo, hi]; observers send free text and
|
|
// one bad value would poison every airtime the observer is later costed with.
|
|
func parseRadioFloat(s string, lo, hi float64) (float64, error) {
|
|
f, err := strconv.ParseFloat(s, 32)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
if !(f >= lo && f <= hi) {
|
|
return 0, fmt.Errorf("radio value %q outside %g..%g", s, lo, hi)
|
|
}
|
|
return f, nil
|
|
}
|
|
|
|
// stripNULs removes NUL bytes, which Postgres text columns reject, and returns the
|
|
// names of the fields that had any.
|
|
func stripNULs(fields map[string]*string) []string {
|
|
var dirty []string
|
|
for name, s := range fields {
|
|
if strings.Contains(*s, "\x00") {
|
|
*s = strings.ReplaceAll(*s, "\x00", "")
|
|
dirty = append(dirty, name)
|
|
}
|
|
}
|
|
slices.Sort(dirty)
|
|
return dirty
|
|
}
|
|
|
|
// cleanText makes over-the-air text storable: invalid UTF-8 replaced, NULs dropped.
|
|
func cleanText(s string) string {
|
|
return strings.ReplaceAll(strings.ToValidUTF8(s, "�"), "\x00", "")
|
|
}
|
|
|
|
// stripJSONNULs drops \u0000 from a JSON document's strings, which jsonb rejects.
|
|
// Payloads without one are returned untouched.
|
|
func stripJSONNULs(raw []byte) ([]byte, bool) {
|
|
if !bytes.Contains(raw, []byte(`\u0000`)) {
|
|
return raw, false
|
|
}
|
|
dec := json.NewDecoder(bytes.NewReader(raw))
|
|
dec.UseNumber()
|
|
var v any
|
|
if err := dec.Decode(&v); err != nil {
|
|
return raw, false
|
|
}
|
|
out, err := json.Marshal(dropNULs(v))
|
|
if err != nil {
|
|
return raw, false
|
|
}
|
|
return out, true
|
|
}
|
|
|
|
func dropNULs(v any) any {
|
|
switch v := v.(type) {
|
|
case string:
|
|
return strings.ReplaceAll(v, "\x00", "")
|
|
case []any:
|
|
for i := range v {
|
|
v[i] = dropNULs(v[i])
|
|
}
|
|
case map[string]any:
|
|
out := make(map[string]any, len(v))
|
|
for k, e := range v {
|
|
out[strings.ReplaceAll(k, "\x00", "")] = dropNULs(e)
|
|
}
|
|
return out
|
|
}
|
|
return v
|
|
}
|
|
|
|
// handleStatus processes a /status message and fans out an observerStatus event.
|
|
func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte) {
|
|
var envelope struct {
|
|
ObserverType string `json:"source"`
|
|
SoftwareVersion string `json:"client_version"`
|
|
HardwareModel string `json:"model"`
|
|
FirmwareVersion string `json:"firmware_version"`
|
|
DisplayName string `json:"origin"`
|
|
RadioString string `json:"radio"`
|
|
Stats statusStats `json:"stats"`
|
|
}
|
|
if err := json.Unmarshal(raw, &envelope); err != nil {
|
|
w.log.Warn(fmt.Sprintf("malformed status envelope from %s", pubkeyHex), "error", err)
|
|
return
|
|
}
|
|
pubkey, err := hex.DecodeString(pubkeyHex)
|
|
if err != nil {
|
|
w.log.Warn(fmt.Sprintf("invalid pubkey hex in status from %s", pubkeyHex), "error", err)
|
|
return
|
|
}
|
|
dirty := stripNULs(map[string]*string{
|
|
"source": &envelope.ObserverType, "client_version": &envelope.SoftwareVersion,
|
|
"model": &envelope.HardwareModel, "firmware_version": &envelope.FirmwareVersion,
|
|
"origin": &envelope.DisplayName, "radio": &envelope.RadioString,
|
|
})
|
|
raw, rawDirty := stripJSONNULs(raw)
|
|
if len(dirty) > 0 || rawDirty {
|
|
w.log.Warn(fmt.Sprintf("stripped NUL bytes from status from %s", pubkeyHex), "fields", dirty, "metadata", rawDirty)
|
|
}
|
|
id, _, err := w.db.UpsertObserver(ctx, pubkey, "")
|
|
if err != nil {
|
|
w.log.Error(fmt.Sprintf("db: upsert observer failed in status from %s", pubkeyHex), "error", err)
|
|
return
|
|
}
|
|
// invalidate cache for observer details
|
|
if w.onObserverUpsert != nil {
|
|
w.onObserverUpsert(ctx, id)
|
|
}
|
|
if err := w.db.UpsertObserverBroker(ctx, id, w.cfg.BrokerName, false); err != nil {
|
|
w.log.Error(fmt.Sprintf("db: upsert observer broker failed in status from %s", pubkeyHex), "error", err)
|
|
}
|
|
params := UpdateObserverStatusParams{
|
|
PublicKey: pubkey,
|
|
StatusMetadata: raw,
|
|
LastStatusAt: time.Now(),
|
|
}
|
|
// A running observer never reports 0, so treat this the same as a missing
|
|
// stats object and leave the existing value alone (COALESCE in the SQL)
|
|
// rather than stomping it with a zero.
|
|
if envelope.Stats.UptimeSeconds != 0 {
|
|
params.UptimeSeconds = &envelope.Stats.UptimeSeconds
|
|
}
|
|
if envelope.Stats.BatteryMV != 0 {
|
|
batteryLevel := float32(envelope.Stats.BatteryMV) / 1000
|
|
params.BatteryLevel = &batteryLevel
|
|
}
|
|
if envelope.SoftwareVersion != "" {
|
|
params.SoftwareVersion = envelope.SoftwareVersion
|
|
}
|
|
params.ObserverType = inferObserverType(envelope.ObserverType, envelope.SoftwareVersion)
|
|
if envelope.DisplayName != "" {
|
|
params.DisplayName = strings.ToValidUTF8(envelope.DisplayName, "\uFFFD")
|
|
}
|
|
if envelope.HardwareModel != "" {
|
|
params.HardwareModel = envelope.HardwareModel
|
|
}
|
|
if envelope.FirmwareVersion != "" {
|
|
params.FirmwareVersion = envelope.FirmwareVersion
|
|
}
|
|
|
|
radio := strings.Split(strings.TrimSpace(envelope.RadioString), ",")
|
|
if len(radio) != 4 {
|
|
w.log.Warn(fmt.Sprintf("missing or malformed radio params in status from %s, skipping radio fields", pubkeyHex))
|
|
} else {
|
|
freq, err := parseRadioFloat(radio[0], minRadioFreqMHz, maxRadioFreqMHz)
|
|
if err != nil {
|
|
w.log.Warn(fmt.Sprintf("error parsing radio freq in status from %s", pubkeyHex), "error", err)
|
|
} else {
|
|
params.RadioFreqMHz = float32(freq)
|
|
}
|
|
bw, err := parseRadioFloat(radio[1], lora.MinBWKHz, lora.MaxBWKHz)
|
|
if err != nil {
|
|
w.log.Warn(fmt.Sprintf("error parsing radio bw in status from %s", pubkeyHex), "error", err)
|
|
} else {
|
|
params.RadioBWKHz = float32(bw)
|
|
}
|
|
sf, err := strconv.ParseInt(radio[2], 10, 16)
|
|
if err != nil {
|
|
w.log.Warn(fmt.Sprintf("error parsing radio sf in status from %s", pubkeyHex), "error", err)
|
|
} else {
|
|
params.RadioSF = int16(sf)
|
|
}
|
|
cr, err := strconv.ParseInt(radio[3], 10, 16)
|
|
if err != nil {
|
|
w.log.Warn(fmt.Sprintf("error parsing radio cr in status from %s", pubkeyHex), "error", err)
|
|
} else {
|
|
params.RadioCR = int16(cr)
|
|
}
|
|
}
|
|
|
|
observerID, err := w.db.UpdateObserverStatus(ctx, params)
|
|
if err != nil {
|
|
w.log.Error(fmt.Sprintf("db: update observer status failed for %s", pubkeyHex), "error", err)
|
|
return
|
|
}
|
|
// Store a telemetry snapshot at the configured resolution. Partial stats are
|
|
// skipped rather than written as zeros that would win the hourly dedup and
|
|
// pollute the telemetry aggregates.
|
|
if !envelope.Stats.usable() {
|
|
w.log.Debug("status has no usable stats; skipping telemetry insert")
|
|
} else {
|
|
resolution := w.cfg.TelemetryResolution
|
|
if resolution == 0 {
|
|
resolution = time.Hour
|
|
}
|
|
reportedAt := time.Now().Truncate(resolution)
|
|
batteryMV := int32(envelope.Stats.BatteryMV)
|
|
txAirSecs := float32(envelope.Stats.TxAirSecs)
|
|
rxAirSecs := float32(envelope.Stats.RxAirSecs)
|
|
queueLen := int32(envelope.Stats.QueueLen)
|
|
debugFlags := int32(envelope.Stats.DebugFlags)
|
|
recvErrors := int32(envelope.Stats.RecvErrors)
|
|
|
|
if err := w.db.InsertObserverTelemetry(
|
|
ctx, observerID, reportedAt, &batteryMV, &txAirSecs, &rxAirSecs,
|
|
envelope.Stats.NoiseFloor, envelope.Stats.UptimeSeconds,
|
|
&queueLen, &debugFlags, &recvErrors,
|
|
); err != nil {
|
|
w.log.Error(fmt.Sprintf("db: insert telemetry failed for %s", pubkeyHex), "error", err)
|
|
}
|
|
}
|
|
|
|
iata, err := w.db.GetObserverLastIATA(ctx, observerID)
|
|
if err != nil {
|
|
iata = "" // non-fatal, continue
|
|
}
|
|
scopes, err := w.db.GetObserverScopes(ctx, observerID)
|
|
if err != nil {
|
|
w.log.Error(fmt.Sprintf("failed to get observer scopes for %s", pubkeyHex), "error", err)
|
|
scopes = []string{}
|
|
}
|
|
var radioStr *string
|
|
if params.RadioFreqMHz != 0 {
|
|
s := fmt.Sprintf("%g,%g,%d", params.RadioFreqMHz, params.RadioBWKHz, params.RadioSF)
|
|
radioStr = &s
|
|
}
|
|
var observerType *string
|
|
if envelope.ObserverType != "" {
|
|
observerType = &envelope.ObserverType
|
|
}
|
|
evt := statusEvent{
|
|
ObserverID: observerID.String(),
|
|
DisplayName: envelope.DisplayName,
|
|
ObserverType: observerType,
|
|
IATA: iata,
|
|
Online: true,
|
|
Radio: radioStr,
|
|
Scopes: scopes,
|
|
BatteryMV: envelope.Stats.BatteryMV,
|
|
UptimeSeconds: envelope.Stats.UptimeSeconds,
|
|
LastStatusAt: time.Now().UnixMilli(),
|
|
}
|
|
payload, err := json.Marshal(evt)
|
|
if err != nil {
|
|
w.log.Error(fmt.Sprintf("failed to marshal status event payload for %s", pubkeyHex), "error", err)
|
|
return
|
|
}
|
|
w.hub.Broadcast(hub.Event{Type: hub.EventObserverStatus, Payload: payload, IATA: iata, ObserverID: evt.ObserverID})
|
|
}
|