From 203c5f1b2fc0f5c20fa9cfe3e354f371fd5cc597 Mon Sep 17 00:00:00 2001 From: "Enot (ded) Skelly" Date: Thu, 28 May 2026 12:22:49 -0700 Subject: [PATCH] add packets api --- README.md | 2 +- db/queries/queries.sql | 43 +++++++--- db/sqlc/queries.sql.go | 115 ++++++++++++++++++------- db/store.go | 134 +++++++++++++++++++++++++++++ internal/api/handlers/packets.go | 139 +++++++++++++++++++++++-------- internal/api/helpers.go | 35 ++++++++ internal/api/reader.go | 92 ++++++++++++++++++++ internal/api/router/router.go | 2 +- 8 files changed, 483 insertions(+), 79 deletions(-) diff --git a/README.md b/README.md index 94c6722..4e29f5e 100644 --- a/README.md +++ b/README.md @@ -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 diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 58ff014..8b78fcb 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -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 diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index cdbb691..60ea503 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -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 } diff --git a/db/store.go b/db/store.go index 96d340e..4725e6d 100644 --- a/db/store.go +++ b/db/store.go @@ -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 +} diff --git a/internal/api/handlers/packets.go b/internal/api/handlers/packets.go index 7548588..d9e8605 100644 --- a/internal/api/handlers/packets.go +++ b/internal/api/handlers/packets.go @@ -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= filter by payload type integer + // payloadTypeName= filter by payload type name (advert, grp_txt, txt_msg, trace, anon_req) + // routeType= filter by route type integer (0=transport_flood, 1=flood, 2=direct, 3=transport_direct) + // iata= filter by latest observation IATA (case-insensitive) + // since= filter by first_heard_at >= since + // until= filter by first_heard_at <= until + // cursor= 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= -// until= -// afterId= used for deterministic WS reconnection backfill -// limit=50 -// cursor= -// -// 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) -} diff --git a/internal/api/helpers.go b/internal/api/helpers.go index 90cffd7..731bd25 100644 --- a/internal/api/helpers.go +++ b/internal/api/helpers.go @@ -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" + } +} diff --git a/internal/api/reader.go b/internal/api/reader.go index 1c427c7..8c29883 100644 --- a/internal/api/reader.go +++ b/internal/api/reader.go @@ -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) } diff --git a/internal/api/router/router.go b/internal/api/router/router.go index 60e781c..091f8ba 100644 --- a/internal/api/router/router.go +++ b/internal/api/router/router.go @@ -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))