diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 3598b48..9584c8a 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -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 diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 5a4c574..d2ec3cf 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -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 diff --git a/db/store.go b/db/store.go index 6662605..438beaf 100644 --- a/db/store.go +++ b/db/store.go @@ -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 diff --git a/internal/api/reader.go b/internal/api/reader.go index 5b9a0a0..fda7c16 100644 --- a/internal/api/reader.go +++ b/internal/api/reader.go @@ -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 diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index c3a714f..80839c7 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -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,