From f0c57fde45e724d45ed6715f42e94637570c211f Mon Sep 17 00:00:00 2001 From: "Enot (ded) Skelly" Date: Fri, 5 Jun 2026 15:20:53 -0700 Subject: [PATCH] add traces endpoint --- db/migrations/001_schema.sql | 2 + db/packets.go | 84 ++------------ db/queries/queries.sql | 41 ++++++- db/sqlc/models.go | 1 + db/sqlc/queries.sql.go | 137 ++++++++++++++++++++++- db/store.go | 71 ++++++++++++ db/traces.go | 90 +++++++++++++++ docs/docs.go | 189 ++++++++++++++++++++++++++++++++ docs/swagger.json | 189 ++++++++++++++++++++++++++++++++ docs/swagger.yaml | 126 +++++++++++++++++++++ internal/api/handlers/traces.go | 106 ++++++++++++++++++ internal/api/reader.go | 4 + internal/api/router/router.go | 1 + internal/api/traces.go | 28 +++++ internal/ingest/packet.go | 4 + 15 files changed, 995 insertions(+), 78 deletions(-) create mode 100644 db/traces.go create mode 100644 internal/api/handlers/traces.go create mode 100644 internal/api/traces.go diff --git a/db/migrations/001_schema.sql b/db/migrations/001_schema.sql index 8c576f3..5487156 100644 --- a/db/migrations/001_schema.sql +++ b/db/migrations/001_schema.sql @@ -235,6 +235,7 @@ CREATE TABLE packets ( parsed_payload JSONB, decrypted BOOLEAN DEFAULT FALSE, channel_hash BYTEA, + trace_tag BYTEA, first_heard_at TIMESTAMPTZ NOT NULL, last_heard_at TIMESTAMPTZ NOT NULL ); @@ -246,6 +247,7 @@ CREATE INDEX idx_packets_origin ON packets(origin_pubkey, first_heard_at DESC) WHERE origin_pubkey IS NOT NULL; CREATE INDEX idx_packets_channel ON packets(channel_hash, first_heard_at DESC) WHERE channel_hash IS NOT NULL; + CREATE INDEX idx_packets_trace_tag ON packets(trace_tag) WHERE trace_tag IS NOT NULL; -- ============================================================ -- PACKET OBSERVATIONS diff --git a/db/packets.go b/db/packets.go index 3178e24..831d485 100644 --- a/db/packets.go +++ b/db/packets.go @@ -43,6 +43,7 @@ func (s *Store) UpsertPacket(ctx context.Context, p ingest.UpsertPacketParams) ( ParsedPayload: p.ParsedPayload, ChannelHash: p.ChannelHash, ScopeID: p.ScopeID, + TraceTag: p.TraceTag, } row, err := s.q.UpsertPacket(ctx, params) if err != nil { @@ -323,82 +324,15 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet, p.Observations = append(p.Observations, obs) } if row.PayloadType == 9 && len(obsRows) > 0 { - var tracePayload struct { - PathHashes []string `json:"pathHashes"` - Flags byte `json:"flags"` - } - if err := json.Unmarshal(row.ParsedPayload, &tracePayload); err == nil && len(tracePayload.PathHashes) > 0 { - hashSize := int(1 << (tracePayload.Flags & 0x03)) - hashes := make([][]byte, 0, len(tracePayload.PathHashes)) - for _, h := range tracePayload.PathHashes { - b, err := hex.DecodeString(h) - if err == nil { - hashes = append(hashes, b) - } - } - - // collect unique IATAs from observations - seenIATAs := make(map[string]struct{}) - for _, v := range obsRows { - seenIATAs[v.Iata] = struct{}{} - } - - // merge results across all IATAs — best confidence per hop wins - // confidence priority: high > ambiguous > none - type hopResult struct { - confidence string - entries []api.ResolvedPathEntry - } - merged := make([]hopResult, len(hashes)) - for i := range merged { - merged[i] = hopResult{confidence: "none"} - } - - confidenceRank := map[string]int{"none": 0, "ambiguous": 1, "high": 2} - - for iata := range seenIATAs { - resolved, err := s.ResolvePathHashes(ctx, iata, hashes) - if err != nil { - log.Printf("store: trace route resolution failed for iata=%s: %v", iata, err) - continue - } - for i, hash := range hashes { - key := hex.EncodeToString(hash[:hashSize]) - entries := resolved[key] - var confidence string - switch len(entries) { - case 0: - confidence = "none" - case 1: - confidence = "high" - default: - confidence = "ambiguous" - } - if confidenceRank[confidence] > confidenceRank[merged[i].confidence] { - merged[i] = hopResult{confidence: confidence, entries: entries} - } - } - } - - route := make([]api.ResolvedHop, 0, len(hashes)) - for _, hr := range merged { - hop := api.ResolvedHop{ - Confidence: hr.confidence, - Nodes: make([]api.ResolvedNode, 0, len(hr.entries)), - } - for _, e := range hr.entries { - hop.Nodes = append(hop.Nodes, api.ResolvedNode{ - ID: e.NodeID, - Name: e.Name, - Latitude: e.Latitude, - Longitude: e.Longitude, - PublicKey: hex.EncodeToString(e.PublicKey), - }) - } - route = append(route, hop) - } - p.ResolvedRoute = route + iatas := make([]string, 0, len(obsRows)) + seen := make(map[string]struct{}) + for _, v := range obsRows { + if _, ok := seen[v.Iata]; !ok { + seen[v.Iata] = struct{}{} + iatas = append(iatas, v.Iata) + } } + p.ResolvedRoute = s.resolveTraceRoute(ctx, row.ParsedPayload, iatas) } return p, nil } diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 9fe3ec6..1c02925 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -267,10 +267,11 @@ INSERT INTO packets ( parsed_payload, channel_hash, scope_id, + trace_tag, first_heard_at, last_heard_at ) VALUES ( - $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, NOW(), NOW() + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, NOW(), NOW() ) ON CONFLICT (packet_hash) DO UPDATE SET last_heard_at = NOW() @@ -290,6 +291,20 @@ LEFT JOIN transport_scopes ts ON ts.id = p.scope_id LEFT JOIN channel_messages cm ON cm.packet_hash = p.packet_hash WHERE p.packet_hash = $1; +-- name: GetPacketsByTraceTag :many +-- Returns all packets for a given trace tag with observations. +SELECT encode(p.packet_hash, 'hex') AS packet_hash_hex, + p.route_type, + p.first_heard_at, + p.last_heard_at, + p.parsed_payload, + p.scope_id, + ts.name AS scope_name +FROM packets p +LEFT JOIN transport_scopes ts ON ts.id = p.scope_id +WHERE p.trace_tag = decode($1, 'hex') +ORDER BY p.first_heard_at ASC; + -- name: GetPacketObservationCount :one SELECT COUNT(*) FROM packet_observations WHERE packet_hash = $1; @@ -782,6 +797,30 @@ INSERT INTO region_iatas (region_id, iata) VALUES ($1, $2) ON CONFLICT (region_id, iata) DO NOTHING; +-- ============================================================ +-- TRACES +-- ============================================================ + +-- name: ListTraceTags :many +-- Returns distinct trace tags with summary info, ordered by most recent first. +SELECT + encode(p.trace_tag, 'hex') AS trace_tag, + MIN(p.first_heard_at)::timestamptz AS first_heard_at, + MAX(p.last_heard_at)::timestamptz AS last_heard_at, + COUNT(DISTINCT p.packet_hash) AS packet_count, + COUNT(DISTINCT po.iata) AS iata_count +FROM packets p +LEFT JOIN packet_observations po ON po.packet_hash = p.packet_hash +WHERE p.trace_tag IS NOT NULL + AND ($1::text = '' OR po.iata = ANY(string_to_array($1::text, ','))) + AND ($2::text = '' OR p.scope_id = (SELECT id FROM transport_scopes WHERE name = $2)) + AND ($3::timestamptz IS NULL OR p.first_heard_at >= $3) + AND ($4::timestamptz IS NULL OR p.first_heard_at <= $4) + AND ($5::timestamptz IS NULL OR p.last_heard_at < $5) +GROUP BY p.trace_tag +ORDER BY MAX(p.last_heard_at) DESC +LIMIT $6; + -- ============================================================ -- HELPERS -- ============================================================ diff --git a/db/sqlc/models.go b/db/sqlc/models.go index 1699fab..9f6626d 100644 --- a/db/sqlc/models.go +++ b/db/sqlc/models.go @@ -199,6 +199,7 @@ type Packet struct { ParsedPayload []byte `json:"parsed_payload"` Decrypted *bool `json:"decrypted"` ChannelHash []byte `json:"channel_hash"` + TraceTag []byte `json:"trace_tag"` FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` } diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index f75c6fd..441f153 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -592,7 +592,7 @@ func (q *Queries) GetObserverTelemetry(ctx context.Context, arg GetObserverTelem } const getPacketByHash = `-- name: GetPacketByHash :one -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, +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.trace_tag, p.first_heard_at, p.last_heard_at, ts.name AS scope_name, cm.sender_name AS cm_sender_name, cm.content AS cm_content, cm.sent_at AS cm_sent_at @@ -617,6 +617,7 @@ type GetPacketByHashRow struct { ParsedPayload []byte `json:"parsed_payload"` Decrypted *bool `json:"decrypted"` ChannelHash []byte `json:"channel_hash"` + TraceTag []byte `json:"trace_tag"` FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` ScopeName *string `json:"scope_name"` @@ -643,6 +644,7 @@ func (q *Queries) GetPacketByHash(ctx context.Context, packetHash []byte) (GetPa &i.ParsedPayload, &i.Decrypted, &i.ChannelHash, + &i.TraceTag, &i.FirstHeardAt, &i.LastHeardAt, &i.ScopeName, @@ -664,6 +666,59 @@ func (q *Queries) GetPacketObservationCount(ctx context.Context, packetHash []by return count, err } +const getPacketsByTraceTag = `-- name: GetPacketsByTraceTag :many +SELECT encode(p.packet_hash, 'hex') AS packet_hash_hex, + p.route_type, + p.first_heard_at, + p.last_heard_at, + p.parsed_payload, + p.scope_id, + ts.name AS scope_name +FROM packets p +LEFT JOIN transport_scopes ts ON ts.id = p.scope_id +WHERE p.trace_tag = decode($1, 'hex') +ORDER BY p.first_heard_at ASC +` + +type GetPacketsByTraceTagRow struct { + PacketHashHex string `json:"packet_hash_hex"` + RouteType int16 `json:"route_type"` + FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` + LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` + ParsedPayload []byte `json:"parsed_payload"` + ScopeID *int32 `json:"scope_id"` + ScopeName *string `json:"scope_name"` +} + +// Returns all packets for a given trace tag with observations. +func (q *Queries) GetPacketsByTraceTag(ctx context.Context, decode string) ([]GetPacketsByTraceTagRow, error) { + rows, err := q.db.Query(ctx, getPacketsByTraceTag, decode) + if err != nil { + return nil, err + } + defer rows.Close() + items := []GetPacketsByTraceTagRow{} + for rows.Next() { + var i GetPacketsByTraceTagRow + if err := rows.Scan( + &i.PacketHashHex, + &i.RouteType, + &i.FirstHeardAt, + &i.LastHeardAt, + &i.ParsedPayload, + &i.ScopeID, + &i.ScopeName, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const getRadioPresets = `-- name: GetRadioPresets :many SELECT preset, iata, source_type, count FROM mv_radio_presets @@ -2434,6 +2489,81 @@ func (q *Queries) ListRegions(ctx context.Context) ([]ListRegionsRow, error) { return items, nil } +const listTraceTags = `-- name: ListTraceTags :many + +SELECT + encode(p.trace_tag, 'hex') AS trace_tag, + MIN(p.first_heard_at)::timestamptz AS first_heard_at, + MAX(p.last_heard_at)::timestamptz AS last_heard_at, + COUNT(DISTINCT p.packet_hash) AS packet_count, + COUNT(DISTINCT po.iata) AS iata_count +FROM packets p +LEFT JOIN packet_observations po ON po.packet_hash = p.packet_hash +WHERE p.trace_tag IS NOT NULL + AND ($1::text = '' OR po.iata = ANY(string_to_array($1::text, ','))) + AND ($2::text = '' OR p.scope_id = (SELECT id FROM transport_scopes WHERE name = $2)) + AND ($3::timestamptz IS NULL OR p.first_heard_at >= $3) + AND ($4::timestamptz IS NULL OR p.first_heard_at <= $4) + AND ($5::timestamptz IS NULL OR p.last_heard_at < $5) +GROUP BY p.trace_tag +ORDER BY MAX(p.last_heard_at) DESC +LIMIT $6 +` + +type ListTraceTagsParams struct { + Column1 string `json:"column_1"` + Column2 string `json:"column_2"` + Column3 pgtype.Timestamptz `json:"column_3"` + Column4 pgtype.Timestamptz `json:"column_4"` + Column5 pgtype.Timestamptz `json:"column_5"` + Limit int32 `json:"limit"` +} + +type ListTraceTagsRow struct { + TraceTag string `json:"trace_tag"` + FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` + LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` + PacketCount int64 `json:"packet_count"` + IataCount int64 `json:"iata_count"` +} + +// ============================================================ +// TRACES +// ============================================================ +// Returns distinct trace tags with summary info, ordered by most recent first. +func (q *Queries) ListTraceTags(ctx context.Context, arg ListTraceTagsParams) ([]ListTraceTagsRow, error) { + rows, err := q.db.Query(ctx, listTraceTags, + arg.Column1, + arg.Column2, + arg.Column3, + arg.Column4, + arg.Column5, + arg.Limit, + ) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ListTraceTagsRow{} + for rows.Next() { + var i ListTraceTagsRow + if err := rows.Scan( + &i.TraceTag, + &i.FirstHeardAt, + &i.LastHeardAt, + &i.PacketCount, + &i.IataCount, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const refreshHourlyStats = `-- name: RefreshHourlyStats :exec REFRESH MATERIALIZED VIEW CONCURRENTLY mv_hourly_iata_stats ` @@ -2946,10 +3076,11 @@ INSERT INTO packets ( parsed_payload, channel_hash, scope_id, + trace_tag, first_heard_at, last_heard_at ) VALUES ( - $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, NOW(), NOW() + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, NOW(), NOW() ) ON CONFLICT (packet_hash) DO UPDATE SET last_heard_at = NOW() @@ -2971,6 +3102,7 @@ type UpsertPacketParams struct { ParsedPayload []byte `json:"parsed_payload"` ChannelHash []byte `json:"channel_hash"` ScopeID *int32 `json:"scope_id"` + TraceTag []byte `json:"trace_tag"` } type UpsertPacketRow struct { @@ -3010,6 +3142,7 @@ func (q *Queries) UpsertPacket(ctx context.Context, arg UpsertPacketParams) (Ups arg.ParsedPayload, arg.ChannelHash, arg.ScopeID, + arg.TraceTag, ) var i UpsertPacketRow err := row.Scan( diff --git a/db/store.go b/db/store.go index cd9ea69..88500f4 100644 --- a/db/store.go +++ b/db/store.go @@ -6,6 +6,7 @@ package db import ( "context" "encoding/hex" + "encoding/json" sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" "github.com/MeshCore-Beacon/beacon-server/internal/api" @@ -92,3 +93,73 @@ func toChannelMessage(id int64, packetHashHex string, channelHash []byte, sender ObservationCount: observationCount, } } + +// resolveTraceRoute resolves the path hashes from a trace parsedPayload across +// all provided IATAs, merging results per hop with best-confidence-wins semantics. +// Returns nil if the payload cannot be parsed or contains no path hashes. +func (s *Store) resolveTraceRoute(ctx context.Context, parsedPayload []byte, iatas []string) []api.ResolvedHop { + var tracePayload struct { + PathHashes []string `json:"pathHashes"` + Flags byte `json:"flags"` + } + if err := json.Unmarshal(parsedPayload, &tracePayload); err != nil || len(tracePayload.PathHashes) == 0 { + return nil + } + hashSize := int(1 << (tracePayload.Flags & 0x03)) + hashes := make([][]byte, 0, len(tracePayload.PathHashes)) + for _, h := range tracePayload.PathHashes { + b, err := hex.DecodeString(h) + if err == nil { + hashes = append(hashes, b) + } + } + confidenceRank := map[string]int{"none": 0, "ambiguous": 1, "high": 2} + type hopResult struct { + confidence string + entries []api.ResolvedPathEntry + } + merged := make([]hopResult, len(hashes)) + for i := range merged { + merged[i] = hopResult{confidence: "none"} + } + for _, iata := range iatas { + resolved, err := s.ResolvePathHashes(ctx, iata, hashes) + if err != nil { + continue + } + for i, hash := range hashes { + key := hex.EncodeToString(hash[:hashSize]) + entries := resolved[key] + var confidence string + switch len(entries) { + case 0: + confidence = "none" + case 1: + confidence = "high" + default: + confidence = "ambiguous" + } + if confidenceRank[confidence] > confidenceRank[merged[i].confidence] { + merged[i] = hopResult{confidence: confidence, entries: entries} + } + } + } + route := make([]api.ResolvedHop, 0, len(hashes)) + for _, hr := range merged { + hop := api.ResolvedHop{ + Confidence: hr.confidence, + Nodes: make([]api.ResolvedNode, 0, len(hr.entries)), + } + for _, e := range hr.entries { + hop.Nodes = append(hop.Nodes, api.ResolvedNode{ + ID: e.NodeID, + Name: e.Name, + Latitude: e.Latitude, + Longitude: e.Longitude, + PublicKey: hex.EncodeToString(e.PublicKey), + }) + } + route = append(route, hop) + } + return route +} diff --git a/db/traces.go b/db/traces.go new file mode 100644 index 0000000..6e19a93 --- /dev/null +++ b/db/traces.go @@ -0,0 +1,90 @@ +package db + +import ( + "context" + "encoding/hex" + "strings" + "time" + + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/jackc/pgx/v5/pgtype" +) + +func (s *Store) ListTraceTags(ctx context.Context, iatas []string, scope string, since, until time.Time, cursor time.Time, limit int32) ([]api.TraceTagSummary, error) { + iataFilter := strings.Join(iatas, ",") + var sinceTS, untilTS, cursorTS pgtype.Timestamptz + if !since.IsZero() { + sinceTS = pgtype.Timestamptz{Time: since, Valid: true} + } + if !until.IsZero() { + untilTS = pgtype.Timestamptz{Time: until, Valid: true} + } + if !cursor.IsZero() { + cursorTS = pgtype.Timestamptz{Time: cursor, Valid: true} + } + rows, err := s.q.ListTraceTags(ctx, sqlc.ListTraceTagsParams{ + Column1: iataFilter, + Column2: scope, + Column3: sinceTS, + Column4: untilTS, + Column5: cursorTS, + Limit: limit, + }) + if err != nil { + return nil, err + } + items := make([]api.TraceTagSummary, 0, len(rows)) + for _, r := range rows { + items = append(items, api.TraceTagSummary{ + TraceTag: r.TraceTag, + FirstHeardAt: r.FirstHeardAt.Time.UnixMilli(), + LastHeardAt: r.LastHeardAt.Time.UnixMilli(), + PacketCount: r.PacketCount, + IATACount: r.IataCount, + }) + } + return items, nil +} + +func (s *Store) GetTraceByTag(ctx context.Context, tag string) (*api.TraceDetail, error) { + rows, err := s.q.GetPacketsByTraceTag(ctx, tag) + if err != nil { + return nil, err + } + if len(rows) == 0 { + return nil, nil + } + detail := &api.TraceDetail{ + TraceTag: tag, + Packets: make([]api.TracePacket, 0, len(rows)), + } + for _, r := range rows { + packet := api.TracePacket{ + PacketHash: r.PacketHashHex, + RouteType: r.RouteType, + RouteTypeName: api.RouteTypeName(r.RouteType), + Scope: r.ScopeName, + FirstHeardAt: r.FirstHeardAt.Time.UnixMilli(), + LastHeardAt: r.LastHeardAt.Time.UnixMilli(), + } + // fetch observations to get IATAs for route resolution + packetHashBytes, err := hex.DecodeString(r.PacketHashHex) + if err == nil { + obsRows, err := s.q.ListObservationsForPacket(ctx, packetHashBytes) + if err == nil && len(obsRows) > 0 { + iatas := make([]string, 0, len(obsRows)) + seen := make(map[string]struct{}) + for _, v := range obsRows { + if _, ok := seen[v.Iata]; !ok { + seen[v.Iata] = struct{}{} + iatas = append(iatas, v.Iata) + } + } + packet.ResolvedRoute = s.resolveTraceRoute(ctx, r.ParsedPayload, iatas) + } + } + detail.Packets = append(detail.Packets, packet) + } + return detail, nil +} diff --git a/docs/docs.go b/docs/docs.go index 09449aa..95da09f 100644 --- a/docs/docs.go +++ b/docs/docs.go @@ -1568,6 +1568,124 @@ const docTemplate = `{ } } } + }, + "/traces": { + "get": { + "produces": [ + "application/json" + ], + "tags": [ + "Traces" + ], + "summary": "List trace tags", + "parameters": [ + { + "type": "string", + "description": "Filter by IATA code(s), comma-separated", + "name": "iatas", + "in": "query" + }, + { + "type": "string", + "description": "Filter by region slug", + "name": "region", + "in": "query" + }, + { + "type": "integer", + "description": "Filter by region ID", + "name": "regionId", + "in": "query" + }, + { + "type": "string", + "description": "Filter by transport scope name", + "name": "scope", + "in": "query" + }, + { + "type": "integer", + "description": "Filter by first_heard_at \u003e= since (epoch ms)", + "name": "since", + "in": "query" + }, + { + "type": "integer", + "description": "Filter by first_heard_at \u003c= until (epoch ms)", + "name": "until", + "in": "query" + }, + { + "type": "integer", + "description": "last_heard_at epoch ms of last item for pagination", + "name": "cursor", + "in": "query" + }, + { + "type": "integer", + "description": "Max results (default 50)", + "name": "limit", + "in": "query" + } + ], + "responses": { + "200": { + "description": "OK", + "schema": { + "type": "array", + "items": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.TraceTagSummary" + } + } + }, + "500": { + "description": "Internal Server Error", + "schema": { + "$ref": "#/definitions/internal_api_handlers.APIError" + } + } + } + } + }, + "/traces/{tag}": { + "get": { + "produces": [ + "application/json" + ], + "tags": [ + "Traces" + ], + "summary": "Get full trace detail by tag", + "parameters": [ + { + "type": "string", + "description": "Trace tag hex e.g. a3f1b2c4", + "name": "tag", + "in": "path", + "required": true + } + ], + "responses": { + "200": { + "description": "OK", + "schema": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.TraceDetail" + } + }, + "404": { + "description": "Not Found", + "schema": { + "$ref": "#/definitions/internal_api_handlers.APIError" + } + }, + "500": { + "description": "Internal Server Error", + "schema": { + "$ref": "#/definitions/internal_api_handlers.APIError" + } + } + } + } } }, "definitions": { @@ -2705,6 +2823,77 @@ const docTemplate = `{ } } }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.TraceDetail": { + "type": "object", + "properties": { + "packets": { + "type": "array", + "items": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.TracePacket" + } + }, + "traceTag": { + "description": "hex-encoded 4-byte tag", + "type": "string" + } + } + }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.TracePacket": { + "type": "object", + "properties": { + "firstHeardAt": { + "description": "epoch ms", + "type": "integer" + }, + "lastHeardAt": { + "description": "epoch ms", + "type": "integer" + }, + "packetHash": { + "type": "string" + }, + "resolvedRoute": { + "type": "array", + "items": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ResolvedHop" + } + }, + "routeType": { + "type": "integer" + }, + "routeTypeName": { + "type": "string" + }, + "scope": { + "type": "string" + } + } + }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.TraceTagSummary": { + "type": "object", + "properties": { + "firstHeardAt": { + "description": "epoch ms", + "type": "integer" + }, + "iataCount": { + "description": "number of distinct IATAs where heard", + "type": "integer" + }, + "lastHeardAt": { + "description": "epoch ms", + "type": "integer" + }, + "packetCount": { + "description": "number of packets with this trace tag", + "type": "integer" + }, + "traceTag": { + "description": "hex-encoded 4-byte tag", + "type": "string" + } + } + }, "internal_api_handlers.APIError": { "type": "object", "properties": { diff --git a/docs/swagger.json b/docs/swagger.json index a369bb4..3b6e94d 100644 --- a/docs/swagger.json +++ b/docs/swagger.json @@ -1566,6 +1566,124 @@ } } } + }, + "/traces": { + "get": { + "produces": [ + "application/json" + ], + "tags": [ + "Traces" + ], + "summary": "List trace tags", + "parameters": [ + { + "type": "string", + "description": "Filter by IATA code(s), comma-separated", + "name": "iatas", + "in": "query" + }, + { + "type": "string", + "description": "Filter by region slug", + "name": "region", + "in": "query" + }, + { + "type": "integer", + "description": "Filter by region ID", + "name": "regionId", + "in": "query" + }, + { + "type": "string", + "description": "Filter by transport scope name", + "name": "scope", + "in": "query" + }, + { + "type": "integer", + "description": "Filter by first_heard_at \u003e= since (epoch ms)", + "name": "since", + "in": "query" + }, + { + "type": "integer", + "description": "Filter by first_heard_at \u003c= until (epoch ms)", + "name": "until", + "in": "query" + }, + { + "type": "integer", + "description": "last_heard_at epoch ms of last item for pagination", + "name": "cursor", + "in": "query" + }, + { + "type": "integer", + "description": "Max results (default 50)", + "name": "limit", + "in": "query" + } + ], + "responses": { + "200": { + "description": "OK", + "schema": { + "type": "array", + "items": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.TraceTagSummary" + } + } + }, + "500": { + "description": "Internal Server Error", + "schema": { + "$ref": "#/definitions/internal_api_handlers.APIError" + } + } + } + } + }, + "/traces/{tag}": { + "get": { + "produces": [ + "application/json" + ], + "tags": [ + "Traces" + ], + "summary": "Get full trace detail by tag", + "parameters": [ + { + "type": "string", + "description": "Trace tag hex e.g. a3f1b2c4", + "name": "tag", + "in": "path", + "required": true + } + ], + "responses": { + "200": { + "description": "OK", + "schema": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.TraceDetail" + } + }, + "404": { + "description": "Not Found", + "schema": { + "$ref": "#/definitions/internal_api_handlers.APIError" + } + }, + "500": { + "description": "Internal Server Error", + "schema": { + "$ref": "#/definitions/internal_api_handlers.APIError" + } + } + } + } } }, "definitions": { @@ -2703,6 +2821,77 @@ } } }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.TraceDetail": { + "type": "object", + "properties": { + "packets": { + "type": "array", + "items": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.TracePacket" + } + }, + "traceTag": { + "description": "hex-encoded 4-byte tag", + "type": "string" + } + } + }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.TracePacket": { + "type": "object", + "properties": { + "firstHeardAt": { + "description": "epoch ms", + "type": "integer" + }, + "lastHeardAt": { + "description": "epoch ms", + "type": "integer" + }, + "packetHash": { + "type": "string" + }, + "resolvedRoute": { + "type": "array", + "items": { + "$ref": "#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ResolvedHop" + } + }, + "routeType": { + "type": "integer" + }, + "routeTypeName": { + "type": "string" + }, + "scope": { + "type": "string" + } + } + }, + "github_com_MeshCore-Beacon_beacon-server_internal_api.TraceTagSummary": { + "type": "object", + "properties": { + "firstHeardAt": { + "description": "epoch ms", + "type": "integer" + }, + "iataCount": { + "description": "number of distinct IATAs where heard", + "type": "integer" + }, + "lastHeardAt": { + "description": "epoch ms", + "type": "integer" + }, + "packetCount": { + "description": "number of packets with this trace tag", + "type": "integer" + }, + "traceTag": { + "description": "hex-encoded 4-byte tag", + "type": "string" + } + } + }, "internal_api_handlers.APIError": { "type": "object", "properties": { diff --git a/docs/swagger.yaml b/docs/swagger.yaml index 2e1f70c..11c8d0d 100644 --- a/docs/swagger.yaml +++ b/docs/swagger.yaml @@ -785,6 +785,55 @@ definitions: observerType: type: string type: object + github_com_MeshCore-Beacon_beacon-server_internal_api.TraceDetail: + properties: + packets: + items: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.TracePacket' + type: array + traceTag: + description: hex-encoded 4-byte tag + type: string + type: object + github_com_MeshCore-Beacon_beacon-server_internal_api.TracePacket: + properties: + firstHeardAt: + description: epoch ms + type: integer + lastHeardAt: + description: epoch ms + type: integer + packetHash: + type: string + resolvedRoute: + items: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.ResolvedHop' + type: array + routeType: + type: integer + routeTypeName: + type: string + scope: + type: string + type: object + github_com_MeshCore-Beacon_beacon-server_internal_api.TraceTagSummary: + properties: + firstHeardAt: + description: epoch ms + type: integer + iataCount: + description: number of distinct IATAs where heard + type: integer + lastHeardAt: + description: epoch ms + type: integer + packetCount: + description: number of packets with this trace tag + type: integer + traceTag: + description: hex-encoded 4-byte tag + type: string + type: object internal_api_handlers.APIError: properties: code: @@ -1828,6 +1877,83 @@ paths: summary: Top N observers by observation count (last 24h by default) tags: - Stats + /traces: + get: + parameters: + - description: Filter by IATA code(s), comma-separated + in: query + name: iatas + type: string + - description: Filter by region slug + in: query + name: region + type: string + - description: Filter by region ID + in: query + name: regionId + type: integer + - description: Filter by transport scope name + in: query + name: scope + type: string + - description: Filter by first_heard_at >= since (epoch ms) + in: query + name: since + type: integer + - description: Filter by first_heard_at <= until (epoch ms) + in: query + name: until + type: integer + - description: last_heard_at epoch ms of last item for pagination + in: query + name: cursor + type: integer + - description: Max results (default 50) + in: query + name: limit + type: integer + produces: + - application/json + responses: + "200": + description: OK + schema: + items: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.TraceTagSummary' + type: array + "500": + description: Internal Server Error + schema: + $ref: '#/definitions/internal_api_handlers.APIError' + summary: List trace tags + tags: + - Traces + /traces/{tag}: + get: + parameters: + - description: Trace tag hex e.g. a3f1b2c4 + in: path + name: tag + required: true + type: string + produces: + - application/json + responses: + "200": + description: OK + schema: + $ref: '#/definitions/github_com_MeshCore-Beacon_beacon-server_internal_api.TraceDetail' + "404": + description: Not Found + schema: + $ref: '#/definitions/internal_api_handlers.APIError' + "500": + description: Internal Server Error + schema: + $ref: '#/definitions/internal_api_handlers.APIError' + summary: Get full trace detail by tag + tags: + - Traces schemes: - http - https diff --git a/internal/api/handlers/traces.go b/internal/api/handlers/traces.go new file mode 100644 index 0000000..6319a81 --- /dev/null +++ b/internal/api/handlers/traces.go @@ -0,0 +1,106 @@ +package handlers + +import ( + "net/http" + "strconv" + "time" + + "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/go-chi/chi/v5" +) + +// TracesRouter mounts all /traces routes onto a subrouter. +// +// GET /traces → listTraceTags +// GET /traces/{tag} → getTrace +func TracesRouter(reader api.Reader) http.Handler { + r := chi.NewRouter() + r.Get("/", listTraceTags(reader)) + r.Get("/{tag}", getTrace(reader)) + return r +} + +// listTraceTags godoc +// +// @Summary List trace tags +// @Tags Traces +// @Produce json +// @Param iatas query string false "Filter by IATA code(s), comma-separated" +// @Param region query string false "Filter by region slug" +// @Param regionId query int false "Filter by region ID" +// @Param scope query string false "Filter by transport scope name" +// @Param since query int false "Filter by first_heard_at >= since (epoch ms)" +// @Param until query int false "Filter by first_heard_at <= until (epoch ms)" +// @Param cursor query int false "last_heard_at epoch ms of last item for pagination" +// @Param limit query int false "Max results (default 50)" +// @Success 200 {object} []api.TraceTagSummary +// @Failure 500 {object} handlers.APIError +// @Router /traces [get] +func listTraceTags(reader api.Reader) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + var limit int32 = 50 + if l := r.URL.Query().Get("limit"); l != "" { + if v, err := strconv.ParseInt(l, 10, 32); err == nil { + limit = int32(v) + } + } + var since, until, cursor time.Time + if v := r.URL.Query().Get("since"); v != "" { + if ms, err := strconv.ParseInt(v, 10, 64); err == nil { + since = time.UnixMilli(ms) + } + } + if v := r.URL.Query().Get("until"); v != "" { + if ms, err := strconv.ParseInt(v, 10, 64); err == nil { + until = time.UnixMilli(ms) + } + } + if v := r.URL.Query().Get("cursor"); v != "" { + if ms, err := strconv.ParseInt(v, 10, 64); err == nil { + cursor = time.UnixMilli(ms) + } + } + iatas := parseIATAs(r) + if regionIDStr := r.URL.Query().Get("regionId"); regionIDStr != "" || r.URL.Query().Get("region") != "" { + regionIATAs, err := resolveRegionIATAs(r.Context(), r.URL.Query().Get("regionId"), r.URL.Query().Get("region"), reader) + if err != nil { + respondError(w, http.StatusBadRequest, err.Error()) + return + } + iatas = append(iatas, regionIATAs...) + } + scope := r.URL.Query().Get("scope") + tags, err := reader.ListTraceTags(r.Context(), iatas, scope, since, until, cursor, limit) + if err != nil { + respondError(w, http.StatusInternalServerError, "internal server error") + return + } + respond(w, http.StatusOK, tags) + } +} + +// getTrace godoc +// +// @Summary Get full trace detail by tag +// @Tags Traces +// @Produce json +// @Param tag path string true "Trace tag hex e.g. a3f1b2c4" +// @Success 200 {object} api.TraceDetail +// @Failure 404 {object} handlers.APIError +// @Failure 500 {object} handlers.APIError +// @Router /traces/{tag} [get] +func getTrace(reader api.Reader) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + tag := chi.URLParam(r, "tag") + trace, err := reader.GetTraceByTag(r.Context(), tag) + if err != nil { + respondError(w, http.StatusInternalServerError, "internal server error") + return + } + if trace == nil { + respondError(w, http.StatusNotFound, "trace not found") + return + } + respond(w, http.StatusOK, trace) + } +} diff --git a/internal/api/reader.go b/internal/api/reader.go index 8251070..1aac9fc 100644 --- a/internal/api/reader.go +++ b/internal/api/reader.go @@ -133,4 +133,8 @@ type Reader interface { // including packet count, observer count, node count, and the list of IATAs it is active in. // Returns nil if the scope is not found. GetScopeByName(ctx context.Context, name string) (*ScopeDetail, error) + // ListTraceTags returns a paginated list of trace tags with aggregate metadata. + 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) } diff --git a/internal/api/router/router.go b/internal/api/router/router.go index b38113a..f998d1b 100644 --- a/internal/api/router/router.go +++ b/internal/api/router/router.go @@ -73,6 +73,7 @@ func New(h *hub.Hub, reader api.Reader, workers []*ingest.Worker, maxConnsPerIP r.Mount("/regions", handlers.RegionsRouter(reader)) r.Mount("/scopes", handlers.ScopesRouter(reader)) r.Mount("/stats", handlers.StatsRouter(reader)) + r.Mount("/traces", handlers.TracesRouter(reader)) }) // Private group — auth middleware applied. diff --git a/internal/api/traces.go b/internal/api/traces.go new file mode 100644 index 0000000..da895a3 --- /dev/null +++ b/internal/api/traces.go @@ -0,0 +1,28 @@ +package api + +// TraceTagSummary is a single trace tag with aggregate metadata. +type TraceTagSummary struct { + TraceTag string `json:"traceTag"` // hex-encoded 4-byte tag + FirstHeardAt int64 `json:"firstHeardAt"` // epoch ms + LastHeardAt int64 `json:"lastHeardAt"` // epoch ms + PacketCount int64 `json:"packetCount"` // number of packets with this trace tag + IATACount int64 `json:"iataCount"` // number of distinct IATAs where heard +} + +// TracePacket is a single packet within a trace series, including its +// resolved route derived from the trace path hashes. +type TracePacket struct { + PacketHash string `json:"packetHash"` + 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 + ResolvedRoute []ResolvedHop `json:"resolvedRoute"` +} + +// TraceDetail is the full trace series for a given trace tag. +type TraceDetail struct { + TraceTag string `json:"traceTag"` // hex-encoded 4-byte tag + Packets []TracePacket `json:"packets"` +} diff --git a/internal/ingest/packet.go b/internal/ingest/packet.go index aaa17b9..e26cd9d 100644 --- a/internal/ingest/packet.go +++ b/internal/ingest/packet.go @@ -31,6 +31,7 @@ type UpsertPacketParams struct { OriginPubkey []byte ChannelHash []byte ScopeID *int32 + TraceTag []byte } // InsertObservationParams mirrors the columns written on packet_observations insert. @@ -261,6 +262,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ var channelHash []byte originPubkey := []byte(nil) var parsedPayload json.RawMessage + var traceTag []byte switch packet.PayloadType() { case meshcore.PayloadTypeGrpTxt: @@ -431,6 +433,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ case meshcore.PayloadTypeTrace: trace, err := meshcore.TraceFromBytes(packet.Payload) if err == nil { + traceTag = uint32ToBytes(trace.Tag) hashSize := int(trace.PathHashSize()) hashes := make([]string, 0) for i := 0; i+hashSize <= len(trace.PathHashes); i += hashSize { @@ -533,6 +536,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ OriginPubkey: originPubkey, ChannelHash: channelHash, ScopeID: scopeID, + TraceTag: traceTag, } isNew, err := w.db.UpsertPacket(ctx, pParams) if err != nil {