dandri/beacon-server
1
0
Fork 0
mirror of https://github.com/MeshCore-Beacon/beacon-server.git synced 2026-09-01 16:48:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(ingest): status and nodeUpdate events on par with summary analogies

Browse Source 
observer status updates are now at data parity with
observer summary and node update events are at data
parity with node summaries

closes #38
This commit is contained in:
Enot (ded) Skelly
2026-06-08 12:17:16 -07:00
parent fef8c8d7b6
commit cace46b004
5 changed files with 77 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
db/observers.go
+5
View File
@@ -286,6 +286,11 @@ func (s *Store) GetObserverScopes(ctx context.Context, observerID uuid.UUID) ([]
return s.q.GetObserverScopes(ctx, observerID)
}
func (s *Store) IsObserverByPubkey(ctx context.Context, pubkey []byte) bool {
_, err := s.q.GetObserverByPubkey(ctx, pubkey)
return err == nil
}
func (s *Store) DeleteOldTelemetry(ctx context.Context, cutoff time.Time) error {
return s.q.DeleteOldTelemetry(ctx, pgtype.Timestamptz{Time: cutoff, Valid: true})
}
internal/ingest/ingest.go
+6
View File
@@ -119,6 +119,12 @@ type DB interface {
// GetObserverRadio returns the current radio settings for the given observer.
GetObserverRadio(ctx context.Context, observerID uuid.UUID) (RadioSettings, error)
// IsObserverByPubkey returns true if the given public key belongs to a known observer.
IsObserverByPubkey(ctx context.Context, pubkey []byte) bool
// GetObserverScopes returns the list of scope names associated with the given observer.
GetObserverScopes(ctx context.Context, observerID uuid.UUID) ([]string, error)
// ResolvePathHashes returns a list of node UUIDs for the given path hash prefixes and IATA.
ResolvePathHashes(ctx context.Context, iata string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error)
internal/ingest/packet.go
+1 -1
View File
@@ -631,7 +631,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
}
w.runCapabilityDetection(ctx, packet.PayloadType(), packet.PathHashSize(), resolvedIDs)
if inserted {
w.handlePayloadTypeSideEffects(ctx, packet, iata, packetHash[:], radio, scopeID)
w.handlePayloadTypeSideEffects(ctx, packet, iata, packetHash[:], radio, scopeID, matchedScope)
evt := packetObservationEvent{}
evt.PacketHash = hex.EncodeToString(packetHash[:])
evt.Packet.PayloadType = packet.PayloadType()
internal/ingest/side_effects.go
+37 -13
View File
@@ -8,6 +8,7 @@ import (
"strings"
"time"
"github.com/MeshCore-Beacon/beacon-server/internal/api"
"github.com/MeshCore-Beacon/beacon-server/internal/hub"
"github.com/MeshCore-Beacon/beacon-server/internal/keystore"
"github.com/meshcore-go/meshcore-go"
@@ -43,19 +44,25 @@ type channelMessageEvent struct {
// nodeUpdateEvent is the JSON payload for a nodeUpdate WS event.
type nodeUpdateEvent struct {
NodeID string `json:"nodeId"` // UUID string
Name string `json:"name"`
NodeType uint8 `json:"nodeType"`
IATA string `json:"iata"`
Lat *float64 `json:"lat,omitempty"`
Lng *float64 `json:"lng,omitempty"`
NodeID string `json:"nodeId"`
PublicKey string `json:"publicKey"`
Name string `json:"name"`
NodeType uint8 `json:"nodeType"`
NodeTypeName string `json:"nodeTypeName"`
IATA string `json:"iata"`
Lat *float64 `json:"lat,omitempty"`
Lng *float64 `json:"lng,omitempty"`
IsObserver bool `json:"isObserver"`
IATAs []api.NodeIATA `json:"iatas"`
DefaultScope *string `json:"defaultScope,omitempty"`
Radio *string `json:"radio,omitempty"`
}
// handlePayloadTypeSideEffects runs payload-type-specific processing after a
// new observation is confirmed inserted. Currently handles:
// - PayloadTypeAdvert (0x04): upsert node and node_iatas
// - PayloadTypeGrpTxt (0x05): decrypt and store channel message if key is known
func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshcore.Packet, iata string, packetHash []byte, radio RadioSettings, scopeID *int32) {
func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshcore.Packet, iata string, packetHash []byte, radio RadioSettings, scopeID *int32, matchedScope *string) {
if packet.PayloadType() == meshcore.PayloadTypeAdvert {
advert, err := meshcore.AdvertFromBytes(packet.Payload)
if err != nil {
@@ -112,13 +119,30 @@ func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshc
if err := w.db.UpsertNodeShortID(ctx, nodeID, iata, prefix4); err != nil {
log.Printf("ingest[%s]: failed to upsert node short ID for %s: %v", w.cfg.BrokerName, hex.EncodeToString(prefix4), err)
}
pubkeyHex := hex.EncodeToString(advert.PublicKey.PublicKeyBytes())
isObserver := w.db.IsObserverByPubkey(ctx, advert.PublicKey.PublicKeyBytes())
var defaultScope *string
if matchedScope != nil {
defaultScope = matchedScope
}
var radioStr *string
if radio.FreqMHz != 0 {
s := fmt.Sprintf("%.1f,%g,%d", radio.FreqMHz, radio.BWKHz, radio.SF)
radioStr = &s
}
evt := nodeUpdateEvent{
NodeID: nodeID.String(),
Name: advert.AppData().Name,
NodeType: advert.Type(),
IATA: iata,
Lat: lat,
Lng: lon,
NodeID: nodeID.String(),
PublicKey: pubkeyHex,
Name: advert.AppData().Name,
NodeType: advert.Type(),
NodeTypeName: api.NodeTypeName(int16(advert.Type())),
IATA: iata,
Lat: lat,
Lng: lon,
IsObserver: isObserver,
IATAs: []api.NodeIATA{{IATA: iata, LastHeard: time.Now().UnixMilli()}},
DefaultScope: defaultScope,
Radio: radioStr,
}
w.broadcast(hub.EventNodeUpdate, iata, meshcore.PayloadTypeAdvert, "", evt)
return
internal/ingest/status.go
+28 -7
View File
@@ -4,6 +4,7 @@ import (
"context"
"encoding/hex"
"encoding/json"
"fmt"
"log"
"strconv"
"strings"
@@ -34,13 +35,16 @@ type UpdateObserverStatusParams struct {
// 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"`
IATA string `json:"iata,omitempty"`
Online bool `json:"online"`
BatteryMV int `json:"batteryMv,omitempty"`
UptimeSeconds int64 `json:"uptimeSeconds"`
LastStatusAt int64 `json:"lastStatusAt"` // epoch ms
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"`
}
// handleStatus processes a /status message and fans out an observerStatus event.
@@ -172,11 +176,28 @@ func (w *Worker) handleStatus(ctx context.Context, pubkeyHex string, raw []byte)
if err != nil {
iata = "" // non-fatal, continue
}
scopes, err := w.db.GetObserverScopes(ctx, observerID)
if err != nil {
log.Printf("ingest[%s]: failed to get observer scopes for %s: %v", w.cfg.BrokerName, pubkeyHex, err)
scopes = []string{}
}
var radioStr *string
if params.RadioFreqMHz != 0 {
s := fmt.Sprintf("%.1f,%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(),