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

include observation count for packets in ws event

Browse Source
This commit is contained in:
Enot (ded) Skelly
2026-05-29 11:17:30 -07:00
parent 2597192e89
commit 2bb602071a
2 changed files with 8 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
db/store.go
+3 -3
View File
@@ -57,7 +57,7 @@ func (s *Store) UpsertIATA(ctx context.Context, iata string) error {
}
// UpsertPacket inserts or bumps the packets row. Returns (isNew, error).
func (s *Store) UpsertPacket(ctx context.Context, p ingest.UpsertPacketParams) (bool, error) {
func (s *Store) UpsertPacket(ctx context.Context, p ingest.UpsertPacketParams) (bool, int64, error) {
var regionCode, subRegionCode *int32
hasTransportCodes := len(p.TransportCodes) == 4
if hasTransportCodes {
@@ -81,9 +81,9 @@ func (s *Store) UpsertPacket(ctx context.Context, p ingest.UpsertPacketParams) (
}
row, err := s.q.UpsertPacket(ctx, params)
if err != nil {
return false, err
return false, 0, err
}
return row.Inserted, nil
return row.Inserted, int64(*row.ObservationCount), nil
}
// InsertObservation inserts a packet_observations row.
internal/ingest/ingest.go
+5 -3
View File
@@ -69,8 +69,8 @@ type DB interface {
// UpsertIATA auto-creates an iata_codes row if it doesn't exist yet.
UpsertIATA(ctx context.Context, iata string) error
// UpsertPacket inserts or bumps the packets row. Returns (isNew, error).
UpsertPacket(ctx context.Context, p UpsertPacketParams) (bool, error)
// UpsertPacket inserts or bumps the packets row. Returns (isNew, observationCount, error).
UpsertPacket(ctx context.Context, p UpsertPacketParams) (bool, int64, error)
// InsertObservation inserts a packet_observations row.
// Returns (inserted, error); inserted=false means ON CONFLICT DO NOTHING fired.
@@ -237,6 +237,7 @@ type packetObservationEvent struct {
PayloadTypeName string `json:"payloadTypeName"`
RouteType uint8 `json:"routeType"`
IsFirstObservation bool `json:"isFirstObservation"`
ObservationCount int64 `json:"observationCount"`
} `json:"packet"`
Observation struct {
ObserverID string `json:"observerId"`
@@ -456,7 +457,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
OriginPubkey: originPubkey,
ChannelHash: channelHash,
}
isNew, err := w.db.UpsertPacket(ctx, pParams)
isNew, observationCount, 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
@@ -512,6 +513,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
evt.Packet.PayloadTypeName = packet.PayloadTypeString()
evt.Packet.RouteType = packet.RouteType()
evt.Packet.IsFirstObservation = isNew
evt.Packet.ObservationCount = observationCount
evt.Observation.ObserverID = id.String()
evt.Observation.ObserverName = observerName
evt.Observation.IATA = iata