From 71f5cbae3b7927ef3faaebd82acb8b272163f66c Mon Sep 17 00:00:00 2001 From: "Enot (ded) Skelly" Date: Thu, 28 May 2026 13:07:22 -0700 Subject: [PATCH] add stats api --- README.md | 60 +++++---- db/migrations/001_schema.sql | 5 +- db/queries/queries.sql | 39 +++++- db/sqlc/models.go | 10 +- db/sqlc/queries.sql.go | 115 ++++++++++++++++- db/store.go | 122 +++++++++++++++++- internal/api/handlers/stats.go | 224 +++++++++++++++++++++------------ internal/api/reader.go | 63 ++++++++++ internal/api/router/router.go | 6 +- 9 files changed, 508 insertions(+), 136 deletions(-) diff --git a/README.md b/README.md index 4e29f5e..714806a 100644 --- a/README.md +++ b/README.md @@ -228,36 +228,31 @@ Base path: `/api/v1` ### Implemented -| Method | Path | Description | -| ------ | ------------------------- | -------------------------------------------------------------------------------------------------- | -| `GET` | `/brokers` | List MQTT brokers and connection status | -| `GET` | `/iatas` | List all known IATA codes | -| `GET` | `/iatas/{iata}` | Get a single IATA code | -| `GET` | `/regions` | List all regions (summary) | -| `GET` | `/regions/{id}` | Get a single region with IATA list | -| `GET` | `/channels` | List channels (optional: `?hash=&iata=&limit=50`) | -| `GET` | `/channels/{id}` | Get channel detail by integer ID | -| `GET` | `/channels/{id}/messages` | List messages for a channel (optional: `?since=&iata=&limit=50`) | -| `GET` | `/messages` | List all messages (optional: `?channelId=&channelHash=&iata=&since=&limit=50`) | -| `GET` | `/observers` | List observers (optional: `?iata=&type=&broker=&status=online\|offline`) | -| `GET` | `/observers/{observerId}` | Get observer detail including broker last-seen timestamps | - -### Stubbed (501 Not Implemented) - -| Method | Path | Description | -| ------ | ----------------------------------- | ---------------------------------- | -| `GET` | `/packets` | List packets with filters | -| `GET` | `/packets/{packetHash}` | Get packet with all observations | -| `GET` | `/nodes` | List nodes | -| `GET` | `/nodes/{nodeId}` | Get node detail | -| `GET` | `/nodes/{nodeId}/observations` | List observations for a node | -| `GET` | `/observers/{observerId}/telemetry` | Observer telemetry history | -| `GET` | `/observers/{observerId}/adverts` | Adverts heard by observer | -| `GET` | `/stats/overview` | Network overview stats | -| `GET` | `/stats/observations` | Observation time series | -| `GET` | `/stats/payloadBreakdown` | Observations by payload type | -| `GET` | `/stats/topNodes` | Top nodes by observation count | -| `GET` | `/stats/topObservers` | Top observers by observation count | +| Method | Path | Description | +| ------ | ----------------------------------- | -------------------------------------------------------------------------------------------------- | +| `GET` | `/brokers` | List MQTT brokers and connection status | +| `GET` | `/iatas` | List all known IATA codes | +| `GET` | `/iatas/{iata}` | Get a single IATA code | +| `GET` | `/regions` | List all regions (summary) | +| `GET` | `/regions/{id}` | Get a single region with IATA list | +| `GET` | `/channels` | List channels (optional: `?hash=&iata=&limit=50`) | +| `GET` | `/channels/{id}` | Get channel detail by integer ID | +| `GET` | `/channels/{id}/messages` | List messages for a channel (optional: `?since=&iata=&limit=50`) | +| `GET` | `/messages` | List all messages (optional: `?channelId=&channelHash=&iata=&since=&limit=50`) | +| `GET` | `/observers` | List observers (optional: `?iata=&type=&broker=&status=online\|offline`) | +| `GET` | `/observers/{observerId}` | Get observer detail including broker last-seen timestamps | +| `GET` | `/packets` | List packets with filters | +| `GET` | `/packets/{packetHash}` | Get packet with all observations | +| `GET` | `/nodes` | List nodes | +| `GET` | `/nodes/{nodeId}` | Get node detail | +| `GET` | `/nodes/{nodeId}/observations` | List observations for a node | +| `GET` | `/observers/{observerId}/telemetry` | Observer telemetry history | +| `GET` | `/observers/{observerId}/adverts` | Adverts heard by observer | +| `GET` | `/stats/overview` | Network overview stats | +| `GET` | `/stats/observations` | Hourly observation time series (last 7 days by default) | +| `GET` | `/stats/payload-breakdown` | Observation counts by payload type (last 24h by default) | +| `GET` | `/stats/top-nodes` | Top N nodes by observation count (from materialized view) | +| `GET` | `/stats/top-observers` | Top N observers by observation count (last 24h by default) | --- @@ -310,10 +305,11 @@ Tower server. - [x] REST API: Pagination - [x] REST API: Nodes (list + detail + observations) - [x] REST API: Packets (list + detail) +- [x] REST API: Stats ### In progress / next -- [ ] REST API: Stats +- [ ] Materialized view refresh (mv_hourly_iata_stats, mv_top_nodes_by_iata) - [ ] Path resolution (node short ID lookup) - [ ] Propagation time calculation - [ ] Routes and traces endpoints @@ -328,4 +324,4 @@ Tower server. - [ ] Region management via API (currently config-file only) - [ ] WebSocket subscription unsubscribe (scaffolded, not implemented) - [ ] Observer owner tracking (schema exists, API excluded by design) -- [ ] Swagger/OpenAPI documentation +- [ ] Swagger/OpenAPI documentation via swaggo/swag diff --git a/db/migrations/001_schema.sql b/db/migrations/001_schema.sql index 714120d..362ca42 100644 --- a/db/migrations/001_schema.sql +++ b/db/migrations/001_schema.sql @@ -312,7 +312,7 @@ CREATE INDEX idx_channel_messages_sent_brin ON channel_messages USING BRIN (sent CREATE MATERIALIZED VIEW mv_hourly_iata_stats AS SELECT iata, - date_trunc('hour', heard_at) AS hour, + date_trunc('hour', heard_at)::timestamptz AS hour, COUNT(*) AS observation_count, COUNT(DISTINCT packet_hash) AS unique_packets, COUNT(DISTINCT observer_id) AS active_observers @@ -320,9 +320,6 @@ FROM packet_observations WHERE heard_at > NOW() - INTERVAL '7 days' GROUP BY iata, date_trunc('hour', heard_at); -CREATE UNIQUE INDEX idx_mv_hourly_iata - ON mv_hourly_iata_stats(iata, hour); - CREATE MATERIALIZED VIEW mv_top_nodes_by_iata AS SELECT ni.iata, diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 8b78fcb..540bf55 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -488,11 +488,12 @@ SELECT COUNT(DISTINCT po.iata) AS active_iatas FROM packet_observations po WHERE po.heard_at > NOW() - INTERVAL '24 hours' - AND ($1::char(3) IS NULL OR po.iata = $1); + AND ($1 = '' OR po.iata ILIKE $1); -- name: GetHourlyStats :many -SELECT * FROM mv_hourly_iata_stats -WHERE ($1::char(3) IS NULL OR iata = $1) +SELECT iata, hour, observation_count, unique_packets, active_observers +FROM mv_hourly_iata_stats +WHERE ($1 = '' OR iata ILIKE $1) AND hour >= NOW() - $2::interval ORDER BY iata, hour; @@ -502,6 +503,38 @@ WHERE ($1::char(3) IS NULL OR iata = $1) ORDER BY observation_count DESC LIMIT $2; +-- name: GetStatsPayloadBreakdown :many +-- Returns observation counts grouped by payload type for the given window and IATA. +SELECT + p.payload_type, + COUNT(*) AS count +FROM packet_observations po +JOIN packets p ON p.packet_hash = po.packet_hash +WHERE po.heard_at > $1 + AND ($2 = '' OR po.iata ILIKE $2) +GROUP BY p.payload_type +ORDER BY count DESC; + +-- name: GetStatsTopObservers :many +-- Returns the top N observers by observation count for the given window and IATA. +SELECT + o.id, + o.display_name, + o.observer_type, + COUNT(*) AS observation_count, + COALESCE(( + SELECT po2.iata FROM packet_observations po2 + WHERE po2.observer_id = o.id + ORDER BY po2.heard_at DESC LIMIT 1 + ), '') AS iata +FROM packet_observations po +JOIN observers o ON o.id = po.observer_id +WHERE po.heard_at > $1 + AND ($2 = '' OR po.iata ILIKE $2) +GROUP BY o.id +ORDER BY observation_count DESC +LIMIT $3; + -- ============================================================ -- REGIONS -- ============================================================ diff --git a/db/sqlc/models.go b/db/sqlc/models.go index 6d423dc..21b64ee 100644 --- a/db/sqlc/models.go +++ b/db/sqlc/models.go @@ -50,11 +50,11 @@ type IataCode struct { } type MvHourlyIataStat struct { - Iata string `json:"iata"` - Hour pgtype.Interval `json:"hour"` - ObservationCount int64 `json:"observation_count"` - UniquePackets int64 `json:"unique_packets"` - ActiveObservers int64 `json:"active_observers"` + Iata string `json:"iata"` + Hour pgtype.Timestamptz `json:"hour"` + ObservationCount int64 `json:"observation_count"` + UniquePackets int64 `json:"unique_packets"` + ActiveObservers int64 `json:"active_observers"` } type MvTopNodesByIatum struct { diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 60ea503..1633b6b 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -150,14 +150,15 @@ func (q *Queries) GetChannelsByHash(ctx context.Context, arg GetChannelsByHashPa } const getHourlyStats = `-- name: GetHourlyStats :many -SELECT iata, hour, observation_count, unique_packets, active_observers FROM mv_hourly_iata_stats -WHERE ($1::char(3) IS NULL OR iata = $1) +SELECT iata, hour, observation_count, unique_packets, active_observers +FROM mv_hourly_iata_stats +WHERE ($1 = '' OR iata ILIKE $1) AND hour >= NOW() - $2::interval ORDER BY iata, hour ` type GetHourlyStatsParams struct { - Column1 string `json:"column_1"` + Column1 interface{} `json:"column_1"` Column2 pgtype.Interval `json:"column_2"` } @@ -552,7 +553,7 @@ SELECT COUNT(DISTINCT po.iata) AS active_iatas FROM packet_observations po WHERE po.heard_at > NOW() - INTERVAL '24 hours' - AND ($1::char(3) IS NULL OR po.iata = $1) + AND ($1 = '' OR po.iata ILIKE $1) ` type GetStatsOverviewRow struct { @@ -565,7 +566,7 @@ type GetStatsOverviewRow struct { // ============================================================ // STATS // ============================================================ -func (q *Queries) GetStatsOverview(ctx context.Context, dollar_1 string) (GetStatsOverviewRow, error) { +func (q *Queries) GetStatsOverview(ctx context.Context, dollar_1 interface{}) (GetStatsOverviewRow, error) { row := q.db.QueryRow(ctx, getStatsOverview, dollar_1) var i GetStatsOverviewRow err := row.Scan( @@ -577,6 +578,110 @@ func (q *Queries) GetStatsOverview(ctx context.Context, dollar_1 string) (GetSta return i, err } +const getStatsPayloadBreakdown = `-- name: GetStatsPayloadBreakdown :many +SELECT + p.payload_type, + COUNT(*) AS count +FROM packet_observations po +JOIN packets p ON p.packet_hash = po.packet_hash +WHERE po.heard_at > $1 + AND ($2 = '' OR po.iata ILIKE $2) +GROUP BY p.payload_type +ORDER BY count DESC +` + +type GetStatsPayloadBreakdownParams struct { + HeardAt pgtype.Timestamptz `json:"heard_at"` + Column2 interface{} `json:"column_2"` +} + +type GetStatsPayloadBreakdownRow struct { + PayloadType int16 `json:"payload_type"` + Count int64 `json:"count"` +} + +// Returns observation counts grouped by payload type for the given window and IATA. +func (q *Queries) GetStatsPayloadBreakdown(ctx context.Context, arg GetStatsPayloadBreakdownParams) ([]GetStatsPayloadBreakdownRow, error) { + rows, err := q.db.Query(ctx, getStatsPayloadBreakdown, arg.HeardAt, arg.Column2) + if err != nil { + return nil, err + } + defer rows.Close() + items := []GetStatsPayloadBreakdownRow{} + for rows.Next() { + var i GetStatsPayloadBreakdownRow + if err := rows.Scan(&i.PayloadType, &i.Count); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const getStatsTopObservers = `-- name: GetStatsTopObservers :many +SELECT + o.id, + o.display_name, + o.observer_type, + COUNT(*) AS observation_count, + COALESCE(( + SELECT po2.iata FROM packet_observations po2 + WHERE po2.observer_id = o.id + ORDER BY po2.heard_at DESC LIMIT 1 + ), '') AS iata +FROM packet_observations po +JOIN observers o ON o.id = po.observer_id +WHERE po.heard_at > $1 + AND ($2 = '' OR po.iata ILIKE $2) +GROUP BY o.id +ORDER BY observation_count DESC +LIMIT $3 +` + +type GetStatsTopObserversParams struct { + HeardAt pgtype.Timestamptz `json:"heard_at"` + Column2 interface{} `json:"column_2"` + Limit int32 `json:"limit"` +} + +type GetStatsTopObserversRow struct { + ID uuid.UUID `json:"id"` + DisplayName *string `json:"display_name"` + ObserverType *string `json:"observer_type"` + ObservationCount int64 `json:"observation_count"` + Iata interface{} `json:"iata"` +} + +// Returns the top N observers by observation count for the given window and IATA. +func (q *Queries) GetStatsTopObservers(ctx context.Context, arg GetStatsTopObserversParams) ([]GetStatsTopObserversRow, error) { + rows, err := q.db.Query(ctx, getStatsTopObservers, arg.HeardAt, arg.Column2, arg.Limit) + if err != nil { + return nil, err + } + defer rows.Close() + items := []GetStatsTopObserversRow{} + for rows.Next() { + var i GetStatsTopObserversRow + if err := rows.Scan( + &i.ID, + &i.DisplayName, + &i.ObserverType, + &i.ObservationCount, + &i.Iata, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const getTopNodes = `-- name: GetTopNodes :many SELECT iata, node_id, name, node_type, observation_count, last_heard FROM mv_top_nodes_by_iata WHERE ($1::char(3) IS NULL OR iata = $1) diff --git a/db/store.go b/db/store.go index 4725e6d..0eb88bd 100644 --- a/db/store.go +++ b/db/store.go @@ -386,14 +386,14 @@ func (s *Store) UpsertRegionIATA(ctx context.Context, regionID int32, iata strin // cursor is last_seen epoch ms of the last item; pass 0 to start from the beginning. // Note: after sqlc generate, verify Column param names match generated types. func (s *Store) ListChannels(ctx context.Context, limit int32, hash []byte, iata string, cursor int64) (api.Page[api.ChannelSummary], error) { - var cursorTs pgtype.Timestamptz + var cursorTS pgtype.Timestamptz if cursor > 0 { - cursorTs = pgtype.Timestamptz{Time: time.UnixMilli(cursor), Valid: true} + cursorTS = pgtype.Timestamptz{Time: time.UnixMilli(cursor), Valid: true} } rows, err := s.q.ListChannels(ctx, sqlc.ListChannelsParams{ Column1: hash, Column2: iata, - Column3: cursorTs, + Column3: cursorTS, Limit: limit + 1, }) if err != nil { @@ -1048,3 +1048,119 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet, } return p, nil } + +// GetStatsOverview returns top-line network figures for the last 24 hours. +func (s *Store) GetStatsOverview(ctx context.Context, iata string) (*api.StatsOverview, error) { + row, err := s.q.GetStatsOverview(ctx, iata) + if err != nil { + return nil, err + } + return &api.StatsOverview{ + TotalPackets: row.TotalPackets, + TotalObservations: row.TotalObservations, + ActiveObservers: row.ActiveObservers, + ActiveIATAs: row.ActiveIatas, + WindowHours: 24, + }, nil +} + +// GetStatsObservations returns hourly observation counts for charting. +func (s *Store) GetStatsObservations(ctx context.Context, iata string, since time.Time) ([]api.ObservationPoint, error) { + if since.IsZero() { + since = time.Now().Add(-7 * 24 * time.Hour) + } + interval := time.Since(since) + rows, err := s.q.GetHourlyStats(ctx, sqlc.GetHourlyStatsParams{ + Column1: iata, + Column2: pgtype.Interval{Microseconds: int64(interval.Hours()) * 3600 * 1e6, Valid: true}, + }) + if err != nil { + return nil, err + } + points := make([]api.ObservationPoint, 0, len(rows)) + for _, v := range rows { + points = append(points, api.ObservationPoint{ + Hour: v.Hour.Time.UnixMilli(), + IATA: v.Iata, + ObservationCount: v.ObservationCount, + UniquePackets: v.UniquePackets, + ActiveObservers: v.ActiveObservers, + }) + } + return points, nil +} + +func (s *Store) GetStatsPayloadBreakdown(ctx context.Context, iata string, since time.Time) ([]api.PayloadBreakdownItem, error) { + if since.IsZero() { + since = time.Now().Add(-24 * time.Hour) + } + rows, err := s.q.GetStatsPayloadBreakdown(ctx, sqlc.GetStatsPayloadBreakdownParams{ + HeardAt: pgtype.Timestamptz{Time: since, Valid: true}, + Column2: iata, + }) + if err != nil { + return nil, err + } + items := make([]api.PayloadBreakdownItem, 0, len(rows)) + for _, v := range rows { + items = append(items, api.PayloadBreakdownItem{ + PayloadType: v.PayloadType, + PayloadTypeName: api.PayloadTypeName(v.PayloadType), + Count: v.Count, + }) + } + return items, nil +} + +func (s *Store) GetStatsTopNodes(ctx context.Context, iata string, limit int32) ([]api.TopNode, error) { + rows, err := s.q.GetTopNodes(ctx, sqlc.GetTopNodesParams{ + Column1: iata, + Limit: limit, + }) + if err != nil { + return nil, err + } + items := make([]api.TopNode, 0, len(rows)) + for _, v := range rows { + var count int64 + if v.ObservationCount != nil { + count = *v.ObservationCount + } + items = append(items, api.TopNode{ + NodeID: v.NodeID, + NodeName: v.Name, + NodeType: v.NodeType, + NodeTypeName: api.NodeTypeName(v.NodeType), + IATA: v.Iata, + ObservationCount: count, + LastHeard: v.LastHeard.Time.UnixMilli(), + }) + } + return items, nil +} + +func (s *Store) GetStatsTopObservers(ctx context.Context, iata string, since time.Time, limit int32) ([]api.TopObserver, error) { + if since.IsZero() { + since = time.Now().Add(-24 * time.Hour) + } + rows, err := s.q.GetStatsTopObservers(ctx, sqlc.GetStatsTopObserversParams{ + HeardAt: pgtype.Timestamptz{Time: since, Valid: true}, + Column2: iata, + Limit: limit, + }) + if err != nil { + return nil, err + } + items := make([]api.TopObserver, 0, len(rows)) + for _, v := range rows { + iata, _ := v.Iata.(string) + items = append(items, api.TopObserver{ + ObserverID: v.ID, + DisplayName: v.DisplayName, + ObserverType: v.ObserverType, + IATA: iata, + ObservationCount: v.ObservationCount, + }) + } + return items, nil +} diff --git a/internal/api/handlers/stats.go b/internal/api/handlers/stats.go index 7481a25..851d6b7 100644 --- a/internal/api/handlers/stats.go +++ b/internal/api/handlers/stats.go @@ -1,99 +1,157 @@ package handlers import ( + "log" "net/http" + "strconv" + "time" + "github.com/MeshCore-Tower/tower-server/internal/api" "github.com/go-chi/chi/v5" ) // StatsRouter mounts all /stats routes onto a subrouter. // -// GET /stats/overview → GetStatsOverview -// GET /stats/observations → GetStatsObservations -// GET /stats/payloadBreakdown → GetStatsPayloadBreakdown -// GET /stats/topNodes → GetStatsTopNodes -// GET /stats/topObservers → GetStatsTopObservers +// GET /stats/overview → GetStatsOverview +// GET /stats/observations → GetStatsObservations +// GET /stats/payload-breakdown → GetStatsPayloadBreakdown +// GET /stats/top-nodes → GetStatsTopNodes +// GET /stats/top-observers → GetStatsTopObservers // -// All endpoints accept either iata= (one or comma-separated) or regionId= -// (expands to all IATAs in that super-region via region_iatas). -func StatsRouter() http.Handler { +// All endpoints accept an optional iata= filter (case-insensitive). +// regionId= expansion and comma-separated IATAs are not yet implemented. +func StatsRouter(reader api.Reader) http.Handler { r := chi.NewRouter() - r.Get("/overview", GetStatsOverview) - r.Get("/observations", GetStatsObservations) - r.Get("/payloadBreakdown", GetStatsPayloadBreakdown) - r.Get("/topNodes", GetStatsTopNodes) - r.Get("/topObservers", GetStatsTopObservers) + // GET /api/v1/stats/overview + // + // Query params (all optional): + // + // iata= filter to a single IATA (case-insensitive) + r.Get("/overview", func(w http.ResponseWriter, r *http.Request) { + iata := r.URL.Query().Get("iata") + overview, err := reader.GetStatsOverview(r.Context(), iata) + if err != nil { + log.Printf("api: GetStatsOverview failed: %v", err) + respondError(w, http.StatusInternalServerError, "internal server error") + return + } + respond(w, http.StatusOK, overview) + }) + + // GET /api/v1/stats/observations + // + // Query params (all optional): + // + // iata= filter to a single IATA (case-insensitive) + // since= start of window; defaults to 7 days ago + r.Get("/observations", func(w http.ResponseWriter, r *http.Request) { + iata := r.URL.Query().Get("iata") + var since 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) + } + points, err := reader.GetStatsObservations(r.Context(), iata, since) + if err != nil { + log.Printf("api: GetStatsObservations failed: %v", err) + respondError(w, http.StatusInternalServerError, "internal server error") + return + } + respond(w, http.StatusOK, points) + }) + + // GET /api/v1/stats/payload-breakdown + // + // Query params (all optional): + // + // iata= filter to a single IATA (case-insensitive) + // since= start of window; defaults to last 24h + r.Get("/payload-breakdown", func(w http.ResponseWriter, r *http.Request) { + iata := r.URL.Query().Get("iata") + var since 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) + } + breakdown, err := reader.GetStatsPayloadBreakdown(r.Context(), iata, since) + if err != nil { + log.Printf("api: GetStatsPayloadBreakdown failed: %v", err) + respondError(w, http.StatusInternalServerError, "internal server error") + return + } + respond(w, http.StatusOK, breakdown) + }) + + // GET /api/v1/stats/top-nodes + // + // Query params (all optional): + // + // iata= filter to a single IATA (case-insensitive) + // limit=10 + r.Get("/top-nodes", func(w http.ResponseWriter, r *http.Request) { + iata := r.URL.Query().Get("iata") + var limit int32 = 10 + 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) + } + nodes, err := reader.GetStatsTopNodes(r.Context(), iata, limit) + if err != nil { + log.Printf("api: GetStatsTopNodes failed: %v", err) + respondError(w, http.StatusInternalServerError, "internal server error") + return + } + respond(w, http.StatusOK, nodes) + }) + + // GET /api/v1/stats/top-observers + // + // Query params (all optional): + // + // iata= filter to a single IATA (case-insensitive) + // since= start of window; defaults to last 24h + // limit=10 + r.Get("/top-observers", func(w http.ResponseWriter, r *http.Request) { + iata := r.URL.Query().Get("iata") + var since 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) + } + var limit int32 = 10 + 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) + } + observers, err := reader.GetStatsTopObservers(r.Context(), iata, since, limit) + if err != nil { + log.Printf("api: GetStatsTopObservers failed: %v", err) + respondError(w, http.StatusInternalServerError, "internal server error") + return + } + respond(w, http.StatusOK, observers) + }) return r } - -// GetStatsOverview handles GET /api/v1/stats/overview -// -// Query params (all optional): -// iata=YOW (one or comma-separated) -// regionId= -// -// Returns top-line figures: total packets and observations last 24h, -// active observers, active IATAs, unique nodes seen. -func GetStatsOverview(w http.ResponseWriter, r *http.Request) { - // TODO: query mv_hourly_iata_stats + live aggregates, write JSON response. - w.WriteHeader(http.StatusNotImplemented) -} - -// GetStatsObservations handles GET /api/v1/stats/observations -// -// Query params (all optional): -// iata=YOW -// regionId= -// range=24h (duration string: 24h, 7d, 30d) -// interval=1h (bucket size: 5m, 1h, 1d) -// -// Returns a time series of observation counts bucketed by interval, -// suitable for charting. -func GetStatsObservations(w http.ResponseWriter, r *http.Request) { - // TODO: query mv_hourly_iata_stats, group by bucket, write JSON response. - w.WriteHeader(http.StatusNotImplemented) -} - -// GetStatsPayloadBreakdown handles GET /api/v1/stats/payloadBreakdown -// -// Query params (all optional): -// iata=YOW -// regionId= -// range=24h -// -// Returns observation counts grouped by payload_type for the given window. -func GetStatsPayloadBreakdown(w http.ResponseWriter, r *http.Request) { - // TODO: query packets + observations filtered by time/iata, group by payload_type. - w.WriteHeader(http.StatusNotImplemented) -} - -// GetStatsTopNodes handles GET /api/v1/stats/topNodes -// -// Query params (all optional): -// iata=YOW -// regionId= -// range=24h -// limit=10 -// -// Returns the top N nodes by observation contribution count using -// mv_top_nodes_by_iata. -func GetStatsTopNodes(w http.ResponseWriter, r *http.Request) { - // TODO: query mv_top_nodes_by_iata, apply filters, write JSON response. - w.WriteHeader(http.StatusNotImplemented) -} - -// GetStatsTopObservers handles GET /api/v1/stats/topObservers -// -// Query params (all optional): -// iata=YOW -// regionId= -// range=24h -// limit=10 -// -// Returns the top N observers by observation count for the given window. -func GetStatsTopObservers(w http.ResponseWriter, r *http.Request) { - // TODO: query packet_observations grouped by observer_id, apply filters. - w.WriteHeader(http.StatusNotImplemented) -} diff --git a/internal/api/reader.go b/internal/api/reader.go index 8c29883..419765e 100644 --- a/internal/api/reader.go +++ b/internal/api/reader.go @@ -9,6 +9,51 @@ import ( "github.com/google/uuid" ) +// StatsOverview is the top-level network summary for the overview endpoint. +type StatsOverview struct { + TotalPackets int64 `json:"totalPackets"` + TotalObservations int64 `json:"totalObservations"` + ActiveObservers int64 `json:"activeObservers"` + ActiveIATAs int64 `json:"activeIatas"` + WindowHours int `json:"windowHours"` // always 24 for now +} + +// ObservationPoint is a single time-bucketed observation count. +type ObservationPoint struct { + Hour int64 `json:"hour"` // epoch ms, start of bucket + IATA string `json:"iata"` + ObservationCount int64 `json:"observationCount"` + UniquePackets int64 `json:"uniquePackets"` + ActiveObservers int64 `json:"activeObservers"` +} + +// PayloadBreakdownItem is a single payload type with its observation count. +type PayloadBreakdownItem struct { + PayloadType int16 `json:"payloadType"` + PayloadTypeName string `json:"payloadTypeName"` + Count int64 `json:"count"` +} + +// TopNode is a node ranked by observation count. +type TopNode struct { + NodeID uuid.UUID `json:"nodeId"` + NodeName *string `json:"nodeName,omitempty"` + NodeType int16 `json:"nodeType"` + NodeTypeName string `json:"nodeTypeName"` + IATA string `json:"iata"` + ObservationCount int64 `json:"observationCount"` + LastHeard int64 `json:"lastHeard"` // epoch ms +} + +// TopObserver is an observer ranked by observation count. +type TopObserver struct { + ObserverID uuid.UUID `json:"observerId"` + DisplayName *string `json:"displayName,omitempty"` + ObserverType *string `json:"observerType,omitempty"` + IATA string `json:"iata"` + ObservationCount int64 `json:"observationCount"` +} + // PacketLatestObserver is the most recent observer summary rolled into a packet list item. type PacketLatestObserver struct { ID uuid.UUID `json:"id"` @@ -339,4 +384,22 @@ type Reader interface { // 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) + // GetStatsOverview returns top-line network figures for the last 24 hours. + // Pass empty string iata to return stats across all IATAs. + GetStatsOverview(ctx context.Context, iata string) (*StatsOverview, error) + // GetStatsObservations returns hourly observation counts for charting. + // Pass empty string iata to return stats across all IATAs. + // since defines the start of the window; pass zero time for default (last 7 days). + GetStatsObservations(ctx context.Context, iata string, since time.Time) ([]ObservationPoint, error) + // GetStatsPayloadBreakdown returns observation counts grouped by payload type. + // Pass empty string iata to return stats across all IATAs. + // since defines the start of the window; pass zero time for default (last 24h). + GetStatsPayloadBreakdown(ctx context.Context, iata string, since time.Time) ([]PayloadBreakdownItem, error) + // GetStatsTopNodes returns the top N nodes by observation count. + // Pass empty string iata to return stats across all IATAs. + GetStatsTopNodes(ctx context.Context, iata string, limit int32) ([]TopNode, error) + // GetStatsTopObservers returns the top N observers by observation count. + // Pass empty string iata to return stats across all IATAs. + // since defines the start of the window; pass zero time for default (last 24h). + GetStatsTopObservers(ctx context.Context, iata string, since time.Time, limit int32) ([]TopObserver, error) } diff --git a/internal/api/router/router.go b/internal/api/router/router.go index 091f8ba..214b431 100644 --- a/internal/api/router/router.go +++ b/internal/api/router/router.go @@ -1,3 +1,7 @@ +// Package router wires all HTTP routes onto the Chi router and injects +// dependencies (hub, reader, ingest workers) into the handler closures. +// All routes are mounted under /api/v1 with public and private groups +// stubbed for future auth middleware. package router import ( @@ -57,7 +61,7 @@ func New(h *hub.Hub, reader api.Reader, workers []*ingest.Worker) http.Handler { r.Mount("/messages", handlers.MessagesRouter(reader)) r.Mount("/iatas", handlers.IATAsRouter(reader)) r.Mount("/regions", handlers.RegionsRouter(reader)) - r.Mount("/stats", handlers.StatsRouter()) + r.Mount("/stats", handlers.StatsRouter(reader)) }) // Private group — auth middleware applied.