From fcc7bb1b23d82ef68eb1de8c7c5bcb6cecbf7ee0 Mon Sep 17 00:00:00 2001 From: "Enot (ded) Skelly" Date: Thu, 4 Jun 2026 10:32:03 -0700 Subject: [PATCH] add transport scopes new config for transport scope keys seeded to db new transport keystore to lookup matching keys and tag - packets with scope key - nodes with default scope - observers with all supported scopes it forwards allow filtering by region scope key on: packets, nodes, observers --- cmd/tower/main.go | 12 ++ config.yaml.example | 6 + db/migrations/001_schema.sql | 25 +++ db/queries/queries.sql | 66 +++++++- db/sqlc/models.go | 18 +++ db/sqlc/queries.sql.go | 244 ++++++++++++++++++++++++++--- db/store.go | 79 +++++++++- docs/docs.go | 43 +++++ docs/swagger.json | 43 +++++ docs/swagger.yaml | 30 ++++ internal/api/handlers/nodes.go | 4 +- internal/api/handlers/observers.go | 4 +- internal/api/handlers/packets.go | 4 +- internal/api/handlers/responses.go | 3 + internal/api/reader.go | 16 +- internal/config/config.go | 8 + internal/config/seed.go | 31 +++- internal/ingest/ingest.go | 78 ++++++++- internal/scopestore/scopestore.go | 39 +++++ 19 files changed, 709 insertions(+), 44 deletions(-) create mode 100644 internal/scopestore/scopestore.go diff --git a/cmd/tower/main.go b/cmd/tower/main.go index e63cb18..9b8772e 100644 --- a/cmd/tower/main.go +++ b/cmd/tower/main.go @@ -19,6 +19,7 @@ import ( "github.com/MeshCore-Tower/tower-server/internal/hub" "github.com/MeshCore-Tower/tower-server/internal/ingest" "github.com/MeshCore-Tower/tower-server/internal/keystore" + "github.com/MeshCore-Tower/tower-server/internal/scopestore" "github.com/jackc/pgx/v5/pgxpool" "github.com/joho/godotenv" @@ -116,6 +117,15 @@ func main() { log.Fatalf("failed to seed config: %v", err) } + // ── Build transport scope keystore ─────────────────────────────────────── + scopes := scopestore.New() + scopeEntries, err := store.GetTransportScopes(ctx) + if err != nil { + log.Fatalf("failed to load transport scopes: %v", err) + } + scopes.Load(scopeEntries) + log.Printf("loaded %d transport scopes", len(scopeEntries)) + // ── Build channel keystore ────────────────────────────────────────────── entries := make(map[string][]keystore.Entry) @@ -166,6 +176,7 @@ func main() { store, h, keys, + scopes, ) broker2 := ingest.New( @@ -179,6 +190,7 @@ func main() { store, h, keys, + scopes, ) go broker1.Start(ctx) diff --git a/config.yaml.example b/config.yaml.example index 96af526..88bca7b 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -59,6 +59,12 @@ channel_keys: key: "8b3387e9c5cdea6ac9e5edbaa115cd72" name: "Public" +# Regional transport scopes for matching TRANSPORT_FLOOD packets. +# Plain names have # prepended automatically (e.g. "bc" → "#bc"). +scopes: + - name: bc + - name: "#west" + telemetry: retention: 672h # 4 weeks resolution: 1h # store one snapshot per observer per hour diff --git a/db/migrations/001_schema.sql b/db/migrations/001_schema.sql index 21dddb6..6859cc1 100644 --- a/db/migrations/001_schema.sql +++ b/db/migrations/001_schema.sql @@ -40,6 +40,21 @@ CREATE TABLE region_iatas ( CREATE INDEX idx_region_iatas_iata ON region_iatas(iata); +-- ============================================================ +-- TRANSPORT SCOPES +-- ============================================================ + +CREATE TABLE transport_scopes ( + id SERIAL PRIMARY KEY, + name TEXT NOT NULL UNIQUE, -- normalized key string e.g. "#bc" + display_name TEXT, -- optional friendly name + transport_key BYTEA NOT NULL, -- 16-byte derived key + key_fingerprint BYTEA NOT NULL, -- SHA256(transport_key)[:8] for fast lookup + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX idx_transport_scopes_fingerprint ON transport_scopes(key_fingerprint); + -- ============================================================ -- NODES (must come before observers due to observer_owners FK) -- ============================================================ @@ -55,6 +70,7 @@ CREATE TABLE nodes ( last_advert_at TIMESTAMPTZ, supports_multibyte_paths BOOLEAN NOT NULL DEFAULT FALSE, supports_multibyte_traces BOOLEAN NOT NULL DEFAULT FALSE, + default_scope_id INT REFERENCES transport_scopes(id), min_firmware_version TEXT GENERATED ALWAYS AS ( CASE WHEN supports_multibyte_paths THEN '1.14.0+' @@ -128,6 +144,14 @@ CREATE TABLE observer_locations ( PRIMARY KEY (observer_id, reported_at) ); +CREATE TABLE observer_scopes ( + observer_id UUID NOT NULL REFERENCES observers(id) ON DELETE CASCADE, + scope_id INT NOT NULL REFERENCES transport_scopes(id) ON DELETE CASCADE, + first_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(), + last_seen TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (observer_id, scope_id) +); + CREATE INDEX idx_observer_locations_recent ON observer_locations(observer_id, reported_at DESC); CREATE TABLE observer_telemetry ( @@ -204,6 +228,7 @@ CREATE TABLE packets ( transport_codes_present BOOLEAN DEFAULT FALSE, region_code INT, sub_region_code INT, + scope_id INT REFERENCES transport_scopes(id), origin_pubkey BYTEA, raw_payload BYTEA NOT NULL, raw_header BYTEA NOT NULL, diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 62c3545..1f89d20 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -20,6 +20,24 @@ UPDATE iata_codes SET approx_lng = $4 WHERE iata = $1; +-- ============================================================ +-- TRANSPORT CODES +-- ============================================================ + +-- name: UpsertTransportScope :exec +INSERT INTO transport_scopes (name, display_name, transport_key, key_fingerprint) +VALUES ($1, $2, $3, $4) +ON CONFLICT (name) DO UPDATE SET + display_name = EXCLUDED.display_name, + transport_key = EXCLUDED.transport_key, + key_fingerprint = EXCLUDED.key_fingerprint; + +-- name: GetTransportScopes :many +SELECT name, transport_key, key_fingerprint FROM transport_scopes ORDER BY name; + +-- name: GetTransportScopeByName :one +SELECT id FROM transport_scopes WHERE name = $1; + -- ============================================================ -- OBSERVERS -- ============================================================ @@ -52,6 +70,18 @@ UPDATE observers SET WHERE public_key = $1 RETURNING id; +-- name: UpsertObserverScope :exec +INSERT INTO observer_scopes (observer_id, scope_id, last_seen) +VALUES ($1, $2, NOW()) +ON CONFLICT (observer_id, scope_id) DO UPDATE SET + last_seen = NOW(); + +-- name: GetObserverScopes :many +SELECT ts.name FROM observer_scopes os +JOIN transport_scopes ts ON ts.id = os.scope_id +WHERE os.observer_id = $1 +ORDER BY ts.name; + -- name: GetObserverByPubkey :one SELECT * FROM observers WHERE public_key = $1; @@ -75,6 +105,7 @@ SELECT o.radio_freq_mhz, o.radio_sf, o.radio_bw_khz, + array_remove(array_agg(DISTINCT ts.name ORDER BY ts.name), NULL)::text[] AS scopes, COALESCE(CASE WHEN o.last_status_at > NOW() - INTERVAL '5 minutes' THEN 'online' ELSE 'offline' @@ -88,6 +119,8 @@ COALESCE(( ), '')::text AS iata FROM observers o LEFT JOIN observer_brokers ob ON ob.observer_id = o.id +LEFT JOIN observer_scopes os ON os.observer_id = o.id +LEFT JOIN transport_scopes ts ON ts.id = os.scope_id WHERE ($1::text = '' OR ( SELECT po.iata FROM packet_observations po @@ -102,6 +135,11 @@ WHERE END = $4) AND ($5 = '' OR o.display_name ILIKE '%' || $5 || '%') AND ($6::timestamptz IS NULL OR o.last_seen < $6) + AND ($8::text = '' OR EXISTS ( + SELECT 1 FROM observer_scopes os2 + JOIN transport_scopes ts2 ON ts2.id = os2.scope_id + WHERE os2.observer_id = o.id AND ts2.name = $8::text + )) GROUP BY o.id ORDER BY o.last_seen DESC LIMIT $7; @@ -183,18 +221,22 @@ INSERT INTO packets ( raw_header, parsed_payload, channel_hash, + scope_id, first_heard_at, last_heard_at ) VALUES ( - $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, NOW(), NOW() + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, NOW(), NOW() ) ON CONFLICT (packet_hash) DO UPDATE SET - last_heard_at = NOW() + last_heard_at = NOW() RETURNING packet_hash, payload_type, payload_version, route_type, transport_codes_present, region_code, sub_region_code, origin_pubkey, raw_payload, raw_header, parsed_payload, decrypted, channel_hash, first_heard_at, last_heard_at, (xmax = 0) AS inserted; -- name: GetPacketByHash :one -SELECT * FROM packets WHERE packet_hash = $1; +SELECT p.*, ts.name AS scope_name +FROM packets p +LEFT JOIN transport_scopes ts ON ts.id = p.scope_id +WHERE p.packet_hash = $1; -- name: GetPacketObservationCount :one SELECT COUNT(*) FROM packet_observations WHERE packet_hash = $1; @@ -208,6 +250,8 @@ SELECT p.route_type, p.first_heard_at, p.last_heard_at, + p.scope_id, + ts.name AS scope_name, (SELECT COUNT(*) FROM packet_observations po2 WHERE po2.packet_hash = p.packet_hash) AS observation_count, po.observer_id AS latest_observer_id, o.display_name AS latest_observer_name, @@ -221,6 +265,7 @@ LEFT JOIN LATERAL ( LIMIT 1 ) po ON true LEFT JOIN observers o ON o.id = po.observer_id +LEFT JOIN transport_scopes ts ON ts.id = p.scope_id WHERE ($1::smallint = -1 OR p.payload_type = $1::smallint) AND ($2::smallint = -1 OR p.route_type = $2::smallint) @@ -232,6 +277,7 @@ WHERE AND ($4::timestamptz IS NULL OR p.first_heard_at >= $4) AND ($5::timestamptz IS NULL OR p.first_heard_at <= $5) AND ($6::timestamptz IS NULL OR p.last_heard_at < $6) + AND ($8::text = '' OR ts.name = $8::text) ORDER BY p.last_heard_at DESC LIMIT $7; @@ -319,33 +365,40 @@ WHERE id = $1 AND supports_multibyte_paths = FALSE; UPDATE nodes SET supports_multibyte_traces = TRUE WHERE id = $1 AND supports_multibyte_traces = FALSE; +-- name: SetNodeDefaultScope :exec +UPDATE nodes SET default_scope_id = $2 WHERE id = $1; + -- name: GetNodeByPubkey :one -SELECT *, +SELECT n.*, ts.name AS default_scope_name, 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, (SELECT json_agg(json_build_object('iata', ni.iata, 'lastHeard', (extract(epoch from ni.last_heard) * 1000)::bigint) ORDER BY ni.last_heard DESC) FROM node_iatas ni WHERE ni.node_id = n.id) AS iatas FROM nodes n +LEFT JOIN transport_scopes ts ON ts.id = n.default_scope_id WHERE n.public_key = $1; -- name: GetNodeByID :one -SELECT *, +SELECT n.*, ts.name AS default_scope_name, 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, (SELECT json_agg(json_build_object('iata', ni.iata, 'lastHeard', (extract(epoch from ni.last_heard) * 1000)::bigint) ORDER BY ni.last_heard DESC) FROM node_iatas ni WHERE ni.node_id = n.id) AS iatas FROM nodes n +LEFT JOIN transport_scopes ts ON ts.id = n.default_scope_id 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, + ts.name AS default_scope_name, json_agg(json_build_object('iata', ni.iata, 'lastHeard', (extract(epoch from ni.last_heard) * 1000)::bigint) ORDER BY ni.last_heard DESC) FILTER (WHERE ni.iata IS NOT NULL) 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 FROM nodes n LEFT JOIN node_iatas ni ON ni.node_id = n.id +LEFT JOIN transport_scopes ts ON ts.id = n.default_scope_id WHERE ($1 = 0 OR n.node_type = $1) AND ($2::text = '' OR n.id IN (SELECT node_id FROM node_iatas WHERE iata = ANY(string_to_array($2::text, ',')))) @@ -362,7 +415,8 @@ WHERE AND ($5::bytea IS NULL OR n.public_key = $5) AND ($6 = '' OR n.name ILIKE '%' || $6 || '%') AND ($7::timestamptz IS NULL OR n.last_seen < $7) -GROUP BY n.id + AND ($9::text = '' OR ts.name = $9::text) +GROUP BY n.id, ts.name ORDER BY n.last_seen DESC LIMIT $8; diff --git a/db/sqlc/models.go b/db/sqlc/models.go index 7001895..1699fab 100644 --- a/db/sqlc/models.go +++ b/db/sqlc/models.go @@ -84,6 +84,7 @@ type Node struct { LastAdvertAt pgtype.Timestamptz `json:"last_advert_at"` SupportsMultibytePaths bool `json:"supports_multibyte_paths"` SupportsMultibyteTraces bool `json:"supports_multibyte_traces"` + DefaultScopeID *int32 `json:"default_scope_id"` MinFirmwareVersion *string `json:"min_firmware_version"` FirstSeen pgtype.Timestamptz `json:"first_seen"` LastSeen pgtype.Timestamptz `json:"last_seen"` @@ -162,6 +163,13 @@ type ObserverOwner struct { UpdatedAt pgtype.Timestamptz `json:"updated_at"` } +type ObserverScope struct { + ObserverID uuid.UUID `json:"observer_id"` + ScopeID int32 `json:"scope_id"` + FirstSeen pgtype.Timestamptz `json:"first_seen"` + LastSeen pgtype.Timestamptz `json:"last_seen"` +} + type ObserverTelemetry struct { ID int64 `json:"id"` ObserverID uuid.UUID `json:"observer_id"` @@ -184,6 +192,7 @@ type Packet struct { TransportCodesPresent *bool `json:"transport_codes_present"` RegionCode *int32 `json:"region_code"` SubRegionCode *int32 `json:"sub_region_code"` + ScopeID *int32 `json:"scope_id"` OriginPubkey []byte `json:"origin_pubkey"` RawPayload []byte `json:"raw_payload"` RawHeader []byte `json:"raw_header"` @@ -232,3 +241,12 @@ type RegionIata struct { Iata string `json:"iata"` AddedAt pgtype.Timestamptz `json:"added_at"` } + +type TransportScope struct { + ID int32 `json:"id"` + Name string `json:"name"` + DisplayName *string `json:"display_name"` + TransportKey []byte `json:"transport_key"` + KeyFingerprint []byte `json:"key_fingerprint"` + CreatedAt pgtype.Timestamptz `json:"created_at"` +} diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 040a624..6fa0215 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -206,12 +206,13 @@ 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, radio_freq_mhz, radio_sf, radio_bw_khz, metadata, +SELECT n.id, n.public_key, n.node_type, n.name, n.latitude, n.longitude, n.location_source, n.last_advert_at, n.supports_multibyte_paths, n.supports_multibyte_traces, n.default_scope_id, n.min_firmware_version, n.first_seen, n.last_seen, n.radio_freq_mhz, n.radio_sf, n.radio_bw_khz, n.metadata, ts.name AS default_scope_name, 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, (SELECT json_agg(json_build_object('iata', ni.iata, 'lastHeard', (extract(epoch from ni.last_heard) * 1000)::bigint) ORDER BY ni.last_heard DESC) FROM node_iatas ni WHERE ni.node_id = n.id) AS iatas FROM nodes n +LEFT JOIN transport_scopes ts ON ts.id = n.default_scope_id WHERE n.id = $1 ` @@ -226,6 +227,7 @@ type GetNodeByIDRow struct { LastAdvertAt pgtype.Timestamptz `json:"last_advert_at"` SupportsMultibytePaths bool `json:"supports_multibyte_paths"` SupportsMultibyteTraces bool `json:"supports_multibyte_traces"` + DefaultScopeID *int32 `json:"default_scope_id"` MinFirmwareVersion *string `json:"min_firmware_version"` FirstSeen pgtype.Timestamptz `json:"first_seen"` LastSeen pgtype.Timestamptz `json:"last_seen"` @@ -233,6 +235,7 @@ type GetNodeByIDRow struct { RadioSf *int16 `json:"radio_sf"` RadioBwKhz *float32 `json:"radio_bw_khz"` Metadata []byte `json:"metadata"` + DefaultScopeName *string `json:"default_scope_name"` IsObserver bool `json:"is_observer"` ObserverID uuid.UUID `json:"observer_id"` Iatas []byte `json:"iatas"` @@ -252,6 +255,7 @@ func (q *Queries) GetNodeByID(ctx context.Context, id uuid.UUID) (GetNodeByIDRow &i.LastAdvertAt, &i.SupportsMultibytePaths, &i.SupportsMultibyteTraces, + &i.DefaultScopeID, &i.MinFirmwareVersion, &i.FirstSeen, &i.LastSeen, @@ -259,6 +263,7 @@ func (q *Queries) GetNodeByID(ctx context.Context, id uuid.UUID) (GetNodeByIDRow &i.RadioSf, &i.RadioBwKhz, &i.Metadata, + &i.DefaultScopeName, &i.IsObserver, &i.ObserverID, &i.Iatas, @@ -267,12 +272,13 @@ 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, radio_freq_mhz, radio_sf, radio_bw_khz, metadata, +SELECT n.id, n.public_key, n.node_type, n.name, n.latitude, n.longitude, n.location_source, n.last_advert_at, n.supports_multibyte_paths, n.supports_multibyte_traces, n.default_scope_id, n.min_firmware_version, n.first_seen, n.last_seen, n.radio_freq_mhz, n.radio_sf, n.radio_bw_khz, n.metadata, ts.name AS default_scope_name, 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, (SELECT json_agg(json_build_object('iata', ni.iata, 'lastHeard', (extract(epoch from ni.last_heard) * 1000)::bigint) ORDER BY ni.last_heard DESC) FROM node_iatas ni WHERE ni.node_id = n.id) AS iatas FROM nodes n +LEFT JOIN transport_scopes ts ON ts.id = n.default_scope_id WHERE n.public_key = $1 ` @@ -287,6 +293,7 @@ type GetNodeByPubkeyRow struct { LastAdvertAt pgtype.Timestamptz `json:"last_advert_at"` SupportsMultibytePaths bool `json:"supports_multibyte_paths"` SupportsMultibyteTraces bool `json:"supports_multibyte_traces"` + DefaultScopeID *int32 `json:"default_scope_id"` MinFirmwareVersion *string `json:"min_firmware_version"` FirstSeen pgtype.Timestamptz `json:"first_seen"` LastSeen pgtype.Timestamptz `json:"last_seen"` @@ -294,6 +301,7 @@ type GetNodeByPubkeyRow struct { RadioSf *int16 `json:"radio_sf"` RadioBwKhz *float32 `json:"radio_bw_khz"` Metadata []byte `json:"metadata"` + DefaultScopeName *string `json:"default_scope_name"` IsObserver bool `json:"is_observer"` ObserverID uuid.UUID `json:"observer_id"` Iatas []byte `json:"iatas"` @@ -313,6 +321,7 @@ func (q *Queries) GetNodeByPubkey(ctx context.Context, publicKey []byte) (GetNod &i.LastAdvertAt, &i.SupportsMultibytePaths, &i.SupportsMultibyteTraces, + &i.DefaultScopeID, &i.MinFirmwareVersion, &i.FirstSeen, &i.LastSeen, @@ -320,6 +329,7 @@ func (q *Queries) GetNodeByPubkey(ctx context.Context, publicKey []byte) (GetNod &i.RadioSf, &i.RadioBwKhz, &i.Metadata, + &i.DefaultScopeName, &i.IsObserver, &i.ObserverID, &i.Iatas, @@ -487,6 +497,33 @@ func (q *Queries) GetObserverRadio(ctx context.Context, id uuid.UUID) (GetObserv return i, err } +const getObserverScopes = `-- name: GetObserverScopes :many +SELECT ts.name FROM observer_scopes os +JOIN transport_scopes ts ON ts.id = os.scope_id +WHERE os.observer_id = $1 +ORDER BY ts.name +` + +func (q *Queries) GetObserverScopes(ctx context.Context, observerID uuid.UUID) ([]string, error) { + rows, err := q.db.Query(ctx, getObserverScopes, observerID) + if err != nil { + return nil, err + } + defer rows.Close() + items := []string{} + for rows.Next() { + var name string + if err := rows.Scan(&name); err != nil { + return nil, err + } + items = append(items, name) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const getObserverTelemetry = `-- name: GetObserverTelemetry :many SELECT id, reported_at, battery_voltage_mv, airtime_tx_pct, airtime_rx_pct, noise_floor_db, uptime_seconds, queue_length, debug_flags, receive_errors @@ -555,12 +592,35 @@ func (q *Queries) GetObserverTelemetry(ctx context.Context, arg GetObserverTelem } const getPacketByHash = `-- name: GetPacketByHash :one -SELECT packet_hash, payload_type, payload_version, route_type, transport_codes_present, region_code, sub_region_code, origin_pubkey, raw_payload, raw_header, parsed_payload, decrypted, channel_hash, first_heard_at, last_heard_at FROM packets WHERE packet_hash = $1 +SELECT p.packet_hash, p.payload_type, p.payload_version, p.route_type, p.transport_codes_present, p.region_code, p.sub_region_code, p.scope_id, p.origin_pubkey, p.raw_payload, p.raw_header, p.parsed_payload, p.decrypted, p.channel_hash, p.first_heard_at, p.last_heard_at, ts.name AS scope_name +FROM packets p +LEFT JOIN transport_scopes ts ON ts.id = p.scope_id +WHERE p.packet_hash = $1 ` -func (q *Queries) GetPacketByHash(ctx context.Context, packetHash []byte) (Packet, error) { +type GetPacketByHashRow struct { + PacketHash []byte `json:"packet_hash"` + PayloadType int16 `json:"payload_type"` + PayloadVersion int16 `json:"payload_version"` + RouteType int16 `json:"route_type"` + TransportCodesPresent *bool `json:"transport_codes_present"` + RegionCode *int32 `json:"region_code"` + SubRegionCode *int32 `json:"sub_region_code"` + ScopeID *int32 `json:"scope_id"` + OriginPubkey []byte `json:"origin_pubkey"` + RawPayload []byte `json:"raw_payload"` + RawHeader []byte `json:"raw_header"` + ParsedPayload []byte `json:"parsed_payload"` + Decrypted *bool `json:"decrypted"` + ChannelHash []byte `json:"channel_hash"` + FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` + LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` + ScopeName *string `json:"scope_name"` +} + +func (q *Queries) GetPacketByHash(ctx context.Context, packetHash []byte) (GetPacketByHashRow, error) { row := q.db.QueryRow(ctx, getPacketByHash, packetHash) - var i Packet + var i GetPacketByHashRow err := row.Scan( &i.PacketHash, &i.PayloadType, @@ -569,6 +629,7 @@ func (q *Queries) GetPacketByHash(ctx context.Context, packetHash []byte) (Packe &i.TransportCodesPresent, &i.RegionCode, &i.SubRegionCode, + &i.ScopeID, &i.OriginPubkey, &i.RawPayload, &i.RawHeader, @@ -577,6 +638,7 @@ func (q *Queries) GetPacketByHash(ctx context.Context, packetHash []byte) (Packe &i.ChannelHash, &i.FirstHeardAt, &i.LastHeardAt, + &i.ScopeName, ) return i, err } @@ -895,6 +957,47 @@ func (q *Queries) GetTopNodes(ctx context.Context, arg GetTopNodesParams) ([]MvT return items, nil } +const getTransportScopeByName = `-- name: GetTransportScopeByName :one +SELECT id FROM transport_scopes WHERE name = $1 +` + +func (q *Queries) GetTransportScopeByName(ctx context.Context, name string) (int32, error) { + row := q.db.QueryRow(ctx, getTransportScopeByName, name) + var id int32 + err := row.Scan(&id) + return id, err +} + +const getTransportScopes = `-- name: GetTransportScopes :many +SELECT name, transport_key, key_fingerprint FROM transport_scopes ORDER BY name +` + +type GetTransportScopesRow struct { + Name string `json:"name"` + TransportKey []byte `json:"transport_key"` + KeyFingerprint []byte `json:"key_fingerprint"` +} + +func (q *Queries) GetTransportScopes(ctx context.Context) ([]GetTransportScopesRow, error) { + rows, err := q.db.Query(ctx, getTransportScopes) + if err != nil { + return nil, err + } + defer rows.Close() + items := []GetTransportScopesRow{} + for rows.Next() { + var i GetTransportScopesRow + if err := rows.Scan(&i.Name, &i.TransportKey, &i.KeyFingerprint); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const insertChannelMessage = `-- name: InsertChannelMessage :one INSERT INTO channel_messages (channel_id, packet_hash, sender_name, content, sent_at) @@ -1432,11 +1535,13 @@ 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, + ts.name AS default_scope_name, json_agg(json_build_object('iata', ni.iata, 'lastHeard', (extract(epoch from ni.last_heard) * 1000)::bigint) ORDER BY ni.last_heard DESC) FILTER (WHERE ni.iata IS NOT NULL) 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 FROM nodes n LEFT JOIN node_iatas ni ON ni.node_id = n.id +LEFT JOIN transport_scopes ts ON ts.id = n.default_scope_id WHERE ($1 = 0 OR n.node_type = $1) AND ($2::text = '' OR n.id IN (SELECT node_id FROM node_iatas WHERE iata = ANY(string_to_array($2::text, ',')))) @@ -1453,7 +1558,8 @@ WHERE AND ($5::bytea IS NULL OR n.public_key = $5) AND ($6 = '' OR n.name ILIKE '%' || $6 || '%') AND ($7::timestamptz IS NULL OR n.last_seen < $7) -GROUP BY n.id + AND ($9::text = '' OR ts.name = $9::text) +GROUP BY n.id, ts.name ORDER BY n.last_seen DESC LIMIT $8 ` @@ -1467,22 +1573,24 @@ type ListNodesParams struct { Column6 interface{} `json:"column_6"` Column7 pgtype.Timestamptz `json:"column_7"` Limit int32 `json:"limit"` + Column9 string `json:"column_9"` } 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"` - RadioFreqMhz *float32 `json:"radio_freq_mhz"` - RadioSf *int16 `json:"radio_sf"` - RadioBwKhz *float32 `json:"radio_bw_khz"` - Iatas []byte `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"` + DefaultScopeName *string `json:"default_scope_name"` + Iatas []byte `json:"iatas"` + IsObserver bool `json:"is_observer"` + ObserverID uuid.UUID `json:"observer_id"` } func (q *Queries) ListNodes(ctx context.Context, arg ListNodesParams) ([]ListNodesRow, error) { @@ -1495,6 +1603,7 @@ func (q *Queries) ListNodes(ctx context.Context, arg ListNodesParams) ([]ListNod arg.Column6, arg.Column7, arg.Limit, + arg.Column9, ) if err != nil { return nil, err @@ -1514,6 +1623,7 @@ func (q *Queries) ListNodes(ctx context.Context, arg ListNodesParams) ([]ListNod &i.RadioFreqMhz, &i.RadioSf, &i.RadioBwKhz, + &i.DefaultScopeName, &i.Iatas, &i.IsObserver, &i.ObserverID, @@ -1731,6 +1841,7 @@ SELECT o.radio_freq_mhz, o.radio_sf, o.radio_bw_khz, + array_remove(array_agg(DISTINCT ts.name ORDER BY ts.name), NULL)::text[] AS scopes, COALESCE(CASE WHEN o.last_status_at > NOW() - INTERVAL '5 minutes' THEN 'online' ELSE 'offline' @@ -1744,6 +1855,8 @@ COALESCE(( ), '')::text AS iata FROM observers o LEFT JOIN observer_brokers ob ON ob.observer_id = o.id +LEFT JOIN observer_scopes os ON os.observer_id = o.id +LEFT JOIN transport_scopes ts ON ts.id = os.scope_id WHERE ($1::text = '' OR ( SELECT po.iata FROM packet_observations po @@ -1758,6 +1871,11 @@ WHERE END = $4) AND ($5 = '' OR o.display_name ILIKE '%' || $5 || '%') AND ($6::timestamptz IS NULL OR o.last_seen < $6) + AND ($8::text = '' OR EXISTS ( + SELECT 1 FROM observer_scopes os2 + JOIN transport_scopes ts2 ON ts2.id = os2.scope_id + WHERE os2.observer_id = o.id AND ts2.name = $8::text + )) GROUP BY o.id ORDER BY o.last_seen DESC LIMIT $7 @@ -1771,6 +1889,7 @@ type ListObserversParams struct { Column5 interface{} `json:"column_5"` Column6 pgtype.Timestamptz `json:"column_6"` Limit int32 `json:"limit"` + Column8 string `json:"column_8"` } type ListObserversRow struct { @@ -1781,6 +1900,7 @@ type ListObserversRow struct { RadioFreqMhz *float32 `json:"radio_freq_mhz"` RadioSf *int16 `json:"radio_sf"` RadioBwKhz *float32 `json:"radio_bw_khz"` + Scopes []string `json:"scopes"` Status string `json:"status"` Iata string `json:"iata"` } @@ -1796,6 +1916,7 @@ func (q *Queries) ListObservers(ctx context.Context, arg ListObserversParams) ([ arg.Column5, arg.Column6, arg.Limit, + arg.Column8, ) if err != nil { return nil, err @@ -1812,6 +1933,7 @@ func (q *Queries) ListObservers(ctx context.Context, arg ListObserversParams) ([ &i.RadioFreqMhz, &i.RadioSf, &i.RadioBwKhz, + &i.Scopes, &i.Status, &i.Iata, ); err != nil { @@ -1832,6 +1954,8 @@ SELECT p.route_type, p.first_heard_at, p.last_heard_at, + p.scope_id, + ts.name AS scope_name, (SELECT COUNT(*) FROM packet_observations po2 WHERE po2.packet_hash = p.packet_hash) AS observation_count, po.observer_id AS latest_observer_id, o.display_name AS latest_observer_name, @@ -1845,6 +1969,7 @@ LEFT JOIN LATERAL ( LIMIT 1 ) po ON true LEFT JOIN observers o ON o.id = po.observer_id +LEFT JOIN transport_scopes ts ON ts.id = p.scope_id WHERE ($1::smallint = -1 OR p.payload_type = $1::smallint) AND ($2::smallint = -1 OR p.route_type = $2::smallint) @@ -1856,6 +1981,7 @@ WHERE AND ($4::timestamptz IS NULL OR p.first_heard_at >= $4) AND ($5::timestamptz IS NULL OR p.first_heard_at <= $5) AND ($6::timestamptz IS NULL OR p.last_heard_at < $6) + AND ($8::text = '' OR ts.name = $8::text) ORDER BY p.last_heard_at DESC LIMIT $7 ` @@ -1868,6 +1994,7 @@ type ListPacketsParams struct { Column5 pgtype.Timestamptz `json:"column_5"` Column6 pgtype.Timestamptz `json:"column_6"` Limit int32 `json:"limit"` + Column8 string `json:"column_8"` } type ListPacketsRow struct { @@ -1876,6 +2003,8 @@ type ListPacketsRow struct { RouteType int16 `json:"route_type"` FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` + ScopeID *int32 `json:"scope_id"` + ScopeName *string `json:"scope_name"` ObservationCount int64 `json:"observation_count"` LatestObserverID uuid.UUID `json:"latest_observer_id"` LatestObserverName *string `json:"latest_observer_name"` @@ -1893,6 +2022,7 @@ func (q *Queries) ListPackets(ctx context.Context, arg ListPacketsParams) ([]Lis arg.Column5, arg.Column6, arg.Limit, + arg.Column8, ) if err != nil { return nil, err @@ -1907,6 +2037,8 @@ func (q *Queries) ListPackets(ctx context.Context, arg ListPacketsParams) ([]Lis &i.RouteType, &i.FirstHeardAt, &i.LastHeardAt, + &i.ScopeID, + &i.ScopeName, &i.ObservationCount, &i.LatestObserverID, &i.LatestObserverName, @@ -1923,7 +2055,7 @@ func (q *Queries) ListPackets(ctx context.Context, arg ListPacketsParams) ([]Lis } const listPacketsAfterID = `-- name: ListPacketsAfterID :many -SELECT p.packet_hash, p.payload_type, p.payload_version, p.route_type, p.transport_codes_present, p.region_code, p.sub_region_code, p.origin_pubkey, p.raw_payload, p.raw_header, p.parsed_payload, p.decrypted, p.channel_hash, p.first_heard_at, p.last_heard_at +SELECT p.packet_hash, p.payload_type, p.payload_version, p.route_type, p.transport_codes_present, p.region_code, p.sub_region_code, p.scope_id, p.origin_pubkey, p.raw_payload, p.raw_header, p.parsed_payload, p.decrypted, p.channel_hash, p.first_heard_at, p.last_heard_at FROM packets p JOIN packet_observations po ON po.packet_hash = p.packet_hash WHERE po.id > $1 @@ -1953,6 +2085,7 @@ func (q *Queries) ListPacketsAfterID(ctx context.Context, arg ListPacketsAfterID &i.TransportCodesPresent, &i.RegionCode, &i.SubRegionCode, + &i.ScopeID, &i.OriginPubkey, &i.RawPayload, &i.RawHeader, @@ -2109,6 +2242,20 @@ func (q *Queries) SetChannelKeyKnown(ctx context.Context, arg SetChannelKeyKnown return err } +const setNodeDefaultScope = `-- name: SetNodeDefaultScope :exec +UPDATE nodes SET default_scope_id = $2 WHERE id = $1 +` + +type SetNodeDefaultScopeParams struct { + ID uuid.UUID `json:"id"` + DefaultScopeID *int32 `json:"default_scope_id"` +} + +func (q *Queries) SetNodeDefaultScope(ctx context.Context, arg SetNodeDefaultScopeParams) error { + _, err := q.db.Exec(ctx, setNodeDefaultScope, arg.ID, arg.DefaultScopeID) + return err +} + const setNodeMultibytePaths = `-- name: SetNodeMultibytePaths :exec UPDATE nodes SET supports_multibyte_paths = TRUE WHERE id = $1 AND supports_multibyte_paths = FALSE @@ -2310,7 +2457,7 @@ ON CONFLICT (public_key) DO UPDATE SET 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 +RETURNING id, public_key, node_type, name, latitude, longitude, location_source, last_advert_at, supports_multibyte_paths, supports_multibyte_traces, default_scope_id, min_firmware_version, first_seen, last_seen, radio_freq_mhz, radio_sf, radio_bw_khz, metadata ` type UpsertNodeParams struct { @@ -2350,6 +2497,7 @@ func (q *Queries) UpsertNode(ctx context.Context, arg UpsertNodeParams) (Node, e &i.LastAdvertAt, &i.SupportsMultibytePaths, &i.SupportsMultibyteTraces, + &i.DefaultScopeID, &i.MinFirmwareVersion, &i.FirstSeen, &i.LastSeen, @@ -2463,6 +2611,23 @@ func (q *Queries) UpsertObserverBroker(ctx context.Context, arg UpsertObserverBr return err } +const upsertObserverScope = `-- name: UpsertObserverScope :exec +INSERT INTO observer_scopes (observer_id, scope_id, last_seen) +VALUES ($1, $2, NOW()) +ON CONFLICT (observer_id, scope_id) DO UPDATE SET + last_seen = NOW() +` + +type UpsertObserverScopeParams struct { + ObserverID uuid.UUID `json:"observer_id"` + ScopeID int32 `json:"scope_id"` +} + +func (q *Queries) UpsertObserverScope(ctx context.Context, arg UpsertObserverScopeParams) error { + _, err := q.db.Exec(ctx, upsertObserverScope, arg.ObserverID, arg.ScopeID) + return err +} + const upsertPacket = `-- name: UpsertPacket :one INSERT INTO packets ( @@ -2478,13 +2643,14 @@ INSERT INTO packets ( raw_header, parsed_payload, channel_hash, + scope_id, first_heard_at, last_heard_at ) VALUES ( - $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, NOW(), NOW() + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, NOW(), NOW() ) ON CONFLICT (packet_hash) DO UPDATE SET - last_heard_at = NOW() + last_heard_at = NOW() RETURNING packet_hash, payload_type, payload_version, route_type, transport_codes_present, region_code, sub_region_code, origin_pubkey, raw_payload, raw_header, parsed_payload, decrypted, channel_hash, first_heard_at, last_heard_at, (xmax = 0) AS inserted ` @@ -2502,6 +2668,7 @@ type UpsertPacketParams struct { RawHeader []byte `json:"raw_header"` ParsedPayload []byte `json:"parsed_payload"` ChannelHash []byte `json:"channel_hash"` + ScopeID *int32 `json:"scope_id"` } type UpsertPacketRow struct { @@ -2540,6 +2707,7 @@ func (q *Queries) UpsertPacket(ctx context.Context, arg UpsertPacketParams) (Ups arg.RawHeader, arg.ParsedPayload, arg.ChannelHash, + arg.ScopeID, ) var i UpsertPacketRow err := row.Scan( @@ -2617,3 +2785,33 @@ func (q *Queries) UpsertRegionIATA(ctx context.Context, arg UpsertRegionIATAPara _, err := q.db.Exec(ctx, upsertRegionIATA, arg.RegionID, arg.Iata) return err } + +const upsertTransportScope = `-- name: UpsertTransportScope :exec + +INSERT INTO transport_scopes (name, display_name, transport_key, key_fingerprint) +VALUES ($1, $2, $3, $4) +ON CONFLICT (name) DO UPDATE SET + display_name = EXCLUDED.display_name, + transport_key = EXCLUDED.transport_key, + key_fingerprint = EXCLUDED.key_fingerprint +` + +type UpsertTransportScopeParams struct { + Name string `json:"name"` + DisplayName *string `json:"display_name"` + TransportKey []byte `json:"transport_key"` + KeyFingerprint []byte `json:"key_fingerprint"` +} + +// ============================================================ +// TRANSPORT CODES +// ============================================================ +func (q *Queries) UpsertTransportScope(ctx context.Context, arg UpsertTransportScopeParams) error { + _, err := q.db.Exec(ctx, upsertTransportScope, + arg.Name, + arg.DisplayName, + arg.TransportKey, + arg.KeyFingerprint, + ) + return err +} diff --git a/db/store.go b/db/store.go index ca52b7f..09755b9 100644 --- a/db/store.go +++ b/db/store.go @@ -17,6 +17,7 @@ import ( sqlc "github.com/MeshCore-Tower/tower-server/db/sqlc" "github.com/MeshCore-Tower/tower-server/internal/api" "github.com/MeshCore-Tower/tower-server/internal/ingest" + "github.com/MeshCore-Tower/tower-server/internal/scopestore" "github.com/google/uuid" "github.com/jackc/pgx/v5" @@ -54,11 +55,60 @@ func (s *Store) UpsertObserverBroker(ctx context.Context, observerID uuid.UUID, return s.q.UpsertObserverBroker(ctx, params) } +func (s *Store) UpsertObserverScope(ctx context.Context, observerID uuid.UUID, scopeID int32) error { + return s.q.UpsertObserverScope(ctx, sqlc.UpsertObserverScopeParams{ + ObserverID: observerID, + ScopeID: scopeID, + }) +} + +func (s *Store) GetObserverScopes(ctx context.Context, observerID uuid.UUID) ([]string, error) { + return nil, nil +} + // UpsertIATA auto-creates an iata_codes row if it doesn't exist yet. func (s *Store) UpsertIATA(ctx context.Context, iata string) error { return s.q.UpsertIATA(ctx, iata) } +// UpsertTransportScope inserts or updates a transport scope derived from config. +// The transport key is SHA256(name)[:16] and the fingerprint is SHA256(key)[:8]. +// Called on startup via config.Seed(); safe to call multiple times. +func (s *Store) UpsertTransportScope(ctx context.Context, name, displayName string, transportKey, keyFingerprint []byte) error { + var dn *string + if displayName != "" { + dn = &displayName + } + return s.q.UpsertTransportScope(ctx, sqlc.UpsertTransportScopeParams{ + Name: name, + DisplayName: dn, + TransportKey: transportKey, + KeyFingerprint: keyFingerprint, + }) +} + +// GetTransportScopes returns all transport scope keys for loading into the scopestore. +func (s *Store) GetTransportScopes(ctx context.Context) ([]scopestore.Entry, error) { + rows, err := s.q.GetTransportScopes(ctx) + if err != nil { + return nil, err + } + entries := make([]scopestore.Entry, 0, len(rows)) + for _, r := range rows { + entries = append(entries, scopestore.Entry{ + Name: r.Name, + TransportKey: r.TransportKey, + KeyFingerprint: r.KeyFingerprint, + }) + } + return entries, nil +} + +// GetTransportScopeByName returns the ID of a transport scope by its normalized name. +func (s *Store) GetTransportScopeByName(ctx context.Context, name string) (int32, error) { + return s.q.GetTransportScopeByName(ctx, name) +} + // UpsertPacket inserts or bumps the packets row. Returns (isNew, error). func (s *Store) UpsertPacket(ctx context.Context, p ingest.UpsertPacketParams) (bool, error) { var regionCode, subRegionCode *int32 @@ -82,6 +132,7 @@ func (s *Store) UpsertPacket(ctx context.Context, p ingest.UpsertPacketParams) ( RawHeader: p.RawHeader, ParsedPayload: p.ParsedPayload, ChannelHash: p.ChannelHash, + ScopeID: p.ScopeID, } row, err := s.q.UpsertPacket(ctx, params) if err != nil { @@ -134,6 +185,15 @@ func (s *Store) SetNodeCapability(ctx context.Context, nodeID uuid.UUID, paths, return errors.Join(errs...) } +// SetNodeDefaultScope records the most recent scope attached to a node advert +// scopes are matched against configured regional transport scopes +func (s *Store) SetNodeDefaultScope(ctx context.Context, nodeID uuid.UUID, scopeID int32) error { + return s.q.SetNodeDefaultScope(ctx, sqlc.SetNodeDefaultScopeParams{ + ID: nodeID, + DefaultScopeID: &scopeID, + }) +} + // UpsertNode upserts a nodes row from an advert payload. func (s *Store) UpsertNode(ctx context.Context, n ingest.UpsertNodeParams, radio ingest.RadioSettings) (uuid.UUID, error) { params := sqlc.UpsertNodeParams{ @@ -612,7 +672,7 @@ func (s *Store) ListChannelMessagesByHash(ctx context.Context, hash []byte, sinc // status is "online" or "offline" derived from last_status_at recency. // ListObservers returns a paginated list of observers with optional filters. // cursor is last_seen epoch ms of the last observer; pass 0 to start from the beginning. -func (s *Store) ListObservers(ctx context.Context, iatas []string, observerType, broker, status, name string, cursor int64, limit int32) (api.Page[api.ObserverSummary], error) { +func (s *Store) ListObservers(ctx context.Context, iatas []string, observerType, broker, status, name, scope string, cursor int64, limit int32) (api.Page[api.ObserverSummary], error) { var cursorTS pgtype.Timestamptz if cursor > 0 { cursorTS = pgtype.Timestamptz{Time: time.UnixMilli(cursor), Valid: true} @@ -626,6 +686,7 @@ func (s *Store) ListObservers(ctx context.Context, iatas []string, observerType, Column5: name, Column6: cursorTS, Limit: limit + 1, + Column8: scope, } rows, err := s.q.ListObservers(ctx, params) if err != nil { @@ -641,6 +702,7 @@ func (s *Store) ListObservers(ctx context.Context, iatas []string, observerType, ID: v.ID, IATA: v.Iata, Status: v.Status, + Scopes: v.Scopes, } if v.RadioFreqMhz != nil && v.RadioSf != nil && v.RadioBwKhz != nil { s := fmt.Sprintf("%.1f,%g,%d", *v.RadioFreqMhz, *v.RadioBwKhz, *v.RadioSf) @@ -703,6 +765,12 @@ func (s *Store) GetObserver(ctx context.Context, observerID uuid.UUID) (*api.Obs LastSeen: obs.LastSeen.Time.UnixMilli(), ObservationCount: *obs.ObservationCount, } + scopes, err := s.GetObserverScopes(ctx, observerID) + if err != nil { + log.Printf("store: GetObserverScopes failed for %s: %v", observerID, err) + scopes = []string{} + } + observer.Scopes = scopes brokers := make([]api.ObserverBroker, 0, len(brokerRows)) for _, v := range brokerRows { var lastPacketAt int64 @@ -877,7 +945,7 @@ func (s *Store) ListObserverAdverts(ctx context.Context, observerID uuid.UUID, c // ListNodes returns a paginated list of nodes with optional filters. // Pass 0 for nodeType, empty string for iata/name, nil for pubkey to skip those filters. // cursor is last_seen epoch ms; pass 0 to start from the beginning. -func (s *Store) ListNodes(ctx context.Context, nodeType int16, iatas []string, supportsMultibytePaths, supportsMultibyteTraces *bool, pubkey []byte, name string, cursor int64, limit int32) (api.Page[api.NodeSummary], error) { +func (s *Store) ListNodes(ctx context.Context, nodeType int16, iatas []string, supportsMultibytePaths, supportsMultibyteTraces *bool, pubkey []byte, name, scope string, cursor int64, limit int32) (api.Page[api.NodeSummary], error) { var cursorTS pgtype.Timestamptz if cursor > 0 { cursorTS = pgtype.Timestamptz{Time: time.UnixMilli(cursor), Valid: true} @@ -892,6 +960,7 @@ func (s *Store) ListNodes(ctx context.Context, nodeType int16, iatas []string, s Column6: name, Column7: cursorTS, Limit: limit + 1, + Column9: scope, }) if err != nil { return api.Page[api.NodeSummary]{}, err @@ -955,6 +1024,7 @@ func (s *Store) GetNode(ctx context.Context, nodeID uuid.UUID) (*api.Node, error Longitude: row.Longitude, IsObserver: row.IsObserver, ObvserverID: nullableUUID(row.ObserverID), + DefaultScope: row.DefaultScopeName, }, LocationSource: row.LocationSource, SupportsMultibytePaths: row.SupportsMultibytePaths, @@ -1020,7 +1090,7 @@ func (s *Store) ListNodeObservations(ctx context.Context, nodeID uuid.UUID, curs }, nil } -func (s *Store) ListPackets(ctx context.Context, payloadType, routeType int16, iatas []string, since, until time.Time, cursor int64, limit int32) (api.Page[api.PacketSummary], error) { +func (s *Store) ListPackets(ctx context.Context, payloadType, routeType int16, iatas []string, scope string, since, until time.Time, cursor int64, limit int32) (api.Page[api.PacketSummary], error) { var cursorTS pgtype.Timestamptz if cursor > 0 { cursorTS = pgtype.Timestamptz{Time: time.UnixMilli(cursor), Valid: true} @@ -1042,6 +1112,7 @@ func (s *Store) ListPackets(ctx context.Context, payloadType, routeType int16, i Column5: untilTS, Column6: cursorTS, Limit: limit + 1, + Column8: scope, }) if err != nil { return api.Page[api.PacketSummary]{}, err @@ -1058,6 +1129,7 @@ func (s *Store) ListPackets(ctx context.Context, payloadType, routeType int16, i PayloadTypeName: api.PayloadTypeName(v.PayloadType), RouteType: v.RouteType, RouteTypeName: api.RouteTypeName(v.RouteType), + Scope: v.ScopeName, FirstHeardAt: v.FirstHeardAt.Time.UnixMilli(), LastHeardAt: v.LastHeardAt.Time.UnixMilli(), ObservationCount: int32(v.ObservationCount), @@ -1105,6 +1177,7 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet, ParsedPayload: row.ParsedPayload, RawPayload: hex.EncodeToString(row.RawPayload), Decrypted: row.Decrypted != nil && *row.Decrypted, + Scope: row.ScopeName, FirstHeardAt: row.FirstHeardAt.Time.UnixMilli(), LastHeardAt: row.LastHeardAt.Time.UnixMilli(), ObservationCount: int32(len(obsRows)), diff --git a/docs/docs.go b/docs/docs.go index acbe0cc..ecfb483 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -388,6 +388,12 @@ const docTemplate = `{ "name": "name", "in": "query" }, + { + "type": "string", + "description": "Filter by transport scope name e.g. %23bc (URL-encoded #bc)", + "name": "scope", + "in": "query" + }, { "type": "string", "description": "Exact public key match (hex)", @@ -591,6 +597,12 @@ const docTemplate = `{ "name": "name", "in": "query" }, + { + "type": "string", + "description": "Filter by transport scope name e.g. %23bc (URL-encoded #bc)", + "name": "scope", + "in": "query" + }, { "type": "integer", "description": "last_seen epoch ms of last item for pagination", @@ -816,6 +828,12 @@ const docTemplate = `{ "name": "iatas", "in": "query" }, + { + "type": "string", + "description": "Filter by transport scope name e.g. %23bc (URL-encoded #bc)", + "name": "scope", + "in": "query" + }, { "type": "integer", "description": "Filter by region ID, expands to member IATAs", @@ -1400,6 +1418,9 @@ const docTemplate = `{ "github_com_MeshCore-Tower_tower-server_internal_api.Node": { "type": "object", "properties": { + "defaultScope": { + "type": "string" + }, "firstSeen": { "description": "epoch ms", "type": "integer" @@ -1486,6 +1507,9 @@ const docTemplate = `{ "github_com_MeshCore-Tower_tower-server_internal_api.NodeSummary": { "type": "object", "properties": { + "defaultScope": { + "type": "string" + }, "iatas": { "type": "array", "items": { @@ -1605,6 +1629,7 @@ const docTemplate = `{ "type": "string" }, "radio": { + "description": "friendly radio param string: freqMhz,BwKhz,SF", "type": "string" }, "radioBwKhz": { @@ -1623,6 +1648,13 @@ const docTemplate = `{ "description": "LoRa spreading factor", "type": "integer" }, + "scopes": { + "description": "list of observer forwarded scopes matched to config", + "type": "array", + "items": { + "type": "string" + } + }, "softwareVersion": { "type": "string" }, @@ -1674,8 +1706,16 @@ const docTemplate = `{ "type": "string" }, "radio": { + "description": "friendly radio param string: freqMhz,BwKhz,SF", "type": "string" }, + "scopes": { + "description": "list of observer forwarded scopes matched to config", + "type": "array", + "items": { + "type": "string" + } + }, "status": { "description": "\"online\" or \"offline\" derived from last_status_at", "type": "string" @@ -1774,6 +1814,9 @@ const docTemplate = `{ "rawPayload": { "type": "string" }, + "scope": { + "type": "string" + }, "transportCodes": { "$ref": "#/definitions/github_com_MeshCore-Tower_tower-server_internal_api.PacketTransportCodes" } diff --git a/docs/swagger.json b/docs/swagger.json index aa4f6bb..feb8ca1 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -386,6 +386,12 @@ "name": "name", "in": "query" }, + { + "type": "string", + "description": "Filter by transport scope name e.g. %23bc (URL-encoded #bc)", + "name": "scope", + "in": "query" + }, { "type": "string", "description": "Exact public key match (hex)", @@ -589,6 +595,12 @@ "name": "name", "in": "query" }, + { + "type": "string", + "description": "Filter by transport scope name e.g. %23bc (URL-encoded #bc)", + "name": "scope", + "in": "query" + }, { "type": "integer", "description": "last_seen epoch ms of last item for pagination", @@ -814,6 +826,12 @@ "name": "iatas", "in": "query" }, + { + "type": "string", + "description": "Filter by transport scope name e.g. %23bc (URL-encoded #bc)", + "name": "scope", + "in": "query" + }, { "type": "integer", "description": "Filter by region ID, expands to member IATAs", @@ -1398,6 +1416,9 @@ "github_com_MeshCore-Tower_tower-server_internal_api.Node": { "type": "object", "properties": { + "defaultScope": { + "type": "string" + }, "firstSeen": { "description": "epoch ms", "type": "integer" @@ -1484,6 +1505,9 @@ "github_com_MeshCore-Tower_tower-server_internal_api.NodeSummary": { "type": "object", "properties": { + "defaultScope": { + "type": "string" + }, "iatas": { "type": "array", "items": { @@ -1603,6 +1627,7 @@ "type": "string" }, "radio": { + "description": "friendly radio param string: freqMhz,BwKhz,SF", "type": "string" }, "radioBwKhz": { @@ -1621,6 +1646,13 @@ "description": "LoRa spreading factor", "type": "integer" }, + "scopes": { + "description": "list of observer forwarded scopes matched to config", + "type": "array", + "items": { + "type": "string" + } + }, "softwareVersion": { "type": "string" }, @@ -1672,8 +1704,16 @@ "type": "string" }, "radio": { + "description": "friendly radio param string: freqMhz,BwKhz,SF", "type": "string" }, + "scopes": { + "description": "list of observer forwarded scopes matched to config", + "type": "array", + "items": { + "type": "string" + } + }, "status": { "description": "\"online\" or \"offline\" derived from last_status_at", "type": "string" @@ -1772,6 +1812,9 @@ "rawPayload": { "type": "string" }, + "scope": { + "type": "string" + }, "transportCodes": { "$ref": "#/definitions/github_com_MeshCore-Tower_tower-server_internal_api.PacketTransportCodes" } diff --git a/docs/swagger.yaml b/docs/swagger.yaml index 10c94da..6cf8bfe 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -112,6 +112,8 @@ definitions: type: object github_com_MeshCore-Tower_tower-server_internal_api.Node: properties: + defaultScope: + type: string firstSeen: description: epoch ms type: integer @@ -172,6 +174,8 @@ definitions: type: object github_com_MeshCore-Tower_tower-server_internal_api.NodeSummary: properties: + defaultScope: + type: string iatas: items: $ref: '#/definitions/github_com_MeshCore-Tower_tower-server_internal_api.NodeIATA' @@ -255,6 +259,7 @@ definitions: description: hex-encoded public key type: string radio: + description: 'friendly radio param string: freqMhz,BwKhz,SF' type: string radioBwKhz: description: bandwidth in kHz @@ -268,6 +273,11 @@ definitions: radioSf: description: LoRa spreading factor type: integer + scopes: + description: list of observer forwarded scopes matched to config + items: + type: string + type: array softwareVersion: type: string status: @@ -304,7 +314,13 @@ definitions: description: e.g. "meshcoretomqtt", "meshcoreha" type: string radio: + description: 'friendly radio param string: freqMhz,BwKhz,SF' type: string + scopes: + description: list of observer forwarded scopes matched to config + items: + type: string + type: array status: description: '"online" or "offline" derived from last_status_at' type: string @@ -370,6 +386,8 @@ definitions: type: array rawPayload: type: string + scope: + type: string transportCodes: $ref: '#/definitions/github_com_MeshCore-Tower_tower-server_internal_api.PacketTransportCodes' type: object @@ -928,6 +946,10 @@ paths: in: query name: name type: string + - description: 'Filter by transport scope name e.g. %23bc (URL-encoded #bc)' + in: query + name: scope + type: string - description: Exact public key match (hex) in: query name: pubkey @@ -1061,6 +1083,10 @@ paths: in: query name: name type: string + - description: 'Filter by transport scope name e.g. %23bc (URL-encoded #bc)' + in: query + name: scope + type: string - description: last_seen epoch ms of last item for pagination in: query name: cursor @@ -1210,6 +1236,10 @@ paths: in: query name: iatas type: string + - description: 'Filter by transport scope name e.g. %23bc (URL-encoded #bc)' + in: query + name: scope + type: string - description: Filter by region ID, expands to member IATAs in: query name: regionId diff --git a/internal/api/handlers/nodes.go b/internal/api/handlers/nodes.go index 0e4e917..a379c0a 100644 --- a/internal/api/handlers/nodes.go +++ b/internal/api/handlers/nodes.go @@ -37,6 +37,7 @@ func NodesRouter(reader api.Reader) http.Handler { // @Param regionId query int false "Filter by region ID, expands to member IATAs" // @Param region query string false "Filter by region slug, expands to member IATAs" // @Param name query string false "Partial case-insensitive name match" +// @Param scope query string false "Filter by transport scope name e.g. %23bc (URL-encoded #bc)" // @Param pubkey query string false "Exact public key match (hex)" // @Param supportsMultibytePaths query bool false "Filter by multibyte path support (true/false); omit for no filter" // @Param supportsMultibyteTraces query bool false "Filter by multibyte trace support (true/false); omit for no filter" @@ -96,6 +97,7 @@ func listNodes(reader api.Reader) http.HandlerFunc { iatas = append(iatas, regionIATAs...) } name := r.URL.Query().Get("name") + scope := r.URL.Query().Get("scope") var supportsMultibytePaths *bool if v := r.URL.Query().Get("supportsMultibytePaths"); v != "" { b, err := strconv.ParseBool(v) @@ -114,7 +116,7 @@ func listNodes(reader api.Reader) http.HandlerFunc { } supportsMultibyteTraces = &b } - nodes, err := reader.ListNodes(r.Context(), nodeType, iatas, supportsMultibytePaths, supportsMultibyteTraces, pubkey, name, cursor, limit) + nodes, err := reader.ListNodes(r.Context(), nodeType, iatas, supportsMultibytePaths, supportsMultibyteTraces, pubkey, name, scope, cursor, limit) if err != nil { respondError(w, http.StatusInternalServerError, "internal server error") return diff --git a/internal/api/handlers/observers.go b/internal/api/handlers/observers.go index 7b3e5b8..7583ce0 100644 --- a/internal/api/handlers/observers.go +++ b/internal/api/handlers/observers.go @@ -40,6 +40,7 @@ func ObserversRouter(reader api.Reader) http.Handler { // @Param broker query string false "Filter by broker name" // @Param status query string false "Filter by status (online or offline)" // @Param name query string false "Partial case-insensitive display name match" +// @Param scope query string false "Filter by transport scope name e.g. %23bc (URL-encoded #bc)" // @Param cursor query int false "last_seen epoch ms of last item for pagination" // @Param limit query int false "Max results (default 50)" // @Success 200 {object} api.Page[api.ObserverSummary] @@ -52,6 +53,7 @@ func listObservers(reader api.Reader) http.HandlerFunc { broker := r.URL.Query().Get("broker") name := r.URL.Query().Get("name") status := r.URL.Query().Get("status") + scope := r.URL.Query().Get("scope") var cursor int64 if cursorParam := r.URL.Query().Get("cursor"); cursorParam != "" { c, err := strconv.ParseInt(cursorParam, 10, 64) @@ -79,7 +81,7 @@ func listObservers(reader api.Reader) http.HandlerFunc { } iatas = append(iatas, regionIATAs...) } - observers, err := reader.ListObservers(r.Context(), iatas, observerType, broker, status, name, cursor, limit) + observers, err := reader.ListObservers(r.Context(), iatas, observerType, broker, status, name, scope, cursor, limit) if err != nil { respondError(w, http.StatusInternalServerError, "failed to get list of observers") return diff --git a/internal/api/handlers/packets.go b/internal/api/handlers/packets.go index e67cf68..7f051c6 100644 --- a/internal/api/handlers/packets.go +++ b/internal/api/handlers/packets.go @@ -31,6 +31,7 @@ func PacketsRouter(reader api.Reader) http.Handler { // @Param routeType query int false "Filter by route type (0=transport_flood, 1=flood, 2=direct, 3=transport_direct)" // @Param iata query string false "Filter by single IATA code (case-insensitive)" // @Param iatas query string false "Filter by multiple IATA codes, comma-separated e.g. YVR,YYJ" +// @Param scope query string false "Filter by transport scope name e.g. %23bc (URL-encoded #bc)" // @Param regionId query int false "Filter by region ID, expands to member IATAs" // @Param region query string false "Filter by region slug, expands to member IATAs" // @Param since query int false "Filter by first_heard_at >= since (epoch ms)" @@ -110,7 +111,8 @@ func listPackets(reader api.Reader) http.HandlerFunc { } iatas = append(iatas, regionIATAs...) } - packets, err := reader.ListPackets(r.Context(), payloadType, routeType, iatas, since, until, cursor, limit) + scope := r.URL.Query().Get("scope") + packets, err := reader.ListPackets(r.Context(), payloadType, routeType, iatas, scope, since, until, cursor, limit) if err != nil { respondError(w, http.StatusInternalServerError, "internal server error") return diff --git a/internal/api/handlers/responses.go b/internal/api/handlers/responses.go index fa835b7..7aa1544 100644 --- a/internal/api/handlers/responses.go +++ b/internal/api/handlers/responses.go @@ -29,6 +29,9 @@ func respond(w http.ResponseWriter, status int, data any) { // respondError writes a standard JSON error response. // The error code is derived automatically from the HTTP status text. func respondError(w http.ResponseWriter, status int, message string) { + if status >= 500 { + log.Printf("api: error %d: %s", status, message) + } code := strings.ToLower(strings.ReplaceAll(http.StatusText(status), " ", "_")) respond(w, status, map[string]APIError{"error": {Code: code, Message: message}}) } diff --git a/internal/api/reader.go b/internal/api/reader.go index 25150ca..9fb7bab 100644 --- a/internal/api/reader.go +++ b/internal/api/reader.go @@ -77,6 +77,7 @@ type PacketSummary struct { PayloadTypeName string `json:"payloadTypeName"` RouteType int16 `json:"routeType"` RouteTypeName string `json:"routeTypeName"` + Scope *string `json:"scope,omitempty"` FirstHeardAt int64 `json:"firstHeardAt"` // epoch ms LastHeardAt int64 `json:"lastHeardAt"` // epoch ms ObservationCount int32 `json:"observationCount"` @@ -162,6 +163,7 @@ type Packet struct { RawPayload string `json:"rawPayload"` Decrypted bool `json:"decrypted"` ChannelHash *string `json:"channelHash,omitempty"` + Scope *string `json:"scope,omitempty"` FirstHeardAt int64 `json:"firstHeardAt"` LastHeardAt int64 `json:"lastHeardAt"` FirstToLastMs int64 `json:"firstToLastMs"` @@ -230,6 +232,7 @@ type NodeSummary struct { Longitude *float64 `json:"lng,omitempty"` Radio *string `json:"radio,omitempty"` IATAs []NodeIATA `json:"iatas"` + DefaultScope *string `json:"defaultScope,omitempty"` } // Node is the full node representation including firmware capability flags, @@ -253,7 +256,8 @@ 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"` + Radio *string `json:"radio,omitempty"` // friendly radio param string: freqMhz,BwKhz,SF + Scopes []string `json:"scopes,omitempty"` // list of observer forwarded scopes matched to config } // ObserverBroker represents a single MQTT broker an observer has been seen on, @@ -399,7 +403,7 @@ type Reader interface { // All filter params are optional — pass empty string or nil to skip a filter. // status is "online" or "offline" derived from last_status_at recency. // cursor is last_seen epoch ms of the last observer; pass 0 to start from the beginning. - ListObservers(ctx context.Context, iatas []string, observerType, broker, status, name string, cursor int64, limit int32) (Page[ObserverSummary], error) + ListObservers(ctx context.Context, iatas []string, observerType, broker, status, name, scope string, cursor int64, limit int32) (Page[ObserverSummary], error) // GetObserver returns full detail for a single observer by UUID. // Returns nil, pgx.ErrNoRows if the observer is not found. GetObserver(ctx context.Context, observerID uuid.UUID) (*Observer, error) @@ -408,13 +412,17 @@ type Reader interface { GetObserverTelemetry(ctx context.Context, observerID uuid.UUID, since, until time.Time, afterID int64) (*ObserverTelemetry, error) // TODO: add interval time.Duration param for server-side bucketing + // GetObserverScopes returns the names of all transport scopes an observer has + // been seen forwarding packets for, ordered alphabetically. + GetObserverScopes(ctx context.Context, observerID uuid.UUID) ([]string, error) + // ListObserverAdverts returns a paginated list of advert packets heard by an observer. // Pass cursor=0 to start from the beginning. ListObserverAdverts(ctx context.Context, observerID uuid.UUID, cursor int64, limit int32) (Page[AdvertObservation], error) // ListNodes returns a paginated list of nodes with optional filters. // Pass 0 for nodeType, nil iatas, nil for pubkey to skip those filters. // cursor is last_seen epoch ms; pass 0 to start from the beginning. - ListNodes(ctx context.Context, nodeType int16, iatas []string, supportsMultibytePaths, supportsMultibyteTraces *bool, pubkey []byte, name string, cursor int64, limit int32) (Page[NodeSummary], error) + ListNodes(ctx context.Context, nodeType int16, iatas []string, supportsMultibytePaths, supportsMultibyteTraces *bool, pubkey []byte, name, scope string, cursor int64, limit int32) (Page[NodeSummary], error) // GetNode returns full detail for a single node by UUID. // Returns nil, pgx.ErrNoRows if the node is not found. @@ -426,7 +434,7 @@ type Reader interface { // Pass 0 for payloadType/routeType to skip those filters. // Pass nil for iatas, zero times for since/until to skip those filters. // cursor is last_heard_at epoch ms; pass 0 to start from the beginning. - ListPackets(ctx context.Context, payloadType, routeType int16, iatas []string, since, until time.Time, cursor int64, limit int32) (Page[PacketSummary], error) + ListPackets(ctx context.Context, payloadType, routeType int16, iatas []string, scope string, since, until time.Time, cursor int64, limit int32) (Page[PacketSummary], error) // GetPacket returns full packet detail including all observations with radio settings. // Returns nil, pgx.ErrNoRows if not found. GetPacket(ctx context.Context, packetHash []byte) (*Packet, error) diff --git a/internal/config/config.go b/internal/config/config.go index 631b804..5d638eb 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -17,6 +17,14 @@ type Config struct { Telemetry TelemetryConfig `yaml:"telemetry"` WebSocket WebSocketConfig `yaml:"websocket"` Packets PacketsConfig `yaml:"packets"` + Scopes []ScopeConfig `yaml:"scopes"` +} + +// ScopeConfig defines a regional transport scope. +// Name can be provided with or without the # or $ prefix. +// Tower normalizes plain names by prepending #. +type ScopeConfig struct { + Name string `yaml:"name"` // e.g. "bc", "#west", "$private" } // TelemetryConfig controls observer telemetry storage behaviour. diff --git a/internal/config/seed.go b/internal/config/seed.go index a848ccd..6c3889c 100644 --- a/internal/config/seed.go +++ b/internal/config/seed.go @@ -2,7 +2,9 @@ package config import ( "context" + "crypto/sha256" "log" + "strings" ) // Seeder is the database interface required to seed config data on startup. @@ -11,12 +13,13 @@ type Seeder interface { UpsertIATADetails(ctx context.Context, iata string, name string, lat, lng *float64) error UpsertRegion(ctx context.Context, slug, name, description string, displayOrder int, centerLat, centerLng *float64, zoomLevel *int) (int32, error) UpsertRegionIATA(ctx context.Context, regionID int32, iata string) error + UpsertTransportScope(ctx context.Context, name, displayName string, transportKey, keyFingerprint []byte) error } // Seed applies config-defined regions, IATA overrides to the database. // It is safe to call on every startup — all operations are upserts. func Seed(ctx context.Context, cfg *Config, db Seeder) error { - log.Printf("config: seeding %d IATAs, %d regions", len(cfg.IATAs), len(cfg.Regions)) + log.Printf("config: seeding %d IATAs, %d regions, %d scopes", len(cfg.IATAs), len(cfg.Regions), len(cfg.Scopes)) // IATA overrides for iata, details := range cfg.IATAs { if err := db.UpsertIATADetails(ctx, iata, details.Name, details.Lat, details.Lng); err != nil { @@ -38,5 +41,31 @@ func Seed(ctx context.Context, cfg *Config, db Seeder) error { } } } + // Transport Codes + for _, s := range cfg.Scopes { + name := normalizeScopeName(s.Name) + key := deriveScopeKey(name) + h := sha256.Sum256(key) + fingerprint := h[:8] + if err := db.UpsertTransportScope(ctx, name, "", key, fingerprint); err != nil { + return err + } + } return nil } + +// normalizeScopeName ensures the scope name has a # or $ prefix. +// Plain names get # prepended: "bc" → "#bc". +func normalizeScopeName(name string) string { + if strings.HasPrefix(name, "#") || strings.HasPrefix(name, "$") { + return name + } + return "#" + name +} + +// deriveScopeKey derives the 16-byte transport key from a normalized scope name. +// key = SHA256(name)[:16] +func deriveScopeKey(name string) []byte { + h := sha256.Sum256([]byte(name)) + return h[:16] +} diff --git a/internal/ingest/ingest.go b/internal/ingest/ingest.go index 4deaf3a..a3c3468 100644 --- a/internal/ingest/ingest.go +++ b/internal/ingest/ingest.go @@ -21,6 +21,8 @@ package ingest import ( "context" + "crypto/hmac" + "crypto/sha256" "encoding/binary" "encoding/hex" "encoding/json" @@ -37,6 +39,7 @@ import ( "github.com/MeshCore-Tower/tower-server/internal/api" "github.com/MeshCore-Tower/tower-server/internal/hub" "github.com/MeshCore-Tower/tower-server/internal/keystore" + "github.com/MeshCore-Tower/tower-server/internal/scopestore" ) // Config holds the connection parameters for one broker. @@ -81,6 +84,10 @@ type DB interface { // for a node, never downgrading an existing TRUE. SetNodeCapability(ctx context.Context, nodeID uuid.UUID, paths, traces bool) error + // SetNodeDefaultScope records the most recent scope attached to a node advert + // scopes are matched against configured regional transport scopes + SetNodeDefaultScope(ctx context.Context, nodeID uuid.UUID, scopeID int32) error + // UpsertNode upserts a nodes row from an advert payload. UpsertNode(ctx context.Context, n UpsertNodeParams, r RadioSettings) (uuid.UUID, error) @@ -121,6 +128,12 @@ type DB interface { UpsertChannelHashOnly(ctx context.Context, channelHash []byte) (int, error) // GetPacketObservationCount returns the number of rows for the packet observations GetPacketObservationCount(ctx context.Context, packetHash []byte) (int64, error) + // GetTransportScopeByName returns the ID of a transport scope by its normalized name. + GetTransportScopeByName(ctx context.Context, name string) (int32, error) + // UpsertObserverScope records or updates a scope association for an observer. + // 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 } // UpsertPacketParams mirrors the columns written on packets upsert. @@ -135,6 +148,7 @@ type UpsertPacketParams struct { ParsedPayload json.RawMessage OriginPubkey []byte ChannelHash []byte + ScopeID *int32 } // InsertObservationParams mirrors the columns written on packet_observations insert. @@ -365,18 +379,24 @@ type ChannelKeyStore interface { GetKey(channelHash []byte) []keystore.Entry } +// ScopeStore provides transport scope key lookup for matching TRANSPORT_FLOOD packets. +type ScopeStore interface { + Entries() []scopestore.Entry +} + // Worker holds the dependencies for one broker's ingest loop. type Worker struct { cfg Config db DB hub *hub.Hub keys ChannelKeyStore + scopes ScopeStore client mqtt.Client } // New creates an ingest Worker. Call Start() to connect and begin processing. -func New(cfg Config, db DB, h *hub.Hub, keys ChannelKeyStore) *Worker { - return &Worker{cfg: cfg, db: db, hub: h, keys: keys} +func New(cfg Config, db DB, h *hub.Hub, keys ChannelKeyStore, scopes ScopeStore) *Worker { + return &Worker{cfg: cfg, db: db, hub: h, keys: keys, scopes: scopes} } // Start connects to the broker and blocks until ctx is cancelled. It @@ -773,6 +793,26 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ parsedPayload, _ = json.Marshal(pr) } + var matchedScope *string + if packet.RouteType() == meshcore.RouteTypeTransportFlood || packet.RouteType() == meshcore.RouteTypeTransportDirect { + for _, entry := range w.scopes.Entries() { + code := computeTransportCode(entry.TransportKey, packet.PayloadType(), packet.Payload) + if code == packet.TransportCode1 { + s := entry.Name + matchedScope = &s + break + } + } + } + var scopeID *int32 + if matchedScope != nil { + id, err := w.db.GetTransportScopeByName(ctx, *matchedScope) + if err != nil { + log.Printf("ingest[%s]: failed to get scope ID for %s: %v", w.cfg.BrokerName, *matchedScope, err) + } else { + scopeID = &id + } + } rawHeader := []byte{packet.Header} if transportCodes != nil { rawHeader = append(rawHeader, transportCodes...) @@ -788,6 +828,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ ParsedPayload: parsedPayload, OriginPubkey: originPubkey, ChannelHash: channelHash, + ScopeID: scopeID, } isNew, err := w.db.UpsertPacket(ctx, pParams) if err != nil { @@ -831,6 +872,12 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ return } + if scopeID != nil && inserted { + if err := w.db.UpsertObserverScope(ctx, id, *scopeID); err != nil { + log.Printf("ingest[%s]: failed to upsert observer scope for %s: %v", w.cfg.BrokerName, id, err) + } + } + resolved, err := w.db.ResolvePathHashes(ctx, iata, packet.PathHashes()) if err != nil { log.Printf("ingest[%s]: path resolution failed: %v", w.cfg.BrokerName, err) @@ -843,7 +890,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ } w.runCapabilityDetection(ctx, packet.PayloadType(), packet.PathHashSize(), resolvedIDs) if inserted { - w.handlePayloadTypeSideEffects(ctx, packet, iata, packetHash[:], radio) + w.handlePayloadTypeSideEffects(ctx, packet, iata, packetHash[:], radio, scopeID) evt := packetObservationEvent{} evt.PacketHash = hex.EncodeToString(packetHash[:]) evt.Packet.PayloadType = packet.PayloadType() @@ -1045,7 +1092,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, radio RadioSettings) { +func (w *Worker) handlePayloadTypeSideEffects(ctx context.Context, packet *meshcore.Packet, iata string, packetHash []byte, radio RadioSettings, scopeID *int32) { if packet.PayloadType() == meshcore.PayloadTypeAdvert { advert, err := meshcore.AdvertFromBytes(packet.Payload) if err != nil { @@ -1074,6 +1121,11 @@ 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) } + if scopeID != nil && (packet.RouteType() == meshcore.RouteTypeTransportFlood || packet.RouteType() == meshcore.RouteTypeTransportDirect) { + if err := w.db.SetNodeDefaultScope(ctx, nodeID, *scopeID); err != nil { + log.Printf("ingest[%s]: failed to set default scope for node %s: %v", w.cfg.BrokerName, hex.EncodeToString(advert.PublicKey.PublicKeyBytes()), 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) @@ -1226,3 +1278,21 @@ func uint32ToBytes(v uint32) []byte { binary.LittleEndian.PutUint32(b, v) return b } + +// computeTransportCode derives transport_code_1 from a transport key and packet payload. +// code = HMAC-SHA256(key, payload_type_byte || payload)[0:2] as little-endian uint16. +// Coerces reserved values 0x0000 → 0x0001 and 0xFFFF → 0xFFFE per §2.4. +func computeTransportCode(key []byte, payloadType uint8, payload []byte) uint16 { + mac := hmac.New(sha256.New, key) + mac.Write([]byte{payloadType}) + mac.Write(payload) + sum := mac.Sum(nil) + code := uint16(sum[0]) | uint16(sum[1])<<8 + if code == 0x0000 { + code = 0x0001 + } + if code == 0xFFFF { + code = 0xFFFE + } + return code +} diff --git a/internal/scopestore/scopestore.go b/internal/scopestore/scopestore.go new file mode 100644 index 0000000..4b4e69b --- /dev/null +++ b/internal/scopestore/scopestore.go @@ -0,0 +1,39 @@ +// Package scopestore provides an in-memory lookup of transport scope keys +// loaded from the database at startup. +package scopestore + +import "sync" + +// Entry holds a single transport scope key and its metadata. +type Entry struct { + Name string + TransportKey []byte // 16 bytes + KeyFingerprint []byte // 8 bytes +} + +// ScopeStore holds all known transport scope keys in memory. +type ScopeStore struct { + mu sync.RWMutex + entries []Entry +} + +// New creates an empty ScopeStore. +func New() *ScopeStore { + return &ScopeStore{} +} + +// Load replaces all entries — call on startup after DB seeding. +func (s *ScopeStore) Load(entries []Entry) { + s.mu.Lock() + defer s.mu.Unlock() + s.entries = entries +} + +// Entries returns a copy of all loaded entries. +func (s *ScopeStore) Entries() []Entry { + s.mu.RLock() + defer s.mu.RUnlock() + result := make([]Entry, len(s.entries)) + copy(result, s.entries) + return result +}