diff --git a/db/migrations/001_schema.sql b/db/migrations/001_schema.sql index 49a1ef9..651297e 100644 --- a/db/migrations/001_schema.sql +++ b/db/migrations/001_schema.sql @@ -64,6 +64,9 @@ CREATE TABLE nodes ( ) STORED, first_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(), last_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(), + radio_freq_mhz REAL, + radio_sf SMALLINT, + radio_bw_khz REAL, metadata JSONB ); diff --git a/db/queries/queries.sql b/db/queries/queries.sql index a84bfde..a244dc9 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -72,6 +72,9 @@ SELECT o.display_name, o.observer_type, o.last_status_at, + o.radio_freq_mhz, + o.radio_sf, + o.radio_bw_khz, COALESCE(CASE WHEN o.last_status_at > NOW() - INTERVAL '5 minutes' THEN 'online' ELSE 'offline' @@ -289,8 +292,8 @@ LIMIT $3; -- ============================================================ -- name: UpsertNode :one -INSERT INTO nodes (public_key, node_type, name, latitude, longitude, location_source, last_advert_at, last_seen) -VALUES ($1, $2, $3, $4, $5, 'advert', NOW(), NOW()) +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) +VALUES ($1, $2, $3, $4, $5, 'advert', NOW(), NOW(), $6, $7, $8) ON CONFLICT (public_key) DO UPDATE SET node_type = EXCLUDED.node_type, name = COALESCE(EXCLUDED.name, nodes.name), @@ -298,7 +301,10 @@ ON CONFLICT (public_key) DO UPDATE SET longitude = COALESCE(EXCLUDED.longitude, nodes.longitude), location_source = CASE WHEN EXCLUDED.latitude IS NOT NULL THEN 'advert' ELSE nodes.location_source END, last_advert_at = NOW(), - last_seen = NOW() + last_seen = NOW(), + radio_freq_mhz = EXCLUDED.radio_freq_mhz, + radio_sf = EXCLUDED.radio_sf, + radio_bw_khz = EXCLUDED.radio_bw_khz RETURNING *; -- name: SetNodeMultibytePaths :exec @@ -326,6 +332,7 @@ WHERE n.id = $1; -- name: ListNodes :many SELECT n.id, n.public_key, n.node_type, n.name, n.latitude, n.longitude, n.last_seen, + n.radio_freq_mhz, n.radio_sf, n.radio_bw_khz, array_remove(array_agg(DISTINCT ni.iata ORDER BY ni.iata), NULL)::text[] AS iatas, EXISTS (SELECT 1 FROM observers o WHERE o.public_key = n.public_key) AS is_observer, (SELECT o.id FROM observers o WHERE o.public_key = n.public_key LIMIT 1) AS observer_id diff --git a/db/sqlc/models.go b/db/sqlc/models.go index e02012a..491bfc9 100644 --- a/db/sqlc/models.go +++ b/db/sqlc/models.go @@ -80,6 +80,9 @@ type Node struct { MinFirmwareVersion *string `json:"min_firmware_version"` FirstSeen pgtype.Timestamptz `json:"first_seen"` LastSeen pgtype.Timestamptz `json:"last_seen"` + RadioFreqMhz *float32 `json:"radio_freq_mhz"` + RadioSf *int16 `json:"radio_sf"` + RadioBwKhz *float32 `json:"radio_bw_khz"` Metadata []byte `json:"metadata"` } diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index ff5bdce..88e726e 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -206,7 +206,7 @@ func (q *Queries) GetIATA(ctx context.Context, iata string) (IataCode, error) { } const getNodeByID = `-- name: GetNodeByID :one -SELECT id, public_key, node_type, name, latitude, longitude, location_source, last_advert_at, supports_multibyte_paths, supports_multibyte_traces, min_firmware_version, first_seen, last_seen, metadata, +SELECT id, public_key, node_type, name, latitude, longitude, location_source, last_advert_at, supports_multibyte_paths, supports_multibyte_traces, min_firmware_version, first_seen, last_seen, radio_freq_mhz, radio_sf, radio_bw_khz, metadata, EXISTS (SELECT 1 FROM observers o WHERE o.public_key = n.public_key) AS is_observer, (SELECT o.id FROM observers o WHERE o.public_key = n.public_key LIMIT 1) AS observer_id FROM nodes n @@ -227,6 +227,9 @@ type GetNodeByIDRow struct { MinFirmwareVersion *string `json:"min_firmware_version"` FirstSeen pgtype.Timestamptz `json:"first_seen"` LastSeen pgtype.Timestamptz `json:"last_seen"` + RadioFreqMhz *float32 `json:"radio_freq_mhz"` + RadioSf *int16 `json:"radio_sf"` + RadioBwKhz *float32 `json:"radio_bw_khz"` Metadata []byte `json:"metadata"` IsObserver bool `json:"is_observer"` ObserverID uuid.UUID `json:"observer_id"` @@ -249,6 +252,9 @@ func (q *Queries) GetNodeByID(ctx context.Context, id uuid.UUID) (GetNodeByIDRow &i.MinFirmwareVersion, &i.FirstSeen, &i.LastSeen, + &i.RadioFreqMhz, + &i.RadioSf, + &i.RadioBwKhz, &i.Metadata, &i.IsObserver, &i.ObserverID, @@ -257,7 +263,7 @@ func (q *Queries) GetNodeByID(ctx context.Context, id uuid.UUID) (GetNodeByIDRow } const getNodeByPubkey = `-- name: GetNodeByPubkey :one -SELECT id, public_key, node_type, name, latitude, longitude, location_source, last_advert_at, supports_multibyte_paths, supports_multibyte_traces, min_firmware_version, first_seen, last_seen, metadata, +SELECT id, public_key, node_type, name, latitude, longitude, location_source, last_advert_at, supports_multibyte_paths, supports_multibyte_traces, min_firmware_version, first_seen, last_seen, radio_freq_mhz, radio_sf, radio_bw_khz, metadata, EXISTS (SELECT 1 FROM observers o WHERE o.public_key = n.public_key) AS is_observer, (SELECT o.id FROM observers o WHERE o.public_key = n.public_key LIMIT 1) AS observer_id FROM nodes n @@ -278,6 +284,9 @@ type GetNodeByPubkeyRow struct { MinFirmwareVersion *string `json:"min_firmware_version"` FirstSeen pgtype.Timestamptz `json:"first_seen"` LastSeen pgtype.Timestamptz `json:"last_seen"` + RadioFreqMhz *float32 `json:"radio_freq_mhz"` + RadioSf *int16 `json:"radio_sf"` + RadioBwKhz *float32 `json:"radio_bw_khz"` Metadata []byte `json:"metadata"` IsObserver bool `json:"is_observer"` ObserverID uuid.UUID `json:"observer_id"` @@ -300,6 +309,9 @@ func (q *Queries) GetNodeByPubkey(ctx context.Context, publicKey []byte) (GetNod &i.MinFirmwareVersion, &i.FirstSeen, &i.LastSeen, + &i.RadioFreqMhz, + &i.RadioSf, + &i.RadioBwKhz, &i.Metadata, &i.IsObserver, &i.ObserverID, @@ -1342,6 +1354,7 @@ func (q *Queries) ListNodeObservations(ctx context.Context, arg ListNodeObservat const listNodes = `-- name: ListNodes :many SELECT n.id, n.public_key, n.node_type, n.name, n.latitude, n.longitude, n.last_seen, + n.radio_freq_mhz, n.radio_sf, n.radio_bw_khz, array_remove(array_agg(DISTINCT ni.iata ORDER BY ni.iata), NULL)::text[] AS iatas, EXISTS (SELECT 1 FROM observers o WHERE o.public_key = n.public_key) AS is_observer, (SELECT o.id FROM observers o WHERE o.public_key = n.public_key LIMIT 1) AS observer_id @@ -1372,16 +1385,19 @@ type ListNodesParams struct { } type ListNodesRow struct { - ID uuid.UUID `json:"id"` - PublicKey []byte `json:"public_key"` - NodeType int16 `json:"node_type"` - Name *string `json:"name"` - Latitude *float64 `json:"latitude"` - Longitude *float64 `json:"longitude"` - LastSeen pgtype.Timestamptz `json:"last_seen"` - Iatas []string `json:"iatas"` - IsObserver bool `json:"is_observer"` - ObserverID uuid.UUID `json:"observer_id"` + ID uuid.UUID `json:"id"` + PublicKey []byte `json:"public_key"` + NodeType int16 `json:"node_type"` + Name *string `json:"name"` + Latitude *float64 `json:"latitude"` + Longitude *float64 `json:"longitude"` + LastSeen pgtype.Timestamptz `json:"last_seen"` + RadioFreqMhz *float32 `json:"radio_freq_mhz"` + RadioSf *int16 `json:"radio_sf"` + RadioBwKhz *float32 `json:"radio_bw_khz"` + Iatas []string `json:"iatas"` + IsObserver bool `json:"is_observer"` + ObserverID uuid.UUID `json:"observer_id"` } func (q *Queries) ListNodes(ctx context.Context, arg ListNodesParams) ([]ListNodesRow, error) { @@ -1410,6 +1426,9 @@ func (q *Queries) ListNodes(ctx context.Context, arg ListNodesParams) ([]ListNod &i.Latitude, &i.Longitude, &i.LastSeen, + &i.RadioFreqMhz, + &i.RadioSf, + &i.RadioBwKhz, &i.Iatas, &i.IsObserver, &i.ObserverID, @@ -1624,6 +1643,9 @@ SELECT o.display_name, o.observer_type, o.last_status_at, + o.radio_freq_mhz, + o.radio_sf, + o.radio_bw_khz, COALESCE(CASE WHEN o.last_status_at > NOW() - INTERVAL '5 minutes' THEN 'online' ELSE 'offline' @@ -1671,6 +1693,9 @@ type ListObserversRow struct { DisplayName *string `json:"display_name"` ObserverType *string `json:"observer_type"` LastStatusAt pgtype.Timestamptz `json:"last_status_at"` + RadioFreqMhz *float32 `json:"radio_freq_mhz"` + RadioSf *int16 `json:"radio_sf"` + RadioBwKhz *float32 `json:"radio_bw_khz"` Status string `json:"status"` Iata string `json:"iata"` } @@ -1699,6 +1724,9 @@ func (q *Queries) ListObservers(ctx context.Context, arg ListObserversParams) ([ &i.DisplayName, &i.ObserverType, &i.LastStatusAt, + &i.RadioFreqMhz, + &i.RadioSf, + &i.RadioBwKhz, &i.Status, &i.Iata, ); err != nil { @@ -2155,8 +2183,8 @@ func (q *Queries) UpsertIATADetails(ctx context.Context, arg UpsertIATADetailsPa const upsertNode = `-- name: UpsertNode :one -INSERT INTO nodes (public_key, node_type, name, latitude, longitude, location_source, last_advert_at, last_seen) -VALUES ($1, $2, $3, $4, $5, 'advert', NOW(), NOW()) +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) +VALUES ($1, $2, $3, $4, $5, 'advert', NOW(), NOW(), $6, $7, $8) ON CONFLICT (public_key) DO UPDATE SET node_type = EXCLUDED.node_type, name = COALESCE(EXCLUDED.name, nodes.name), @@ -2164,16 +2192,22 @@ ON CONFLICT (public_key) DO UPDATE SET longitude = COALESCE(EXCLUDED.longitude, nodes.longitude), location_source = CASE WHEN EXCLUDED.latitude IS NOT NULL THEN 'advert' ELSE nodes.location_source END, last_advert_at = NOW(), - last_seen = NOW() -RETURNING id, public_key, node_type, name, latitude, longitude, location_source, last_advert_at, supports_multibyte_paths, supports_multibyte_traces, min_firmware_version, first_seen, last_seen, metadata + last_seen = NOW(), + radio_freq_mhz = EXCLUDED.radio_freq_mhz, + radio_sf = EXCLUDED.radio_sf, + radio_bw_khz = EXCLUDED.radio_bw_khz +RETURNING id, public_key, node_type, name, latitude, longitude, location_source, last_advert_at, supports_multibyte_paths, supports_multibyte_traces, min_firmware_version, first_seen, last_seen, radio_freq_mhz, radio_sf, radio_bw_khz, metadata ` type UpsertNodeParams struct { - PublicKey []byte `json:"public_key"` - NodeType int16 `json:"node_type"` - Name *string `json:"name"` - Latitude *float64 `json:"latitude"` - Longitude *float64 `json:"longitude"` + PublicKey []byte `json:"public_key"` + NodeType int16 `json:"node_type"` + Name *string `json:"name"` + Latitude *float64 `json:"latitude"` + Longitude *float64 `json:"longitude"` + RadioFreqMhz *float32 `json:"radio_freq_mhz"` + RadioSf *int16 `json:"radio_sf"` + RadioBwKhz *float32 `json:"radio_bw_khz"` } // ============================================================ @@ -2186,6 +2220,9 @@ func (q *Queries) UpsertNode(ctx context.Context, arg UpsertNodeParams) (Node, e arg.Name, arg.Latitude, arg.Longitude, + arg.RadioFreqMhz, + arg.RadioSf, + arg.RadioBwKhz, ) var i Node err := row.Scan( @@ -2202,6 +2239,9 @@ func (q *Queries) UpsertNode(ctx context.Context, arg UpsertNodeParams) (Node, e &i.MinFirmwareVersion, &i.FirstSeen, &i.LastSeen, + &i.RadioFreqMhz, + &i.RadioSf, + &i.RadioBwKhz, &i.Metadata, ) return i, err diff --git a/db/store.go b/db/store.go index b9113c0..955651c 100644 --- a/db/store.go +++ b/db/store.go @@ -8,6 +8,7 @@ import ( "encoding/binary" "encoding/hex" "errors" + "fmt" "log" "time" @@ -132,7 +133,7 @@ func (s *Store) SetNodeCapability(ctx context.Context, nodeID uuid.UUID, paths, } // UpsertNode upserts a nodes row from an advert payload. -func (s *Store) UpsertNode(ctx context.Context, n ingest.UpsertNodeParams) (uuid.UUID, error) { +func (s *Store) UpsertNode(ctx context.Context, n ingest.UpsertNodeParams, radio ingest.RadioSettings) (uuid.UUID, error) { params := sqlc.UpsertNodeParams{ PublicKey: n.PublicKey, NodeType: int16(n.NodeType), @@ -140,6 +141,11 @@ func (s *Store) UpsertNode(ctx context.Context, n ingest.UpsertNodeParams) (uuid Latitude: n.Latitude, Longitude: n.Longitude, } + if radio.FreqMHz != 0 { + params.RadioFreqMhz = &radio.FreqMHz + params.RadioSf = &radio.SF + params.RadioBwKhz = &radio.BWKHz + } row, err := s.q.UpsertNode(ctx, params) if err != nil { return uuid.Nil, err @@ -582,6 +588,10 @@ func (s *Store) ListObservers(ctx context.Context, iata, observerType, broker, s IATA: v.Iata, Status: v.Status, } + if v.RadioFreqMhz != nil && v.RadioSf != nil && v.RadioBwKhz != nil { + s := fmt.Sprintf("%.1f,%g,%d", *v.RadioFreqMhz, *v.RadioBwKhz, *v.RadioSf) + observer.Radio = &s + } if v.DisplayName != nil { observer.DisplayName = v.DisplayName } @@ -831,7 +841,7 @@ func (s *Store) ListNodes(ctx context.Context, nodeType int16, iata string, supp } items := make([]api.NodeSummary, 0, len(rows)) for _, v := range rows { - items = append(items, api.NodeSummary{ + node := api.NodeSummary{ ID: v.ID, PublicKey: hex.EncodeToString(v.PublicKey), NodeType: v.NodeType, @@ -842,7 +852,12 @@ func (s *Store) ListNodes(ctx context.Context, nodeType int16, iata string, supp IATAs: v.Iatas, IsObserver: v.IsObserver, ObvserverID: nullableUUID(v.ObserverID), - }) + } + if v.RadioFreqMhz != nil && v.RadioSf != nil && v.RadioBwKhz != nil { + s := fmt.Sprintf("%.1f,%g,%d", *v.RadioFreqMhz, *v.RadioBwKhz, *v.RadioSf) + node.Radio = &s + } + items = append(items, node) } var nextCursor *int64 if hasMore && len(items) > 0 { @@ -888,6 +903,10 @@ func (s *Store) GetNode(ctx context.Context, nodeID uuid.UUID) (*api.Node, error LastSeen: row.LastSeen.Time.UnixMilli(), Metadata: row.Metadata, } + if row.RadioFreqMhz != nil && row.RadioSf != nil && row.RadioBwKhz != nil { + s := fmt.Sprintf("%.1f,%g,%d", *row.RadioFreqMhz, *row.RadioBwKhz, *row.RadioSf) + node.Radio = &s + } if row.LastAdvertAt.Valid { ms := row.LastAdvertAt.Time.UnixMilli() node.LastAdvertAt = &ms diff --git a/internal/api/reader.go b/internal/api/reader.go index 3156bf4..a574e82 100644 --- a/internal/api/reader.go +++ b/internal/api/reader.go @@ -191,6 +191,7 @@ type NodeSummary struct { ObvserverID *uuid.UUID `json:"observerId,omitempty"` Latitude *float64 `json:"lat,omitempty"` Longitude *float64 `json:"lng,omitempty"` + Radio *string `json:"radio,omitempty"` IATAs []string `json:"iatas"` } @@ -215,6 +216,7 @@ type ObserverSummary struct { ObserverType *string `json:"observerType,omitempty"` // e.g. "meshcoretomqtt", "meshcoreha" IATA string `json:"iata"` // most recently heard IATA Status string `json:"status"` // "online" or "offline" derived from last_status_at + Radio *string `json:"radio,omitempty"` } // ObserverBroker represents a single MQTT broker an observer has been seen on, diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index 26bfb4f..71485dc 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -81,7 +81,7 @@ type DB interface { SetNodeCapability(ctx context.Context, nodeID uuid.UUID, paths, traces bool) error // UpsertNode upserts a nodes row from an advert payload. - UpsertNode(ctx context.Context, n UpsertNodeParams) (uuid.UUID, error) + UpsertNode(ctx context.Context, n UpsertNodeParams, r RadioSettings) (uuid.UUID, error) // UpsertNodeIATA upserts a node_iatas row. UpsertNodeIATA(ctx context.Context, nodeID uuid.UUID, iata string) error @@ -513,7 +513,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ } if inserted { w.runCapabilityDetection(ctx, packet.PayloadType(), packet.PathHashSize(), resolvedIDs) - w.handlePayloadTypeSideEffects(ctx, packet, iata, packetHash[:]) + w.handlePayloadTypeSideEffects(ctx, packet, iata, packetHash[:], radio) evt := packetObservationEvent{} evt.PacketHash = hex.EncodeToString(packetHash[:]) evt.Packet.PayloadType = packet.PayloadType() @@ -706,7 +706,7 @@ func (w *Worker) runCapabilityDetection(ctx context.Context, payloadType uint8, // 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) { +func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshcore.Packet, iata string, packetHash []byte, radio RadioSettings) { if packet.PayloadType() == meshcore.PayloadTypeAdvert { advert, err := meshcore.AdvertFromBytes(packet.Payload) if err != nil { @@ -727,7 +727,7 @@ func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshc Latitude: lat, Longitude: lon, } - nodeID, err := w.db.UpsertNode(ctx, params) + nodeID, err := w.db.UpsertNode(ctx, params, radio) if err != nil { log.Printf("ingest[%s]: db: upsert node failed: %v", w.cfg.BrokerName, err) return