From 65cb8c9d617b4ec2340b89d81bcd00bedcdb4556 Mon Sep 17 00:00:00 2001 From: "Enot (ded) Skelly" Date: Fri, 5 Jun 2026 15:57:08 -0700 Subject: [PATCH] add routes api any route observer where all hops are high confidence are now stored and available at /routes you can also search for a src dest route and it will return any recored route subset that can link the src and dest hash --- db/migrations/001_schema.sql | 18 +++++ db/queries/queries.sql | 33 +++++++++ db/routes.go | 77 +++++++++++++++++++ db/sqlc/models.go | 10 +++ db/sqlc/queries.sql.go | 127 ++++++++++++++++++++++++++++++++ internal/api/handlers/routes.go | 92 +++++++++++++++++++++++ internal/api/reader.go | 4 + internal/api/router/router.go | 1 + internal/api/routes.go | 21 ++++++ internal/ingest/ingest.go | 3 + internal/ingest/packet.go | 21 ++++++ 11 files changed, 407 insertions(+) create mode 100644 db/routes.go create mode 100644 internal/api/handlers/routes.go create mode 100644 internal/api/routes.go diff --git a/db/migrations/001_schema.sql b/db/migrations/001_schema.sql index 5487156..c62cead 100644 --- a/db/migrations/001_schema.sql +++ b/db/migrations/001_schema.sql @@ -335,6 +335,24 @@ CREATE TABLE channel_messages ( CREATE INDEX idx_channel_messages_channel ON channel_messages(channel_id, sent_at DESC); CREATE INDEX idx_channel_messages_sent_brin ON channel_messages USING BRIN (sent_at); +-- ============================================================ +-- ROUTES +-- ============================================================ + +CREATE TABLE known_routes ( + id BIGSERIAL PRIMARY KEY, + node_ids UUID[] NOT NULL, -- resolved node UUIDs in hop order + hash_prefix BYTEA[] NOT NULL, -- raw hash bytes in hop order for prefix matching + iata CHAR(3) NOT NULL REFERENCES iata_codes(iata) ON DELETE CASCADE, + hop_count INT NOT NULL, + first_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(), + last_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(), + UNIQUE (node_ids, iata) +); + +CREATE INDEX idx_known_routes_iata ON known_routes(iata); +CREATE INDEX idx_known_routes_hop_count ON known_routes(iata, hop_count); + -- ============================================================ -- MATERIALIZED VIEWS -- ============================================================ diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 1c02925..e2a1615 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -821,6 +821,39 @@ GROUP BY p.trace_tag ORDER BY MAX(p.last_heard_at) DESC LIMIT $6; +-- ============================================================ +-- ROUTES +-- ============================================================ + +-- name: UpsertKnownRoute :exec +-- Inserts or updates a known route (all hops resolved to high confidence). +-- node_ids and hash_prefix are ordered arrays of the resolved node UUIDs and +-- their hash bytes. last_seen is bumped on conflict. +INSERT INTO known_routes (node_ids, hash_prefix, iata, hop_count) +VALUES ($1, $2, $3, $4) +ON CONFLICT (node_ids, iata) DO UPDATE SET + last_seen = NOW(); + +-- name: ListKnownRoutes :many +-- Returns known routes filtered by IATA, ordered by most recently seen. +SELECT id, node_ids, hash_prefix, iata, hop_count, first_seen, last_seen +FROM known_routes +WHERE ($1 = '' OR iata = $1) + AND ($2 = 0 OR hop_count = $2) + AND ($3 = 0 OR id < $3) +ORDER BY last_seen DESC +LIMIT $4; + +-- name: SearchKnownRoutes :many +-- Returns known routes containing a subsequence from source to destination hash prefix. +-- Matches routes where source hash appears before destination hash in the hash_prefix array. +SELECT id, node_ids, hash_prefix, iata, hop_count, first_seen, last_seen +FROM known_routes +WHERE iata = $1 + AND hash_prefix @> ARRAY[$2::bytea] + AND hash_prefix @> ARRAY[$3::bytea] +ORDER BY hop_count ASC, last_seen DESC; + -- ============================================================ -- HELPERS -- ============================================================ diff --git a/db/routes.go b/db/routes.go new file mode 100644 index 0000000..e2be41e --- /dev/null +++ b/db/routes.go @@ -0,0 +1,77 @@ +package db + +import ( + "context" + "encoding/hex" + + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/google/uuid" +) + +func (s *Store) UpsertKnownRoute(ctx context.Context, nodeIDs []uuid.UUID, hashPrefix [][]byte, iata string, hopCount int32) error { + return s.q.UpsertKnownRoute(ctx, sqlc.UpsertKnownRouteParams{ + NodeIds: nodeIDs, + HashPrefix: hashPrefix, + Iata: iata, + HopCount: int32(hopCount), + }) +} + +func (s *Store) ListKnownRoutes(ctx context.Context, iata string, hopCount int32, cursor int64, limit int32) ([]api.KnownRoute, error) { + rows, err := s.q.ListKnownRoutes(ctx, sqlc.ListKnownRoutesParams{ + Column1: iata, + Column2: hopCount, + Column3: cursor, + Limit: limit, + }) + if err != nil { + return nil, err + } + return toKnownRoutes(rows), nil +} + +func (s *Store) SearchKnownRoutes(ctx context.Context, iata, fromHash, toHash string) ([]api.KnownRoute, error) { + fromBytes, err := hex.DecodeString(fromHash) + if err != nil { + return nil, err + } + toBytes, err := hex.DecodeString(toHash) + if err != nil { + return nil, err + } + rows, err := s.q.SearchKnownRoutes(ctx, sqlc.SearchKnownRoutesParams{ + Iata: iata, + Column2: fromBytes, + Column3: toBytes, + }) + if err != nil { + return nil, err + } + return toKnownRoutes(rows), nil +} + +func toKnownRoutes(rows []sqlc.KnownRoute) []api.KnownRoute { + items := make([]api.KnownRoute, 0, len(rows)) + for _, r := range rows { + hops := make([]api.RouteHop, 0, len(r.NodeIds)) + for i, nodeID := range r.NodeIds { + hop := api.RouteHop{ + NodeID: nodeID, + } + if i < len(r.HashPrefix) { + hop.HashBytes = hex.EncodeToString(r.HashPrefix[i]) + } + hops = append(hops, hop) + } + items = append(items, api.KnownRoute{ + ID: r.ID, + IATA: r.Iata, + HopCount: r.HopCount, + Hops: hops, + FirstSeen: r.FirstSeen.Time.UnixMilli(), + LastSeen: r.LastSeen.Time.UnixMilli(), + }) + } + return items +} diff --git a/db/sqlc/models.go b/db/sqlc/models.go index 9f6626d..22fc413 100644 --- a/db/sqlc/models.go +++ b/db/sqlc/models.go @@ -49,6 +49,16 @@ type IataCode struct { AddedAt pgtype.Timestamptz `json:"added_at"` } +type KnownRoute struct { + ID int64 `json:"id"` + NodeIds []uuid.UUID `json:"node_ids"` + HashPrefix [][]byte `json:"hash_prefix"` + Iata string `json:"iata"` + HopCount int32 `json:"hop_count"` + FirstSeen pgtype.Timestamptz `json:"first_seen"` + LastSeen pgtype.Timestamptz `json:"last_seen"` +} + type MvHourlyIataStat struct { Iata string `json:"iata"` Hour pgtype.Timestamptz `json:"hour"` diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 441f153..2600845 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -1713,6 +1713,57 @@ func (q *Queries) ListIATAs(ctx context.Context) ([]IataCode, error) { return items, nil } +const listKnownRoutes = `-- name: ListKnownRoutes :many +SELECT id, node_ids, hash_prefix, iata, hop_count, first_seen, last_seen +FROM known_routes +WHERE ($1 = '' OR iata = $1) + AND ($2 = 0 OR hop_count = $2) + AND ($3 = 0 OR id < $3) +ORDER BY last_seen DESC +LIMIT $4 +` + +type ListKnownRoutesParams struct { + Column1 interface{} `json:"column_1"` + Column2 interface{} `json:"column_2"` + Column3 interface{} `json:"column_3"` + Limit int32 `json:"limit"` +} + +// Returns known routes filtered by IATA, ordered by most recently seen. +func (q *Queries) ListKnownRoutes(ctx context.Context, arg ListKnownRoutesParams) ([]KnownRoute, error) { + rows, err := q.db.Query(ctx, listKnownRoutes, + arg.Column1, + arg.Column2, + arg.Column3, + arg.Limit, + ) + if err != nil { + return nil, err + } + defer rows.Close() + items := []KnownRoute{} + for rows.Next() { + var i KnownRoute + if err := rows.Scan( + &i.ID, + &i.NodeIds, + &i.HashPrefix, + &i.Iata, + &i.HopCount, + &i.FirstSeen, + &i.LastSeen, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const listMessagesAfterID = `-- name: ListMessagesAfterID :many SELECT DISTINCT ON (cm.id) cm.id, cm.channel_id, cm.packet_hash, cm.sender_name, cm.sender_pubkey, cm.content, cm.sent_at, encode(cm.packet_hash, 'hex') as packet_hash_hex, c.channel_hash, (SELECT COUNT(*) FROM packet_observations po2 WHERE po2.packet_hash = cm.packet_hash) AS observation_count @@ -2650,6 +2701,51 @@ func (q *Queries) ResolvePathHashes(ctx context.Context, arg ResolvePathHashesPa return items, nil } +const searchKnownRoutes = `-- name: SearchKnownRoutes :many +SELECT id, node_ids, hash_prefix, iata, hop_count, first_seen, last_seen +FROM known_routes +WHERE iata = $1 + AND hash_prefix @> ARRAY[$2::bytea] + AND hash_prefix @> ARRAY[$3::bytea] +ORDER BY hop_count ASC, last_seen DESC +` + +type SearchKnownRoutesParams struct { + Iata string `json:"iata"` + Column2 []byte `json:"column_2"` + Column3 []byte `json:"column_3"` +} + +// Returns known routes containing a subsequence from source to destination hash prefix. +// Matches routes where source hash appears before destination hash in the hash_prefix array. +func (q *Queries) SearchKnownRoutes(ctx context.Context, arg SearchKnownRoutesParams) ([]KnownRoute, error) { + rows, err := q.db.Query(ctx, searchKnownRoutes, arg.Iata, arg.Column2, arg.Column3) + if err != nil { + return nil, err + } + defer rows.Close() + items := []KnownRoute{} + for rows.Next() { + var i KnownRoute + if err := rows.Scan( + &i.ID, + &i.NodeIds, + &i.HashPrefix, + &i.Iata, + &i.HopCount, + &i.FirstSeen, + &i.LastSeen, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const setChannelKeyKnown = `-- name: SetChannelKeyKnown :exec UPDATE channels SET key_known = TRUE WHERE channel_hash = $1 AND key_fingerprint = $2 @@ -2874,6 +2970,37 @@ func (q *Queries) UpsertIATADetails(ctx context.Context, arg UpsertIATADetailsPa return err } +const upsertKnownRoute = `-- name: UpsertKnownRoute :exec + +INSERT INTO known_routes (node_ids, hash_prefix, iata, hop_count) +VALUES ($1, $2, $3, $4) +ON CONFLICT (node_ids, iata) DO UPDATE SET + last_seen = NOW() +` + +type UpsertKnownRouteParams struct { + NodeIds []uuid.UUID `json:"node_ids"` + HashPrefix [][]byte `json:"hash_prefix"` + Iata string `json:"iata"` + HopCount int32 `json:"hop_count"` +} + +// ============================================================ +// ROUTES +// ============================================================ +// Inserts or updates a known route (all hops resolved to high confidence). +// node_ids and hash_prefix are ordered arrays of the resolved node UUIDs and +// their hash bytes. last_seen is bumped on conflict. +func (q *Queries) UpsertKnownRoute(ctx context.Context, arg UpsertKnownRouteParams) error { + _, err := q.db.Exec(ctx, upsertKnownRoute, + arg.NodeIds, + arg.HashPrefix, + arg.Iata, + arg.HopCount, + ) + return err +} + const upsertNode = `-- name: UpsertNode :one INSERT INTO nodes (public_key, node_type, name, latitude, longitude, location_source, last_advert_at, last_seen, radio_freq_mhz, radio_sf, radio_bw_khz) diff --git a/internal/api/handlers/routes.go b/internal/api/handlers/routes.go new file mode 100644 index 0000000..fedc8bb --- /dev/null +++ b/internal/api/handlers/routes.go @@ -0,0 +1,92 @@ +package handlers + +import ( + "net/http" + "strconv" + + "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/go-chi/chi/v5" +) + +// RoutesRouter mounts all /routes routes onto a subrouter. +// +// GET /routes → listKnownRoutes +// GET /routes/search → searchKnownRoutes +func RoutesRouter(reader api.Reader) http.Handler { + r := chi.NewRouter() + r.Get("/", listKnownRoutes(reader)) + r.Get("/search", searchKnownRoutes(reader)) + return r +} + +// listKnownRoutes godoc +// +// @Summary List known routes +// @Tags Routes +// @Produce json +// @Param iata query string false "Filter by IATA code" +// @Param hopCount query int false "Filter by exact hop count" +// @Param cursor query int false "Route ID of last item for pagination" +// @Param limit query int false "Max results (default 50)" +// @Success 200 {object} []api.KnownRoute +// @Failure 500 {object} handlers.APIError +// @Router /routes [get] +func listKnownRoutes(reader api.Reader) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + iata := r.URL.Query().Get("iata") + var hopCount int32 + if v := r.URL.Query().Get("hopCount"); v != "" { + if h, err := strconv.ParseInt(v, 10, 32); err == nil { + hopCount = int32(h) + } + } + var cursor int64 + if v := r.URL.Query().Get("cursor"); v != "" { + if c, err := strconv.ParseInt(v, 10, 64); err == nil { + cursor = c + } + } + var limit int32 = 50 + if v := r.URL.Query().Get("limit"); v != "" { + if l, err := strconv.ParseInt(v, 10, 32); err == nil { + limit = int32(l) + } + } + routes, err := reader.ListKnownRoutes(r.Context(), iata, hopCount, cursor, limit) + if err != nil { + respondError(w, http.StatusInternalServerError, "internal server error") + return + } + respond(w, http.StatusOK, routes) + } +} + +// searchKnownRoutes godoc +// +// @Summary Search known routes by source and destination hash +// @Tags Routes +// @Produce json +// @Param iata query string true "IATA code to search within" +// @Param from query string true "Source node hash prefix (hex)" +// @Param to query string true "Destination node hash prefix (hex)" +// @Success 200 {object} []api.KnownRoute +// @Failure 400 {object} handlers.APIError +// @Failure 500 {object} handlers.APIError +// @Router /routes/search [get] +func searchKnownRoutes(reader api.Reader) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + iata := r.URL.Query().Get("iata") + from := r.URL.Query().Get("from") + to := r.URL.Query().Get("to") + if iata == "" || from == "" || to == "" { + respondError(w, http.StatusBadRequest, "iata, from and to are required") + return + } + routes, err := reader.SearchKnownRoutes(r.Context(), iata, from, to) + if err != nil { + respondError(w, http.StatusInternalServerError, "internal server error") + return + } + respond(w, http.StatusOK, routes) + } +} diff --git a/internal/api/reader.go b/internal/api/reader.go index 1aac9fc..8a1cd75 100644 --- a/internal/api/reader.go +++ b/internal/api/reader.go @@ -137,4 +137,8 @@ type Reader interface { ListTraceTags(ctx context.Context, iatas []string, scope string, since, until time.Time, cursor time.Time, limit int32) ([]TraceTagSummary, error) // GetTraceByTag returns all packets for a given trace tag with resolved routes. GetTraceByTag(ctx context.Context, tag string) (*TraceDetail, error) + // ListKnownRoutes returns known routes filtered by IATA and optional hop count. + ListKnownRoutes(ctx context.Context, iata string, hopCount int32, cursor int64, limit int32) ([]KnownRoute, error) + // SearchKnownRoutes returns known routes containing a path from source to destination hash. + SearchKnownRoutes(ctx context.Context, iata, fromHash, toHash string) ([]KnownRoute, error) } diff --git a/internal/api/router/router.go b/internal/api/router/router.go index f998d1b..eb576ff 100644 --- a/internal/api/router/router.go +++ b/internal/api/router/router.go @@ -71,6 +71,7 @@ func New(h *hub.Hub, reader api.Reader, workers []*ingest.Worker, maxConnsPerIP r.Mount("/messages", handlers.MessagesRouter(reader)) r.Mount("/iatas", handlers.IATAsRouter(reader)) r.Mount("/regions", handlers.RegionsRouter(reader)) + r.Mount("/routes", handlers.RoutesRouter(reader)) r.Mount("/scopes", handlers.ScopesRouter(reader)) r.Mount("/stats", handlers.StatsRouter(reader)) r.Mount("/traces", handlers.TracesRouter(reader)) diff --git a/internal/api/routes.go b/internal/api/routes.go new file mode 100644 index 0000000..fd4f646 --- /dev/null +++ b/internal/api/routes.go @@ -0,0 +1,21 @@ +package api + +import "github.com/google/uuid" + +// RouteHop is a single resolved hop in a known route. +type RouteHop struct { + NodeID uuid.UUID `json:"nodeId"` + HashBytes string `json:"hashBytes"` // hex-encoded hash prefix + Node *ResolvedNode `json:"node,omitempty"` // populated when node details are available +} + +// KnownRoute is a fully resolved path through the mesh where all hops +// have been confirmed as high confidence. +type KnownRoute struct { + ID int64 `json:"id"` + IATA string `json:"iata"` + HopCount int32 `json:"hopCount"` + Hops []RouteHop `json:"hops"` + FirstSeen int64 `json:"firstSeen"` // epoch ms + LastSeen int64 `json:"lastSeen"` // epoch ms +} diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index 2981757..aa51af9 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -142,6 +142,9 @@ type DB interface { // 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 + + // UpsertKnownRoute stores a fully resolved path where all hops have high confidence. + UpsertKnownRoute(ctx context.Context, nodeIDs []uuid.UUID, hashPrefix [][]byte, iata string, hopCount int32) error } // ChannelKeyStore is a read-only view of the channel keys loaded from config. diff --git a/internal/ingest/packet.go b/internal/ingest/packet.go index e26cd9d..fb7d885 100644 --- a/internal/ingest/packet.go +++ b/internal/ingest/packet.go @@ -596,6 +596,27 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ resolvedIDs = append(resolvedIDs, e.NodeID) } } + hashes := packet.PathHashes() + if len(hashes) > 0 && resolved != nil { + allHigh := true + nodeIDs := make([]uuid.UUID, 0, len(hashes)) + hashPrefixes := make([][]byte, 0, len(hashes)) + for _, hash := range hashes { + key := hex.EncodeToString(hash) + entries := resolved[key] + if len(entries) != 1 { + allHigh = false + break + } + nodeIDs = append(nodeIDs, entries[0].NodeID) + hashPrefixes = append(hashPrefixes, hash) + } + if allHigh && len(nodeIDs) > 0 { + if err := w.db.UpsertKnownRoute(ctx, nodeIDs, hashPrefixes, iata, int32(len(nodeIDs))); err != nil { + log.Printf("ingest[%s]: failed to upsert known route: %v", w.cfg.BrokerName, err) + } + } + } w.runCapabilityDetection(ctx, packet.PayloadType(), packet.PathHashSize(), resolvedIDs) if inserted { w.handlePayloadTypeSideEffects(ctx, packet, iata, packetHash[:], radio, scopeID)