mirror of
https://github.com/MeshCore-Beacon/beacon-server.git
synced 2026-10-23 09:52:32 +00:00
add packets api
This commit is contained in:
8 files changed
+483
-79
No files matched your search
@@ -309,10 +309,10 @@ Tower server.
|
||||
- [x] REST API: Brokers (list with connection status)
|
||||
- [x] REST API: Pagination
|
||||
- [x] REST API: Nodes (list + detail + observations)
|
||||
- [x] REST API: Packets (list + detail)
|
||||
|
||||
### In progress / next
|
||||
|
||||
- [ ] REST API: Packets (list + detail)
|
||||
- [ ] REST API: Stats
|
||||
- [ ] Path resolution (node short ID lookup)
|
||||
- [ ] Propagation time calculation
|
||||
|
||||
+33
-10
@@ -186,19 +186,40 @@ ON CONFLICT (packet_hash) DO UPDATE SET
|
||||
observation_count = packets.observation_count + 1
|
||||
RETURNING *, (xmax = 0) AS inserted;
|
||||
|
||||
-- name: GetPacket :one
|
||||
-- name: GetPacketByHash :one
|
||||
SELECT * FROM packets WHERE packet_hash = $1;
|
||||
|
||||
-- name: ListPackets :many
|
||||
SELECT p.*
|
||||
-- Returns packets with the latest observation rolled in for display.
|
||||
-- Pass cursor=0 to start from the beginning.
|
||||
SELECT
|
||||
p.packet_hash,
|
||||
p.payload_type,
|
||||
p.route_type,
|
||||
p.first_heard_at,
|
||||
p.last_heard_at,
|
||||
p.observation_count,
|
||||
po.observer_id AS latest_observer_id,
|
||||
o.display_name AS latest_observer_name,
|
||||
po.iata AS latest_observer_iata
|
||||
FROM packets p
|
||||
LEFT JOIN LATERAL (
|
||||
SELECT observer_id, iata
|
||||
FROM packet_observations
|
||||
WHERE packet_hash = p.packet_hash
|
||||
ORDER BY heard_at DESC
|
||||
LIMIT 1
|
||||
) po ON true
|
||||
LEFT JOIN observers o ON o.id = po.observer_id
|
||||
WHERE
|
||||
($1::smallint IS NULL OR p.payload_type = $1)
|
||||
AND ($2::smallint IS NULL OR p.route_type = $2)
|
||||
AND ($3::timestamptz IS NULL OR p.first_heard_at >= $3)
|
||||
AND ($4::timestamptz IS NULL OR p.first_heard_at <= $4)
|
||||
($1 = 0 OR p.payload_type = $1)
|
||||
AND ($2 = 0 OR p.route_type = $2)
|
||||
AND ($3 = '' OR po.iata ILIKE $3)
|
||||
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)
|
||||
ORDER BY p.last_heard_at DESC
|
||||
LIMIT $5;
|
||||
LIMIT $7;
|
||||
|
||||
-- name: ListPacketsAfterID :many
|
||||
SELECT p.*
|
||||
@@ -237,9 +258,11 @@ ON CONFLICT (packet_hash, observer_id, heard_at) DO NOTHING
|
||||
RETURNING *;
|
||||
|
||||
-- name: ListObservationsForPacket :many
|
||||
SELECT * FROM packet_observations
|
||||
WHERE packet_hash = $1
|
||||
ORDER BY heard_at ASC;
|
||||
SELECT po.*, o.display_name AS observer_name
|
||||
FROM packet_observations po
|
||||
LEFT JOIN observers o ON o.id = po.observer_id
|
||||
WHERE po.packet_hash = $1
|
||||
ORDER BY po.heard_at ASC;
|
||||
|
||||
-- name: ListObservationsForObserver :many
|
||||
SELECT * FROM packet_observations
|
||||
|
||||
+85
-30
@@ -459,12 +459,12 @@ func (q *Queries) GetObserverTelemetry(ctx context.Context, arg GetObserverTelem
|
||||
return items, nil
|
||||
}
|
||||
|
||||
const getPacket = `-- name: GetPacket :one
|
||||
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, parsed_payload, decrypted, channel_hash, first_heard_at, last_heard_at, observation_count FROM packets WHERE packet_hash = $1
|
||||
`
|
||||
|
||||
func (q *Queries) GetPacket(ctx context.Context, packetHash []byte) (Packet, error) {
|
||||
row := q.db.QueryRow(ctx, getPacket, packetHash)
|
||||
func (q *Queries) GetPacketByHash(ctx context.Context, packetHash []byte) (Packet, error) {
|
||||
row := q.db.QueryRow(ctx, getPacketByHash, packetHash)
|
||||
var i Packet
|
||||
err := row.Scan(
|
||||
&i.PacketHash,
|
||||
@@ -1268,20 +1268,43 @@ func (q *Queries) ListObservationsForObserver(ctx context.Context, arg ListObser
|
||||
}
|
||||
|
||||
const listObservationsForPacket = `-- name: ListObservationsForPacket :many
|
||||
SELECT id, packet_hash, observer_id, iata, heard_at, path_length_byte, hash_size, hop_count, path_bytes, rssi, snr, propagation_time_ms, radio_freq_mhz, spread_factor, bandwidth_khz, coding_rate, source_broker FROM packet_observations
|
||||
WHERE packet_hash = $1
|
||||
ORDER BY heard_at ASC
|
||||
SELECT po.id, po.packet_hash, po.observer_id, po.iata, po.heard_at, po.path_length_byte, po.hash_size, po.hop_count, po.path_bytes, po.rssi, po.snr, po.propagation_time_ms, po.radio_freq_mhz, po.spread_factor, po.bandwidth_khz, po.coding_rate, po.source_broker, o.display_name AS observer_name
|
||||
FROM packet_observations po
|
||||
LEFT JOIN observers o ON o.id = po.observer_id
|
||||
WHERE po.packet_hash = $1
|
||||
ORDER BY po.heard_at ASC
|
||||
`
|
||||
|
||||
func (q *Queries) ListObservationsForPacket(ctx context.Context, packetHash []byte) ([]PacketObservation, error) {
|
||||
type ListObservationsForPacketRow struct {
|
||||
ID int64 `json:"id"`
|
||||
PacketHash []byte `json:"packet_hash"`
|
||||
ObserverID uuid.UUID `json:"observer_id"`
|
||||
Iata string `json:"iata"`
|
||||
HeardAt pgtype.Timestamptz `json:"heard_at"`
|
||||
PathLengthByte int16 `json:"path_length_byte"`
|
||||
HashSize int16 `json:"hash_size"`
|
||||
HopCount int16 `json:"hop_count"`
|
||||
PathBytes []byte `json:"path_bytes"`
|
||||
Rssi *int16 `json:"rssi"`
|
||||
Snr *float32 `json:"snr"`
|
||||
PropagationTimeMs *int32 `json:"propagation_time_ms"`
|
||||
RadioFreqMhz *float32 `json:"radio_freq_mhz"`
|
||||
SpreadFactor *int16 `json:"spread_factor"`
|
||||
BandwidthKhz *float32 `json:"bandwidth_khz"`
|
||||
CodingRate *int16 `json:"coding_rate"`
|
||||
SourceBroker *string `json:"source_broker"`
|
||||
ObserverName *string `json:"observer_name"`
|
||||
}
|
||||
|
||||
func (q *Queries) ListObservationsForPacket(ctx context.Context, packetHash []byte) ([]ListObservationsForPacketRow, error) {
|
||||
rows, err := q.db.Query(ctx, listObservationsForPacket, packetHash)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
items := []PacketObservation{}
|
||||
items := []ListObservationsForPacketRow{}
|
||||
for rows.Next() {
|
||||
var i PacketObservation
|
||||
var i ListObservationsForPacketRow
|
||||
if err := rows.Scan(
|
||||
&i.ID,
|
||||
&i.PacketHash,
|
||||
@@ -1300,6 +1323,7 @@ func (q *Queries) ListObservationsForPacket(ctx context.Context, packetHash []by
|
||||
&i.BandwidthKhz,
|
||||
&i.CodingRate,
|
||||
&i.SourceBroker,
|
||||
&i.ObserverName,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -1480,56 +1504,87 @@ func (q *Queries) ListObservers(ctx context.Context, arg ListObserversParams) ([
|
||||
}
|
||||
|
||||
const listPackets = `-- name: ListPackets :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.parsed_payload, p.decrypted, p.channel_hash, p.first_heard_at, p.last_heard_at, p.observation_count
|
||||
SELECT
|
||||
p.packet_hash,
|
||||
p.payload_type,
|
||||
p.route_type,
|
||||
p.first_heard_at,
|
||||
p.last_heard_at,
|
||||
p.observation_count,
|
||||
po.observer_id AS latest_observer_id,
|
||||
o.display_name AS latest_observer_name,
|
||||
po.iata AS latest_observer_iata
|
||||
FROM packets p
|
||||
LEFT JOIN LATERAL (
|
||||
SELECT observer_id, iata
|
||||
FROM packet_observations
|
||||
WHERE packet_hash = p.packet_hash
|
||||
ORDER BY heard_at DESC
|
||||
LIMIT 1
|
||||
) po ON true
|
||||
LEFT JOIN observers o ON o.id = po.observer_id
|
||||
WHERE
|
||||
($1::smallint IS NULL OR p.payload_type = $1)
|
||||
AND ($2::smallint IS NULL OR p.route_type = $2)
|
||||
AND ($3::timestamptz IS NULL OR p.first_heard_at >= $3)
|
||||
AND ($4::timestamptz IS NULL OR p.first_heard_at <= $4)
|
||||
($1 = 0 OR p.payload_type = $1)
|
||||
AND ($2 = 0 OR p.route_type = $2)
|
||||
AND ($3 = '' OR po.iata ILIKE $3)
|
||||
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)
|
||||
ORDER BY p.last_heard_at DESC
|
||||
LIMIT $5
|
||||
LIMIT $7
|
||||
`
|
||||
|
||||
type ListPacketsParams struct {
|
||||
Column1 int16 `json:"column_1"`
|
||||
Column2 int16 `json:"column_2"`
|
||||
Column3 pgtype.Timestamptz `json:"column_3"`
|
||||
Column1 interface{} `json:"column_1"`
|
||||
Column2 interface{} `json:"column_2"`
|
||||
Column3 interface{} `json:"column_3"`
|
||||
Column4 pgtype.Timestamptz `json:"column_4"`
|
||||
Column5 pgtype.Timestamptz `json:"column_5"`
|
||||
Column6 pgtype.Timestamptz `json:"column_6"`
|
||||
Limit int32 `json:"limit"`
|
||||
}
|
||||
|
||||
func (q *Queries) ListPackets(ctx context.Context, arg ListPacketsParams) ([]Packet, error) {
|
||||
type ListPacketsRow struct {
|
||||
PacketHash []byte `json:"packet_hash"`
|
||||
PayloadType int16 `json:"payload_type"`
|
||||
RouteType int16 `json:"route_type"`
|
||||
FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"`
|
||||
LastHeardAt pgtype.Timestamptz `json:"last_heard_at"`
|
||||
ObservationCount *int32 `json:"observation_count"`
|
||||
LatestObserverID uuid.UUID `json:"latest_observer_id"`
|
||||
LatestObserverName *string `json:"latest_observer_name"`
|
||||
LatestObserverIata string `json:"latest_observer_iata"`
|
||||
}
|
||||
|
||||
// Returns packets with the latest observation rolled in for display.
|
||||
// Pass cursor=0 to start from the beginning.
|
||||
func (q *Queries) ListPackets(ctx context.Context, arg ListPacketsParams) ([]ListPacketsRow, error) {
|
||||
rows, err := q.db.Query(ctx, listPackets,
|
||||
arg.Column1,
|
||||
arg.Column2,
|
||||
arg.Column3,
|
||||
arg.Column4,
|
||||
arg.Column5,
|
||||
arg.Column6,
|
||||
arg.Limit,
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
items := []Packet{}
|
||||
items := []ListPacketsRow{}
|
||||
for rows.Next() {
|
||||
var i Packet
|
||||
var i ListPacketsRow
|
||||
if err := rows.Scan(
|
||||
&i.PacketHash,
|
||||
&i.PayloadType,
|
||||
&i.PayloadVersion,
|
||||
&i.RouteType,
|
||||
&i.TransportCodesPresent,
|
||||
&i.RegionCode,
|
||||
&i.SubRegionCode,
|
||||
&i.OriginPubkey,
|
||||
&i.RawPayload,
|
||||
&i.ParsedPayload,
|
||||
&i.Decrypted,
|
||||
&i.ChannelHash,
|
||||
&i.FirstHeardAt,
|
||||
&i.LastHeardAt,
|
||||
&i.ObservationCount,
|
||||
&i.LatestObserverID,
|
||||
&i.LatestObserverName,
|
||||
&i.LatestObserverIata,
|
||||
); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
+134
@@ -914,3 +914,137 @@ func (s *Store) ListNodeObservations(ctx context.Context, nodeID uuid.UUID, curs
|
||||
HasMore: hasMore,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *Store) ListPackets(ctx context.Context, payloadType, routeType int16, iata 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}
|
||||
}
|
||||
var sinceTS pgtype.Timestamptz
|
||||
if !since.IsZero() {
|
||||
sinceTS = pgtype.Timestamptz{Time: since, Valid: true}
|
||||
}
|
||||
var untilTS pgtype.Timestamptz
|
||||
if !until.IsZero() {
|
||||
untilTS = pgtype.Timestamptz{Time: until, Valid: true}
|
||||
}
|
||||
rows, err := s.q.ListPackets(ctx, sqlc.ListPacketsParams{
|
||||
Column1: payloadType,
|
||||
Column2: routeType,
|
||||
Column3: iata,
|
||||
Column4: sinceTS,
|
||||
Column5: untilTS,
|
||||
Column6: cursorTS,
|
||||
Limit: limit + 1,
|
||||
})
|
||||
if err != nil {
|
||||
return api.Page[api.PacketSummary]{}, err
|
||||
}
|
||||
hasMore := len(rows) > int(limit)
|
||||
if hasMore {
|
||||
rows = rows[:limit]
|
||||
}
|
||||
items := make([]api.PacketSummary, 0, len(rows))
|
||||
for _, v := range rows {
|
||||
item := api.PacketSummary{
|
||||
PacketHash: hex.EncodeToString(v.PacketHash),
|
||||
PayloadType: v.PayloadType,
|
||||
PayloadTypeName: api.PayloadTypeName(v.PayloadType),
|
||||
RouteType: v.RouteType,
|
||||
RouteTypeName: api.RouteTypeName(v.RouteType),
|
||||
FirstHeardAt: v.FirstHeardAt.Time.UnixMilli(),
|
||||
LastHeardAt: v.LastHeardAt.Time.UnixMilli(),
|
||||
ObservationCount: *v.ObservationCount,
|
||||
}
|
||||
if v.LatestObserverID != (uuid.UUID{}) {
|
||||
item.LatestObserver = &api.PacketLatestObserver{
|
||||
ID: v.LatestObserverID,
|
||||
DisplayName: v.LatestObserverName,
|
||||
IATA: v.LatestObserverIata,
|
||||
}
|
||||
}
|
||||
items = append(items, item)
|
||||
}
|
||||
var nextCursor *int64
|
||||
if hasMore && len(items) > 0 {
|
||||
last := items[len(items)-1].LastHeardAt
|
||||
nextCursor = &last
|
||||
}
|
||||
return api.Page[api.PacketSummary]{
|
||||
Items: items,
|
||||
NextCursor: nextCursor,
|
||||
HasMore: hasMore,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet, error) {
|
||||
row, err := s.q.GetPacketByHash(ctx, packetHash)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
obsRows, err := s.q.ListObservationsForPacket(ctx, packetHash)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
p := &api.Packet{
|
||||
PacketHash: hex.EncodeToString(row.PacketHash),
|
||||
PayloadType: row.PayloadType,
|
||||
PayloadTypeName: api.PayloadTypeName(row.PayloadType),
|
||||
PayloadVersion: row.PayloadVersion,
|
||||
RouteType: row.RouteType,
|
||||
RouteTypeName: api.RouteTypeName(row.RouteType),
|
||||
RawPayload: hex.EncodeToString(row.RawPayload),
|
||||
ParsedPayload: row.ParsedPayload,
|
||||
Decrypted: row.Decrypted != nil && *row.Decrypted,
|
||||
FirstHeardAt: row.FirstHeardAt.Time.UnixMilli(),
|
||||
LastHeardAt: row.LastHeardAt.Time.UnixMilli(),
|
||||
Observations: make([]api.PacketObservationDetail, 0, len(obsRows)),
|
||||
}
|
||||
if row.ObservationCount != nil {
|
||||
p.ObservationCount = *row.ObservationCount
|
||||
}
|
||||
if row.OriginPubkey != nil {
|
||||
s := hex.EncodeToString(row.OriginPubkey)
|
||||
p.OriginPubkey = &s
|
||||
}
|
||||
if row.ChannelHash != nil {
|
||||
ch := hex.EncodeToString(row.ChannelHash)
|
||||
p.ChannelHash = &ch
|
||||
}
|
||||
if row.TransportCodesPresent != nil && *row.TransportCodesPresent {
|
||||
// TODO: decode transport codes from region_code/sub_region_code
|
||||
}
|
||||
for _, v := range obsRows {
|
||||
obs := api.PacketObservationDetail{
|
||||
ID: v.ID,
|
||||
ObserverID: v.ObserverID,
|
||||
ObserverName: v.ObserverName,
|
||||
IATA: v.Iata,
|
||||
HeardAt: v.HeardAt.Time.UnixMilli(),
|
||||
PathLengthByte: v.PathLengthByte,
|
||||
HashSize: v.HashSize,
|
||||
HopCount: v.HopCount,
|
||||
RSSI: v.Rssi,
|
||||
SNR: v.Snr,
|
||||
SourceBroker: *v.SourceBroker,
|
||||
ResolvedPath: []api.ResolvedHop{}, // TODO: implement path resolution
|
||||
}
|
||||
if v.PathBytes != nil {
|
||||
pb := hex.EncodeToString(v.PathBytes)
|
||||
obs.PathBytes = &pb
|
||||
}
|
||||
if v.PropagationTimeMs != nil {
|
||||
obs.PropagationTimeMs = v.PropagationTimeMs
|
||||
}
|
||||
if v.RadioFreqMhz != nil || v.SpreadFactor != nil || v.BandwidthKhz != nil || v.CodingRate != nil {
|
||||
obs.Radio = &api.PacketRadio{
|
||||
FreqMHz: v.RadioFreqMhz,
|
||||
SpreadFactor: v.SpreadFactor,
|
||||
BandwidthKHz: v.BandwidthKhz,
|
||||
CodingRate: v.CodingRate,
|
||||
}
|
||||
}
|
||||
p.Observations = append(p.Observations, obs)
|
||||
}
|
||||
return p, nil
|
||||
}
|
||||
@@ -1,8 +1,12 @@
|
||||
package handlers
|
||||
|
||||
import (
|
||||
"encoding/hex"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/MeshCore-Tower/tower-server/internal/api"
|
||||
"github.com/go-chi/chi/v5"
|
||||
)
|
||||
|
||||
@@ -10,45 +14,106 @@ import (
|
||||
//
|
||||
// GET /packets → ListPackets
|
||||
// GET /packets/{packetHash} → GetPacket
|
||||
func PacketsRouter() http.Handler {
|
||||
func PacketsRouter(reader api.Reader) http.Handler {
|
||||
r := chi.NewRouter()
|
||||
|
||||
r.Get("/", ListPackets)
|
||||
r.Get("/{packetHash}", GetPacket)
|
||||
// GET /api/v1/packets
|
||||
//
|
||||
// Query params (all optional):
|
||||
//
|
||||
// payloadType=<int> filter by payload type integer
|
||||
// payloadTypeName=<string> filter by payload type name (advert, grp_txt, txt_msg, trace, anon_req)
|
||||
// routeType=<int> filter by route type integer (0=transport_flood, 1=flood, 2=direct, 3=transport_direct)
|
||||
// iata=<code> filter by latest observation IATA (case-insensitive)
|
||||
// since=<epoch ms> filter by first_heard_at >= since
|
||||
// until=<epoch ms> filter by first_heard_at <= until
|
||||
// cursor=<int> last_heard_at epoch ms of last item for pagination
|
||||
// limit=50
|
||||
r.Get("/", func(w http.ResponseWriter, r *http.Request) {
|
||||
var payloadType int16
|
||||
if p := r.URL.Query().Get("payloadType"); p != "" {
|
||||
t, err := strconv.ParseInt(p, 10, 16)
|
||||
if err != nil {
|
||||
respondError(w, http.StatusBadRequest, "payloadType must be an integer")
|
||||
return
|
||||
}
|
||||
payloadType = int16(t)
|
||||
}
|
||||
if payloadType == 0 {
|
||||
if p := r.URL.Query().Get("payloadTypeName"); p != "" {
|
||||
payloadType = api.PayloadTypeFromString(p)
|
||||
}
|
||||
}
|
||||
var routeType int16
|
||||
if p := r.URL.Query().Get("routeType"); p != "" {
|
||||
t, err := strconv.ParseInt(p, 10, 16)
|
||||
if err != nil {
|
||||
respondError(w, http.StatusBadRequest, "routeType must be an integer")
|
||||
return
|
||||
}
|
||||
routeType = int16(t)
|
||||
}
|
||||
var since, until time.Time
|
||||
if p := r.URL.Query().Get("since"); p != "" {
|
||||
ms, err := strconv.ParseInt(p, 10, 64)
|
||||
if err != nil {
|
||||
respondError(w, http.StatusBadRequest, "since must be epoch milliseconds")
|
||||
return
|
||||
}
|
||||
since = time.UnixMilli(ms)
|
||||
}
|
||||
if p := r.URL.Query().Get("until"); p != "" {
|
||||
ms, err := strconv.ParseInt(p, 10, 64)
|
||||
if err != nil {
|
||||
respondError(w, http.StatusBadRequest, "until must be epoch milliseconds")
|
||||
return
|
||||
}
|
||||
until = time.UnixMilli(ms)
|
||||
}
|
||||
var cursor int64
|
||||
if p := r.URL.Query().Get("cursor"); p != "" {
|
||||
c, err := strconv.ParseInt(p, 10, 64)
|
||||
if err != nil {
|
||||
respondError(w, http.StatusBadRequest, "cursor must be an integer")
|
||||
return
|
||||
}
|
||||
cursor = c
|
||||
}
|
||||
var limit int32 = 50
|
||||
if p := r.URL.Query().Get("limit"); p != "" {
|
||||
l, err := strconv.ParseInt(p, 10, 32)
|
||||
if err != nil {
|
||||
respondError(w, http.StatusBadRequest, "limit must be an integer")
|
||||
return
|
||||
}
|
||||
limit = int32(l)
|
||||
}
|
||||
iata := r.URL.Query().Get("iata")
|
||||
packets, err := reader.ListPackets(r.Context(), payloadType, routeType, iata, since, until, cursor, limit)
|
||||
if err != nil {
|
||||
respondError(w, http.StatusInternalServerError, "internal server error")
|
||||
return
|
||||
}
|
||||
respond(w, http.StatusOK, packets)
|
||||
})
|
||||
|
||||
// GET /api/v1/packets/{packetHash}
|
||||
//
|
||||
// Returns full packet detail including all observations and resolved paths.
|
||||
r.Get("/{packetHash}", func(w http.ResponseWriter, r *http.Request) {
|
||||
hashHex := chi.URLParam(r, "packetHash")
|
||||
hash, err := hex.DecodeString(hashHex)
|
||||
if err != nil {
|
||||
respondError(w, http.StatusBadRequest, "invalid packet hash")
|
||||
return
|
||||
}
|
||||
packet, err := reader.GetPacket(r.Context(), hash)
|
||||
if err != nil {
|
||||
respondError(w, http.StatusNotFound, "packet not found")
|
||||
return
|
||||
}
|
||||
respond(w, http.StatusOK, packet)
|
||||
})
|
||||
|
||||
return r
|
||||
}
|
||||
|
||||
// ListPackets handles GET /api/v1/packets
|
||||
//
|
||||
// Query params (all optional):
|
||||
//
|
||||
// iata=YOW
|
||||
// payloadType=4
|
||||
// routeType=1
|
||||
// since=<epoch ms>
|
||||
// until=<epoch ms>
|
||||
// afterId=<observation id> used for deterministic WS reconnection backfill
|
||||
// limit=50
|
||||
// cursor=<opaque>
|
||||
//
|
||||
// Returns a paginated list of packet summaries with the latest observation
|
||||
// rolled in, newest first.
|
||||
func ListPackets(w http.ResponseWriter, r *http.Request) {
|
||||
// TODO: parse query params, query DB/cache, write JSON response.
|
||||
//
|
||||
// afterId (int64) takes precedence over cursor for reconnection backfill:
|
||||
// WHERE id > afterId ORDER BY id ASC LIMIT limit
|
||||
// Normal pagination uses cursor (opaque, encodes last seen id+timestamp).
|
||||
w.WriteHeader(http.StatusNotImplemented)
|
||||
}
|
||||
|
||||
// GetPacket handles GET /api/v1/packets/{packetHash}
|
||||
//
|
||||
// Returns the full packet with all observations and each observation's
|
||||
// resolved path inline.
|
||||
func GetPacket(w http.ResponseWriter, r *http.Request) {
|
||||
// packetHash := chi.URLParam(r, "packetHash")
|
||||
// TODO: fetch packet + observations, resolve paths, write JSON response.
|
||||
w.WriteHeader(http.StatusNotImplemented)
|
||||
}
|
||||
@@ -28,6 +28,25 @@ func PayloadTypeName(t int16) string {
|
||||
}
|
||||
}
|
||||
|
||||
// PayloadTypeFromString returns the integer payload type for a given name.
|
||||
// Returns 0 (no filter) if the string is empty or unrecognized.
|
||||
func PayloadTypeFromString(s string) int16 {
|
||||
switch strings.ToLower(s) {
|
||||
case "advert":
|
||||
return int16(meshcore.PayloadTypeAdvert)
|
||||
case "grp_txt", "grptxt", "group_text":
|
||||
return int16(meshcore.PayloadTypeGrpTxt)
|
||||
case "txt_msg", "txtmsg", "text":
|
||||
return int16(meshcore.PayloadTypeTxtMsg)
|
||||
case "trace":
|
||||
return int16(meshcore.PayloadTypeTrace)
|
||||
case "anon_req", "anonreq":
|
||||
return int16(meshcore.PayloadTypeAnonReq)
|
||||
default:
|
||||
return 0
|
||||
}
|
||||
}
|
||||
|
||||
// NodeTypeName returns a human-readable name for a node type integer.
|
||||
func NodeTypeName(t int16) string {
|
||||
switch byte(t) {
|
||||
@@ -60,3 +79,19 @@ func NodeTypeFromString(s string) int16 {
|
||||
return 0
|
||||
}
|
||||
}
|
||||
|
||||
// RouteTypeName returns a human-readable name for a route type integer.
|
||||
func RouteTypeName(t int16) string {
|
||||
switch byte(t) {
|
||||
case meshcore.RouteTypeFlood:
|
||||
return "FLOOD"
|
||||
case meshcore.RouteTypeDirect:
|
||||
return "DIRECT"
|
||||
case meshcore.RouteTypeTransportFlood:
|
||||
return "TRANSPORT_FLOOD"
|
||||
case meshcore.RouteTypeTransportDirect:
|
||||
return "TRANSPORT_DIRECT"
|
||||
default:
|
||||
return "unknown"
|
||||
}
|
||||
}
|
||||
@@ -9,6 +9,90 @@ import (
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
// PacketLatestObserver is the most recent observer summary rolled into a packet list item.
|
||||
type PacketLatestObserver struct {
|
||||
ID uuid.UUID `json:"id"`
|
||||
DisplayName *string `json:"displayName,omitempty"`
|
||||
IATA string `json:"iata"`
|
||||
}
|
||||
|
||||
// PacketSummary is the minimal packet representation used in list responses.
|
||||
// Includes the latest observation rolled in for display purposes.
|
||||
type PacketSummary struct {
|
||||
PacketHash string `json:"packetHash"` // hex-encoded
|
||||
PayloadType int16 `json:"payloadType"`
|
||||
PayloadTypeName string `json:"payloadTypeName"`
|
||||
RouteType int16 `json:"routeType"`
|
||||
RouteTypeName string `json:"routeTypeName"`
|
||||
FirstHeardAt int64 `json:"firstHeardAt"` // epoch ms
|
||||
LastHeardAt int64 `json:"lastHeardAt"` // epoch ms
|
||||
ObservationCount int32 `json:"observationCount"`
|
||||
LatestObserver *PacketLatestObserver `json:"latestObserver,omitempty"`
|
||||
Summary *string `json:"summary,omitempty"` // human-readable payload summary
|
||||
}
|
||||
|
||||
// PacketObservationDetail is a full observation including radio settings and resolved path.
|
||||
type PacketObservationDetail struct {
|
||||
ID int64 `json:"id"`
|
||||
ObserverID uuid.UUID `json:"observerId"`
|
||||
ObserverName *string `json:"observerName,omitempty"`
|
||||
IATA string `json:"iata"`
|
||||
HeardAt int64 `json:"heardAt"` // epoch ms
|
||||
PathLengthByte int16 `json:"pathLengthByte"`
|
||||
HashSize int16 `json:"hashSize"`
|
||||
HopCount int16 `json:"hopCount"`
|
||||
PathBytes *string `json:"pathBytes,omitempty"` // hex-encoded
|
||||
RSSI *int16 `json:"rssi,omitempty"`
|
||||
SNR *float32 `json:"snr,omitempty"`
|
||||
PropagationTimeMs *int32 `json:"propagationTimeMs,omitempty"`
|
||||
Radio *PacketRadio `json:"radio,omitempty"`
|
||||
SourceBroker string `json:"sourceBroker"`
|
||||
ResolvedPath []ResolvedHop `json:"resolvedPath"`
|
||||
}
|
||||
|
||||
// PacketRadio holds the radio settings from the observation.
|
||||
type PacketRadio struct {
|
||||
FreqMHz *float32 `json:"freqMhz,omitempty"`
|
||||
SpreadFactor *int16 `json:"spreadFactor,omitempty"`
|
||||
BandwidthKHz *float32 `json:"bandwidthKhz,omitempty"`
|
||||
CodingRate *int16 `json:"codingRate,omitempty"`
|
||||
}
|
||||
|
||||
// ResolvedHop is a single hop in a packet's resolved path.
|
||||
type ResolvedHop struct {
|
||||
Confidence string `json:"confidence"` // "high", "low", "unknown"
|
||||
Node *ResolvedNode `json:"node,omitempty"`
|
||||
}
|
||||
|
||||
// ResolvedNode is a node reference within a resolved path hop.
|
||||
type ResolvedNode struct {
|
||||
ID uuid.UUID `json:"id"`
|
||||
Name *string `json:"name,omitempty"`
|
||||
PublicKey string `json:"publicKey"` // hex-encoded prefix
|
||||
Latitude *float64 `json:"latitude,omitempty"`
|
||||
Longitude *float64 `json:"longitude,omitempty"`
|
||||
}
|
||||
|
||||
// Packet is the full packet representation including all observations and resolved paths.
|
||||
type Packet struct {
|
||||
PacketHash string `json:"packetHash"` // hex-encoded
|
||||
PayloadType int16 `json:"payloadType"`
|
||||
PayloadTypeName string `json:"payloadTypeName"`
|
||||
PayloadVersion int16 `json:"payloadVersion"`
|
||||
RouteType int16 `json:"routeType"`
|
||||
RouteTypeName string `json:"routeTypeName"`
|
||||
TransportCodes *string `json:"transportCodes,omitempty"` // hex-encoded
|
||||
OriginPubkey *string `json:"originPubkey,omitempty"` // hex-encoded
|
||||
ParsedPayload json.RawMessage `json:"parsedPayload,omitempty"`
|
||||
RawPayload string `json:"rawPayload"` // hex-encoded
|
||||
Decrypted bool `json:"decrypted"`
|
||||
ChannelHash *string `json:"channelHash,omitempty"` // hex-encoded
|
||||
FirstHeardAt int64 `json:"firstHeardAt"` // epoch ms
|
||||
LastHeardAt int64 `json:"lastHeardAt"` // epoch ms
|
||||
ObservationCount int32 `json:"observationCount"`
|
||||
Observations []PacketObservationDetail `json:"observations"`
|
||||
}
|
||||
|
||||
// AdvertObservation extends PacketObservationSummary with node identity fields
|
||||
// specific to advert packets (payload_type=4).
|
||||
type AdvertObservation struct {
|
||||
@@ -247,4 +331,12 @@ type Reader interface {
|
||||
// ListNodeObservations returns a paginated list of packet observations originating from a node.
|
||||
// Pass cursor=0 to start from the beginning.
|
||||
ListNodeObservations(ctx context.Context, nodeID uuid.UUID, cursor int64, limit int32) (Page[PacketObservationSummary], error)
|
||||
// ListPackets returns a paginated list of packets with the latest observation rolled in.
|
||||
// Pass 0 for payloadType/routeType to skip those filters.
|
||||
// Pass empty string for iata, 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, iata 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)
|
||||
}
|
||||
@@ -49,7 +49,7 @@ func New(h *hub.Hub, reader api.Reader, workers []*ingest.Worker) http.Handler {
|
||||
r.Route("/api/v1", func(r chi.Router) {
|
||||
// Public group — no authentication required (all of v1 is public).
|
||||
r.Group(func(r chi.Router) {
|
||||
r.Mount("/packets", handlers.PacketsRouter())
|
||||
r.Mount("/packets", handlers.PacketsRouter(reader))
|
||||
r.Mount("/nodes", handlers.NodesRouter(reader))
|
||||
r.Mount("/brokers", handlers.BrokersRouter(workers))
|
||||
r.Mount("/observers", handlers.ObserversRouter(reader))
|
||||
|
||||
Reference in new issue
Block a user