From 4425e3ffa28bc9f7e814a2d0c9f2653a8a2f6f41 Mon Sep 17 00:00:00 2001 From: "Enot (ded) Skelly" Date: Thu, 4 Jun 2026 14:29:25 -0700 Subject: [PATCH] refactor ingest package --- internal/ingest/capability.go | 29 + internal/ingest/ingest.go | 990 +------------------------------- internal/ingest/packet.go | 642 +++++++++++++++++++++ internal/ingest/side_effects.go | 168 ++++++ internal/ingest/status.go | 190 ++++++ 5 files changed, 1037 insertions(+), 982 deletions(-) create mode 100644 internal/ingest/capability.go create mode 100644 internal/ingest/packet.go create mode 100644 internal/ingest/side_effects.go create mode 100644 internal/ingest/status.go diff --git a/internal/ingest/capability.go b/internal/ingest/capability.go new file mode 100644 index 0000000..d7065fc --- /dev/null +++ b/internal/ingest/capability.go @@ -0,0 +1,29 @@ +package ingest + +import ( + "context" + + "github.com/google/uuid" +) + +// runCapabilityDetection checks hash sizes and flips firmware capability flags. +// Called only when the observation INSERT succeeded (no dedup conflict). +// +// Rules (from design doc): +// - hash_size == 1: do nothing (proves nothing about firmware) +// - duplicate hash prefixes within the path: skip entirely +// - non-trace + hash_size 2 or 3 → supports_multibyte_paths = TRUE +// - trace (0x09) + hash_size 2 or 4 → supports_multibyte_traces = TRUE +func (w *Worker) runCapabilityDetection(ctx context.Context, payloadType uint8, hashSize uint8, resolvedNodeIDs []uuid.UUID) { + if hashSize < 2 { + return + } + for _, nodeID := range resolvedNodeIDs { + switch { + case payloadType != 0x09 && (hashSize == 2 || hashSize == 3): + _ = w.db.SetNodeCapability(ctx, nodeID, true, false) + case payloadType == 0x09 && (hashSize == 2 || hashSize == 4): + _ = w.db.SetNodeCapability(ctx, nodeID, false, true) + } + } +} diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index a3c3468..50963a3 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -9,9 +9,10 @@ // 2. Decode hex payload via meshcore-go PacketFromBytes // 3. Compute content-based packet hash (PacketHash) // 4. Upsert observers + observer_brokers + iata_codes -// 5. Upsert packets row (ON CONFLICT bump last_heard_at + observation_count) +// 5. Upsert packets row (ON CONFLICT bump last_heard_at) // 6. Insert packet_observations (ON CONFLICT DO NOTHING for cross-broker dedup) -// 7. If INSERT succeeded: capability detection, payload-type side effects, fan-out +// 7. Match transport codes against known scopes (TRANSPORT_FLOOD/DIRECT only) +// 8. If INSERT succeeded: capability detection, payload-type side effects, fan-out // // Pipeline per incoming /status message: // 1. Parse topic → extract publisher pubkey @@ -21,10 +22,7 @@ package ingest import ( "context" - "crypto/hmac" - "crypto/sha256" "encoding/binary" - "encoding/hex" "encoding/json" "fmt" "log" @@ -34,7 +32,6 @@ import ( mqtt "github.com/eclipse/paho.mqtt.golang" "github.com/google/uuid" - "github.com/meshcore-go/meshcore-go" "github.com/MeshCore-Tower/tower-server/internal/api" "github.com/MeshCore-Tower/tower-server/internal/hub" @@ -126,252 +123,19 @@ type DB interface { // hash-only row exists per channel hash. The return value is the channel ID // but can be safely ignored since unknown-key channels have no messages. UpsertChannelHashOnly(ctx context.Context, channelHash []byte) (int, error) + // GetPacketObservationCount returns the number of rows for the packet observations GetPacketObservationCount(ctx context.Context, packetHash []byte) (int64, error) + // GetTransportScopeByName returns the ID of a transport scope by its normalized name. GetTransportScopeByName(ctx context.Context, name string) (int32, error) + // UpsertObserverScope records or updates a scope association for an observer. // Called when a TRANSPORT_FLOOD packet is observed, linking the observer to // the matched regional transport scope. UpsertObserverScope(ctx context.Context, observerID uuid.UUID, scopeID int32) error } -// UpsertPacketParams mirrors the columns written on packets upsert. -type UpsertPacketParams struct { - PacketHash []byte - RouteType uint8 - PayloadType uint8 - PayloadVersion uint8 - TransportCodes []byte // nil if not FLOOD/DIRECT - RawHeader []byte - RawPayload []byte - ParsedPayload json.RawMessage - OriginPubkey []byte - ChannelHash []byte - ScopeID *int32 -} - -// InsertObservationParams mirrors the columns written on packet_observations insert. -type InsertObservationParams struct { - PacketHash []byte - ObserverID uuid.UUID - IATA string - HeardAt time.Time - PathLengthByte uint8 - HashSize uint8 - HopCount uint8 - PathBytes []byte - RSSI int16 - SNR float32 - PropagationTimeMs int32 - RadioFreqMHz float32 - SpreadFactor int16 - BandwidthKHz float32 - CodingRate int16 - SourceBroker string -} - -// UpsertNodeParams carries the fields extracted from a payload type 0x04 advert. -type UpsertNodeParams struct { - PublicKey []byte - Name string - NodeType uint8 // 1=companion, 2=repeater, 3=room server - Latitude *float64 - Longitude *float64 -} - -// InsertChannelMessageParams carries a decrypted group text message. -type InsertChannelMessageParams struct { - ChannelID int - PacketHash []byte - SenderName string - Content string - SentAt time.Time -} - -// 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 -} - -// RadioSettings holds the radio configuration for an observer, populated from -// /status messages and copied onto each observation row for RF analysis. -type RadioSettings struct { - FreqMHz float32 // MHz, e.g. 910.525 - SF int16 // LoRa spreading factor, e.g. 7 - BWKHz float32 // bandwidth in kHz, e.g. 62.5 - CR int16 // coding rate denominator, e.g. 5 means 4/5 -} - -// 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 -} - -// channelMessageEvent is the JSON payload for a channelMessage WS event. -type channelMessageEvent struct { - ChannelID int `json:"channelId"` - ChannelHash string `json:"channelHash"` // hex-encoded single byte - PacketHash string `json:"packetHash"` // hex-encoded - SenderName string `json:"senderName"` - Content string `json:"content"` - SentAt int64 `json:"sentAt"` // epoch ms -} - -// 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"` -} - -// packetObservationEvent is the JSON payload for a packetObservation WS event. -// Shape matches the design doc § Server → Client events. -type packetObservationEvent struct { - PacketHash string `json:"packetHash"` - Packet struct { - PayloadType uint8 `json:"payloadType"` - PayloadTypeName string `json:"payloadTypeName"` - RouteType uint8 `json:"routeType"` - RouteTypeName string `json:"routeTypeName"` - IsFirstObservation bool `json:"isFirstObservation"` - ObservationCount int64 `json:"observationCount"` - } `json:"packet"` - Observation struct { - ObserverID string `json:"observerId"` - ObserverName string `json:"observerName"` - IATA string `json:"iata"` - HeardAt int64 `json:"heardAt"` - RSSI int16 `json:"rssi"` - SNR float32 `json:"snr"` - SourceBroker string `json:"sourceBroker"` - PathBytes string `json:"pathBytes"` - PathLength struct { - Raw string `json:"raw"` - HashSize uint8 `json:"hashSize"` - HopCount uint8 `json:"hopCount"` - } `json:"pathLength"` - PropagationTimeMs int32 `json:"propagationTimeMs"` - } `json:"observation"` -} - -type parsedAnonReq struct { - Raw string `json:"raw"` - Type string `json:"type"` - Destination byte `json:"destination"` - EphemeralPubKey string `json:"ephemeralPubKey"` // hex -} - -type advertFlags struct { - Raw string `json:"raw"` - DeviceRole int `json:"deviceRole"` - DeviceRoleName string `json:"deviceRoleName"` - HasLocation bool `json:"hasLocation"` - HasName bool `json:"hasName"` - HasFeature1 bool `json:"hasFeature1"` - HasFeature2 bool `json:"hasFeature2"` -} - -type advertAppData struct { - Raw string `json:"raw"` - Flags advertFlags `json:"flags"` - Latitude *float64 `json:"latitude"` - Longitude *float64 `json:"longitude"` - Feature1 *uint16 `json:"feature1"` - Feature2 *uint16 `json:"feature2"` - Name *string `json:"name"` -} - -type parsedAdvert struct { - Type string `json:"type"` - Raw string `json:"raw"` - PublicKey string `json:"publicKey"` - Timestamp uint32 `json:"timestamp"` - Signature string `json:"signature"` - AppData advertAppData `json:"appData"` -} - -type parsedEnvelope struct { - Raw string `json:"raw"` - Type string `json:"type"` - DestinationHash string `json:"destinationHash"` - SourceHash string `json:"sourceHash"` - CipherMac string `json:"cipherMac"` - Ciphertext string `json:"ciphertext"` - CiphertextLength int `json:"ciphertextLength"` - Decrypted any `json:"decrypted"` -} - -type parsedGroupEnvelope struct { - Raw string `json:"raw"` - Type string `json:"type"` - ChannelHash string `json:"channelHash"` - CipherMac string `json:"cipherMac"` - Ciphertext string `json:"ciphertext"` - CiphertextLength int `json:"ciphertextLength"` - Decrypted any `json:"decrypted"` -} - -type parsedTrace struct { - Raw string `json:"raw"` - Type string `json:"type"` - TraceTag string `json:"traceTag"` - AuthCode uint32 `json:"authCode"` - Flags byte `json:"flags"` - PathHashes []string `json:"pathHashes"` - SNRValues []float32 `json:"snrValues"` -} - -type parsedAck struct { - Raw string `json:"raw"` - Type string `json:"type"` - Checksum string `json:"checksum"` -} - -type parsedMultipart struct { - Raw string `json:"raw"` - Type string `json:"type"` - Remaining uint8 `json:"remaining"` - WrappedType byte `json:"wrappedType"` - WrappedPayload string `json:"wrappedPayload"` -} - -type parsedControl struct { - Raw string `json:"raw"` - Type string `json:"type"` - Flags byte `json:"flags"` - Data string `json:"data"` -} - -type parsedRaw struct { - Type string `json:"type"` - Raw string `json:"raw"` -} - // ChannelKeyStore is a read-only view of the channel keys loaded from config. type ChannelKeyStore interface { // GetKey returns all known key entries for the given channel hash byte. @@ -462,6 +226,8 @@ func (w *Worker) subscribe(client mqtt.Client) { } // handleMessage dispatches incoming MQTT messages by subtopic. +// Each message is processed with a 30s timeout to prevent slow DB calls +// from blocking the MQTT receive goroutine indefinitely. func (w *Worker) handleMessage(msg mqtt.Message) { // Topic shape: meshcore/{IATA}/{pubkey}/{subtopic} parts := strings.SplitN(msg.Topic(), "/", 4) @@ -482,728 +248,6 @@ func (w *Worker) handleMessage(msg mqtt.Message) { } } -// handlePacket runs the full observation pipeline for a /packets message. -func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw []byte) { - var envelope struct { - Raw string `json:"raw"` // hex-encoded raw LoRa packet bytes - Timestamp string `json:"timestamp"` - Hash string `json:"hash"` - Origin string `json:"origin"` - Type string `json:"type"` - Direction string `json:"direction"` - Time string `json:"time"` - Date string `json:"date"` - Len string `json:"len"` - PacketType string `json:"packet_type"` - Route string `json:"route"` - PayloadLen string `json:"payload_len"` - OriginID string `json:"origin_id"` - SNR json.RawMessage `json:"SNR"` - RSSI json.RawMessage `json:"RSSI"` - } - err := json.Unmarshal(raw, &envelope) - if err != nil { - log.Printf("ingest[%s]: malformed packet envelope from %s/%s", w.cfg.BrokerName, iata, pubkeyHex) - return - } - if envelope.Raw == "" { - if envelope.Len == "0" || envelope.Len == "" { - return // observer keepalive with no packet data - } - log.Printf("ingest[%s]: malformed packet envelope from %s/%s", w.cfg.BrokerName, iata, pubkeyHex) - return - } - hexBytes, err := hex.DecodeString(strings.ReplaceAll(envelope.Raw, " ", "")) - if err != nil { - log.Printf("ingest[%s]: invalid hex from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) - return - } - packet, err := meshcore.PacketFromBytes(hexBytes) - if err != nil { - log.Printf("ingest[%s]: error decoding packet from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) - return - } - - pubkeyBytes, err := hex.DecodeString(pubkeyHex) - if err != nil { - log.Printf("ingest[%s]: invalid pubkey hex from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) - return - } - - id, observerName, err := w.db.UpsertObserver(ctx, pubkeyBytes) - if err != nil { - log.Printf("ingest[%s]: db: upsert observer failed with packet from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) - return - } - err = w.db.UpsertObserverBroker(ctx, id, w.cfg.BrokerName) - if err != nil { - log.Printf("ingest[%s]: db: update observer broker failed with packet from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) - return - } - err = w.db.UpsertIATA(ctx, iata) - if err != nil { - log.Printf("ingest[%s]: db: upsert IATA failed with packet from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) - return - } - packetHash := packet.PacketHash() - var transportCodes []byte - if packet.IsTransport() { - transportCodes = make([]byte, 4) - binary.LittleEndian.PutUint16(transportCodes[0:2], packet.TransportCode1) - binary.LittleEndian.PutUint16(transportCodes[2:4], packet.TransportCode2) - } - - // begin parse payloads - var channelHash []byte - originPubkey := []byte(nil) - var parsedPayload json.RawMessage - - switch packet.PayloadType() { - case meshcore.PayloadTypeGrpTxt: - grpTxt, err := meshcore.GroupTextFromBytes(packet.Payload) - if err == nil { - channelHash = []byte{grpTxt.ChannelHash} - pg := parsedGroupEnvelope{ - Raw: hex.EncodeToString(packet.Payload), - Type: "GROUP_TEXT", - ChannelHash: hex.EncodeToString([]byte{grpTxt.ChannelHash}), - CipherMac: hex.EncodeToString(grpTxt.MAC[:]), - Ciphertext: hex.EncodeToString(grpTxt.EncryptedPayload), - CiphertextLength: len(grpTxt.EncryptedPayload), - } - parsedPayload, _ = json.Marshal(pg) - } - - case meshcore.PayloadTypeGrpData: - grpData, err := meshcore.GroupDataFromBytes(packet.Payload) - if err == nil { - pg := parsedGroupEnvelope{ - Raw: hex.EncodeToString(packet.Payload), - Type: "GROUP_DATA", - ChannelHash: hex.EncodeToString([]byte{grpData.ChannelHash}), - CipherMac: hex.EncodeToString(grpData.MAC[:]), - Ciphertext: hex.EncodeToString(grpData.EncryptedPayload), - CiphertextLength: len(grpData.EncryptedPayload), - } - parsedPayload, _ = json.Marshal(pg) - } - - case meshcore.PayloadTypeAdvert: - advert, err := meshcore.AdvertFromBytes(packet.Payload) - if err == nil { - originPubkey = advert.PublicKey.PublicKeyBytes() - appData := advert.AppData() - flags := advert.Flags() - - hasLocation := flags&0x10 != 0 - hasFeature1 := flags&0x20 != 0 - hasFeature2 := flags&0x40 != 0 - hasName := flags&0x80 != 0 - - var lat, lon *float64 - if hasLocation { - la := float64(appData.Lat) / 1e6 - lo := float64(appData.Lon) / 1e6 - lat = &la - lon = &lo - } - - var feat1, feat2 *uint16 - if hasFeature1 { - feat1 = &appData.Feat1 - } - if hasFeature2 { - feat2 = &appData.Feat2 - } - - var name *string - if hasName { - n := strings.ToValidUTF8(appData.Name, "\uFFFD") - name = &n - } - - deviceRole := int(flags & 0x0F) - - pa := parsedAdvert{ - Type: "ADVERT", - Raw: hex.EncodeToString(packet.Payload), - PublicKey: hex.EncodeToString(advert.PublicKey.PublicKeyBytes()), - Timestamp: advert.Timestamp, - Signature: hex.EncodeToString(advert.Signature), - AppData: advertAppData{ - Raw: hex.EncodeToString(advert.RawAppData), - Flags: advertFlags{ - Raw: fmt.Sprintf("%02x", flags), - DeviceRole: deviceRole, - DeviceRoleName: appData.Type, - HasLocation: hasLocation, - HasName: hasName, - HasFeature1: hasFeature1, - HasFeature2: hasFeature2, - }, - Latitude: lat, - Longitude: lon, - Feature1: feat1, - Feature2: feat2, - Name: name, - }, - } - parsedPayload, _ = json.Marshal(pa) - } - - case meshcore.PayloadTypeAnonReq: - anonReq, err := meshcore.AnonReqFromBytes(packet.Payload) - if err == nil { - originPubkey = anonReq.EphemeralPubKey[:] - par := parsedAnonReq{ - Raw: hex.EncodeToString(packet.Payload), - Type: "ANON_REQUEST", - Destination: anonReq.Destination, - EphemeralPubKey: hex.EncodeToString(anonReq.EphemeralPubKey[:]), - } - parsedPayload, _ = json.Marshal(par) - } - - case meshcore.PayloadTypeReq: - req, err := meshcore.RequestFromBytes(packet.Payload) - if err == nil { - pe := parsedEnvelope{ - Raw: hex.EncodeToString(packet.Payload), - Type: "REQUEST", - DestinationHash: hex.EncodeToString([]byte{req.Destination}), - SourceHash: hex.EncodeToString([]byte{req.Source}), - CipherMac: hex.EncodeToString(req.MAC[:]), - Ciphertext: hex.EncodeToString(req.EncryptedPayload), - CiphertextLength: len(req.EncryptedPayload), - } - parsedPayload, _ = json.Marshal(pe) - } - - case meshcore.PayloadTypeResponse: - resp, err := meshcore.ResponseFromBytes(packet.Payload) - if err == nil { - pe := parsedEnvelope{ - Raw: hex.EncodeToString(packet.Payload), - Type: "RESPONSE", - DestinationHash: hex.EncodeToString([]byte{resp.Destination}), - SourceHash: hex.EncodeToString([]byte{resp.Source}), - CipherMac: hex.EncodeToString(resp.MAC[:]), - Ciphertext: hex.EncodeToString(resp.EncryptedPayload), - CiphertextLength: len(resp.EncryptedPayload), - } - parsedPayload, _ = json.Marshal(pe) - } - - case meshcore.PayloadTypeTxtMsg: - txt, err := meshcore.TextMessageFromBytes(packet.Payload) - if err == nil { - pe := parsedEnvelope{ - Raw: hex.EncodeToString(packet.Payload), - Type: "TEXT_MESSAGE", - DestinationHash: hex.EncodeToString([]byte{txt.Destination}), - SourceHash: hex.EncodeToString([]byte{txt.Source}), - CipherMac: hex.EncodeToString(txt.MAC[:]), - Ciphertext: hex.EncodeToString(txt.EncryptedPayload), - CiphertextLength: len(txt.EncryptedPayload), - } - parsedPayload, _ = json.Marshal(pe) - } - - case meshcore.PayloadTypePath: - path, err := meshcore.PathFromBytes(packet.Payload) - if err == nil { - pe := parsedEnvelope{ - Raw: hex.EncodeToString(packet.Payload), - Type: "PATH", - DestinationHash: hex.EncodeToString([]byte{path.Destination}), - SourceHash: hex.EncodeToString([]byte{path.Source}), - CipherMac: hex.EncodeToString(path.MAC[:]), - Ciphertext: hex.EncodeToString(path.EncryptedPayload), - CiphertextLength: len(path.EncryptedPayload), - } - parsedPayload, _ = json.Marshal(pe) - } - - case meshcore.PayloadTypeTrace: - trace, err := meshcore.TraceFromBytes(packet.Payload) - if err == nil { - hashSize := int(trace.PathHashSize()) - hashes := make([]string, 0) - for i := 0; i+hashSize <= len(trace.PathHashes); i += hashSize { - hashes = append(hashes, hex.EncodeToString(trace.PathHashes[i:i+hashSize])) - } - // SNR values are in packet.Path, one signed int8 per consumed hop - snrValues := make([]float32, 0, len(packet.Path)) - for _, b := range packet.Path { - snrValues = append(snrValues, float32(int8(b))/4.0) - } - pt := parsedTrace{ - Raw: hex.EncodeToString(packet.Payload), - Type: "TRACE", - TraceTag: hex.EncodeToString(uint32ToBytes(trace.Tag)), - AuthCode: trace.AuthCode, - Flags: trace.Flags, - PathHashes: hashes, - SNRValues: snrValues, - } - parsedPayload, _ = json.Marshal(pt) - } - - case meshcore.PayloadTypeAck: - ack, err := meshcore.AckFromBytes(packet.Payload) - if err == nil { - pa := parsedAck{ - Raw: hex.EncodeToString(packet.Payload), - Type: "ACK", - Checksum: hex.EncodeToString(uint32ToBytes(ack.AckCRC)), - } - parsedPayload, _ = json.Marshal(pa) - } - - case meshcore.PayloadTypeMultiPart: - mp, err := meshcore.MultiPartFromBytes(packet.Payload) - if err == nil { - pm := parsedMultipart{ - Raw: hex.EncodeToString(packet.Payload), - Type: "MULTIPART", - Remaining: mp.Remaining, - WrappedType: mp.WrappedType, - WrappedPayload: hex.EncodeToString(mp.WrappedPayload), - } - parsedPayload, _ = json.Marshal(pm) - } - - case meshcore.PayloadTypeControl: - ctrl, err := meshcore.ControlFromBytes(packet.Payload) - if err == nil { - pc := parsedControl{ - Raw: hex.EncodeToString(packet.Payload), - Type: "CONTROL", - Flags: ctrl.Flags, - Data: hex.EncodeToString(ctrl.Data), - } - parsedPayload, _ = json.Marshal(pc) - } - - default: - pr := parsedRaw{ - Type: "RAW", - Raw: hex.EncodeToString(packet.Payload), - } - parsedPayload, _ = json.Marshal(pr) - } - - var matchedScope *string - if packet.RouteType() == meshcore.RouteTypeTransportFlood || packet.RouteType() == meshcore.RouteTypeTransportDirect { - for _, entry := range w.scopes.Entries() { - code := computeTransportCode(entry.TransportKey, packet.PayloadType(), packet.Payload) - if code == packet.TransportCode1 { - s := entry.Name - matchedScope = &s - break - } - } - } - var scopeID *int32 - if matchedScope != nil { - id, err := w.db.GetTransportScopeByName(ctx, *matchedScope) - if err != nil { - log.Printf("ingest[%s]: failed to get scope ID for %s: %v", w.cfg.BrokerName, *matchedScope, err) - } else { - scopeID = &id - } - } - rawHeader := []byte{packet.Header} - if transportCodes != nil { - rawHeader = append(rawHeader, transportCodes...) - } - pParams := UpsertPacketParams{ - PacketHash: packetHash[:], - RouteType: packet.RouteType(), - PayloadType: packet.PayloadType(), - PayloadVersion: packet.PayloadVer(), - TransportCodes: transportCodes, - RawHeader: rawHeader, - RawPayload: packet.Payload, - ParsedPayload: parsedPayload, - OriginPubkey: originPubkey, - ChannelHash: channelHash, - ScopeID: scopeID, - } - isNew, err := w.db.UpsertPacket(ctx, pParams) - if err != nil { - log.Printf("ingest[%s]: db: upsert packet failed from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) - return - } - heardAt, err := time.Parse("2006-01-02T15:04:05.000000", envelope.Timestamp) - if err != nil { - heardAt, err = time.Parse("2006-01-02T15:04:05", envelope.Timestamp) - if err != nil { - log.Printf("ingest[%s]: error parsing time from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) - return - } - } - - radio, err := w.db.GetObserverRadio(ctx, id) - if err != nil { - log.Printf("ingest[%s]: db: get observer radio failed for %s: %v", w.cfg.BrokerName, pubkeyHex, err) - } - oParams := InsertObservationParams{ - PacketHash: packetHash[:], - ObserverID: id, - IATA: iata, - HeardAt: heardAt, - PathLengthByte: packet.PathLength, - HashSize: packet.PathHashSize(), - HopCount: packet.PathHashCount(), - PathBytes: packet.Path, - RSSI: int16(parseNumber(envelope.RSSI)), - SNR: float32(parseNumber(envelope.SNR)), - PropagationTimeMs: 0, // TODO: figure out how to calculate this - RadioFreqMHz: radio.FreqMHz, - SpreadFactor: radio.SF, - BandwidthKHz: radio.BWKHz, - CodingRate: radio.CR, - SourceBroker: w.cfg.BrokerName, - } - inserted, err := w.db.InsertObservation(ctx, oParams) - if err != nil { - log.Printf("ingest[%s]: db: insert observation failed from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) - return - } - - if scopeID != nil && inserted { - if err := w.db.UpsertObserverScope(ctx, id, *scopeID); err != nil { - log.Printf("ingest[%s]: failed to upsert observer scope for %s: %v", w.cfg.BrokerName, id, err) - } - } - - resolved, err := w.db.ResolvePathHashes(ctx, iata, packet.PathHashes()) - if err != nil { - log.Printf("ingest[%s]: path resolution failed: %v", w.cfg.BrokerName, err) - } - var resolvedIDs []uuid.UUID - for _, entries := range resolved { - for _, e := range entries { - resolvedIDs = append(resolvedIDs, e.NodeID) - } - } - w.runCapabilityDetection(ctx, packet.PayloadType(), packet.PathHashSize(), resolvedIDs) - if inserted { - w.handlePayloadTypeSideEffects(ctx, packet, iata, packetHash[:], radio, scopeID) - evt := packetObservationEvent{} - evt.PacketHash = hex.EncodeToString(packetHash[:]) - evt.Packet.PayloadType = packet.PayloadType() - evt.Packet.PayloadTypeName = packet.PayloadTypeString() - evt.Packet.RouteType = packet.RouteType() - evt.Packet.RouteTypeName = api.RouteTypeName(int16(packet.RouteType())) - evt.Packet.IsFirstObservation = isNew - evt.Observation.ObserverID = id.String() - evt.Observation.ObserverName = observerName - evt.Observation.IATA = iata - evt.Observation.HeardAt = heardAt.UnixMilli() - evt.Observation.RSSI = oParams.RSSI - evt.Observation.SNR = oParams.SNR - evt.Observation.SourceBroker = w.cfg.BrokerName - evt.Observation.PathBytes = hex.EncodeToString(packet.Path) - evt.Observation.PathLength.Raw = fmt.Sprintf("%02x", packet.PathLength) - evt.Observation.PathLength.HashSize = packet.PathHashSize() - evt.Observation.PathLength.HopCount = packet.PathHashCount() - evt.Observation.PropagationTimeMs = 0 // not yet calculated - count, err := w.db.GetPacketObservationCount(ctx, packetHash[:]) - if err != nil { - log.Printf("ingest[%s]: failed to get observation count: %v", w.cfg.BrokerName, err) - count = 0 - } - evt.Packet.ObservationCount = count - w.broadcast(hub.EventPacketObservation, iata, packet.PayloadType(), "", evt) - } -} - -// 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 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"` - } `json:"stats"` - } - if err := json.Unmarshal(raw, &envelope); err != nil { - log.Printf("ingest[%s]: malformed status envelope from %s: %v", w.cfg.BrokerName, pubkeyHex, err) - return - } - pubkey, err := hex.DecodeString(pubkeyHex) - if err != nil { - log.Printf("ingest[%s]: invalid pubkey hex in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err) - return - } - id, _, err := w.db.UpsertObserver(ctx, pubkey) - if err != nil { - log.Printf("ingest[%s]: db: upsert observer failed in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err) - return - } - if err := w.db.UpsertObserverBroker(ctx, id, w.cfg.BrokerName); err != nil { - log.Printf("ingest[%s]: db: upsert observer broker failed in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err) - } - params := UpdateObserverStatusParams{ - PublicKey: pubkey, - StatusMetadata: raw, - LastStatusAt: time.Now(), - } - 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 { - log.Printf("ingest[%s]: missing or malformed radio params in status from %s, skipping radio fields", w.cfg.BrokerName, pubkeyHex) - } else { - freq, err := strconv.ParseFloat(radio[0], 32) - if err != nil { - log.Printf("ingest[%s]: error parsing radio freq in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err) - } else { - params.RadioFreqMHz = float32(freq) - } - bw, err := strconv.ParseFloat(radio[1], 32) - if err != nil { - log.Printf("ingest[%s]: error parsing radio bw in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err) - } else { - params.RadioBWKHz = float32(bw) - } - sf, err := strconv.ParseInt(radio[2], 10, 16) - if err != nil { - log.Printf("ingest[%s]: error parsing radio sf in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err) - } else { - params.RadioSF = int16(sf) - } - cr, err := strconv.ParseInt(radio[3], 10, 16) - if err != nil { - log.Printf("ingest[%s]: error parsing radio cr in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err) - } else { - params.RadioCR = int16(cr) - } - } - - observerID, err := w.db.UpdateObserverStatus(ctx, params) - if err != nil { - log.Printf("ingest[%s]: db: update observer status failed for %s: %v", w.cfg.BrokerName, pubkeyHex, err) - return - } - // Store a telemetry snapshot at the configured resolution. - 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 { - log.Printf("ingest[%s]: db: insert telemetry failed for %s: %v", w.cfg.BrokerName, pubkeyHex, err) - } - - iata, err := w.db.GetObserverLastIATA(ctx, observerID) - if err != nil { - iata = "" // non-fatal, continue - } - evt := statusEvent{ - ObserverID: observerID.String(), - DisplayName: envelope.DisplayName, - IATA: iata, - Online: true, - BatteryMV: envelope.Stats.BatteryMV, - UptimeSeconds: envelope.Stats.UptimeSeconds, - LastStatusAt: time.Now().UnixMilli(), - } - payload, err := json.Marshal(evt) - if err != nil { - log.Printf("ingest[%s]: failed to marshal status event payload for %s: %v", w.cfg.BrokerName, pubkeyHex, err) - return - } - w.hub.Broadcast(hub.Event{Type: hub.EventObserverStatus, Payload: payload, IATA: iata}) -} - -// runCapabilityDetection checks hash sizes and flips firmware capability flags. -// Called only when the observation INSERT succeeded (no dedup conflict). -// -// Rules (from design doc): -// - hash_size == 1: do nothing (proves nothing about firmware) -// - duplicate hash prefixes within the path: skip entirely -// - non-trace + hash_size 2 or 3 → supports_multibyte_paths = TRUE -// - trace (0x09) + hash_size 2 or 4 → supports_multibyte_traces = TRUE -func (w *Worker) runCapabilityDetection(ctx context.Context, payloadType uint8, hashSize uint8, resolvedNodeIDs []uuid.UUID) { - if hashSize < 2 { - return - } - for _, nodeID := range resolvedNodeIDs { - switch { - case payloadType != 0x09 && (hashSize == 2 || hashSize == 3): - _ = w.db.SetNodeCapability(ctx, nodeID, true, false) - case payloadType == 0x09 && (hashSize == 2 || hashSize == 4): - _ = w.db.SetNodeCapability(ctx, nodeID, false, true) - } - } -} - -// 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) { - if packet.PayloadType() == meshcore.PayloadTypeAdvert { - advert, err := meshcore.AdvertFromBytes(packet.Payload) - if err != nil { - log.Printf("ingest[%s]: error decoding advert payload: %v", w.cfg.BrokerName, err) - return - } - var lat, lon *float64 - if advert.AppData().Lat != 0 || advert.AppData().Lon != 0 { - la := float64(advert.AppData().Lat) / 1e6 - lo := float64(advert.AppData().Lon) / 1e6 - lat = &la - lon = &lo - } - params := UpsertNodeParams{ - PublicKey: advert.PublicKey.PublicKeyBytes(), - Name: strings.ToValidUTF8(advert.AppData().Name, "\uFFFD"), - NodeType: advert.Type(), - Latitude: lat, - Longitude: lon, - } - nodeID, err := w.db.UpsertNode(ctx, params, radio) - if err != nil { - log.Printf("ingest[%s]: db: upsert node failed: %v", w.cfg.BrokerName, err) - return - } - if err := w.db.UpsertNodeIATA(ctx, nodeID, iata); err != nil { - log.Printf("ingest[%s]: db: upsert node IATA failed: %v", w.cfg.BrokerName, err) - } - if scopeID != nil && (packet.RouteType() == meshcore.RouteTypeTransportFlood || packet.RouteType() == meshcore.RouteTypeTransportDirect) { - if err := w.db.SetNodeDefaultScope(ctx, nodeID, *scopeID); err != nil { - log.Printf("ingest[%s]: failed to set default scope for node %s: %v", w.cfg.BrokerName, hex.EncodeToString(advert.PublicKey.PublicKeyBytes()), err) - } - } - prefix4 := advert.PublicKey.PublicKeyBytes()[:4] - 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) - } - evt := nodeUpdateEvent{ - NodeID: nodeID.String(), - Name: advert.AppData().Name, - NodeType: advert.Type(), - IATA: iata, - Lat: lat, - Lng: lon, - } - w.broadcast(hub.EventNodeUpdate, iata, meshcore.PayloadTypeAdvert, "", evt) - return - } - if packet.PayloadType() == meshcore.PayloadTypeGrpTxt { - grpTxt, err := meshcore.GroupTextFromBytes(packet.Payload) - if err != nil { - log.Printf("ingest[%s]: error decoding group text payload: %v", w.cfg.BrokerName, err) - return - } - channelHashBytes := []byte{grpTxt.ChannelHash} - - // Always upsert a hash-only row so unknown channels are recorded. - _, _ = w.db.UpsertChannelHashOnly(ctx, channelHashBytes) - - // Try each known key entry for this hash. - entries := w.keys.GetKey(channelHashBytes) - if len(entries) == 0 { - return // channel key unknown; message stored as encrypted blob only - } - var payload *meshcore.GroupTextPayload - var usedEntry keystore.Entry - for _, entry := range entries { - if p, err := grpTxt.DecryptStruct(entry.Key); err == nil { - payload = p - usedEntry = entry - break - } - } - if payload == nil { - return // none of the keys worked - } - - // Upsert the keyed channel row — messages are associated with this row. - channelID, err := w.db.UpsertChannel(ctx, channelHashBytes, usedEntry.Fingerprint, usedEntry.Name, usedEntry.Hashtag) - if err != nil { - log.Printf("ingest[%s]: db: upsert keyed channel failed: %v", w.cfg.BrokerName, err) - return - } - params := InsertChannelMessageParams{ - ChannelID: channelID, - PacketHash: packetHash[:], - SenderName: strings.ReplaceAll(strings.ToValidUTF8(payload.Sender, "\uFFFD"), "\x00", ""), - SentAt: time.Unix(int64(payload.Timestamp), 0), - Content: strings.ReplaceAll(strings.ToValidUTF8(payload.Text, "\uFFFD"), "\x00", ""), - } - newMsg, err := w.db.InsertChannelMessage(ctx, params) - if err != nil { - log.Printf("ingest[%s]: db: insert channel message failed: %v", w.cfg.BrokerName, err) - return - } - - if newMsg { - evt := channelMessageEvent{ - ChannelID: channelID, - ChannelHash: hex.EncodeToString(channelHashBytes), - PacketHash: hex.EncodeToString(packetHash), - SenderName: strings.ReplaceAll(strings.ToValidUTF8(payload.Sender, "\uFFFD"), "\x00", ""), - Content: strings.ReplaceAll(strings.ToValidUTF8(payload.Text, "\uFFFD"), "\x00", ""), - SentAt: time.Unix(int64(payload.Timestamp), 0).UnixMilli(), - } - w.broadcast(hub.EventChannelMessage, iata, 0, fmt.Sprintf("%02x", grpTxt.ChannelHash), evt) - } - return - } -} - func (w *Worker) broadcast(eventType hub.EventType, iata string, payloadType uint8, channelHash string, payload any) { b, err := json.Marshal(payload) if err != nil { @@ -1278,21 +322,3 @@ func uint32ToBytes(v uint32) []byte { binary.LittleEndian.PutUint32(b, v) return b } - -// computeTransportCode derives transport_code_1 from a transport key and packet payload. -// code = HMAC-SHA256(key, payload_type_byte || payload)[0:2] as little-endian uint16. -// Coerces reserved values 0x0000 → 0x0001 and 0xFFFF → 0xFFFE per §2.4. -func computeTransportCode(key []byte, payloadType uint8, payload []byte) uint16 { - mac := hmac.New(sha256.New, key) - mac.Write([]byte{payloadType}) - mac.Write(payload) - sum := mac.Sum(nil) - code := uint16(sum[0]) | uint16(sum[1])<<8 - if code == 0x0000 { - code = 0x0001 - } - if code == 0xFFFF { - code = 0xFFFE - } - return code -} diff --git a/internal/ingest/packet.go b/internal/ingest/packet.go new file mode 100644 index 0000000..a8505b8 --- /dev/null +++ b/internal/ingest/packet.go @@ -0,0 +1,642 @@ +package ingest + +import ( + "context" + "crypto/hmac" + "crypto/sha256" + "encoding/binary" + "encoding/hex" + "encoding/json" + "fmt" + "log" + "strings" + "time" + + "github.com/MeshCore-Tower/tower-server/internal/api" + "github.com/MeshCore-Tower/tower-server/internal/hub" + "github.com/google/uuid" + "github.com/meshcore-go/meshcore-go" +) + +// UpsertPacketParams mirrors the columns written on packets upsert. +type UpsertPacketParams struct { + PacketHash []byte + RouteType uint8 + PayloadType uint8 + PayloadVersion uint8 + TransportCodes []byte // nil if not FLOOD/DIRECT + RawHeader []byte + RawPayload []byte + ParsedPayload json.RawMessage + OriginPubkey []byte + ChannelHash []byte + ScopeID *int32 +} + +// InsertObservationParams mirrors the columns written on packet_observations insert. +type InsertObservationParams struct { + PacketHash []byte + ObserverID uuid.UUID + IATA string + HeardAt time.Time + PathLengthByte uint8 + HashSize uint8 + HopCount uint8 + PathBytes []byte + RSSI int16 + SNR float32 + PropagationTimeMs int32 + RadioFreqMHz float32 + SpreadFactor int16 + BandwidthKHz float32 + CodingRate int16 + SourceBroker string +} + +// RadioSettings holds the radio configuration for an observer, populated from +// /status messages and copied onto each observation row for RF analysis. +type RadioSettings struct { + FreqMHz float32 // MHz, e.g. 910.525 + SF int16 // LoRa spreading factor, e.g. 7 + BWKHz float32 // bandwidth in kHz, e.g. 62.5 + CR int16 // coding rate denominator, e.g. 5 means 4/5 +} + +// packetObservationEvent is the JSON payload for a packetObservation WS event. +// Shape matches the design doc § Server → Client events. +type packetObservationEvent struct { + PacketHash string `json:"packetHash"` + Packet struct { + PayloadType uint8 `json:"payloadType"` + PayloadTypeName string `json:"payloadTypeName"` + RouteType uint8 `json:"routeType"` + RouteTypeName string `json:"routeTypeName"` + IsFirstObservation bool `json:"isFirstObservation"` + ObservationCount int64 `json:"observationCount"` + } `json:"packet"` + Observation struct { + ObserverID string `json:"observerId"` + ObserverName string `json:"observerName"` + IATA string `json:"iata"` + HeardAt int64 `json:"heardAt"` + RSSI int16 `json:"rssi"` + SNR float32 `json:"snr"` + SourceBroker string `json:"sourceBroker"` + PathBytes string `json:"pathBytes"` + PathLength struct { + Raw string `json:"raw"` + HashSize uint8 `json:"hashSize"` + HopCount uint8 `json:"hopCount"` + } `json:"pathLength"` + PropagationTimeMs int32 `json:"propagationTimeMs"` + } `json:"observation"` +} + +type parsedAnonReq struct { + Raw string `json:"raw"` + Type string `json:"type"` + Destination byte `json:"destination"` + EphemeralPubKey string `json:"ephemeralPubKey"` // hex +} + +type advertFlags struct { + Raw string `json:"raw"` + DeviceRole int `json:"deviceRole"` + DeviceRoleName string `json:"deviceRoleName"` + HasLocation bool `json:"hasLocation"` + HasName bool `json:"hasName"` + HasFeature1 bool `json:"hasFeature1"` + HasFeature2 bool `json:"hasFeature2"` +} + +type advertAppData struct { + Raw string `json:"raw"` + Flags advertFlags `json:"flags"` + Latitude *float64 `json:"latitude"` + Longitude *float64 `json:"longitude"` + Feature1 *uint16 `json:"feature1"` + Feature2 *uint16 `json:"feature2"` + Name *string `json:"name"` +} + +type parsedAdvert struct { + Type string `json:"type"` + Raw string `json:"raw"` + PublicKey string `json:"publicKey"` + Timestamp uint32 `json:"timestamp"` + Signature string `json:"signature"` + AppData advertAppData `json:"appData"` +} + +type parsedEnvelope struct { + Raw string `json:"raw"` + Type string `json:"type"` + DestinationHash string `json:"destinationHash"` + SourceHash string `json:"sourceHash"` + CipherMac string `json:"cipherMac"` + Ciphertext string `json:"ciphertext"` + CiphertextLength int `json:"ciphertextLength"` + Decrypted any `json:"decrypted"` +} + +type parsedGroupEnvelope struct { + Raw string `json:"raw"` + Type string `json:"type"` + ChannelHash string `json:"channelHash"` + CipherMac string `json:"cipherMac"` + Ciphertext string `json:"ciphertext"` + CiphertextLength int `json:"ciphertextLength"` + Decrypted any `json:"decrypted"` +} + +type parsedTrace struct { + Raw string `json:"raw"` + Type string `json:"type"` + TraceTag string `json:"traceTag"` + AuthCode uint32 `json:"authCode"` + Flags byte `json:"flags"` + PathHashes []string `json:"pathHashes"` + SNRValues []float32 `json:"snrValues"` +} + +type parsedAck struct { + Raw string `json:"raw"` + Type string `json:"type"` + Checksum string `json:"checksum"` +} + +type parsedMultipart struct { + Raw string `json:"raw"` + Type string `json:"type"` + Remaining uint8 `json:"remaining"` + WrappedType byte `json:"wrappedType"` + WrappedPayload string `json:"wrappedPayload"` +} + +type parsedControl struct { + Raw string `json:"raw"` + Type string `json:"type"` + Flags byte `json:"flags"` + Data string `json:"data"` +} + +type parsedRaw struct { + Type string `json:"type"` + Raw string `json:"raw"` +} + +// handlePacket runs the full observation pipeline for a /packets message. +func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw []byte) { + var envelope struct { + Raw string `json:"raw"` // hex-encoded raw LoRa packet bytes + Timestamp string `json:"timestamp"` + Hash string `json:"hash"` + Origin string `json:"origin"` + Type string `json:"type"` + Direction string `json:"direction"` + Time string `json:"time"` + Date string `json:"date"` + Len string `json:"len"` + PacketType string `json:"packet_type"` + Route string `json:"route"` + PayloadLen string `json:"payload_len"` + OriginID string `json:"origin_id"` + SNR json.RawMessage `json:"SNR"` + RSSI json.RawMessage `json:"RSSI"` + } + err := json.Unmarshal(raw, &envelope) + if err != nil { + log.Printf("ingest[%s]: malformed packet envelope from %s/%s", w.cfg.BrokerName, iata, pubkeyHex) + return + } + if envelope.Raw == "" { + if envelope.Len == "0" || envelope.Len == "" { + return // observer keepalive with no packet data + } + log.Printf("ingest[%s]: malformed packet envelope from %s/%s", w.cfg.BrokerName, iata, pubkeyHex) + return + } + hexBytes, err := hex.DecodeString(strings.ReplaceAll(envelope.Raw, " ", "")) + if err != nil { + log.Printf("ingest[%s]: invalid hex from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) + return + } + packet, err := meshcore.PacketFromBytes(hexBytes) + if err != nil { + log.Printf("ingest[%s]: error decoding packet from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) + return + } + + pubkeyBytes, err := hex.DecodeString(pubkeyHex) + if err != nil { + log.Printf("ingest[%s]: invalid pubkey hex from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) + return + } + + id, observerName, err := w.db.UpsertObserver(ctx, pubkeyBytes) + if err != nil { + log.Printf("ingest[%s]: db: upsert observer failed with packet from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) + return + } + err = w.db.UpsertObserverBroker(ctx, id, w.cfg.BrokerName) + if err != nil { + log.Printf("ingest[%s]: db: update observer broker failed with packet from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) + return + } + err = w.db.UpsertIATA(ctx, iata) + if err != nil { + log.Printf("ingest[%s]: db: upsert IATA failed with packet from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) + return + } + packetHash := packet.PacketHash() + var transportCodes []byte + if packet.IsTransport() { + transportCodes = make([]byte, 4) + binary.LittleEndian.PutUint16(transportCodes[0:2], packet.TransportCode1) + binary.LittleEndian.PutUint16(transportCodes[2:4], packet.TransportCode2) + } + + // begin parse payloads + var channelHash []byte + originPubkey := []byte(nil) + var parsedPayload json.RawMessage + + switch packet.PayloadType() { + case meshcore.PayloadTypeGrpTxt: + grpTxt, err := meshcore.GroupTextFromBytes(packet.Payload) + if err == nil { + channelHash = []byte{grpTxt.ChannelHash} + pg := parsedGroupEnvelope{ + Raw: hex.EncodeToString(packet.Payload), + Type: "GROUP_TEXT", + ChannelHash: hex.EncodeToString([]byte{grpTxt.ChannelHash}), + CipherMac: hex.EncodeToString(grpTxt.MAC[:]), + Ciphertext: hex.EncodeToString(grpTxt.EncryptedPayload), + CiphertextLength: len(grpTxt.EncryptedPayload), + } + parsedPayload, _ = json.Marshal(pg) + } + + case meshcore.PayloadTypeGrpData: + grpData, err := meshcore.GroupDataFromBytes(packet.Payload) + if err == nil { + pg := parsedGroupEnvelope{ + Raw: hex.EncodeToString(packet.Payload), + Type: "GROUP_DATA", + ChannelHash: hex.EncodeToString([]byte{grpData.ChannelHash}), + CipherMac: hex.EncodeToString(grpData.MAC[:]), + Ciphertext: hex.EncodeToString(grpData.EncryptedPayload), + CiphertextLength: len(grpData.EncryptedPayload), + } + parsedPayload, _ = json.Marshal(pg) + } + + case meshcore.PayloadTypeAdvert: + advert, err := meshcore.AdvertFromBytes(packet.Payload) + if err == nil { + originPubkey = advert.PublicKey.PublicKeyBytes() + appData := advert.AppData() + flags := advert.Flags() + + hasLocation := flags&0x10 != 0 + hasFeature1 := flags&0x20 != 0 + hasFeature2 := flags&0x40 != 0 + hasName := flags&0x80 != 0 + + var lat, lon *float64 + if hasLocation { + la := float64(appData.Lat) / 1e6 + lo := float64(appData.Lon) / 1e6 + lat = &la + lon = &lo + } + + var feat1, feat2 *uint16 + if hasFeature1 { + feat1 = &appData.Feat1 + } + if hasFeature2 { + feat2 = &appData.Feat2 + } + + var name *string + if hasName { + n := strings.ToValidUTF8(appData.Name, "\uFFFD") + name = &n + } + + deviceRole := int(flags & 0x0F) + + pa := parsedAdvert{ + Type: "ADVERT", + Raw: hex.EncodeToString(packet.Payload), + PublicKey: hex.EncodeToString(advert.PublicKey.PublicKeyBytes()), + Timestamp: advert.Timestamp, + Signature: hex.EncodeToString(advert.Signature), + AppData: advertAppData{ + Raw: hex.EncodeToString(advert.RawAppData), + Flags: advertFlags{ + Raw: fmt.Sprintf("%02x", flags), + DeviceRole: deviceRole, + DeviceRoleName: appData.Type, + HasLocation: hasLocation, + HasName: hasName, + HasFeature1: hasFeature1, + HasFeature2: hasFeature2, + }, + Latitude: lat, + Longitude: lon, + Feature1: feat1, + Feature2: feat2, + Name: name, + }, + } + parsedPayload, _ = json.Marshal(pa) + } + + case meshcore.PayloadTypeAnonReq: + anonReq, err := meshcore.AnonReqFromBytes(packet.Payload) + if err == nil { + originPubkey = anonReq.EphemeralPubKey[:] + par := parsedAnonReq{ + Raw: hex.EncodeToString(packet.Payload), + Type: "ANON_REQUEST", + Destination: anonReq.Destination, + EphemeralPubKey: hex.EncodeToString(anonReq.EphemeralPubKey[:]), + } + parsedPayload, _ = json.Marshal(par) + } + + case meshcore.PayloadTypeReq: + req, err := meshcore.RequestFromBytes(packet.Payload) + if err == nil { + pe := parsedEnvelope{ + Raw: hex.EncodeToString(packet.Payload), + Type: "REQUEST", + DestinationHash: hex.EncodeToString([]byte{req.Destination}), + SourceHash: hex.EncodeToString([]byte{req.Source}), + CipherMac: hex.EncodeToString(req.MAC[:]), + Ciphertext: hex.EncodeToString(req.EncryptedPayload), + CiphertextLength: len(req.EncryptedPayload), + } + parsedPayload, _ = json.Marshal(pe) + } + + case meshcore.PayloadTypeResponse: + resp, err := meshcore.ResponseFromBytes(packet.Payload) + if err == nil { + pe := parsedEnvelope{ + Raw: hex.EncodeToString(packet.Payload), + Type: "RESPONSE", + DestinationHash: hex.EncodeToString([]byte{resp.Destination}), + SourceHash: hex.EncodeToString([]byte{resp.Source}), + CipherMac: hex.EncodeToString(resp.MAC[:]), + Ciphertext: hex.EncodeToString(resp.EncryptedPayload), + CiphertextLength: len(resp.EncryptedPayload), + } + parsedPayload, _ = json.Marshal(pe) + } + + case meshcore.PayloadTypeTxtMsg: + txt, err := meshcore.TextMessageFromBytes(packet.Payload) + if err == nil { + pe := parsedEnvelope{ + Raw: hex.EncodeToString(packet.Payload), + Type: "TEXT_MESSAGE", + DestinationHash: hex.EncodeToString([]byte{txt.Destination}), + SourceHash: hex.EncodeToString([]byte{txt.Source}), + CipherMac: hex.EncodeToString(txt.MAC[:]), + Ciphertext: hex.EncodeToString(txt.EncryptedPayload), + CiphertextLength: len(txt.EncryptedPayload), + } + parsedPayload, _ = json.Marshal(pe) + } + + case meshcore.PayloadTypePath: + path, err := meshcore.PathFromBytes(packet.Payload) + if err == nil { + pe := parsedEnvelope{ + Raw: hex.EncodeToString(packet.Payload), + Type: "PATH", + DestinationHash: hex.EncodeToString([]byte{path.Destination}), + SourceHash: hex.EncodeToString([]byte{path.Source}), + CipherMac: hex.EncodeToString(path.MAC[:]), + Ciphertext: hex.EncodeToString(path.EncryptedPayload), + CiphertextLength: len(path.EncryptedPayload), + } + parsedPayload, _ = json.Marshal(pe) + } + + case meshcore.PayloadTypeTrace: + trace, err := meshcore.TraceFromBytes(packet.Payload) + if err == nil { + hashSize := int(trace.PathHashSize()) + hashes := make([]string, 0) + for i := 0; i+hashSize <= len(trace.PathHashes); i += hashSize { + hashes = append(hashes, hex.EncodeToString(trace.PathHashes[i:i+hashSize])) + } + // SNR values are in packet.Path, one signed int8 per consumed hop + snrValues := make([]float32, 0, len(packet.Path)) + for _, b := range packet.Path { + snrValues = append(snrValues, float32(int8(b))/4.0) + } + pt := parsedTrace{ + Raw: hex.EncodeToString(packet.Payload), + Type: "TRACE", + TraceTag: hex.EncodeToString(uint32ToBytes(trace.Tag)), + AuthCode: trace.AuthCode, + Flags: trace.Flags, + PathHashes: hashes, + SNRValues: snrValues, + } + parsedPayload, _ = json.Marshal(pt) + } + + case meshcore.PayloadTypeAck: + ack, err := meshcore.AckFromBytes(packet.Payload) + if err == nil { + pa := parsedAck{ + Raw: hex.EncodeToString(packet.Payload), + Type: "ACK", + Checksum: hex.EncodeToString(uint32ToBytes(ack.AckCRC)), + } + parsedPayload, _ = json.Marshal(pa) + } + + case meshcore.PayloadTypeMultiPart: + mp, err := meshcore.MultiPartFromBytes(packet.Payload) + if err == nil { + pm := parsedMultipart{ + Raw: hex.EncodeToString(packet.Payload), + Type: "MULTIPART", + Remaining: mp.Remaining, + WrappedType: mp.WrappedType, + WrappedPayload: hex.EncodeToString(mp.WrappedPayload), + } + parsedPayload, _ = json.Marshal(pm) + } + + case meshcore.PayloadTypeControl: + ctrl, err := meshcore.ControlFromBytes(packet.Payload) + if err == nil { + pc := parsedControl{ + Raw: hex.EncodeToString(packet.Payload), + Type: "CONTROL", + Flags: ctrl.Flags, + Data: hex.EncodeToString(ctrl.Data), + } + parsedPayload, _ = json.Marshal(pc) + } + + default: + pr := parsedRaw{ + Type: "RAW", + Raw: hex.EncodeToString(packet.Payload), + } + parsedPayload, _ = json.Marshal(pr) + } + + var matchedScope *string + if packet.RouteType() == meshcore.RouteTypeTransportFlood || packet.RouteType() == meshcore.RouteTypeTransportDirect { + for _, entry := range w.scopes.Entries() { + code := computeTransportCode(entry.TransportKey, packet.PayloadType(), packet.Payload) + if code == packet.TransportCode1 { + s := entry.Name + matchedScope = &s + break + } + } + } + var scopeID *int32 + if matchedScope != nil { + id, err := w.db.GetTransportScopeByName(ctx, *matchedScope) + if err != nil { + log.Printf("ingest[%s]: failed to get scope ID for %s: %v", w.cfg.BrokerName, *matchedScope, err) + } else { + scopeID = &id + } + } + rawHeader := []byte{packet.Header} + if transportCodes != nil { + rawHeader = append(rawHeader, transportCodes...) + } + pParams := UpsertPacketParams{ + PacketHash: packetHash[:], + RouteType: packet.RouteType(), + PayloadType: packet.PayloadType(), + PayloadVersion: packet.PayloadVer(), + TransportCodes: transportCodes, + RawHeader: rawHeader, + RawPayload: packet.Payload, + ParsedPayload: parsedPayload, + OriginPubkey: originPubkey, + ChannelHash: channelHash, + ScopeID: scopeID, + } + isNew, err := w.db.UpsertPacket(ctx, pParams) + if err != nil { + log.Printf("ingest[%s]: db: upsert packet failed from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) + return + } + heardAt, err := time.Parse("2006-01-02T15:04:05.000000", envelope.Timestamp) + if err != nil { + heardAt, err = time.Parse("2006-01-02T15:04:05", envelope.Timestamp) + if err != nil { + log.Printf("ingest[%s]: error parsing time from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) + return + } + } + + radio, err := w.db.GetObserverRadio(ctx, id) + if err != nil { + log.Printf("ingest[%s]: db: get observer radio failed for %s: %v", w.cfg.BrokerName, pubkeyHex, err) + } + oParams := InsertObservationParams{ + PacketHash: packetHash[:], + ObserverID: id, + IATA: iata, + HeardAt: heardAt, + PathLengthByte: packet.PathLength, + HashSize: packet.PathHashSize(), + HopCount: packet.PathHashCount(), + PathBytes: packet.Path, + RSSI: int16(parseNumber(envelope.RSSI)), + SNR: float32(parseNumber(envelope.SNR)), + PropagationTimeMs: 0, // TODO: figure out how to calculate this + RadioFreqMHz: radio.FreqMHz, + SpreadFactor: radio.SF, + BandwidthKHz: radio.BWKHz, + CodingRate: radio.CR, + SourceBroker: w.cfg.BrokerName, + } + inserted, err := w.db.InsertObservation(ctx, oParams) + if err != nil { + log.Printf("ingest[%s]: db: insert observation failed from %s/%s: %v", w.cfg.BrokerName, iata, pubkeyHex, err) + return + } + + if scopeID != nil && inserted { + if err := w.db.UpsertObserverScope(ctx, id, *scopeID); err != nil { + log.Printf("ingest[%s]: failed to upsert observer scope for %s: %v", w.cfg.BrokerName, id, err) + } + } + + resolved, err := w.db.ResolvePathHashes(ctx, iata, packet.PathHashes()) + if err != nil { + log.Printf("ingest[%s]: path resolution failed: %v", w.cfg.BrokerName, err) + } + var resolvedIDs []uuid.UUID + for _, entries := range resolved { + for _, e := range entries { + resolvedIDs = append(resolvedIDs, e.NodeID) + } + } + w.runCapabilityDetection(ctx, packet.PayloadType(), packet.PathHashSize(), resolvedIDs) + if inserted { + w.handlePayloadTypeSideEffects(ctx, packet, iata, packetHash[:], radio, scopeID) + evt := packetObservationEvent{} + evt.PacketHash = hex.EncodeToString(packetHash[:]) + evt.Packet.PayloadType = packet.PayloadType() + evt.Packet.PayloadTypeName = packet.PayloadTypeString() + evt.Packet.RouteType = packet.RouteType() + evt.Packet.RouteTypeName = api.RouteTypeName(int16(packet.RouteType())) + evt.Packet.IsFirstObservation = isNew + evt.Observation.ObserverID = id.String() + evt.Observation.ObserverName = observerName + evt.Observation.IATA = iata + evt.Observation.HeardAt = heardAt.UnixMilli() + evt.Observation.RSSI = oParams.RSSI + evt.Observation.SNR = oParams.SNR + evt.Observation.SourceBroker = w.cfg.BrokerName + evt.Observation.PathBytes = hex.EncodeToString(packet.Path) + evt.Observation.PathLength.Raw = fmt.Sprintf("%02x", packet.PathLength) + evt.Observation.PathLength.HashSize = packet.PathHashSize() + evt.Observation.PathLength.HopCount = packet.PathHashCount() + evt.Observation.PropagationTimeMs = 0 // not yet calculated + count, err := w.db.GetPacketObservationCount(ctx, packetHash[:]) + if err != nil { + log.Printf("ingest[%s]: failed to get observation count: %v", w.cfg.BrokerName, err) + count = 0 + } + evt.Packet.ObservationCount = count + w.broadcast(hub.EventPacketObservation, iata, packet.PayloadType(), "", evt) + } +} + +// computeTransportCode derives transport_code_1 from a transport key and packet payload. +// code = HMAC-SHA256(key, payload_type_byte || payload)[0:2] as little-endian uint16. +// Coerces reserved values 0x0000 → 0x0001 and 0xFFFF → 0xFFFE per §2.4. +func computeTransportCode(key []byte, payloadType uint8, payload []byte) uint16 { + mac := hmac.New(sha256.New, key) + mac.Write([]byte{payloadType}) + mac.Write(payload) + sum := mac.Sum(nil) + code := uint16(sum[0]) | uint16(sum[1])<<8 + if code == 0x0000 { + code = 0x0001 + } + if code == 0xFFFF { + code = 0xFFFE + } + return code +} diff --git a/internal/ingest/side_effects.go b/internal/ingest/side_effects.go new file mode 100644 index 0000000..9860f3a --- /dev/null +++ b/internal/ingest/side_effects.go @@ -0,0 +1,168 @@ +package ingest + +import ( + "context" + "encoding/hex" + "fmt" + "log" + "strings" + "time" + + "github.com/MeshCore-Tower/tower-server/internal/hub" + "github.com/MeshCore-Tower/tower-server/internal/keystore" + "github.com/meshcore-go/meshcore-go" +) + +// UpsertNodeParams carries the fields extracted from a payload type 0x04 advert. +type UpsertNodeParams struct { + PublicKey []byte + Name string + NodeType uint8 // 1=companion, 2=repeater, 3=room server + Latitude *float64 + Longitude *float64 +} + +// InsertChannelMessageParams carries a decrypted group text message. +type InsertChannelMessageParams struct { + ChannelID int + PacketHash []byte + SenderName string + Content string + SentAt time.Time +} + +// channelMessageEvent is the JSON payload for a channelMessage WS event. +type channelMessageEvent struct { + ChannelID int `json:"channelId"` + ChannelHash string `json:"channelHash"` // hex-encoded single byte + PacketHash string `json:"packetHash"` // hex-encoded + SenderName string `json:"senderName"` + Content string `json:"content"` + SentAt int64 `json:"sentAt"` // epoch ms +} + +// 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"` +} + +// 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) { + if packet.PayloadType() == meshcore.PayloadTypeAdvert { + advert, err := meshcore.AdvertFromBytes(packet.Payload) + if err != nil { + log.Printf("ingest[%s]: error decoding advert payload: %v", w.cfg.BrokerName, err) + return + } + var lat, lon *float64 + if advert.AppData().Lat != 0 || advert.AppData().Lon != 0 { + la := float64(advert.AppData().Lat) / 1e6 + lo := float64(advert.AppData().Lon) / 1e6 + lat = &la + lon = &lo + } + params := UpsertNodeParams{ + PublicKey: advert.PublicKey.PublicKeyBytes(), + Name: strings.ToValidUTF8(advert.AppData().Name, "\uFFFD"), + NodeType: advert.Type(), + Latitude: lat, + Longitude: lon, + } + nodeID, err := w.db.UpsertNode(ctx, params, radio) + if err != nil { + log.Printf("ingest[%s]: db: upsert node failed: %v", w.cfg.BrokerName, err) + return + } + if err := w.db.UpsertNodeIATA(ctx, nodeID, iata); err != nil { + log.Printf("ingest[%s]: db: upsert node IATA failed: %v", w.cfg.BrokerName, err) + } + if scopeID != nil && (packet.RouteType() == meshcore.RouteTypeTransportFlood || packet.RouteType() == meshcore.RouteTypeTransportDirect) { + if err := w.db.SetNodeDefaultScope(ctx, nodeID, *scopeID); err != nil { + log.Printf("ingest[%s]: failed to set default scope for node %s: %v", w.cfg.BrokerName, hex.EncodeToString(advert.PublicKey.PublicKeyBytes()), err) + } + } + prefix4 := advert.PublicKey.PublicKeyBytes()[:4] + 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) + } + evt := nodeUpdateEvent{ + NodeID: nodeID.String(), + Name: advert.AppData().Name, + NodeType: advert.Type(), + IATA: iata, + Lat: lat, + Lng: lon, + } + w.broadcast(hub.EventNodeUpdate, iata, meshcore.PayloadTypeAdvert, "", evt) + return + } + if packet.PayloadType() == meshcore.PayloadTypeGrpTxt { + grpTxt, err := meshcore.GroupTextFromBytes(packet.Payload) + if err != nil { + log.Printf("ingest[%s]: error decoding group text payload: %v", w.cfg.BrokerName, err) + return + } + channelHashBytes := []byte{grpTxt.ChannelHash} + + // Always upsert a hash-only row so unknown channels are recorded. + _, _ = w.db.UpsertChannelHashOnly(ctx, channelHashBytes) + + // Try each known key entry for this hash. + entries := w.keys.GetKey(channelHashBytes) + if len(entries) == 0 { + return // channel key unknown; message stored as encrypted blob only + } + var payload *meshcore.GroupTextPayload + var usedEntry keystore.Entry + for _, entry := range entries { + if p, err := grpTxt.DecryptStruct(entry.Key); err == nil { + payload = p + usedEntry = entry + break + } + } + if payload == nil { + return // none of the keys worked + } + + // Upsert the keyed channel row — messages are associated with this row. + channelID, err := w.db.UpsertChannel(ctx, channelHashBytes, usedEntry.Fingerprint, usedEntry.Name, usedEntry.Hashtag) + if err != nil { + log.Printf("ingest[%s]: db: upsert keyed channel failed: %v", w.cfg.BrokerName, err) + return + } + params := InsertChannelMessageParams{ + ChannelID: channelID, + PacketHash: packetHash[:], + SenderName: strings.ReplaceAll(strings.ToValidUTF8(payload.Sender, "\uFFFD"), "\x00", ""), + SentAt: time.Unix(int64(payload.Timestamp), 0), + Content: strings.ReplaceAll(strings.ToValidUTF8(payload.Text, "\uFFFD"), "\x00", ""), + } + newMsg, err := w.db.InsertChannelMessage(ctx, params) + if err != nil { + log.Printf("ingest[%s]: db: insert channel message failed: %v", w.cfg.BrokerName, err) + return + } + + if newMsg { + evt := channelMessageEvent{ + ChannelID: channelID, + ChannelHash: hex.EncodeToString(channelHashBytes), + PacketHash: hex.EncodeToString(packetHash), + SenderName: strings.ReplaceAll(strings.ToValidUTF8(payload.Sender, "\uFFFD"), "\x00", ""), + Content: strings.ReplaceAll(strings.ToValidUTF8(payload.Text, "\uFFFD"), "\x00", ""), + SentAt: time.Unix(int64(payload.Timestamp), 0).UnixMilli(), + } + w.broadcast(hub.EventChannelMessage, iata, 0, fmt.Sprintf("%02x", grpTxt.ChannelHash), evt) + } + return + } +} diff --git a/internal/ingest/status.go b/internal/ingest/status.go new file mode 100644 index 0000000..23f34f1 --- /dev/null +++ b/internal/ingest/status.go @@ -0,0 +1,190 @@ +package ingest + +import ( + "context" + "encoding/hex" + "encoding/json" + "log" + "strconv" + "strings" + "time" + + "github.com/MeshCore-Tower/tower-server/internal/hub" +) + +// 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"` + IATA string `json:"iata,omitempty"` + Online bool `json:"online"` + BatteryMV int `json:"batteryMv,omitempty"` + UptimeSeconds int64 `json:"uptimeSeconds"` + LastStatusAt int64 `json:"lastStatusAt"` // epoch ms +} + +// 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 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"` + } `json:"stats"` + } + if err := json.Unmarshal(raw, &envelope); err != nil { + log.Printf("ingest[%s]: malformed status envelope from %s: %v", w.cfg.BrokerName, pubkeyHex, err) + return + } + pubkey, err := hex.DecodeString(pubkeyHex) + if err != nil { + log.Printf("ingest[%s]: invalid pubkey hex in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err) + return + } + id, _, err := w.db.UpsertObserver(ctx, pubkey) + if err != nil { + log.Printf("ingest[%s]: db: upsert observer failed in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err) + return + } + if err := w.db.UpsertObserverBroker(ctx, id, w.cfg.BrokerName); err != nil { + log.Printf("ingest[%s]: db: upsert observer broker failed in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err) + } + params := UpdateObserverStatusParams{ + PublicKey: pubkey, + StatusMetadata: raw, + LastStatusAt: time.Now(), + } + 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 { + log.Printf("ingest[%s]: missing or malformed radio params in status from %s, skipping radio fields", w.cfg.BrokerName, pubkeyHex) + } else { + freq, err := strconv.ParseFloat(radio[0], 32) + if err != nil { + log.Printf("ingest[%s]: error parsing radio freq in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err) + } else { + params.RadioFreqMHz = float32(freq) + } + bw, err := strconv.ParseFloat(radio[1], 32) + if err != nil { + log.Printf("ingest[%s]: error parsing radio bw in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err) + } else { + params.RadioBWKHz = float32(bw) + } + sf, err := strconv.ParseInt(radio[2], 10, 16) + if err != nil { + log.Printf("ingest[%s]: error parsing radio sf in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err) + } else { + params.RadioSF = int16(sf) + } + cr, err := strconv.ParseInt(radio[3], 10, 16) + if err != nil { + log.Printf("ingest[%s]: error parsing radio cr in status from %s: %v", w.cfg.BrokerName, pubkeyHex, err) + } else { + params.RadioCR = int16(cr) + } + } + + observerID, err := w.db.UpdateObserverStatus(ctx, params) + if err != nil { + log.Printf("ingest[%s]: db: update observer status failed for %s: %v", w.cfg.BrokerName, pubkeyHex, err) + return + } + // Store a telemetry snapshot at the configured resolution. + 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 { + log.Printf("ingest[%s]: db: insert telemetry failed for %s: %v", w.cfg.BrokerName, pubkeyHex, err) + } + + iata, err := w.db.GetObserverLastIATA(ctx, observerID) + if err != nil { + iata = "" // non-fatal, continue + } + evt := statusEvent{ + ObserverID: observerID.String(), + DisplayName: envelope.DisplayName, + IATA: iata, + Online: true, + BatteryMV: envelope.Stats.BatteryMV, + UptimeSeconds: envelope.Stats.UptimeSeconds, + LastStatusAt: time.Now().UnixMilli(), + } + payload, err := json.Marshal(evt) + if err != nil { + log.Printf("ingest[%s]: failed to marshal status event payload for %s: %v", w.cfg.BrokerName, pubkeyHex, err) + return + } + w.hub.Broadcast(hub.Event{Type: hub.EventObserverStatus, Payload: payload, IATA: iata}) +}