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(),