dandri/beacon-server
1
0 Fork 0
mirror of https://github.com/MeshCore-Beacon/beacon-server.git synced 2026-10-22 14:11:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

add traces endpoint

Browse Source
This commit is contained in:
Enot (ded) Skelly committed 2026-06-05 15:20:53 -07:00
1 parent 8b23f8a4a2
commit f0c57fde45
15 files changed
+995 -78
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

db/migrations/001_schema.sql
+2
View File
@@ -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
db/packets.go
+9 -75
View File
@@ -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
}
db/queries/queries.sql
+40 -1
View File
@@ -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
-- ============================================================
db/sqlc/models.go
+1
View File
@@ -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"`
}
db/sqlc/queries.sql.go
+135 -2
View File
@@ -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(
db/store.go
+71
View File
@@ -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
}
db/traces.go
+90
View File
@@ -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
}
docs/docs.go
+189
View File
@@ -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": {
docs/swagger.json
+189
View File
@@ -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": {
docs/swagger.yaml
+126
View File
@@ -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
internal/api/handlers/traces.go
+106
View File
@@ -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)
}
}
internal/api/reader.go
+4
View File
@@ -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)
}
internal/api/router/router.go
+1
View File
@@ -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.
internal/api/traces.go
+28
View File
@@ -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"`
}
internal/ingest/packet.go
+4
View File
@@ -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 {