From f6f94acad8910cb69361b11ff30436f40eedf8fb Mon Sep 17 00:00:00 2001 From: "Enot (ded) Skelly" Date: Tue, 16 Jun 2026 14:47:30 -0700 Subject: [PATCH] feat(neighbors): add zero hop advert neighbors for observers w/ snr --- internal/ingest/packet.go | 2 +- internal/ingest/side_effects.go | 16 +++++++++++++++- 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/internal/ingest/packet.go b/internal/ingest/packet.go index c67516c..6a4dd53 100644 --- a/internal/ingest/packet.go +++ b/internal/ingest/packet.go @@ -760,7 +760,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, matchedScope) + w.handlePayloadTypeSideEffects(ctx, packet, iata, packetHash[:], radio, scopeID, matchedScope, pubkeyBytes, float32(parseNumber(envelope.SNR))) 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 2b43849..a154fb6 100644 --- a/internal/ingest/side_effects.go +++ b/internal/ingest/side_effects.go @@ -65,7 +65,7 @@ type nodeUpdateEvent struct { // 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, matchedScope *string) { +func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshcore.Packet, iata string, packetHash []byte, radio RadioSettings, scopeID *int32, matchedScope *string, observerPubkey []byte, rxSNR float32) { if packet.PayloadType() == meshcore.PayloadTypeAdvert { advert, err := meshcore.AdvertFromBytes(packet.Payload) if err != nil { @@ -99,6 +99,7 @@ func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshc if w.onNodeUpsert != nil { w.onNodeUpsert(ctx, nodeID) } + // record advertiser neighbors // if the advert was forwarded, the first hop is a neighbor if packet.PathHashCount() > 0 && (advert.Type() == meshcore.AdvertTypeRepeater || advert.Type() == meshcore.AdvertTypeRoom) { firstHop := packet.PathHashes() @@ -114,6 +115,19 @@ func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshc } } } + // record observer neighbors + // if heard directly (zero-hop), record the observer's own RX SNR + // of hearing this advertiser as a node_neighbors edge. Skipped if + // the observer has no node row yet (hasn't advertised itself). + if packet.PathHashCount() == 0 && (advert.Type() == meshcore.AdvertTypeRepeater || advert.Type() == meshcore.AdvertTypeRoom) { + observerNodeID, oErr := w.db.GetNodeByPubkey(ctx, observerPubkey) + if oErr == nil && observerNodeID != nodeID { + snr := rxSNR + if err := w.db.UpsertNodeNeighbor(ctx, observerNodeID, nodeID, iata, &snr); err != nil { + log.Printf("ingest[%s]: failed to upsert observer-advert neighbor: %v", w.cfg.BrokerName, err) + } + } + } if err := w.db.UpsertNodeIATA(ctx, nodeID, iata); err != nil { log.Printf("ingest[%s]: db: upsert node IATA failed: %v", w.cfg.BrokerName, err) }