From cace46b004aebf636c4daf270ed7ec5e20fb61d7 Mon Sep 17 00:00:00 2001 From: "Enot (ded) Skelly" Date: Mon, 8 Jun 2026 12:16:55 -0700 Subject: [PATCH] feat(ingest): status and nodeUpdate events on par with summary analogies observer status updates are now at data parity with observer summary and node update events are at data parity with node summaries closes #38 --- db/observers.go | 5 ++++ internal/ingest/ingest.go | 6 ++++ internal/ingest/packet.go | 2 +- internal/ingest/side_effects.go | 50 ++++++++++++++++++++++++--------- internal/ingest/status.go | 35 ++++++++++++++++++----- 5 files changed, 77 insertions(+), 21 deletions(-) diff --git a/db/observers.go b/db/observers.go index 3d460da..efeee07 100644 --- a/db/observers.go +++ b/db/observers.go @@ -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}) } diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index 3d41fa7..4815d3d 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -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) diff --git a/internal/ingest/packet.go b/internal/ingest/packet.go index c45f1c1..b12c70f 100644 --- a/internal/ingest/packet.go +++ b/internal/ingest/packet.go @@ -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() diff --git a/internal/ingest/side_effects.go b/internal/ingest/side_effects.go index a744ccc..4ae3e90 100644 --- a/internal/ingest/side_effects.go +++ b/internal/ingest/side_effects.go @@ -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 diff --git a/internal/ingest/status.go b/internal/ingest/status.go index 24fe475..e9dacbf 100644 --- a/internal/ingest/status.go +++ b/internal/ingest/status.go @@ -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(),