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

add path resolution

Browse Source
This commit is contained in:
Enot (ded) Skelly
2026-06-02 16:26:32 -07:00
parent c148472502
commit 335ff9662e
5 changed files with 118 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
db/queries/queries.sql
+1 -1
View File
@@ -619,7 +619,7 @@ ON CONFLICT (region_id, iata) DO NOTHING;
-- ============================================================
-- name: ResolvePathHashes :many
SELECT DISTINCT n.id
SELECT ns.prefix_4 AS hash, n.id AS node_id, n.name, n.latitude, n.longitude, n.public_key
FROM node_short_ids ns
JOIN nodes n ON n.id = ns.node_id
WHERE ns.iata = $1
db/sqlc/queries.sql.go
+22 -6
View File
@@ -1955,7 +1955,7 @@ func (q *Queries) RefreshTopNodes(ctx context.Context) error {
const resolvePathHashes = `-- name: ResolvePathHashes :many
SELECT DISTINCT n.id
SELECT ns.prefix_4 AS hash, n.id AS node_id, n.name, n.latitude, n.longitude, n.public_key
FROM node_short_ids ns
JOIN nodes n ON n.id = ns.node_id
WHERE ns.iata = $1
@@ -1973,22 +1973,38 @@ type ResolvePathHashesParams struct {
Column2 [][]byte `json:"column_2"`
}
type ResolvePathHashesRow struct {
Hash []byte `json:"hash"`
NodeID uuid.UUID `json:"node_id"`
Name *string `json:"name"`
Latitude *float64 `json:"latitude"`
Longitude *float64 `json:"longitude"`
PublicKey []byte `json:"public_key"`
}
// ============================================================
// HELPERS
// ============================================================
func (q *Queries) ResolvePathHashes(ctx context.Context, arg ResolvePathHashesParams) ([]uuid.UUID, error) {
func (q *Queries) ResolvePathHashes(ctx context.Context, arg ResolvePathHashesParams) ([]ResolvePathHashesRow, error) {
rows, err := q.db.Query(ctx, resolvePathHashes, arg.Iata, arg.Column2)
if err != nil {
return nil, err
}
defer rows.Close()
items := []uuid.UUID{}
items := []ResolvePathHashesRow{}
for rows.Next() {
var id uuid.UUID
if err := rows.Scan(&id); err != nil {
var i ResolvePathHashesRow
if err := rows.Scan(
&i.Hash,
&i.NodeID,
&i.Name,
&i.Latitude,
&i.Longitude,
&i.PublicKey,
); err != nil {
return nil, err
}
items = append(items, id)
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
db/store.go
+65 -7
View File
@@ -160,6 +160,14 @@ func (s *Store) UpsertNodeIATA(ctx context.Context, nodeID uuid.UUID, iata strin
return s.q.UpsertNodeIATA(ctx, params)
}
func (s *Store) UpsertNodeShortID(ctx context.Context, nodeID uuid.UUID, iata string, prefix4 []byte) error {
return s.q.UpsertNodeShortID(ctx, sqlc.UpsertNodeShortIDParams{
NodeID: nodeID,
Iata: iata,
Prefix4: prefix4,
})
}
// InsertChannelMessage stores a decrypted group text message.
func (s *Store) InsertChannelMessage(ctx context.Context, m ingest.InsertChannelMessageParams) (bool, error) {
params := sqlc.InsertChannelMessageParams{ChannelID: int32(m.ChannelID), PacketHash: m.PacketHash, SenderName: &m.SenderName, Content: &m.Content, SentAt: pgtype.Timestamptz{Time: m.SentAt, Valid: true}}
@@ -207,9 +215,13 @@ func (s *Store) GetObserverRadio(ctx context.Context, observerID uuid.UUID) (ing
return settings, nil
}
// ResolvePathHashes returns a list of node UUIDs for the given path hash prefixes and IATA.
// ResolvePathHashes returns a map of path hash prefix → matching node UUIDs for the given IATA.
// Hash size is inferred from the length of the first element in hashes.
func (s *Store) ResolvePathHashes(ctx context.Context, iata string, hashes [][]byte) ([]uuid.UUID, error) {
// Confidence: len == 1 → HIGH, len > 1 → AMBIGUOUS, len == 0 → NONE.
func (s *Store) ResolvePathHashes(ctx context.Context, iata string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error) {
if len(hashes) == 0 {
return nil, nil
}
rows, err := s.q.ResolvePathHashes(ctx, sqlc.ResolvePathHashesParams{
Iata: iata,
Column2: hashes,
@@ -217,10 +229,18 @@ func (s *Store) ResolvePathHashes(ctx context.Context, iata string, hashes [][]b
if err != nil {
return nil, err
}
ids := make([]uuid.UUID, len(rows))
copy(ids, rows)
return ids, nil
result := make(map[string][]api.ResolvedPathEntry)
for _, row := range rows {
key := hex.EncodeToString(row.Hash[:len(hashes[0])])
result[key] = append(result[key], api.ResolvedPathEntry{
NodeID: row.NodeID,
Name: row.Name,
Latitude: row.Latitude,
Longitude: row.Longitude,
PublicKey: row.PublicKey,
})
}
return result, nil
}
// UpsertChannel upserts a channel row by (hash, keyFingerprint) and returns its integer ID.
@@ -1071,8 +1091,46 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet,
RSSI: v.Rssi,
SNR: v.Snr,
SourceBroker: *v.SourceBroker,
ResolvedPath: []api.ResolvedHop{}, // TODO: implement path resolution
}
resolvedPath := []api.ResolvedHop{}
if v.PathBytes != nil && v.HashSize > 0 {
hashSize := int(v.HashSize)
hashes := make([][]byte, 0, len(v.PathBytes)/hashSize)
for i := 0; i+hashSize <= len(v.PathBytes); i += hashSize {
hashes = append(hashes, v.PathBytes[i:i+hashSize])
}
resolved, err := s.ResolvePathHashes(ctx, v.Iata, hashes)
if err != nil {
log.Printf("store: path resolution failed for observation %d: %v", v.ID, err)
} else {
for _, hash := range hashes {
key := hex.EncodeToString(hash)
entries := resolved[key]
hop := api.ResolvedHop{
Nodes: make([]api.ResolvedNode, 0, len(entries)),
}
switch len(entries) {
case 0:
hop.Confidence = "none"
case 1:
hop.Confidence = "high"
default:
hop.Confidence = "ambiguous"
}
for _, e := range entries {
hop.Nodes = append(hop.Nodes, api.ResolvedNode{
ID: e.NodeID,
Name: e.Name,
Latitude: e.Latitude,
Longitude: e.Longitude,
PublicKey: hex.EncodeToString(e.PublicKey),
})
}
resolvedPath = append(resolvedPath, hop)
}
}
}
obs.ResolvedPath = resolvedPath
if v.PathBytes != nil {
pb := hex.EncodeToString(v.PathBytes)
obs.PathBytes = &pb
internal/api/reader.go
+10 -2
View File
@@ -105,8 +105,8 @@ type PacketRadio struct {
// ResolvedHop is a single hop in a packet's resolved path.
type ResolvedHop struct {
Confidence string `json:"confidence"` // "high", "low", "unknown"
Node *ResolvedNode `json:"node,omitempty"`
Confidence string `json:"confidence"` // "high", "low", "unknown"
Nodes []ResolvedNode `json:"nodes"`
}
// ResolvedNode is a node reference within a resolved path hop.
@@ -118,6 +118,14 @@ type ResolvedNode struct {
Longitude *float64 `json:"longitude,omitempty"`
}
type ResolvedPathEntry struct {
NodeID uuid.UUID
Name *string
Latitude *float64
Longitude *float64
PublicKey []byte
}
// Packet is the full packet representation including all observations and resolved paths.
type Packet struct {
PacketHash string `json:"packetHash"` // hex-encoded
internal/ingest/ingest.go
+20 -7
View File
@@ -34,6 +34,7 @@ import (
"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"
"github.com/MeshCore-Tower/tower-server/internal/keystore"
)
@@ -86,6 +87,9 @@ type DB interface {
// UpsertNodeIATA upserts a node_iatas row.
UpsertNodeIATA(ctx context.Context, nodeID uuid.UUID, iata string) error
// UpsertNodeShortID upserts a node_short_ids row for path resolution.
UpsertNodeShortID(ctx context.Context, nodeID uuid.UUID, iata string, prefix4 []byte) error
// InsertChannelMessage stores a decrypted group text message. Returns insert success and an error.
InsertChannelMessage(ctx context.Context, m InsertChannelMessageParams) (bool, error)
@@ -104,7 +108,7 @@ type DB interface {
GetObserverRadio(ctx context.Context, observerID uuid.UUID) (RadioSettings, error)
// ResolvePathHashes returns a list of node UUIDs for the given path hash prefixes and IATA.
ResolvePathHashes(ctx context.Context, iata string, hashes [][]byte) ([]uuid.UUID, error)
ResolvePathHashes(ctx context.Context, iata string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error)
// UpsertChannel upserts a channel row by (hash, keyFingerprint) and returns its integer ID.
// Pass nil keyFingerprint to record a hash-only row when the key is unknown.
@@ -532,13 +536,18 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [
return
}
resolvedIDs, err := w.db.ResolvePathHashes(ctx, iata, packet.PathHashes())
resolved, err := w.db.ResolvePathHashes(ctx, iata, packet.PathHashes())
if err != nil {
log.Printf("ingest[%s]: db: resolve path hashes failed: %v", w.cfg.BrokerName, err)
resolvedIDs = []uuid.UUID{}
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.runCapabilityDetection(ctx, packet.PayloadType(), packet.PathHashSize(), resolvedIDs)
w.handlePayloadTypeSideEffects(ctx, packet, iata, packetHash[:], radio)
evt := packetObservationEvent{}
evt.PacketHash = hex.EncodeToString(packetHash[:])
@@ -741,8 +750,8 @@ func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshc
}
var lat, lon *float64
if advert.AppData().Lat != 0 || advert.AppData().Lon != 0 {
la := float64(advert.AppData().Lat)
lo := float64(advert.AppData().Lon)
la := float64(advert.AppData().Lat) / 1e7
lo := float64(advert.AppData().Lon) / 1e7
lat = &la
lon = &lo
}
@@ -761,6 +770,10 @@ func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshc
if err := w.db.UpsertNodeIATA(ctx, nodeID, iata); err != nil {
log.Printf("ingest[%s]: db: upsert node IATA failed: %v", w.cfg.BrokerName, 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,